src/endopath/dictionary.py

"""The target data dictionary an Intake submission points the tool at.

Intake points at a report source and at a target dictionary (docs/prd.md section
8.1). A dictionary names the variables to capture, each variable's permitted
values, and whether confirming a variable requires a licensed reference source.
The built-in dictionary is the CAP endometrium checklist, derived from
`endopath.fields` so it never drifts from `GET /api/fields`. An uploaded
dictionary arrives on the ingestion request in the same shape.
"""

from __future__ import annotations

import json
from typing import Optional

from pydantic import BaseModel, Field
from sqlalchemy import Connection

from endopath import fields, storage

# The CAP protocol the built-in checklist transcribes, recorded in
# data/taxonomy/sources.csv. The dictionary carries the source_id so a reviewer
# can trace every variable to the standard it came from.
BUILTIN_DICTIONARY_ID = "cap_uterus/5.1.0.0"
BUILTIN_DICTIONARY_TITLE = "CAP endometrium checklist"


class DictionaryVariable(BaseModel):
    """One variable to capture. `permitted_values` is None for an open
    vocabulary (a stage code, a lymph node string) and a closed list for an
    enumerated field. `licence_required` marks a variable whose confirmation
    needs a licensed reference source, so a deployment without that licence can
    gate it."""

    name: str
    label: Optional[str] = None
    permitted_values: Optional[list[str]] = None
    licence_required: bool = False


class DataDictionary(BaseModel):
    """A target dictionary: an id, a title, and the variables to capture. Both
    the built-in CAP checklist and an uploaded dictionary use this shape."""

    id: str
    title: str
    variables: list[DictionaryVariable] = Field(default_factory=list)

    @property
    def licence_required_variables(self) -> list[str]:
        return [v.name for v in self.variables if v.licence_required]


def register(conn: Connection, dictionary: DataDictionary) -> DataDictionary:
    """Store a submitted dictionary so the compilation job (#91) can read it back by id after the
    request that submitted it has returned. Returns the stored dictionary."""
    storage.upsert_dictionary(
        conn,
        dictionary_id=dictionary.id,
        title=dictionary.title,
        variables_json=json.dumps([v.model_dump() for v in dictionary.variables]),
    )
    return dictionary


def get_registered(conn: Connection, dictionary_id: str) -> Optional[DataDictionary]:
    """The stored dictionary, reconstructed from its row, or None if it was never submitted."""
    row = storage.get_dictionary(conn, dictionary_id)
    if row is None:
        return None
    variables = [DictionaryVariable(**item) for item in json.loads(row["variables_json"])]
    return DataDictionary(id=row["id"], title=row["title"], variables=variables)


def reset(conn: Connection) -> None:
    """Clear the stored dictionaries, the queue, and their artifacts. For tests and a fresh process."""
    storage.clear_intake_state(conn)


def builtin_cap_dictionary() -> DataDictionary:
    """The CAP endometrium checklist as a dictionary, one variable per checklist
    field, permitted values from each field's closed vocabulary (None for the
    open ones). No variable is licence-gated: every value is extracted from the
    report and stored with its citation, not drawn from a licensed source."""
    variables = [
        DictionaryVariable(
            name=spec.name,
            label=spec.label,
            permitted_values=list(spec.enum_values) if spec.enum_values else None,
            licence_required=False,
        )
        for spec in fields.FIELDS
    ]
    return DataDictionary(
        id=BUILTIN_DICTIONARY_ID,
        title=BUILTIN_DICTIONARY_TITLE,
        variables=variables,
    )