Observation builder

main.py

@@ -62,7 +62,7 @@ def run_pipeline(preprocessing_input: Path, base_root: Path, trial: str = "IMPRE
         meta=_meta,
     )
 
-    print(f"Harmonized: {harmonized_result.patients[0:10]}")
+    # print(f"Harmonized: {harmonized_result.patients[0:10]}")
 
     # run semantic mapping
     semantic_mapper = SemanticService(outdir=base_root, layout=Layout.TRIAL_TIMESTAMP_RUN)
@@ -89,7 +89,7 @@ def run_pipeline(preprocessing_input: Path, base_root: Path, trial: str = "IMPRE
     tables: OmopTables = omop_service.build(harmonized_result.patients)
 
     # todo: remove
-    print(f"built tables: {tables}")
+    # print(f"built tables: {tables}")
 
     # export concept lookup tracking (missed lookups, coverage stats)
     concept_service.export(formats="csv")

src/omop_etl/concept_mapping/core/models.py

@@ -6,7 +6,7 @@
 
 
 def _norm(v: str | None) -> str:
-    """Lowercase + strip a CSV value, defaulting None to empty string."""
+    """Lowercase and strip a CSV value, defaulting None to empty string."""
     return (v or "").lower().strip()
 
 

src/omop_etl/concept_mapping/service.py

@@ -157,7 +157,7 @@ def lookup_static(
             validity=c.validity,
         )
         if not _concept_matches_filter(concept, domains, vocabs, validity):
-            self._result.record_miss("static", value_set, local_value)
+            # concept mapped but rejected by filter
             return None
 
         self._result.record_match("static", value_set, local_value, concept)
@@ -198,7 +198,7 @@ def lookup_structural(
             validity=c.validity,
         )
         if not _concept_matches_filter(concept, domains, vocabs, validity):
-            self._result.record_miss("structural", value_set, "")
+            # concept mapped but rejected by filter
             return None
 
         self._result.record_match("structural", value_set, "", concept)

src/omop_etl/harmonization/harmonizers/base.py

@@ -66,6 +66,7 @@ class CollectionSpec(SpecBase):
     order_by: tuple[str, ...] = ()
     require_order_by: bool = False
     items_col: str = "items"
+    on_natural_key_conflict: Literal["error", "warn"] = "warn"
 
 
 # union type for all specs
@@ -84,6 +85,37 @@ def _derived_name(fn: Callable[..., Any]) -> str:
     return name.removeprefix("_process_")
 
 
+def _check_natural_key_conflicts(
+    objs: list[DomainBase],
+    *,
+    patient_id: str,
+    item_type: type[DomainBase],
+    policy: Literal["error", "warn"],
+) -> None:
+    """
+    Detect natural-key collisions where the rows have differing data.
+
+    Identical duplicates (same NK, same data) are assumed to be deduplicated
+    upstream by the collection processor, so this only flags conflicts.
+    Keeps the first occurrence.
+    """
+    seen: dict[tuple, DomainBase] = {}
+    fields = item_type.data_fields()
+    for obj in objs:
+        nk = obj.natural_key()
+        prior = seen.get(nk)
+        if prior is None:
+            seen[nk] = obj
+            continue
+        if all(getattr(prior, f) == getattr(obj, f) for f in fields):
+            continue
+        diffs = {f: (getattr(prior, f), getattr(obj, f)) for f in fields if getattr(prior, f) != getattr(obj, f)}
+        msg = f"{item_type.__name__} natural-key conflict for patient {patient_id}: NK={nk} has conflicting values: {diffs}"
+        if policy == "error":
+            raise ValueError(msg)
+        log.warning(msg)
+
+
 def scalar(
     *,
     name: str | None = None,
@@ -166,6 +198,7 @@ def collection(
     skip_missing_patients: bool = False,
     subject_col: str = "SubjectId",
     strict_schema: bool | None = None,
+    on_natural_key_conflict: Literal["error", "warn"] = "warn",
 ) -> Callable[[_F], _F]:
     """
     Decorator: register a method as a collection-domain processor.
@@ -187,6 +220,7 @@ def decorator(fn: _F) -> _F:
             skip_missing_patients=skip_missing_patients,
             subject_col=subject_col,
             strict_schema=strict_schema,
+            on_natural_key_conflict=on_natural_key_conflict,
         )
         setattr(fn, _SPEC_ATTR, spec)
         return fn
@@ -418,6 +452,7 @@ def _run_spec(self, spec: ProcessorSpec) -> None:
                     items_col=spec.items_col,
                     skip_missing_patients=spec.skip_missing_patients,
                     mode=spec.mode,
+                    on_natural_key_conflict=spec.on_natural_key_conflict,
                 )
 
             elif isinstance(spec, SingletonSpec):
@@ -595,6 +630,7 @@ def hydrate_collection_field(
         item_type: type[DomainBase],
         patients: dict[str, Patient],
         mode: Literal["replace", "extend"] = "replace",
+        on_natural_key_conflict: Literal["error", "warn"] = "warn",
     ) -> None:
         """
         Instantiate collection domain models onto Patient after schema validation.
@@ -611,6 +647,9 @@ def hydrate_collection_field(
            item_type: Target domain class (used to resolve Patient attribute).
            patients: Map of patient_id to Patient instance.
            mode: "replace" overwrites, "extend" appends to existing collection.
+           on_natural_key_conflict: "warn" logs a warning when two instances share a natural key
+               but differ in other field values; "error" raises ValueError. Identical duplicates
+               (same NK, same data) are assumed to be deduplicated upstream.
         """
         target_attr = Patient.get_attr_for_type(item_type)
         build = builder or item_type.from_row
@@ -627,6 +666,14 @@ def hydrate_collection_field(
             except Exception as e:
                 raise ValueError(f"{item_type.__name__} collection hydration failed for {sid=}") from e
 
+            if item_type.NATURAL_KEY_FIELDS:
+                _check_natural_key_conflicts(
+                    objs,
+                    patient_id=sid,
+                    item_type=item_type,
+                    policy=on_natural_key_conflict,
+                )
+
             if mode == "extend":
                 existing = getattr(patient, target_attr, ()) or ()
                 objs = list(existing) + objs