ods_to_rdf

ods_to_rdf

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

Pin rows carry their own id and link by id in a links column; a link's predicate follows what it points at, so there is no mapping to configure. Concepts never link out: a tag is recorded on the pin that carries it, so a region is on the map only because some pin points at it.

The use-case subdirectory under src/ontology/ is selected by COMPASS_USE_CASE (default oceancare).

turtle-generator/ods_to_rdf.py            regenerate
turtle-generator/ods_to_rdf.py --check    exit 1 if the committed files are stale

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

SheetError

Bases: Exception

A defect in the tables that the operator must resolve.

Problems(items=list()) dataclass

Collects every defect in one pass, so a run reports all of them at once.

Fallbacks(counts=dict()) dataclass

Counts German cells that were empty and took the English text instead.

_resolve_use_case()

Return COMPASS_USE_CASE from the environment or repo .env.

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

def _resolve_use_case() -> str:
    """Return ``COMPASS_USE_CASE`` from the environment or 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():
            stripped = line.strip()
            if not stripped.startswith("COMPASS_USE_CASE="):
                continue
            raw = stripped.split("=", 1)[1].strip().strip("\"'")
            if " #" in raw:
                raw = raw.split(" #", 1)[0].strip()
            if raw:
                return raw
    return "oceancare"

display(path)

Repo-relative where possible, absolute otherwise, so messages never crash.

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

def display(path: Path) -> str:
    """Repo-relative where possible, absolute otherwise, so messages never crash."""
    try:
        return str(path.relative_to(REPO))
    except ValueError:
        return str(path)

_cell_text(cell)

A cell's value, preferring the stored number over its displayed form.

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

def _cell_text(cell) -> str:
    """A cell's value, preferring the stored number over its displayed form."""
    if cell.getAttribute("valuetype") == "float":
        stored = cell.getAttribute("value")
        if stored is not None:
            # Trim the trailing .0 a spreadsheet adds to whole numbers.
            return stored[:-2] if stored.endswith(".0") else stored
    return "\n".join(str(p) for p in cell.getElementsByType(P)).strip()

_row_values(row, width)

Expand a row's cells, honouring the repeat counts spreadsheets pack with.

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

def _row_values(row, width: int) -> list[str]:
    """Expand a row's cells, honouring the repeat counts spreadsheets pack with."""
    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_table(name, columns)

Read one sheet of the workbook, requiring exactly the expected header.

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

def read_table(name: str, columns: list[str]) -> list[Row]:
    """Read one sheet of the workbook, requiring exactly the expected header."""
    sheet_rows = _sheet(name).getElementsByType(TableRow)
    if not sheet_rows:
        raise SheetError(f"sheet {name!r} is empty")

    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]
        detail = ", ".join(
            part
            for part in (
                f"missing {missing}" if missing else "",
                f"unexpected {extra}" if extra else "",
                "" if (missing or extra) else "columns are in the wrong order",
            )
            if part
        )
        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))
        # A repeated row is spreadsheet padding, so only a filled one counts.
        for _ in range(repeat if any(values) else 1):
            number += 1
            if any(values):
                rows.append(
                    Row(
                        number=number,
                        table=name,
                        # _row_values pads to len(columns), so the two always match.
                        cells=dict(zip(columns, values, strict=True)),
                    )
                )
    return rows

index_terms(concepts, pins, problems)

Map every id to the dimension or class it belongs to.

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 id to the dimension or class it belongs to."""
    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.table, row.number, "row has no id")
                continue
            if identifier in seen:
                problems.add(
                    row.table,
                    row.number,
                    f"id {identifier!r} is already used by "
                    f"{seen[identifier].table}:{seen[identifier].number}",
                )
                continue
            if row[column] not in allowed:
                problems.add(
                    row.table,
                    row.number,
                    f"{column} {row[column]!r} is not one of {sorted(allowed)}",
                )
                continue
            seen[identifier] = row
            kinds[identifier] = row[column]
    return kinds

parse_links(row, kinds, problems)

Resolve a links cell into objects grouped by the predicate they become.

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]]:
    """Resolve a links cell into objects grouped by the predicate they become."""
    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.table, row.number, f"{target!r} links to itself")
            continue
        if target not in kinds:
            problems.add(
                row.table,
                row.number,
                f"link to unknown id {target!r} -- check the spelling, or add the row",
            )
            continue
        predicate = TAG_PREDICATE[kinds[target]]
        prefix = "compass:" if kinds[target] in DIMENSIONS else "ocinst:"
        grouped.setdefault(predicate, set()).add(prefix + target)
    return grouped

number(row, column, problems, kind)

Validate a numeric cell, reporting rather than raising on a bad value.

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

def number(row: Row, column: str, problems: Problems, kind: type) -> str:
    """Validate a numeric cell, reporting rather than raising on a bad value."""
    value = row[column]
    if not value:
        return ""
    try:
        kind(value)
    except ValueError:
        problems.add(row.table, row.number, f"{column} {value!r} is not a number")
        return ""
    return value

bilingual(row, column, fallbacks)

The English and German text of a <column>_en / <column>_de pair.

An empty German cell takes the English text, so a German reader never gets a blank where an English one gets prose. Each substitution is counted, so a missing translation stays visible instead of silently shipping.

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

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

    An empty German cell takes the English text, so a German reader never gets
    a blank where an English one gets prose. Each substitution is counted, so a
    missing translation stays visible instead of silently shipping.
    """
    english = row[f"{column}_en"]
    if not english:
        return None
    german = row[f"{column}_de"]
    if not german:
        fallbacks.note(row.table, f"{column}_de")
        german = english
    return english, german

order_key(triple)

Total order over a subject's triples: predicate, then language, then value.

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

def order_key(triple: tuple[str, str]) -> tuple[int, str, int, str]:
    """Total order over a subject's triples: predicate, then language, then value."""
    predicate, value = triple
    if predicate == "a":
        rank = -1
    elif predicate in PREDICATE_ORDER:
        rank = PREDICATE_ORDER.index(predicate)
    else:
        rank = len(PREDICATE_ORDER)
    suffix = value[-3:]
    language = (
        LANGUAGE_ORDER.index(suffix) if suffix in LANGUAGE_ORDER else len(LANGUAGE_ORDER)
    )
    return (rank, predicate if rank == len(PREDICATE_ORDER) else "", language, value)

generate(fallbacks=None)

The two Turtle files. Pass a Fallbacks to learn which translations are missing.

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

def generate(fallbacks: Fallbacks | None = None) -> tuple[str, str]:
    """The two Turtle files. Pass a Fallbacks to learn which translations are missing."""
    schemes = read_table(SCHEMES, SCHEME_COLUMNS)
    concepts = read_table(CONCEPTS, CONCEPT_COLUMNS)
    pins = read_table(PINS, PIN_COLUMNS)

    problems = Problems()
    fallbacks = fallbacks if fallbacks is not None else Fallbacks()
    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