diff --git a/changelog.d/1263-us-acs-local-live-stored-inputs.fixed.md b/changelog.d/1263-us-acs-local-live-stored-inputs.fixed.md new file mode 100644 index 000000000..be9cf1d2b --- /dev/null +++ b/changelog.d/1263-us-acs-local-live-stored-inputs.fixed.md @@ -0,0 +1 @@ +The ACS local-area release's consumer artifact stores the donor's WIC take-up draw under the live input `takes_up_wic_if_eligible`, not the retired `would_claim_wic`, and no longer stores the retired `medicare_part_b_premiums`, so the package stage's stored-inputs gate accepts it. Donor decisions are kept. ACS people, whom the staging builder gives no decision, take WIC up when eligible: the engine's default, and the assumption policyengine.py applies to the same cells today. It is disclosed as the interim limitation `acs_wic_takeup_assumed_interim` until the staging generates ACS participation (microcosm#1154, #1156). diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/retired_stored_inputs.py b/packages/microcosm-build/src/microcosm/build/us_runtime/retired_stored_inputs.py new file mode 100644 index 000000000..203d5dbb2 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/retired_stored_inputs.py @@ -0,0 +1,207 @@ +"""Store the donor's retired engine inputs under the engine's live inputs. + +The ACS local-area chain inherits its donor's person table. The donor named by +the current staging, the reported-receipt child +``populace-us-2024-spm-receipts-20260923``, stores two columns under engine +input names that policyengine-us 2.2.1 no longer defines: + +- ``would_claim_wic``, the WIC take-up draw. policyengine-us 1.777.0 renamed + the input ``takes_up_wic_if_eligible`` (Person, MONTH, Boolean, default + ``True``). +- ``medicare_part_b_premiums``, the ASEC ``PEMCPREM`` transfer target. Its + successor name ``medicare_part_b_premiums_reported`` is read by no engine + version from 1.690.7 to 2.15.1, and microcosm#590 dropped the transfer. + +The stored-inputs contract (:mod:`microcosm.data.stored_inputs`, decision d271) +refuses an artifact that stores either column, because the engine ignores it +by name: every WIC-eligible person would take WIC up, whatever the draw says. +Its register (``US_STORED_NON_VARIABLE_COLUMNS``) leaves both out on purpose +and names the remedy, which :func:`with_live_stored_inputs` applies to the +consumer artifact (microcosm#1263): + +- ``would_claim_wic`` becomes ``takes_up_wic_if_eligible``. Every stored + decision is kept as it is, and a person without one takes WIC up when + eligible (``True``), the engine's own default. The staging builder generates + no decision for ACS people (microcosm#1154; the generator is + microcosm#1156), and the default is the assumption policyengine.py applies + to exactly those cells today (``legacy_inputs._temporary_acs_wic_values``). + The caller says which people may lack a decision (the release tool passes + the ACS spine); a missing decision on anyone else is refused, never filled. + The rule itself reads no provenance column. +- ``medicare_part_b_premiums`` is no longer stored. + +Every count recorded is an aggregate; no row value is. +""" + +from __future__ import annotations + +from collections.abc import Collection +from dataclasses import dataclass +from numbers import Real +from typing import Any + +import numpy as np +import pandas as pd + +from microcosm.frame import Frame + +#: The WIC take-up draw: the retired engine input and the live one. +RETIRED_WIC_INPUT = "would_claim_wic" +LIVE_WIC_INPUT = "takes_up_wic_if_eligible" +#: The retired Part B premium input, which the artifact stops storing. +RETIRED_PART_B_INPUT = "medicare_part_b_premiums" + +#: The interim WIC rule, quoted in the provenance and the release limitation. +ACS_WIC_ASSUMPTION = ( + "An ACS person without a stored WIC take-up decision takes WIC up when " + "eligible: takes_up_wic_if_eligible = True, the engine's own default and " + "the assumption policyengine.py applies to the same cells today." +) +#: When the interim WIC rule ends. +ACS_WIC_FLIP_CONDITION = ( + "Drop the assumption in the first refresh whose staging generates ACS WIC " + "participation (microcosm#1154, generator microcosm#1156)." +) + + +@dataclass(frozen=True) +class LiveStoredInputsResult: + """The frame with live inputs, and what was changed (counts only).""" + + frame: Frame + provenance: dict[str, Any] + + +def with_live_stored_inputs( + frame: Frame, + *, + engine_variables: Collection[str], + may_lack_decision: np.ndarray, +) -> LiveStoredInputsResult: + """Return ``frame`` with the retired stored inputs moved to the live ones. + + ``engine_variables`` are the installed policyengine-us's variable names. + The rename applies only while the engine defines the live input and not + the retired one; anything else is refused, because then the remedy is no + longer the one the stored-inputs register prescribes. + ``may_lack_decision`` is a Boolean vector aligned with the person table's + rows: the people whose missing WIC decision takes the default. + + Raises: + ValueError: if the engine's variables do not fit the remedy, if both + WIC names are stored, if a stored decision is not a Boolean (or + 0/1), if ``may_lack_decision`` is not aligned with the person + table, or if a decision is missing for anyone it does not cover. + """ + + names = set(engine_variables) + if RETIRED_WIC_INPUT in names or LIVE_WIC_INPUT not in names: + raise ValueError( + f"The installed policyengine-us does not fit the {RETIRED_WIC_INPUT} " + f"-> {LIVE_WIC_INPUT} remedy: it must define {LIVE_WIC_INPUT!r} and " + f"not {RETIRED_WIC_INPUT!r}. Review microcosm.data.stored_inputs " + "before packaging against it." + ) + if RETIRED_PART_B_INPUT in names: + raise ValueError( + f"The installed policyengine-us defines {RETIRED_PART_B_INPUT!r}, " + "so it is no longer a retired input to stop storing. Review " + "microcosm.data.stored_inputs before packaging against it." + ) + + person = frame.table("person") + allowed = np.asarray(may_lack_decision) + if allowed.dtype != bool or allowed.shape != (len(person),): + raise ValueError( + "may_lack_decision must be a Boolean vector with one entry per " + f"person ({len(person)}); got dtype {allowed.dtype} and shape " + f"{allowed.shape}." + ) + provenance: dict[str, Any] = { + "decision": "microcosm#1263", + "authority": "microcosm.data.stored_inputs (d271)", + } + changed = person + if RETIRED_WIC_INPUT in person.columns: + changed, provenance["wic"] = _live_wic(changed, allowed) + else: + provenance["wic"] = {"renamed": False} + if RETIRED_PART_B_INPUT in changed.columns: + provenance["medicare_part_b_premiums"] = { + "dropped": True, + "stored_values": int(changed[RETIRED_PART_B_INPUT].notna().sum()), + } + changed = changed.drop(columns=RETIRED_PART_B_INPUT) + else: + provenance["medicare_part_b_premiums"] = {"dropped": False} + if changed is person: + return LiveStoredInputsResult(frame, provenance) + tables = {entity: frame.table(entity) for entity in frame.entities} + tables["person"] = changed + remedied = Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + frame.strata, + mass_log=frame.mass_log, + metadata=frame.metadata, + ) + return LiveStoredInputsResult(remedied, provenance) + + +def _live_wic( + person: pd.DataFrame, allowed: np.ndarray +) -> tuple[pd.DataFrame, dict[str, Any]]: + if LIVE_WIC_INPUT in person.columns: + raise ValueError( + f"The person table stores both {RETIRED_WIC_INPUT!r} and " + f"{LIVE_WIC_INPUT!r}; refusing to choose between them." + ) + stored = person[RETIRED_WIC_INPUT] + missing = stored.isna().to_numpy() + outside = int((missing & ~allowed).sum()) + if outside: + raise ValueError( + f"{RETIRED_WIC_INPUT!r} is missing for {outside} people the caller " + "did not allow to lack a decision; refusing to fill them." + ) + decisions = np.ones(len(person), dtype=bool) + present = ~missing + decisions[present] = _booleans(stored[present]) + renamed = person.drop(columns=RETIRED_WIC_INPUT).assign( + **{LIVE_WIC_INPUT: decisions} + ) + assumed = missing + return renamed, { + "renamed": True, + "from": RETIRED_WIC_INPUT, + "to": LIVE_WIC_INPUT, + "stored_decisions_kept": int(present.sum()), + "stored_decisions_true": int(decisions[present].sum()), + "people_assumed_true": int(assumed.sum()), + "rule": ACS_WIC_ASSUMPTION, + "status": "interim", + "flip_condition": ACS_WIC_FLIP_CONDITION, + } + + +def _booleans(values: pd.Series) -> np.ndarray: + """Stored decisions as Booleans: bools, or real numbers equal to 0 or 1.""" + + out = np.empty(len(values), dtype=bool) + for position, value in enumerate(values.to_numpy(dtype=object)): + if isinstance(value, (bool, np.bool_)): + out[position] = bool(value) + elif ( + isinstance(value, Real) + and not isinstance(value, (bool, np.bool_)) + and value in (0, 1) + ): + out[position] = bool(value) + else: + raise ValueError( + f"{RETIRED_WIC_INPUT!r} stores a value that is not a Boolean " + f"or 0/1 ({type(value).__name__}); refusing to read it as a " + "decision." + ) + return out diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py b/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py index efaca664a..946a79696 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py @@ -310,6 +310,96 @@ def ids(): assert "social_security_cps_reporters_erased_inherited" not in ids() +def test_the_acs_wic_assumption_is_disclosed_exactly_when_it_was_applied() -> None: + """microcosm#1263: the limitation follows the consumer export's receipt.""" + + module = _load_tool_module() + staging_summary = {"base": {"donor_release": {}}, "reviewed_limitations": []} + diagnostics = {"effective_sample_size": 1.0, "ess_fraction": 0.5} + + def limitation(consumer_export): + return { + item["id"]: item + for item in module.finalize_reviewed_limitations( + staging_summary, diagnostics, {}, consumer_export=consumer_export + ) + }.get("acs_wic_takeup_assumed_interim") + + applied = { + "live_stored_inputs": { + "wic": { + "renamed": True, + "from": "would_claim_wic", + "to": "takes_up_wic_if_eligible", + "people_assumed_true": 12, + "flip_condition": "Drop it once microcosm#1156 lands.", + } + } + } + entry = limitation(applied) + assert entry["status"] == "reviewed_interim_assumption" + assert entry["column"] == "takes_up_wic_if_eligible" + assert entry["people_assumed_true"] == 12 + assert entry["treatment"] == "Drop it once microcosm#1156 lands." + assert entry["calibration_blocker"] is False + assert limitation(None) is None + assert limitation({}) is None + assert limitation({"live_stored_inputs": {"wic": {"renamed": False}}}) is None + + +def _spine_frame(spines, decisions): + from microcosm.frame import Frame + from test_support.microcosm_build.us_acs_spm_role import acs_frame + + frame = acs_frame([[(20, 40.0)] for _ in spines]) + person = frame.table("person").copy() + person["person_spine"] = spines + person["would_claim_wic"] = pd.Series(decisions, dtype=object) + person["medicare_part_b_premiums"] = [ + 0.0 if decision is not None else np.nan for decision in decisions + ] + tables = {entity: frame.table(entity) for entity in frame.entities} + tables["person"] = person + return Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + ) + + +def test_only_acs_spine_people_may_lack_a_wic_decision() -> None: + """The tool hands the rule the ACS spine, read from the spine tag.""" + + module = _load_tool_module() + frame = _spine_frame( + ["asec_puf", "acs_2024_1yr", "acs_2024_1yr"], [False, None, None] + ) + + result = module._with_live_stored_inputs(frame) + + person = result.frame.table("person") + assert person["takes_up_wic_if_eligible"].tolist() == [False, True, True] + assert "would_claim_wic" not in person.columns + assert "medicare_part_b_premiums" not in person.columns + assert result.provenance["wic"]["people_assumed_true"] == 2 + # Neither retired name, nor the live one, is left for the package gate to + # refuse. + engine = stored_inputs.installed_us_engine() + refused = stored_inputs.undefined_stored_inputs( + person.columns, engine_variables=engine.variables + ) + assert not { + "would_claim_wic", + "medicare_part_b_premiums", + "takes_up_wic_if_eligible", + } & set(refused) + + with pytest.raises(SystemExit, match="did not allow"): + module._with_live_stored_inputs( + _spine_frame(["asec_puf", "acs_2024_1yr"], [None, None]) + ) + + def test_parse_args_enforces_stage_requirements(tmp_path: Path) -> None: module = _load_tool_module() base = [ diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_retired_stored_inputs.py b/packages/microcosm-build/tests/engine_free/us/test_us_retired_stored_inputs.py new file mode 100644 index 000000000..62fdd90cd --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/us/test_us_retired_stored_inputs.py @@ -0,0 +1,250 @@ +"""Moving the donor's retired stored inputs to the live ones (microcosm#1263). + +Every frame here is invented. The engine is a variable-name set, so no +policyengine-us is imported; ``test_us_retired_stored_inputs_engine`` checks +the names against the installed engine. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd +import pytest + +hypothesis = pytest.importorskip("hypothesis") +from hypothesis import given, settings # noqa: E402 +from hypothesis import strategies as st # noqa: E402 + +from microcosm.build.us_runtime.retired_stored_inputs import ( # noqa: E402 + LIVE_WIC_INPUT, + RETIRED_PART_B_INPUT, + RETIRED_WIC_INPUT, + with_live_stored_inputs, +) +from microcosm.data.stored_inputs import undefined_stored_inputs # noqa: E402 +from microcosm.frame import Frame # noqa: E402 +from test_support.microcosm_build.us_acs_spm_role import acs_frame # noqa: E402 + +#: The live engine as far as these tests need it: the live WIC input, and +#: neither retired name. +ENGINE = frozenset({LIVE_WIC_INPUT, "age", "person_id", "wic"}) +ACS_CHANNEL = "acs_2024_1yr" + + +def _remedy(frame, engine=ENGINE, allowed=None): + """The rule with the release tool's mask: the ACS people may lack one.""" + + if allowed is None: + channels = frame.table("person")["person_support_channel"].map( + lambda value: value.decode() if isinstance(value, bytes) else value + ) + allowed = channels.eq(ACS_CHANNEL).to_numpy(dtype=bool) + return with_live_stored_inputs( + frame, engine_variables=engine, may_lack_decision=allowed + ) + + +def _frame(channels, decisions=None, part_b=None) -> Frame: + """One person per household, with the given channels and stored columns.""" + + frame = acs_frame([[(20, 40.0)] for _ in channels]) + person = frame.table("person").copy() + person["person_support_channel"] = channels + if decisions is not None: + person[RETIRED_WIC_INPUT] = pd.Series(decisions, dtype=object) + if part_b is not None: + person[RETIRED_PART_B_INPUT] = part_b + tables = {entity: frame.table(entity) for entity in frame.entities} + tables["person"] = person + return Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + ) + + +def test_donor_decisions_are_kept_and_acs_gaps_take_wic_up(): + frame = _frame( + ["asec", "puf_tax_detail", ACS_CHANNEL, ACS_CHANNEL], + [True, False, None, np.nan], + ) + + result = _remedy(frame) + + person = result.frame.table("person") + assert RETIRED_WIC_INPUT not in person.columns + assert person[LIVE_WIC_INPUT].dtype == bool + assert person[LIVE_WIC_INPUT].tolist() == [True, False, True, True] + wic = result.provenance["wic"] + assert wic["renamed"] is True + assert wic["stored_decisions_kept"] == 2 + assert wic["stored_decisions_true"] == 1 + assert wic["people_assumed_true"] == 2 + assert wic["status"] == "interim" + assert "microcosm#1156" in wic["flip_condition"] + + +def test_part_b_premiums_stop_being_stored(): + frame = _frame(["asec", ACS_CHANNEL, "puf_tax_detail"], part_b=[120.0, np.nan, 0.0]) + + result = _remedy(frame) + + assert RETIRED_PART_B_INPUT not in result.frame.table("person").columns + assert result.provenance["medicare_part_b_premiums"] == { + "dropped": True, + "stored_values": 2, + } + + +def test_the_remedied_columns_pass_the_stored_inputs_rule(): + frame = _frame(["asec", ACS_CHANNEL], [False, None], part_b=[10.0, np.nan]) + before = undefined_stored_inputs( + frame.table("person").columns, engine_variables=ENGINE + ) + assert set(before) >= {RETIRED_WIC_INPUT, RETIRED_PART_B_INPUT} + + person = _remedy(frame).frame.table("person") + after = undefined_stored_inputs(person.columns, engine_variables=ENGINE) + + assert RETIRED_WIC_INPUT not in after + assert RETIRED_PART_B_INPUT not in after + assert LIVE_WIC_INPUT not in after + + +def test_a_frame_without_retired_inputs_is_returned_unchanged(): + frame = _frame(["asec", ACS_CHANNEL]) + + result = _remedy(frame) + + assert result.frame is frame + assert result.provenance["wic"] == {"renamed": False} + assert result.provenance["medicare_part_b_premiums"] == {"dropped": False} + + +def test_ids_weights_and_other_columns_are_unchanged(): + frame = _frame( + ["asec", ACS_CHANNEL, ACS_CHANNEL], [True, None, None], part_b=[1.0, None, None] + ) + + result = _remedy(frame).frame + + before, after = frame.table("person"), result.table("person") + kept = [ + c for c in before.columns if c not in (RETIRED_WIC_INPUT, RETIRED_PART_B_INPUT) + ] + pd.testing.assert_frame_equal(before[kept], after[kept]) + for entity in frame.entities: + if entity != "person": + pd.testing.assert_frame_equal(frame.table(entity), result.table(entity)) + np.testing.assert_array_equal( + frame.weights_for("household").values, result.weights_for("household").values + ) + + +def test_the_rule_reads_no_provenance_column(): + """Only the caller's mask says who may lack a decision.""" + + frame = _frame(["asec", ACS_CHANNEL], [None, False]) + + person = _remedy(frame, allowed=np.array([True, False])).frame.table("person") + + assert person[LIVE_WIC_INPUT].tolist() == [True, False] + + +@pytest.mark.parametrize( + ("channels", "decisions", "match"), + [ + (["asec", ACS_CHANNEL], [None, None], "did not allow"), + (["puf_tax_detail"], [np.nan], "did not allow"), + (["asec"], ["yes"], "not a Boolean"), + (["asec"], [2], "not a Boolean"), + (["asec"], [1 + 0j], "not a Boolean"), + ], +) +def test_it_refuses_rather_than_guessing(channels, decisions, match): + with pytest.raises(ValueError, match=match): + _remedy(_frame(channels, decisions)) + + +def test_zero_one_numbers_are_read_as_decisions(): + frame = _frame(["asec", "asec", "asec"], [1, 0.0, np.float64(1.0)]) + + person = _remedy(frame).frame.table("person") + + assert person[LIVE_WIC_INPUT].tolist() == [True, False, True] + + +def test_it_refuses_both_wic_names(): + frame = _frame(["asec"], [True]) + tables = {entity: frame.table(entity) for entity in frame.entities} + tables["person"] = tables["person"].assign(**{LIVE_WIC_INPUT: [False]}) + both = Frame( + tables, + frame.schema, + {entity: frame.weights_for(entity) for entity in frame.weighted_entities}, + ) + + with pytest.raises(ValueError, match="both"): + _remedy(both) + + +@pytest.mark.parametrize( + "allowed", + [np.array([True]), np.array([1, 0]), np.array([[True, False]])], +) +def test_the_mask_must_be_one_boolean_per_person(allowed): + with pytest.raises(ValueError, match="one entry per person"): + _remedy(_frame(["asec", ACS_CHANNEL], [True, None]), allowed=allowed) + + +@pytest.mark.parametrize( + "engine", + [ + ENGINE - {LIVE_WIC_INPUT}, + ENGINE | {RETIRED_WIC_INPUT}, + ENGINE | {RETIRED_PART_B_INPUT}, + ], +) +def test_it_refuses_an_engine_the_remedy_does_not_fit(engine): + with pytest.raises(ValueError, match="Review microcosm.data.stored_inputs"): + _remedy(_frame(["asec"], [True]), engine=engine) + + +_CHANNELS = st.sampled_from(["asec", "puf_tax_detail", ACS_CHANNEL]) + + +@st.composite +def _people(draw): + """Channels and decisions where only ACS people may lack one.""" + + channels = draw(st.lists(_CHANNELS, min_size=1, max_size=40)) + decisions = [] + for channel in channels: + if channel == ACS_CHANNEL and draw(st.booleans()): + decisions.append(None) + else: + decisions.append(draw(st.booleans())) + return channels, decisions + + +@settings(max_examples=150, deadline=None) +@given(people=_people()) +def test_stored_decisions_survive_and_only_acs_gaps_are_filled(people): + channels, decisions = people + + result = _remedy(_frame(channels, decisions)) + + live = result.frame.table("person")[LIVE_WIC_INPUT].tolist() + assert len(live) == len(decisions) + for stored, value, channel in zip(decisions, live, channels, strict=True): + if stored is None: + assert channel == ACS_CHANNEL + assert value is True + else: + assert value is stored + wic = result.provenance["wic"] + assert wic["stored_decisions_kept"] + wic["people_assumed_true"] == len(live) + assert wic["stored_decisions_true"] == sum(d is True for d in decisions) + # A second pass finds nothing retired and changes nothing. + again = _remedy(result.frame) + assert again.frame is result.frame diff --git a/packages/microcosm-build/tests/engine_free/us/test_us_spine_blindness.py b/packages/microcosm-build/tests/engine_free/us/test_us_spine_blindness.py index c9be7ff67..77338f117 100644 --- a/packages/microcosm-build/tests/engine_free/us/test_us_spine_blindness.py +++ b/packages/microcosm-build/tests/engine_free/us/test_us_spine_blindness.py @@ -342,6 +342,10 @@ "spm_universe_source.py", # Reviewed held-out SSI/Social Security diagnostic; no population treatment. "ssi_social_security_holdout.py", + # Consumer-artifact remedy for two retired stored inputs (microcosm#1263): + # renames the WIC draw and drops a dead column. It reads no provenance + # column; the release tool passes the people who may lack a decision. + "retired_stored_inputs.py", "stacked_spine.py", # Provenance owner (#578 revision); see owners list. "support_provenance.py", "take_up.py", diff --git a/packages/microcosm-data/src/microcosm/data/stored_inputs.py b/packages/microcosm-data/src/microcosm/data/stored_inputs.py index ffe292684..5193e0cfc 100644 --- a/packages/microcosm-data/src/microcosm/data/stored_inputs.py +++ b/packages/microcosm-data/src/microcosm/data/stored_inputs.py @@ -309,6 +309,10 @@ #: build, and the rehearsal export no longer stores it, so the remedy is to #: stop storing it. #: +#: The ACS local-area chain inherits both from its donor and applies these +#: remedies to its consumer artifact +#: (``microcosm.build.us_runtime.retired_stored_inputs``, microcosm#1263). +#: #: An entry must never be a variable of the certified engine. The engine reads #: such a column as an input, so the entry would be dead, and its reason (that #: the engine ignores the column) would be false. Tests hold the register to diff --git a/tools/build_us_acs_local_release.py b/tools/build_us_acs_local_release.py index 3112f0eea..ebe36a549 100644 --- a/tools/build_us_acs_local_release.py +++ b/tools/build_us_acs_local_release.py @@ -2218,6 +2218,49 @@ def _resample_like_materialize(frame, identity: dict): return sampled +def _with_live_stored_inputs(frame): + """Move the donor's retired stored inputs to the installed engine's live ones. + + :func:`microcosm.build.us_runtime.retired_stored_inputs.with_live_stored_inputs` + owns the rule; the engine is the installed policyengine-us, the one the + package stage checks the artifact against and records as built-with. Only + ACS-spine people may lack a WIC take-up decision: the staging builder + generates none for them (microcosm#1154), and every donor row carries its + own. + """ + + from microcosm.build.us_runtime.acs_pums import ACS_2024_1YR_SPINE + from microcosm.build.us_runtime.base_pool import spine_column + from microcosm.build.us_runtime.retired_stored_inputs import ( + with_live_stored_inputs, + ) + from microcosm.data import stored_inputs + + person = frame.table("person") + tag = spine_column("person") + if tag not in person.columns: + raise SystemExit( + f"Refusing to write the consumer artifact: the person table has no " + f"{tag!r}, so it cannot show which people may lack a WIC decision." + ) + acs_spine = person[tag].eq(ACS_2024_1YR_SPINE).fillna(False).to_numpy(dtype=bool) + + try: + engine = stored_inputs.installed_us_engine() + except ImportError as error: + raise SystemExit( + "Refusing to write the consumer artifact: the installed " + "policyengine-us cannot be imported, so its retired stored inputs " + f"cannot be moved to the live ones: {error}" + ) from error + try: + return with_live_stored_inputs( + frame, engine_variables=engine.variables, may_lack_decision=acs_spine + ) + except ValueError as error: + raise SystemExit(f"Refusing to write the consumer artifact: {error}") from error + + def _write_calibrated_artifact( args, weights: np.ndarray, identity: dict, settings: dict ) -> None: @@ -2266,6 +2309,11 @@ def _write_calibrated_artifact( manifest_path=args.checkpoint_dir / "consumer_reviewed_null_fills.json", period=PERIOD, ) + # microcosm#1263: the donor stores two retired engine inputs that the + # stored-inputs contract refuses at package time; store them as its + # register prescribes. + live_inputs = _with_live_stored_inputs(projected) + projected = live_inputs.frame calibrated = Frame( {entity: projected.table(entity) for entity in projected.entities}, projected.schema, @@ -2292,6 +2340,7 @@ def _write_calibrated_artifact( "held_back_total": sum(len(v) for v in dropped.values()), "filled_columns": len(fills), "total_values_filled": sum(f["filled_rows"] for f in fills), + "live_stored_inputs": live_inputs.provenance, "note": ( "The engine-pass contract (input-schema projection + " "reviewed-null default fill) is applied to the published " @@ -2567,10 +2616,42 @@ def _social_security_erasure_limitation(donor_release: dict) -> dict | None: } +def _acs_wic_takeup_limitation(consumer_export: dict | None) -> dict | None: + """The interim WIC take-up assumption for ACS people, when it was applied.""" + + wic = ((consumer_export or {}).get("live_stored_inputs") or {}).get("wic") or {} + assumed = wic.get("people_assumed_true") + if not assumed: + return None + return { + "id": "acs_wic_takeup_assumed_interim", + "status": "reviewed_interim_assumption", + "issue": "microcosm#1154", + "affected_spines": ["acs_2024_1yr"], + "column": wic.get("to"), + "people_assumed_true": assumed, + "reason": ( + "The staging builder generates no WIC take-up decision for ACS " + "people, and the donor stores its decisions under the retired " + f"input name {wic.get('from')!r}, which the installed " + "policyengine-us ignores. The consumer artifact stores them as " + f"{wic.get('to')!r}: every donor decision is kept, and every ACS " + "person is set to take WIC up when eligible. That is the engine's " + "default and the assumption policyengine.py applies to the same " + "cells today; it is a modelling assumption, not observed " + "participation." + ), + "treatment": wic.get("flip_condition"), + "calibration_blocker": False, + } + + def finalize_reviewed_limitations( staging_summary: dict, diagnostics: dict, spine_qa: dict, + *, + consumer_export: dict | None = None, ) -> list[dict]: """The reviewed-limitations register for the buildo-acs-local lineage. @@ -2682,6 +2763,9 @@ def finalize_reviewed_limitations( social_security = _social_security_erasure_limitation(donor_release) if social_security is not None: limitations.append(social_security) + acs_wic = _acs_wic_takeup_limitation(consumer_export) + if acs_wic is not None: + limitations.append(acs_wic) deduped: dict[str, dict] = {} for limitation in limitations: deduped[limitation["id"]] = limitation @@ -3024,7 +3108,9 @@ def do_finalize(args) -> None: "artifact_sha256": spine_qa.get("artifact_sha256"), } - limitations = finalize_reviewed_limitations(staging_summary, diagnostics, spine_qa) + limitations = finalize_reviewed_limitations( + staging_summary, diagnostics, spine_qa, consumer_export=consumer_export + ) limitations += state_cd_reviewed_limitations(materialize_rss) hard_failures = [ name