diff --git a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py index 44448303..78ddd486 100644 --- a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py @@ -6873,3 +6873,30 @@ def test_main_runs_cross_register_and_take_up_contract_preflights() -> None: "assert_take_up_treatments_consistent", ): assert preflight in called, f"main() no longer calls {preflight}" + + +def test_spm_unit_state_fips_collapses_person_broadcast_per_unit() -> None: + """_spm_unit_state_fips must not ask Frame.broadcast for a group target. + + Frame.broadcast only supports the person entity; the SNAP state take-up + stage previously crashed on every release run by broadcasting to + spm_unit. The helper must collapse the person-level broadcast to one + zero-padded FIPS text code per SPM unit, in spm_unit table order. + """ + builder = _load_builder_module() + sys.path.insert(0, str(Path(__file__).parent)) + try: + from test_us_snap_state_take_up import _us_frame + finally: + sys.path.pop(0) + + frame = _us_frame( + [ + {"person_id": 1, "person_spm_unit_id": 10, "state_fips": 6}, + {"person_id": 2, "person_spm_unit_id": 10, "state_fips": 6}, + {"person_id": 3, "person_spm_unit_id": 20, "state_fips": 36}, + {"person_id": 4, "person_spm_unit_id": 30, "state_fips": 2}, + ] + ) + out = builder._spm_unit_state_fips(frame) + assert list(out) == ["06", "36", "02"] diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index f89e59fc..b5e95e26 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -2732,10 +2732,28 @@ def _snap_state_target_table(target_specs: tuple) -> pd.DataFrame: def _spm_unit_state_fips(frame: Frame) -> np.ndarray: - """SPM-unit-aligned state FIPS text codes via the frame's linkage.""" - return np.asarray( - _state_fips_text(frame.broadcast("state_fips", to="spm_unit").to_numpy()) + """SPM-unit-aligned state FIPS text codes via the person linkage. + + ``Frame.broadcast`` only targets the person entity, so state_fips is + broadcast to persons and collapsed to one value per SPM unit (members of + an SPM unit share a household, hence a state). + """ + person = frame.table("person") + person_state = frame.broadcast("state_fips", to="person").to_numpy() + first_by_unit = ( + pd.Series(person_state, index=person["person_spm_unit_id"].to_numpy()) + .groupby(level=0) + .first() ) + spm_unit_ids = frame.table("spm_unit")["spm_unit_id"].to_numpy() + aligned = first_by_unit.reindex(spm_unit_ids) + if aligned.isna().any(): + missing = int(aligned.isna().sum()) + raise RuntimeError( + f"SNAP state take-up: {missing} spm_unit(s) have no member " + "persons to carry state_fips." + ) + return np.asarray(_state_fips_text(aligned.to_numpy())) def _snap_spm_unit_eligibility(