Merge pull request #853 from PolicyEngine/maria/pipeline_fix

Fix Modal data-build pipeline failures (TANF retry, rent ETERNITY workaround, db diagnostics)

Skipping, given the H5 publication pathway is broken anyway

Author: anth-volk

changelog.d/cps-rent-eternity-workaround.fixed.md

@@ -0,0 +1 @@
+Read `is_household_head` directly from the ACS and CPS H5 datasets in `add_rent` and require `policyengine-core>=3.25.4` for PolicyEngine/policyengine-core#482, where user-supplied ETERNITY inputs were dropped after `_invalidate_all_caches`. Removes the empty-train-frame `ValueError` from `sample(10_000)` that was failing the Modal CPS build.

changelog.d/modal-pipeline-build-failures.fixed.md

@@ -0,0 +1 @@
+Improve Modal data-build diagnostics so database-step failures preserve their stdout and stderr in the build log.

changelog.d/tanf-acf-retry.fixed.md

@@ -0,0 +1 @@
+Skip the flaky ACF HTML landing page in `etl_tanf` and fetch the FY-stamped workbooks directly via a per-year `TANF_WORKBOOK_URLS` constant; keep the tenacity retry / extended timeout around the workbook GET so transient `acf.gov` slowness does not fail `make database` on Modal builds.

modal_app/data_build.py

@@ -479,12 +479,10 @@ def build_datasets(
         log_file=log_file,
     )
     # Build policy_data.db from source
-    subprocess.run(
-        ["make", "database"],
-        check=True,
-        cwd="/root/policyengine-us-data",
-        env=env,
-    )
+    env["PYTHONUNBUFFERED"] = "1"
+    log_file.write(f"\n{'=' * 60}\nStarting make database...\n{'=' * 60}\n")
+    log_file.flush()
+    run_script_logged(["make", "database"], log_file, env)
     # Checkpoint policy_data.db immediately after build so it survives
     # test failures and can be restored on retries.
     save_checkpoint(

policyengine_us_data/datasets/cps/cps.py

@@ -262,6 +262,19 @@ def add_rent(self, cps: h5py.File, person: DataFrame, household: DataFrame):
     ]
     IMPUTATIONS = ["rent", "real_estate_taxes"]
     train_df = acs.calculate_dataframe(PREDICTORS + IMPUTATIONS, map_to="person")
+    # TODO(PolicyEngine/policyengine-core#482): policyengine-core 3.24.0+
+    # silently drops user-supplied ETERNITY inputs on dataset reload because
+    # _user_input_keys records the user-supplied period instead of the
+    # canonicalized ETERNITY key. is_household_head therefore comes back as
+    # all False from calculate_dataframe and the household-head filter below
+    # produces an empty frame. For ACS we read it directly from the source
+    # H5; for CPS we use the in-memory dict (already populated upstream in
+    # add_id_variables). Remove both overrides once pyproject.toml's
+    # policyengine-core upper bound is lifted.
+    with h5py.File(ACS_2022.file_path, "r") as acs_h5:
+        train_df["is_household_head"] = np.asarray(
+            acs_h5["is_household_head"], dtype=bool
+        )
     train_df.tenure_type = train_df.tenure_type.map(
         {
             "OWNED_OUTRIGHT": "OWNED_WITH_MORTGAGE",
@@ -270,6 +283,7 @@ def add_rent(self, cps: h5py.File, person: DataFrame, household: DataFrame):
     ).fillna(train_df.tenure_type)
     train_df = train_df[train_df.is_household_head].sample(10_000)
     inference_df = cps_sim.calculate_dataframe(PREDICTORS, map_to="person")
+    inference_df["is_household_head"] = np.asarray(cps["is_household_head"], dtype=bool)
     mask = inference_df.is_household_head.values
     inference_df = inference_df[mask]
 

policyengine_us_data/db/etl_tanf.py

@@ -5,6 +5,13 @@
 import pandas as pd
 import requests
 from sqlmodel import Session, create_engine
+from tenacity import (
+    before_sleep_log,
+    retry,
+    retry_if_exception_type,
+    stop_after_attempt,
+    wait_exponential,
+)
 
 from policyengine_us_data.storage import STORAGE_FOLDER
 from policyengine_us_data.db.create_database_tables import (
@@ -19,14 +26,53 @@
 logger = logging.getLogger(__name__)
 
 ACF_DATA_YEAR = 2024
-CASELOAD_PAGE_URL = "https://www.acf.hhs.gov/ofa/data/tanf-caseload-data-2024"
-FINANCIAL_PAGE_URL = "https://www.acf.hhs.gov/ofa/data/tanf-financial-data-fy-2024"
-CASELOAD_URL_PATTERN = re.compile(
-    r"https://acf\.gov/sites/default/files/documents/ofa/fy\d{4}_tanf_caseload\.xlsx"
-)
-FINANCIAL_URL_PATTERN = re.compile(
-    r"https://acf\.gov/sites/default/files/documents/ofa/fy-\d{4}-tanf-moe-financial-data\.xlsx"
+ACF_REQUEST_TIMEOUT = 60
+
+# Direct URLs for the FY-stamped ACF workbooks. The previous implementation
+# scraped the HTML landing page (`acf.gov/ofa/data/tanf-...`) to discover
+# these links, but that page is intermittently unreachable on `acf.gov` and
+# was the dominant source of `make database` build failures (see #852). The
+# workbook URLs themselves on `acf.gov/sites/default/files/documents/ofa/`
+# return 200 reliably, so we hit them directly and skip the page entirely.
+#
+# Update this dict when:
+#   - ACF publishes a new fiscal year's workbooks (add a new top-level key
+#     and bump `ACF_DATA_YEAR` / `_validate_supported_year`),
+#   - or ACF renames an existing FY's workbook on disk. A 404 from
+#     `_acf_get` is the early signal — that's an authoritative file
+#     rename, not a transient outage, and the new path needs to be
+#     copied in by hand from the corresponding ACF page.
+TANF_WORKBOOK_URLS: dict[int, dict[str, str]] = {
+    2024: {
+        "caseload": (
+            "https://acf.gov/sites/default/files/documents/ofa/"
+            "fy2024_tanf_caseload.xlsx"
+        ),
+        "financial": (
+            "https://acf.gov/sites/default/files/documents/ofa/"
+            "fy-2024-tanf-moe-financial-data.xlsx"
+        ),
+    },
+}
+
+
+@retry(
+    stop=stop_after_attempt(5),
+    wait=wait_exponential(multiplier=2, min=5, max=60),
+    retry=retry_if_exception_type(
+        (
+            requests.exceptions.Timeout,
+            requests.exceptions.ConnectionError,
+            requests.exceptions.ChunkedEncodingError,
+        )
+    ),
+    before_sleep=before_sleep_log(logger, logging.WARNING),
+    reraise=True,
 )
+def _acf_get(session: requests.Session, url: str) -> requests.Response:
+    response = session.get(url, timeout=ACF_REQUEST_TIMEOUT)
+    response.raise_for_status()
+    return response
 
 
 def _validate_supported_year(year: int) -> None:
@@ -37,9 +83,7 @@ def _validate_supported_year(year: int) -> None:
         )
 
 
-def _download_acf_excel(
-    page_url: str, cache_file: str, url_pattern: re.Pattern
-) -> bytes:
+def _download_acf_excel(workbook_url: str, cache_file: str) -> bytes:
     if is_cached(cache_file):
         logger.info("Using cached %s", cache_file)
         return load_bytes(cache_file)
@@ -54,25 +98,16 @@ def _download_acf_excel(
         }
     )
 
-    page_response = session.get(page_url, timeout=30)
-    page_response.raise_for_status()
-    match = url_pattern.search(page_response.text)
-    if match is None:
-        raise ValueError(f"Could not find TANF workbook URL on {page_url}")
-
-    workbook_url = match.group(0)
-    workbook_response = session.get(workbook_url, timeout=60)
-    workbook_response.raise_for_status()
+    workbook_response = _acf_get(session, workbook_url)
     save_bytes(cache_file, workbook_response.content)
     return workbook_response.content
 
 
 def extract_tanf_caseload_data(year: int) -> pd.DataFrame:
     _validate_supported_year(year)
     workbook = _download_acf_excel(
-        CASELOAD_PAGE_URL,
+        TANF_WORKBOOK_URLS[ACF_DATA_YEAR]["caseload"],
         f"tanf_caseload_{ACF_DATA_YEAR}.xlsx",
-        CASELOAD_URL_PATTERN,
     )
     return pd.read_excel(io.BytesIO(workbook), sheet_name="TFam", header=3)
 
@@ -115,9 +150,8 @@ def extract_tanf_financial_data(
 ) -> tuple[pd.DataFrame, dict[str, pd.DataFrame]]:
     _validate_supported_year(year)
     workbook = _download_acf_excel(
-        FINANCIAL_PAGE_URL,
+        TANF_WORKBOOK_URLS[ACF_DATA_YEAR]["financial"],
         f"tanf_financial_{ACF_DATA_YEAR}.xlsx",
-        FINANCIAL_URL_PATTERN,
     )
     xls = pd.ExcelFile(io.BytesIO(workbook))
     national_df = pd.read_excel(

pyproject.toml

@@ -23,7 +23,9 @@ classifiers = [
 ]
 dependencies = [
     "policyengine-us>=1.674.1",
-    "policyengine-core>=3.23.6",
+    # policyengine-core 3.25.4 fixes PolicyEngine/policyengine-core#482
+    # (user-set ETERNITY inputs lost after _invalidate_all_caches).
+    "policyengine-core>=3.25.4,<3.26",
     "pandas>=2.3.1",
     "requests>=2.25.0",
     "tqdm>=4.60.0",

tests/unit/datasets/test_cps_file_handles.py

@@ -1,5 +1,6 @@
 from types import SimpleNamespace
 
+import h5py
 import numpy as np
 import pandas as pd
 
@@ -249,8 +250,25 @@ def recording_hdfstore(path, mode="a", *args, **kwargs):
         opened_modes.append(mode)
         return real_hdfstore(path, mode=mode, *args, **kwargs)
 
-    def fail_h5py_file(*args, **kwargs):
-        raise AssertionError("add_rent should not reopen the existing H5 with h5py")
+    # ACS fixture H5 backing the policyengine-core#482 workaround in add_rent
+    # (it reads is_household_head straight from the source H5 since
+    # calculate_dataframe drops user-set ETERNITY inputs under
+    # policyengine-core 3.24.0+). 10_000 True values match the size of the
+    # FakeMicrosimulation train frame below.
+    acs_fixture_path = tmp_path / "acs_fixture.h5"
+    with h5py.File(acs_fixture_path, "w") as acs_fixture:
+        acs_fixture["is_household_head"] = np.ones(10_000, dtype=bool)
+
+    real_h5py_file = cps_module.h5py.File
+    opened_h5_paths = []
+
+    def recording_h5py_file(path, mode="r", *args, **kwargs):
+        opened_h5_paths.append((str(path), mode))
+        if str(path) == str(existing_path):
+            raise AssertionError(
+                "add_rent should not reopen the existing CPS H5 with h5py"
+            )
+        return real_h5py_file(path, mode=mode, *args, **kwargs)
 
     class FakeQRF:
         def fit(self, X_train, predictors, imputed_variables):
@@ -314,26 +332,33 @@ def save_dataset(self, data):
             self.saved.append(data)
 
     monkeypatch.setattr(cps_module.pd, "HDFStore", recording_hdfstore)
-    monkeypatch.setattr(cps_module.h5py, "File", fail_h5py_file)
+    monkeypatch.setattr(cps_module.h5py, "File", recording_h5py_file)
     monkeypatch.setattr(cps_module, "QRF", FakeQRF)
 
     import policyengine_us
     import policyengine_us_data.datasets.acs.acs as acs_module
 
+    class FakeACS_2022:
+        file_path = acs_fixture_path
+
     monkeypatch.setattr(policyengine_us, "Microsimulation", FakeMicrosimulation)
-    monkeypatch.setattr(acs_module, "ACS_2022", object())
+    monkeypatch.setattr(acs_module, "ACS_2022", FakeACS_2022)
 
     dataset = FakeDataset()
     cps = {
         "age": np.array([40], dtype=np.int32),
         "spm_unit_capped_housing_subsidy_reported": np.array([0.0]),
+        # add_id_variables populates this upstream of add_rent in the real
+        # pipeline; see the policyengine-core#482 workaround override below.
+        "is_household_head": np.array([True]),
     }
     person = pd.DataFrame({"dummy": [1]})
     household = pd.DataFrame({"H_TENURE": [2]})
 
     add_rent(dataset, cps, person, household)
 
     assert opened_modes == ["r"]
+    assert opened_h5_paths == [(str(acs_fixture_path), "r")]
     assert not existing_path.exists()
     np.testing.assert_array_equal(cps["rent"], np.array([1_000.0]))
     np.testing.assert_array_equal(cps["real_estate_taxes"], np.array([250.0]))