Infer QUDT/SOSA/PROV annotations for FAIR exports

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
@@ -204,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",
@@ -222,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.
@@ -231,33 +268,42 @@ 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
 
-    # 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", "")}
+            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
 
-    # 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"}
             if target.get("qudt_quantity_kind"):
                 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
@@ -274,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)
@@ -288,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
@@ -539,6 +589,7 @@ def _build_rocrate(
             **({
                 "sameAs": {"@id": equipment_ontology_uri}
             } if equipment_ontology_uri else {}),
+            **_inference_note(domain_info.get("ontology_inferred")),
         },
     ]
 
@@ -560,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)
@@ -875,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
@@ -897,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 [
@@ -916,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
@@ -934,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] = {