Address hardcoded 2018 MECS in GHG methods

bedrock/extract/eia/EIA_MECS.py

@@ -20,8 +20,16 @@
 from bedrock.extract.generateflowbyactivity import generateFlowByActivity
 from bedrock.transform.flowbyclean import load_prepare_clean_source
 from bedrock.transform.flowbyfunctions import assign_fips_location_system
-from bedrock.utils.config.common import WITHDRAWN_KEYWORD
-from bedrock.utils.io.gcp import download_extract_input_from_gcs_if_not_exists
+from bedrock.utils.config.common import WITHDRAWN_KEYWORD, get_catalog_info
+from bedrock.utils.economic.units import (
+    HEATING_OIL_MMBTU_PER_GALLON,
+    PROPANE_MMBTU_PER_GALLON,
+)
+from bedrock.utils.io.gcp import (
+    download_extract_input_from_gcs_if_not_exists,
+    load_from_gcs,
+)
+from bedrock.utils.io.gcp_paths import gcs_extract_input_path
 from bedrock.utils.io.local_extract_input_data import local_extract_input_dir
 from bedrock.utils.logging.flowsa_log import log
 from bedrock.utils.mapping.location import (
@@ -656,6 +664,189 @@ def estimate_suppressed_mecs_energy(
     return unsuppressed.drop(columns='Suppressed')
 
 
+def keep_chemical_manufacturing(fba: FlowByActivity, **_kwargs: Any) -> FlowByActivity:
+    '''Keep MECS NAICS 325 (chemicals) and descendants. CEDA NEU NG is chemicals-only.'''
+    return fba.query("ActivityConsumedBy.str.startswith('325')").reset_index(drop=True)
+
+
+# 2013 MECS Other energy / (energy + nonfuel) when the mapped NAICS has no
+# Table 2.1 / 3.1 Other ratio. Keys are MECS NAICS prefixes.
+PETROL_FUEL_RATIO_FALLBACKS: dict[str, float] = {
+    '311221': 0.93,
+    '324110': 0.43,
+    '324199': 0.45,
+    '324121': 0.45,
+    '324122': 0.45,
+    '325110': 0.95,
+    '325180': 0.91,
+    '325194': 0.76,
+    '325211': 0.99,
+    '325212': 0.84,
+    '325311': 0.50,
+    '327211': 0.00,
+    '327310': 0.98,
+    '331110': 0.96,
+    '3312': 0.70,
+    '336': 0.83,
+}
+
+
+def multiply_bea_by_mecs_petroleum_energy_fraction(
+    fba: FlowByActivity, **_kwargs: Any
+) -> FlowByActivity:
+    '''
+    clean_fba_after_attribution. Industrial petrol weights are BEA 324110
+    purchases times MECS Other energy fraction Table 3.1 / (Table 2.1 + Table 3.1).
+    Join on mapped NAICS (SectorConsumedBy after Cornerstone_2025), prefix-matching
+    published MECS lines. Sectors with no MECS ratio use 2013 fallback values,
+    else 1.0. Restricts to ag/mining/construction/manufacturing plus 221200.
+    '''
+    clean_source = fba.config['clean_source']
+    if isinstance(clean_source, str):
+        name, src_config = clean_source, {}
+    else:
+        ((name, src_config),) = clean_source.items()
+    year = int(src_config['year'])
+    mecs = FlowByActivity(
+        getFlowByActivity(name, year),
+        full_name=name,
+        config={**get_catalog_info(name), **src_config, 'year': year},
+    )
+    mecs = mecs.function_socket('estimate_suppressed').select_by_fields()
+    t21 = (
+        mecs.query("Description == 'Table 2.1'")
+        .groupby('ActivityConsumedBy')['FlowAmount']
+        .sum()
+    )
+    t31 = (
+        mecs.query("Description == 'Table 3.1'")
+        .groupby('ActivityConsumedBy')['FlowAmount']
+        .sum()
+    )
+    idx = t21.index.union(t31.index)
+    t21 = t21.reindex(idx).fillna(0)
+    t31 = t31.reindex(idx).fillna(0)
+    den = t21 + t31
+    ratios = (t31 / den).where(den != 0).fillna(1.0)
+
+    sector_col = (
+        'SectorConsumedBy'
+        if 'SectorConsumedBy' in fba.columns
+        else 'ActivityConsumedBy'
+    )
+    sector = fba[sector_col].astype(str)
+    industrial = sector.str.startswith(
+        ('11', '21', '23', '31', '32', '33')
+    ) | sector.eq('221200')
+    out = fba.loc[industrial].copy()
+    ratio_index = set(ratios.index.astype(str))
+
+    def ratio_for_sector(naics: str) -> float:
+        for n in range(len(naics), 0, -1):
+            key = naics[:n]
+            if key in ratio_index:
+                return float(ratios[key])
+            if key in PETROL_FUEL_RATIO_FALLBACKS:
+                return PETROL_FUEL_RATIO_FALLBACKS[key]
+        return 1.0
+
+    matched = out[sector_col].astype(str).map(ratio_for_sector)
+    return out.assign(FlowAmount=out['FlowAmount'] * matched)
+
+
+def household_petroleum_transport_fraction(config: dict[str, Any]) -> float:
+    '''
+    Share of PCE gasoline-and-other-energy that is transport rather than
+    residential heat oil/propane. Same formula CEDA uses on F01000.
+    '''
+    clean_parameter = config['clean_parameter']
+    year = int(clean_parameter.get('year', config['year']))
+    pce_cfg = clean_parameter['pce']
+    propane_cfg = clean_parameter['propane_price']
+    heat_oil_cfg = clean_parameter['heating_oil_price']
+    heat_fuels = clean_parameter['residential_heat_fuels']
+
+    pce_tbl = load_from_gcs(
+        name=pce_cfg['file'],
+        sub_bucket=gcs_extract_input_path(pce_cfg['extract_input']),
+        local_dir=local_extract_input_dir(pce_cfg['extract_input']),
+        loader=lambda pth: pd.read_csv(
+            pth,
+            skiprows=int(pce_cfg.get('skiprows', 3)),
+            index_col=int(pce_cfg.get('index_col', 1)),
+        )
+        .dropna()
+        .drop(columns=list(pce_cfg.get('drop_columns', ['Line']))),
+    )
+    pce_tbl.index = pce_tbl.index.str.strip()
+    pce_tbl.columns = pce_tbl.columns.astype(int)
+    pce_year = int(max(int(c) for c in pce_tbl.columns if int(c) <= year))
+    pce = float(pce_tbl.loc[pce_cfg['line'], pce_year])
+
+    prices: dict[str, float] = {}
+    for key, price_cfg in (
+        ('propane_price', propane_cfg),
+        ('heating_oil_price', heat_oil_cfg),
+    ):
+        skiprows = int(price_cfg.get('skiprows', 4))
+        price_tbl = load_from_gcs(
+            name=price_cfg['file'],
+            sub_bucket=gcs_extract_input_path(price_cfg['extract_input']),
+            local_dir=local_extract_input_dir(price_cfg['extract_input']),
+            loader=lambda pth: pd.read_csv(pth, skiprows=skiprows),
+        )
+        month_col = price_cfg.get('month_column', 'Month')
+        value_col = price_cfg['value_column']
+        price_tbl['Year'] = price_tbl[month_col].str.extract(r'(\d{4})').astype(int)
+        by_year = price_tbl.groupby('Year')[value_col].mean()
+        years = [y for y in by_year.index if y <= year] or list(by_year.index)
+        prices[key] = float(by_year[int(max(years))])
+
+    clean_source = config['clean_source']
+    if isinstance(clean_source, str):
+        name, src_config = clean_source, {}
+    else:
+        ((name, src_config),) = clean_source.items()
+    annex_year = int(src_config.get('year', year))
+    annex = FlowByActivity(
+        getFlowByActivity(name, annex_year),
+        full_name=name,
+        config={**get_catalog_info(name), **src_config, 'year': annex_year},
+    ).select_by_fields()
+
+    kerosene_lpg = float(
+        annex.loc[
+            annex['FlowName'].isin(list(heat_fuels['propane_priced'])), 'FlowAmount'
+        ].sum()
+    )
+    distillate = float(
+        annex.loc[
+            annex['FlowName'].isin(list(heat_fuels['heating_oil_priced'])),
+            'FlowAmount',
+        ].sum()
+    )
+    res_heat = (kerosene_lpg * (prices['propane_price'] / PROPANE_MMBTU_PER_GALLON)) + (
+        distillate * (prices['heating_oil_price'] / HEATING_OIL_MMBTU_PER_GALLON)
+    )
+    return (pce - res_heat) / pce
+
+
+def scale_household_petroleum_to_transport_share(
+    fba: FlowByActivity, **_kwargs: Any
+) -> FlowByActivity:
+    '''
+    clean_fba_after_attribution. Scale F01000 petroleum by the transport share
+    of PCE gasoline-and-other-energy (residential heat oil/propane removed).
+    '''
+    scale = household_petroleum_transport_fraction(fba.config)
+    mask = fba['ActivityConsumedBy'] == 'F01000'
+    if 'SectorConsumedBy' in fba.columns:
+        mask = mask | (fba['SectorConsumedBy'] == 'F01000')
+    out = fba.copy()
+    out.loc[mask, 'FlowAmount'] = out.loc[mask, 'FlowAmount'] * scale
+    return out
+
+
 def clean_mapped_mecs_energy_fba_to_state(
     fba: FlowByActivity, **_: Any
 ) -> FlowByActivity:
@@ -738,41 +929,6 @@ def mecs_land_fba_cleanup(fba: FlowByActivity, **_: Any) -> FlowByActivity:
     return fba
 
 
-def clean_mecs_energy_fba_for_bea_summary(
-    fba: FlowByActivity, **_kwargs: Any
-) -> FlowByActivity:
-    naics_3 = fba.query('ActivityConsumedBy.str.len() == 3')
-    naics_4 = fba.query(
-        'ActivityConsumedBy.str.len() == 4 '
-        '& ActivityConsumedBy.str.startswith("336")'
-    )
-    naics_4_sum = (
-        naics_4.assign(ActivityConsumedBy='336')
-        .aggregate_flowby()[['Flowable', 'FlowAmount', 'Unit', 'ActivityConsumedBy']]
-        .rename(columns={'FlowAmount': 'naics_4_sum'})
-    )
-
-    merged = naics_3.merge(naics_4_sum, how='left').fillna({'naics_4_sum': 0})
-    subtracted = merged.assign(FlowAmount=merged.FlowAmount - merged.naics_4_sum).drop(
-        columns='naics_4_sum'
-    )
-
-    subtracted.config['naics_4_list'] = list(naics_4.ActivityConsumedBy.unique())
-
-    return subtracted
-
-
-def clean_mapped_mecs_energy_fba_for_bea_summary(
-    fba: FlowByActivity, **_kwargs: Any
-) -> FlowByActivity:
-    _naics_4_list = fba.config['naics_4_list']
-
-    return fba.query(
-        '~(SectorConsumedBy in @_naics_4_list '
-        '& ActivityConsumedBy != SectorConsumedBy)'
-    )
-
-
 if __name__ == "__main__":
     generateFlowByActivity(source='EIA_MECS_Energy', year=2018)
     fba = getFlowByActivity('EIA_MECS_Energy', 2018)

bedrock/extract/epa/EPA_GHGI.py

@@ -168,7 +168,11 @@ def ghg_load_gcs(**kwargs: dict[str, Any]) -> List[pd.DataFrame]:
             if year == '2023' and table == '3-25b':
                 # Skip 3-25b for current year (use 3-25 instead)
                 continue
-            df = _load_ghg_table(table)
+            try:
+                df = _load_ghg_table(table)
+            except Exception as exc:
+                log.warning(f'Skipping EPA GHGI table {table} for year {year}: {exc}')
+                continue
             if df is not None and len(df.columns) > 1:
                 years = YEARS.copy()
                 years.remove(year)

bedrock/extract/flowbyactivity.py

@@ -715,37 +715,31 @@ def prepare_fbs(
             drop_cols = []
 
         if 'activity_sets' in self.config:
-            try:
-                return FlowBySector(
-                    pd.concat(
-                        [
-                            fba.prepare_fbs(
-                                external_config_path=external_config_path,
-                                download_sources_ok=download_sources_ok,
-                                skip_select_by=True,
-                                retain_activity_columns=retain_activity_columns,
-                                fbs_method_name=fbs_method_name,
-                            )
-                            for fba in (
-                                self.select_by_fields()
-                                .function_socket('clean_fba_before_activity_sets')
-                                .activity_sets()
-                            )
-                        ]
-                    ).reset_index(drop=True),
-                    convert_df_to_flowby=True,
-                )
-            except ValueError:
-                # This discards every activity_set, not just the one that
-                # failed, so a fault in one silently zeroes the whole method.
-                # Log it - the silence is what makes this class of bug expensive
-                # to find.
-                log.exception(
-                    'Discarding ALL activity_sets for %s: one of them raised '
-                    'while being prepared. The method will return no rows.',
-                    self.full_name,
-                )
-                return FlowBySector(pd.DataFrame(), convert_df_to_flowby=True)
+            prepared = []
+            for fba in (
+                self.select_by_fields()
+                .function_socket('clean_fba_before_activity_sets')
+                .activity_sets()
+            ):
+                try:
+                    prepared.append(
+                        fba.prepare_fbs(
+                            external_config_path=external_config_path,
+                            download_sources_ok=download_sources_ok,
+                            skip_select_by=True,
+                            retain_activity_columns=retain_activity_columns,
+                            fbs_method_name=fbs_method_name,
+                        )
+                    )
+                except ValueError as exc:
+                    log.exception(f'{fba.full_name} failed while preparing FBS: {exc}')
+                    raise ValueError(
+                        f'{fba.full_name} failed while preparing FBS: {exc}'
+                    ) from exc
+            return FlowBySector(
+                pd.concat(prepared).reset_index(drop=True),
+                convert_df_to_flowby=True,
+            )
         log.info(f'Processing FlowBySector for {self.full_name}')
         # Primary FlowBySector generation approach:
         return FlowBySector(