feat: dctap-python files

dctap/__init__.py

@@ -0,0 +1,12 @@
+"""Convert a Tabular Application Profile from CSV into JSON."""
+
+from dctap.tapclasses import TAPShape, TAPStatementTemplate
+from dctap.csvreader import csvreader
+
+__version__ = "0.4.5"
+
+# Keep version number in sync with:
+# - https://github.com/dcmi/dctap-python/blob/main/docs/conf.py#L28
+#   ../docs/conf.py
+# - https://github.com/dcmi/dctap-python/blob/main/dctap/cli.py#L20
+#   ../dctap/cli.py

dctap/cli.py

@@ -0,0 +1,97 @@
+"""DC Tabular Application Profiles (DCTAP) command-line utility."""
+
+import sys
+import json as j
+from ruamel.yaml import YAML
+import click
+from dctap.defaults import CONFIGFILE
+from dctap.config import get_config, write_configfile
+from dctap.csvreader import csvreader
+from dctap.inspect import pprint_tapshapes, print_warnings
+from dctap.loggers import stderr_logger
+from dctap.utils import expand_uri_prefixes
+
+# pylint: disable=unused-argument,no-value-for-parameter
+# => unused-argument: Allows placeholders for now.
+# => no-value-for-parameter: Okay in cli.py
+
+
+@click.group()
+@click.version_option("0.4.5", help="Show version and exit")
+@click.help_option(help="Show help and exit")
+@click.pass_context
+def cli(context):
+    """DC Tabular Application Profiles parser and base module
+
+    Examples (see https://dctap-python.rtfd.io):
+
+    \b
+    Write starter config file:
+    $ dctap init                           # Write dctap.yaml
+    \b
+    Show normalized view of TAP:
+    $ dctap read x.csv                     # Output as plain text
+    $ dctap read --json x.csv              # Output as JSON
+    $ dctap read --yaml x.csv              # Output as YAML
+    $ dctap read --warnings x.csv          # Also show warnings
+    """
+
+
+@cli.command()
+@click.help_option(help="Show help and exit")
+@click.pass_context
+def init(context, hidden):
+    """Write config file: 'dctap.yaml'."""
+    configfile = CONFIGFILE
+    write_configfile(configfile)
+
+
+@cli.command()
+@click.argument("csvfile_obj", type=click.File(mode="r", encoding="utf-8-sig"))
+@click.option("--config", type=click.Path(exists=True), help="Alternative config file")
+@click.option("--expand-prefixes", is_flag=True, help="Expand compact IRIs")
+@click.option("--warnings", is_flag=True, help="Print warnings to stderr")
+@click.option("--json", is_flag=True, help="Print JSON to stdout")
+@click.option("--yaml", is_flag=True, help="Print YAML to stdout")
+@click.help_option(help="Show help and exit")
+@click.pass_context
+def read(context, csvfile_obj, config, expand_prefixes, warnings, json, yaml):
+    """Show TAP as TXT, JSON, or YAML."""
+    # pylint: disable=too-many-locals,too-many-arguments
+
+    if config:
+        config_dict = get_config(nondefault_configfile_name=config)
+    else:
+        config_dict = get_config()
+    tapshapes_dict = csvreader(open_csvfile_obj=csvfile_obj, config_dict=config_dict)
+
+    if expand_prefixes:
+        tapshapes_dict = expand_uri_prefixes(tapshapes_dict, config_dict)
+
+    if json and yaml:
+        # Quick fix for mutually exclusive options, a better fix in future.
+        echo = stderr_logger()
+        echo.warning("Please use either --json or --yaml")
+        click.Context.exit(0)
+
+    if json:
+        if not warnings:
+            del tapshapes_dict["warnings"]
+        json_output = j.dumps(tapshapes_dict, indent=2)
+        print(json_output)
+
+    if yaml:
+        if not warnings:
+            del tapshapes_dict["warnings"]
+        y = YAML()
+        y.indent(mapping=2, sequence=4, offset=2)
+        y.dump(tapshapes_dict, sys.stdout)
+
+    if not (json or yaml):
+        pprint_output = pprint_tapshapes(
+            tapshapes_dict=tapshapes_dict, config_dict=config_dict
+        )
+        for line in pprint_output:
+            print(line, file=sys.stdout)
+        if warnings:
+            print_warnings(tapshapes_dict["warnings"])

dctap/config.py

@@ -0,0 +1,156 @@
+"""Default settings."""
+
+import sys
+from dataclasses import asdict
+from pathlib import Path
+from dctap.defaults import CONFIGFILE, CONFIGYAML
+from dctap.exceptions import ConfigError
+from dctap.tapclasses import TAPShape, TAPStatementTemplate
+from dctap.utils import load_yaml_to_dict, coerce_concise
+
+
+def get_config(
+    nondefault_configyaml_str=None,
+    nondefault_configfile_name=None,
+    default_configyaml_str=CONFIGYAML,
+    default_configfile_name=CONFIGFILE,
+    default_shape_class=TAPShape,
+    default_state_class=TAPStatementTemplate,
+):
+    """Get configuration dictionary from package defaults (or from non-defaults)."""
+    # pylint: disable=too-many-arguments
+    yamlstring = None
+    configdict_from_yamlstring = None
+    config_dict = _initialize_config_dict(default_shape_class, default_state_class)
+
+    if nondefault_configfile_name and nondefault_configyaml_str:
+        raise ConfigError("Can load YAML from either string or file, not both.")
+
+    if nondefault_configfile_name:
+        nondefault_configfile = Path(nondefault_configfile_name)
+        try:
+            yamlstring = nondefault_configfile.read_text(encoding="utf-8")
+        except FileNotFoundError as err:
+            message = f"Config file '{nondefault_configfile_name}' not found."
+            raise ConfigError(message) from err
+        configdict_from_yamlstring = load_yaml_to_dict(yamlstring=yamlstring)
+        if configdict_from_yamlstring is not None:
+            config_dict.update(configdict_from_yamlstring)
+        config_dict = _add_extra_element_aliases(config_dict)
+        config_dict = _add_colons_to_prefixes_if_needed(config_dict)
+        return config_dict
+
+    if nondefault_configyaml_str:
+        configdict_from_yamlstring = load_yaml_to_dict(
+            yamlstring=nondefault_configyaml_str
+        )
+        if configdict_from_yamlstring is not None:
+            if not configdict_from_yamlstring.get("default_shape_identifier"):
+                configdict_from_yamlstring["default_shape_identifier"] = "default"
+            config_dict.update(configdict_from_yamlstring)
+        config_dict = _add_extra_element_aliases(config_dict)
+        config_dict = _add_colons_to_prefixes_if_needed(config_dict)
+        return config_dict
+
+    try:
+        yamlstring = Path(default_configfile_name).read_text(encoding="utf-8")
+    except FileNotFoundError:
+        yamlstring = default_configyaml_str
+    configdict_from_yamlstring = load_yaml_to_dict(yamlstring=yamlstring)
+    if configdict_from_yamlstring is not None:
+        config_dict.update(configdict_from_yamlstring)
+    config_dict = _add_extra_element_aliases(config_dict)
+    config_dict = _add_colons_to_prefixes_if_needed(config_dict)
+    return config_dict
+
+
+def _add_extra_element_aliases(config_dict):
+    """If extra element aliases are specified, add them to the configuration dict."""
+    extras = config_dict.get("extra_element_aliases")
+    if extras:
+        try:
+            extras = {coerce_concise(str(k).lower()): v for (k, v) in extras.items()}
+        except AttributeError:
+            extras = {}
+        config_dict["element_aliases"].update(extras)
+    return config_dict
+
+
+def _add_colons_to_prefixes_if_needed(config_dict):
+    """Reconstitute config_dict.prefixes, ensuring that each prefix ends in colon."""
+    prefixes = config_dict.get("prefixes")
+    new_prefixes = {}
+    if prefixes:
+        for prefix in prefixes:
+            if not prefix.endswith(":"):
+                new_prefixes[prefix + ":"] = prefixes[prefix]
+            else:
+                new_prefixes[prefix] = prefixes[prefix]
+    config_dict["prefixes"] = new_prefixes
+    return config_dict
+
+
+def _get_shems(shape):
+    """List TAP elements supported by given shape class."""
+    main_shems = list(asdict(shape()))
+    main_shems.remove("state_list")
+    main_shems.remove("shape_warns")
+    main_shems.remove("shape_extras")
+    return main_shems
+
+
+def _get_stems(state):
+    """List TAP elements supported by given statement template class."""
+    main_stems = list(asdict(state()))
+    main_stems.remove("state_warns")
+    main_stems.remove("state_extras")
+    return main_stems
+
+
+def _initialize_config_dict(shapeclass, stateclass):
+    """Initialize config dict with element lists (computed) and placeholder keys."""
+    config_dict = {}
+    ems_dict = {}
+    shems = ems_dict["shape_elements"] = _get_shems(shapeclass)
+    stems = ems_dict["statement_template_elements"] = _get_stems(stateclass)
+    ems_dict["csv_elements"] = shems + stems
+    config_dict["element_aliases"] = {}
+    config_dict["element_aliases"].update(_get_aliases_dict(ems_dict["csv_elements"]))
+    config_dict["default_shape_identifier"] = "default"
+    config_dict["prefixes"] = {}
+    config_dict["extra_shape_elements"] = []
+    config_dict["extra_statement_template_elements"] = []
+    config_dict["picklist_elements"] = []
+    config_dict["picklist_item_separator"] = " "
+    config_dict["extra_value_node_types"] = []
+    config_dict["extra_element_aliases"] = {}
+    config_dict.update(ems_dict)
+    return config_dict
+
+
+def write_configfile(
+    nondefault_configyaml_str=None,
+    default_configfile_name=CONFIGFILE,
+    default_configyaml_str=CONFIGYAML,
+):
+    """Write initial config file or exit trying."""
+    if Path(default_configfile_name).exists():
+        raise ConfigError(f"'{default_configfile_name}' exists - will not overwrite.")
+    if nondefault_configyaml_str:  # useful for testing
+        default_configyaml_str = nondefault_configyaml_str
+    try:
+        with open(default_configfile_name, "w", encoding="utf-8") as outfile:
+            outfile.write(default_configyaml_str)
+            print(f"Settings written to '{default_configfile_name}'.", file=sys.stderr)
+    except FileNotFoundError as error:
+        raise ConfigError(f"'{default_configfile_name}' not writeable.") from error
+
+
+def _get_aliases_dict(csv_elements_list=None):
+    """Short/lowerkey-to-element dict from CSV elements, minus privates."""
+    aliases_to_elements = {}
+    for csv_elem in csv_elements_list:
+        if csv_elem[0] != "_":
+            lowerkey = csv_elem.lower()
+            aliases_to_elements[lowerkey] = csv_elem  # { foobar: fooBar }
+    return aliases_to_elements

dctap/csvreader.py

@@ -0,0 +1,278 @@
+"""Parse TAP, return two-item tuple: (list of shape objects, list of warnings)."""
+
+import re
+from collections import defaultdict
+from csv import DictReader
+from io import StringIO as StringBuffer
+from dataclasses import asdict
+from dctap.exceptions import DctapError, NoDataError
+from dctap.tapclasses import TAPShape, TAPStatementTemplate
+from dctap.utils import coerce_concise
+
+
+def csvreader(
+    csvfile_str=None,
+    config_dict=None,
+    open_csvfile_obj=None,
+    shape_class=TAPShape,
+    state_class=TAPStatementTemplate,
+):
+    """From open CSV file object, return shapes dict."""
+    if csvfile_str:
+        (csvrows, csvwarns) = _get_rows(
+            csvfile_str=csvfile_str,
+            config_dict=config_dict,
+        )
+    elif open_csvfile_obj:
+        (csvrows, csvwarns) = _get_rows(
+            open_csvfile_obj=open_csvfile_obj,
+            config_dict=config_dict,
+        )
+    else:
+        raise DctapError("No data provided.")
+
+    (tapshapes, tapwarns) = _get_tapshapes(
+        rows=csvrows,
+        config_dict=config_dict,
+        shape_class=shape_class,
+        state_class=state_class,
+    )
+    tapwarns = {**csvwarns, **tapwarns}
+    prefixes_used = _get_prefixes_actually_used(csvrows)
+    tapshapes = _add_namespaces(tapshapes, config_dict, prefixes_used)
+    tapshapes = _add_tapwarns(tapshapes, tapwarns)
+    return tapshapes
+
+
+def _add_namespaces(tapshapes=None, config_dict=None, prefixes_used=None):
+    """Adds key 'namespaces' to tapshapes dict."""
+    tapshapes["namespaces"] = {}
+    if config_dict.get("prefixes"):
+        for prefix in prefixes_used:
+            if config_dict["prefixes"].get(prefix):
+                tapshapes["namespaces"][prefix] = config_dict["prefixes"].get(prefix)
+    return tapshapes
+
+
+def _add_tapwarns(tapshapes=None, tapwarns=None):
+    """Adds key 'warnings' to tapshapes dict."""
+    tapshapes["warnings"] = tapwarns
+    return tapshapes
+
+
+def _get_prefixes_actually_used(csvrows):
+    """List strings before colons in values of elements that could take URI prefixes."""
+    prefixes = set()
+    for row in csvrows:
+        for element in [
+            "shapeID",
+            "propertyID",
+            "valueDataType",
+            "valueShape",
+        ]:
+            if row.get(element):
+                prefix_plus_uri_pair = re.match(r"([^:]*):", row.get(element))
+                if prefix_plus_uri_pair:  # if there is at least one
+                    prefix_as_provided = prefix_plus_uri_pair.group(0)
+                    prefixes.add(prefix_as_provided)
+        if row.get("valueConstraint"):
+            pattern = r"\b\w+:"
+            used_in_valueconstraint = re.findall(pattern, row.get("valueConstraint"))
+            prefixes = set(list(prefixes) + list(used_in_valueconstraint))
+    return list(prefixes)
+
+
+def _get_rows(
+    csvfile_str=None,
+    config_dict=None,
+    open_csvfile_obj=None,
+):
+    """Extract from _io.TextIOWrapper object a list of CSV file rows as dicts."""
+    # pylint: disable=too-many-locals
+    # pylint: disable=too-many-branches
+    if csvfile_str:
+        csvfile_contents_str = csvfile_str
+    elif open_csvfile_obj:
+        csvfile_contents_str = open_csvfile_obj.read()
+    else:
+        raise NoDataError("No data to process.")
+
+    tmp_buffer = StringBuffer(csvfile_contents_str)
+    csvlines_stripped = [line.strip() for line in tmp_buffer]
+    csvlines_stripped = [
+        line for line in csvlines_stripped if not re.match("#", line.strip())
+    ]
+    if len(csvlines_stripped) < 2:
+        raise NoDataError("No data to process.")
+    raw_header_line_list = csvlines_stripped[0].split(",")
+    new_header_line_list = []
+
+    recognized_elements = config_dict.get("csv_elements")
+    xtra_shems = config_dict.get("extra_shape_elements")
+    xtra_stems = config_dict.get("extra_statement_template_elements")
+    if xtra_shems:
+        recognized_elements.extend(xtra_shems)
+        for element in xtra_shems:
+            config_dict["element_aliases"][element.lower()] = element
+    if xtra_stems:
+        recognized_elements.extend(xtra_stems)
+        for element in xtra_stems:
+            config_dict["element_aliases"][element.lower()] = element
+    recognized_elements = [elem.lower() for elem in recognized_elements]
+
+    for column in raw_header_line_list:
+        column = coerce_concise(column)
+        column = _normalize_element_name(column, config_dict.get("element_aliases"))
+        new_header_line_list.append(column)
+
+    csv_warns = defaultdict(dict)
+    for column in new_header_line_list:
+        if column.lower() not in recognized_elements:
+            warn = f"Non-DCTAP element '{column}' not configured as extra element."
+            csv_warns["csv"] = {}
+            csv_warns["csv"]["column"] = []
+            csv_warns["csv"]["column"].append(warn)
+
+    new_header_line_str = ",".join(new_header_line_list)
+    csvlines_stripped[0] = new_header_line_str
+    if not csvlines_stripped[0]:
+        raise NoDataError("No data to process.")
+    if "propertyID" not in csvlines_stripped[0]:
+        raise DctapError("Valid DCTAP CSV must have a 'propertyID' column.")
+
+    tmp_buffer2 = StringBuffer("".join([line + "\n" for line in csvlines_stripped]))
+    csv_rows = list(DictReader(tmp_buffer2))
+    for row in csv_rows:
+        for key, value in row.items():
+            if isinstance(value, str):  # ignore if instance of NoneType or list
+                row[key] = value.strip()
+    csv_warns = dict(csv_warns)
+    return (csv_rows, csv_warns)
+
+
+def _get_tapshapes(rows=None, config_dict=None, shape_class=None, state_class=None):
+    """Return tuple: (shapes dict, warnings dict)."""
+    # pylint: disable=too-many-locals
+    # pylint: disable=too-many-branches
+    # pylint: disable=too-many-statements
+
+    default_shape_id = config_dict["default_shape_identifier"]
+    main_stems = config_dict.get("statement_template_elements")
+    xtra_stems = config_dict.get("extra_statement_template_elements")
+    shapes = {}  # dict for shapeID-to-TAPShape_list
+    warns = defaultdict(dict)  # dict for shapeID-to-warnings_list
+
+    for row in rows:
+        shape_id = ""
+        if row.get("propertyID"):
+            if row.get("shapeID"):
+                shape_id = row.get("shapeID")
+            elif not row.get("shapeID"):
+                try:
+                    shape_id = list(shapes)[-1]
+                except IndexError:
+                    shape_id = row["shapeID"] = default_shape_id
+        elif row.get("shapeID"):
+            shape_id = row.get("shapeID")
+
+        if shape_id:
+            if shape_id not in list(shapes):
+                shape_obj = _make_shape(
+                    row_dict=row,
+                    config_dict=config_dict,
+                    shape_class=shape_class,
+                )
+                shape_obj.normalize(config_dict)
+                shapes[shape_id] = shape_obj
+                warns[shape_id] = {}
+
+            shape_warnings = shape_obj.get_warnings()
+            for (elem, warn) in shape_warnings.items():
+                try:
+                    warns[shape_id][elem].append(warn)
+                except KeyError:
+                    warns[shape_id][elem] = []
+                    warns[shape_id][elem].append(warn)
+
+        if not row.get("propertyID"):
+            continue
+
+        state_class_obj = state_class()
+        for col in row:
+            if col in main_stems:
+                setattr(state_class_obj, col, row[col])
+            elif col in xtra_stems:
+                state_class_obj.state_extras[col] = row[col]
+
+        state_class_obj.normalize(config_dict)
+        shapes[shape_id].state_list.append(state_class_obj)
+
+        warns_dict = dict(warns)
+        shapes_dict = {}
+        shapes_dict["shapes"] = []
+
+        for shape_obj in list(shapes.values()):
+            sh_dict = asdict(shape_obj)
+            sh_dict["statement_templates"] = sh_dict.pop("state_list")
+            shapes_dict["shapes"].append(sh_dict)
+
+        shapes_dict = _simplify(shapes_dict)
+
+    return (shapes_dict, warns_dict)
+
+
+def _make_shape(row_dict=None, config_dict=None, shape_class=None):
+    """Populates shape fields of dataclass shape object from dict for one row.
+
+    Args:
+        row_dict: Dictionary of all columns headers (keys) and cell values (values)
+            found in a given row, with no distinction between shape elements and
+            statement template elements.
+        config_dict: Dictionary of settings, built-in or as read from config file.
+
+    Returns:
+        Unpopulated instance of shape class, for example:
+        TAPShape(shapeID='', state_list=[], shape_warns={}, state_extras={}, ...)
+    """
+    main_shems = config_dict.get("shape_elements")
+    xtra_shems = config_dict.get("extra_shape_elements")
+    tapshape_obj = shape_class()
+    for key in row_dict:
+        if key in main_shems:
+            setattr(tapshape_obj, key, row_dict[key])
+        elif key in xtra_shems:
+            tapshape_obj.shape_extras[key] = row_dict[key]
+    return tapshape_obj
+
+
+def _normalize_element_name(some_str, element_aliases_dict=None):
+    """Given header string, return converted if aliased, else return unchanged."""
+    some_str = coerce_concise(some_str)
+    if element_aliases_dict:
+        for key in element_aliases_dict.keys():
+            if key == some_str:
+                some_str = element_aliases_dict[key]
+    return some_str
+
+
+def _simplify(shapes_dict):
+    """Remove elements from shapes dictionary with falsy values."""
+    for shape in shapes_dict["shapes"]:
+        for state in shape["statement_templates"]:
+            if state.get("state_extras"):
+                for (k, v) in state["state_extras"].items():
+                    state[k] = v
+                del state["state_extras"]
+            if state.get("state_warns"):
+                del state["state_warns"]
+            for empty_element in [key for key in state if not state[key]]:
+                del state[empty_element]
+        if shape.get("shape_extras"):
+            for (k, v) in shape["shape_extras"].items():
+                shape[k] = v
+            del shape["shape_extras"]
+        if shape.get("shape_warns"):
+            del shape["shape_warns"]
+        for empty_element in [key for key in shape if not shape[key]]:
+            del shape[empty_element]
+    return shapes_dict

dctap/defaults.py

@@ -0,0 +1,22 @@
+"""Default settings."""
+
+CONFIGFILE = "dctap.yaml"
+CONFIGYAML = """\
+# dctap configuration file (in YAML format)
+# See https://dctap-python.readthedocs.io/en/latest/config/ for more options
+
+prefixes:
+    ":":        "http://example.org/"
+    "dc:":      "http://purl.org/dc/elements/1.1/"
+    "dcterms:": "http://purl.org/dc/terms/"
+    "dct:":     "http://purl.org/dc/terms/"
+    "foaf:":    "http://xmlns.com/foaf/0.1/"
+    "owl:":     "http://www.w3.org/2002/07/owl#"
+    "rdf:":     "http://www.w3.org/1999/02/22-rdf-syntax-ns#"
+    "rdfs:":    "http://www.w3.org/2000/01/rdf-schema#"
+    "schema:":  "http://schema.org/"
+    "skos:":    "http://www.w3.org/2004/02/skos/core#"
+    "skosxl:":  "http://www.w3.org/2008/05/skos-xl#"
+    "wdt:":     "http://www.wikidata.org/prop/direct/"
+    "xsd:":     "http://www.w3.org/2001/XMLSchema#"
+"""

dctap/exceptions.py

@@ -0,0 +1,13 @@
+"""Exception classes for dctap."""
+
+
+class DctapError(SystemExit):
+    """Exceptions related to Dctap generally."""
+
+
+class ConfigError(DctapError):
+    """Exceptions related to configuration."""
+
+
+class NoDataError(DctapError):
+    """Exception raised if there is no (TAP) data to process."""

dctap/inspect.py

@@ -0,0 +1,57 @@
+"""Print CSV contents and warnings."""
+
+import sys
+from dctap.loggers import stderr_logger
+
+
+def pprint_tapshapes(tapshapes_dict=None, config_dict=None):
+    """Pretty-print TAPShape objects to output list, ready for printing to console."""
+    # pylint: disable=too-many-branches
+    main_shems = config_dict.get("shape_elements")
+    xtra_shems = config_dict.get("extra_shape_elements")
+    main_stems = config_dict.get("statement_template_elements")
+    xtra_stems = config_dict.get("extra_statement_template_elements")
+    pprint_output = []
+    pprint_output.append("Tabular Application Profile (TAP)")
+    for tapshape_dict in tapshapes_dict.get("shapes"):
+        pprint_output.append("    Shape")
+        for key in main_shems:
+            indent08 = 8 * " " + key + " "
+            while len(indent08) < 33:
+                indent08 += " "
+            if tapshape_dict.get(key):
+                pprint_output.append(indent08 + str(tapshape_dict.get(key)))
+        for key in xtra_shems:
+            indent08 = 8 * " " + "[" + key + "] "
+            while len(indent08) < 33:
+                indent08 += " "
+            if tapshape_dict.get(key):
+                pprint_output.append(indent08 + str(tapshape_dict.get(key)))
+
+        for sc_dict in tapshape_dict.get("statement_templates"):
+            pprint_output.append("        Statement Template")
+            for key in main_stems:
+                if sc_dict.get(key):
+                    indent12 = 12 * " " + key + " "
+                    while len(indent12) < 33:
+                        indent12 += " "
+                    pprint_output.append(indent12 + str(sc_dict.get(key)))
+            for key in xtra_stems:
+                indent08 = 12 * " " + "[" + key + "] "
+                while len(indent08) < 33:
+                    indent08 += " "
+                if sc_dict.get(key):
+                    pprint_output.append(indent08 + str(sc_dict.get(key)))
+
+    return pprint_output
+
+
+def print_warnings(warnings_dict):
+    """Print warnings to stdout."""
+    # pylint: disable=logging-fstring-interpolation
+    print("", file=sys.stderr)
+    echo = stderr_logger()
+    for (shapeid, warns) in warnings_dict.items():
+        for (elem, warn_list) in warns.items():
+            for warning in warn_list:
+                echo.warning(f"[{shapeid}/{elem}] {warning}")

dctap/loggers.py

@@ -0,0 +1,43 @@
+"""Loggers: warnings to stderr, warnings to file, verbose debug info to file."""
+
+import logging
+import sys
+
+PLAIN_FORMATTER = logging.Formatter("%(levelname)s %(message)s")
+
+TIMESTAMP_FORMATTER = logging.Formatter(
+    "%(asctime)s %(levelname)s %(message)s", "%Y-%m-%d %H:%M"
+)
+
+
+def stderr_logger():
+    """Initialize logger for printing messages, INFO and higher, to standard error."""
+    stderr_info_logger = logging.getLogger("stderr_logger")
+    stderr_info_logger.setLevel(logging.INFO)
+    stderr_info_logger_handler = logging.StreamHandler(sys.stderr)
+    stderr_info_logger_handler.setLevel(logging.INFO)
+    stderr_info_logger.addHandler(stderr_info_logger_handler)
+    stderr_info_logger_handler.setFormatter(PLAIN_FORMATTER)
+    return stderr_info_logger
+
+
+def warning_logger(output_filename="warnings.log"):
+    """Logger for printing time-stamped messages, WARNING and higher, to file."""
+    warningfile_logger = logging.getLogger("warningfile_logger")
+    warningfile_logger.setLevel(logging.INFO)
+    warningfile_logger_handler = logging.FileHandler(output_filename, mode="w")
+    warningfile_logger_handler.setLevel(logging.WARNING)
+    warningfile_logger.addHandler(warningfile_logger_handler)
+    warningfile_logger_handler.setFormatter(TIMESTAMP_FORMATTER)
+    return warningfile_logger
+
+
+def debug_logger(output_filename="debug.log"):
+    """Logger for printing time-stamped messages, DEBUG and higher, to file."""
+    debugfile_logger = logging.getLogger("debugfile_logger")
+    debugfile_logger.setLevel(logging.DEBUG)
+    debugfile_logger_handler = logging.FileHandler(output_filename, mode="w")
+    debugfile_logger_handler.setLevel(logging.DEBUG)
+    debugfile_logger.addHandler(debugfile_logger_handler)
+    debugfile_logger_handler.setFormatter(TIMESTAMP_FORMATTER)
+    return debug_logger

dctap/tapclasses.py

@@ -0,0 +1,298 @@
+"""Classes for Python objects derived from CSV files."""
+
+import re
+from dataclasses import dataclass, field
+from typing import List
+from dctap.utils import coerce_integer, coerce_numeric, looks_like_uri_or_curie
+
+
+@dataclass
+class TAPStatementTemplate:
+    """Instances hold TAP/CSV elements related to statement templates."""
+
+    # pylint: disable=too-many-instance-attributes # It's a dataclass, right?
+    # pylint: disable=invalid-name # for elements not named in snake case.
+
+    propertyID: str = ""
+    propertyLabel: str = ""
+    mandatory: str = ""
+    repeatable: str = ""
+    valueNodeType: str = ""
+    valueDataType: str = ""
+    valueConstraint: str = ""
+    valueConstraintType: str = ""
+    valueShape: str = ""
+    note: str = ""
+    state_warns: dict = field(default_factory=dict)
+    state_extras: dict = field(default_factory=dict)
+
+    def normalize(self, config_dict):
+        """Normalizes specific fields."""
+        # pylint: disable=attribute-defined-outside-init
+        self._normalize_booleans()
+        self._valueConstraintType_pattern_warn_if_valueConstraint_not_valid_regex()
+        self._valueConstraintType_pattern_warn_if_used_with_value_shape()
+        self._valueConstraintType_iristem_parse()
+        self._valueConstraintType_iristem_warn_if_list_items_not_IRIs()
+        self._valueConstraintType_languageTag_parse(config_dict)
+        self._valueConstraintType_minmaxlength_warn_if_not_nonnegative_integer()
+        self._valueConstraintType_minmaxinclusive_parse()
+        self._valueConstraintType_minmaxinclusive_warn_if_value_not_numeric()
+        self._valueConstraintType_warn_if_used_without_valueConstraint()
+        self._valueDataType_warn_if_used_with_valueNodeType_IRI()
+        self._valueNodeType_warn_if_valueNodeType_literal_used_with_any_valueShape()
+        self._valueConstraintType_picklist_parse(config_dict)
+        self._valueNodeType_is_from_enumerated_list(config_dict)
+        self._parse_elements_configured_as_picklist_elements(config_dict)
+        return self
+
+    def _warn_if_value_not_urilike(self):
+        """Warns when values of given elements do not look like URIs."""
+        elements_that_may_take_uris = ["propertyID", "valueDataType", "valueShape"]
+        for elem in elements_that_may_take_uris:
+            value = getattr(self, elem)
+            warning = f"Value '{value}' does not look like a URI."
+            if value:
+                if not looks_like_uri_or_curie(value):
+                    self.state_warns[elem] = warning
+        return self
+
+    def _normalize_booleans(self):
+        """Coerces supported Boolean values to 'true' or 'false' or leaves unchanged."""
+        valid_values_for_true = ["true", "TRUE", "True", "1"]
+        valid_values_for_false = ["false", "FALSE", "False", "0"]
+        boolean_elements = ["mandatory", "repeatable"]
+        for elem in boolean_elements:
+            value = getattr(self, elem)
+            if value:
+                warning_message = f"'{value}' is not a supported Boolean value."
+                if value in valid_values_for_true:
+                    setattr(self, elem, "true")
+                elif value in valid_values_for_false:
+                    setattr(self, elem, "false")
+                else:
+                    self.state_warns[elem] = warning_message
+        return self
+
+    def _valueConstraintType_iristem_parse(self):
+        """If valueConstraintType is Iristem, split valueConstraint on whitespace."""
+        self.valueConstraintType = self.valueConstraintType.lower()
+        if self.valueConstraintType == "iristem":
+            if self.valueConstraint:
+                self.valueConstraint = self.valueConstraint.split()
+        return self
+
+    def _valueConstraintType_iristem_warn_if_list_items_not_IRIs(self):
+        """If IRIStem, warn if valueConstraint list items do not look like IRIs."""
+        self.valueConstraintType = self.valueConstraintType.lower()
+        if self.valueConstraintType == "iristem":
+            for list_item in self.valueConstraint:
+                if not looks_like_uri_or_curie(list_item):
+                    self.state_warns["valueConstraint"] = (
+                        f"Value constraint type is '{self.valueConstraintType}', "
+                        f"but '{list_item}' does not look like an IRI or "
+                        "Compact IRI."
+                    )
+        return self
+
+    def _valueConstraintType_minmaxlength_warn_if_not_nonnegative_integer(self):
+        """
+        Tries to coerce valueConstraint to integer (or leaves string untouched).
+        Warns if valueConstraint for minLength is not a nonnegative integer.
+        """
+        vctype = self.valueConstraintType.lower()
+        vc = self.valueConstraint = coerce_integer(self.valueConstraint)
+        bad_vc_warning = (
+            f"Value constraint type is '{self.valueConstraintType}', "
+            f"but '{self.valueConstraint}' is not a positive integer."
+        )
+        if vctype in ("minlength", "maxlength"):
+            if isinstance(vc, int):
+                if vc < 0:
+                    self.state_warns["valueConstraint"] = bad_vc_warning
+            elif not isinstance(vc, int):
+                self.state_warns["valueConstraint"] = bad_vc_warning
+
+        return self
+
+    def _valueConstraintType_minmaxinclusive_parse(self):
+        """
+        If value of valueConstraintType is 'minInclusive' or 'maxInclusive',
+        value of valueConstraint should be numeric (int or float).
+        """
+        self.valueConstraintType = self.valueConstraintType.lower()
+        value_constraint = self.valueConstraint
+        if self.valueConstraintType in ("mininclusive", "maxinclusive"):
+            if value_constraint:
+                self.valueConstraint = coerce_numeric(value_constraint)
+        return self
+
+    def _valueConstraintType_minmaxinclusive_warn_if_value_not_numeric(self):
+        """Warns if valueConstraint for minInclusive not coercable to float."""
+        self.valueConstraintType = self.valueConstraintType.lower()
+        if self.valueConstraintType in ("mininclusive", "maxinclusive"):
+            try:
+                float(self.valueConstraint)
+            except (ValueError, TypeError):
+                self.state_warns["valueConstraint"] = (
+                    f"Value constraint type is '{self.valueConstraintType}', "
+                    f"but '{self.valueConstraint}' is not numeric."
+                )
+        return self
+
+    def _valueConstraintType_pattern_warn_if_valueConstraint_not_valid_regex(self):
+        """If valueConstraintType Pattern, warn if valueConstraint not valid regex."""
+        self.valueConstraintType = self.valueConstraintType.lower()
+        if self.valueConstraintType == "pattern":
+            try:
+                re.compile(self.valueConstraint)
+            except (re.error, TypeError):
+                self.state_warns["valueConstraint"] = (
+                    f"Value constraint type is '{self.valueConstraintType}', but "
+                    f"'{self.valueConstraint}' is not a valid regular expression."
+                )
+        return self
+
+    def _valueConstraintType_pattern_warn_if_used_with_value_shape(self):
+        """Regular expressions cannot conform to value shapes."""
+        self.valueConstraintType = self.valueConstraintType.lower()
+        if self.valueConstraintType == "pattern":
+            if self.valueShape:
+                self.state_warns["valueConstraintType"] = (
+                    f"Values of constraint type '{self.valueConstraintType}' "
+                    "cannot conform to a value shape."
+                )
+
+    def _valueConstraintType_languageTag_parse(self, config_dict):
+        """For valueConstraintType languageTag, splits valueConstraint on whitespace."""
+        self.valueConstraintType = self.valueConstraintType.lower()
+        sep = config_dict.get("picklist_item_separator", " ")
+        if self.valueConstraintType == "languagetag":
+            if self.valueConstraint:
+                self.valueConstraint = self.valueConstraint.split(sep)
+                self.valueConstraint = [x.strip() for x in self.valueConstraint if x]
+        return self
+
+    def _valueConstraintType_warn_if_used_without_valueConstraint(self):
+        """Warns if valueConstraintType used without valueConstraint."""
+        if self.valueConstraintType:
+            if not self.valueConstraint:
+                self.state_warns["valueConstraint"] = (
+                    f"Value constraint type '{self.valueConstraintType}' "
+                    "has no corresponding value constraint."
+                )
+        return self
+
+    def _valueConstraintType_picklist_parse(self, config_dict):
+        """If valueConstraintType is Picklist, split valueConstraint on whitespace."""
+        self.valueConstraintType = self.valueConstraintType.lower()
+        sep = config_dict.get("picklist_item_separator", " ")
+        if self.valueConstraintType == "picklist":
+            if self.valueConstraint:
+                self.valueConstraint = self.valueConstraint.split(sep)
+                self.valueConstraint = [x.strip() for x in self.valueConstraint if x]
+        return self
+
+    def _valueNodeType_is_from_enumerated_list(self, config_dict):
+        """Take valueNodeType from configurable enumerated list, case-insensitive."""
+        warning = f"'{self.valueNodeType}' is not a valid node type."
+        valid_types = ["iri", "bnode", "literal"]
+        # This should be moved out to defaults dictionary.
+        if config_dict.get("extra_value_node_types"):
+            valid_types += [v.lower() for v in config_dict["extra_value_node_types"]]
+        if self.valueNodeType:
+            self.valueNodeType = self.valueNodeType.lower()
+            if self.valueNodeType not in valid_types:
+                self.state_warns["valueNodeType"] = warning
+        return self
+
+    def _valueNodeType_warn_if_valueNodeType_literal_used_with_any_valueShape(self):
+        """Value with node type Literal cannot conform to a value shape."""
+        warning = "Values of node type 'literal' cannot conform to value shapes."
+        self.valueNodeType = self.valueNodeType.lower()
+        if self.valueShape:
+            if self.valueNodeType == "literal":
+                self.state_warns["valueDataType"] = warning
+        return self
+
+    def _valueDataType_warn_if_used_with_valueShape(self):
+        """Value with any datatype cannot conform to a value shape."""
+        warning = "Values with datatypes (literals) cannot conform to value shapes."
+        if self.valueShape:
+            if self.valueDataType:
+                self.state_warns["valueDataType"] = warning
+        return self
+
+    def _valueDataType_warn_if_used_with_valueNodeType_IRI(self):
+        """Value with datatype implies Literal and cannot be node type IRI."""
+        node_type = self.valueNodeType
+        data_type = self.valueDataType
+        warning = f"Datatype '{data_type}' incompatible with node type '{node_type}'."
+        node_type = self.valueNodeType.lower()
+        if node_type in ("iri", "uri", "bnode"):
+            if self.valueDataType:
+                self.state_warns["valueDataType"] = warning
+        return self
+
+    def _parse_elements_configured_as_picklist_elements(self, config_dict):
+        """Parse elements configured as list elementss."""
+        if config_dict.get("picklist_item_separator"):
+            separator = config_dict.get("picklist_item_separator")
+        else:
+            separator = " "
+
+        if config_dict.get("picklist_elements"):
+            picklist_elements = config_dict.get("picklist_elements")
+        else:
+            picklist_elements = []
+
+        for element in picklist_elements:
+            if getattr(self, element):
+                setattr(self, element, getattr(self, element).split(separator))
+
+        return self
+
+    def get_warnings(self):
+        """Emit self.state_warns as populated by self.normalize()."""
+        return dict(self.state_warns)
+
+
+@dataclass
+class TAPShape:
+    """An instance holds TAP/CSV row elements related to one given, named shape."""
+
+    # pylint: disable=invalid-name
+    # True that propertyID, etc, do not conform to snake-case naming style.
+
+    shapeID: str = ""
+    shapeLabel: str = ""
+    state_list: List[TAPStatementTemplate] = field(default_factory=list)
+    shape_warns: dict = field(default_factory=dict)
+    shape_extras: dict = field(default_factory=dict)
+
+    def normalize(self, config_dict):
+        """Normalize values where required."""
+        self._normalize_default_shapeID(config_dict)
+        self._warn_if_value_not_urilike()
+        return self
+
+    def _normalize_default_shapeID(self, config_dict):
+        """If shapeID not specified, looks first in config, else sets "default"."""
+        if not self.shapeID:
+            self.shapeID = config_dict.get("default_shape_identifier", "default")
+        return self
+
+    def _warn_if_value_not_urilike(self):
+        """Warns when values of given elements do not look like URIs."""
+        elements_that_may_take_uris = ["shapeID"]
+        for elem in elements_that_may_take_uris:
+            value = getattr(self, elem)
+            warning = f"Value '{value}' does not look like a URI."
+            if value:
+                if not looks_like_uri_or_curie(value):
+                    self.shape_warns[elem] = warning
+        return self
+
+    def get_warnings(self):
+        """Emit warnings dictionary self.shape_warns, populated by normalize() method."""
+        return dict(self.shape_warns)

dctap/utils.py

@@ -0,0 +1,113 @@
+"""Utilities."""
+
+import sys
+import re
+from pathlib import Path
+from urllib.parse import urlparse
+from ruamel.yaml import YAML, YAMLError
+from ruamel.yaml.scanner import ScannerError
+from ruamel.yaml.parser import ParserError
+from dctap.exceptions import ConfigError, DctapError
+
+
+def load_yaml_to_dict(yamlstring=None, yamlfile=None):
+    """Convert YAML from string or file (filename or Path object) into Python dict."""
+    dict_from_yamlstring = {}
+    if yamlfile and yamlstring:
+        raise DctapError("Can load YAML from string or file, but not both.")
+
+    if yamlfile:
+        try:
+            yamlstring = Path(yamlfile).read_text(encoding="UTF-8")
+        except FileNotFoundError:
+            print(f"File '{yamlfile}' not found.", file=sys.stderr)
+
+    if yamlstring is not None:
+        yaml = YAML(typ="safe", pure=True)
+        try:
+            dict_from_yamlstring = yaml.load(yamlstring)
+        except (YAMLError, ScannerError, ParserError):
+            dict_from_yamlstring = None
+            if yamlfile:
+                print(f"YAML in '{yamlfile}' is badly formed.", file=sys.stderr)
+            else:
+                print("YAML is badly formed.", file=sys.stderr)
+
+    return dict_from_yamlstring
+
+
+def coerce_concise(some_str=None):
+    """
+    For given string:
+    - delete spaces, underscores, dashes, commas
+    - lowercase
+    - delete surrounding single and double quotes
+    """
+    some_str = some_str.replace(" ", "")
+    some_str = some_str.replace("_", "")
+    some_str = some_str.replace("-", "")
+    some_str = some_str.replace(",", "")
+    some_str = some_str.lower()
+    some_str = some_str.strip('"')
+    some_str = some_str.strip("'")
+    return some_str
+
+
+def coerce_integer(value_constraint=None):
+    """Coerces string to integer or returns string untouched."""
+    try:
+        value_constraint = int(value_constraint)
+    except (ValueError, TypeError):
+        pass  # pass the valueConstraint through untouched
+    return value_constraint
+
+
+def coerce_numeric(value_constraint=None):
+    """Coerces string to numeric type or returns string untouched."""
+    try:
+        if value_constraint == str(float(value_constraint)):
+            value_constraint = float(value_constraint)
+        else:
+            value_constraint = int(value_constraint)
+    except (ValueError, TypeError):
+        pass  # pass the valueConstraint through untouched
+    return value_constraint
+
+
+def expand_uri_prefixes(shapes_dict=None, config_dict=None):
+    """Expand namespace prefixes, eg: dc:date to http://purl.org/dc/terms/date."""
+    # pylint: disable=too-many-nested-blocks
+    if not config_dict.get("prefixes"):
+        raise ConfigError("No 'prefixes' section found in config file.")
+    for shape in shapes_dict["shapes"]:
+        for prefix in config_dict["prefixes"]:
+            if re.match(prefix, shape["shapeID"]):
+                prefix_expanded = config_dict["prefixes"][prefix]
+                shape["shapeID"] = re.sub(prefix, prefix_expanded, shape["shapeID"])
+        for sc in shape["statement_templates"]:
+            for element in ["propertyID", "valueDataType", "valueShape"]:
+                if sc.get(element):
+                    for prefix in config_dict["prefixes"]:
+                        if re.match(prefix, sc.get(element)):
+                            prefix_expanded = config_dict["prefixes"][prefix]
+                            sc[element] = re.sub(prefix, prefix_expanded, sc[element])
+    return shapes_dict
+
+
+def looks_like_uri_or_curie(url_string):
+    """True if string superficially looks like a URI or Compact URI."""
+    if not isinstance(url_string, str):
+        return False
+
+    url_parsed = urlparse(url_string)
+    has_prefix = bool(url_parsed.scheme)  # could be URI scheme or CURIE prefix
+    has_net_location = bool(url_parsed.netloc)  # something with a dot
+    has_name = bool(re.match(r"^:", url_parsed.path))  # starts with a colon
+
+    if has_prefix and has_net_location:
+        return True
+    if has_prefix:
+        return True
+    if has_name:
+        return True
+    return False