Map ACS local Arch facts into Microplex targets

Author: MaxGhenis

src/microplex_us/targets/arch.py

@@ -573,6 +573,8 @@
         "medicare_part_b_premiums",
         "AMOUNT",
     ),
+    "census_acs.household_count": ("household_count", "COUNT"),
+    "census_acs.person_count": ("population", "COUNT"),
     "census_decennial.resident_population": ("population", "COUNT"),
     "census_decennial.occupied_housing_units": ("household_count", "COUNT"),
     "census_pep.resident_population": ("population", "COUNT"),
@@ -781,10 +783,12 @@
     "tanf_cash_assistance": (),
     "tanf_caseload": (),
     "liheap_state_programs": (),
+    "total_population": (),
 }
 
 ARCH_FACT_CONSTRAINT_VARIABLE_ALIASES = {
     "age": "age",
+    "snap_receipt_status": "snap_receipt_status",
     "us.tax.earned_income_credit_qualifying_children": "eitc_child_count",
     "us_social_security_and_ssi.program_payment_type": "program_payment_type",
     "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income",
@@ -4870,6 +4874,11 @@ def _arch_constraint_display_label(
         return f"state senate district {value_text}"
     if variable == "sldl_id" and canonical_operator == "==":
         return f"state house district {value_text}"
+    if variable == "snap_receipt_status" and canonical_operator == "==":
+        if value_text == "receiving_food_stamps_snap":
+            return "SNAP > 0"
+        if value_text == "not_receiving_food_stamps_snap":
+            return "SNAP = 0"
     positive_feature = ARCH_POSITIVE_CONSTRAINT_ALIASES.get(variable)
     if positive_feature is not None and canonical_operator == "==":
         feature_label = _humanize_snake_label(positive_feature)
@@ -4936,6 +4945,13 @@ def _canonical_filters_for_arch_constraints(
                 )
             )
             continue
+        if variable == "snap_receipt_status":
+            if value == "receiving_food_stamps_snap" and canonical_operator == "==":
+                filters.append(TargetFilter(feature="snap", operator=">", value=0))
+                continue
+            if value == "not_receiving_food_stamps_snap" and canonical_operator == "==":
+                filters.append(TargetFilter(feature="snap", operator="==", value=0))
+                continue
         positive_feature = ARCH_POSITIVE_CONSTRAINT_ALIASES.get(variable)
         if positive_feature is not None:
             filters.append(
@@ -5017,7 +5033,7 @@ def _county_fips_from_arch_geography_id(geography_id: Any) -> str:
 
 def _congressional_district_from_arch_geography_id(geography_id: Any) -> str:
     raw = str(geography_id)
-    if raw.startswith("5001800US"):
+    if raw.startswith(("5001800US", "5001900US")):
         return raw[-4:]
     return raw
 
@@ -5662,6 +5678,12 @@ def _arch_gap_expected_source(cell: dict[str, Any]) -> str | None:
         return "BEA"
     if variable == "tax_unit_count" and "aca_ptc" in domain_variables:
         return "IRS_SOI"
+    if (
+        variable in {"household_count", "person_count"}
+        and "snap" in domain_variables
+        and _normalize_geo_level(cell.get("geo_level")) == "district"
+    ):
+        return "CENSUS_ACS"
     if variable == "snap" or "snap" in domain_variables:
         return "USDA_SNAP"
     if variable == "tanf" or "tanf" in domain_variables:
@@ -5683,6 +5705,11 @@ def _arch_gap_expected_source(cell: dict[str, Any]) -> str | None:
     if variable == "person_count":
         if _normalize_geo_level(cell.get("geo_level")) in {"sldu", "sldl"}:
             return "CENSUS_DECENNIAL"
+        if (
+            _normalize_geo_level(cell.get("geo_level")) == "district"
+            and "age" in domain_variables
+        ):
+            return "CENSUS_ACS"
         if "adjusted_gross_income" in domain_variables:
             return "IRS_SOI"
         if "age" in domain_variables or not domain_variables:

tests/targets/test_arch_facts.py

@@ -1492,6 +1492,148 @@ def test_arch_consumer_fact_jsonl_provider_maps_state_soi_rows(
     }
 
 
+def test_arch_consumer_fact_jsonl_provider_maps_acs_district_age_rows(
+    tmp_path: Path,
+) -> None:
+    consumer_jsonl = tmp_path / "consumer_facts.jsonl"
+    rows = [
+        _consumer_fact(
+            "acs-cd-al01-age-0-4",
+            concept="census_acs.person_count",
+            domain="total_population",
+            source_name="census_acs",
+            source_table="ACS S0101 congressional district age",
+            period={"type": "calendar_year", "value": 2024},
+            geography={
+                "level": "congressional_district",
+                "id": "5001900US0101",
+                "name": "Congressional District 1 (119th Congress), Alabama",
+            },
+            value=39_908,
+            constraints=(
+                {
+                    "variable": "age",
+                    "operator": ">=",
+                    "value": 0,
+                    "unit": "years",
+                    "role": "filter",
+                },
+                {
+                    "variable": "age",
+                    "operator": "<",
+                    "value": 5,
+                    "unit": "years",
+                    "role": "filter",
+                },
+            ),
+        ),
+    ]
+    consumer_jsonl.write_text(
+        "\n".join(json.dumps(row, sort_keys=True) for row in rows) + "\n"
+    )
+
+    target_set = ArchConsumerFactJSONLTargetProvider(consumer_jsonl).load_target_set(
+        TargetQuery(
+            period=2024,
+            provider_filters={
+                "sources": ["CENSUS_ACS"],
+                "target_cells": [
+                    {
+                        "variable": "person_count",
+                        "geo_level": "district",
+                        "geographic_id": "0101",
+                        "domain_variable": "age",
+                    },
+                ],
+            },
+        )
+    )
+
+    assert len(target_set.targets) == 1
+    target = target_set.targets[0]
+    assert target.value == 39_908
+    assert target.metadata["variable"] == "person_count"
+    assert target.metadata["geo_level"] == "district"
+    assert target.metadata["source"] == "CENSUS_ACS"
+    assert _target_filter_tuples(target) == {
+        ("age", ">=", "0"),
+        ("age", "<", "5"),
+        ("congressional_district_geoid", "==", "0101"),
+    }
+
+
+def test_arch_consumer_fact_jsonl_provider_maps_acs_district_snap_rows(
+    tmp_path: Path,
+) -> None:
+    consumer_jsonl = tmp_path / "consumer_facts.jsonl"
+    geography = {
+        "level": "congressional_district",
+        "id": "5001900US0101",
+        "name": "Congressional District 1 (119th Congress), Alabama",
+    }
+    rows = [
+        _consumer_fact(
+            "acs-cd-al01-households-total",
+            concept="census_acs.household_count",
+            domain="households",
+            source_name="census_acs",
+            source_table="ACS S2201 congressional district SNAP households",
+            period={"type": "calendar_year", "value": 2024},
+            geography=geography,
+            value=300_636,
+        ),
+        _consumer_fact(
+            "acs-cd-al01-households-snap",
+            concept="census_acs.household_count",
+            domain="households",
+            source_name="census_acs",
+            source_table="ACS S2201 congressional district SNAP households",
+            period={"type": "calendar_year", "value": 2024},
+            geography=geography,
+            value=34_742,
+            constraints=(
+                {
+                    "variable": "snap_receipt_status",
+                    "operator": "==",
+                    "value": "receiving_food_stamps_snap",
+                    "role": "filter",
+                },
+            ),
+        ),
+    ]
+    consumer_jsonl.write_text(
+        "\n".join(json.dumps(row, sort_keys=True) for row in rows) + "\n"
+    )
+
+    target_set = ArchConsumerFactJSONLTargetProvider(consumer_jsonl).load_target_set(
+        TargetQuery(
+            period=2024,
+            provider_filters={
+                "sources": ["CENSUS_ACS"],
+                "target_cells": [
+                    {
+                        "variable": "household_count",
+                        "geo_level": "district",
+                        "geographic_id": "0101",
+                        "domain_variable": "snap",
+                    },
+                ],
+            },
+        )
+    )
+
+    assert len(target_set.targets) == 1
+    target = target_set.targets[0]
+    assert target.value == 34_742
+    assert target.metadata["variable"] == "household_count"
+    assert target.metadata["geo_level"] == "district"
+    assert target.metadata["source"] == "CENSUS_ACS"
+    assert _target_filter_tuples(target) == {
+        ("congressional_district_geoid", "==", "0101"),
+        ("snap", ">", "0"),
+    }
+
+
 def test_arch_consumer_fact_jsonl_provider_maps_state_broad_soi_concepts(
     tmp_path: Path,
 ) -> None: