ods_to_rdf

ods_to_rdf

Generate compass.ttl and vocab.ttl from a use case's source-data.ods.

COMPASS_USE_CASE (default oceancare) selects the directory under src/ontology/. Pins link to other rows by id in their links column; the predicate a link becomes depends on the dimension or class of its target. Concepts carry no links.

Subject order, predicate order and float precision are fixed, so unchanged input produces byte-identical output.

SheetError

Bases: Exception

A defect in the workbook, to be fixed there.

Row(number, sheet, cells) dataclass

One non-empty data row of a sheet.

Problems(items=list()) dataclass

Collect workbook defects so one run reports all of them.

Fallbacks(counts=Counter()) dataclass

Count the empty German cells that took the English text.

_resolve_use_case()

Return COMPASS_USE_CASE from the environment, else the repo .env.

Source code in src/turtle-generator/ods_to_rdf.py

def _resolve_use_case() -> str:
    """Return ``COMPASS_USE_CASE`` from the environment, else the repo ``.env``."""
    value = os.environ.get("COMPASS_USE_CASE", "").strip()
    if value:
        return value
    env_file = REPO / ".env"
    if env_file.is_file():
        for line in env_file.read_text(encoding="utf-8").splitlines():
            if line.strip().startswith("COMPASS_USE_CASE="):
                value = line.split("=", 1)[1].split(" #", 1)[0].strip().strip("\"'")
                if value:
                    return value
    return "oceancare"

repo_relative(path)

Return path relative to the repo root, or absolute if it lies outside.

Source code in src/turtle-generator/ods_to_rdf.py

def repo_relative(path: Path) -> str:
    """Return ``path`` relative to the repo root, or absolute if it lies outside."""
    try:
        return str(path.relative_to(REPO))
    except ValueError:
        return str(path)

_cell_text(cell)

Return a cell's stored number if it has one, else its displayed text.

Source code in src/turtle-generator/ods_to_rdf.py

def _cell_text(cell) -> str:
    """Return a cell's stored number if it has one, else its displayed text."""
    if cell.getAttribute("valuetype") == "float":
        stored = cell.getAttribute("value")
        if stored is not None:
            return stored.removesuffix(".0")
    # str() drops the <text:s/> elements that pack runs of spaces; teletype expands them.
    return "\n".join(teletype.extractText(p) for p in cell.getElementsByType(P)).strip()

_row_values(row, width)

Return the first width cell texts of a row, expanding repeated cells.

Source code in src/turtle-generator/ods_to_rdf.py

def _row_values(row, width: int) -> list[str]:
    """Return the first ``width`` cell texts of a row, expanding repeated cells."""
    values: list[str] = []
    for cell in row.getElementsByType(TableCell):
        if len(values) >= width:
            break
        repeat = int(cell.getAttribute("numbercolumnsrepeated") or 1)
        values.extend([_cell_text(cell)] * min(repeat, width - len(values)))
    return values + [""] * (width - len(values))

read_sheet(name, columns)

Read the non-empty rows of a workbook sheet whose header must equal columns.

Raises:

Type	Description
SheetError	The workbook or sheet is missing or empty, or the header differs.

Source code in src/turtle-generator/ods_to_rdf.py

def read_sheet(name: str, columns: list[str]) -> list[Row]:
    """Read the non-empty rows of a workbook sheet whose header must equal ``columns``.

    Raises:
        SheetError: The workbook or sheet is missing or empty, or the header differs.
    """
    sheet_rows = _sheet(name).getElementsByType(TableRow)
    if not sheet_rows:
        raise SheetError(f"sheet {name!r} is empty")

    # Read past the expected width so that extra columns are reported.
    header = [h for h in _row_values(sheet_rows[0], len(columns) + 8) if h]
    if header != columns:
        missing = [c for c in columns if c not in header]
        extra = [c for c in header if c not in columns]
        parts = []
        if missing:
            parts.append(f"missing {missing}")
        if extra:
            parts.append(f"unexpected {extra}")
        detail = ", ".join(parts) or "columns are in the wrong order"
        raise SheetError(f"sheet {name!r}: {detail}")

    rows = []
    number = 1
    for sheet_row in sheet_rows[1:]:
        repeat = int(sheet_row.getAttribute("numberrowsrepeated") or 1)
        values = _row_values(sheet_row, len(columns))
        if any(values):
            rows += [
                Row(number + offset, name, dict(zip(columns, values, strict=True)))
                for offset in range(1, repeat + 1)
            ]
        number += repeat
    return rows

index_terms(concepts, pins, problems)

Map every concept and pin id to its dimension or class.

Source code in src/turtle-generator/ods_to_rdf.py

def index_terms(concepts: list[Row], pins: list[Row], problems: Problems) -> dict[str, str]:
    """Map every concept and pin id to its dimension or class."""
    kinds: dict[str, str] = {}
    seen: dict[str, Row] = {}
    for rows, column, allowed in (
        (concepts, "dimension", DIMENSIONS),
        (pins, "class", CLASSES),
    ):
        for row in rows:
            identifier = row["id"]
            if not identifier:
                problems.add(row, "row has no id")
            elif identifier in seen:
                first = seen[identifier]
                problems.add(
                    row,
                    f"id {identifier!r} is already used by {first.sheet}:{first.number}",
                )
            elif row[column] not in allowed:
                problems.add(
                    row, f"{column} {row[column]!r} is not one of {sorted(allowed)}"
                )
            else:
                seen[identifier] = row
                kinds[identifier] = row[column]
    return kinds

parse_links(row, kinds, problems)

Group the targets in a pin's links cell by the predicate each becomes.

Source code in src/turtle-generator/ods_to_rdf.py

def parse_links(row: Row, kinds: dict[str, str], problems: Problems) -> dict[str, set[str]]:
    """Group the targets in a pin's ``links`` cell by the predicate each becomes."""
    grouped: dict[str, set[str]] = {}
    for target in (part.strip() for part in row["links"].split(",")):
        if not target:
            continue
        if target == row["id"]:
            problems.add(row, f"{target!r} links to itself")
        elif target not in kinds:
            problems.add(
                row, f"link to unknown id {target!r} -- check the spelling, or add the row"
            )
        else:
            prefix = "compass:" if kinds[target] in DIMENSIONS else "ocinst:"
            grouped.setdefault(LINK_PREDICATE[kinds[target]], set()).add(prefix + target)
    return grouped

numeric_cell(row, column, problems, parse)

Return the cell text, or "" after reporting a value parse rejects.

Source code in src/turtle-generator/ods_to_rdf.py

def numeric_cell(row: Row, column: str, problems: Problems, parse: type) -> str:
    """Return the cell text, or "" after reporting a value ``parse`` rejects."""
    value = row[column]
    if not value:
        return ""
    try:
        parse(value)
    except ValueError:
        problems.add(row, f"{column} {value!r} is not a number")
        return ""
    return value

bilingual(row, column, fallbacks)

Return the English and German text of <column>_en / <column>_de.

An empty German cell takes the English text and is noted in fallbacks. Returns None when the English cell is empty.

Source code in src/turtle-generator/ods_to_rdf.py

def bilingual(row: Row, column: str, fallbacks: Fallbacks) -> tuple[str, str] | None:
    """Return the English and German text of ``<column>_en`` / ``<column>_de``.

    An empty German cell takes the English text and is noted in ``fallbacks``.
    Returns None when the English cell is empty.
    """
    english = row[f"{column}_en"]
    if not english:
        return None
    german = row[f"{column}_de"]
    if not german:
        fallbacks.note(row.sheet, f"{column}_de")
        german = english
    return english, german

order_key(triple)

Sort key: predicate, then language, then value.

Source code in src/turtle-generator/ods_to_rdf.py

def order_key(triple: tuple[str, str]) -> tuple[int, int, str]:
    """Sort key: predicate, then language, then value."""
    predicate, value = triple
    suffix = value[-3:]
    language = (
        LANGUAGE_ORDER.index(suffix) if suffix in LANGUAGE_ORDER else len(LANGUAGE_ORDER)
    )
    return PREDICATE_ORDER.index(predicate), language, value

validate(data, vocab)

Raise SheetError unless the generated Turtle conforms to shapes.ttl.

Source code in src/turtle-generator/ods_to_rdf.py

def validate(data: str, vocab: str) -> None:
    """Raise SheetError unless the generated Turtle conforms to shapes.ttl."""
    shapes = Graph().parse(SHAPES, format="turtle")
    graph = Graph()
    graph.parse(data=data, format="turtle")
    graph.parse(data=vocab, format="turtle")
    graph.parse(SHAPES, format="turtle")
    conforms, _, report = pyshacl.validate(
        graph, shacl_graph=shapes, inference="rdfs", abort_on_first=False
    )
    if not conforms:
        raise SheetError(f"SHACL validation failed:\n{report}")

generate(fallbacks)

Return the text of compass.ttl and vocab.ttl, noting German fallbacks.

Raises:

Type	Description
SheetError	The workbook has defects; the message lists all of them.

Source code in src/turtle-generator/ods_to_rdf.py

def generate(fallbacks: Fallbacks) -> tuple[str, str]:
    """Return the text of compass.ttl and vocab.ttl, noting German fallbacks.

    Raises:
        SheetError: The workbook has defects; the message lists all of them.
    """
    schemes = read_sheet(SCHEME_SHEET, SCHEME_COLUMNS)
    concepts = read_sheet(CONCEPT_SHEET, CONCEPT_COLUMNS)
    pins = read_sheet(PIN_SHEET, PIN_COLUMNS)

    problems = Problems()
    kinds = index_terms(concepts, pins, problems)
    problems.raise_if_any()  # ids must be sound before links can be checked

    vocab = build_vocab(schemes, concepts, problems, fallbacks)
    data = build_data(pins, kinds, problems, fallbacks)
    problems.raise_if_any()
    return data, vocab

main(argv=None)

Run the command line; return the exit status.

Source code in src/turtle-generator/ods_to_rdf.py

def main(argv: list[str] | None = None) -> int:
    """Run the command line; return the exit status."""
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--check",
        action="store_true",
        help="verify the committed files match a fresh run; write nothing",
    )
    parser.add_argument(
        "--skip-validation",
        action="store_true",
        help="skip SHACL validation (for inspecting output that does not yet conform)",
    )
    args = parser.parse_args(argv)

    fallbacks = Fallbacks()
    try:
        data, vocab = generate(fallbacks)
        if not args.skip_validation:
            validate(data, vocab)
    except SheetError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1

    if args.check:
        stale = [
            repo_relative(path)
            for path, fresh in ((OUT_DATA, data), (OUT_VOCAB, vocab))
            if not path.exists() or path.read_text(encoding="utf-8") != fresh
        ]
        if stale:
            print(
                f"error: {', '.join(stale)} differ from a fresh run. "
                "Run `just data::generate`.",
                file=sys.stderr,
            )
            return 1
        print("ontology is up to date")
        return 0

    OUT_DATA.write_text(data, encoding="utf-8", newline="\n")
    OUT_VOCAB.write_text(vocab, encoding="utf-8", newline="\n")
    print(f"wrote {repo_relative(OUT_DATA)} and {repo_relative(OUT_VOCAB)}")
    print(fallbacks.summary())
    return 0