From 9a81a7f7f6fa367f1b5d28d7c3a05c23e8874922 Mon Sep 17 00:00:00 2001 From: Matthew Horridge Date: Mon, 29 Jun 2026 16:05:18 -0700 Subject: [PATCH 1/4] Add Case and Coalesce multi-source combinator primitives Two combinators for "the same quantity arrives in one of several columns, each possibly needing its own conversion": - Case: switch on a selector source; the first branch whose `when` contains the selector value wins. Models data with an authoritative unit/type flag (e.g. a "units: kg/lbs" column). - Coalesce: pick the first branch whose source is non-null. Models "whichever field was filled in". Both address sources BY NAME (the primitive stores its own ordered source-name list and zips it with the positional values transform() receives) rather than by fragile positional index. A branch is composable: either a single source + op-chain, or several `terms` (each source + op-chain) combined with a reduction -- so a branch can e.g. convert feet->inches, take inches as-is, and sum them. Branch op-chains reuse existing primitives (ConvertUnits, Scale, Round, ...) and round-trip through the factory like MapEach's nested ops. 14 new tests (single-source weight kg/lbs flag, multi-term height feet+inches and meters+cm summing, serialization round-trips, null/default edges). Full suite: 196 passing. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../primitives/__init__.py | 4 +- .../primitives/case.py | 169 ++++++++++++++++++ .../primitives/coalesce.py | 101 +++++++++++ .../primitives/factory.py | 6 + .../primitives/vocabulary.py | 2 + tests/test_case_coalesce.py | 159 ++++++++++++++++ 6 files changed, 440 insertions(+), 1 deletion(-) create mode 100644 src/harmonization_framework/primitives/case.py create mode 100644 src/harmonization_framework/primitives/coalesce.py create mode 100644 tests/test_case_coalesce.py diff --git a/src/harmonization_framework/primitives/__init__.py b/src/harmonization_framework/primitives/__init__.py index 9d972d9..7f43daf 100644 --- a/src/harmonization_framework/primitives/__init__.py +++ b/src/harmonization_framework/primitives/__init__.py @@ -1,6 +1,8 @@ from .base import PrimitiveOperation -from .bin_primitive import Bin +from .bin_primitive import Bin +from .case import Case from .cast import Cast +from .coalesce import Coalesce from .dates import ConvertDate from .donothing import DoNothing from .enum2enum import EnumToEnum diff --git a/src/harmonization_framework/primitives/case.py b/src/harmonization_framework/primitives/case.py new file mode 100644 index 0000000..331ab93 --- /dev/null +++ b/src/harmonization_framework/primitives/case.py @@ -0,0 +1,169 @@ +from typing import Any, Dict, List + +from .base import PrimitiveOperation, isnull + + +def _normalize_terms(branch): + """A branch produces one value from one or more 'terms'. Each term computes a + value from one source via its own op-chain; a branch with multiple terms combines + them with `combine` (a Reduction name string, e.g. 'sum'). + + Accepts three authoring forms, normalized to the canonical term list: + - {"source": "x", "operations": [...]} -> one term + - {"sources": ["a","b"], ... } is NOT auto-summed; use explicit terms instead + - {"terms": [{"source": "a", "operations": [...]}, ...], "combine": "sum"} + """ + if "terms" in branch: + terms = [{"source": t["source"], "operations": list(t.get("operations", []))} + for t in branch["terms"]] + combine = branch.get("combine") + else: + terms = [{"source": branch["source"], "operations": list(branch.get("operations", []))}] + combine = None + return terms, combine + + +def _apply_terms(terms, combine, by_name): + """Compute each term's value, then combine. Returns the branch value.""" + from .reduce import Reduce, Reduction + + vals = [] + for t in terms: + current = by_name[t["source"]] + for op in t["operations"]: + current = op(current) + vals.append(current) + if combine is None: + return vals[0] if len(vals) == 1 else vals + return Reduce(Reduction(combine)).transform(vals) + + +def _term_sources(branch): + terms, _ = _normalize_terms(branch) + return [t["source"] for t in terms] + + +class Case(PrimitiveOperation): + """ + Choose one of several branches by switching on a selector source, then compute + that branch's value from one or more source terms. + + A multi-source combinator. The rule's `sources` list names every column this + primitive may read; `transform` receives the values in that same order and + `Case` zips them back to names internally, so branches address sources by NAME, + never by fragile positional index. + + A branch is either a single source with an op-chain, or several `terms` (each a + source + op-chain) combined with a reduction. The first branch whose `when` + contains the selector value wins; null/unmatched selector -> `default`. + + Single-source branch (RADx-UP weight: unit flag selects pounds-as-is or kg->lbs): + + Case( + sources=["weight_units", "weight_lbs", "weight_kgs"], + selector="weight_units", + branches=[ + {"when": ["2"], "source": "weight_lbs", "operations": []}, + {"when": ["1"], "source": "weight_kgs", + "operations": [ConvertUnits(Unit.KILOGRAM, Unit.POUNDS), Round(0)]}, + ], + ) + + Multi-term branch (RADx-UP height: feet+inches OR meters+cm, summed in inches): + + Case( + sources=["height_units", "ft", "in", "m", "cm"], + selector="height_units", + branches=[ + {"when": ["1"], "combine": "sum", "terms": [ + {"source": "ft", "operations": [ConvertUnits(Unit.FEET, Unit.INCH)]}, + {"source": "in", "operations": []}, + ]}, + {"when": ["2"], "combine": "sum", "terms": [ + {"source": "m", "operations": [ConvertUnits(Unit.METER, Unit.INCH)]}, + {"source": "cm", "operations": [ConvertUnits(Unit.CENTIMETER, Unit.INCH)]}, + ]}, + ], + ) + """ + + def __init__(self, sources, selector, branches, default=None): + self.sources = list(sources) + self.selector = selector + self.branches = [] + for b in branches: + terms, combine = _normalize_terms(b) + self.branches.append({"when": [str(w) for w in b["when"]], + "terms": terms, "combine": combine}) + self.default = default + if self.selector not in self.sources: + raise ValueError(f"Case selector {self.selector!r} not in sources {self.sources}") + for b in self.branches: + for t in b["terms"]: + if t["source"] not in self.sources: + raise ValueError(f"Case branch source {t['source']!r} not in sources {self.sources}") + + def __str__(self): + lines = [f"Switch on {self.selector}:"] + for b in self.branches: + srcs = "+".join(t["source"] for t in b["terms"]) + comb = f" ({b['combine']})" if b["combine"] else "" + lines.append(f" when {b['when']} -> {srcs}{comb}") + lines.append(f" else -> {self.default!r}") + return "\n".join(lines) + + def _by_name(self, values): + if not isinstance(values, (list, tuple)): + raise TypeError(f"Case expects a list of source values, got {type(values).__name__}") + if len(values) != len(self.sources): + raise ValueError(f"Case received {len(values)} values but has {len(self.sources)} sources") + return dict(zip(self.sources, values)) + + def transform(self, values: Any) -> Any: + by_name = self._by_name(values) + sel = by_name[self.selector] + if isnull(sel): + return self.default + sel_key = str(sel) + for b in self.branches: + if sel_key in b["when"]: + return _apply_terms(b["terms"], b["combine"], by_name) + return self.default + + def to_dict(self): + return { + "operation": "case", + "sources": list(self.sources), + "selector": self.selector, + "branches": [ + { + "when": list(b["when"]), + "combine": b["combine"], + "terms": [ + {"source": t["source"], "operations": [op.to_dict() for op in t["operations"]]} + for t in b["terms"] + ], + } + for b in self.branches + ], + "default": self.default, + } + + @classmethod + def from_serialization(cls, serialization): + from .factory import deserialize_operation + + branches = [] + for b in serialization["branches"]: + terms = [ + {"source": t["source"], + "operations": [deserialize_operation(op) for op in t.get("operations", [])]} + for t in b["terms"] + ] + branches.append({"when": b["when"], "combine": b.get("combine"), "terms": terms}) + return cls( + sources=serialization["sources"], + selector=serialization["selector"], + branches=branches, + default=serialization.get("default"), + ) diff --git a/src/harmonization_framework/primitives/coalesce.py b/src/harmonization_framework/primitives/coalesce.py new file mode 100644 index 0000000..fd8df1a --- /dev/null +++ b/src/harmonization_framework/primitives/coalesce.py @@ -0,0 +1,101 @@ +from typing import Any + +from .base import PrimitiveOperation, isnull +from .case import _normalize_terms, _apply_terms + + +class Coalesce(PrimitiveOperation): + """ + Pick the first branch whose (primary) source is non-null, then compute that + branch's value from one or more source terms. + + A multi-source combinator, like Case but with the branch chosen by + populated-ness rather than an explicit selector. Use Case when the data carries + an authoritative unit/type flag; use Coalesce when "whichever field was filled + in" is the right rule. + + A branch is a single source with an op-chain, or several `terms` (each a + source + op-chain) combined with a reduction (same shape as Case branches). A + branch is considered populated when its first term's source is non-null. Branch + order defines precedence; all sources null -> `default`. + + Coalesce( + sources=["weight_lbs", "weight_kgs"], + branches=[ + {"source": "weight_lbs", "operations": []}, + {"source": "weight_kgs", + "operations": [ConvertUnits(Unit.KILOGRAM, Unit.POUNDS), Round(0)]}, + ], + ) + """ + + def __init__(self, sources, branches, default=None): + self.sources = list(sources) + self.branches = [] + for b in branches: + terms, combine = _normalize_terms(b) + self.branches.append({"terms": terms, "combine": combine}) + self.default = default + for b in self.branches: + for t in b["terms"]: + if t["source"] not in self.sources: + raise ValueError(f"Coalesce branch source {t['source']!r} not in sources {self.sources}") + + def __str__(self): + lines = ["Coalesce (first non-null):"] + for b in self.branches: + srcs = "+".join(t["source"] for t in b["terms"]) + comb = f" ({b['combine']})" if b["combine"] else "" + lines.append(f" {srcs}{comb}") + lines.append(f" else -> {self.default!r}") + return "\n".join(lines) + + def _by_name(self, values): + if not isinstance(values, (list, tuple)): + raise TypeError(f"Coalesce expects a list of source values, got {type(values).__name__}") + if len(values) != len(self.sources): + raise ValueError(f"Coalesce received {len(values)} values but has {len(self.sources)} sources") + return dict(zip(self.sources, values)) + + def transform(self, values: Any) -> Any: + by_name = self._by_name(values) + for b in self.branches: + primary = by_name[b["terms"][0]["source"]] + if not isnull(primary): + return _apply_terms(b["terms"], b["combine"], by_name) + return self.default + + def to_dict(self): + return { + "operation": "coalesce", + "sources": list(self.sources), + "branches": [ + { + "combine": b["combine"], + "terms": [ + {"source": t["source"], "operations": [op.to_dict() for op in t["operations"]]} + for t in b["terms"] + ], + } + for b in self.branches + ], + "default": self.default, + } + + @classmethod + def from_serialization(cls, serialization): + from .factory import deserialize_operation + + branches = [] + for b in serialization["branches"]: + terms = [ + {"source": t["source"], + "operations": [deserialize_operation(op) for op in t.get("operations", [])]} + for t in b["terms"] + ] + branches.append({"combine": b.get("combine"), "terms": terms}) + return cls( + sources=serialization["sources"], + branches=branches, + default=serialization.get("default"), + ) diff --git a/src/harmonization_framework/primitives/factory.py b/src/harmonization_framework/primitives/factory.py index bf19b2f..e2943bc 100644 --- a/src/harmonization_framework/primitives/factory.py +++ b/src/harmonization_framework/primitives/factory.py @@ -11,7 +11,9 @@ from .base import PrimitiveOperation from .bin_primitive import Bin +from .case import Case from .cast import Cast +from .coalesce import Coalesce from .dates import ConvertDate from .donothing import DoNothing from .enum2enum import EnumToEnum @@ -44,8 +46,12 @@ def deserialize_operation(operation: Dict[str, Any]) -> PrimitiveOperation: match name: case PrimitiveVocabulary.BIN.value: return Bin.from_serialization(operation) + case PrimitiveVocabulary.CASE.value: + return Case.from_serialization(operation) case PrimitiveVocabulary.CAST.value: return Cast.from_serialization(operation) + case PrimitiveVocabulary.COALESCE.value: + return Coalesce.from_serialization(operation) case PrimitiveVocabulary.CONVERT_DATE.value: return ConvertDate.from_serialization(operation) case PrimitiveVocabulary.CONVERT_UNITS.value: diff --git a/src/harmonization_framework/primitives/vocabulary.py b/src/harmonization_framework/primitives/vocabulary.py index c8deb63..999001b 100644 --- a/src/harmonization_framework/primitives/vocabulary.py +++ b/src/harmonization_framework/primitives/vocabulary.py @@ -2,7 +2,9 @@ class PrimitiveVocabulary(Enum): BIN = "bin" + CASE = "case" CAST = "cast" + COALESCE = "coalesce" CONVERT_DATE = "convert_date" CONVERT_UNITS = "convert_units" DO_NOTHING = "do_nothing" diff --git a/tests/test_case_coalesce.py b/tests/test_case_coalesce.py new file mode 100644 index 0000000..7b09f3e --- /dev/null +++ b/tests/test_case_coalesce.py @@ -0,0 +1,159 @@ +"""Tests for the Case and Coalesce multi-source combinator primitives.""" +import pandas as pd + +from harmonization_framework.harmonization_rule import HarmonizationRule +from harmonization_framework.harmonize import harmonize_dataset +from harmonization_framework.primitives import ( + Case, Coalesce, ConvertUnits, DoNothing, Round, Scale) +from harmonization_framework.primitives.units import Unit +from harmonization_framework.rule_registry import RuleSet + + +# --- Case ---------------------------------------------------------------- + +def _weight_case(): + return Case( + sources=["weight_units", "weight_lbs", "weight_kgs"], + selector="weight_units", + branches=[ + {"when": ["2"], "source": "weight_lbs", "operations": [DoNothing()]}, + {"when": ["1"], "source": "weight_kgs", "operations": [Scale(2.20462), Round(0)]}, + ], + default=None, + ) + + +def test_case_selects_pounds_branch_as_is(): + c = _weight_case() + # units=2 (pounds): take weight_lbs unchanged + assert c.transform(["2", 150, 68]) == 150 + + +def test_case_selects_kg_branch_and_converts(): + c = _weight_case() + # units=1 (kilograms): convert weight_kgs -> lbs, rounded + assert c.transform(["1", 150, 68]) == round(68 * 2.20462) + + +def test_case_int_selector_matches_string_when(): + c = _weight_case() + # selector arrives as an int 1 (e.g. pandas) but `when` is "1" + assert c.transform([1, 150, 68]) == round(68 * 2.20462) + + +def test_case_null_selector_returns_default(): + c = _weight_case() + assert c.transform([None, 150, 68]) is None + + +def test_case_unmatched_selector_returns_default(): + c = _weight_case() + assert c.transform(["99", 150, 68]) is None + + +def test_case_serialization_roundtrip(): + rule = HarmonizationRule( + ["weight_units", "weight_lbs", "weight_kgs"], "nih_weight", [_weight_case()] + ) + payload = rule.serialize() + assert payload["operations"][0]["operation"] == "case" + roundtrip = HarmonizationRule.from_serialization(payload) + assert roundtrip.serialize() == payload + assert roundtrip.transform(["1", 150, 68]) == round(68 * 2.20462) + + +def test_case_in_harmonize_dataset(): + rules = RuleSet() + rules.add_rule( + HarmonizationRule( + ["weight_units", "weight_lbs", "weight_kgs"], "nih_weight", [_weight_case()] + ) + ) + df = pd.DataFrame([ + {"weight_units": "2", "weight_lbs": 150, "weight_kgs": None}, # pounds + {"weight_units": "1", "weight_lbs": None, "weight_kgs": 68}, # kilograms + ]) + out = harmonize_dataset(df, rules, dataset_name="t") + assert out["nih_weight"].tolist() == [150, round(68 * 2.20462)] + + +# --- Case with multi-term branches (height: feet+inches OR meters+cm) ---- + +def _height_case(): + return Case( + sources=["height_units", "ft", "inch", "m", "cm"], + selector="height_units", + branches=[ + {"when": ["1"], "combine": "sum", "terms": [ + {"source": "ft", "operations": [ConvertUnits(Unit.FEET, Unit.INCH)]}, + {"source": "inch", "operations": []}, + ]}, + {"when": ["2"], "combine": "sum", "terms": [ + {"source": "m", "operations": [ConvertUnits(Unit.METER, Unit.INCH)]}, + {"source": "cm", "operations": [ConvertUnits(Unit.CENTIMETER, Unit.INCH)]}, + ]}, + ], + default=None, + ) + + +def test_case_multiterm_feet_inches_sum(): + c = _height_case() + # 5 ft + 7 in -> 60 + 7 = 67 inches + assert c.transform(["1", 5, 7, None, None]) == 67 + + +def test_case_multiterm_meters_cm_sum(): + c = _height_case() + # 1 m + 70 cm -> ~39.37 + ~27.56 inches + result = c.transform(["2", None, None, 1, 70]) + assert abs(result - (1 / 0.0254 + 70 / 2.54)) < 0.01 + + +def test_case_multiterm_serialization_roundtrip(): + rule = HarmonizationRule( + ["height_units", "ft", "inch", "m", "cm"], "nih_height", [_height_case()] + ) + payload = rule.serialize() + roundtrip = HarmonizationRule.from_serialization(payload) + assert roundtrip.serialize() == payload + assert roundtrip.transform(["1", 5, 7, None, None]) == 67 + + +# --- Coalesce ------------------------------------------------------------ + +def _weight_coalesce(): + return Coalesce( + sources=["weight_lbs", "weight_kgs"], + branches=[ + {"source": "weight_lbs", "operations": [DoNothing()]}, + {"source": "weight_kgs", "operations": [Scale(2.20462), Round(0)]}, + ], + default=None, + ) + + +def test_coalesce_first_non_null_wins(): + c = _weight_coalesce() + assert c.transform([150, None]) == 150 # lbs present + assert c.transform([None, 68]) == round(68 * 2.20462) # only kg present + + +def test_coalesce_precedence_when_both_present(): + c = _weight_coalesce() + # both populated -> first branch (lbs) wins + assert c.transform([150, 68]) == 150 + + +def test_coalesce_all_null_returns_default(): + c = _weight_coalesce() + assert c.transform([None, None]) is None + + +def test_coalesce_serialization_roundtrip(): + rule = HarmonizationRule(["weight_lbs", "weight_kgs"], "nih_weight", [_weight_coalesce()]) + payload = rule.serialize() + assert payload["operations"][0]["operation"] == "coalesce" + roundtrip = HarmonizationRule.from_serialization(payload) + assert roundtrip.serialize() == payload + assert roundtrip.transform([None, 68]) == round(68 * 2.20462) From c63d4cf15fa93d63b2102f2f540d8efb3439f321 Mon Sep 17 00:00:00 2001 From: Matthew Horridge Date: Mon, 29 Jun 2026 16:47:44 -0700 Subject: [PATCH 2/4] Rename branch operand list 'terms' -> 'operands' The multi-source branch sub-computations were called 'terms', which only read naturally when combine='sum'. 'operands' is precise (an operand is what a combining op acts on) and combine-agnostic. Pure rename of the serialization key, helper functions, and docstrings; no behavior change. Also drops the unused _term_sources helper. 196 tests passing. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../primitives/case.py | 84 +++++++++---------- .../primitives/coalesce.py | 42 +++++----- tests/test_case_coalesce.py | 12 +-- 3 files changed, 66 insertions(+), 72 deletions(-) diff --git a/src/harmonization_framework/primitives/case.py b/src/harmonization_framework/primitives/case.py index 331ab93..ab8066c 100644 --- a/src/harmonization_framework/primitives/case.py +++ b/src/harmonization_framework/primitives/case.py @@ -3,34 +3,33 @@ from .base import PrimitiveOperation, isnull -def _normalize_terms(branch): - """A branch produces one value from one or more 'terms'. Each term computes a - value from one source via its own op-chain; a branch with multiple terms combines - them with `combine` (a Reduction name string, e.g. 'sum'). - - Accepts three authoring forms, normalized to the canonical term list: - - {"source": "x", "operations": [...]} -> one term - - {"sources": ["a","b"], ... } is NOT auto-summed; use explicit terms instead - - {"terms": [{"source": "a", "operations": [...]}, ...], "combine": "sum"} +def _normalize_operands(branch): + """A branch produces one value from one or more 'operands'. Each operand computes + a value from one source via its own op-chain; a branch with multiple operands + combines them with `combine` (a Reduction name string, e.g. 'sum'). + + Accepts two authoring forms, normalized to the canonical operand list: + - {"source": "x", "operations": [...]} -> one operand + - {"operands": [{"source": "a", "operations": [...]}, ...], "combine": "sum"} """ - if "terms" in branch: - terms = [{"source": t["source"], "operations": list(t.get("operations", []))} - for t in branch["terms"]] + if "operands" in branch: + operands = [{"source": o["source"], "operations": list(o.get("operations", []))} + for o in branch["operands"]] combine = branch.get("combine") else: - terms = [{"source": branch["source"], "operations": list(branch.get("operations", []))}] + operands = [{"source": branch["source"], "operations": list(branch.get("operations", []))}] combine = None - return terms, combine + return operands, combine -def _apply_terms(terms, combine, by_name): - """Compute each term's value, then combine. Returns the branch value.""" +def _apply_operands(operands, combine, by_name): + """Compute each operand's value, then combine. Returns the branch value.""" from .reduce import Reduce, Reduction vals = [] - for t in terms: - current = by_name[t["source"]] - for op in t["operations"]: + for o in operands: + current = by_name[o["source"]] + for op in o["operations"]: current = op(current) vals.append(current) if combine is None: @@ -38,23 +37,18 @@ def _apply_terms(terms, combine, by_name): return Reduce(Reduction(combine)).transform(vals) -def _term_sources(branch): - terms, _ = _normalize_terms(branch) - return [t["source"] for t in terms] - - class Case(PrimitiveOperation): """ Choose one of several branches by switching on a selector source, then compute - that branch's value from one or more source terms. + that branch's value from one or more source operands. A multi-source combinator. The rule's `sources` list names every column this primitive may read; `transform` receives the values in that same order and `Case` zips them back to names internally, so branches address sources by NAME, never by fragile positional index. - A branch is either a single source with an op-chain, or several `terms` (each a - source + op-chain) combined with a reduction. The first branch whose `when` + A branch is either a single source with an op-chain, or several `operands` (each + a source + op-chain) combined with a reduction. The first branch whose `when` contains the selector value wins; null/unmatched selector -> `default`. Single-source branch (RADx-UP weight: unit flag selects pounds-as-is or kg->lbs): @@ -69,17 +63,17 @@ class Case(PrimitiveOperation): ], ) - Multi-term branch (RADx-UP height: feet+inches OR meters+cm, summed in inches): + Multi-operand branch (RADx-UP height: feet+inches OR meters+cm, summed in inches): Case( sources=["height_units", "ft", "in", "m", "cm"], selector="height_units", branches=[ - {"when": ["1"], "combine": "sum", "terms": [ + {"when": ["1"], "combine": "sum", "operands": [ {"source": "ft", "operations": [ConvertUnits(Unit.FEET, Unit.INCH)]}, {"source": "in", "operations": []}, ]}, - {"when": ["2"], "combine": "sum", "terms": [ + {"when": ["2"], "combine": "sum", "operands": [ {"source": "m", "operations": [ConvertUnits(Unit.METER, Unit.INCH)]}, {"source": "cm", "operations": [ConvertUnits(Unit.CENTIMETER, Unit.INCH)]}, ]}, @@ -92,21 +86,21 @@ def __init__(self, sources, selector, branches, default=None): self.selector = selector self.branches = [] for b in branches: - terms, combine = _normalize_terms(b) + operands, combine = _normalize_operands(b) self.branches.append({"when": [str(w) for w in b["when"]], - "terms": terms, "combine": combine}) + "operands": operands, "combine": combine}) self.default = default if self.selector not in self.sources: raise ValueError(f"Case selector {self.selector!r} not in sources {self.sources}") for b in self.branches: - for t in b["terms"]: - if t["source"] not in self.sources: - raise ValueError(f"Case branch source {t['source']!r} not in sources {self.sources}") + for o in b["operands"]: + if o["source"] not in self.sources: + raise ValueError(f"Case branch source {o['source']!r} not in sources {self.sources}") def __str__(self): lines = [f"Switch on {self.selector}:"] for b in self.branches: - srcs = "+".join(t["source"] for t in b["terms"]) + srcs = "+".join(o["source"] for o in b["operands"]) comb = f" ({b['combine']})" if b["combine"] else "" lines.append(f" when {b['when']} -> {srcs}{comb}") lines.append(f" else -> {self.default!r}") @@ -127,7 +121,7 @@ def transform(self, values: Any) -> Any: sel_key = str(sel) for b in self.branches: if sel_key in b["when"]: - return _apply_terms(b["terms"], b["combine"], by_name) + return _apply_operands(b["operands"], b["combine"], by_name) return self.default def to_dict(self): @@ -139,9 +133,9 @@ def to_dict(self): { "when": list(b["when"]), "combine": b["combine"], - "terms": [ - {"source": t["source"], "operations": [op.to_dict() for op in t["operations"]]} - for t in b["terms"] + "operands": [ + {"source": o["source"], "operations": [op.to_dict() for op in o["operations"]]} + for o in b["operands"] ], } for b in self.branches @@ -155,12 +149,12 @@ def from_serialization(cls, serialization): branches = [] for b in serialization["branches"]: - terms = [ - {"source": t["source"], - "operations": [deserialize_operation(op) for op in t.get("operations", [])]} - for t in b["terms"] + operands = [ + {"source": o["source"], + "operations": [deserialize_operation(op) for op in o.get("operations", [])]} + for o in b["operands"] ] - branches.append({"when": b["when"], "combine": b.get("combine"), "terms": terms}) + branches.append({"when": b["when"], "combine": b.get("combine"), "operands": operands}) return cls( sources=serialization["sources"], selector=serialization["selector"], diff --git a/src/harmonization_framework/primitives/coalesce.py b/src/harmonization_framework/primitives/coalesce.py index fd8df1a..fa51932 100644 --- a/src/harmonization_framework/primitives/coalesce.py +++ b/src/harmonization_framework/primitives/coalesce.py @@ -1,23 +1,23 @@ from typing import Any from .base import PrimitiveOperation, isnull -from .case import _normalize_terms, _apply_terms +from .case import _normalize_operands, _apply_operands class Coalesce(PrimitiveOperation): """ Pick the first branch whose (primary) source is non-null, then compute that - branch's value from one or more source terms. + branch's value from one or more source operands. A multi-source combinator, like Case but with the branch chosen by populated-ness rather than an explicit selector. Use Case when the data carries an authoritative unit/type flag; use Coalesce when "whichever field was filled in" is the right rule. - A branch is a single source with an op-chain, or several `terms` (each a + A branch is a single source with an op-chain, or several `operands` (each a source + op-chain) combined with a reduction (same shape as Case branches). A - branch is considered populated when its first term's source is non-null. Branch - order defines precedence; all sources null -> `default`. + branch is considered populated when its first operand's source is non-null. + Branch order defines precedence; all sources null -> `default`. Coalesce( sources=["weight_lbs", "weight_kgs"], @@ -33,18 +33,18 @@ def __init__(self, sources, branches, default=None): self.sources = list(sources) self.branches = [] for b in branches: - terms, combine = _normalize_terms(b) - self.branches.append({"terms": terms, "combine": combine}) + operands, combine = _normalize_operands(b) + self.branches.append({"operands": operands, "combine": combine}) self.default = default for b in self.branches: - for t in b["terms"]: - if t["source"] not in self.sources: - raise ValueError(f"Coalesce branch source {t['source']!r} not in sources {self.sources}") + for o in b["operands"]: + if o["source"] not in self.sources: + raise ValueError(f"Coalesce branch source {o['source']!r} not in sources {self.sources}") def __str__(self): lines = ["Coalesce (first non-null):"] for b in self.branches: - srcs = "+".join(t["source"] for t in b["terms"]) + srcs = "+".join(o["source"] for o in b["operands"]) comb = f" ({b['combine']})" if b["combine"] else "" lines.append(f" {srcs}{comb}") lines.append(f" else -> {self.default!r}") @@ -60,9 +60,9 @@ def _by_name(self, values): def transform(self, values: Any) -> Any: by_name = self._by_name(values) for b in self.branches: - primary = by_name[b["terms"][0]["source"]] + primary = by_name[b["operands"][0]["source"]] if not isnull(primary): - return _apply_terms(b["terms"], b["combine"], by_name) + return _apply_operands(b["operands"], b["combine"], by_name) return self.default def to_dict(self): @@ -72,9 +72,9 @@ def to_dict(self): "branches": [ { "combine": b["combine"], - "terms": [ - {"source": t["source"], "operations": [op.to_dict() for op in t["operations"]]} - for t in b["terms"] + "operands": [ + {"source": o["source"], "operations": [op.to_dict() for op in o["operations"]]} + for o in b["operands"] ], } for b in self.branches @@ -88,12 +88,12 @@ def from_serialization(cls, serialization): branches = [] for b in serialization["branches"]: - terms = [ - {"source": t["source"], - "operations": [deserialize_operation(op) for op in t.get("operations", [])]} - for t in b["terms"] + operands = [ + {"source": o["source"], + "operations": [deserialize_operation(op) for op in o.get("operations", [])]} + for o in b["operands"] ] - branches.append({"combine": b.get("combine"), "terms": terms}) + branches.append({"combine": b.get("combine"), "operands": operands}) return cls( sources=serialization["sources"], branches=branches, diff --git a/tests/test_case_coalesce.py b/tests/test_case_coalesce.py index 7b09f3e..9a3a0dd 100644 --- a/tests/test_case_coalesce.py +++ b/tests/test_case_coalesce.py @@ -77,18 +77,18 @@ def test_case_in_harmonize_dataset(): assert out["nih_weight"].tolist() == [150, round(68 * 2.20462)] -# --- Case with multi-term branches (height: feet+inches OR meters+cm) ---- +# --- Case with multi-operand branches (height: feet+inches OR meters+cm) ---- def _height_case(): return Case( sources=["height_units", "ft", "inch", "m", "cm"], selector="height_units", branches=[ - {"when": ["1"], "combine": "sum", "terms": [ + {"when": ["1"], "combine": "sum", "operands": [ {"source": "ft", "operations": [ConvertUnits(Unit.FEET, Unit.INCH)]}, {"source": "inch", "operations": []}, ]}, - {"when": ["2"], "combine": "sum", "terms": [ + {"when": ["2"], "combine": "sum", "operands": [ {"source": "m", "operations": [ConvertUnits(Unit.METER, Unit.INCH)]}, {"source": "cm", "operations": [ConvertUnits(Unit.CENTIMETER, Unit.INCH)]}, ]}, @@ -97,20 +97,20 @@ def _height_case(): ) -def test_case_multiterm_feet_inches_sum(): +def test_case_multioperand_feet_inches_sum(): c = _height_case() # 5 ft + 7 in -> 60 + 7 = 67 inches assert c.transform(["1", 5, 7, None, None]) == 67 -def test_case_multiterm_meters_cm_sum(): +def test_case_multioperand_meters_cm_sum(): c = _height_case() # 1 m + 70 cm -> ~39.37 + ~27.56 inches result = c.transform(["2", None, None, 1, 70]) assert abs(result - (1 / 0.0254 + 70 / 2.54)) < 0.01 -def test_case_multiterm_serialization_roundtrip(): +def test_case_multioperand_serialization_roundtrip(): rule = HarmonizationRule( ["height_units", "ft", "inch", "m", "cm"], "nih_height", [_height_case()] ) From 11550a38e7d8878d8b638d9194bc871994b76a38 Mon Sep 17 00:00:00 2001 From: Matthew Horridge Date: Mon, 29 Jun 2026 16:50:54 -0700 Subject: [PATCH 3/4] Add Coalesce test exercising combine over a multi-operand branch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Coverage gap: combine was only exercised via Case (height sum). Add a Coalesce case where one branch is stone+pounds summed (Scale(14) on stone + leftover pounds, combine="sum") and the other is a single pounds field — no flag, so the populated branch wins. Demonstrates combine doing real work in Coalesce and adds a serialization round-trip for a multi-operand Coalesce branch. 198 tests passing. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/test_case_coalesce.py | 42 +++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/tests/test_case_coalesce.py b/tests/test_case_coalesce.py index 9a3a0dd..1de4a3e 100644 --- a/tests/test_case_coalesce.py +++ b/tests/test_case_coalesce.py @@ -157,3 +157,45 @@ def test_coalesce_serialization_roundtrip(): roundtrip = HarmonizationRule.from_serialization(payload) assert roundtrip.serialize() == payload assert roundtrip.transform([None, 68]) == round(68 * 2.20462) + + +# --- Coalesce with multi-operand branches + combine ---------------------- +# Weight reported either as a single pounds field, OR as stone + pounds (two +# fields summed). No unit flag, so Coalesce picks whichever branch is populated; +# the stone+pounds branch uses `combine: "sum"` over two operands. + +def _weight_coalesce_combine(): + return Coalesce( + sources=["weight_lbs", "weight_stone", "weight_stone_lbs"], + branches=[ + {"source": "weight_lbs", "operations": [DoNothing()]}, + {"combine": "sum", "operands": [ + {"source": "weight_stone", "operations": [Scale(14)]}, # stone -> lbs + {"source": "weight_stone_lbs", "operations": []}, # leftover pounds, as-is + ]}, + ], + default=None, + ) + + +def test_coalesce_combine_sums_multioperand_branch(): + c = _weight_coalesce_combine() + # single pounds field populated -> first branch wins, value as-is + assert c.transform([150, None, None]) == 150 + # only stone+pounds populated -> second branch: 10 st * 14 + 7 lb = 147 + assert c.transform([None, 10, 7]) == 147 + + +def test_coalesce_combine_serialization_roundtrip(): + rule = HarmonizationRule( + ["weight_lbs", "weight_stone", "weight_stone_lbs"], "nih_weight", + [_weight_coalesce_combine()], + ) + payload = rule.serialize() + # the multi-operand branch carries combine="sum" and two operands + branch = payload["operations"][0]["branches"][1] + assert branch["combine"] == "sum" + assert [o["source"] for o in branch["operands"]] == ["weight_stone", "weight_stone_lbs"] + roundtrip = HarmonizationRule.from_serialization(payload) + assert roundtrip.serialize() == payload + assert roundtrip.transform([None, 10, 7]) == 147 From 7359e4c64e2d2fb03a281c3906d2fd5234adb407 Mon Sep 17 00:00:00 2001 From: Matthew Horridge Date: Mon, 29 Jun 2026 17:02:40 -0700 Subject: [PATCH 4/4] Use explicit DoNothing() for passthrough operands in examples/tests A passthrough operand had an empty operations list ([]); make it [DoNothing()] to match the single-source examples and read clearly as "passes through unchanged" at every no-op site. Functionally identical (empty chain == DoNothing). 16 tests passing. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/harmonization_framework/primitives/case.py | 4 ++-- tests/test_case_coalesce.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/harmonization_framework/primitives/case.py b/src/harmonization_framework/primitives/case.py index ab8066c..87dce01 100644 --- a/src/harmonization_framework/primitives/case.py +++ b/src/harmonization_framework/primitives/case.py @@ -57,7 +57,7 @@ class Case(PrimitiveOperation): sources=["weight_units", "weight_lbs", "weight_kgs"], selector="weight_units", branches=[ - {"when": ["2"], "source": "weight_lbs", "operations": []}, + {"when": ["2"], "source": "weight_lbs", "operations": [DoNothing()]}, {"when": ["1"], "source": "weight_kgs", "operations": [ConvertUnits(Unit.KILOGRAM, Unit.POUNDS), Round(0)]}, ], @@ -71,7 +71,7 @@ class Case(PrimitiveOperation): branches=[ {"when": ["1"], "combine": "sum", "operands": [ {"source": "ft", "operations": [ConvertUnits(Unit.FEET, Unit.INCH)]}, - {"source": "in", "operations": []}, + {"source": "in", "operations": [DoNothing()]}, ]}, {"when": ["2"], "combine": "sum", "operands": [ {"source": "m", "operations": [ConvertUnits(Unit.METER, Unit.INCH)]}, diff --git a/tests/test_case_coalesce.py b/tests/test_case_coalesce.py index 1de4a3e..2533ee5 100644 --- a/tests/test_case_coalesce.py +++ b/tests/test_case_coalesce.py @@ -86,7 +86,7 @@ def _height_case(): branches=[ {"when": ["1"], "combine": "sum", "operands": [ {"source": "ft", "operations": [ConvertUnits(Unit.FEET, Unit.INCH)]}, - {"source": "inch", "operations": []}, + {"source": "inch", "operations": [DoNothing()]}, ]}, {"when": ["2"], "combine": "sum", "operands": [ {"source": "m", "operations": [ConvertUnits(Unit.METER, Unit.INCH)]}, @@ -171,7 +171,7 @@ def _weight_coalesce_combine(): {"source": "weight_lbs", "operations": [DoNothing()]}, {"combine": "sum", "operands": [ {"source": "weight_stone", "operations": [Scale(14)]}, # stone -> lbs - {"source": "weight_stone_lbs", "operations": []}, # leftover pounds, as-is + {"source": "weight_stone_lbs", "operations": [DoNothing()]}, # leftover pounds, as-is ]}, ], default=None,