Improve data retrieval and workflow robustness

workflow/config/config.common.yaml

@@ -1,3 +1,7 @@
+pudl_cache:
+  version: v2025.5.0
+  base_url: https://s3.us-west-2.amazonaws.com/pudl.catalyst.coop
+  directory: repo_data/pudl
 pudl_path: s3://pudl.catalyst.coop/v2025.2.0
 
 foresight: 'perfect'

workflow/repo_data/config/config.common.yaml

@@ -1,5 +1,10 @@
 pudl_path: s3://pudl.catalyst.coop/v2025.5.0
 
+pudl_cache:
+  version: v2025.5.0
+  base_url: https://s3.us-west-2.amazonaws.com/pudl.catalyst.coop
+  directory: repo_data/pudl
+
 # docs :
 renewable:
   EGS:

workflow/rules/build_electricity.smk

@@ -1,6 +1,107 @@
 ################# ----------- Rules to Build Network ---------- #################
 
 from itertools import chain
+from pathlib import Path
+
+
+PUDL_VERSION = config["pudl_cache"]["version"]
+PUDL_BASE_URL = config["pudl_cache"]["base_url"].rstrip("/")
+PUDL_DIRECTORY = Path(config["pudl_cache"]["directory"]) / PUDL_VERSION
+
+PUDL_POWERPLANT_FILES = [
+    "out_eia__monthly_generators.parquet",
+    "out_eia__yearly_generators.parquet",
+    "core_eia860__scd_generators_energy_storage.parquet",
+    "core_eia860__scd_plants.parquet",
+]
+
+PUDL_COST_FILES = [
+    "core_nrelatb__yearly_projected_financial_cases_by_scenario.parquet",
+    "core_nrelatb__yearly_projected_cost_performance.parquet",
+    "core_eiaaeo__yearly_projected_fuel_cost_in_electric_sector_by_type.parquet",
+]
+
+PUDL_DEMAND_FILES = [
+    "censusdp1tract.state_2010census_dp1.parquet",
+    "core_eiaaeo__yearly_projected_generation_in_electric_sector_by_technology.parquet",
+]
+
+PUDL_FUEL_PRICE_FILES = [
+    "out_eia__monthly_generators.parquet",
+    "out_eia__yearly_generators.parquet",
+    "core_eia860__scd_plants.parquet",
+]
+
+PUDL_FILES = sorted(
+    set(
+        PUDL_POWERPLANT_FILES
+        + PUDL_COST_FILES
+        + PUDL_DEMAND_FILES
+        + PUDL_FUEL_PRICE_FILES
+    )
+)
+
+
+def pudl_paths(filenames):
+    return [str(PUDL_DIRECTORY / filename) for filename in filenames]
+
+
+PUDL_POWERPLANT_PATHS = pudl_paths(PUDL_POWERPLANT_FILES)
+PUDL_COST_PATHS = pudl_paths(PUDL_COST_FILES)
+PUDL_FUEL_PRICE_PATHS = pudl_paths(PUDL_FUEL_PRICE_FILES)
+
+
+def demand_pudl_paths(wildcards):
+    profile = config["electricity"]["demand"]["profile"]
+
+    if profile == "ferc":
+        return pudl_paths(PUDL_DEMAND_FILES)
+
+    if profile == "eia":
+        return pudl_paths(
+            [
+                "core_eiaaeo__yearly_projected_generation_in_electric_sector_by_technology.parquet",
+            ]
+        )
+
+    return []
+
+
+rule retrieve_pudl_file:
+    output:
+        str(PUDL_DIRECTORY / "{pudl_file}.parquet"),
+    params:
+        url=lambda wildcards: (
+            f"{PUDL_BASE_URL}/{PUDL_VERSION}/" f"{wildcards.pudl_file}.parquet"
+        ),
+    wildcard_constraints:
+        pudl_file="|".join(Path(filename).stem for filename in PUDL_FILES),
+    log:
+        "logs/retrieve_pudl/{pudl_file}.log",
+    resources:
+        mem_mb=1000,
+        walltime="00:30:00",
+    shell:
+        r"""
+        mkdir -p "$(dirname {output:q})" "$(dirname {log:q})"
+
+        curl -fL \
+          --retry 5 \
+          --retry-all-errors \
+          --connect-timeout 20 \
+          --max-time 1800 \
+          {params.url:q} \
+          --output {output:q}.part \
+          > {log:q} 2>&1
+
+        test -s {output:q}.part
+
+        python -c \
+          "import duckdb; duckdb.sql(\"SELECT * FROM read_parquet('{output}.part') LIMIT 1\").fetchall()" \
+          >> {log:q} 2>&1
+
+        mv {output:q}.part {output:q}
+        """
 
 
 rule build_shapes:
@@ -101,8 +202,9 @@ rule build_bus_regions:
 rule build_cost_data:
     params:
         aeo=config_provider("costs", "aeo"),
-        pudl_path=config_provider("pudl_path"),
+        pudl_path=str(PUDL_DIRECTORY),
     input:
+        pudl=PUDL_COST_PATHS,
         efs_tech_costs="repo_data/costs/EFS_Technology_Data.xlsx",
         efs_icev_costs="repo_data/costs/efs_icev_costs.csv",
         eia_tech_costs="repo_data/costs/eia_tech_costs.csv",
@@ -344,7 +446,6 @@ def demand_raw_data(wildcards):
     elif profile == "ferc":
         return [
             DATA + "pudl/out_ferc714__hourly_estimated_state_demand.parquet",
-            DATA + "pudl/censusdp1tract.sqlite",
         ]
     elif profile == "eulp":
         return [
@@ -422,8 +523,9 @@ rule build_electrical_demand:
         planning_horizons=config["scenario"]["planning_horizons"],
         renewable_weather_years=config["renewable_weather_years"],
         snapshots=config["snapshots"],
-        pudl_path=config_provider("pudl_path"),
+        pudl_path=str(PUDL_DIRECTORY),
     input:
+        pudl=demand_pudl_paths,
         network=RESOURCES + "{interconnect}/elec_base_network.nc",
         demand_files=demand_raw_data,
         demand_scaling_file=demand_scaling_data,
@@ -480,7 +582,6 @@ rule build_industry_demand:
         profile_year=pd.to_datetime(config["snapshots"]["start"]).year,
         eia_api=config_provider("api", "eia"),
         snapshots=config_provider("snapshots"),
-        pudl_path=config_provider("pudl_path"),
     input:
         network=RESOURCES + "{interconnect}/elec_base_network.nc",
         demand_files=demand_raw_data,
@@ -629,8 +730,9 @@ rule build_fuel_prices:
     params:
         snapshots=config["snapshots"],
         api_eia=config["api"]["eia"],
-        pudl_path=config_provider("pudl_path"),
+        pudl_path=str(PUDL_DIRECTORY),
     input:
+        pudl=PUDL_FUEL_PRICE_PATHS,
         gas_balancing_area=ba_gas_dynamic_fuel_price_files,
     output:
         state_ng_fuel_prices=RESOURCES + "{interconnect}/state_ng_power_prices.csv",
@@ -665,8 +767,10 @@ def dynamic_fuel_price_files(wildcards):
 
 rule build_powerplants:
     params:
-        pudl_path=config_provider("pudl_path"),
+        pudl_path=str(PUDL_DIRECTORY),
+        renewable_weather_year=config_provider("renewable_weather_years"),
     input:
+        pudl=PUDL_POWERPLANT_PATHS,
         wecc_ads="repo_data/WECC_ADS_public",
         eia_ads_generator_mapping="repo_data/WECC_ADS_public/eia_ads_generator_mapping_updated.csv",
         fuel_costs="repo_data/plants/fuelCost22.csv",

workflow/rules/common.smk

@@ -7,8 +7,11 @@ from functools import partial, lru_cache
 
 import os, sys, glob
 
-path = workflow.source_path("../scripts/_helpers.py")
-sys.path.insert(0, os.path.dirname(path))
+# Cache local modules before importing them from Snakemake's source cache.
+for source in ("../scripts/_helpers.py", "../scripts/constants.py"):
+    source_dir = os.path.dirname(workflow.source_path(source))
+    if source_dir not in sys.path:
+        sys.path.insert(0, source_dir)
 
 from _helpers import validate_checksum, update_config_from_wildcards
 from constants import HOURS_PER_YEAR

workflow/scripts/build_cost_data.py

@@ -63,17 +63,8 @@
 ]  # https://github.com/NREL/ReEDS-2.0/blob/e65ed5ed4ffff973071839481309f77d12d802cd/inputs/plant_characteristics/maxage.csv#L4
 
 
-def create_duckdb_instance():
-    """Set up DuckDB to read parquet files directly."""
-    duckdb.connect(database=":memory:", read_only=False)
-    # Install httpfs extension to access remote files if needed
-    duckdb.query("INSTALL httpfs;")
-
-
 def load_pudl_atb_data(parquet_path: str):
     """Loads ATB data directly from parquet files."""
-    create_duckdb_instance()
-
     query = f"""
     WITH finance_cte AS (
         SELECT

workflow/scripts/build_cutout.py

@@ -139,4 +139,8 @@
     logging.info(f"Preparing cutout with parameters {cutout_params}.")
     features = cutout_params.pop("features", None)
     cutout = atlite.Cutout(snakemake.output[0], **cutout_params)
-    cutout.prepare(features=features)
+    cutout.prepare(
+        features=features,
+        monthly_requests=True,
+        concurrent_requests=False,
+    )

workflow/scripts/build_demand.py

@@ -255,9 +255,6 @@ def _read_data(self) -> pd.DataFrame:
 
     def _read_census_data(self) -> pd.DataFrame:
         """Reads in census data for population weighting using parquet."""
-        duckdb.connect(database=":memory:", read_only=False)
-        duckdb.query("INSTALL httpfs;")
-
         parquet_path = snakemake.params.pudl_path
 
         sql = f"""
@@ -699,7 +696,7 @@ class ReadCliu(ReadStrategy):
     - MECS is used to scale county level data
 
     Sources:
-    - https://data.nrel.gov/submissions/97
+    - https://data.nlr.gov/submissions/97
     - https://www.eia.gov/consumption/manufacturing/data/2014/#r3 (Table 3.2)
     - https://loadshape.epri.com/enduse
     - https://github.com/NREL/Industry-Energy-Tool/
@@ -1814,7 +1811,7 @@ class WriteIndustrial(WriteStrategy):
     """
     Based on county level energy use from 2014.
 
-    https://data.nrel.gov/submissions/97
+    https://data.nlr.gov/submissions/97
     """
 
     def __init__(self, n: pypsa.Network, filepath: str) -> None:
@@ -2022,9 +2019,7 @@ def assign_scaler(self):  # type DemandScaler
             assert self.api, "Must provide eia api key"
             return AeoEnergyScaler(self.api)
         elif self.scaling_method == "aeo_electricity":
-            assert self.filepath.startswith(
-                "s3://pudl.catalyst.coop/",
-            ), "Must provide pudl S3 URL (s3://pudl.catalyst.coop/...)"
+            assert self.filepath, "Must provide a local PUDL directory"
             return AeoElectricityScaler(self.filepath)
         elif self.scaling_method == "efs":
             assert self.filepath.endswith(".csv"), "Must provide EFS.csv data"
@@ -2126,9 +2121,6 @@ def get_projections(self) -> pd.DataFrame:
         | 2049 |  ###  |  ###  |
         | 2050 |  ###  |  ###  |
         """
-        duckdb.connect(database=":memory:", read_only=False)
-        duckdb.query("INSTALL httpfs;")
-
         query = f"""
         SELECT
             projection_year,

workflow/scripts/build_natural_gas.py

@@ -853,7 +853,7 @@ def _expand_costs(self, n: pypsa.Network, costs: pd.DataFrame) -> pd.DataFrame:
         expanded_costs = []
         for invesetment_period in n.investment_periods:
             # reindex to match any tsa
-            cost = costs.copy()
+            cost = costs.loc[costs.index.year == self.year].copy()
             cost.index = cost.index.map(lambda x: x.replace(year=invesetment_period))
             cost = cost.reindex(n.snapshots.get_level_values(1), method="nearest")
             # set investment periods

workflow/scripts/build_powerplants.py

@@ -12,11 +12,6 @@
 logger = logging.getLogger(__name__)
 
 
-def initialize_duckdb():
-    duckdb.connect(database=":memory:", read_only=False)
-    duckdb.query("INSTALL httpfs;")
-
-
 def load_eia_operable_data(parquet_path: str):
     """Queries the parquet files directly for operable plant data."""
     return duckdb.query(
@@ -694,9 +689,9 @@ def apply_cems_heat_rates(plants, crosswalk_fn, cems_fn):
     start_date = f"{data_year}-01-01"
     end_date = f"{data_year + 1}-01-01"
 
-    initialize_duckdb()
-    eia_data_operable = load_eia_operable_data(snakemake.params.pudl_path)
-    heat_rates = load_heat_rates_data(snakemake.params.pudl_path, start_date, end_date)
+    pudl_path = snakemake.params.pudl_path
+    eia_data_operable = load_eia_operable_data(pudl_path)
+    heat_rates = load_heat_rates_data(pudl_path, start_date, end_date)
 
     eia_data_operable = merge_fc_hr_data(
         eia_data_operable,

workflow/scripts/build_sector_costs.py

@@ -2,7 +2,7 @@
 Builds costs data from EFS studies.
 
 https://www.nrel.gov/docs/fy18osti/70485.pdf
-https://data.nrel.gov/submissions/78
+https://data.nlr.gov/submissions/78
 """
 
 from abc import ABC, abstractmethod
@@ -22,7 +22,7 @@
 
 # fixed maintenance costs in %/year
 # From https://www.nrel.gov/docs/fy18osti/71500.pdf
-# Data in Sheet "13" at https://data.nrel.gov/submissions/93
+# Data in Sheet "13" at https://data.nlr.gov/submissions/93
 # manually claculated as fixed/capex
 ICE_MAINTENANCE_COSTS = {  # units of % / year
     "light_duty_cars": 11.34,
@@ -43,7 +43,7 @@
 # maintenance costs in $/mile. This gave very high maintenance costs
 # when using exogenous lifetime miles assumption.
 # From https://www.nrel.gov/docs/fy18osti/71500.pdf
-# Data in Sheet "A2" at https://data.nrel.gov/submissions/93
+# Data in Sheet "A2" at https://data.nlr.gov/submissions/93
 """
 ICE_MAINTENANCE_COSTS = {  # units of $ / miles
     "light_duty_cars": 0.033,
@@ -96,7 +96,7 @@ class EfsTechnologyData:
 
     args:
         file_path: str
-            Path to this file https://data.nrel.gov/submissions/78 which is the
+            Path to this file https://data.nlr.gov/submissions/78 which is the
             data from this report https://www.nrel.gov/docs/fy18osti/70485.pdf
 
     Buildings sector will return both commercial and residential data
@@ -254,7 +254,7 @@ def get_capex(self):  # noqa: D102
         df = df[
             (df.Sector == "Transportation") & (df["EFS Case"] == self.efs_case) & (df.Metric == "Capital Cost")
         ].copy()
-        source = "NREL EFS at https://data.nrel.gov/submissions/78"
+        source = "NREL EFS at https://data.nlr.gov/submissions/78"
         description = "Supplemented with NREL ATB"
         df["Metric"] = "investment"
         df = self._format_data_structure(df, source=source, description=description)
@@ -278,7 +278,7 @@ def get_efficiency(self):
         df = df[
             (df.Sector == "Transportation") & (df["EFS Case"] == self.efs_case) & (df.Metric == "Main Efficiency")
         ].copy()
-        source = "NREL EFS at https://data.nrel.gov/submissions/78"
+        source = "NREL EFS at https://data.nlr.gov/submissions/78"
         df["Metric"] = "efficiency"
         df = self._format_data_structure(df, source=source, description="")
         df = self._correct_efficiency_units(df)
@@ -329,7 +329,7 @@ class EfsIceTransportationData:
     Only contains ICE vehicles, as this data is manually scraped.
 
     - See Table 5 in https://www.nrel.gov/docs/fy18osti/70485.pdf for the sources
-    - See this file for the raw data https://data.nrel.gov/submissions/93
+    - See this file for the raw data https://data.nlr.gov/submissions/93
 
     Args:
         file_path: str
@@ -511,7 +511,7 @@ def get_capex(self):  # noqa: D102
             & (df.Metric == "Installed Cost")
             & (df.Units.str.startswith("2016$/kBtu/hr"))
         ].copy()
-        source = "NREL EFS at https://data.nrel.gov/submissions/78"
+        source = "NREL EFS at https://data.nlr.gov/submissions/78"
         description = ""
         df["Metric"] = "investment"
         df = self._format_data_structure(df, source=source, description=description)
@@ -529,7 +529,7 @@ def get_lifetime(self):  # noqa: D102
     def get_efficiency(self):  # noqa: D102
         df = self.data.copy()
         df = df[(df.Sector == "Buildings") & (df["EFS Case"] == self.efs_case) & (df.Metric == "Efficiency")].copy()
-        source = "NREL EFS at https://data.nrel.gov/submissions/78"
+        source = "NREL EFS at https://data.nlr.gov/submissions/78"
         df["Metric"] = "efficiency"
         df["Units"] = "per unit"
         df = self._format_data_structure(df, source=source, description="")