Rubrics

def annotate_node(  # noqa: PLR0913 - one argument per kind of annotation
    graph: Graph,
    rubric: Rubric,
    node_id: str,
    *,
    node_type: str | None = None,
    settings: Mapping[str, Any] | None = None,
    rests_on: Sequence[str] = (),
    verdict: str | None = None,
    reason: str | None = None,
) -> Graph:
    """Type, fill in, link and assess a node, validating each write against the rubric.

    The steps run in that order, so a type given here is the one the form and links are checked
    against. Required form fields may be left out: they are filled one at a time, and
    ``check_graph`` at ``typed`` reports the ones still missing.

    Args:
        graph: The graph, left unchanged.
        rubric: The rubric.
        node_id: The node.
        node_type: A declared type allowed for the node's origin, or ``unassigned``, which removes
            the type with the node's form, assessment and base. Assigning a type with a ``base``
            sets it when the node has none.
        settings: Form fields to set, by name; each must be declared by the node's type and its value
            must have the field type and be found in the node's text.
        rests_on: Nodes to add the type's edge-rule relation from, with the rule's strength.
        verdict: ``holds``, ``fails`` or ``undetermined``.
        reason: Why the node was not assessed; sets the verdict ``not_assessed``.

    Returns:
        The new graph.

    Raises:
        ValidationError: If any write breaks the rubric, or a relation would close a cycle.
    """
    node = graph.nodes[node_id]
    origin = origin_of(node)
    if origin is None:
        msg = f"node {node_id!r} has no valid attributes.origin, so the rubric cannot apply to it"
        raise ValidationError(msg)
    if node.kind == COMPOUND:
        msg = f"node {node_id!r} is a compound; annotate its children instead"
        raise ValidationError(msg)
    if verdict is not None and reason is not None:
        msg = "give a verdict or a reason for not assessing, not both"
        raise ValidationError(msg)
    if node_type is not None:
        node = _typed(rubric, node, origin, node_type)
    if settings:
        node = _filled(graph, rubric, node, origin, settings)
    if verdict is not None or reason is not None:
        node = _assessed(rubric, node, verdict, reason)
    relations = _links(graph, rubric, node, rests_on)
    return _rebuilt(graph, {node_id: node}, relations=relations)

def check_graph(graph: Graph, rubric: Rubric, level: str = TYPED) -> list[Finding]:
    """Check a graph against a rubric at a validation level and every level below it.

    Args:
        graph: The graph.
        rubric: The rubric.
        level: ``anchored``, ``typed`` or ``assessed``.

    Returns:
        Every violation: the ``anchored`` ones in graph order, then the ``typed`` ones, then the
        ``assessed`` ones. Empty when the graph passes. The graph is compiled only at ``assessed``.

    Raises:
        ValueError: If the level is not one of ``LEVELS``.
    """
    if level not in LEVELS:
        msg = f"level must be one of {list(LEVELS)}, got {level!r}"
        raise ValueError(msg)
    structure = _Structure.of(graph)
    findings = _anchored(graph, rubric, structure)
    if level in (TYPED, ASSESSED):
        findings.extend(_typed(graph, rubric, structure))
    if level == ASSESSED:
        findings.extend(_assessed(graph, rubric))
    return findings

def coverage(graph: Graph, rubric: Rubric) -> dict[str, object]:
    """Report how far classification has got. Never fails: it is a report, not a gate.

    Args:
        graph: The graph.
        rubric: The rubric.

    Returns:
        ``{"rubric", "nodes", "by_origin", "by_type", "untyped", "unexamined", "not_assessed"}``.
        ``by_type`` counts document nodes only, children and compounds included, in the order types
        first appear, each with its ``share`` of the document nodes. ``untyped`` lists the
        ``unassigned`` nodes and ``unexamined`` the nodes of ``assess = true`` types without an
        assessment, in graph order. A node without a valid origin is counted in ``nodes`` only.
    """
    by_origin = dict.fromkeys(ORIGINS, 0)
    by_type: dict[str, int] = {}
    untyped: list[str] = []
    unexamined: list[str] = []
    reasons: dict[str, object] = {}
    for node in graph:
        origin = origin_of(node)
        if origin is not None:
            by_origin[origin] += 1
        if origin == DOCUMENT:
            by_type[node.kind] = by_type.get(node.kind, 0) + 1
        if node.kind == UNASSIGNED:
            untyped.append(node.id)
        node_type = rubric.types.get(node.kind)
        assessment = node.attributes.get("assessment")
        if node_type is not None and node_type.assess and assessment is None:
            unexamined.append(node.id)
        if isinstance(assessment, Mapping) and assessment.get("verdict") == NOT_ASSESSED:
            reasons[node.id] = assessment.get("reason")
    documents = by_origin[DOCUMENT]
    return {
        "rubric": rubric.identity(),
        "nodes": len(graph),
        "by_origin": by_origin,
        "by_type": {name: {"count": count, "share": count / documents} for name, count in by_type.items()},
        "untyped": untyped,
        "unexamined": unexamined,
        "not_assessed": {"count": len(reasons), "reasons": reasons},
    }

def has_eqref(text: str, label: str) -> bool:
    r"""Tell whether a text references an equation.

    Args:
        text: The text.
        label: An equation label, matched by ``\ref``, ``\eqref``, ``\cref`` or ``\Cref``, or a
            printed number, matched as ``(label)``.

    Returns:
        Whether the text references it.
    """
    commands = ("ref", "eqref", "cref", "Cref")
    return any(f"\\{command}{{{label}}}" in text for command in commands) or f"({label})" in text

def has_numeral(text: str, number: float) -> bool:
    """Tell whether a number appears as a numeral in a text.

    Args:
        text: The text.
        number: The number.

    Returns:
        Whether some numeral of the text parses to the same float.
    """
    return float(number) in numerals(text)

def has_unit(text: str, unit: str) -> bool:
    r"""Tell whether a unit appears in a text as a token, delimited by whitespace or punctuation.

    The argument of ``\mathrm{…}``, ``\text{…}`` and ``\si{…}`` is delimited by its braces, so it
    counts as a token too. A boundary is required only where the unit begins or ends with a letter
    or digit: it keeps ``Hz`` from matching inside ``kHz`` or ``100Hz``, while a unit that starts
    with punctuation, such as the ``%`` of ``1.1%``, is itself delimited from the numeral before it.
    The operators ``/``, ``^``, ``·``, ``⋅``, ``*`` and the multiplication sign join symbols into one
    compound unit, and the subscript operator ``_`` joins a symbol to its subscript, rather than
    delimit them, so ``m`` is not a token of ``m/s``, ``m^2`` or ``m_e``; the compound itself is.

    Args:
        text: The text.
        unit: The unit, such as ``Hz``.

    Returns:
        Whether the unit is a token of the text.
    """
    before = rf"(?<![^\W_]|{_JOINERS})" if _ALPHANUMERIC.match(unit) else ""
    after = rf"(?![^\W_]|{_JOINERS})" if _ALPHANUMERIC.match(unit[-1:]) else ""
    return re.search(before + re.escape(unit) + after, text) is not None

def load_rubric(path: str | Path) -> Rubric:
    """Read a rubric file.

    Args:
        path: The TOML file.

    Returns:
        The rubric.

    Raises:
        RubricError: If the file is not TOML or breaks a rule of the rubric format.
        OSError: If the file cannot be read.
    """
    return loads_rubric(Path(path).read_text(encoding="utf-8"))

def loads_rubric(text: str) -> Rubric:
    """Read a rubric from TOML text.

    Args:
        text: The TOML document.

    Returns:
        The rubric.

    Raises:
        RubricError: If the text is not TOML or breaks a rule of the rubric format.
    """
    try:
        data = tomllib.loads(text)
    except tomllib.TOMLDecodeError as error:
        msg = f"not valid TOML: {error}"
        raise RubricError(msg) from None
    return rubric_from_dict(data)

def normalise_text(text: str) -> str:
    """Normalise a text for storage: Unicode NFC, whitespace runs collapsed to one space, ends trimmed.

    Args:
        text: The text as read from the source.

    Returns:
        The normalised text.
    """
    return _WHITESPACE.sub(" ", unicodedata.normalize("NFC", text)).strip()

def numerals(text: str) -> set[float]:
    """Return the values of every numeral in a text.

    A numeral followed by ``%`` contributes both its value and its value divided by 100. Values are
    computed in decimal arithmetic and rounded once, so ``1.1%`` gives exactly the float ``0.011``.

    Args:
        text: The text.

    Returns:
        The values.
    """
    values: set[float] = set()
    for match in _NUMERAL.finditer(text):
        digits = re.sub(_GROUP, "", match["digits"])
        value = Decimal(digits)
        exponent = match["e"] or match["braced"] or match["bare"]
        if exponent is not None:
            value = value.scaleb(int(exponent.replace(_MINUS, "-")))
        if match["sign"] in ("-", _MINUS):
            value = -value
        values.add(float(value))
        if match["percent"]:
            values.add(float(value / 100))
    return values

def parse_value(text: str) -> Any:
    """Read a form value from the command line: as JSON if it parses, else as the string itself.

    Args:
        text: The value as typed.

    Returns:
        The value.
    """
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return text

def rubric_from_dict(data: Any) -> Rubric:
    """Build a rubric from the data of a rubric file.

    Args:
        data: The parsed TOML document.

    Returns:
        The rubric.

    Raises:
        RubricError: If the data breaks a rule of the rubric format, or its format is unknown.
    """
    top = _table(data, "the rubric", {"rubric", "parameters", "initial", "types"}, frozenset({"rubric", "types"}))
    header = _table(top["rubric"], "rubric", {"format", "name", "version"}, frozenset({"format", "name", "version"}))
    if type(header["format"]) is not int or header["format"] != RUBRIC_FORMAT:
        msg = f"rubric.format {header['format']!r} is not a known rubric format; this version reads {RUBRIC_FORMAT}"
        raise RubricError(msg)
    parameters = {}
    for name, value in _table(top.get("parameters", {}), "parameters").items():
        table = _table(value, f"parameters.{name}", {"unit", "domain"}, frozenset({"unit"}))
        domain = _domain(table["domain"], f"parameters.{name}.domain") if "domain" in table else (None, None)
        parameters[name] = Parameter(name, _string(table["unit"], f"parameters.{name}.unit"), domain)
    types_table = _table(top["types"], "types")
    types = {name: _node_type(name, value) for name, value in types_table.items()}
    for node_type in types.values():
        rule = node_type.rests_on
        for name in rule.types if rule else ():
            if name not in types:
                msg = f"types.{node_type.name}.rests_on.types names {name!r}, which is not a declared type"
                raise RubricError(msg)
    initial = {}
    for unit, name in _table(top.get("initial", {}), "initial").items():
        if not isinstance(name, str) or name not in types or DOCUMENT not in types[name].origins:
            msg = f"initial.{unit} names {name!r}, which is not a declared type allowed for document nodes"
            raise RubricError(msg)
        initial[unit] = name
    return Rubric(
        _string(header["name"], "rubric.name"),
        _string(header["version"], "rubric.version"),
        MappingProxyType(types),
        MappingProxyType(parameters),
        MappingProxyType(initial),
    )

def split_node(
    graph: Graph, node_id: str, clauses: Sequence[str] = (), *, fields: bool = False
) -> tuple[Graph, list[str]]:
    """Make a document chunk a compound, with one span child per clause or one field child.

    A span child's text is a clause of the chunk, its locator the chunk's plus ``;chars=<start>-<end>``.
    A field child has no text, a copy of the chunk's anchor, and gets its content from a form later.
    Children are ``<node id>a``, ``<node id>b``, … in the order created, joined to the chunk by a
    ``part-of`` relation, and start ``unassigned``. Splitting a compound adds children to it.

    Args:
        graph: The graph, left unchanged.
        node_id: The chunk to split.
        clauses: The clauses, each occurring exactly once in the chunk's text.
        fields: Add one field child instead of span children.

    Returns:
        The new graph, and the ids of the children added.

    Raises:
        ValidationError: If the node is not a top-level document chunk with a locator, carries a form,
            an assessment or a base, takes part in a relation a compound may not, or a clause does not
            occur exactly once in its text; or if both or neither of ``clauses`` and ``fields`` is given.
    """
    if bool(clauses) == fields:
        msg = "split needs either clauses or fields, not both"
        raise ValidationError(msg)
    node = graph.nodes[node_id]
    _check_splittable(graph, node)
    anchor = node.sources[0]
    spans = []
    for clause in clauses:
        starts = _occurrences(node.statement, clause) if clause else []
        if len(starts) != 1:
            msg = f"clause {clause!r} occurs {len(starts)} times in the text of {node_id!r}, not exactly once"
            raise ValidationError(msg)
        spans.append((starts[0], clause))
    unit = node.attributes.get("unit")
    base_attributes = {"origin": DOCUMENT, **({"unit": unit} if unit is not None else {})}
    suffixes = (suffix for suffix in _suffixes() if f"{node_id}{suffix}" not in graph)
    children = []
    for start, clause in spans or [(None, None)]:
        child_id = f"{node_id}{next(suffixes)}"
        if start is None:
            child = Node(child_id, kind=UNASSIGNED, sources=node.sources, attributes=dict(base_attributes))
        else:
            locator = f"{anchor.locator};chars={start}-{start + len(clause)}"
            source = SourceAnchor(anchor.document, locator, clause, text_digest(clause))
            child = Node(
                child_id, kind=UNASSIGNED, statement=clause, sources=(source,), attributes=dict(base_attributes)
            )
        children.append(child)
    relations = [Relation(f"{child.id}-{PART_OF}-{node_id}", PART_OF, child.id, node_id) for child in children]
    for relation in relations:
        if relation.id in graph.relations:
            msg = f"a relation with id {relation.id!r} already exists"
            raise ValidationError(msg)
    existing = [
        relation.source
        for relation in graph.relations.values()
        if relation.type == PART_OF and relation.target == node_id
    ]
    order = list(graph.nodes)
    after = max([node_id, *existing], key=order.index)
    compound = replace(node, kind=COMPOUND)
    return _rebuilt(graph, {node_id: compound}, after, children, relations), [child.id for child in children]

def text_digest(text: str) -> str:
    """Return the digest of a text: ``sha256:`` and the hex SHA-256 of its normalised UTF-8 bytes.

    Args:
        text: The text.

    Returns:
        The digest, such as ``sha256:9f86d0...``.
    """
    return DIGEST_PREFIX + hashlib.sha256(normalise_text(text).encode("utf-8")).hexdigest()