diff --git a/changelog.d/us-spm-measurement-universe-source.added.md b/changelog.d/us-spm-measurement-universe-source.added.md new file mode 100644 index 0000000000..bb9db4f5ba --- /dev/null +++ b/changelog.d/us-spm-measurement-universe-source.added.md @@ -0,0 +1 @@ +Add `us_runtime.spm_universe_source`, the source-owned producer for policyengine-us#9462's `spm_unit_spm_universe_status`: ACS housing units and every ASEC record are `INCLUDED`, ACS group quarters are `OUTSIDE`, and twelve named refusals fail closed rather than guess. Declared from the source's own record type, never from a unit's composition. Not wired into any build path and not exported. diff --git a/docs/us-spm-measurement-universe.md b/docs/us-spm-measurement-universe.md new file mode 100644 index 0000000000..b1c4f9beee --- /dev/null +++ b/docs/us-spm-measurement-universe.md @@ -0,0 +1,183 @@ +# The US SPM measurement universe + +PolicyEngine/policyengine-us#9462 makes the SPM measurement universe a +**source declaration**. A dataset supplies `spm_unit_spm_universe_status` ∈ +{`INCLUDED`, `OUTSIDE`, `UNRESOLVED`} for every SPM unit; `OUTSIDE` units get +missing SPM amounts and poverty indicators and never reach the threshold +provider, and an undeclared dataset is `UNRESOLVED` and raises +`SPM_UNIVERSE_REQUIRED`. The engine never infers scope from a unit's +composition, its tenure, its age or its geography. + +This page covers the Microcosm side. The full design — the source evidence, +the registry survey, the consumer enumeration and the open questions — is the +design note `docs/us-spm-measurement-universe-producer.md` on branch +`us-spm-measurement-universe-producer-design`. + +## The producer + +`microcosm.build.us_runtime.spm_universe_source` declares the universe from +each record's own source record type, never from what the record contains: + +| Source record | Status | +| --- | --- | +| ACS housing unit (`TYPEHUGQ == 1`) | `INCLUDED` | +| ACS group quarters (`TYPEHUGQ` ∈ {2, 3}) | `OUTSIDE` | +| ASEC record | `INCLUDED` | + +Both arms come from Census SEHSD-WP2020-09, pages 6–7, and they are not +symmetric. The ACS exclusion is a *data limitation* Census states explicitly: +the public ACS PUMS cannot distinguish noninstitutional group quarters, so the +ACS SPM sample is limited to people living in households, and Microcosm's ACS +spine inherits exactly that limitation. The ASEC ruling is the opposite shape: +the CPS ASEC sample frame already equals the CPS SPM universe, so there is no +ASEC record outside it to declare. The ASEC ruling is therefore made at the +spine rather than per record, and it carries its own refusal — if a frame ever +arrives carrying an ASEC household record-type field, the producer refuses +rather than applying a ruling that was made on the premise that no such field +is read. + +`UNRESOLVED` is the engine's *absence* sentinel, not a producer value. A +producer that cannot decide refuses. + +### Never an inference + +The ACS problem was discovered as SPM units with no classified adult, and the +tempting wrong design is to mark zero-adult units `OUTSIDE`. That would brand +a genuine household-unit data defect as out-of-universe and hide it, and it +would make the universe move whenever the role column moved. The input is +`TYPEHUGQ`, which the source fixes before any composition is computed. +`test_a_zero_adult_housing_unit_is_included` pins this, and +`test_the_declaration_is_invariant_to_age_and_role` pins it in both +directions. + +### Never a target, never a gate, never an agreement surface + +The status is not a calibration target, not a selection criterion, and not a +release gate. Poverty in this repository is a comparison only. The status is +also deliberately **not** a `spine_agreement` surface: it is *intended* to +differ by spine — that is its entire content — so registering it there would +assert that the ACS and ASEC arms should agree about a fact they are defined +to disagree about. + +`OUTSIDE` changes no weight, no membership, and no record's presence in the +file. The fiscal population is unchanged; only its SPM measurability is +declared. + +### Refusals + +Every refusal is a named, fail-closed `ValueError`: + +| Code | Condition | +| --- | --- | +| `SPM_UNIVERSE_COLUMN_EXISTS` | any entity table already carries a declaration | +| `SPM_UNIVERSE_UNDECLARED_SPINE` | a provenance channel with no universe ruling, an untagged frame with no caller declaration, or a caller declaration over a frame that already declares | +| `SPM_UNIVERSE_UNKNOWN_HOUSEHOLD_KIND` | an ACS household whose `TYPEHUGQ` ∉ {1, 2, 3}, missing included | +| `SPM_UNIVERSE_OFF_ARM_HOUSEHOLD_KIND` | a non-ACS household carrying an ACS record type | +| `SPM_UNIVERSE_MIXED_HOUSEHOLD` | one household id resolving to more than one kind | +| `SPM_UNIVERSE_GQ_MULTI_PERSON` | an ACS group-quarters placeholder with more than one native person | +| `SPM_UNIVERSE_ORPHAN_MEMBER` | a membership id matching no owning entity row | +| `SPM_UNIVERSE_EMPTY_UNIT` | an SPM unit with no member | +| `SPM_UNIVERSE_UNIT_SPANS_KINDS` | one SPM unit whose members sit in households of different kinds | +| `SPM_UNIVERSE_DEGRADED_PARTITION` | the ASEC arm's SPM partition is not the native one: `SPM_ID` absent, missing, or not one-to-one with the frame's units per support-clone copy | +| `SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED` | the frame carries an ASEC record-type field the spine-level ruling was not derived against | +| `SPM_UNIVERSE_INVALID_CLONE_INDEX` | the person table carries a support-clone index that is missing, non-finite, negative or non-integral, so a row's clone copy is undecidable | + +`SPM_UNIVERSE_UNIT_SPANS_KINDS` is natively unreachable — ACS group-quarters +households are one-person placeholders — and earns its keep the moment a +future spine, a donor merge or a clone-index bug puts a group-quarters person +into a multi-person unit. Silently picking a majority kind there would be a +fabricated universe decision. + +`SPM_UNIVERSE_DEGRADED_PARTITION` closes the largest silent-failure mode in +the area. `assign_us_unit_structure` uses the native `SPM_ID` when it is fully +present and falls back to `household_id` otherwise, and *any* missing `SPM_ID` +triggers that fallback for the whole frame — while `asec_pool` deliberately +tolerates a missing `SPM_ID` when globalizing. Declaring `INCLUDED` over a +degraded partition would attest to a unit structure that does not exist. The +check is ASEC-only on purpose: ACS PUMS supplies no `SPM_ID` at all, so the +ACS arm reaches `assign_us_unit_structure` without one and takes the household +fallback by construction — there, one SPM unit per household *is* the native +partition and carries no information. + +The native unit key is `(support-clone copy, SPM_ID)`, not `SPM_ID` alone. +The PUF support clone (`puf_support._clone_entity_table`) deep-copies the +person table and remaps only the id and membership columns, so clone copy 1 +carries the native `SPM_ID` under new SPM unit ids; keyed on `SPM_ID` alone, +every support-cloned ASEC frame would read as degraded. +`test_the_real_support_clone_operator_output_is_accepted` runs the real clone +operator on invented rows to pin that. When the frame carries a clone-index +column, a missing, non-finite, negative or non-integral value is refused +(`SPM_UNIVERSE_INVALID_CLONE_INDEX`) before either this check or the +one-native-person group-quarters check runs: the row's copy is undecidable, +and reading it as native could merge a genuine cross-copy collision into an +apparent bijection. A frame with no clone-index column was never +support-cloned, and every person in it is native. + +## What is wired, and what is not + +**Wired.** The producer, its refusals, and its tests. It is classified in the +spine-blindness inventory as a reviewed non-registry module and as a +source-spine provenance owner (the provenance tag is its input, not something +it must be blind to). + +**Not wired, deliberately.** + +- **No build path calls it.** Neither `tools/build_us_acs_local_release.py`, + `tools/build_us_fiscal_refresh_release.py` nor the legacy multispine base + invokes it. Adding a call site is separate work. +- **No export.** `microcosm.frame.adapters.policyengine_us` is untouched, so + the column reaches no release H5. The adapter change is blocked upstream: + see the design note's §4 and Q-A. +- **No enrichment-lane carry.** `microcosm.data.h5_enrichment` writes one + Boolean person column at the HDF type level; carrying an `spm_unit`-entity + string column is a separate, larger piece of work (design note §3). +- **No consumer change here.** `check_spm_composition` still grades every SPM + unit with no universe predicate. `reform_validation._person_rate` already + reads the poverty indicator as nullable (#976, merged 2026-09-22), so a + person with a missing indicator leaves both numerator and denominator; on + the pinned engine, where `in_poverty` is Boolean, nothing is missing. +- **No producer source pin.** The module is deliberately *not* added to + `source_enrichment.PRODUCER_SOURCE_FILES`: that tuple is the SPM-role + enrichment producer's own reviewed source closure, and + `validate_source_enrichment_candidate` refuses a recorded inventory whose key + set differs from it. This module is not in it until the enrichment lane + carries the column (ruled 2026-09-22; see "Rulings" below). + +## Open questions + +These are the design note's, carried here so they are answerable from the +repository. The note states each in full. + +- **Q-A — the formula-owned collision.** The engine variable as drafted cannot + be exported by Microcosm at all, for a reason unrelated to the pin. Resolve + upstream with a source-declaration marker, ask upstream to move the + household default, or add a Microcosm-side allowlist? +- **Q-B — sequencing.** Ship the held-back frame column now, wait for the + engine release, or both? +- **Q-C — group-quarters housing.** ACS group-quarters records are drawing a + housing-receipt flag today with no housing unit. Force the flags `False` on + the group-quarters mask, or supply an explicit reported-subsidy path? This + is a defect on `main` independent of the universe rule. +- **Q-D — the published-rate discontinuity.** The 104 state SPM rates move + when outside persons leave the denominator, correctly but visibly. +- **Q-E — the scope of the ASEC ruling.** Encoded here as a named, cited + constant with its own refusal, per the note's recommendation; signed off + 2026-09-22 (below). + +## Rulings (Max Ghenis, 2026-09-22) + +- `spm_universe_source.py` stays out of + `source_enrichment.PRODUCER_SOURCE_FILES`. +- The missing-`SPM_ID` refusal on the ASEC arm stays strict: an absent or + partly missing `SPM_ID` refuses rather than falling back. +- `puf_tax_detail` stays unruled: a household whose only provenance tag is + that clone-operator channel refuses with `SPM_UNIVERSE_UNDECLARED_SPINE`. +- The ASEC spine judgement is signed off as the named, cited constant + `ASEC_SPM_INCLUDED_AUTHORITY` (Census SEHSD-WP2020-09, pages 6–7). + +## Upstream + +- Engine PR: PolicyEngine/policyengine-us#9462, fixing #9461. +- The dataset must carry `spm_unit_spm_universe_status` on `spm_unit` for + **every year** SPM is requested; a single-year H5 resolves only its base + year. Encode as member names. diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py b/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py new file mode 100644 index 0000000000..7fd990025f --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/spm_universe_source.py @@ -0,0 +1,835 @@ +"""Declare the US SPM measurement universe from each source's record type. + +PolicyEngine/policyengine-us#9462 makes the SPM measurement universe a +**source declaration**: a dataset supplies ``spm_unit_spm_universe_status`` in +``{INCLUDED, OUTSIDE, UNRESOLVED}`` for every SPM unit and year, and the model +never infers scope from composition. This module is Microcosm's producer for +that input. + +It is a source declaration, not a model. Nothing here fits, draws, ages, +weights, or imputes: the status is read off ``TYPEHUGQ`` — the ACS PUMS +household record type the source already puts on the frame's household table +(``acs_pums.py`` ``_HOUSEHOLD_FRAME_COLUMNS``) — and off the record's +provenance channel. ``OUTSIDE`` changes no weight, no membership, and no row's +presence in the file; it declares only that the source does not measure that +unit's SPM poverty. + +The rule +-------- + +=============================================== ========== +ACS housing unit (``TYPEHUGQ == 1``) INCLUDED +ACS group quarters (``TYPEHUGQ`` in ``{2, 3}``) OUTSIDE +ASEC record INCLUDED +=============================================== ========== + +Both arms come from Census SEHSD-WP2020-09 pages 6-7 (see +:data:`SPM_UNIVERSE_METHODOLOGY`), and they are not symmetric. The ACS +exclusion is a *data limitation* Census states explicitly — the public PUMS +cannot separate institutional from noninstitutional group quarters, so the ACS +SPM sample is limited to people living in households. The ASEC ruling is the +opposite shape: the CPS ASEC *sample frame* already equals the CPS SPM +universe, so there is no ASEC record outside it to declare. That ruling is +therefore made at the spine, not per record, and carries its own refusal +(:data:`ASEC_RECORD_TYPE_FIELDS`) so that a future ASEC vintage which admits a +record-type field cannot be silently declared in-universe. + +``UNRESOLVED`` is the engine's *absence* sentinel, not a value a producer +emits. A producer that cannot decide refuses; it never writes ``UNRESOLVED``. + +Never an inference +------------------ + +The ACS problem was discovered as SPM units with no classified adult, and it +would be easy — and wrong — to mark zero-adult units ``OUTSIDE``. That would +brand a genuine household-unit data defect as out-of-universe and hide it, and +it would make the universe move whenever the role column moved. The input is +``TYPEHUGQ``, which the source fixes before any composition is computed. The +anti-inference property is pinned by test: a zero-adult *housing* unit is +``INCLUDED``. + +Never a spine-agreement surface +------------------------------- + +A new categorical ``spm_unit`` column looks like something that belongs in +``spine_agreement``. It does not. The universe status is *intended* to differ +by spine — that is its entire content — and registering it there would assert +that the ACS and ASEC arms should agree about a fact they are defined to +disagree about. The status must never become a declared transfer, a derived +transfer, a take-up contract entry or a simulated output, which is what would +be required to admit it to that gate. + +Never a target +-------------- + +The status is not a calibration target, not a selection criterion, and not a +gate. Poverty in this repository is a comparison only. + +Import weight +------------- + +This producer is a leaf: a preflight or a release tool must be able to call it +without dragging the ACS loader into the process. Both +``microcosm.build.us_runtime.acs_pums`` and ``...stacked_spine`` pull the +country engine: re-measured in separate processes on 2026-09-22 after merging +main, each import took ~36-38 s on a warm cache (the authoring lane measured +~90 s for ``stacked_spine`` cold). So the four channel +tag values below are named here rather than imported from those modules. They +are not a second source of truth: :mod:`test_us_spm_universe_source` parses +each canonical definition out of its defining module with :mod:`ast` — no +import — and fails if this table drifts from it. The one canonical constant +that is free to import (:data:`BASE_ASEC_SUPPORT_CHANNEL`) is imported. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from types import MappingProxyType +from typing import Any + +import numpy as np +import pandas as pd + +from .support_provenance import ( + BASE_ASEC_SUPPORT_CHANNEL, + support_channel_column, + support_clone_index_column, +) + +__all__ = [ + "ACS_ARM", + "ACS_GROUP_QUARTERS_KINDS", + "ACS_HOUSEHOLD_KIND_COLUMN", + "ACS_HOUSEHOLD_KINDS", + "ACS_HOUSING_UNIT_KIND", + "ACS_SPM_OUTSIDE_AUTHORITY", + "ASEC_ARM", + "ASEC_NATIVE_UNIT_COLUMN", + "ASEC_RECORD_TYPE_FIELDS", + "ASEC_SPM_INCLUDED_AUTHORITY", + "INCLUDED", + "OUTSIDE", + "SPM_UNIVERSE_CHANNEL_ARMS", + "SPM_UNIVERSE_METHODOLOGY", + "SPM_UNIVERSE_PROFILE_NAME", + "SPM_UNIVERSE_REFUSALS", + "SPM_UNIVERSE_RULE", + "SPM_UNIVERSE_STATUSES", + "SpmUniverseClassification", + "SpmUniverseSourceResult", + "UNIVERSE_INPUT", + "UNRESOLVED", + "attach_spm_universe_status", + "classify_spm_universe", +] + +#: The engine input this producer supplies (policyengine-us#9462). The name +#: and the three-state vocabulary below are deliberately identical to the +#: graph-native line's ``spm_input_contract`` (unmerged native-SPM branches, +#: first at commit 774a90091; not on main); there is no reason for two +#: spellings of the same declaration. +UNIVERSE_INPUT = "spm_unit_spm_universe_status" + +INCLUDED = "INCLUDED" +OUTSIDE = "OUTSIDE" +#: The engine's absence sentinel. Read back from a defaulted dataset; never +#: written by this producer. +UNRESOLVED = "UNRESOLVED" +SPM_UNIVERSE_STATUSES = (INCLUDED, OUTSIDE, UNRESOLVED) + +#: The two source universes this producer rules on. +ACS_ARM = "acs" +ASEC_ARM = "asec" + +#: The ACS PUMS household record type, as the source stages it. +ACS_HOUSEHOLD_KIND_COLUMN = "TYPEHUGQ" +ACS_HOUSING_UNIT_KIND = 1 +ACS_GROUP_QUARTERS_KINDS = (2, 3) +ACS_HOUSEHOLD_KINDS = (ACS_HOUSING_UNIT_KIND, *ACS_GROUP_QUARTERS_KINDS) + +#: The ASEC native SPM unit id. Its presence and completeness is what proves +#: the frame's SPM partition is the native one rather than the documented +#: household fallback (``microcosm.frame.units.assign_us_unit_structure``). +ASEC_NATIVE_UNIT_COLUMN = "SPM_ID" + +#: ASEC household record-type fields. Microcosm reads none of them at this +#: head, which is why the ASEC ruling is made at the spine. If one ever +#: arrives on a frame, the spine-level ruling has to be re-derived against it +#: rather than silently applied, so its presence refuses. +ASEC_RECORD_TYPE_FIELDS = ("H_TYPE", "HRHTYPE", "H_HHTYPE", "HUNITS", "H_LIVQRT") + +SPM_UNIVERSE_METHODOLOGY = ( + "Census SEHSD-WP2020-09, pages 6-7: all people living in group quarters " + "are excluded from the ACS SPM because the public ACS PUMS cannot " + "distinguish noninstitutional group quarters, while the CPS ASEC sample " + "frame already equals the CPS SPM universe." +) +#: Named authority for the ACS exclusion; identical to the graph-native line's +#: ``acs_spm_scope.ACS_SPM_OUTSIDE_AUTHORITY`` (same unmerged branches). +ACS_SPM_OUTSIDE_AUTHORITY = "outside_acs_household_universe" +#: Named authority for the ASEC ruling. This one has no upstream: it is this +#: producer's own source-universe judgement, made on the Census quote above +#: plus the zero-adult reconciliation ``spm_role_source`` already enforces +#: against the pinned complete Census ASEC files. Signed off as this named, +#: cited constant by Max Ghenis on 2026-09-22 (microcosm#977). +ASEC_SPM_INCLUDED_AUTHORITY = "asec_sample_frame_equals_cps_spm_universe" +#: The named analysis scope a release declares when it carries this status. +SPM_UNIVERSE_PROFILE_NAME = "us_acs_household_and_asec_frame_universe_v1" + +SPM_UNIVERSE_RULE = ( + "TYPEHUGQ == 1 -> INCLUDED; TYPEHUGQ in {2, 3} -> OUTSIDE; ASEC record -> INCLUDED" +) + +#: Provenance tag values this producer has a universe ruling for. Anything +#: else refuses; see the module docstring on why these are named rather than +#: imported. +SPM_UNIVERSE_CHANNEL_ARMS: Mapping[str, str] = MappingProxyType( + { + # microcosm.build.us_runtime.acs_pums.ACS_2024_1YR_SPINE + "acs_2024_1yr": ACS_ARM, + # microcosm.build.us_runtime.stacked_spine.ACS_STACKED_SUPPORT_CHANNEL + "acs": ACS_ARM, + # microcosm.build.us_runtime.base_pool.ASEC_PUF_SPINE + "asec_puf": ASEC_ARM, + BASE_ASEC_SUPPORT_CHANNEL: ASEC_ARM, + } +) + +#: The household column naming each row's base spine +#: (``base_pool.spine_column("household")``), then the assembled support +#: channel, in the order the producer looks for them. +HOUSEHOLD_SPINE_COLUMN = "household_spine" +HOUSEHOLD_SUPPORT_CHANNEL_COLUMN = support_channel_column("household") +_CHANNEL_COLUMN_ORDER = (HOUSEHOLD_SPINE_COLUMN, HOUSEHOLD_SUPPORT_CHANNEL_COLUMN) +_PERSON_CLONE_INDEX_COLUMN = support_clone_index_column("person") +_NATIVE_CLONE_INDEX = 0 + +SPM_UNIVERSE_REFUSALS = ( + "SPM_UNIVERSE_COLUMN_EXISTS", + "SPM_UNIVERSE_UNDECLARED_SPINE", + "SPM_UNIVERSE_UNKNOWN_HOUSEHOLD_KIND", + "SPM_UNIVERSE_OFF_ARM_HOUSEHOLD_KIND", + "SPM_UNIVERSE_MIXED_HOUSEHOLD", + "SPM_UNIVERSE_GQ_MULTI_PERSON", + "SPM_UNIVERSE_ORPHAN_MEMBER", + "SPM_UNIVERSE_EMPTY_UNIT", + "SPM_UNIVERSE_UNIT_SPANS_KINDS", + "SPM_UNIVERSE_DEGRADED_PARTITION", + "SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED", + "SPM_UNIVERSE_INVALID_CLONE_INDEX", +) + +_MAX_REPORTED_EXAMPLES = 5 + + +@dataclass(frozen=True) +class SpmUniverseClassification: + """One frame's universe declaration, plus the evidence it was made from.""" + + #: Member-name strings in ``unit_ids`` order. + status: np.ndarray + #: ``unit_ids`` as classified, in the order given. + unit_ids: np.ndarray + #: ``ACS_ARM``/``ASEC_ARM`` per household row. + household_arm: pd.Series + #: ACS record type per household row; missing on the ASEC arm. + household_kind: pd.Series + provenance: dict[str, Any] + + +@dataclass(frozen=True) +class SpmUniverseSourceResult: + """A frame carrying the declaration, with the classification behind it.""" + + frame: Any + classification: SpmUniverseClassification + + @property + def status(self) -> np.ndarray: + return self.classification.status + + @property + def provenance(self) -> dict[str, Any]: + return self.classification.provenance + + +def _require(condition: bool, code: str, detail: str) -> None: + """Refuse, named and fail-closed, in ``spm_role_source``'s idiom.""" + if not condition: + raise ValueError(f"{code}: {detail}") + + +def _examples(values: Any) -> str: + unique = pd.unique(np.asarray(values, dtype=object)) + shown = ", ".join(repr(value) for value in unique[:_MAX_REPORTED_EXAMPLES]) + if len(unique) > _MAX_REPORTED_EXAMPLES: + shown = f"{shown}, ... ({len(unique)} distinct)" + return shown + + +def _resolve_channel( + household: pd.DataFrame, + *, + declared_channel: str | None, + channel_column: str | None, +) -> tuple[pd.Series, str]: + """Return each household's provenance tag and where it came from. + + Resolution order is explicit and fail-closed: an operator-named column, the + base-spine tag, the assembled support channel, then a caller declaration + for a frame that carries no tag at all. A caller may not declare over a + frame that already declares for itself. + """ + present = [name for name in _CHANNEL_COLUMN_ORDER if name in household.columns] + if channel_column is not None: + _require( + channel_column in household.columns, + "SPM_UNIVERSE_UNDECLARED_SPINE", + f"the household table carries no requested channel column " + f"{channel_column!r}; present: {sorted(household.columns)[:20]}.", + ) + _require( + declared_channel is None, + "SPM_UNIVERSE_UNDECLARED_SPINE", + "declared_channel and channel_column are mutually exclusive.", + ) + return household[channel_column].astype(object), channel_column + if present: + _require( + declared_channel is None, + "SPM_UNIVERSE_UNDECLARED_SPINE", + f"the household table already declares its own provenance in " + f"{present[0]!r}; a caller declaration would override it.", + ) + return household[present[0]].astype(object), present[0] + _require( + declared_channel is not None, + "SPM_UNIVERSE_UNDECLARED_SPINE", + "the household table carries no provenance tag " + f"({' or '.join(_CHANNEL_COLUMN_ORDER)}) and the caller declared no " + "channel, so no universe ruling applies to these rows.", + ) + return ( + pd.Series(declared_channel, index=household.index, dtype=object), + f"declared:{declared_channel}", + ) + + +def _household_arms(channel: pd.Series) -> pd.Series: + arm = channel.map(SPM_UNIVERSE_CHANNEL_ARMS) + unruled = channel[arm.isna()] + _require( + unruled.empty, + "SPM_UNIVERSE_UNDECLARED_SPINE", + f"{len(unruled)} household row(s) carry a provenance channel this " + f"producer has no universe ruling for: {_examples(unruled)}; ruled " + f"channels: {sorted(SPM_UNIVERSE_CHANNEL_ARMS)}.", + ) + return arm.astype(object) + + +def _household_kinds(household: pd.DataFrame, arm: pd.Series) -> pd.Series: + """Validate the ACS record type and its confinement to the ACS arm.""" + if ACS_HOUSEHOLD_KIND_COLUMN in household.columns: + kind = pd.to_numeric( + household[ACS_HOUSEHOLD_KIND_COLUMN], errors="coerce" + ).astype(float) + else: + kind = pd.Series(np.nan, index=household.index, dtype=float) + acs = arm.eq(ACS_ARM).to_numpy() + unknown = acs & ~kind.isin(ACS_HOUSEHOLD_KINDS).to_numpy() + if ACS_HOUSEHOLD_KIND_COLUMN in household.columns: + offending = _examples(household.loc[unknown, ACS_HOUSEHOLD_KIND_COLUMN]) + else: + offending = "" + _require( + not unknown.any(), + "SPM_UNIVERSE_UNKNOWN_HOUSEHOLD_KIND", + f"{int(unknown.sum())} ACS household row(s) carry a " + f"{ACS_HOUSEHOLD_KIND_COLUMN} outside {list(ACS_HOUSEHOLD_KINDS)} " + f"(missing counts as outside): {offending}.", + ) + # TYPEHUGQ == 1 means "ACS housing unit", not "not group quarters": ASEC + # rows carry it as missing by design, so a non-missing value on the ASEC + # arm is a bad merge, not a housing unit. + off_arm = (~acs) & kind.notna().to_numpy() + _require( + not off_arm.any(), + "SPM_UNIVERSE_OFF_ARM_HOUSEHOLD_KIND", + f"{int(off_arm.sum())} non-ACS household row(s) carry an ACS " + f"{ACS_HOUSEHOLD_KIND_COLUMN} value " + f"({_examples(kind[off_arm])}); the ACS record type does not " + "describe them and cannot be read as one.", + ) + return kind + + +def _kind_keys(arm: pd.Series, kind: pd.Series) -> pd.Series: + """One hashable ``(arm, record type)`` per household row.""" + return pd.Series( + [ + (str(a), None if pd.isna(k) else int(k)) + for a, k in zip(arm.to_numpy(), kind.to_numpy(), strict=True) + ], + index=arm.index, + dtype=object, + ) + + +def _assert_one_kind_per_household( + household_ids: pd.Series, kind_keys: pd.Series +) -> None: + """A household id resolving to more than one kind is a bad merge. + + Unreachable through a validated :class:`~microcosm.frame.Frame`, whose + constructor already refuses a duplicated group id. It is reachable when the + producer is handed tables directly, and picking a majority kind there would + be a fabricated universe decision. + """ + distinct = kind_keys.groupby(household_ids.to_numpy(), sort=False).nunique() + mixed = distinct[distinct > 1] + _require( + mixed.empty, + "SPM_UNIVERSE_MIXED_HOUSEHOLD", + f"{len(mixed)} household id(s) resolve to more than one " + f"(arm, record type): {_examples(mixed.index)}.", + ) + + +def _person_clone_copies(person: pd.DataFrame) -> np.ndarray: + """Each person's support-clone copy; refuses an undecidable clone index. + + A support clone deep-copies every column of a table and remaps only the + structural id and membership columns + (``puf_support._clone_entity_table``), so clone copy ``k`` carries the + native row's source fields under new ids. With no clone-index column the + frame was never support-cloned and every person is native. When the column + is present, a missing, non-finite, negative or non-integral value is + undecidable provenance, and the module refuses it rather than guess a copy: + reading it as native could merge a genuine cross-copy collision into an + apparent bijection in the degraded-partition check. The maintained readers + of this column refuse the same values (``acs_transfer`` raises on an + invalid clone index; ``stacked_spine`` and ``puf_support`` parse it with + ``errors="raise"`` or refuse a missing one). + """ + if _PERSON_CLONE_INDEX_COLUMN not in person.columns: + return np.full(len(person), _NATIVE_CLONE_INDEX, dtype=float) + values = pd.to_numeric( + person[_PERSON_CLONE_INDEX_COLUMN], errors="coerce" + ).to_numpy(dtype=float) + invalid = ~np.isfinite(values) | (values < 0) | (values != np.floor(values)) + _require( + not invalid.any(), + "SPM_UNIVERSE_INVALID_CLONE_INDEX", + f"{int(invalid.sum())} person row(s) carry a missing, non-finite, " + f"negative or non-integral {_PERSON_CLONE_INDEX_COLUMN}; the row's " + "support-clone copy is undecidable, so neither the group-quarters nor " + "the degraded-partition check can be judged.", + ) + return values + + +def _assert_group_quarters_are_single_native_records( + *, + household_ids: pd.Series, + arm: pd.Series, + kind: pd.Series, + person_household_ids: pd.Series, + clone_copies: np.ndarray, +) -> int: + """Mirror the stacked spine's one-native-person-per-GQ-placeholder rule. + + ACS group-quarters rows are one-person ``WGTP == 0`` placeholders at the + source (``acs_pums._occupied_households``), and the stacked assembly + re-asserts it. Clones of such a placeholder are expected and are not + counted here; more than one *native* person in one is not. A person + whose clone index is missing or invalid is refused before this check runs + (see :func:`_person_clone_copies`), so a pool that lost the index on one + arm cannot pass it vacuously. + """ + gq = (arm.eq(ACS_ARM) & kind.isin(ACS_GROUP_QUARTERS_KINDS)).to_numpy() + gq_ids = pd.Index(household_ids.to_numpy()[gq]).unique() + if len(gq_ids) == 0: + return 0 + native = clone_copies == _NATIVE_CLONE_INDEX + counts = ( + pd.Series(1, index=person_household_ids.to_numpy()[native]) + .groupby(level=0) + .sum() + .reindex(gq_ids, fill_value=0) + ) + crowded = counts[counts > 1] + _require( + crowded.empty, + "SPM_UNIVERSE_GQ_MULTI_PERSON", + f"{len(crowded)} ACS group-quarters household row(s) carry more than " + f"one native person: {_examples(crowded.index)}.", + ) + return int(len(gq_ids)) + + +def _assert_membership( + values: pd.Series, owner_ids: np.ndarray, *, column: str, owner: str +) -> None: + """Both directions of a membership join, independently of the frame kernel.""" + owner_index = pd.Index(owner_ids) + orphan = values[~pd.Index(values).isin(owner_index)] + _require( + orphan.empty, + "SPM_UNIVERSE_ORPHAN_MEMBER", + f"{len(orphan)} person row(s) carry a {column!r} value matching no " + f"{owner} id: {_examples(orphan)}.", + ) + if owner == "spm_unit": + referenced = pd.Index(pd.unique(values.to_numpy())) + empty = owner_index[~owner_index.isin(referenced)] + _require( + empty.empty, + "SPM_UNIVERSE_EMPTY_UNIT", + f"{len(empty)} {owner} id(s) have no member in {column!r}: " + f"{_examples(empty)}; an empty unit has no source record to read " + "a universe from.", + ) + + +def _assert_native_asec_partition( + *, + person: pd.DataFrame, + asec_person: np.ndarray, + unit_membership_column: str, + clone_copies: np.ndarray, +) -> None: + """Refuse a silently degraded SPM partition on the ASEC arm. + + ``assign_us_unit_structure`` uses the native ``SPM_ID`` when it is fully + present and falls back to ``household_id`` otherwise — and *any* missing + ``SPM_ID`` triggers that fallback for the whole frame, while ``asec_pool`` + deliberately tolerates a missing ``SPM_ID`` when globalizing. Declaring + ``INCLUDED`` over a degraded partition would attest to a unit structure + that does not exist, so the ASEC arm must prove its partition is the + native one. + + A support clone copies ``SPM_ID`` verbatim and gives each copy new unit + ids (see :func:`_person_clone_copies`), so every clone copy is its own + instance of the native partition. The native unit key is therefore + ``(clone copy, SPM_ID)``, and the frame's units must be a one-to-one + densification of that key. Keying on ``SPM_ID`` alone would read every + support-cloned ASEC frame as degraded. + + The check is deliberately ASEC-only. ACS PUMS supplies no ``SPM_ID`` at + all, so the ACS arm reaches ``assign_us_unit_structure`` without one and + takes the household fallback by construction — there, one SPM unit per + household *is* the native partition and carries no information. + """ + if not asec_person.any(): + return + _require( + ASEC_NATIVE_UNIT_COLUMN in person.columns, + "SPM_UNIVERSE_DEGRADED_PARTITION", + f"the person table carries no {ASEC_NATIVE_UNIT_COLUMN!r}, so the ASEC " + "arm's SPM partition cannot be shown to be the native one rather than " + "the documented household fallback.", + ) + native = person.loc[asec_person, ASEC_NATIVE_UNIT_COLUMN] + missing = int(native.isna().sum()) + _require( + missing == 0, + "SPM_UNIVERSE_DEGRADED_PARTITION", + f"{missing} ASEC person row(s) carry a missing " + f"{ASEC_NATIVE_UNIT_COLUMN}; any missing value silently converts every " + "SPM unit in the frame into a household.", + ) + units = person.loc[asec_person, unit_membership_column].to_numpy() + copies = clone_copies[asec_person] + native_units = len(pd.MultiIndex.from_arrays([copies, native.to_numpy()]).unique()) + frame_units = len(pd.unique(units)) + pairings = len( + pd.MultiIndex.from_arrays([copies, native.to_numpy(), units]).unique() + ) + _require( + pairings == native_units == frame_units, + "SPM_UNIVERSE_DEGRADED_PARTITION", + f"the ASEC arm's {unit_membership_column!r} is not a one-to-one " + f"densification of (clone copy, {ASEC_NATIVE_UNIT_COLUMN}) " + f"({native_units} native unit(s), {frame_units} frame unit(s), " + f"{pairings} distinct pairing(s)), so the native partition did not " + "survive assembly.", + ) + + +def _assert_asec_record_type_unreviewed( + *, household: pd.DataFrame, person: pd.DataFrame, asec_any: bool +) -> None: + """Guard the spine-level ASEC ruling against a vintage that outgrows it.""" + if not asec_any: + return + found = sorted( + field + for field in ASEC_RECORD_TYPE_FIELDS + if field in household.columns or field in person.columns + ) + _require( + not found, + "SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED", + f"the frame carries ASEC household record-type field(s) {found}. The " + f"ASEC ruling ({ASEC_SPM_INCLUDED_AUTHORITY}) was made at the spine " + "precisely because Microcosm reads none of these; it must be " + "re-derived against the field rather than applied over it.", + ) + + +def classify_spm_universe( + *, + household: pd.DataFrame, + person: pd.DataFrame, + unit_ids: np.ndarray, + household_id_column: str = "household_id", + household_membership_column: str = "person_household_id", + unit_membership_column: str = "person_spm_unit_id", + declared_channel: str | None = None, + channel_column: str | None = None, +) -> SpmUniverseClassification: + """Classify every SPM unit's measurement universe from source record type. + + Args: + household: The household table, carrying its provenance tag and, on + the ACS arm, ``TYPEHUGQ``. + person: The person table, carrying household and SPM-unit membership. + unit_ids: The SPM unit ids to classify, in the order the caller wants + the returned status. + household_id_column: Id column on ``household``. + household_membership_column: Person column linking to ``household``. + unit_membership_column: Person column linking to the SPM unit table. + declared_channel: Universe channel for a frame that carries no + provenance tag of its own. Refused when the frame does carry one. + channel_column: Explicit provenance column to read instead of the + default resolution order. + + Returns: + The per-unit status as member-name strings, the household-level + evidence it was derived from, and a provenance dict. + + Raises: + ValueError: Any of :data:`SPM_UNIVERSE_REFUSALS`. Every refusal is + fail-closed: this producer never resolves an ambiguity by guessing, + and never writes ``UNRESOLVED`` to mean "I could not tell". + """ + for table, columns, label in ( + (household, (household_id_column,), "household"), + (person, (household_membership_column, unit_membership_column), "person"), + ): + missing = sorted(set(columns) - set(table.columns)) + _require( + not missing, + "SPM_UNIVERSE_ORPHAN_MEMBER", + f"the {label} table is missing required column(s) {missing}.", + ) + + unit_ids = np.asarray(unit_ids) + household_ids = household[household_id_column] + channel, channel_source = _resolve_channel( + household, declared_channel=declared_channel, channel_column=channel_column + ) + arm = _household_arms(channel) + kind = _household_kinds(household, arm) + kind_keys = _kind_keys(arm, kind) + _assert_one_kind_per_household(household_ids, kind_keys) + + person_household_ids = person[household_membership_column] + person_units = person[unit_membership_column] + _assert_membership( + person_household_ids, + household_ids.to_numpy(), + column=household_membership_column, + owner="household", + ) + _assert_membership( + person_units, unit_ids, column=unit_membership_column, owner="spm_unit" + ) + + clone_copies = _person_clone_copies(person) + group_quarters_households = _assert_group_quarters_are_single_native_records( + household_ids=household_ids, + arm=arm, + kind=kind, + person_household_ids=person_household_ids, + clone_copies=clone_copies, + ) + + kind_by_household = pd.Series( + kind_keys.to_numpy(), index=pd.Index(household_ids.to_numpy()) + ) + # Every duplicate id resolves to one kind by now, so dropping repeats + # cannot change a lookup; it only keeps ``map`` from fanning rows out. + kind_by_household = kind_by_household[~kind_by_household.index.duplicated()] + person_kind = person_household_ids.map(kind_by_household) + asec_person = np.asarray( + [key[0] == ASEC_ARM for key in person_kind.to_numpy()], dtype=bool + ) + _assert_asec_record_type_unreviewed( + household=household, person=person, asec_any=bool(asec_person.any()) + ) + _assert_native_asec_partition( + person=person, + asec_person=asec_person, + unit_membership_column=unit_membership_column, + clone_copies=clone_copies, + ) + + by_unit = pd.DataFrame( + {"unit": person_units.to_numpy(), "kind": person_kind.to_numpy()} + ) + distinct = by_unit.groupby("unit", sort=False)["kind"].nunique() + spanning = distinct[distinct > 1] + _require( + spanning.empty, + "SPM_UNIVERSE_UNIT_SPANS_KINDS", + f"{len(spanning)} SPM unit(s) have members in households of different " + f"(arm, record type): {_examples(spanning.index)}. Picking a majority " + "kind would be a fabricated universe decision.", + ) + + unit_kind = ( + by_unit.groupby("unit", sort=False)["kind"].first().reindex(pd.Index(unit_ids)) + ) + status = np.asarray( + [ + OUTSIDE + if key[0] == ACS_ARM and key[1] in ACS_GROUP_QUARTERS_KINDS + else INCLUDED + for key in unit_kind.to_numpy() + ], + dtype=object, + ) + # A producer that cannot decide refuses above; UNRESOLVED is the engine's + # absence sentinel and must never be emitted. + _require( + UNRESOLVED not in set(status), + "SPM_UNIVERSE_UNDECLARED_SPINE", + f"the classification emitted {UNRESOLVED}, which is the engine's " + "absence sentinel rather than a producer value.", + ) + + arm_counts = arm.value_counts() + kind_counts = { + int(value): int(count) + for value, count in kind.value_counts(dropna=True).items() + } + provenance = { + "declared_variable": UNIVERSE_INPUT, + "declared_entity": "spm_unit", + "declared_period": "YEAR", + "stored_encoding": "enum member name string", + "profile": SPM_UNIVERSE_PROFILE_NAME, + "rule": SPM_UNIVERSE_RULE, + "acs_authority": ACS_SPM_OUTSIDE_AUTHORITY, + "asec_authority": ASEC_SPM_INCLUDED_AUTHORITY, + "methodology": SPM_UNIVERSE_METHODOLOGY, + "evidence_status": ( + "Source record type read from the frame's own household table " + "(ACS TYPEHUGQ), plus a spine-level ruling for the ASEC arm whose " + "sample frame equals the CPS SPM universe. No composition, age, " + "tenure, geography or role is read." + ), + "channel_source": channel_source, + "channel_arms": dict(SPM_UNIVERSE_CHANNEL_ARMS), + "household_rows": int(len(household)), + "households_by_arm": {str(k): int(v) for k, v in arm_counts.items()}, + "households_by_acs_record_type": kind_counts, + "acs_group_quarters_households": group_quarters_households, + "person_rows": int(len(person)), + "spm_units": int(len(unit_ids)), + "units_included": int((status == INCLUDED).sum()), + "units_outside": int((status == OUTSIDE).sum()), + "units_unresolved": 0, + "refusals": list(SPM_UNIVERSE_REFUSALS), + "composition_read": False, + "weights_used": False, + "ages_changed": False, + "is_calibration_target": False, + "is_spine_agreement_surface": False, + "period_handling": ( + "The declaration is a construction fact about the source record, " + "constant across the years a build requests." + ), + } + return SpmUniverseClassification( + status=status, + unit_ids=unit_ids, + household_arm=arm, + household_kind=kind, + provenance=provenance, + ) + + +def attach_spm_universe_status( + frame: Any, + *, + unit_entity: str = "spm_unit", + declared_channel: str | None = None, + channel_column: str | None = None, +) -> SpmUniverseSourceResult: + """Attach :data:`UNIVERSE_INPUT` to ``frame``'s SPM unit table. + + The status is stored as the enum **member name** — ``"OUTSIDE"``, not + ``2``, not ``"SPMUniverseStatus.OUTSIDE"``, and not the label text — + because the export adapter normalizes a stored cell to a member name and + validates it cell-by-cell against the engine's own enum domain. + + Nothing else about the frame changes: no weight, no membership, no row. + + Args: + frame: A validated :class:`~microcosm.frame.Frame`. + unit_entity: The SPM unit entity. + declared_channel: See :func:`classify_spm_universe`. + channel_column: See :func:`classify_spm_universe`. + + Returns: + A new frame carrying the column, and the classification behind it. + + Raises: + ValueError: Any of :data:`SPM_UNIVERSE_REFUSALS`, including + ``SPM_UNIVERSE_COLUMN_EXISTS`` when any entity table already + carries the declaration — a producer overwriting an existing + declaration would silently replace someone else's source ruling, + and a copy on another entity would collide with it under the + frame's global column-name rule. + """ + from microcosm.frame import Frame + + schema = frame.schema + unit_table = frame.table(unit_entity) + carriers = [ + entity + for entity in frame.entities + if UNIVERSE_INPUT in frame.table(entity).columns + ] + _require( + not carriers, + "SPM_UNIVERSE_COLUMN_EXISTS", + f"the {carriers} table(s) already carry {UNIVERSE_INPUT!r}; this " + "producer never overwrites an existing universe declaration.", + ) + classification = classify_spm_universe( + household=frame.table("household"), + person=frame.table(schema.person_entity), + unit_ids=unit_table[schema.id_column(unit_entity)].to_numpy(), + household_id_column=schema.id_column("household"), + household_membership_column=schema.membership_column("household"), + unit_membership_column=schema.membership_column(unit_entity), + declared_channel=declared_channel, + channel_column=channel_column, + ) + tables = {entity: frame.table(entity).copy() for entity in frame.entities} + tables.update({name: frame.link(name).copy() for name in frame.links}) + tables[unit_entity][UNIVERSE_INPUT] = classification.status + declared = 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 SpmUniverseSourceResult(frame=declared, classification=classification) diff --git a/packages/microcosm-build/tests/test_us_spine_blindness.py b/packages/microcosm-build/tests/test_us_spine_blindness.py index 70924220fa..5c8e67b6d3 100644 --- a/packages/microcosm-build/tests/test_us_spine_blindness.py +++ b/packages/microcosm-build/tests/test_us_spine_blindness.py @@ -107,6 +107,10 @@ "puf_support.py", # Validates provenance at the clone boundary. "spine_agreement.py", # Pre-calibration distribution comparison. "spine_assembly.py", # New pre-operator assembly seam. + # Declares the SPM measurement universe per source arm. The ACS and + # ASEC rulings differ by construction, so the provenance tag is the + # producer's input rather than something it must be blind to. + "spm_universe_source.py", # Stacked-spine pilot (#578 revision): stacking, gap-fill donor # routing, activation authority, the completeness gate, and the # by-origin battery are origin-aware by charter. @@ -300,6 +304,9 @@ # Pinned ASEC role reconstruction and exact parent join; no population # treatment. Remains subject to the all-runtime source-identity scan. "spm_role_source.py", + # Source-owned SPM measurement-universe declaration; reads the source + # record type and no population attribute. Provenance owner above. + "spm_universe_source.py", "stacked_spine.py", # Provenance owner (#578 revision); see owners list. "support_provenance.py", "take_up.py", diff --git a/packages/microcosm-build/tests/test_us_spm_universe_source.py b/packages/microcosm-build/tests/test_us_spm_universe_source.py new file mode 100644 index 0000000000..e4718596eb --- /dev/null +++ b/packages/microcosm-build/tests/test_us_spm_universe_source.py @@ -0,0 +1,874 @@ +"""The SPM measurement-universe producer: the rule, and every refusal. + +Every fixture here is invented. Nothing in this file opens a build artifact, +a pinned source, or a release. +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.us_runtime.spm_universe_source import ( + ACS_ARM, + ACS_GROUP_QUARTERS_KINDS, + ACS_HOUSEHOLD_KIND_COLUMN, + ACS_HOUSING_UNIT_KIND, + ASEC_ARM, + ASEC_NATIVE_UNIT_COLUMN, + ASEC_RECORD_TYPE_FIELDS, + HOUSEHOLD_SPINE_COLUMN, + HOUSEHOLD_SUPPORT_CHANNEL_COLUMN, + INCLUDED, + OUTSIDE, + SPM_UNIVERSE_CHANNEL_ARMS, + SPM_UNIVERSE_REFUSALS, + UNIVERSE_INPUT, + UNRESOLVED, + attach_spm_universe_status, + classify_spm_universe, +) +from microcosm.frame import US_SCHEMA, Frame, WeightKind, Weights + +_US_RUNTIME = ( + Path(__file__).resolve().parents[1] / "src" / "microcosm" / "build" / "us_runtime" +) +_ACS_CHANNEL = "acs_2024_1yr" +_ASEC_CHANNEL = "asec_puf" + + +def _tables( + persons: list[dict[str, object]], + *, + channel_column: str = HOUSEHOLD_SPINE_COLUMN, + household_extra: dict[int, dict[str, object]] | None = None, + person_extra_columns: dict[str, list[object]] | None = None, + extra_household_rows: list[dict[str, object]] | None = None, + extra_unit_ids: list[int] | None = None, + without_support_metadata: bool = False, +) -> tuple[pd.DataFrame, pd.DataFrame, np.ndarray]: + """Household, person and SPM unit ids from a list of invented persons. + + Each person dict carries ``hid`` (household), ``unit`` (SPM unit), + ``chan`` (provenance tag), and optionally ``kind`` (``TYPEHUGQ``), + ``spm_id`` (the ASEC native unit id) and ``clone`` (clone index). + """ + prows = [] + households: dict[int, dict[str, object]] = {} + for index, spec in enumerate(persons, start=1): + hid = int(spec["hid"]) + row: dict[str, object] = { + "person_id": index, + "person_household_id": hid, + "person_spm_unit_id": int(spec["unit"]), + "person_tax_unit_id": hid, + "person_family_id": hid, + "person_marital_unit_id": index, + # ``"clone": None`` models a pool that lost the index on one arm; + # a float passes through unchanged so invalid indices can be tested. + "person_support_clone_index": ( + np.nan + if spec.get("clone", 0) is None + else spec.get("clone", 0) + if isinstance(spec.get("clone", 0), float) + else int(spec.get("clone", 0)) + ), + } + if "spm_id" in spec: + row[ASEC_NATIVE_UNIT_COLUMN] = spec["spm_id"] + prows.append(row) + household = {"household_id": hid, channel_column: spec["chan"]} + if "kind" in spec: + household[ACS_HOUSEHOLD_KIND_COLUMN] = spec["kind"] + households.setdefault(hid, household) + person = pd.DataFrame(prows) + if without_support_metadata: + person = person.drop(columns=["person_support_clone_index"]) + if person_extra_columns: + for name, values in person_extra_columns.items(): + person[name] = values + household_rows = [households[hid] for hid in sorted(households)] + if household_extra: + for row in household_rows: + row.update(household_extra.get(int(row["household_id"]), {})) + if extra_household_rows: + household_rows.extend(extra_household_rows) + household = pd.DataFrame(household_rows) + if ACS_HOUSEHOLD_KIND_COLUMN in household.columns: + household[ACS_HOUSEHOLD_KIND_COLUMN] = household[ + ACS_HOUSEHOLD_KIND_COLUMN + ].astype(object) + unit_ids = np.asarray( + sorted({int(spec["unit"]) for spec in persons} | set(extra_unit_ids or [])) + ) + return household, person, unit_ids + + +def _classify(persons: list[dict[str, object]], **kwargs) -> dict[int, str]: + table_kwargs = { + key: kwargs.pop(key) + for key in ( + "channel_column", + "household_extra", + "person_extra_columns", + "extra_household_rows", + "extra_unit_ids", + ) + if key in kwargs + } + household, person, unit_ids = _tables(persons, **table_kwargs) + result = classify_spm_universe( + household=household, person=person, unit_ids=unit_ids, **kwargs + ) + return dict(zip(result.unit_ids.tolist(), result.status.tolist(), strict=True)) + + +def _frame(persons: list[dict[str, object]], **kwargs) -> Frame: + household, person, unit_ids = _tables(persons, **kwargs) + household_ids = household["household_id"].to_numpy() + return Frame( + { + "person": person, + "household": household, + "tax_unit": pd.DataFrame({"tax_unit_id": household_ids}), + "spm_unit": pd.DataFrame({"spm_unit_id": unit_ids}), + "family": pd.DataFrame({"family_id": household_ids}), + "marital_unit": pd.DataFrame( + {"marital_unit_id": person["person_id"].to_numpy()} + ), + }, + US_SCHEMA, + { + "household": Weights( + np.full(len(household_ids), 100.0), WeightKind.CALIBRATED + ) + }, + metadata={"stage": "invented"}, + ) + + +# -------------------------------------------------------------------------- +# The rule +# -------------------------------------------------------------------------- + + +def test_acs_housing_unit_is_included(): + statuses = _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": ACS_HOUSING_UNIT_KIND}, + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": ACS_HOUSING_UNIT_KIND}, + ] + ) + assert statuses == {1: INCLUDED} + + +@pytest.mark.parametrize("kind", ACS_GROUP_QUARTERS_KINDS) +def test_acs_group_quarters_are_outside_for_each_code(kind): + statuses = _classify([{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": kind}]) + assert statuses == {1: OUTSIDE} + + +def test_asec_records_are_included(): + """The ASEC ruling is spine-level: no record-type field is consulted.""" + statuses = _classify( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 1, "unit": 2, "chan": _ASEC_CHANNEL, "spm_id": 12}, + {"hid": 2, "unit": 3, "chan": _ASEC_CHANNEL, "spm_id": 13}, + ] + ) + assert statuses == {1: INCLUDED, 2: INCLUDED, 3: INCLUDED} + + +def test_a_zero_adult_housing_unit_is_included(): + """The anti-inference test. + + A housing unit whose only members are children is a household-unit data + defect, not an out-of-universe record. Marking it ``OUTSIDE`` would hide + the defect and make the universe move whenever the role column moved. Only + the record type decides: the identical composition under ``TYPEHUGQ`` 2 is + ``OUTSIDE``, and it is the record type that moved it. + """ + children = [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": ACS_HOUSING_UNIT_KIND}, + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": ACS_HOUSING_UNIT_KIND}, + ] + ages = {"age": [3, 7], "is_spm_independent_minor_role": [False, False]} + assert _classify(children, person_extra_columns=ages) == {1: INCLUDED} + + group_quarters = [dict(spec, kind=2) for spec in children[:1]] + assert _classify(group_quarters, person_extra_columns={"age": [3]}) == {1: OUTSIDE} + + +def test_the_declaration_is_invariant_to_age_and_role(): + """Composition cannot move the answer, in either direction.""" + persons = [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": ACS_HOUSING_UNIT_KIND}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 2}, + ] + children = _classify( + persons, + person_extra_columns={ + "age": [2, 3], + "is_spm_independent_minor_role": [False, False], + }, + ) + adults = _classify( + persons, + person_extra_columns={ + "age": [44, 51], + "is_spm_independent_minor_role": [True, True], + }, + ) + assert children == adults == {1: INCLUDED, 2: OUTSIDE} + + +def test_a_clone_inherits_its_native_rows_status(): + """Cloning a record carries its record type, so it carries its universe.""" + statuses = _classify( + [ + # Native group-quarters placeholder and its clone. + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": 0}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 2, "clone": 1}, + # Native housing unit and its clone. + {"hid": 3, "unit": 3, "chan": _ACS_CHANNEL, "kind": 1, "clone": 0}, + {"hid": 4, "unit": 4, "chan": _ACS_CHANNEL, "kind": 1, "clone": 1}, + ] + ) + assert statuses == {1: OUTSIDE, 2: OUTSIDE, 3: INCLUDED, 4: INCLUDED} + + +def test_a_pooled_frame_declares_each_arm_from_its_own_rule(): + household, person, unit_ids = _tables( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 3}, + {"hid": 3, "unit": 3, "chan": _ASEC_CHANNEL, "spm_id": 31}, + ] + ) + result = classify_spm_universe( + household=household, person=person, unit_ids=unit_ids + ) + assert result.status.tolist() == [INCLUDED, OUTSIDE, INCLUDED] + assert result.provenance["households_by_arm"] == {ACS_ARM: 2, ASEC_ARM: 1} + assert result.provenance["households_by_acs_record_type"] == {1: 1, 3: 1} + assert result.provenance["acs_group_quarters_households"] == 1 + assert result.provenance["units_included"] == 2 + assert result.provenance["units_outside"] == 1 + assert result.provenance["units_unresolved"] == 0 + + +def test_the_support_channel_column_is_read_when_no_spine_tag_is_present(): + statuses = _classify( + [ + {"hid": 1, "unit": 1, "chan": "acs", "kind": 1}, + {"hid": 2, "unit": 2, "chan": "asec", "spm_id": 21}, + ], + channel_column=HOUSEHOLD_SUPPORT_CHANNEL_COLUMN, + ) + assert statuses == {1: INCLUDED, 2: INCLUDED} + + +def test_the_spine_tag_takes_precedence_over_the_support_channel(): + """A PUF clone's support channel names its operator role, not its spine.""" + statuses = _classify( + [{"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}], + household_extra={1: {HOUSEHOLD_SUPPORT_CHANNEL_COLUMN: "puf_tax_detail"}}, + ) + assert statuses == {1: INCLUDED} + + +def test_an_explicit_channel_column_overrides_the_default_order(): + household, person, unit_ids = _tables( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2}], + household_extra={1: {"operator_channel": _ACS_CHANNEL}}, + ) + result = classify_spm_universe( + household=household.drop(columns=[HOUSEHOLD_SPINE_COLUMN]).assign( + **{HOUSEHOLD_SUPPORT_CHANNEL_COLUMN: "puf_tax_detail"} + ), + person=person, + unit_ids=unit_ids, + channel_column="operator_channel", + ) + assert result.status.tolist() == [OUTSIDE] + assert result.provenance["channel_source"] == "operator_channel" + + +@pytest.mark.parametrize( + ("channel_column", "declared_channel", "reason"), + [ + ("operator_channel", None, "no requested channel column"), + (HOUSEHOLD_SPINE_COLUMN, _ACS_CHANNEL, "mutually exclusive"), + ], +) +def test_refuses_a_bad_explicit_channel_column( + channel_column, declared_channel, reason +): + household, person, unit_ids = _tables( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}] + ) + with pytest.raises( + ValueError, match=rf"^SPM_UNIVERSE_UNDECLARED_SPINE: .*{reason}" + ): + classify_spm_universe( + household=household, + person=person, + unit_ids=unit_ids, + channel_column=channel_column, + declared_channel=declared_channel, + ) + + +def test_an_untagged_table_accepts_an_explicit_caller_declaration(): + household = pd.DataFrame( + {"household_id": [1], ACS_HOUSEHOLD_KIND_COLUMN: [2]}, + ) + person = pd.DataFrame( + {"person_id": [1], "person_household_id": [1], "person_spm_unit_id": [1]} + ) + result = classify_spm_universe( + household=household, + person=person, + unit_ids=np.asarray([1]), + declared_channel=_ACS_CHANNEL, + ) + assert result.status.tolist() == [OUTSIDE] + assert result.provenance["channel_source"] == f"declared:{_ACS_CHANNEL}" + + +# -------------------------------------------------------------------------- +# Stored encoding +# -------------------------------------------------------------------------- + + +def test_the_status_is_stored_as_the_enum_member_name(): + household, person, unit_ids = _tables( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 2}, + ] + ) + status = classify_spm_universe( + household=household, person=person, unit_ids=unit_ids + ).status + assert status.dtype == np.dtype(object) + assert status.tolist() == ["INCLUDED", "OUTSIDE"] + # Negative controls: not the integer index, not repr(), not the label. + assert 2 not in set(status.tolist()) + assert "SPMUniverseStatus.OUTSIDE" not in set(status.tolist()) + assert not any(" " in value for value in status.tolist()) + # UNRESOLVED is the engine's absence sentinel; a producer never emits it. + assert UNRESOLVED not in set(status.tolist()) + + +# -------------------------------------------------------------------------- +# Refusals +# -------------------------------------------------------------------------- + + +def _refuses(code: str): + return pytest.raises(ValueError, match=rf"^{code}: ") + + +def test_refuses_a_preexisting_declaration(): + frame = _frame([{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}]) + declared = attach_spm_universe_status(frame).frame + with _refuses("SPM_UNIVERSE_COLUMN_EXISTS"): + attach_spm_universe_status(declared) + + +def test_refuses_a_declaration_already_on_another_entity_table(): + frame = _frame( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}], + person_extra_columns={UNIVERSE_INPUT: [INCLUDED]}, + ) + with _refuses("SPM_UNIVERSE_COLUMN_EXISTS"): + attach_spm_universe_status(frame) + + +def test_refuses_a_channel_with_no_universe_ruling(): + with _refuses("SPM_UNIVERSE_UNDECLARED_SPINE"): + _classify([{"hid": 1, "unit": 1, "chan": "puf_tax_detail", "kind": 1}]) + + +def test_refuses_a_frame_with_no_provenance_tag_and_no_declaration(): + household = pd.DataFrame({"household_id": [1], ACS_HOUSEHOLD_KIND_COLUMN: [1]}) + person = pd.DataFrame( + {"person_id": [1], "person_household_id": [1], "person_spm_unit_id": [1]} + ) + with _refuses("SPM_UNIVERSE_UNDECLARED_SPINE"): + classify_spm_universe( + household=household, person=person, unit_ids=np.asarray([1]) + ) + + +def test_refuses_a_caller_declaration_over_a_tagged_frame(): + with _refuses("SPM_UNIVERSE_UNDECLARED_SPINE"): + _classify( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}], + declared_channel=_ASEC_CHANNEL, + ) + + +@pytest.mark.parametrize("kind", [4, 0, -1, None, float("nan")]) +def test_refuses_an_unknown_acs_record_type(kind): + with _refuses("SPM_UNIVERSE_UNKNOWN_HOUSEHOLD_KIND"): + _classify([{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": kind}]) + + +def test_refuses_an_acs_record_type_on_a_non_acs_row(): + """``TYPEHUGQ == 1`` means "ACS housing unit", never "not group quarters".""" + with _refuses("SPM_UNIVERSE_OFF_ARM_HOUSEHOLD_KIND"): + _classify( + [{"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "kind": 1, "spm_id": 11}] + ) + + +def test_refuses_a_household_resolving_to_more_than_one_kind(): + household = pd.DataFrame( + { + "household_id": [1, 1], + HOUSEHOLD_SPINE_COLUMN: [_ACS_CHANNEL, _ACS_CHANNEL], + ACS_HOUSEHOLD_KIND_COLUMN: [1, 2], + } + ) + person = pd.DataFrame( + {"person_id": [1], "person_household_id": [1], "person_spm_unit_id": [1]} + ) + with _refuses("SPM_UNIVERSE_MIXED_HOUSEHOLD"): + classify_spm_universe( + household=household, person=person, unit_ids=np.asarray([1]) + ) + + +def test_refuses_a_group_quarters_placeholder_with_two_native_persons(): + with _refuses("SPM_UNIVERSE_GQ_MULTI_PERSON"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": 0}, + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": 0}, + ] + ) + + +def test_a_group_quarters_placeholder_with_a_clone_is_accepted(): + """Negative control for the refusal above: clones are expected.""" + statuses = _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": 0}, + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": 1}, + ] + ) + assert statuses == {1: OUTSIDE} + + +def test_refuses_a_member_of_no_declared_unit(): + with _refuses("SPM_UNIVERSE_ORPHAN_MEMBER"): + household, person, _ = _tables( + [{"hid": 1, "unit": 7, "chan": _ACS_CHANNEL, "kind": 1}] + ) + classify_spm_universe( + household=household, person=person, unit_ids=np.asarray([1]) + ) + + +def test_refuses_a_member_of_no_declared_household(): + household, person, unit_ids = _tables( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}] + ) + person["person_household_id"] = [9] + with _refuses("SPM_UNIVERSE_ORPHAN_MEMBER"): + classify_spm_universe(household=household, person=person, unit_ids=unit_ids) + + +def test_refuses_a_unit_with_no_members(): + with _refuses("SPM_UNIVERSE_EMPTY_UNIT"): + _classify( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}], + extra_unit_ids=[2], + ) + + +def test_refuses_a_unit_spanning_two_record_types(): + """Natively unreachable; a majority vote here would fabricate a universe.""" + with _refuses("SPM_UNIVERSE_UNIT_SPANS_KINDS"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2}, + ] + ) + + +def test_refuses_a_unit_spanning_two_arms(): + with _refuses("SPM_UNIVERSE_UNIT_SPANS_KINDS"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 21}, + ] + ) + + +def test_refuses_an_asec_arm_with_no_native_unit_column(): + with _refuses("SPM_UNIVERSE_DEGRADED_PARTITION"): + _classify([{"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL}]) + + +def test_refuses_a_missing_native_unit_id_on_the_asec_arm(): + """One missing ``SPM_ID`` degrades every SPM unit in the frame.""" + with _refuses("SPM_UNIVERSE_DEGRADED_PARTITION"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 2, "unit": 2, "chan": _ASEC_CHANNEL, "spm_id": np.nan}, + ] + ) + + +def test_refuses_an_asec_partition_that_is_not_the_native_one(): + """Two native units collapsed into one frame unit: the fallback's shape.""" + with _refuses("SPM_UNIVERSE_DEGRADED_PARTITION"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 12}, + ] + ) + + +def test_a_support_cloned_asec_frame_keeps_its_native_partition(): + """Each support-clone copy is its own instance of the native partition. + + ``puf_support._clone_entity_table`` deep-copies the person table and remaps + only the id and membership columns, so clone copy 1 carries the native + ``SPM_ID`` under new unit ids. Keying the check on ``SPM_ID`` alone read + every support-cloned ASEC frame as degraded. + """ + statuses = _classify( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11, "clone": 0}, + {"hid": 1, "unit": 2, "chan": _ASEC_CHANNEL, "spm_id": 12, "clone": 0}, + {"hid": 2, "unit": 3, "chan": _ASEC_CHANNEL, "spm_id": 21, "clone": 0}, + {"hid": 11, "unit": 11, "chan": _ASEC_CHANNEL, "spm_id": 11, "clone": 1}, + {"hid": 11, "unit": 12, "chan": _ASEC_CHANNEL, "spm_id": 12, "clone": 1}, + {"hid": 12, "unit": 13, "chan": _ASEC_CHANNEL, "spm_id": 21, "clone": 1}, + ] + ) + assert statuses == dict.fromkeys((1, 2, 3, 11, 12, 13), INCLUDED) + + +@pytest.mark.parametrize( + "clone_rows", + [ + pytest.param( + [ + {"hid": 11, "unit": 11, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 11, "unit": 11, "chan": _ASEC_CHANNEL, "spm_id": 12}, + ], + id="two-native-units-collapsed-inside-the-clone-copy", + ), + pytest.param( + [ + {"hid": 11, "unit": 11, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 11, "unit": 12, "chan": _ASEC_CHANNEL, "spm_id": 11}, + ], + id="one-native-unit-split-inside-the-clone-copy", + ), + ], +) +def test_the_clone_copy_key_still_refuses_a_degraded_clone_copy(clone_rows): + """Negative control: keying on the copy does not relax the check within it.""" + native = [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11, "clone": 0}, + {"hid": 1, "unit": 2, "chan": _ASEC_CHANNEL, "spm_id": 12, "clone": 0}, + ] + with _refuses("SPM_UNIVERSE_DEGRADED_PARTITION"): + _classify(native + [dict(row, clone=1) for row in clone_rows]) + + +def test_a_missing_clone_index_is_refused_on_the_asec_arm(): + """A clone row that lost its index has an undecidable copy: refused.""" + with _refuses("SPM_UNIVERSE_INVALID_CLONE_INDEX"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11, "clone": 0}, + { + "hid": 11, + "unit": 11, + "chan": _ASEC_CHANNEL, + "spm_id": 11, + "clone": None, + }, + ] + ) + + +def test_a_missing_clone_index_is_refused_in_group_quarters(): + """A lost clone index cannot make the one-person GQ check pass vacuously.""" + with _refuses("SPM_UNIVERSE_INVALID_CLONE_INDEX"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": None}, + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 2, "clone": None}, + ] + ) + + +def test_a_missing_clone_index_cannot_collapse_a_cross_copy_collision(): + """The gate's counterexample: two copies sharing one frame unit. + + With both indices present the rows are two native keys in one frame unit + and the partition check refuses. Reading the missing indices as native + would merge them into one key and accept; the module refuses first. + """ + rows = [ + {"hid": 1, "unit": 5, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 1, "unit": 5, "chan": _ASEC_CHANNEL, "spm_id": 11}, + ] + with _refuses("SPM_UNIVERSE_DEGRADED_PARTITION"): + _classify([dict(rows[0], clone=1), dict(rows[1], clone=2)]) + with _refuses("SPM_UNIVERSE_INVALID_CLONE_INDEX"): + _classify([dict(row, clone=None) for row in rows]) + + +@pytest.mark.parametrize("bad", [-1.0, 0.5, float("inf")]) +def test_an_invalid_clone_index_is_refused(bad): + with _refuses("SPM_UNIVERSE_INVALID_CLONE_INDEX"): + _classify( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11, "clone": 0}, + { + "hid": 2, + "unit": 2, + "chan": _ASEC_CHANNEL, + "spm_id": 11, + "clone": bad, + }, + ] + ) + + +@pytest.mark.requires_us +def test_the_real_support_clone_operator_output_is_accepted(): + """The copy-key rule holds on ``clone_us_frame_for_puf_support`` itself. + + Invented rows. Imported lazily: the operator's module pulls the country + engine, which the rest of this file never needs. + """ + from microcosm.build.us_runtime.puf_support import ( + clone_us_frame_for_puf_support, + ) + + frame = _frame( + [ + {"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}, + {"hid": 1, "unit": 2, "chan": _ASEC_CHANNEL, "spm_id": 12}, + {"hid": 2, "unit": 3, "chan": _ASEC_CHANNEL, "spm_id": 21}, + ], + without_support_metadata=True, + ) + cloned = clone_us_frame_for_puf_support(frame) + person = cloned.table("person") + # The operator's shape this rule depends on: SPM_ID copied verbatim onto + # a second copy whose SPM units are new ids. + assert sorted(person["person_support_clone_index"].unique().tolist()) == [0, 1] + by_copy = person.groupby("person_support_clone_index") + assert by_copy[ASEC_NATIVE_UNIT_COLUMN].apply(sorted).tolist() == [ + [11, 12, 21], + [11, 12, 21], + ] + assert person["person_spm_unit_id"].nunique() == 6 + + result = attach_spm_universe_status(cloned) + assert result.status.tolist() == [INCLUDED] * 6 + + +def test_the_degraded_partition_check_does_not_fire_on_the_acs_arm(): + """ACS PUMS supplies no ``SPM_ID``; one unit per household is native there.""" + assert _classify( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 1}, + ] + ) == {1: INCLUDED, 2: INCLUDED} + + +@pytest.mark.parametrize("field", ASEC_RECORD_TYPE_FIELDS) +def test_refuses_an_asec_vintage_carrying_a_record_type_field(field): + with _refuses("SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED"): + _classify( + [{"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}], + household_extra={1: {field: 1}}, + ) + + +def test_refuses_an_asec_record_type_field_on_the_person_table(): + with _refuses("SPM_UNIVERSE_ASEC_RECORD_TYPE_UNREVIEWED"): + _classify( + [{"hid": 1, "unit": 1, "chan": _ASEC_CHANNEL, "spm_id": 11}], + person_extra_columns={"H_TYPE": [1]}, + ) + + +def test_an_acs_only_frame_is_unaffected_by_the_asec_record_type_guard(): + """Negative control: the guard is scoped to the arm whose ruling it guards.""" + assert _classify( + [{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}], + household_extra={1: {"H_LIVQRT": 1}}, + ) == {1: INCLUDED} + + +# -------------------------------------------------------------------------- +# The frame wrapper +# -------------------------------------------------------------------------- + + +def test_attach_declares_the_status_and_changes_nothing_else(): + frame = _frame( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 2}, + {"hid": 3, "unit": 3, "chan": _ACS_CHANNEL, "kind": 3}, + ] + ) + result = attach_spm_universe_status(frame) + declared = result.frame + + assert declared.table("spm_unit")[UNIVERSE_INPUT].tolist() == [ + INCLUDED, + OUTSIDE, + OUTSIDE, + ] + # The input frame is untouched: this is a new frame, not a mutation. + assert UNIVERSE_INPUT not in frame.table("spm_unit").columns + # No weight, no membership, no row moves. + assert declared.weights_for("household").total == pytest.approx( + frame.weights_for("household").total + ) + for entity in frame.entities: + assert declared.n(entity) == frame.n(entity) + before = frame.table(entity) + after = declared.table(entity) + assert set(after.columns) - set(before.columns) <= {UNIVERSE_INPUT} + pd.testing.assert_frame_equal(after[before.columns], before) + assert declared.metadata == frame.metadata + assert declared.mass_log == frame.mass_log + + +def test_attach_reports_the_classification_it_declared(): + frame = _frame( + [ + {"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}, + {"hid": 2, "unit": 2, "chan": _ACS_CHANNEL, "kind": 2}, + ] + ) + result = attach_spm_universe_status(frame) + assert result.status.tolist() == [INCLUDED, OUTSIDE] + assert result.provenance["declared_variable"] == UNIVERSE_INPUT + assert result.provenance["declared_entity"] == "spm_unit" + assert result.provenance["stored_encoding"] == "enum member name string" + + +def test_the_provenance_states_what_the_producer_is_not(): + frame = _frame([{"hid": 1, "unit": 1, "chan": _ACS_CHANNEL, "kind": 1}]) + provenance = attach_spm_universe_status(frame).provenance + assert provenance["weights_used"] is False + assert provenance["ages_changed"] is False + assert provenance["composition_read"] is False + assert provenance["is_calibration_target"] is False + assert provenance["is_spine_agreement_surface"] is False + + +# -------------------------------------------------------------------------- +# Constants parity and refusal-code completeness +# -------------------------------------------------------------------------- + + +def _module_constant(module: str, name: str) -> str: + """Read one module-level string constant without importing the module. + + ``acs_pums`` and ``stacked_spine`` both pull the country engine and cost + tens of seconds to import; parsing is exact and free. + """ + tree = ast.parse((_US_RUNTIME / module).read_text()) + for node in tree.body: + if isinstance(node, ast.Assign) and any( + isinstance(target, ast.Name) and target.id == name + for target in node.targets + ): + assert isinstance(node.value, ast.Constant), (module, name) + return node.value.value + raise AssertionError(f"{module} declares no module-level {name}") + + +@pytest.mark.parametrize( + ("module", "constant", "arm"), + [ + ("acs_pums.py", "ACS_2024_1YR_SPINE", ACS_ARM), + ("stacked_spine.py", "ACS_STACKED_SUPPORT_CHANNEL", ACS_ARM), + ("base_pool.py", "ASEC_PUF_SPINE", ASEC_ARM), + ("support_provenance.py", "BASE_ASEC_SUPPORT_CHANNEL", ASEC_ARM), + ], +) +def test_every_ruled_channel_matches_its_canonical_definition(module, constant, arm): + value = _module_constant(module, constant) + assert SPM_UNIVERSE_CHANNEL_ARMS.get(value) == arm, ( + f"{module}:{constant} == {value!r}, which this producer rules as " + f"{SPM_UNIVERSE_CHANNEL_ARMS.get(value)!r} rather than {arm!r}" + ) + + +def test_the_ruling_table_declares_nothing_beyond_the_canonical_channels(): + canonical = { + _module_constant("acs_pums.py", "ACS_2024_1YR_SPINE"), + _module_constant("stacked_spine.py", "ACS_STACKED_SUPPORT_CHANNEL"), + _module_constant("base_pool.py", "ASEC_PUF_SPINE"), + _module_constant("support_provenance.py", "BASE_ASEC_SUPPORT_CHANNEL"), + } + assert set(SPM_UNIVERSE_CHANNEL_ARMS) == canonical + + +def test_the_spine_column_name_matches_its_canonical_factory(): + tree = ast.parse((_US_RUNTIME / "base_pool.py").read_text()) + factory = next( + node + for node in tree.body + if isinstance(node, ast.FunctionDef) and node.name == "spine_column" + ) + template = next( + node for node in ast.walk(factory) if isinstance(node, ast.JoinedStr) + ) + suffix = "".join( + part.value for part in template.values if isinstance(part, ast.Constant) + ) + assert HOUSEHOLD_SPINE_COLUMN == f"household{suffix}" + + +def test_every_refusal_the_module_raises_is_declared(): + """No refusal code may exist only inside a message string.""" + source = (_US_RUNTIME / "spm_universe_source.py").read_text() + tree = ast.parse(source) + raised = set() + for node in ast.walk(tree): + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "_require" + and len(node.args) >= 2 + and isinstance(node.args[1], ast.Constant) + ): + raised.add(node.args[1].value) + assert raised, "the producer raises no named refusal at all" + assert raised <= set(SPM_UNIVERSE_REFUSALS), sorted( + raised - set(SPM_UNIVERSE_REFUSALS) + ) + assert len(SPM_UNIVERSE_REFUSALS) == len(set(SPM_UNIVERSE_REFUSALS)) + # SPM_UNIVERSE_COLUMN_EXISTS is raised by the frame wrapper, every other + # declared code by the classifier; together they must cover the tuple. + assert raised == set(SPM_UNIVERSE_REFUSALS)