feat(nowcast): detail IO loader router for Step 8 (Phase A)

bedrock/extract/iot/__tests__/test_detail_io.py

@@ -0,0 +1,48 @@
+# ruff: noqa: PLC0415
+
+from collections.abc import Callable
+
+import pandas as pd
+import pytest
+
+from bedrock.extract.iot.detail_io import (
+    load_detail_margins_usa,
+    load_detail_Uimp_usa,
+    load_detail_Utot_usa,
+    load_detail_V_usa,
+    load_detail_value_added_usa,
+    load_detail_Ytot_usa,
+)
+from bedrock.extract.iot.io_2017 import (
+    load_2017_margins_usa,
+    load_2017_Uimp_usa,
+    load_2017_Utot_usa,
+    load_2017_V_usa,
+    load_2017_value_added_usa,
+    load_2017_Ytot_usa,
+)
+from bedrock.utils.config.usa_config import reset_usa_config, set_global_usa_config
+
+
+@pytest.fixture(autouse=True)
+def _reset_config() -> None:
+    reset_usa_config(should_reset_env_var=True)
+
+
+@pytest.mark.parametrize(
+    ('detail_loader', 'published_loader'),
+    [
+        (load_detail_V_usa, load_2017_V_usa),
+        (load_detail_Utot_usa, load_2017_Utot_usa),
+        (load_detail_Uimp_usa, load_2017_Uimp_usa),
+        (load_detail_margins_usa, load_2017_margins_usa),
+        (load_detail_Ytot_usa, load_2017_Ytot_usa),
+        (load_detail_value_added_usa, load_2017_value_added_usa),
+    ],
+)
+def test_bea_published_detail_loaders_match_2017(
+    detail_loader: Callable[[], pd.DataFrame],
+    published_loader: Callable[[], pd.DataFrame],
+) -> None:
+    set_global_usa_config('test_usa_config.yaml')
+    pd.testing.assert_frame_equal(detail_loader(), published_loader())

bedrock/extract/iot/detail_io.py

@@ -0,0 +1,68 @@
+"""Router for BEA 2017 Detail IO tables used by the Cornerstone pipeline.
+
+``usa_detail_io_source`` selects published BEA tables (``bea_published``) or
+nowcast replay artifacts (``nowcast``). Cornerstone correspondence and
+expansion run unchanged downstream; only the loader inputs differ.
+"""
+
+from __future__ import annotations
+
+import pandas as pd
+
+from bedrock.extract.iot.io_2017 import (
+    load_2017_margins_usa,
+    load_2017_Uimp_usa,
+    load_2017_Utot_usa,
+    load_2017_V_usa,
+    load_2017_value_added_usa,
+    load_2017_Ytot_usa,
+)
+from bedrock.extract.iot.nowcast_mut_storage import (
+    load_nowcast_detail_margins_usa,
+    load_nowcast_detail_Uimp_usa,
+    load_nowcast_detail_Utot_usa,
+    load_nowcast_detail_V_usa,
+    load_nowcast_detail_value_added_usa,
+    load_nowcast_detail_Ytot_usa,
+)
+from bedrock.utils.config.usa_config import get_usa_config
+
+
+def _detail_io_source() -> str:
+    return get_usa_config().usa_detail_io_source
+
+
+def load_detail_V_usa() -> pd.DataFrame:
+    if _detail_io_source() == 'bea_published':
+        return load_2017_V_usa()
+    return load_nowcast_detail_V_usa()
+
+
+def load_detail_Utot_usa() -> pd.DataFrame:
+    if _detail_io_source() == 'bea_published':
+        return load_2017_Utot_usa()
+    return load_nowcast_detail_Utot_usa()
+
+
+def load_detail_Uimp_usa() -> pd.DataFrame:
+    if _detail_io_source() == 'bea_published':
+        return load_2017_Uimp_usa()
+    return load_nowcast_detail_Uimp_usa()
+
+
+def load_detail_margins_usa() -> pd.DataFrame:
+    if _detail_io_source() == 'bea_published':
+        return load_2017_margins_usa()
+    return load_nowcast_detail_margins_usa()
+
+
+def load_detail_Ytot_usa() -> pd.DataFrame:
+    if _detail_io_source() == 'bea_published':
+        return load_2017_Ytot_usa()
+    return load_nowcast_detail_Ytot_usa()
+
+
+def load_detail_value_added_usa() -> pd.DataFrame:
+    if _detail_io_source() == 'bea_published':
+        return load_2017_value_added_usa()
+    return load_nowcast_detail_value_added_usa()

bedrock/extract/iot/nowcast_mut_storage.py

@@ -0,0 +1,78 @@
+"""Nowcast BEA-detail MUT artifact storage (GCS layout TBD).
+
+Published artifacts are the Make, Use, Import, and Margins tables in BEA 2017
+Detail space before and after redefinition. The detail IO router loads those
+artifacts and applies existing Cornerstone correspondence at read time.
+
+Physical GCS URIs are not finalized; ``resolve_nowcast_mut_uri`` and the
+``load_nowcast_detail_*`` entry points raise ``NotImplementedError`` until
+wired.
+"""
+
+from __future__ import annotations
+
+import typing as ta
+
+import pandas as pd
+
+from bedrock.utils.config.usa_config import get_usa_config
+
+
+def resolve_nowcast_mut_uri(
+    *,
+    vintage: str,
+    year: int,
+    stage: ta.Literal['before', 'after'],
+    table: ta.Literal['Make', 'Use', 'Import', 'Margins', 'Ytot', 'ValueAdded'],
+) -> str:
+    """Return the GCS URI for a nowcast MUT artifact.
+
+    Logical layout: ``{vintage}/{year}/{stage}/{table}`` — physical bucket and
+    prefix are not finalized.
+    """
+    raise NotImplementedError(
+        'GCS layout for nowcast MUT artifacts is not finalized; '
+        f'vintage={vintage!r}, year={year}, stage={stage!r}, table={table!r}'
+    )
+
+
+def _resolve_configured_nowcast_uri(
+    table: ta.Literal['Make', 'Use', 'Import', 'Margins', 'Ytot', 'ValueAdded'],
+) -> str:
+    cfg = get_usa_config()
+    return resolve_nowcast_mut_uri(
+        vintage=cfg.nowcast_mut_vintage or '',
+        year=cfg.usa_base_io_data_year,
+        stage=cfg.iot_before_or_after_redefinition,
+        table=table,
+    )
+
+
+def load_nowcast_detail_V_usa() -> pd.DataFrame:
+    _resolve_configured_nowcast_uri('Make')
+    raise AssertionError('resolve_nowcast_mut_uri must raise NotImplementedError')
+
+
+def load_nowcast_detail_Utot_usa() -> pd.DataFrame:
+    _resolve_configured_nowcast_uri('Use')
+    raise AssertionError('resolve_nowcast_mut_uri must raise NotImplementedError')
+
+
+def load_nowcast_detail_Uimp_usa() -> pd.DataFrame:
+    _resolve_configured_nowcast_uri('Import')
+    raise AssertionError('resolve_nowcast_mut_uri must raise NotImplementedError')
+
+
+def load_nowcast_detail_margins_usa() -> pd.DataFrame:
+    _resolve_configured_nowcast_uri('Margins')
+    raise AssertionError('resolve_nowcast_mut_uri must raise NotImplementedError')
+
+
+def load_nowcast_detail_Ytot_usa() -> pd.DataFrame:
+    _resolve_configured_nowcast_uri('Ytot')
+    raise AssertionError('resolve_nowcast_mut_uri must raise NotImplementedError')
+
+
+def load_nowcast_detail_value_added_usa() -> pd.DataFrame:
+    _resolve_configured_nowcast_uri('ValueAdded')
+    raise AssertionError('resolve_nowcast_mut_uri must raise NotImplementedError')

bedrock/transform/eeio/cornerstone_bea_intermediates.py

@@ -11,10 +11,10 @@
 
 import pandas as pd
 
-from bedrock.extract.iot.io_2017 import (
-    load_2017_Uimp_usa,
-    load_2017_Utot_usa,
-    load_2017_V_usa,
+from bedrock.extract.iot.detail_io import (
+    load_detail_Uimp_usa,
+    load_detail_Utot_usa,
+    load_detail_V_usa,
 )
 from bedrock.utils.math.formulas import (
     compute_A_matrix,
@@ -33,13 +33,13 @@
 @functools.cache
 def bea_x() -> pd.Series[float]:
     """Industry total output in BEA 2017 space."""
-    return compute_x(V=load_2017_V_usa())
+    return compute_x(V=load_detail_V_usa())
 
 
 @functools.cache
 def bea_q() -> pd.Series[float]:
     """Commodity total output in BEA 2017 space."""
-    return compute_q(V=load_2017_V_usa())
+    return compute_q(V=load_detail_V_usa())
 
 
 # ---------------------------------------------------------------------------
@@ -55,7 +55,7 @@ def bea_Vnorm_scrap_corrected() -> pd.DataFrame:
     divided by (1 − scrap_j / q_j), where scrap_j is the scrap output of the
     industry sharing code j.
     """
-    V = load_2017_V_usa()
+    V = load_detail_V_usa()
     q = bea_q()
     Vnorm = compute_Vnorm_matrix(V=V, q=q)
     scrap = V.loc[:, 'S00401']
@@ -73,8 +73,8 @@ def bea_Aq() -> tuple[pd.DataFrame, pd.DataFrame, pd.Series[float]]:
     x = bea_x()
     Vnorm = bea_Vnorm_scrap_corrected()
 
-    Utot = load_2017_Utot_usa()
-    Uimp = load_2017_Uimp_usa()
+    Utot = load_detail_Utot_usa()
+    Uimp = load_detail_Uimp_usa()
     Udom = handle_negative_matrix_values(Utot - Uimp)
     Uimp_clean = handle_negative_matrix_values(Uimp)
 

bedrock/transform/eeio/cornerstone_disagg_pipeline.py

@@ -27,12 +27,12 @@
     EEIOWasteDisaggConfig,
     effective_waste_disagg_config,
 )
-from bedrock.extract.iot.io_2017 import (
-    load_2017_Uimp_usa,
-    load_2017_Utot_usa,
-    load_2017_V_usa,
-    load_2017_value_added_usa,
-    load_2017_Ytot_usa,
+from bedrock.extract.iot.detail_io import (
+    load_detail_Uimp_usa,
+    load_detail_Utot_usa,
+    load_detail_V_usa,
+    load_detail_value_added_usa,
+    load_detail_Ytot_usa,
 )
 from bedrock.transform.eeio.cornerstone_expansion import (
     commodity_corresp,
@@ -125,7 +125,7 @@ class CornerstoneDisaggIOBundle:
 
 
 def derive_cornerstone_V_after_waste() -> pd.DataFrame:
-    V_2017 = load_2017_V_usa()
+    V_2017 = load_detail_V_usa()
     V = industry_corresp() @ V_2017 @ commodity_corresp().T
     V.index.name = 'sector'
     V.columns.name = 'sector'
@@ -138,8 +138,8 @@ def derive_cornerstone_V_after_waste() -> pd.DataFrame:
 
 
 def derive_cornerstone_U_after_waste() -> tuple[pd.DataFrame, pd.DataFrame]:
-    Utot = load_2017_Utot_usa()
-    Uimp = load_2017_Uimp_usa()
+    Utot = load_detail_Utot_usa()
+    Uimp = load_detail_Uimp_usa()
     Udom = Utot - Uimp
 
     com_c = commodity_corresp()
@@ -164,7 +164,7 @@ def derive_cornerstone_U_after_waste() -> tuple[pd.DataFrame, pd.DataFrame]:
 
 def _derive_y_before_electricity_disagg() -> pd.DataFrame:
     """Correspondence-mapped Y after waste disagg, before electricity row split."""
-    ytot_orig = load_2017_Ytot_usa()
+    ytot_orig = load_detail_Ytot_usa()
     ytot = commodity_corresp() @ ytot_orig
     ytot.index.name = 'sector'
     weights = get_waste_disagg_weights()
@@ -175,7 +175,7 @@ def _derive_y_before_electricity_disagg() -> pd.DataFrame:
 
 
 def derive_cornerstone_VA_after_waste() -> pd.DataFrame:
-    VA = load_2017_value_added_usa() @ industry_corresp().T
+    VA = load_detail_value_added_usa() @ industry_corresp().T
     VA.columns.name = 'sector'
     weights = get_waste_disagg_weights()
     if weights is not None:
@@ -208,7 +208,7 @@ def derive_disagg_io_bundle() -> CornerstoneDisaggIOBundle:
 @functools.cache
 def derive_disagg_Ytot_with_trade() -> pd.DataFrame:
     """Correspondence-mapped Y with optional waste and electricity disagg."""
-    Ytot_orig = load_2017_Ytot_usa()
+    Ytot_orig = load_detail_Ytot_usa()
     Ytot = commodity_corresp() @ Ytot_orig
     Ytot.index.name = 'sector'
     weights = get_waste_disagg_weights()