eng-lab/hwe/requirements.py

"""Requirements: what the drone shall do, and whether the current design does it.

A requirement here is a record in design/requirements.toml (EL01, section 2.11):

    [[requirement]]
    id = "SYS-001"
    text = "The drone shall have a thrust-to-weight ratio of at least 2.0 at full charge."
    rationale = "Hovering at or below half of maximum thrust leaves headroom to control."
    verification = "Analysis"        # Inspection, Analysis, Demonstration, or Test
    result = "hover.thrust_to_weight" # the engine result it is checked against
    op = ">="                         # ">=" or "<="
    threshold = "2.0"                 # with a unit whenever the result has one
    parent = ""                       # the requirement this one was derived from

`check` compares each requirement with the pipeline's results and reports its
status and margin. `validate` finds mistakes in the file itself (a missing
"shall", an unknown verification method, a threshold in the wrong unit) and is
what EL01 lab step 7 tests.
"""
from __future__ import annotations

import tomllib
from dataclasses import dataclass
from pathlib import Path

from hwe.units import Q_, Quantity, is_quantity, parse

METHODS = ("Inspection", "Analysis", "Demonstration", "Test")
OPS = (">=", "<=")


@dataclass
class Requirement:
    id: str
    text: str
    verification: str
    result: str
    op: str
    threshold: Quantity
    rationale: str = ""
    parent: str = ""


@dataclass
class Status:
    requirement: Requirement
    value: Quantity | None
    state: str            # "pass", "fail", "open" (no value yet), or "locked"
    margin: float | None  # fraction by which the value beats (+) or misses (-) the threshold
    note: str = ""


def load(path: Path | str) -> list[Requirement]:
    with open(path, "rb") as f:
        raw = tomllib.load(f)
    out = []
    for r in raw.get("requirement", []):
        out.append(Requirement(
            id=r.get("id", ""), text=r.get("text", ""), verification=r.get("verification", ""),
            result=r.get("result", ""), op=r.get("op", ""),
            threshold=parse(r["threshold"], default_unit="dimensionless") if "threshold" in r else None,
            rationale=r.get("rationale", ""), parent=r.get("parent", "")))
    return out


def validate(reqs: list[Requirement], known_results: dict[str, Quantity]) -> list[str]:
    """Every problem found in the requirement records, as readable sentences."""
    problems = []
    seen = set()
    for r in reqs:
        where = r.id or "(missing id)"
        if not r.id:
            problems.append("a requirement has no id")
        elif r.id in seen:
            problems.append(f"{where}: duplicate id")
        seen.add(r.id)
        if " shall " not in f" {r.text} ":
            problems.append(f"{where}: the text must state what the system 'shall' do")
        if r.verification not in METHODS:
            problems.append(f"{where}: verification must be one of {', '.join(METHODS)}")
        if r.op not in OPS:
            problems.append(f"{where}: op must be '>=' or '<='")
        if r.threshold is None:
            problems.append(f"{where}: no threshold")
        if r.result and r.result not in known_results and not r.result.startswith("future."):
            problems.append(f"{where}: unknown result '{r.result}' (known: {', '.join(sorted(known_results))})")
        elif r.result in known_results and r.threshold is not None:
            value = known_results[r.result]
            if is_quantity(value) and value.dimensionality != r.threshold.dimensionality:
                problems.append(f"{where}: threshold unit {r.threshold.units:~P} does not match "
                                f"{r.result} ({value.units:~P})")
    return problems


def check(reqs: list[Requirement], results: dict[str, Quantity], locked: set[str] = frozenset()) -> list[Status]:
    """Compare each requirement with the current results."""
    statuses = []
    for r in reqs:
        if r.result.startswith("future.") or r.result in locked:
            statuses.append(Status(r, None, "locked", None, "a later module computes this"))
            continue
        value = results.get(r.result)
        if value is None:
            statuses.append(Status(r, None, "open", None, "not computed yet (is your core code written?)"))
            continue
        v = value.to(r.threshold.units).m if is_quantity(value) else float(value)
        t = r.threshold.m
        if r.op == ">=":
            ok, margin = v >= t, (v - t) / abs(t) if t else None
        else:
            ok, margin = v <= t, (t - v) / abs(t) if t else None
        statuses.append(Status(r, value, "pass" if ok else "fail", margin))
    return statuses