From 4a0c7c7952a7c1a1b4ec44bd8c5a29448936c86f Mon Sep 17 00:00:00 2001 From: navidgh67 Date: Fri, 2 Oct 2026 09:12:03 -0400 Subject: [PATCH 1/5] Infer QUDT/SOSA/PROV annotations for registered equipment (dry run) Equipment registered through the wizard stores plain unit strings and a free-text equipment type, so FAIR exports for it lack QUDT, SOSA and PROV-O terms. Add fair_ontology.infer_ontology_annotations(), which derives them from what the owner already entered without overwriting existing values and marks each inferred value under ontology_inferred. Ambiguous or unknown units (A, a.u., blank) are reported, not guessed. Add a read-only report script (api/scripts/fair_ontology_dry_run.py) that lists, per equipment, what would be filled and what still needs an owner, from domains/configs or equipment_metadata.config_json. It also flags NanoM-PER-MIN in etcher.json, which does not exist on qudt.org. Co-Authored-By: Claude Opus 5.5 (1M context) --- api/fair_ontology.py | 380 +++++++++++++++++++++++++++ api/scripts/fair_ontology_dry_run.py | 126 +++++++++ api/tests/test_fair_ontology.py | 161 ++++++++++++ 3 files changed, 667 insertions(+) create mode 100644 api/fair_ontology.py create mode 100644 api/scripts/fair_ontology_dry_run.py create mode 100644 api/tests/test_fair_ontology.py diff --git a/api/fair_ontology.py b/api/fair_ontology.py new file mode 100644 index 0000000..869b609 --- /dev/null +++ b/api/fair_ontology.py @@ -0,0 +1,380 @@ +""" +fair_ontology.py — Infer ontology annotations for registered equipment. + +Equipment registered through the web wizard stores plain unit strings +("sccm", "mTorr") and a free-text equipment type ("Etcher"). The FAIR exporter +(fair_archiver.py) can emit QUDT / SOSA / PROV-O terms, but only when the +domain config carries them. This module derives those terms from what the +owner already entered, so existing registrations do not have to be re-edited. + +Rules: + - Never overwrite a value that is already present. + - Every inferred value is listed under ``ontology_inferred`` on the item it + was added to, so it can be told apart from an owner-confirmed value. + - Anything that cannot be resolved unambiguously is reported, not guessed. + +Every QUDT URI below was checked to resolve on qudt.org (2026-10-02). +""" + +from __future__ import annotations + +import copy +import re +from typing import Any + +QUDT_UNIT_BASE = "http://qudt.org/vocab/unit/" +QUDT_QK_BASE = "http://qudt.org/vocab/quantitykind/" +SOSA_BASE = "http://www.w3.org/ns/sosa/" + +# QUDT unit code -> QUDT quantity kind. A ``None`` quantity kind means the unit +# alone does not say what is being measured (e.g. percent). +QUDT_UNITS: dict[str, str | None] = { + "CentiM3-PER-MIN": "VolumeFlowRate", + "L-PER-MIN": "VolumeFlowRate", + "W": "Power", + "KiloW": "Power", + "MilliW": "Power", + "TORR": "Pressure", + "MilliTORR": "Pressure", + "PA": "Pressure", + "KiloPA": "Pressure", + "BAR": "Pressure", + "MilliBAR": "Pressure", + "PSI": "Pressure", + "ATM": "Pressure", + "M": "Length", + "CentiM": "Length", + "MilliM": "Length", + "MicroM": "Length", + "NanoM": "Length", + "ANGSTROM": "Length", + "M-PER-MIN": "LinearVelocity", + "M-PER-SEC": "LinearVelocity", + "MicroM-PER-MIN": "LinearVelocity", + "MicroM-PER-SEC": "LinearVelocity", + "DEG_C": "CelsiusTemperature", + "K": "ThermodynamicTemperature", + "SEC": "Time", + "MilliSEC": "Time", + "MIN": "Time", + "HR": "Time", + "V": "Voltage", + "MilliV": "Voltage", + "KiloV": "Voltage", + "A": "ElectricCurrent", + "MilliA": "ElectricCurrent", + "HZ": "Frequency", + "KiloHZ": "Frequency", + "MegaHZ": "Frequency", + "GM": "Mass", + "MilliGM": "Mass", + "KiloGM": "Mass", + "N-M": "Torque", + "REV-PER-MIN": "AngularVelocity", + "EV": "Energy", + "OHM": "Resistance", + "DEG": "PlaneAngle", + "PERCENT": None, +} + +# Units QUDT has a quantity kind for but no unit individual. nm/min is the +# etch-rate unit; qudt.org/vocab/unit/NanoM-PER-MIN does not exist. +QUANTITY_KIND_ONLY: dict[str, str] = { + "nm/min": "LinearVelocity", + "nm/s": "LinearVelocity", + "Å/min": "LinearVelocity", +} + +# Unit URIs seen in configs that return 404 on qudt.org (checked 2026-10-02). +KNOWN_MISSING_QUDT_UNITS = {"NanoM-PER-MIN", "NanoM-PER-SEC", "NanoM-PER-HR", "ANGSTROM-PER-MIN"} + +# Unit spellings an owner might type -> QUDT unit code. Matching is exact +# first; case-insensitive matching is limited to spellings where case cannot +# change the meaning (see _case_insensitive_ok). +UNIT_ALIASES: dict[str, str] = { + "sccm": "CentiM3-PER-MIN", + "cm3/min": "CentiM3-PER-MIN", + "cc/min": "CentiM3-PER-MIN", + "slm": "L-PER-MIN", + "slpm": "L-PER-MIN", + "L/min": "L-PER-MIN", + "lpm": "L-PER-MIN", + "W": "W", + "watt": "W", + "watts": "W", + "kW": "KiloW", + "mW": "MilliW", + "Torr": "TORR", + "mTorr": "MilliTORR", + "millitorr": "MilliTORR", + "Pa": "PA", + "kPa": "KiloPA", + "bar": "BAR", + "mbar": "MilliBAR", + "psi": "PSI", + "atm": "ATM", + "m": "M", + "cm": "CentiM", + "mm": "MilliM", + "um": "MicroM", + "micron": "MicroM", + "microns": "MicroM", + "nm": "NanoM", + "Å": "ANGSTROM", + "angstrom": "ANGSTROM", + "angstroms": "ANGSTROM", + "m/min": "M-PER-MIN", + "m/s": "M-PER-SEC", + "um/min": "MicroM-PER-MIN", + "um/s": "MicroM-PER-SEC", + "°C": "DEG_C", + "degC": "DEG_C", + "celsius": "DEG_C", + "K": "K", + "kelvin": "K", + "s": "SEC", + "sec": "SEC", + "seconds": "SEC", + "ms": "MilliSEC", + "min": "MIN", + "minutes": "MIN", + "h": "HR", + "hr": "HR", + "hours": "HR", + "V": "V", + "volt": "V", + "volts": "V", + "mV": "MilliV", + "kV": "KiloV", + "amp": "A", + "amps": "A", + "mA": "MilliA", + "Hz": "HZ", + "kHz": "KiloHZ", + "MHz": "MegaHZ", + "g": "GM", + "mg": "MilliGM", + "kg": "KiloGM", + "N-m": "N-M", + "Nm": "N-M", + "N*m": "N-M", + "newton meter": "N-M", + "rpm": "REV-PER-MIN", + "eV": "EV", + "ohm": "OHM", + "ohms": "OHM", + "Ω": "OHM", + "deg": "DEG", + "°": "DEG", + "degrees": "DEG", + "%": "PERCENT", + "percent": "PERCENT", +} + +# Spellings that name more than one unit in practice; reported, never guessed. +AMBIGUOUS_UNITS: dict[str, str] = { + "A": "ampere or an ASCII stand-in for angstrom", + "a": "ampere or an ASCII stand-in for angstrom", +} + +# Short symbols are case-sensitive (mW vs MW, mHz vs MHz, s vs S), so only +# words and these unambiguous abbreviations match case-insensitively. +_CASE_INSENSITIVE_SHORT = {"slm", "lpm", "psi", "atm", "bar", "rpm", "sec", "min", "ohm", "deg"} + + +def _case_insensitive_ok(alias: str) -> bool: + return len(alias) >= 4 or alias in _CASE_INSENSITIVE_SHORT + + +def _normalize_unit_text(unit: str) -> str: + text = unit.strip().replace("µ", "u").replace("μ", "u") + text = text.replace("·", "*").replace("⋅", "*") + text = re.sub(r"\s*/\s*", "/", text) + text = re.sub(r"\s+", " ", text) + return text + + +def _build_case_insensitive_aliases() -> dict[str, str]: + by_lower: dict[str, set[str]] = {} + for alias, code in UNIT_ALIASES.items(): + if not _case_insensitive_ok(alias): + continue + by_lower.setdefault(alias.lower(), set()).add(code) + # Drop any spelling that would still map to two different units. + return {key: next(iter(codes)) for key, codes in by_lower.items() if len(codes) == 1} + + +_ALIASES_CI = _build_case_insensitive_aliases() +_QUANTITY_KIND_ONLY_CI = {key.lower(): qk for key, qk in QUANTITY_KIND_ONLY.items()} + + +def infer_qudt(unit: Any) -> dict[str, str]: + """Return ``{"qudt_unit": uri, "qudt_quantity_kind": uri}`` for a unit + string, with either key omitted when it cannot be determined. + + Accepts plain spellings ("sccm"), QUDT codes ("MilliTORR") and full QUDT + unit URIs. + """ + if not isinstance(unit, str) or not unit.strip(): + return {} + text = _normalize_unit_text(unit) + if text in AMBIGUOUS_UNITS: + return {} + + code: str | None = None + if text.startswith(QUDT_UNIT_BASE): + candidate = text[len(QUDT_UNIT_BASE):] + code = candidate if candidate in QUDT_UNITS else None + elif text in QUDT_UNITS: + code = text + elif text in UNIT_ALIASES: + code = UNIT_ALIASES[text] + elif text.lower() in _ALIASES_CI: + code = _ALIASES_CI[text.lower()] + + if code is not None: + result = {"qudt_unit": QUDT_UNIT_BASE + code} + quantity_kind = QUDT_UNITS[code] + if quantity_kind: + result["qudt_quantity_kind"] = QUDT_QK_BASE + quantity_kind + return result + + quantity_kind = QUANTITY_KIND_ONLY.get(text) or _QUANTITY_KIND_ONLY_CI.get(text.lower()) + if quantity_kind: + return {"qudt_quantity_kind": QUDT_QK_BASE + quantity_kind} + return {} + + +# Equipment-type keywords -> SOSA class. Tools that change a sample act on it +# (Actuator); tools that measure it observe it (Sensor). +_ACTUATOR_KEYWORDS = { + "etch", "etcher", "etching", "rie", "icp", "deposition", "cvd", "pecvd", + "lpcvd", "pvd", "ald", "sputter", "sputtering", "evaporator", "evaporation", + "furnace", "anneal", "annealing", "oxidation", "lithography", "litho", + "aligner", "implant", "implanter", "diffusion", "coater", "dicing", "bonder", +} +_SENSOR_KEYWORDS = { + "metrology", "profilometer", "ellipsometer", "ellipsometry", "microscope", + "microscopy", "sem", "tem", "afm", "xrd", "xps", "spectrometer", + "spectroscopy", "cytometer", "cytometry", "interferometer", "reflectometer", + "characterization", "inspection", "sensor", "analyzer", +} + + +def infer_sosa_type(equipment_type: Any) -> tuple[str | None, str]: + """Return ``(sosa_uri_or_None, reason)`` for a free-text equipment type.""" + if not isinstance(equipment_type, str) or not equipment_type.strip(): + return None, "no equipment_type" + tokens = set(re.findall(r"[a-z0-9]+", equipment_type.lower())) + is_actuator = bool(tokens & _ACTUATOR_KEYWORDS) + is_sensor = bool(tokens & _SENSOR_KEYWORDS) + if is_actuator and not is_sensor: + return SOSA_BASE + "Actuator", f"equipment_type {equipment_type!r}" + if is_sensor and not is_actuator: + return SOSA_BASE + "Sensor", f"equipment_type {equipment_type!r}" + if is_actuator and is_sensor: + return None, f"equipment_type {equipment_type!r} matches both Actuator and Sensor" + return None, f"equipment_type {equipment_type!r} not recognized" + + +def _mark_inferred(item: dict[str, Any], field: str) -> None: + inferred = item.setdefault("ontology_inferred", []) + if field not in inferred: + inferred.append(field) + + +def _annotate_variable( + item: dict[str, Any], + *, + section: str, + name: str, + direction: str | None, + report: dict[str, Any], +) -> None: + label = f"{section}:{name}" + unit = item.get("unit", "") + + if direction and not item.get("prov_direction"): + item["prov_direction"] = direction + _mark_inferred(item, "prov_direction") + report["filled"].append({"item": label, "field": "prov_direction", "value": direction}) + + existing_unit = str(item.get("qudt_unit") or "") + if existing_unit.rsplit("/", 1)[-1] in KNOWN_MISSING_QUDT_UNITS: + report["warnings"].append({ + "item": label, + "field": "qudt_unit", + "reason": f"{existing_unit} does not exist on qudt.org", + }) + + missing = [f for f in ("qudt_unit", "qudt_quantity_kind") if not item.get(f)] + if not missing: + return + inferred = infer_qudt(unit) + for field in missing: + if field in inferred: + item[field] = inferred[field] + _mark_inferred(item, field) + report["filled"].append({"item": label, "field": field, "value": inferred[field]}) + else: + unit_text = _normalize_unit_text(unit) if isinstance(unit, str) else "" + if not unit_text: + reason = "no unit entered" + elif unit_text in AMBIGUOUS_UNITS: + reason = f"unit {unit!r} is ambiguous ({AMBIGUOUS_UNITS[unit_text]})" + else: + reason = f"unit {unit!r} not recognized" + if field == "qudt_quantity_kind" and "qudt_unit" in inferred: + reason = f"unit {unit!r} does not identify a quantity kind" + if field == "qudt_unit" and "qudt_quantity_kind" in inferred: + reason = f"QUDT has no unit for {unit!r}" + report["unresolved"].append({"item": label, "field": field, "reason": reason}) + + +def infer_ontology_annotations(config: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]: + """Return ``(annotated_copy, report)`` for a domain config. + + The input is not modified. ``report`` has ``filled`` (what was inferred) + ``unresolved`` (what still needs an owner) and ``warnings`` (existing + values that look wrong), each a list of dicts. + """ + annotated = copy.deepcopy(config) if isinstance(config, dict) else {} + report: dict[str, Any] = {"filled": [], "unresolved": [], "warnings": []} + + domain = annotated.get("domain") + if not isinstance(domain, dict): + domain = {} + annotated["domain"] = domain + if not (domain.get("sosa_type") or annotated.get("sosa_type")): + equipment_type = domain.get("equipment_type") or annotated.get("equipment_type") + sosa_type, reason = infer_sosa_type(equipment_type) + if sosa_type: + domain["sosa_type"] = sosa_type + _mark_inferred(domain, "sosa_type") + report["filled"].append({"item": "domain", "field": "sosa_type", "value": sosa_type, "basis": reason}) + else: + report["unresolved"].append({"item": "domain", "field": "sosa_type", "reason": reason}) + + for feature in annotated.get("features") or []: + if isinstance(feature, dict) and feature.get("id"): + _annotate_variable(feature, section="feature", name=str(feature["id"]), + direction="input", report=report) + + for key in ("primary", "secondary"): + target = (annotated.get("targets") or {}).get(key) + if isinstance(target, dict) and target.get("id"): + _annotate_variable(target, section=f"target.{key}", name=str(target["id"]), + direction=None, report=report) + + # fair_archiver already treats registered parameters as inputs and + # outputs as outputs, so only their units need annotating. + for section in ("parameters", "outputs"): + for item in annotated.get(section) or []: + if not isinstance(item, dict): + continue + name = item.get("name") or item.get("id") + if name: + _annotate_variable(item, section=section.rstrip("s"), name=str(name), + direction=None, report=report) + + return annotated, report diff --git a/api/scripts/fair_ontology_dry_run.py b/api/scripts/fair_ontology_dry_run.py new file mode 100644 index 0000000..6f21822 --- /dev/null +++ b/api/scripts/fair_ontology_dry_run.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +"""Report which ontology annotations could be inferred for registered equipment. + +Read-only: nothing is written to files or to the database. For each equipment +config it lists what fair_ontology.infer_ontology_annotations() would fill in +(QUDT unit / quantity kind, SOSA type, PROV direction), what would still +need the owner, and existing annotations that look wrong. + +Usage: + # Domain config files on disk (default: domains/configs) + python api/scripts/fair_ontology_dry_run.py + + # Registered equipment in PostgreSQL (equipment_metadata.config_json), + # using a read-only session + python api/scripts/fair_ontology_dry_run.py --from-db + + python api/scripts/fair_ontology_dry_run.py --json +""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +import sys + +API_ROOT = Path(__file__).resolve().parents[1] +if str(API_ROOT) not in sys.path: + sys.path.insert(0, str(API_ROOT)) + +from fair_ontology import infer_ontology_annotations # noqa: E402 + +DEFAULT_CONFIGS_DIR = API_ROOT.parent / "domains" / "configs" +SKIPPED_FILES = {"domain_config.schema.json", "template.json"} + + +def load_configs_from_dir(configs_dir: Path) -> list[dict]: + entries = [] + for path in sorted(configs_dir.glob("*.json")): + if path.name in SKIPPED_FILES: + continue + config = json.loads(path.read_text(encoding="utf-8")) + entries.append({"equipment": path.stem, "source": str(path), "config": config}) + return entries + + +def load_configs_from_db() -> list[dict]: + from data_loader_pg import get_pg_connection + from psycopg2.extras import RealDictCursor + + conn = get_pg_connection() + try: + conn.set_session(readonly=True) + with conn.cursor(cursor_factory=RealDictCursor) as cur: + cur.execute( + "SELECT domain_id, status, config_json FROM equipment_metadata ORDER BY domain_id" + ) + rows = cur.fetchall() + finally: + conn.close() + return [ + { + "equipment": row["domain_id"], + "source": f"equipment_metadata ({row['status']})", + "config": row["config_json"] if isinstance(row["config_json"], dict) else {}, + } + for row in rows + ] + + +def build_report(entries: list[dict]) -> list[dict]: + results = [] + for entry in entries: + _, report = infer_ontology_annotations(entry["config"]) + results.append( + { + "equipment": entry["equipment"], + "source": entry["source"], + "filled": report["filled"], + "unresolved": report["unresolved"], + "warnings": report["warnings"], + } + ) + return results + + +def render_text(results: list[dict]) -> str: + lines = [] + total_filled = total_unresolved = total_warnings = 0 + for result in results: + filled, unresolved = result["filled"], result["unresolved"] + total_filled += len(filled) + total_unresolved += len(unresolved) + total_warnings += len(result["warnings"]) + lines.append(f"== {result['equipment']} [{result['source']}]") + lines.append(f" would fill: {len(filled)} needs owner: {len(unresolved)}") + for item in filled: + lines.append(f" + {item['item']:<40} {item['field']:<20} {item['value']}") + for item in unresolved: + lines.append(f" ? {item['item']:<40} {item['field']:<20} {item['reason']}") + for item in result["warnings"]: + lines.append(f" ! {item['item']:<40} {item['field']:<20} {item['reason']}") + lines.append("") + lines.append( + f"Total: {len(results)} equipment, {total_filled} values inferable, " + f"{total_unresolved} need an owner, {total_warnings} warnings." + ) + return "\n".join(lines) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0]) + parser.add_argument("--from-db", action="store_true", + help="read equipment_metadata.config_json (read-only session)") + parser.add_argument("--configs-dir", type=Path, default=DEFAULT_CONFIGS_DIR) + parser.add_argument("--json", action="store_true", help="print JSON instead of text") + args = parser.parse_args() + + entries = load_configs_from_db() if args.from_db else load_configs_from_dir(args.configs_dir) + results = build_report(entries) + print(json.dumps(results, indent=2) if args.json else render_text(results)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/api/tests/test_fair_ontology.py b/api/tests/test_fair_ontology.py new file mode 100644 index 0000000..963243c --- /dev/null +++ b/api/tests/test_fair_ontology.py @@ -0,0 +1,161 @@ +import json +import sys +import unittest +from pathlib import Path + +API_DIR = Path(__file__).resolve().parents[1] +if str(API_DIR) not in sys.path: + sys.path.insert(0, str(API_DIR)) + +from fair_ontology import ( + QUDT_QK_BASE, + QUDT_UNIT_BASE, + SOSA_BASE, + infer_ontology_annotations, + infer_qudt, + infer_sosa_type, +) + +CONFIGS_DIR = API_DIR.parent / "domains" / "configs" + + +class InferQudtTests(unittest.TestCase): + def test_common_spellings(self): + self.assertEqual( + infer_qudt("sccm"), + { + "qudt_unit": QUDT_UNIT_BASE + "CentiM3-PER-MIN", + "qudt_quantity_kind": QUDT_QK_BASE + "VolumeFlowRate", + }, + ) + self.assertEqual(infer_qudt("mTorr")["qudt_unit"], QUDT_UNIT_BASE + "MilliTORR") + self.assertEqual(infer_qudt(" Watts ")["qudt_unit"], QUDT_UNIT_BASE + "W") + self.assertEqual(infer_qudt("µm")["qudt_unit"], QUDT_UNIT_BASE + "MicroM") + self.assertEqual(infer_qudt("um / min")["qudt_unit"], QUDT_UNIT_BASE + "MicroM-PER-MIN") + + def test_accepts_qudt_codes_and_uris(self): + self.assertEqual(infer_qudt("MilliTORR")["qudt_unit"], QUDT_UNIT_BASE + "MilliTORR") + uri = QUDT_UNIT_BASE + "NanoM" + self.assertEqual(infer_qudt(uri)["qudt_quantity_kind"], QUDT_QK_BASE + "Length") + + def test_case_matters_for_short_symbols(self): + self.assertEqual(infer_qudt("mW")["qudt_unit"], QUDT_UNIT_BASE + "MilliW") + self.assertEqual(infer_qudt("MW"), {}) # megawatt, not milliwatt + self.assertEqual(infer_qudt("MHz")["qudt_unit"], QUDT_UNIT_BASE + "MegaHZ") + self.assertEqual(infer_qudt("mHz"), {}) # millihertz, not megahertz + self.assertEqual(infer_qudt("S"), {}) # siemens, not seconds + self.assertEqual(infer_qudt("SCCM")["qudt_unit"], QUDT_UNIT_BASE + "CentiM3-PER-MIN") + + def test_ambiguous_and_unknown_units_are_not_guessed(self): + self.assertEqual(infer_qudt("A"), {}) + self.assertEqual(infer_qudt("C"), {}) + self.assertEqual(infer_qudt("a.u."), {}) + self.assertEqual(infer_qudt(""), {}) + self.assertEqual(infer_qudt(None), {}) + + def test_percent_has_unit_but_no_quantity_kind(self): + self.assertEqual(infer_qudt("%"), {"qudt_unit": QUDT_UNIT_BASE + "PERCENT"}) + + def test_etch_rate_has_quantity_kind_only(self): + # qudt.org has no NanoM-PER-MIN unit individual. + self.assertEqual(infer_qudt("nm/min"), {"qudt_quantity_kind": QUDT_QK_BASE + "LinearVelocity"}) + + +class InferSosaTypeTests(unittest.TestCase): + def test_actuator_and_sensor(self): + self.assertEqual(infer_sosa_type("Etcher")[0], SOSA_BASE + "Actuator") + self.assertEqual(infer_sosa_type("ALD Deposition")[0], SOSA_BASE + "Actuator") + self.assertEqual(infer_sosa_type("Metrology")[0], SOSA_BASE + "Sensor") + self.assertEqual(infer_sosa_type("SEM")[0], SOSA_BASE + "Sensor") + + def test_unknown_or_conflicting_types_are_not_guessed(self): + self.assertIsNone(infer_sosa_type("")[0]) + self.assertIsNone(infer_sosa_type("Glovebox")[0]) + self.assertIsNone(infer_sosa_type("Etch + Inspection")[0]) + self.assertIsNone(infer_sosa_type("Semiconductor tool")[0]) # "sem" only as a word + + +class InferOntologyAnnotationsTests(unittest.TestCase): + def registered_config(self): + return { + "domain": {"id": "sputter1", "display_name": "Sputter 1"}, + "equipment_type": "Sputtering", + "features": [ + {"id": "power", "unit": "W"}, + {"id": "mystery", "unit": "a.u."}, + ], + "targets": { + "primary": {"id": "thickness", "unit": "nm"}, + "secondary": {"id": "uniformity", "unit": "%"}, + }, + "parameters": [{"name": "pressure", "unit": "mTorr"}], + "outputs": [{"name": "rate", "unit": ""}], + } + + def test_fills_missing_values_and_marks_them_inferred(self): + source = self.registered_config() + annotated, report = infer_ontology_annotations(source) + + self.assertEqual(annotated["domain"]["sosa_type"], SOSA_BASE + "Actuator") + self.assertEqual(annotated["domain"]["ontology_inferred"], ["sosa_type"]) + + power = annotated["features"][0] + self.assertEqual(power["qudt_unit"], QUDT_UNIT_BASE + "W") + self.assertEqual(power["qudt_quantity_kind"], QUDT_QK_BASE + "Power") + self.assertEqual(power["prov_direction"], "input") + self.assertEqual( + sorted(power["ontology_inferred"]), + ["prov_direction", "qudt_quantity_kind", "qudt_unit"], + ) + + pressure = annotated["parameters"][0] + self.assertEqual(pressure["qudt_unit"], QUDT_UNIT_BASE + "MilliTORR") + self.assertNotIn("prov_direction", pressure) # exporter already treats it as input + + self.assertEqual(annotated["targets"]["secondary"]["qudt_unit"], QUDT_UNIT_BASE + "PERCENT") + + unresolved = {(item["item"], item["field"]) for item in report["unresolved"]} + self.assertIn(("feature:mystery", "qudt_unit"), unresolved) + self.assertIn(("target.secondary:uniformity", "qudt_quantity_kind"), unresolved) + self.assertIn(("output:rate", "qudt_unit"), unresolved) + + # The input is left untouched. + self.assertEqual(source, self.registered_config()) + + def test_never_overwrites_existing_values(self): + config = self.registered_config() + config["domain"]["sosa_type"] = SOSA_BASE + "Sensor" + config["features"][0].update( + qudt_unit="custom:unit", qudt_quantity_kind="custom:qk", prov_direction="output" + ) + annotated, report = infer_ontology_annotations(config) + + self.assertEqual(annotated["domain"]["sosa_type"], SOSA_BASE + "Sensor") + self.assertNotIn("ontology_inferred", annotated["domain"]) + power = annotated["features"][0] + self.assertEqual( + (power["qudt_unit"], power["qudt_quantity_kind"], power["prov_direction"]), + ("custom:unit", "custom:qk", "output"), + ) + self.assertNotIn("ontology_inferred", power) + self.assertFalse([f for f in report["filled"] if f["item"] in ("domain", "feature:power")]) + + def test_canonical_etcher_needs_nothing_but_flags_missing_qudt_unit(self): + etcher = json.loads((CONFIGS_DIR / "etcher.json").read_text()) + _, report = infer_ontology_annotations(etcher) + + self.assertEqual(report["filled"], []) + self.assertEqual(report["unresolved"], []) + self.assertEqual( + [(w["item"], w["field"]) for w in report["warnings"]], + [("target.primary:AvgEtchRate", "qudt_unit")], + ) + + def test_tolerates_empty_and_malformed_configs(self): + for config in ({}, {"features": None, "targets": None}, {"features": ["x", 3]}): + _, report = infer_ontology_annotations(config) + self.assertEqual(report["filled"], []) + + +if __name__ == "__main__": + unittest.main() From af1ebd64b32ea1b36ea6c49ea5dd856143e4e7e4 Mon Sep 17 00:00:00 2001 From: navidgh67 Date: Fri, 2 Oct 2026 09:17:38 -0400 Subject: [PATCH 2/5] Apply ontology inference when building FAIR exports Equipment edits rebuild config_json from the form, so ontology values written back to the database would be lost on the next edit. Infer them at export time instead: _load_domain_config now normalizes and then runs infer_ontology_annotations, so run JSON-LD and RO-Crate bundles for registered equipment carry QUDT units/quantity kinds, a SOSA type and feature PROV direction. Existing values still win; if inference raises, the export proceeds without it. Drop the etcher's qudt_unit NanoM-PER-MIN, which does not exist on qudt.org; the target keeps its LinearVelocity quantity kind and its nm/min unitText. Units QUDT has no individual for are no longer reported as owner work, and the schema example now uses a real unit URI. Co-Authored-By: Claude Opus 5.5 (1M context) --- api/fair_archiver.py | 23 ++++++- api/fair_ontology.py | 13 +++- api/tests/test_fair_ontology.py | 74 +++++++++++++++++++++-- domains/configs/domain_config.schema.json | 2 +- domains/configs/etcher.json | 3 +- 5 files changed, 103 insertions(+), 12 deletions(-) diff --git a/api/fair_archiver.py b/api/fair_archiver.py index 72aa6d5..ff8ec9d 100644 --- a/api/fair_archiver.py +++ b/api/fair_archiver.py @@ -161,7 +161,7 @@ def _load_domain_config(equipment_id: str) -> dict[str, Any] | None: from domain_configs import load_domain_config config = load_domain_config(equipment_id) if config: - return _normalize_pg_domain_config(config) + return _prepare_domain_config(config) except Exception: pass @@ -170,7 +170,7 @@ def _load_domain_config(equipment_id: str) -> dict[str, Any] | None: config_path = configs_dir / f"{equipment_id}.json" if config_path.exists(): try: - return _normalize_pg_domain_config(json.loads(config_path.read_text())) + return _prepare_domain_config(json.loads(config_path.read_text())) except Exception as exc: logger.debug("Could not load domain config %s: %s", equipment_id, exc) @@ -185,12 +185,29 @@ def _load_domain_config(equipment_id: str) -> dict[str, Any] | None: if equipment: config = equipment.get("config_json") if isinstance(config, dict) and config: - return _normalize_pg_domain_config(config) + return _prepare_domain_config(config) except Exception as exc: logger.debug("Could not load PG config for %s: %s", equipment_id, exc) return None +def _prepare_domain_config(config: dict[str, Any]) -> dict[str, Any]: + """Normalize a domain config, then fill in ontology annotations (QUDT, + SOSA, PROV direction) the owner did not enter. Inference never overwrites + an existing value; if it fails, the normalized config is used as is.""" + normalized = _normalize_pg_domain_config(config) + if not isinstance(normalized, dict): + return normalized + try: + from fair_ontology import infer_ontology_annotations + + annotated, _ = infer_ontology_annotations(normalized) + return annotated + except Exception as exc: + logger.warning("Ontology inference failed; exporting without it: %s", exc) + return normalized + + def _normalize_pg_domain_config(config: dict[str, Any]) -> dict[str, Any]: """Normalize a registered equipment config (whether loaded from its JSON snapshot or from PostgreSQL) so the FAIR metadata generator (which reads diff --git a/api/fair_ontology.py b/api/fair_ontology.py index 869b609..de1d137 100644 --- a/api/fair_ontology.py +++ b/api/fair_ontology.py @@ -208,6 +208,13 @@ def _build_case_insensitive_aliases() -> dict[str, str]: _QUANTITY_KIND_ONLY_CI = {key.lower(): qk for key, qk in QUANTITY_KIND_ONLY.items()} +def _quantity_kind_only(unit: Any) -> bool: + if not isinstance(unit, str): + return False + text = _normalize_unit_text(unit) + return text in QUANTITY_KIND_ONLY or text.lower() in _QUANTITY_KIND_ONLY_CI + + def infer_qudt(unit: Any) -> dict[str, str]: """Return ``{"qudt_unit": uri, "qudt_quantity_kind": uri}`` for a unit string, with either key omitted when it cannot be determined. @@ -326,8 +333,10 @@ def _annotate_variable( reason = f"unit {unit!r} not recognized" if field == "qudt_quantity_kind" and "qudt_unit" in inferred: reason = f"unit {unit!r} does not identify a quantity kind" - if field == "qudt_unit" and "qudt_quantity_kind" in inferred: - reason = f"QUDT has no unit for {unit!r}" + if field == "qudt_unit" and _quantity_kind_only(unit): + # QUDT itself has no unit for this (e.g. nm/min); the export + # carries the quantity kind plus unitText, so nothing to ask. + continue report["unresolved"].append({"item": label, "field": field, "reason": reason}) diff --git a/api/tests/test_fair_ontology.py b/api/tests/test_fair_ontology.py index 963243c..3a58be0 100644 --- a/api/tests/test_fair_ontology.py +++ b/api/tests/test_fair_ontology.py @@ -140,22 +140,88 @@ def test_never_overwrites_existing_values(self): self.assertNotIn("ontology_inferred", power) self.assertFalse([f for f in report["filled"] if f["item"] in ("domain", "feature:power")]) - def test_canonical_etcher_needs_nothing_but_flags_missing_qudt_unit(self): + def test_canonical_etcher_is_fully_annotated(self): etcher = json.loads((CONFIGS_DIR / "etcher.json").read_text()) _, report = infer_ontology_annotations(etcher) - self.assertEqual(report["filled"], []) - self.assertEqual(report["unresolved"], []) + self.assertEqual(report, {"filled": [], "unresolved": [], "warnings": []}) + + def test_flags_existing_qudt_unit_that_does_not_exist(self): + config = {"targets": {"primary": { + "id": "rate", "unit": "nm/min", + "qudt_unit": QUDT_UNIT_BASE + "NanoM-PER-MIN", + "qudt_quantity_kind": QUDT_QK_BASE + "LinearVelocity", + }}} + _, report = infer_ontology_annotations(config) self.assertEqual( [(w["item"], w["field"]) for w in report["warnings"]], - [("target.primary:AvgEtchRate", "qudt_unit")], + [("target.primary:rate", "qudt_unit")], ) + def test_unit_without_qudt_individual_is_not_owner_work(self): + config = {"features": [{"id": "rate", "unit": "nm/min"}]} + annotated, report = infer_ontology_annotations(config) + self.assertEqual(annotated["features"][0]["qudt_quantity_kind"], QUDT_QK_BASE + "LinearVelocity") + self.assertNotIn("qudt_unit", annotated["features"][0]) + self.assertEqual([u for u in report["unresolved"] if u["item"] == "feature:rate"], []) + def test_tolerates_empty_and_malformed_configs(self): for config in ({}, {"features": None, "targets": None}, {"features": ["x", 3]}): _, report = infer_ontology_annotations(config) self.assertEqual(report["filled"], []) +class ExportUsesInferenceTests(unittest.TestCase): + """fair_archiver applies inference when it loads a config for export.""" + + REGISTERED = { + "domain": {"id": "sputter1", "display_name": "Sputter 1"}, + "equipment_type": "Sputtering", + "features": [{"id": "power", "unit": "W"}], + "outputs": [{"name": "thickness", "unit": "nm"}], + } + + def test_prepare_domain_config_adds_ontology_terms(self): + import fair_archiver + + prepared = fair_archiver._prepare_domain_config(self.REGISTERED) + self.assertEqual(prepared["domain"]["sosa_type"], SOSA_BASE + "Actuator") + mapping = fair_archiver._get_feature_ontology_map(prepared) + self.assertEqual(mapping["power"]["qudt_unit"], QUDT_UNIT_BASE + "W") + self.assertEqual(mapping["power"]["prov_direction"], "input") + self.assertEqual(mapping["thickness"]["qudt_unit"], QUDT_UNIT_BASE + "NanoM") + + def test_run_jsonld_for_registered_equipment_carries_qudt_and_sosa(self): + from unittest.mock import patch + + import fair_archiver + + run = { + "equipment_id": "sputter1", + "equipment_name": "Sputter 1", + "features": {"power": 150.0}, + "outputs": {"thickness": 42.0}, + } + prepared = fair_archiver._prepare_domain_config(self.REGISTERED) + with patch.object(fair_archiver, "_load_domain_config", return_value=prepared): + jsonld = fair_archiver.run_to_jsonld(run) + + nodes = jsonld.get("@graph", [jsonld]) + text = json.dumps(nodes) + self.assertIn(QUDT_UNIT_BASE + "W", text) + self.assertIn(QUDT_UNIT_BASE + "NanoM", text) + self.assertIn(SOSA_BASE + "Actuator", text) + + def test_inference_failure_falls_back_to_normalized_config(self): + from unittest.mock import patch + + import fair_archiver + + with patch("fair_ontology.infer_ontology_annotations", side_effect=RuntimeError("boom")): + prepared = fair_archiver._prepare_domain_config(self.REGISTERED) + self.assertNotIn("sosa_type", prepared["domain"]) + self.assertEqual(prepared["domain"]["equipment_type"], "Sputtering") + + if __name__ == "__main__": unittest.main() diff --git a/domains/configs/domain_config.schema.json b/domains/configs/domain_config.schema.json index e45e464..2bb5657 100644 --- a/domains/configs/domain_config.schema.json +++ b/domains/configs/domain_config.schema.json @@ -371,7 +371,7 @@ "qudt_unit": { "type": "string", "format": "uri", - "description": "QUDT unit URI (e.g. http://qudt.org/vocab/unit/NanoM-PER-MIN)" + "description": "QUDT unit URI (e.g. http://qudt.org/vocab/unit/MilliTORR)" } }, "additionalProperties": false diff --git a/domains/configs/etcher.json b/domains/configs/etcher.json index e7e8006..d8d3a02 100644 --- a/domains/configs/etcher.json +++ b/domains/configs/etcher.json @@ -153,8 +153,7 @@ "max": 110 }, "outlier_filter": 250, - "qudt_quantity_kind": "http://qudt.org/vocab/quantitykind/LinearVelocity", - "qudt_unit": "http://qudt.org/vocab/unit/NanoM-PER-MIN" + "qudt_quantity_kind": "http://qudt.org/vocab/quantitykind/LinearVelocity" }, "secondary": { "id": "RangeEtchRate", From 0185620dab77d7ae5e6e8a3d2059bd2b9991b4e6 Mon Sep 17 00:00:00 2001 From: navidgh67 Date: Fri, 2 Oct 2026 09:49:39 -0400 Subject: [PATCH 3/5] Handle production equipment configs in ontology inference and export A read-only dry run against the 13 registered equipment showed: - Workflow-example equipment store domain as a string and features and targets as name lists. Both the inference and fair_archiver's _get_feature_ontology_map / _normalize_pg_domain_config assumed dicts and raised, so FAIR export for those runs was already failing. - Most unresolved items were not owner work: text/date/flag fields (skipped when their declared type is not numeric), the primary_metric/secondary_metric placeholder targets, percent positions (now DimensionlessRatio), count/unitless/ratio and sub-second times (NUM, UNITLESS, MicroSEC/NanoSEC/PicoSEC; URIs checked). - Equipment typed "other" now get a SOSA type from the equipment name. - Corrupted units (U+FFFD) are reported as such. Unresolved items drop from 322 to 28. The dry-run script also gains --output and keeps going when one config fails. Co-Authored-By: Claude Opus 5.5 (1M context) --- api/fair_archiver.py | 20 ++++-- api/fair_ontology.py | 97 ++++++++++++++++++++++++++-- api/scripts/fair_ontology_dry_run.py | 21 +++++- api/tests/test_fair_ontology.py | 78 +++++++++++++++++++++- 4 files changed, 199 insertions(+), 17 deletions(-) diff --git a/api/fair_archiver.py b/api/fair_archiver.py index ff8ec9d..328d801 100644 --- a/api/fair_archiver.py +++ b/api/fair_archiver.py @@ -221,7 +221,12 @@ def _normalize_pg_domain_config(config: dict[str, Any]) -> dict[str, Any]: if not isinstance(config, dict): return config normalized = dict(config) - domain = dict(normalized.get("domain") or {}) + raw_domain = normalized.get("domain") + if isinstance(raw_domain, dict): + domain = dict(raw_domain) + else: + # Workflow-example equipment store ``domain`` as the bare domain id. + domain = {"id": raw_domain} if isinstance(raw_domain, str) and raw_domain else {} for field in ( "manufacturer", "model", @@ -252,8 +257,11 @@ def _get_feature_ontology_map(domain_config: dict[str, Any] | None) -> dict[str, if not domain_config: return mapping - # Features - for feature in domain_config.get("features", []): + # Features. Workflow-example equipment list bare feature names; those + # names are also registered ``parameters``, which are mapped below. + for feature in domain_config.get("features") or []: + if not isinstance(feature, dict): + continue fid = feature.get("id", "") if fid: entry: dict[str, str] = {"unit": feature.get("unit", "")} @@ -265,9 +273,11 @@ def _get_feature_ontology_map(domain_config: dict[str, Any] | None) -> dict[str, entry["prov_direction"] = feature["prov_direction"] mapping[fid] = entry - # Targets (always output) + # Targets (always output). Workflow-example equipment list target names + # instead of target objects; their units come from ``outputs`` below. + targets = domain_config.get("targets") for target_key in ("primary", "secondary"): - target = domain_config.get("targets", {}).get(target_key) + target = targets.get(target_key) if isinstance(targets, dict) else None if target and target.get("id"): tid = target["id"] entry = {"unit": target.get("unit", ""), "prov_direction": "output"} diff --git a/api/fair_ontology.py b/api/fair_ontology.py index de1d137..28a40e1 100644 --- a/api/fair_ontology.py +++ b/api/fair_ontology.py @@ -56,6 +56,9 @@ "K": "ThermodynamicTemperature", "SEC": "Time", "MilliSEC": "Time", + "MicroSEC": "Time", + "NanoSEC": "Time", + "PicoSEC": "Time", "MIN": "Time", "HR": "Time", "V": "Voltage", @@ -74,7 +77,10 @@ "EV": "Energy", "OHM": "Resistance", "DEG": "PlaneAngle", - "PERCENT": None, + # Percent of travel/range (valve and match positions, duty cycles). + "PERCENT": "DimensionlessRatio", + "NUM": "Count", + "UNITLESS": "Dimensionless", } # Units QUDT has a quantity kind for but no unit individual. nm/min is the @@ -136,6 +142,9 @@ "sec": "SEC", "seconds": "SEC", "ms": "MilliSEC", + "us": "MicroSEC", + "ns": "NanoSEC", + "ps": "PicoSEC", "min": "MIN", "minutes": "MIN", "h": "HR", @@ -169,8 +178,24 @@ "degrees": "DEG", "%": "PERCENT", "percent": "PERCENT", + "count": "NUM", + "counts": "NUM", + "unitless": "UNITLESS", + "dimensionless": "UNITLESS", + "ratio": "UNITLESS", } +# Spellings owners use to say "this field has no unit". +NO_UNIT_MARKERS = {"n/a", "na", "none", "-", "--"} + +# Declared value types that are numeric quantities. Fields of any other +# declared type (text, dates, booleans, file links) are not physical +# quantities, so QUDT does not apply to them. +NUMERIC_TYPES = {"float", "int", "integer", "number", "numeric", "double", "real", "decimal"} + +# Target ids build_domain_config() writes when the owner set no target. +PLACEHOLDER_TARGET_IDS = {"primary_metric", "secondary_metric"} + # Spellings that name more than one unit in practice; reported, never guessed. AMBIGUOUS_UNITS: dict[str, str] = { "A": "ampere or an ASCII stand-in for angstrom", @@ -259,6 +284,7 @@ def infer_qudt(unit: Any) -> dict[str, str]: "lpcvd", "pvd", "ald", "sputter", "sputtering", "evaporator", "evaporation", "furnace", "anneal", "annealing", "oxidation", "lithography", "litho", "aligner", "implant", "implanter", "diffusion", "coater", "dicing", "bonder", + "hdpcvd", "drie", } _SENSOR_KEYWORDS = { "metrology", "profilometer", "ellipsometer", "ellipsometry", "microscope", @@ -284,6 +310,29 @@ def infer_sosa_type(equipment_type: Any) -> tuple[str | None, str]: return None, f"equipment_type {equipment_type!r} not recognized" +# Equipment types that carry no information; fall back to the equipment name. +_GENERIC_EQUIPMENT_TYPES = {"", "other", "unknown", "n/a", "none"} + + +def infer_sosa_type_for_equipment( + equipment_type: Any, names: list[Any] +) -> tuple[str | None, str]: + """Like infer_sosa_type, but when the type is blank or generic ("other") + try the equipment's display name and id (e.g. "tube_furnace_1").""" + type_text = equipment_type.strip().lower() if isinstance(equipment_type, str) else "" + if type_text not in _GENERIC_EQUIPMENT_TYPES: + return infer_sosa_type(equipment_type) + for name in names: + if not isinstance(name, str) or not name.strip(): + continue + sosa_type, _ = infer_sosa_type(name.replace("_", " ")) + if sosa_type: + return sosa_type, f"equipment name {name!r} (equipment_type {equipment_type!r})" + shown = equipment_type if type_text else "no equipment_type" + return None, (f"equipment_type {shown!r} not recognized and the name gives no hint" + if type_text else "no equipment_type and the name gives no hint") + + def _mark_inferred(item: dict[str, Any], field: str) -> None: inferred = item.setdefault("ontology_inferred", []) if field not in inferred: @@ -297,6 +346,7 @@ def _annotate_variable( name: str, direction: str | None, report: dict[str, Any], + declared_type: str = "", ) -> None: label = f"{section}:{name}" unit = item.get("unit", "") @@ -317,6 +367,18 @@ def _annotate_variable( missing = [f for f in ("qudt_unit", "qudt_quantity_kind") if not item.get(f)] if not missing: return + value_type = str(item.get("type") or declared_type or "").strip().lower() + if value_type and value_type not in NUMERIC_TYPES: + return # text, date, boolean, file: not a physical quantity + unit_text = _normalize_unit_text(unit) if isinstance(unit, str) else "" + if unit_text.lower() in NO_UNIT_MARKERS: + kind = "numeric field" if value_type else "field with no declared type" + report["unresolved"].append({ + "item": label, "field": "qudt_unit", + "reason": (f"{kind} has unit {unit!r}; enter its real unit, 'unitless', " + "or declare it as text"), + }) + return inferred = infer_qudt(unit) for field in missing: if field in inferred: @@ -324,7 +386,6 @@ def _annotate_variable( _mark_inferred(item, field) report["filled"].append({"item": label, "field": field, "value": inferred[field]}) else: - unit_text = _normalize_unit_text(unit) if isinstance(unit, str) else "" if not unit_text: reason = "no unit entered" elif unit_text in AMBIGUOUS_UNITS: @@ -333,6 +394,8 @@ def _annotate_variable( reason = f"unit {unit!r} not recognized" if field == "qudt_quantity_kind" and "qudt_unit" in inferred: reason = f"unit {unit!r} does not identify a quantity kind" + if "\ufffd" in unit_text: + reason = f"unit {unit!r} is corrupted (character encoding); re-enter it" if field == "qudt_unit" and _quantity_kind_only(unit): # QUDT itself has no unit for this (e.g. nm/min); the export # carries the quantity kind plus unitText, so nothing to ask. @@ -350,13 +413,15 @@ def infer_ontology_annotations(config: dict[str, Any]) -> tuple[dict[str, Any], annotated = copy.deepcopy(config) if isinstance(config, dict) else {} report: dict[str, Any] = {"filled": [], "unresolved": [], "warnings": []} + # Workflow-example equipment store ``domain`` as the bare domain id. domain = annotated.get("domain") if not isinstance(domain, dict): - domain = {} + domain = {"id": domain} if isinstance(domain, str) and domain else {} annotated["domain"] = domain if not (domain.get("sosa_type") or annotated.get("sosa_type")): equipment_type = domain.get("equipment_type") or annotated.get("equipment_type") - sosa_type, reason = infer_sosa_type(equipment_type) + names = [domain.get("display_name"), annotated.get("equipment_name"), domain.get("id")] + sosa_type, reason = infer_sosa_type_for_equipment(equipment_type, names) if sosa_type: domain["sosa_type"] = sosa_type _mark_inferred(domain, "sosa_type") @@ -364,16 +429,34 @@ def infer_ontology_annotations(config: dict[str, Any]) -> tuple[dict[str, Any], else: report["unresolved"].append({"item": "domain", "field": "sosa_type", "reason": reason}) + # Value types are declared on columns/parameters/outputs; features and + # targets refer to those by name. + declared_types: dict[str, str] = {} + for section in ("columns", "parameters", "outputs"): + for item in annotated.get(section) or []: + if isinstance(item, dict) and item.get("type"): + name = item.get("name") or item.get("id") + if name: + declared_types.setdefault(str(name), str(item["type"])) + for feature in annotated.get("features") or []: if isinstance(feature, dict) and feature.get("id"): _annotate_variable(feature, section="feature", name=str(feature["id"]), - direction="input", report=report) + direction="input", report=report, + declared_type=declared_types.get(str(feature["id"]), "")) + # Targets are {"primary": {...}, "secondary": {...}} for wizard-registered + # equipment; workflow examples list output names instead, and those + # outputs are annotated below. + targets = annotated.get("targets") for key in ("primary", "secondary"): - target = (annotated.get("targets") or {}).get(key) + target = targets.get(key) if isinstance(targets, dict) else None if isinstance(target, dict) and target.get("id"): + if target["id"] in PLACEHOLDER_TARGET_IDS and not str(target.get("unit") or "").strip(): + continue # owner configured no target; nothing to annotate _annotate_variable(target, section=f"target.{key}", name=str(target["id"]), - direction=None, report=report) + direction=None, report=report, + declared_type=declared_types.get(str(target["id"]), "")) # fair_archiver already treats registered parameters as inputs and # outputs as outputs, so only their units need annotating. diff --git a/api/scripts/fair_ontology_dry_run.py b/api/scripts/fair_ontology_dry_run.py index 6f21822..e723e62 100644 --- a/api/scripts/fair_ontology_dry_run.py +++ b/api/scripts/fair_ontology_dry_run.py @@ -15,6 +15,9 @@ python api/scripts/fair_ontology_dry_run.py --from-db python api/scripts/fair_ontology_dry_run.py --json + + # Write the report to a file instead of the terminal + python api/scripts/fair_ontology_dry_run.py --from-db --json --output report.json """ from __future__ import annotations @@ -71,7 +74,15 @@ def load_configs_from_db() -> list[dict]: def build_report(entries: list[dict]) -> list[dict]: results = [] for entry in entries: - _, report = infer_ontology_annotations(entry["config"]) + try: + _, report = infer_ontology_annotations(entry["config"]) + except Exception as exc: # report the bad config, keep going + report = { + "filled": [], + "unresolved": [], + "warnings": [{"item": "config", "field": "-", + "reason": f"inference failed: {type(exc).__name__}: {exc}"}], + } results.append( { "equipment": entry["equipment"], @@ -114,11 +125,17 @@ def main() -> int: help="read equipment_metadata.config_json (read-only session)") parser.add_argument("--configs-dir", type=Path, default=DEFAULT_CONFIGS_DIR) parser.add_argument("--json", action="store_true", help="print JSON instead of text") + parser.add_argument("--output", type=Path, help="write the report to this file") args = parser.parse_args() entries = load_configs_from_db() if args.from_db else load_configs_from_dir(args.configs_dir) results = build_report(entries) - print(json.dumps(results, indent=2) if args.json else render_text(results)) + rendered = json.dumps(results, indent=2) if args.json else render_text(results) + if args.output: + args.output.write_text(rendered + "\n", encoding="utf-8") + print(f"Report for {len(results)} equipment written to {args.output}") + else: + print(rendered) return 0 diff --git a/api/tests/test_fair_ontology.py b/api/tests/test_fair_ontology.py index 3a58be0..16cd951 100644 --- a/api/tests/test_fair_ontology.py +++ b/api/tests/test_fair_ontology.py @@ -53,8 +53,19 @@ def test_ambiguous_and_unknown_units_are_not_guessed(self): self.assertEqual(infer_qudt(""), {}) self.assertEqual(infer_qudt(None), {}) - def test_percent_has_unit_but_no_quantity_kind(self): - self.assertEqual(infer_qudt("%"), {"qudt_unit": QUDT_UNIT_BASE + "PERCENT"}) + def test_dimensionless_units(self): + self.assertEqual( + infer_qudt("%"), + {"qudt_unit": QUDT_UNIT_BASE + "PERCENT", + "qudt_quantity_kind": QUDT_QK_BASE + "DimensionlessRatio"}, + ) + self.assertEqual(infer_qudt("count")["qudt_quantity_kind"], QUDT_QK_BASE + "Count") + self.assertEqual(infer_qudt("unitless")["qudt_unit"], QUDT_UNIT_BASE + "UNITLESS") + self.assertEqual(infer_qudt("ratio")["qudt_unit"], QUDT_UNIT_BASE + "UNITLESS") + + def test_sub_second_times(self): + self.assertEqual(infer_qudt("µs")["qudt_unit"], QUDT_UNIT_BASE + "MicroSEC") + self.assertEqual(infer_qudt("ps")["qudt_unit"], QUDT_UNIT_BASE + "PicoSEC") def test_etch_rate_has_quantity_kind_only(self): # qudt.org has no NanoM-PER-MIN unit individual. @@ -74,6 +85,19 @@ def test_unknown_or_conflicting_types_are_not_guessed(self): self.assertIsNone(infer_sosa_type("Etch + Inspection")[0]) self.assertIsNone(infer_sosa_type("Semiconductor tool")[0]) # "sem" only as a word + def test_generic_type_falls_back_to_equipment_name(self): + from fair_ontology import infer_sosa_type_for_equipment + + self.assertEqual( + infer_sosa_type_for_equipment("other", ["Tube Furnace 1", "tube_furnace_1_wl"])[0], + SOSA_BASE + "Actuator", + ) + self.assertEqual(infer_sosa_type_for_equipment("", [None, "fiji200_ald"])[0], + SOSA_BASE + "Actuator") + # A specific type is never overridden by the name. + self.assertIsNone(infer_sosa_type_for_equipment("Glovebox", ["furnace"])[0]) + self.assertIsNone(infer_sosa_type_for_equipment("other", ["Thing 7"])[0]) + class InferOntologyAnnotationsTests(unittest.TestCase): def registered_config(self): @@ -116,7 +140,6 @@ def test_fills_missing_values_and_marks_them_inferred(self): unresolved = {(item["item"], item["field"]) for item in report["unresolved"]} self.assertIn(("feature:mystery", "qudt_unit"), unresolved) - self.assertIn(("target.secondary:uniformity", "qudt_quantity_kind"), unresolved) self.assertIn(("output:rate", "qudt_unit"), unresolved) # The input is left untouched. @@ -165,6 +188,32 @@ def test_unit_without_qudt_individual_is_not_owner_work(self): self.assertNotIn("qudt_unit", annotated["features"][0]) self.assertEqual([u for u in report["unresolved"] if u["item"] == "feature:rate"], []) + def test_non_numeric_fields_and_placeholders_are_skipped(self): + config = { + "domain": {"id": "furnace"}, + "parameters": [ + {"name": "Recipe Name", "unit": "N/A", "type": "string"}, + {"name": "Start Date", "unit": "unitless", "type": "datetime"}, + {"name": "Heating Rate", "unit": "N/A", "type": "float"}, + ], + "features": [{"id": "Recipe Name", "unit": "N/A"}], + "targets": {"primary": {"id": "primary_metric", "unit": ""}}, + } + annotated, report = infer_ontology_annotations(config) + + unresolved = [(u["item"], u["field"]) for u in report["unresolved"] if u["item"] != "domain"] + # Only the numeric field marked N/A needs the owner. + self.assertEqual(unresolved, [("parameter:Heating Rate", "qudt_unit")]) + self.assertNotIn("qudt_unit", annotated["parameters"][0]) + self.assertNotIn("qudt_unit", annotated["features"][0]) + + def test_corrupted_unit_is_reported_as_such(self): + config = {"outputs": [{"name": "Pulse Period", "unit": "\ufffds", "type": "float"}]} + _, report = infer_ontology_annotations(config) + reasons = [u["reason"] for u in report["unresolved"] if u["item"] == "output:Pulse Period"] + self.assertTrue(reasons) + self.assertTrue(all("corrupted" in r for r in reasons)) + def test_tolerates_empty_and_malformed_configs(self): for config in ({}, {"features": None, "targets": None}, {"features": ["x", 3]}): _, report = infer_ontology_annotations(config) @@ -212,6 +261,29 @@ def test_run_jsonld_for_registered_equipment_carries_qudt_and_sosa(self): self.assertIn(QUDT_UNIT_BASE + "NanoM", text) self.assertIn(SOSA_BASE + "Actuator", text) + def test_workflow_example_config_shape_exports(self): + # Shape written by metadata_pg._ensure_workflow_example_equipment. + import fair_archiver + + config = { + "domain": "wf_profilometer", + "equipment_type": "Profilometer", + "parameters": [{"name": "scan_length", "unit": "um", "type": "float"}], + "outputs": [{"name": "step_height", "unit": "nm", "type": "float"}], + "features": ["scan_length"], + "targets": ["step_height"], + } + annotated, _ = infer_ontology_annotations(config) + self.assertEqual(annotated["domain"]["id"], "wf_profilometer") + self.assertEqual(annotated["domain"]["sosa_type"], SOSA_BASE + "Sensor") + + prepared = fair_archiver._prepare_domain_config(config) + mapping = fair_archiver._get_feature_ontology_map(prepared) + self.assertEqual(mapping["scan_length"]["qudt_unit"], QUDT_UNIT_BASE + "MicroM") + self.assertEqual(mapping["scan_length"]["prov_direction"], "input") + self.assertEqual(mapping["step_height"]["qudt_unit"], QUDT_UNIT_BASE + "NanoM") + self.assertEqual(mapping["step_height"]["prov_direction"], "output") + def test_inference_failure_falls_back_to_normalized_config(self): from unittest.mock import patch From d992bcac07576ee23bfa353be09c7e72058cfc66 Mon Sep 17 00:00:00 2001 From: navidgh67 Date: Thu, 8 Oct 2026 21:56:20 -0400 Subject: [PATCH 4/5] =?UTF-8?q?Map=20=C2=B0C/min,=20keep=20unit=20case=20s?= =?UTF-8?q?trict,=20and=20mark=20inferred=20terms=20in=20exports?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add QUDT DEG_C-PER-MIN (TemperaturePerTime) for the registration form's °C/min option. - Match units case-insensitively only for an explicit list of spellings, so "Mbar" (megabar) is no longer read as millibar. - Carry ontology_inferred into the JSON-LD and RO-Crate exports as a schema.org additionalProperty, so inferred terms can be told apart from owner-entered ones. Co-Authored-By: Claude Opus 5.5 (1M context) --- api/fair_archiver.py | 35 +++++++++++++++++++-- api/fair_ontology.py | 18 ++++++++--- api/tests/test_fair_ontology.py | 54 +++++++++++++++++++++++++++++++++ 3 files changed, 100 insertions(+), 7 deletions(-) diff --git a/api/fair_archiver.py b/api/fair_archiver.py index 328d801..d1fc42f 100644 --- a/api/fair_archiver.py +++ b/api/fair_archiver.py @@ -244,7 +244,22 @@ def _normalize_pg_domain_config(config: dict[str, Any]) -> dict[str, Any]: return normalized -def _get_feature_ontology_map(domain_config: dict[str, Any] | None) -> dict[str, dict[str, str]]: +def _inference_note(fields: Any) -> dict[str, Any]: + """Mark ontology terms the platform inferred (fair_ontology) rather than + the owner entering them, so consumers can tell the two apart.""" + if not isinstance(fields, list) or not fields: + return {} + return { + "additionalProperty": { + "@type": "PropertyValue", + "name": "ontologyInferred", + "description": "Ontology terms inferred by the platform from the registered unit and equipment type, not entered by the owner.", + "value": [str(field) for field in fields], + } + } + + +def _get_feature_ontology_map(domain_config: dict[str, Any] | None) -> dict[str, dict[str, Any]]: """ Build a mapping from feature/target IDs to their QUDT ontology URIs and PROV-O direction. @@ -253,7 +268,7 @@ def _get_feature_ontology_map(domain_config: dict[str, Any] | None) -> dict[str, {"Etch_AvgO2Flow": {"qudt_quantity_kind": "...", "qudt_unit": "...", "unit": "sccm", "prov_direction": "input"}, ...} """ - mapping: dict[str, dict[str, str]] = {} + mapping: dict[str, dict[str, Any]] = {} if not domain_config: return mapping @@ -264,11 +279,13 @@ def _get_feature_ontology_map(domain_config: dict[str, Any] | None) -> dict[str, continue fid = feature.get("id", "") if fid: - entry: dict[str, str] = {"unit": feature.get("unit", "")} + entry: dict[str, Any] = {"unit": feature.get("unit", "")} if feature.get("qudt_quantity_kind"): entry["qudt_quantity_kind"] = feature["qudt_quantity_kind"] if feature.get("qudt_unit"): entry["qudt_unit"] = feature["qudt_unit"] + if feature.get("ontology_inferred"): + entry["ontology_inferred"] = list(feature["ontology_inferred"]) if feature.get("prov_direction"): entry["prov_direction"] = feature["prov_direction"] mapping[fid] = entry @@ -285,6 +302,8 @@ def _get_feature_ontology_map(domain_config: dict[str, Any] | None) -> dict[str, entry["qudt_quantity_kind"] = target["qudt_quantity_kind"] if target.get("qudt_unit"): entry["qudt_unit"] = target["qudt_unit"] + if target.get("ontology_inferred"): + entry["ontology_inferred"] = list(target["ontology_inferred"]) mapping[tid] = entry # Registered input parameters (set-points captured at registration; always @@ -301,6 +320,8 @@ def _get_feature_ontology_map(domain_config: dict[str, Any] | None) -> dict[str, entry["qudt_quantity_kind"] = parameter["qudt_quantity_kind"] if parameter.get("qudt_unit"): entry["qudt_unit"] = parameter["qudt_unit"] + if parameter.get("ontology_inferred"): + entry["ontology_inferred"] = list(parameter["ontology_inferred"]) mapping.setdefault(str(pid), entry) # Registered outputs (measured results captured at registration; always output) @@ -315,6 +336,8 @@ def _get_feature_ontology_map(domain_config: dict[str, Any] | None) -> dict[str, entry["qudt_quantity_kind"] = output["qudt_quantity_kind"] if output.get("qudt_unit"): entry["qudt_unit"] = output["qudt_unit"] + if output.get("ontology_inferred"): + entry["ontology_inferred"] = list(output["ontology_inferred"]) mapping.setdefault(str(oid), entry) return mapping @@ -566,6 +589,7 @@ def _build_rocrate( **({ "sameAs": {"@id": equipment_ontology_uri} } if equipment_ontology_uri else {}), + **_inference_note(domain_info.get("ontology_inferred")), }, ] @@ -587,6 +611,7 @@ def _build_rocrate( entity["prov:qualifiedUsage"] = "prov:used" elif direction == "output": entity["prov:qualifiedGeneration"] = "prov:generated" + entity.update(_inference_note(onto.get("ontology_inferred"))) graph.append(entity) graph.extend(file_entities) @@ -902,6 +927,7 @@ def run_to_jsonld(run: dict[str, Any]) -> dict[str, Any]: prop["prov:qualifiedUsage"] = "prov:used" elif direction == "output": prop["prov:qualifiedGeneration"] = "prov:generated" + prop.update(_inference_note(onto.get("ontology_inferred"))) variables_measured.append(prop) # Build ontology-enriched result list. The canonical etcher uses typed @@ -924,6 +950,7 @@ def run_to_jsonld(run: dict[str, Any]) -> dict[str, Any]: if onto.get("qudt_unit"): result_prop["qudt:hasUnit"] = {"@id": onto["qudt_unit"]} result_prop["prov:qualifiedGeneration"] = "prov:generated" + result_prop.update(_inference_note(onto.get("ontology_inferred"))) results.append(result_prop) else: for target_id, target_label, target_key in [ @@ -943,6 +970,7 @@ def run_to_jsonld(run: dict[str, Any]) -> dict[str, Any]: if onto.get("qudt_unit"): result_prop["qudt:hasUnit"] = {"@id": onto["qudt_unit"]} result_prop["prov:qualifiedGeneration"] = "prov:generated" + result_prop.update(_inference_note(onto.get("ontology_inferred"))) results.append(result_prop) # Build equipment entity with SAREF metadata @@ -961,6 +989,7 @@ def run_to_jsonld(run: dict[str, Any]) -> dict[str, Any]: equipment_entity["s4inma:hasSerialNumber"] = domain_info["serial_number"] if domain_info.get("location"): equipment_entity["location"] = domain_info["location"] + equipment_entity.update(_inference_note(domain_info.get("ontology_inferred"))) # Build author entity with schema:Person and optional ORCID author_entity: dict[str, Any] = { diff --git a/api/fair_ontology.py b/api/fair_ontology.py index 28a40e1..bad2d62 100644 --- a/api/fair_ontology.py +++ b/api/fair_ontology.py @@ -53,6 +53,7 @@ "MicroM-PER-MIN": "LinearVelocity", "MicroM-PER-SEC": "LinearVelocity", "DEG_C": "CelsiusTemperature", + "DEG_C-PER-MIN": "TemperaturePerTime", "K": "ThermodynamicTemperature", "SEC": "Time", "MilliSEC": "Time", @@ -136,6 +137,8 @@ "°C": "DEG_C", "degC": "DEG_C", "celsius": "DEG_C", + "°C/min": "DEG_C-PER-MIN", + "degC/min": "DEG_C-PER-MIN", "K": "K", "kelvin": "K", "s": "SEC", @@ -202,13 +205,20 @@ "a": "ampere or an ASCII stand-in for angstrom", } -# Short symbols are case-sensitive (mW vs MW, mHz vs MHz, s vs S), so only -# words and these unambiguous abbreviations match case-insensitively. -_CASE_INSENSITIVE_SHORT = {"slm", "lpm", "psi", "atm", "bar", "rpm", "sec", "min", "ohm", "deg"} +# Unit symbols are case-sensitive (mW vs MW, mbar vs Mbar, s vs S), so only +# these spellings, where case cannot change the meaning, also match in any +# case. Everything else must match exactly. +_CASE_INSENSITIVE_ALIASES = { + "sccm", "slm", "slpm", "lpm", "psi", "atm", "torr", "millitorr", "watt", "watts", + "micron", "microns", "angstrom", "angstroms", "celsius", "kelvin", "sec", + "seconds", "min", "minutes", "hr", "hours", "volt", "volts", "amp", "amps", + "rpm", "ohm", "ohms", "deg", "degrees", "percent", "count", "counts", + "unitless", "dimensionless", "ratio", "newton meter", +} def _case_insensitive_ok(alias: str) -> bool: - return len(alias) >= 4 or alias in _CASE_INSENSITIVE_SHORT + return alias.lower() in _CASE_INSENSITIVE_ALIASES def _normalize_unit_text(unit: str) -> str: diff --git a/api/tests/test_fair_ontology.py b/api/tests/test_fair_ontology.py index 16cd951..18ab7d5 100644 --- a/api/tests/test_fair_ontology.py +++ b/api/tests/test_fair_ontology.py @@ -46,6 +46,25 @@ def test_case_matters_for_short_symbols(self): self.assertEqual(infer_qudt("S"), {}) # siemens, not seconds self.assertEqual(infer_qudt("SCCM")["qudt_unit"], QUDT_UNIT_BASE + "CentiM3-PER-MIN") + def test_case_never_changes_the_scale_of_a_prefixed_unit(self): + self.assertEqual(infer_qudt("mbar")["qudt_unit"], QUDT_UNIT_BASE + "MilliBAR") + self.assertEqual(infer_qudt("Mbar"), {}) # megabar, not millibar + self.assertEqual(infer_qudt("MTORR"), {}) + self.assertEqual(infer_qudt("MV"), {}) # megavolt, not millivolt + self.assertEqual(infer_qudt("TORR")["qudt_unit"], QUDT_UNIT_BASE + "TORR") + self.assertEqual(infer_qudt("torr")["qudt_unit"], QUDT_UNIT_BASE + "TORR") + self.assertEqual(infer_qudt("Angstroms")["qudt_unit"], QUDT_UNIT_BASE + "ANGSTROM") + + def test_registration_unit_options(self): + # Units offered by the equipment registration form. + self.assertEqual( + infer_qudt("°C/min"), + {"qudt_unit": QUDT_UNIT_BASE + "DEG_C-PER-MIN", + "qudt_quantity_kind": QUDT_QK_BASE + "TemperaturePerTime"}, + ) + self.assertEqual(infer_qudt("h")["qudt_unit"], QUDT_UNIT_BASE + "HR") + self.assertEqual(infer_qudt("°")["qudt_unit"], QUDT_UNIT_BASE + "DEG") + def test_ambiguous_and_unknown_units_are_not_guessed(self): self.assertEqual(infer_qudt("A"), {}) self.assertEqual(infer_qudt("C"), {}) @@ -261,6 +280,41 @@ def test_run_jsonld_for_registered_equipment_carries_qudt_and_sosa(self): self.assertIn(QUDT_UNIT_BASE + "NanoM", text) self.assertIn(SOSA_BASE + "Actuator", text) + def test_exports_mark_inferred_terms(self): + import tempfile + from unittest.mock import patch + + import fair_archiver + + config = dict(self.REGISTERED, features=[ + {"id": "power", "unit": "W"}, + {"id": "ramp", "unit": "°C/min", "qudt_unit": QUDT_UNIT_BASE + "DEG_C-PER-MIN", + "qudt_quantity_kind": QUDT_QK_BASE + "TemperaturePerTime", "prov_direction": "input"}, + ]) + prepared = fair_archiver._prepare_domain_config(config) + + def note(entity): + return entity.get("additionalProperty", {}).get("value") + + run = {"equipment_id": "sputter1", "equipment_name": "Sputter 1", + "features": {"power": 150.0, "ramp": 5.0}, "outputs": {"thickness": 42.0}} + with patch.object(fair_archiver, "_load_domain_config", return_value=prepared): + jsonld = fair_archiver.run_to_jsonld(run) + with tempfile.TemporaryDirectory() as tmp: + crate = fair_archiver._build_rocrate( + runs_data=[run], project_meta={"equipment_id": "sputter1"}, + fimap_files=[], crate_dir=Path(tmp) / "crate") + manifest = json.loads((crate / "ro-crate-metadata.json").read_text()) + + variables = {p["name"]: p for p in jsonld["variableMeasured"]} + self.assertEqual(sorted(note(variables["power"])), ["prov_direction", "qudt_quantity_kind", "qudt_unit"]) + self.assertIsNone(note(variables["ramp"])) # owner-entered: no marker + + graph = {node["@id"]: node for node in manifest["@graph"]} + self.assertEqual(sorted(note(graph["#param_power"])), ["prov_direction", "qudt_quantity_kind", "qudt_unit"]) + self.assertIsNone(note(graph["#param_ramp"])) + self.assertEqual(note(graph["#equipment_sputter1"]), ["sosa_type"]) + def test_workflow_example_config_shape_exports(self): # Shape written by metadata_pg._ensure_workflow_example_equipment. import fair_archiver From b22bddd18d15f8d04473b2ef5fdb1b81ace93642 Mon Sep 17 00:00:00 2001 From: navidgh67 Date: Thu, 8 Oct 2026 21:59:21 -0400 Subject: [PATCH 5/5] Match quantity-kind-only units exactly so Nm/s is not read as nm/s Co-Authored-By: Claude Opus 5.5 (1M context) --- api/fair_ontology.py | 6 +++--- api/tests/test_fair_ontology.py | 7 +++++++ 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/api/fair_ontology.py b/api/fair_ontology.py index bad2d62..55577a7 100644 --- a/api/fair_ontology.py +++ b/api/fair_ontology.py @@ -240,14 +240,13 @@ def _build_case_insensitive_aliases() -> dict[str, str]: _ALIASES_CI = _build_case_insensitive_aliases() -_QUANTITY_KIND_ONLY_CI = {key.lower(): qk for key, qk in QUANTITY_KIND_ONLY.items()} def _quantity_kind_only(unit: Any) -> bool: if not isinstance(unit, str): return False text = _normalize_unit_text(unit) - return text in QUANTITY_KIND_ONLY or text.lower() in _QUANTITY_KIND_ONLY_CI + return text in QUANTITY_KIND_ONLY def infer_qudt(unit: Any) -> dict[str, str]: @@ -281,7 +280,8 @@ def infer_qudt(unit: Any) -> dict[str, str]: result["qudt_quantity_kind"] = QUDT_QK_BASE + quantity_kind return result - quantity_kind = QUANTITY_KIND_ONLY.get(text) or _QUANTITY_KIND_ONLY_CI.get(text.lower()) + # Exact match only: these are prefixed symbols (nm/s vs Nm/s). + quantity_kind = QUANTITY_KIND_ONLY.get(text) if quantity_kind: return {"qudt_quantity_kind": QUDT_QK_BASE + quantity_kind} return {} diff --git a/api/tests/test_fair_ontology.py b/api/tests/test_fair_ontology.py index 18ab7d5..88b847c 100644 --- a/api/tests/test_fair_ontology.py +++ b/api/tests/test_fair_ontology.py @@ -54,6 +54,8 @@ def test_case_never_changes_the_scale_of_a_prefixed_unit(self): self.assertEqual(infer_qudt("TORR")["qudt_unit"], QUDT_UNIT_BASE + "TORR") self.assertEqual(infer_qudt("torr")["qudt_unit"], QUDT_UNIT_BASE + "TORR") self.assertEqual(infer_qudt("Angstroms")["qudt_unit"], QUDT_UNIT_BASE + "ANGSTROM") + self.assertEqual(infer_qudt("nm/s"), {"qudt_quantity_kind": QUDT_QK_BASE + "LinearVelocity"}) + self.assertEqual(infer_qudt("Nm/s"), {}) # newton metres per second, not nm/s def test_registration_unit_options(self): # Units offered by the equipment registration form. @@ -207,6 +209,11 @@ def test_unit_without_qudt_individual_is_not_owner_work(self): self.assertNotIn("qudt_unit", annotated["features"][0]) self.assertEqual([u for u in report["unresolved"] if u["item"] == "feature:rate"], []) + def test_miscased_compound_unit_is_reported(self): + _, report = infer_ontology_annotations({"features": [{"id": "x", "unit": "Nm/s"}]}) + self.assertIn(("feature:x", "qudt_unit"), + {(entry["item"], entry["field"]) for entry in report["unresolved"]}) + def test_non_numeric_fields_and_placeholders_are_skipped(self): config = { "domain": {"id": "furnace"},