diff --git a/context/README.md b/context/README.md index 6da12bdd..0b8cbfd9 100644 --- a/context/README.md +++ b/context/README.md @@ -127,7 +127,8 @@ ResStock, EIA, NYISO, ISO-NE: where data comes from, how to read and prepare it. | File | Purpose | | -------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | resstock_data_preparation_run_order.md | Run order for ResStock `_sb` release prep: recommended `main.py` invocation (single command covers fetch → metadata → approx → MF adj → utility assignment → monthly loads → upload), plus legacy individual Justfile step reference for debugging | -| resstock_sb_release_pipeline_main_py.md | Unified `data/resstock/main.py` pipeline: step-by-step reference for all implemented steps (2b through 2d), constants (`_SB_EXCLUDED_FILE_TYPES`, `sb_file_types`, column sets), config.yaml, state_configs.yaml (`SUPPORTED_UTILITY_STATES`, polygon filenames, per-state settings), sample mode, `path_raw` vs `path_sb` data flow, file types included/excluded in `_sb`, how to expand `_sb` scope | +| resstock_sb_release_pipeline_main_py.md | Unified `data/resstock/main.py` pipeline: step-by-step reference for all implemented steps (2b through 2d), constants (`SB_CLONE_EXCLUDED_FILE_TYPES`, `sb_file_types`, column sets), config.yaml, state_configs.yaml (`SUPPORTED_UTILITY_STATES`, polygon filenames, per-state settings), sample mode, `path_raw` vs `path_sb` data flow, file types included/excluded in `_sb`, how to expand `_sb` scope | +| resstock_sb_annual_load_curves.md | `_sb` `load_curve_annual` from modified hourly (combined with monthly in `aggregate_loads`): one hourly read → monthly write + annual row; keep/drop/aggregate decisions; upgrade 01–05 savings; `upgrade_name`; energy_delivered 4× bug (bsf before 1.6.6); pipeline step 2d | | resstock_lmi_metadata_guide.md | ResStock 2024.2 parquet metadata: columns for LMI tier assignment, FPL/SMI, income | | parquet_reads_local_vs_s3.md | Reading Parquet from local disk vs S3: per-GET overhead, file discovery, Hive filters vs path construction, best practices for whole-state and per-utility reads | | polars_laziness_and_validation.md | Polars LazyFrame best practices: when to collect, runtime data-quality asserts vs laziness/streaming, strategies for small and large data | @@ -145,7 +146,7 @@ ResStock, EIA, NYISO, ISO-NE: where data comes from, how to read and prepare it. | ny_genability_charge_fetch_map.md | Exhaustive charge-level table for NY Genability tariffs: tariffRateId, fetch_type, variableRateKey, master_charge, decision; used to fetch 2025 monthly rates from Arcadia API for top-up implementation. | | psegli_revenue_requirement_estimation.md | PSEG-LI revenue requirement estimation: why no rate case, TOU overcount problem, bill-proportional method from LIPA budget, charge classification, implementation via `estimate_psegli_rr.py`. | | tariff_generation_pipeline.md | Tariff generation pipeline: Genability → monthly_rates YAML → flat and default-structure URDB v7 JSONs. Covers fetch_monthly_rates.py, charge classification, monthly_rates YAML structure, create_flat_tariffs.py (top-down from RR), create_default_structure_tariffs.py (bottom-up from filed rates), BASE_TARIFF_PATTERN, Justfile wiring (all-pre), CAIRO calibration, and why flat vs default rates differ. | -| md_tariff_fetch.md | MD electric and gas tariff fetching pipeline: how `fetch_electric_tariffs_genability.py` produces URDB JSONs for all 10 MD electric utilities directly from Arcadia (bypassing the NY/RI monthly_rates YAML pipeline), gas fetching via RateAcuity + Chesapeake county split + UGI PGC, three `tariff_fetch` bug patches (empty-band applicability, calculationFactor fold, duplicate inline+rider dedup), verified rate values per utility, and remaining work (UGI/Easton commodity merge). | +| md_tariff_fetch.md | MD electric and gas tariff pipeline: Arcadia electric URDBs, RateAcuity gas tariffs, Chesapeake county and RES-1/RES-2 mapping from `_sb` annual therms, UGI/Easton commodity rates, `tariff_fetch` bug patches, and verified rate values. | ### code/marginal_costs/ diff --git a/context/code/data/md_tariff_fetch.md b/context/code/data/md_tariff_fetch.md index 39755f97..35a88f12 100644 --- a/context/code/data/md_tariff_fetch.md +++ b/context/code/data/md_tariff_fetch.md @@ -228,6 +228,15 @@ based on **annual consumption** reviewed at calendar year-end: - **RES-1** (≤ 150 therms/year): lower volumetric rate, lower fixed charge (`$8/month`) - **RES-2** (> 150 therms/year): lower volumetric rate but higher fixed charge (`$10/month`) +`utils/pre/gas_tariff_mapper.py` derives annual therms from +`out.natural_gas.total.energy_consumption.kwh` in the generated Switchbox +`res_2024_amy2018_2_sb/load_curve_annual` dataset (÷ 29.3001 kWh/therm). The shared +`rate_design/hp_rates/Justfile` passes the `_sb` annual directory for the requested +upgrade. This is intentional: `_sb` annual totals are rebuilt from modified `_sb` +hourly curves, so Chesapeake tariff classes reflect non-HP approximation and any +other Switchbox load adjustments. The mapper does not fall back to raw NREL annual +loads when the `_sb` annual directory is missing. + The "lower rate, higher fixed charge" pattern at higher usage is consistent with Maryland rate design practice: high-use customers (predominantly space-heating customers) pay more in fixed charges to reduce volumetric cross-subsidization. diff --git a/context/code/data/resstock_sb_annual_load_curves.md b/context/code/data/resstock_sb_annual_load_curves.md new file mode 100644 index 00000000..dfa74ff9 --- /dev/null +++ b/context/code/data/resstock_sb_annual_load_curves.md @@ -0,0 +1,245 @@ +# `_sb` annual load curves from modified hourly + +How we build `load_curve_annual` for the Switchbox `_sb` ResStock release (`res_2024_amy2018_2_sb`), and why specific columns are kept, aggregated, or dropped. + +**Implementation:** `data/resstock/load_curve/aggregate_loads.py` — one hourly read per building conditionally produces monthly parquet **and/or** an in-memory annual row, controlled by `add_monthly` / `add_annual` flags in `process_upgrade`. Rows are concatenated into one ResStock-style annual file after the thread pool finishes. + +**Pipeline:** `data/resstock/main.py` step 2d (`_add_aggregate_loads`) runs when `--add-monthly-loads` and/or `--add-annual-loads` is True (both default True) and `load_curve_hourly` is in `--file-types`. Manifest step `add_aggregate_loads` records whichever of `load_curve_monthly` / `load_curve_annual` were produced. + +**Related:** `resstock_sb_release_pipeline_main_py.md` (pipeline / `SB_CLONE_EXCLUDED_FILE_TYPES`), `approximate_non_hp_load.md` (hourly HVAC rewrite), `investigate_resstock_eia_load_discrepancy.md` / MF adj (hourly electricity rewrite). + +**Downstream use:** Maryland gas tariff mapping reads this `_sb` annual dataset. +`utils/pre/gas_tariff_mapper.py` converts +`out.natural_gas.total.energy_consumption.kwh` to annual therms and assigns +Chesapeake RES-1 (≤150 therms) or RES-2 (>150 therms). It deliberately does not +use or fall back to raw NREL annual totals. + +--- + +## Why this exists + +The `_sb` release modifies **hourly** load curves in place: + +1. Non-HP approximation (upgrade 02): rewrites heating/cooling energy consumption and `out.load.{heating,cooling}.energy_delivered.kbtu` for selected MF highrise buildings. +2. Multifamily electricity adjustment: scales selected electricity columns. + +NREL’s shipped `load_curve_annual` is computed from **unmodified** 15-minute/hourly simulations. Copying raw annual into `_sb` would disagree with `_sb` hourly. Raw annual remains listed in `SB_CLONE_EXCLUDED_FILE_TYPES` so **clone** never copies it; `_sb` annual is generated from modified hourly instead. + +**Prerequisite:** Hourly curves must have been downloaded/aggregated with **buildstock-fetch ≥ 1.6.6**. See [Energy delivered and the 4× bug](#energy-delivered-and-the-4-bug) below. Re-download raw hourly before building `_sb` if older hourly (mean-aggregated delivered) is still on disk or S3. + +--- + +## High-level procedure (combined monthly + annual) + +I/O dominates, so each worker does **one** hourly read: + +For each `(state, upgrade)` building parquet under `_sb` `load_curve_hourly/`: + +1. Read hourly parquet once into memory. +2. **Monthly** (when `add_monthly=True`): `group_by("month")` with the **full** bsf column-aggregation CSV (sum for energy / emissions / energy_delivered; mean for temperatures; first for `bldg_id`) → write one monthly parquet per building. +3. **Annual** (when `add_annual=True`): apply the **same** bsf CSV but only the [annual metric subset](#columns-aggregated-from-hourly--sb-annual) → one in-memory row (`bldg_id` + energy `.kwh` + delivered). + +After all buildings for the upgrade: + +4. Concatenate annual rows. +5. From raw NREL `load_curve_annual`, keep `bldg_id`, `upgrade`, `weight`, `out.params.*`, and `upgrade_name` when present. +6. Left-join aggregated metrics onto that slim slice on `bldg_id`. +7. Write one consolidated parquet under `_sb` `load_curve_annual/…`. +8. Pipeline syncs whichever of `load_curve_monthly/state=/` and `load_curve_annual/state=/` were produced to S3. + +CLI (both monthly + annual): + +```bash +uv run python data/resstock/load_curve/aggregate_loads.py \ + --path-input /ebs/data/nrel/resstock/res_2024_amy2018_2_sb \ + --path-output /ebs/data/nrel/resstock/res_2024_amy2018_2_sb \ + --path-annual-raw /ebs/data/nrel/resstock/res_2024_amy2018_2 \ + --state CT --upgrade-ids "00 02" \ + --bsf-release res_2024_amy2018_2 --workers 50 \ + --add-monthly --add-annual +``` + +Annual-only (same CLI, omit `--add-monthly`): + +```bash +uv run python data/resstock/load_curve/aggregate_loads.py \ + --path-input /ebs/data/nrel/resstock/res_2024_amy2018_2_sb \ + --path-output /ebs/data/nrel/resstock/res_2024_amy2018_2_sb \ + --path-annual-raw /ebs/data/nrel/resstock/res_2024_amy2018_2 \ + --state CT --upgrade-ids "00 02" \ + --bsf-release res_2024_amy2018_2 --workers 50 \ + --add-annual +``` + +`load_curve_annual` stays in `SB_CLONE_EXCLUDED_FILE_TYPES` so clone does not overwrite generated `_sb` annual with raw NREL. The file type becomes expected on `_sb` once the manifest records an `add_aggregate_loads` step. + +--- + +## Schemas (ResStock 2024.2, verified MD/CT) + +| Source | Typical width | Notes | +| --------------------------- | ------------- | ------------------------------------------------------------- | +| Hourly (all upgrades 00–05) | ~142 cols | Identical energy/delivered/temp column set across upgrades | +| Annual upgrade 00 | ~112 cols | Absolute metrics only | +| Annual upgrades 01–05 | ~200–202 cols | Absolute metrics **plus** baseline-relative extras (~90 cols) | + +Hourly energy and the three load-delivered columns map 1:1 onto annual absolute columns for upgrades **00–05** (50 energy columns; 0 missing / 0 extra after `+.kwh` rename). + +--- + +## Columns aggregated from hourly → `_sb` annual + +Annual uses the **same** bsf aggregation CSV as monthly (`res_2024_amy2018_2.csv`, etc.), filtered to: + +- columns ending in `.energy_consumption` but **not** `.energy_consumption_intensity` +- any column whose name contains `energy_delivered` + +Aggregation function comes from the CSV (`sum` for these under bsf ≥ 1.6.6). Energy columns are renamed with a `.kwh` suffix for the annual file. + +### Energy consumption (sum + rename) + +| Hourly name | Annual name | Aggregation | +| -------------------------------------- | -------------------------- | ----------- | +| `*.energy_consumption` (not intensity) | `*.energy_consumption.kwh` | `sum` | + +Example: `out.electricity.heating.energy_consumption` → `out.electricity.heating.energy_consumption.kwh`. + +There are **50** such end-use/fuel columns in 2024.2 (electricity, natural gas, fuel oil, propane, site_energy nets/totals, etc.). Mapping was checked for MD upgrades 00–05: every hourly energy column has an annual `.kwh` counterpart and vice versa. + +**Why aggregate these:** They are the quantities `_sb` modifies (HVAC approximation, MF electricity adj). Annual must reflect those edits. Units are already kWh-equivalent in the hourly series; annual only adds the `.kwh` suffix in the column name. + +**Why not intensities:** Annual NREL files have no intensity columns; intensities are rates, not annual totals. Monthly still aggregates them via the full bsf rule set. + +### Load delivered (sum, no rename) + +| Column | Aggregation | +| ------------------------------------------ | ----------- | +| `out.load.heating.energy_delivered.kbtu` | `sum` | +| `out.load.cooling.energy_delivered.kbtu` | `sum` | +| `out.load.hot_water.energy_delivered.kbtu` | `sum` | + +Same names in hourly and annual (already `.kbtu`). + +**Why include them:** Heating/cooling delivered are rewritten by non-HP approximation on upgrade 02; hot water is not rewritten there but is still a physical annual total that should stay consistent with the (corrected) hourly series. After bsf ≥ 1.6.6, Σ(hourly) matches NREL annual for unmodified buildings. + +**No unit conversion:** Do not apply a kWh↔kBtu factor; both schemas label these columns as kBtu. + +--- + +## Columns kept from raw NREL annual (not recomputed) + +| Column(s) | When present | Why keep | +| -------------- | ----------------------- | -------------------------------------------------------------------------------------------------------- | +| `bldg_id` | Always | Join key | +| `upgrade` | Always | Upgrade id | +| `weight` | Always | Sample weight; not in hourly files | +| `out.params.*` | Always (~20) | Building geometry/area parameters only in annual | +| `upgrade_name` | Upgrades **01–05** only | Human-readable package label (e.g. “High efficiency cold-climate heat pump…”); not derivable from hourly | + +Upgrade **00** has no `upgrade_name`; the selector keeps it only if the column exists. + +These fields are independent of the hourly load edits, so copying them from raw annual is correct. + +--- + +## Columns deliberately not produced + +### From hourly: not summed into `_sb` annual + +| Hourly columns | Why dropped | +| ------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------- | +| `*.energy_consumption_intensity` (~50) | Annual has no intensity columns; intensities are rates, not annual totals | +| Zone / outdoor temperatures (`*.c`, ~13) | Annual has no temps; hourly aggregation uses **mean** for these in bsf (correct for temperatures). Summing would be meaningless for our annual file | +| Time keys (`timestamp`, `year`, `month`, `day`, `hour`) | Not annual attributes | +| Hourly `out.total.lrmer_*` emissions (~20) | We do not rebuild emissions for `_sb` annual (see below). Could be summed later if needed | + +### From raw annual: not copied into `_sb` annual + +Everything not listed in [Columns kept from raw NREL annual](#columns-kept-from-raw-nrel-annual-not-recomputed) is dropped from the raw file, including absolute energy/load/bills/peaks/emissions that would otherwise conflict with recomputed `_sb` totals. + +#### Absolute metrics (present on upgrade 00 and upgrades 01–05) + +| Category | Examples | Why drop | +| ------------------ | ------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Energy consumption | `*.energy_consumption.kwh` | Replaced by Σ(`_sb` hourly) | +| Load delivered | `out.load.*.energy_delivered.kbtu` | Replaced by Σ(`_sb` hourly) | +| Emissions | `out.emissions.*` / LRMER columns | Not recomputed from `_sb` hourly; raw values would not match modified loads | +| Bills | `out.bills.*.usd` | Tariff/assumption-dependent; not from `_sb` load edits | +| Peaks | `out.electricity.*.peak.kw`, `out.load.*.peak.kbtu_hr` | Need **max** (or NREL’s peak definition) over the year, not a sum; after HVAC/MF edits, peaks would need a separate max-over-hourly pass we are not doing yet | +| Hot water volumes | `out.hot_water.*.gal` | Not in hourly files | +| Unmet hours | `out.unmet_hours.*.hour` | Not in hourly files | +| Energy burden | `out.energy_burden.percentage` | Not from load curves | + +#### Baseline-relative extras (upgrades **01–05** only; ~90 columns) + +NREL annual files for non-baseline upgrades add columns that compare the upgrade to upgrade 00. Empirically (MD), for energy: + +`*.savings` ≈ `upgrade_00_value − upgrade_N_value` (positive = savings). + +| Category | Count (typical) | Pattern | Why drop | +| -------------------------- | --------------- | ------------------------------------------------ | --------------------------------------------------------------------------- | +| Energy consumption savings | 50 | `*.energy_consumption.kwh.savings` | Precomputed vs **NREL** baseline, not `_sb` u00; invalid after hourly edits | +| Load delivered savings | 3 | `out.load.*.energy_delivered.kbtu.savings` | Same | +| Bill savings | 5 | `out.bills.*.usd.savings` | Same + tariff-dependent | +| Peak savings | 4 | `*.peak.*.savings` | Same + peaks not rebuilt | +| Hot water volume savings | 4 | `out.hot_water.*.gal.savings` | Not in hourly; baseline-relative | +| Unmet hours savings | 2 | `out.unmet_hours.*.hour.savings` | Same | +| Energy burden savings | 1 | `out.energy_burden.percentage.savings` | Same | +| Emissions reductions | 20 | `out.emissions_reduction.{fuel}.lrmer_*.co2e_kg` | Same idea as savings; named `emissions_reduction` rather than `*.savings` | + +**Kept among the upgrade-only fields:** only `upgrade_name` (see above). + +Recomputing savings correctly would mean differencing `_sb` upgrade _N_ annual against `_sb` upgrade 00 annual after both are rebuilt — explicitly out of scope for this step. + +#### Upgrade 03 quirk (ignored) + +On MD, upgrade 03’s annual file is missing absolute `out.energy_burden.percentage` while still having `.savings`. We do not special-case this: same keep list as other upgrades (`bldg_id`, `upgrade`, `weight`, optional `upgrade_name`, `out.params.*`). + +--- + +## Energy delivered and the 4× bug + +### Symptom + +For older hourly downloads, Σ(hourly `out.load.*.energy_delivered.kbtu`) was **exactly 1/4** of NREL annual (and of Σ(15-minute)). Electricity `*.energy_consumption` summed correctly (ratio ≈ 1). + +### Cause + +`buildstock-fetch` builds hourly from NREL’s native **15-minute** timeseries using per-column rules in `buildstock_fetch/data/load_curve_column_map/*.csv`. Energy columns use `sum`; temperatures use `mean` (correct). + +A Jan 2026 change set the three `energy_delivered` columns to `mean` (treating kBtu as a “power-like” quantity). Averaging four 15-minute energy intervals into one hour understates hourly energy by 4×, so the annual sum of hourly is 4× low. + +### Fix + +| Artifact | Status | +| -------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------ | +| GitHub `main` / **buildstock-fetch ≥ 1.6.6** | `energy_delivered` → `sum` again in 2022/2024 maps (and 2025 maps with NREL’s `energy_delivered..kbtu` double-dot names) | +| PyPI **1.6.5 and earlier** | Still had `mean` | +| This repo | Pins `buildstock-fetch>=1.6.6` in `pyproject.toml` | + +Verified with bsf 1.6.6: fresh hourly downloads and re-aggregation of stored 15-minute files give annual / Σ(hourly) ≈ **1.000** for heating/cooling/hot-water delivered. + +**Operational implication:** Hourly already on S3/local from older bsf must be **re-downloaded** (or re-aggregated from 15-minute) before `_sb` annual is trusted for delivered columns. Energy consumption columns were never affected by this bug. + +Temperature columns remaining as `mean` in the CSV maps is intentional and unrelated. + +--- + +## What `_sb` annual contains (summary) + +Per building, one row with approximately: + +- Identity / sample: `bldg_id`, `upgrade`, `weight`, optional `upgrade_name` +- ~20 `out.params.*` +- 50 `*.energy_consumption.kwh` from Σ(`_sb` hourly) +- 3 `out.load.*.energy_delivered.kbtu` from Σ(`_sb` hourly) + +No savings, emissions, bills, peaks, gallons, unmet hours, or energy burden. + +--- + +## Validation notes used when designing this + +- **Name mapping:** MD upgrades 00–05 — 50 hourly energy ↔ 50 annual `.kwh`; three delivered names match. +- **Unmodified buildings:** With bsf ≥ 1.6.6, Σ(hourly) matches NREL annual for energy and delivered (float noise ~1e−5 relative). +- **Modified buildings:** `_sb` annual will **diverge** from NREL annual wherever hourly was rewritten; that divergence is the point of the `_sb` annual file. diff --git a/context/code/data/resstock_sb_release_pipeline_main_py.md b/context/code/data/resstock_sb_release_pipeline_main_py.md index ab304040..47acf55c 100644 --- a/context/code/data/resstock_sb_release_pipeline_main_py.md +++ b/context/code/data/resstock_sb_release_pipeline_main_py.md @@ -19,8 +19,8 @@ For the older Justfile-based workflow, see `context/code/data/resstock_data_prep | 5. Approximate non-HP load | Step 2c-i: `_approximate_non_hp_load` | Implemented | | 6. Adjust MF electricity | Step 2c-ii: `_adjust_mf_electricity` | Implemented | | 7. Sync `_sb` to EBS | N/A (pipeline writes directly to EBS) | N/A | -| 8. Add monthly load curves | Step 2d: `_add_monthly_loads` | Implemented | -| 9. Upload monthly to S3 | Step 2d: `_add_monthly_loads` | Implemented | +| 8. Aggregate load curves | Step 2d: `_add_aggregate_loads` | Implemented | +| 9. Upload aggregated to S3 | Step 2d: `_add_aggregate_loads` | Implemented | | Upload raw + `_sb` to S3 | Step 3: `_upload` | Implemented | --- @@ -52,7 +52,8 @@ Release-level defaults are loaded from `data/resstock/config.yaml`. State-specif | `--gas-poly-filename` | from `state_configs.yaml` | Gas utility polygon CSV; overrides config default | | `--path-s3-gis-dir` | `s3://data.sb/gis/utility_boundaries/` | S3 directory for NY utility polygon CSVs | | `--add-monthly-loads` | `True` | Aggregate hourly → monthly and upload (needs `load_curve_hourly` in `--file-types`) | -| `--monthly-workers` | `50` | Parallel worker count for monthly aggregation | +| `--add-annual-loads` | `True` | Aggregate hourly → annual and upload (needs `load_curve_hourly` in `--file-types`) | +| `--aggregation-workers` | `50` | Parallel worker count for load curve aggregation | | `--path-output-dir` | `/ebs/data/nrel/resstock` | Local EBS output root | | `--path-s3-dir` | `s3://data.sb/nrel/resstock` | S3 mirror root | @@ -127,22 +128,24 @@ RI: **Dynamic dispatch:** `assign_utility.py` uses `importlib.import_module()` to load the module at `utility_assignment.module`, merges `utility_assignment.kwargs` with any CLI overrides (`--electric-poly-filename`, `--gas-poly-filename`, `--path-s3-gis-dir`), and calls `mod.assign_utility(metadata, **kwargs)`. Each state module is responsible for its own data loading and assignment logic. -### `_SB_EXCLUDED_FILE_TYPES` (module-level constant in `main.py`) +### `SB_CLONE_EXCLUDED_FILE_TYPES` (module-level constant in `constants.py`) ```python -_SB_EXCLUDED_FILE_TYPES: frozenset[str] = frozenset({"load_curve_annual"}) +SB_CLONE_EXCLUDED_FILE_TYPES: frozenset[str] = frozenset({"load_curve_annual"}) ``` File types that are fetched for the raw NREL release but **never copied to `_sb`**, never uploaded under the `_sb` prefix, and never validated against `_sb`. Currently contains only `load_curve_annual`. -**Why `load_curve_annual` is excluded:** The `_sb` release modifies `load_curve_hourly` in place (non-HP approximation, MF electricity adjustment). There is no mechanism to re-derive `load_curve_annual` from the modified hourly data, so including the unmodified raw annual in `_sb` would be misleading — it would not reflect the approximation or adjustment. The correct sub-annual aggregation is `load_curve_monthly`, which is derived from the modified hourly in step 2d. +**Why `load_curve_annual` is excluded (today):** The `_sb` release modifies `load_curve_hourly` in place (non-HP approximation, MF electricity adjustment). Copying unmodified NREL annual into `_sb` would disagree with those hourly edits. Sub-annual aggregation already in the pipeline is `load_curve_monthly` (step 2d). -**How to expand `_sb` to include `load_curve_annual` in the future:** If a need arises for an `_sb` annual file (e.g., a downstream consumer requires it), the steps would be: +**Status of `_sb` annual:** Produced in pipeline step 2d (`_add_aggregate_loads`) when `--add-annual-loads True`: one hourly read → annual row; consolidated annual joined to raw params. Details in **`resstock_sb_annual_load_curves.md`**. Raw `load_curve_annual` remains in `SB_CLONE_EXCLUDED_FILE_TYPES` so **clone** never copies NREL annual onto `_sb`; the generated file is uploaded from the aggregation step and expected by the manifest when `add_aggregate_loads` has run. -1. Implement an hourly-to-annual aggregation script (analogous to `data/resstock/load_curve/add_monthly_loads.py` but aggregating to 1 row per building). -2. Add a pipeline step after all hourly modifications are complete (after step 2c-ii) that runs the aggregation on the `_sb` hourly files and writes `load_curve_annual/` under `path_sb`. -3. Remove `"load_curve_annual"` from `_SB_EXCLUDED_FILE_TYPES` so the clone, upload, and validation steps include it. -4. Ensure `_modify_metadata` still reads `load_curve_annual` from `path_raw` (the raw release) for the `identify_natgas_connection` step, since that runs before any hourly modifications. (Or, if the new annual file is generated after modifications, decide whether natgas identification should use the pre- or post-modification annual.) +**How `_sb` annual stays consistent:** + +1. Fetch hourly with **buildstock-fetch ≥ 1.6.6**. +2. After hourly modifications (after step 2c-ii), run step 2d with `--add-monthly-loads True --add-annual-loads True` (both default True). +3. Keep `_modify_metadata` reading `load_curve_annual` from `path_raw` for `identify_natgas_connection`. +4. Do not remove `load_curve_annual` from `SB_CLONE_EXCLUDED_FILE_TYPES` unless clone logic is changed to skip overwriting generated annual. ### `data/resstock/constants.py` @@ -208,7 +211,7 @@ S3 paths: Computed at the top of `main()`: ```python -sb_file_types = [ft for ft in args.file_types if ft not in _SB_EXCLUDED_FILE_TYPES] +sb_file_types = [ft for ft in args.file_types if ft not in SB_CLONE_EXCLUDED_FILE_TYPES] ``` This list drives the clone, clone validation, `_sb` upload, and `_sb` S3 validation. With current defaults, `sb_file_types` = `["metadata", "load_curve_hourly"]`. @@ -217,7 +220,7 @@ This list drives the clone, clone validation, `_sb` upload, and `_sb` S3 validat A run record is created immediately via `new_run_record` from `data/resstock/manifest.py`. This captures the git commit, branch, CLI arguments, and all flags. The record is updated after each step via `record_step` and `upsert_run`, providing a persistent audit trail in `manifest.yaml` (one per release directory, synced to S3). -If any file type in `--file-types` is in `_SB_EXCLUDED_FILE_TYPES`, a warning is printed to stdout and recorded in the run record under a `warnings` key. This ensures the user is aware that `load_curve_annual` (or any future excluded type) is not part of the `_sb` release. +If any file type in `--file-types` is in `SB_CLONE_EXCLUDED_FILE_TYPES`, a warning is printed to stdout and recorded in the run record under a `warnings` key. This ensures the user is aware that `load_curve_annual` (or any future excluded type) is not part of the `_sb` clone. ### Crash recording @@ -334,38 +337,37 @@ Logic is in `_adjust_mf_electricity()`. For each (state, upgrade) pair in `["00" In sample mode, a warning is printed that ratios are derived from the sampled buildings only. If the sample has no MF buildings, the step is skipped; if fewer than 2 SF buildings, ratios default to 1.0. -### Step 2d: Add monthly load curves +### Step 2d: Aggregate load curves -**Function:** `_add_monthly_loads` +**Function:** `_add_aggregate_loads` -**Gate condition:** `args.add_monthly_loads and "load_curve_hourly" in args.file_types` +**Gate condition:** `(args.add_monthly_loads or args.add_annual_loads) and "load_curve_hourly" in args.file_types` -Aggregates the modified `_sb` hourly load curves into monthly load curves and uploads them to S3. This step runs after all hourly modifications (steps 2c-i, 2c-ii) so that monthly curves reflect non-HP approximation and MF electricity adjustment. +Aggregates modified `_sb` hourly into **monthly** and/or **annual** load curves and uploads produced outputs to S3. Runs after all hourly modifications (steps 2c-i, 2c-ii) so outputs reflect non-HP approximation and MF electricity adjustment. See **`resstock_sb_annual_load_curves.md`**. **Logic:** -1. Loads bsf column aggregation rules via `load_aggregation_rules(release)` (using the **raw** release name, e.g. `res_2024_amy2018_2`, not the `_sb` variant). The rules CSV lives in the bsf package (`buildstock_fetch.constants.LOAD_CURVE_COLUMN_AGGREGATION`). -2. For each (state, upgrade) pair, calls `process_upgrade(path_sb, path_sb, state, upgrade, agg_rules, workers)`, which: - - Reads all hourly parquets from `path_sb/load_curve_hourly/state=/upgrade=/` - - Groups by `month`, applies sum/mean/first rules per column, reconstructs a `timestamp` datetime column - - Writes one monthly parquet per building to `path_sb/load_curve_monthly/state=/upgrade=/` - - Runs up to `--monthly-workers` (default 50) files in parallel via `ThreadPoolExecutor` -3. After all upgrades for a state are done, uploads `path_sb/load_curve_monthly/state=/` to `s3://.../load_curve_monthly/state=/` via `aws s3 sync`. The upload is per state (covers all upgrades in one sync call). -4. Returns the list of `"state= upgrade="` labels that were processed, for manifest recording. +1. Loads bsf column aggregation rules via `load_bsf_aggregation_map(release)` (raw release name, e.g. `res_2024_amy2018_2`). CSV in `buildstock_fetch.constants.LOAD_CURVE_COLUMN_AGGREGATION`. +2. For each (state, upgrade), calls `process_upgrade(..., add_monthly=True/False, add_annual=True/False)`, which for each building (ThreadPool, `--aggregation-workers` default 50): + - Reads the hourly parquet **once** + - If `add_monthly`: writes one monthly parquet (`group_by month`, full bsf sum/mean/first rules) + - If `add_annual`: returns one annual row (same rules, subset: energy consumption + energy_delivered only) + - After the pool (when `add_annual`): concat annual rows, join raw annual params from `path_raw`, write one consolidated `_sb` annual parquet +3. After all upgrades for a state, syncs whichever of `load_curve_monthly/state=/` and `load_curve_annual/state=/` were produced to S3. +4. Returns `"state= upgrade="` labels for the manifest. Step name: `add_aggregate_loads`; `_STEP_FILE_TYPES` maps it to both `load_curve_monthly` and `load_curve_annual` (also maps individual `add_monthly_loads`/`add_annual_loads` names for legacy/future use). -**Important:** `load_curve_monthly` is **not** in `args.file_types` (which controls what is fetched from NREL and uploaded by the main `_upload` call). Monthly files are generated locally and uploaded exclusively by `_add_monthly_loads`. The main `_upload` step at the end does not touch `load_curve_monthly`. +**Important:** Neither monthly nor generated annual is in `sb_file_types` used by the final `_upload` of cloned types. Both are uploaded exclusively by `_add_aggregate_loads`. Raw `load_curve_annual` stays in `SB_CLONE_EXCLUDED_FILE_TYPES` so clone does not overwrite generated annual. -**Sample mode:** When `--sample N` is active, only N hourly files exist for each (state, upgrade). `process_upgrade` processes whatever files are present, producing N monthly files. A `NOTE` is printed but the step is not skipped. This is expected behavior for development/testing. +**Sample mode:** When `--sample N` is active, only N hourly files exist; the step writes N monthly files and/or an N-row annual file. A `NOTE` is printed but the step is not skipped. ### Step 3: Upload to S3 Uploads both releases to S3 via `aws s3 sync`: -- **Raw release**: all `args.file_types` (including `load_curve_annual`). -- **`_sb` release**: only `sb_file_types` (excludes `load_curve_annual`). - -Validation: `validate_s3_objects` spot-checks up to 5 S3 objects per `(file_type, state, upgrade)`. The `_sb` validation uses `sb_file_types`. +- **Raw release**: all `args.file_types` (including raw `load_curve_annual`). +- **`_sb` release**: `sb_file_types` (excludes raw annual from clone set). Generated monthly/annual were already uploaded in step 2d. +Validation: `validate_s3_objects` spot-checks up to 5 S3 objects per `(file_type, state, upgrade)`. The `_sb` validation uses `sb_file_types` (plus integrity expects step-produced monthly/annual when `add_aggregate_loads` ran). Manifests are uploaded separately via `_upload_manifest`. ### Finalization @@ -383,13 +385,13 @@ On success, the run record is marked `completed` and written to both manifests. | `metadata` | Cloned from raw, then modified in place (`metadata-sb.parquet`) | Contains all SB-specific columns | | `load_curve_hourly` | Cloned from raw, then modified in place by approximation and MF electricity adjustment | One parquet per building | | `metadata_utility` | Generated by `_assign_utility` (step 2b), uploaded to S3 immediately | Contains only `bldg_id`, `sb.electric_utility`, `sb.gas_utility` | -| `load_curve_monthly` | Derived by `_add_monthly_loads` (step 2d) from the modified `load_curve_hourly` | One parquet per building; synced to S3 per state after generation | +| `load_curve_monthly` | Derived by `_add_aggregate_loads` (step 2d) from the modified `load_curve_hourly` | One parquet per building; synced to S3 per state after generation | ### Currently excluded -| File type | Reason | Path to inclusion | -| ------------------- | -------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------- | -| `load_curve_annual` | No mechanism to re-derive from modified hourly; raw annual would be inconsistent with `_sb` hourly | Build an hourly-to-annual aggregation step, run it after all modifications, remove from `_SB_EXCLUDED_FILE_TYPES` | +| File type | Reason | Path to inclusion | +| ------------------- | ------------------------------------------------------------------------------------------------------ | ----------------------------------------------------------------------------------------------- | +| `load_curve_annual` | Not cloned from raw; generated from modified hourly in `_add_aggregate_loads` (with `add_annual=True`) | See `resstock_sb_annual_load_curves.md`; stays in `SB_CLONE_EXCLUDED_FILE_TYPES` for clone only | --- @@ -446,13 +448,13 @@ When `--sample N` is passed (N > 0): | `data/resstock/utility/assign_utility.py` | Dynamic dispatch facade; imports state module from `state_configs.yaml` via `importlib` | | `data/resstock/utility/assign_utility_ny.py` | NY-specific thin wrapper: builds name map, passes excluded gas utilities to generic `create_hh_utilities` in `utils.py` | | `data/resstock/utility/assign_utility_ri.py` | Deterministic utility assignment for RI (single utility) | -| `data/resstock/load_curve/add_monthly_loads.py` | Hourly-to-monthly aggregation; called directly by `_add_monthly_loads` (step 2d) | +| `data/resstock/load_curve/aggregate_loads.py` | Hourly-to-monthly/annual aggregation; called by `_add_aggregate_loads` (step 2d) | --- ## Known limitations and TODO items -1. **No `load_curve_annual` in `_sb`**: Intentional. See the `_SB_EXCLUDED_FILE_TYPES` section above for how to change this if needed. +1. **`load_curve_annual` on `_sb`:** Not cloned from raw. Generated with monthly in step 2d from modified hourly. See `SB_CLONE_EXCLUDED_FILE_TYPES` and **`resstock_sb_annual_load_curves.md`**. 2. **`has_natgas_connection` has two sources of truth**: For non-approximated buildings, it comes from `load_curve_annual` in the raw release (step 2a). For approximated buildings, it is re-derived from the modified `load_curve_hourly` in `_sb` (step 2c-i). This is correct behavior but worth understanding when debugging metadata values. diff --git a/data/resstock/Justfile b/data/resstock/Justfile index a4df9f2b..cbdc15fa 100644 --- a/data/resstock/Justfile +++ b/data/resstock/Justfile @@ -262,13 +262,26 @@ adjust-mf-electricity state input_release output_release upgrade_ids: # Aggregate _sb hourly load curves into monthly for a state. add-monthly-loads state upgrade_ids: - uv run python "{{ path_local_repo }}/data/resstock/load_curve/add_monthly_loads.py" \ + uv run python "{{ path_local_repo }}/data/resstock/load_curve/aggregate_loads.py" \ --path-input "{{ path_local_parquet }}/{{ resstock_release_sb }}" \ --path-output "{{ path_local_parquet }}/{{ resstock_release_sb }}" \ --state "{{ state }}" \ --upgrade-ids "{{ upgrade_ids }}" \ --bsf-release "{{ resstock_release }}" \ - --workers 50 + --workers 50 \ + --add-monthly + +# Aggregate _sb hourly load curves into annual for a state. +add-annual-loads state upgrade_ids: + uv run python "{{ path_local_repo }}/data/resstock/load_curve/aggregate_loads.py" \ + --path-input "{{ path_local_parquet }}/{{ resstock_release_sb }}" \ + --path-output "{{ path_local_parquet }}/{{ resstock_release_sb }}" \ + --path-annual-raw "{{ path_local_parquet }}/{{ resstock_release }}" \ + --state "{{ state }}" \ + --upgrade-ids "{{ upgrade_ids }}" \ + --bsf-release "{{ resstock_release }}" \ + --workers 50 \ + --add-annual # Upload monthly load curves for a state to S3. upload-monthly-loads state upgrade_ids: diff --git a/data/resstock/constants.py b/data/resstock/constants.py index 0ab46fdd..1ffbf109 100644 --- a/data/resstock/constants.py +++ b/data/resstock/constants.py @@ -55,12 +55,12 @@ } ) -# File types that belong only to the raw NREL release and must never be copied -# to the _sb release, uploaded under _sb, or validated against _sb. -# load_curve_annual has no post-approximation equivalent: the only valid -# aggregation of the modified _sb load curves is load_curve_monthly (derived -# from load_curve_hourly by add_monthly_loads after all modifications are done). -SB_EXCLUDED_FILE_TYPES: frozenset[str] = frozenset({"load_curve_annual"}) +# File types excluded from cloning raw → _sb. These are fetched for the raw +# release but NOT copied to _sb via clone because _sb produces its own version +# from modified hourly data. ``load_curve_annual`` is fetched raw for natgas ID, +# MF ratios, and annual params, but the _sb annual is generated by the +# aggregation step. See ``context/code/data/resstock_sb_annual_load_curves.md``. +SB_CLONE_EXCLUDED_FILE_TYPES: frozenset[str] = frozenset({"load_curve_annual"}) # ── CLI for Justfile config access ──────────────────────────────────────────── diff --git a/data/resstock/load_curve/add_monthly_loads.py b/data/resstock/load_curve/add_monthly_loads.py deleted file mode 100644 index d82da7c6..00000000 --- a/data/resstock/load_curve/add_monthly_loads.py +++ /dev/null @@ -1,215 +0,0 @@ -"""Aggregate hourly ResStock load curves into monthly load curves. - -Reads per-building hourly parquet files, aggregates to monthly using the same -column-level rules as buildstock-fetch (sum for energy/emissions, mean for -load/temperature), and writes one monthly parquet per building. - -Usage (from project root): - uv run python data/resstock/load_curve/add_monthly_loads.py \ - --path-input /ebs/data/nrel/resstock/res_2024_amy2018_2_sb \ - --path-output /ebs/data/nrel/resstock/res_2024_amy2018_2_sb \ - --state NY --upgrade-ids "00 02" \ - --bsf-release res_2024_amy2018_2 --workers 50 -""" - -from __future__ import annotations - -import argparse -import sys -import time -from concurrent.futures import ThreadPoolExecutor, as_completed -from pathlib import Path -from typing import cast - -import polars as pl - -from buildstock_fetch.constants import LOAD_CURVE_COLUMN_AGGREGATION - - -def load_aggregation_rules(release: str) -> list[pl.Expr]: - """Build polars aggregation expressions from bsf's column map CSV.""" - csv_path = LOAD_CURVE_COLUMN_AGGREGATION / f"{release}.csv" - if not csv_path.exists(): - raise FileNotFoundError( - f"bsf column map not found: {csv_path}. " - f"Available: {[p.stem for p in LOAD_CURVE_COLUMN_AGGREGATION.glob('*.csv')]}" - ) - rules_df = pl.read_csv(csv_path) - rules = dict( - zip( - rules_df["name"].to_list(), - rules_df["Aggregate_function"].to_list(), - strict=True, - ) - ) - - exprs: list[pl.Expr] = [] - for col, agg in rules.items(): - if col == "timestamp": - continue - match agg: - case "sum": - exprs.append(pl.col(col).sum()) - case "mean": - exprs.append(pl.col(col).mean()) - case "first": - exprs.append(pl.col(col).first()) - case _: - raise ValueError( - f"Unknown aggregation function '{agg}' for column '{col}'" - ) - return exprs - - -def aggregate_one_building( - input_path: Path, - output_path: Path, - agg_rules: list[pl.Expr], -) -> None: - """Read one hourly parquet, aggregate to monthly, write output.""" - output_path.parent.mkdir(parents=True, exist_ok=True) - - # bsf's column map covers out.* and bldg_id but not year/day/hour/timestamp - # (those are added by bsf after aggregation). We add year.first() here and - # reconstruct timestamp from the result. - all_rules = [*agg_rules, pl.col("year").first()] - - df = cast( - pl.DataFrame, - pl.scan_parquet(input_path) - .group_by("month") - .agg(all_rules) - .sort("month") - .collect(), - ) - - year = df["year"][0] - - df = df.with_columns( - pl.datetime(year=year, month=pl.col("month"), day=1).alias("timestamp"), - pl.col("year").cast(pl.Int32), - pl.col("month").cast(pl.Int8), - ) - - out_cols = [c for c in df.columns if c.startswith("out.")] - col_order = ["timestamp", *out_cols, "bldg_id", "year", "month"] - df = df.select(col_order) - - df.write_parquet(output_path) - - -def process_upgrade( - path_input: Path, - path_output: Path, - state: str, - upgrade: str, - agg_rules: list[pl.Expr], - workers: int, -) -> None: - """Process all building files for one upgrade.""" - input_dir = path_input / f"load_curve_hourly/state={state}/upgrade={upgrade}" - output_dir = path_output / f"load_curve_monthly/state={state}/upgrade={upgrade}" - - if not input_dir.exists(): - print(f" Input directory does not exist, skipping: {input_dir}") - return - - files = sorted(input_dir.glob("*.parquet")) - n_files = len(files) - if n_files == 0: - print(f" No parquet files found in {input_dir}") - return - - print(f" Found {n_files:,} hourly files in {input_dir}") - print(f" Output: {output_dir}") - output_dir.mkdir(parents=True, exist_ok=True) - - t0 = time.time() - done = 0 - errors = 0 - - def _process(src: Path) -> str | None: - dst = output_dir / src.name - try: - aggregate_one_building(src, dst, agg_rules) - return None - except Exception as e: - return f"{src.name}: {e}" - - with ThreadPoolExecutor(max_workers=workers) as pool: - futures = {pool.submit(_process, f): f for f in files} - for future in as_completed(futures): - result = future.result() - done += 1 - if result is not None: - errors += 1 - print(f" ERROR {result}") - if done % 5000 == 0 or done == n_files: - elapsed = time.time() - t0 - rate = done / elapsed if elapsed > 0 else 0 - print( - f" {done:,}/{n_files:,} ({rate:.0f} files/s, {elapsed:.1f}s elapsed)" - ) - - elapsed = time.time() - t0 - print(f" Done: {done:,} files in {elapsed:.1f}s ({errors} errors)") - - -def main() -> None: - parser = argparse.ArgumentParser( - description="Aggregate hourly ResStock load curves to monthly." - ) - parser.add_argument( - "--path-input", - required=True, - help="Root of the ResStock release (local), e.g. /ebs/data/nrel/resstock/res_2024_amy2018_2_sb", - ) - parser.add_argument( - "--path-output", - required=True, - help="Root to write monthly load curves into (local), e.g. /ebs/data/nrel/resstock/res_2024_amy2018_2_sb", - ) - parser.add_argument("--state", required=True, help="Two-letter state code, e.g. NY") - parser.add_argument( - "--upgrade-ids", - required=True, - help='Space-separated upgrade IDs, e.g. "00 02"', - ) - parser.add_argument( - "--bsf-release", - required=True, - help="bsf release key for column aggregation rules, e.g. res_2024_amy2018_2", - ) - parser.add_argument( - "--workers", - type=int, - default=50, - help="Number of parallel workers (default: 50)", - ) - args = parser.parse_args() - - path_input = Path(args.path_input) - path_output = Path(args.path_output) - upgrade_ids = args.upgrade_ids.split() - - if not path_input.exists(): - print(f"Error: input path does not exist: {path_input}", file=sys.stderr) - sys.exit(1) - - print(f"Loading bsf aggregation rules for release '{args.bsf_release}'...") - agg_rules = load_aggregation_rules(args.bsf_release) - print(f" {len(agg_rules)} column rules loaded") - - for upgrade in upgrade_ids: - print(f"\n{'=' * 60}") - print(f"Processing state={args.state}, upgrade={upgrade}") - print(f"{'=' * 60}") - process_upgrade( - path_input, path_output, args.state, upgrade, agg_rules, args.workers - ) - - print("\nAll done.") - - -if __name__ == "__main__": - main() diff --git a/data/resstock/load_curve/aggregate_loads.py b/data/resstock/load_curve/aggregate_loads.py new file mode 100644 index 00000000..7bb353cb --- /dev/null +++ b/data/resstock/load_curve/aggregate_loads.py @@ -0,0 +1,552 @@ +"""Aggregate hourly ResStock load curves into coarser time resolutions. + +Reads each per-building hourly parquet **once** in a ThreadPoolExecutor, then +conditionally performs one or both of: + +- **Monthly**: group by month using the full bsf column-level rules (sum for + energy / emissions / energy_delivered, mean for temperatures); writes one + monthly parquet per building. +- **Annual**: sum only the ``*.energy_consumption`` (not intensity) and + ``out.load.*.energy_delivered.kbtu`` columns into a single row per building; + after all buildings finish, concatenate and join onto slim raw-NREL annual + params (``bldg_id``, ``upgrade``, ``weight``, ``out.params.*``, + ``upgrade_name``); write one consolidated annual parquet per upgrade. + +Which aggregations are performed is controlled by the ``add_monthly`` and +``add_annual`` flags in :func:`process_upgrade`. The expensive hourly read +happens only once regardless of which outputs are enabled. + +Annual column selection from hourly (bsf aggregation rules, subset only): + - ``*.energy_consumption`` (not intensity) -> sum -> rename to ``.kwh`` + - ``out.load.*.energy_delivered.kbtu`` -> sum (same names) + +From raw NREL annual: ``bldg_id``, ``upgrade``, ``weight``, ``out.params.*``, +and ``upgrade_name`` when present. All ``*.savings``, emissions reductions, +peaks, bills, etc. are dropped. + +Usage (from project root):: + + uv run python data/resstock/load_curve/aggregate_loads.py \\ + --path-input /ebs/data/nrel/resstock/res_2024_amy2018_2_sb \\ + --path-output /ebs/data/nrel/resstock/res_2024_amy2018_2_sb \\ + --path-annual-raw /ebs/data/nrel/resstock/res_2024_amy2018_2 \\ + --state NY --upgrade-ids "00 02" \\ + --bsf-release res_2024_amy2018_2 --workers 50 \\ + --add-monthly --add-annual +""" + +from __future__ import annotations + +import argparse +import sys +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +from collections.abc import Mapping, Sequence +from pathlib import Path +from typing import cast + +import polars as pl +from buildstock_fetch.constants import LOAD_CURVE_COLUMN_AGGREGATION + +# ── Constants ──────────────────────────────────────────────────────────────── + +_HOURLY_ENERGY_SUFFIX = ".energy_consumption" +_INTENSITY_SUFFIX = ".energy_consumption_intensity" + +LOAD_HEATING = "out.load.heating.energy_delivered.kbtu" +LOAD_COOLING = "out.load.cooling.energy_delivered.kbtu" +LOAD_HOT_WATER = "out.load.hot_water.energy_delivered.kbtu" + +LOAD_ENERGY_DELIVERED_COLS: tuple[str, ...] = ( + LOAD_HEATING, + LOAD_COOLING, + LOAD_HOT_WATER, +) + +# ── BSF rule loading ───────────────────────────────────────────────────────── + + +def load_bsf_aggregation_map(release: str) -> dict[str, str]: + """Load bsf column -> Aggregate_function map for a release key. + + Args: + release: e.g. ``res_2024_amy2018_2`` (matches CSV stem under + ``buildstock_fetch/data/load_curve_column_map/``). + """ + csv_path = LOAD_CURVE_COLUMN_AGGREGATION / f"{release}.csv" + if not csv_path.exists(): + raise FileNotFoundError( + f"bsf column map not found: {csv_path}. " + f"Available: {[p.stem for p in LOAD_CURVE_COLUMN_AGGREGATION.glob('*.csv')]}" + ) + rules_df = pl.read_csv(csv_path) + return dict( + zip( + rules_df["name"].to_list(), + rules_df["Aggregate_function"].to_list(), + strict=True, + ) + ) + + +# ── Annual helpers ─────────────────────────────────────────────────────────── + + +def is_annual_metric_column(name: str) -> bool: + """True if this hourly column should be rolled into ``_sb`` annual. + + Includes plain ``*.energy_consumption`` and any ``energy_delivered`` column; + excludes intensities and everything else (temps, emissions, etc.). + """ + if name.endswith(_INTENSITY_SUFFIX): + return False + if name.endswith(_HOURLY_ENERGY_SUFFIX): + return True + return "energy_delivered" in name + + +def hourly_energy_col_to_annual(hourly_col: str) -> str | None: + """Map an hourly energy column name to its annual ``.kwh`` counterpart. + + Returns ``None`` if *hourly_col* is not a plain energy-consumption column + (e.g. intensity columns are excluded). + """ + if hourly_col.endswith(_INTENSITY_SUFFIX): + return None + if not hourly_col.endswith(_HOURLY_ENERGY_SUFFIX): + return None + return hourly_col + ".kwh" + + +def annual_column_name(hourly_col: str) -> str: + """Annual output name for a metric column (``.kwh`` suffix for energy).""" + renamed = hourly_energy_col_to_annual(hourly_col) + return renamed if renamed is not None else hourly_col + + +def build_annual_agg_exprs( + rules: Mapping[str, str], + schema_names: Sequence[str], +) -> list[pl.Expr]: + """Build annual aggregation exprs from a subset of bsf rules. + + Only columns in both *rules* and *schema_names* that pass + :func:`is_annual_metric_column` are included. Aggregation function comes + from the CSV (``sum`` for energy and delivered under bsf >= 1.6.6). + """ + schema_set = set(schema_names) + if "bldg_id" not in schema_set: + raise ValueError("hourly schema must contain 'bldg_id'") + + exprs: list[pl.Expr] = [pl.col("bldg_id").first().alias("bldg_id")] + n_metrics = 0 + for col, agg in rules.items(): + if col not in schema_set or not is_annual_metric_column(col): + continue + out_name = annual_column_name(col) + match agg: + case "sum": + exprs.append(pl.col(col).sum().alias(out_name)) + case "mean": + exprs.append(pl.col(col).mean().alias(out_name)) + case "first": + exprs.append(pl.col(col).first().alias(out_name)) + case _: + raise ValueError( + f"Unknown aggregation function '{agg}' for column '{col}'" + ) + n_metrics += 1 + + if n_metrics == 0: + raise ValueError( + "no annual metric columns found in hourly schema " + "(expected *.energy_consumption and/or energy_delivered)" + ) + return exprs + + +def aggregate_hourly_df_to_annual_row( + hourly_df: pl.DataFrame, + rules: Mapping[str, str], +) -> pl.DataFrame: + """Aggregate an in-memory hourly building DataFrame to one annual row.""" + exprs = build_annual_agg_exprs(rules, hourly_df.columns) + return hourly_df.select(exprs) + + +def select_annual_params_weight_upgrade(annual_lf: pl.LazyFrame) -> pl.LazyFrame: + """Keep identity/params columns from raw annual; drop metrics rebuilt from hourly. + + Always keeps ``bldg_id``, ``upgrade``, ``weight``, and all ``out.params.*``. + Also keeps ``upgrade_name`` when present (NREL upgrades 01-05 only). + """ + schema_names = annual_lf.collect_schema().names() + required = ("bldg_id", "upgrade", "weight") + missing = [c for c in required if c not in schema_names] + if missing: + raise ValueError( + f"load_curve_annual missing required columns {missing}; " + f"found: {schema_names[:20]}..." + ) + + keep: list[str] = ["bldg_id", "upgrade", "weight"] + if "upgrade_name" in schema_names: + keep.append("upgrade_name") + keep.extend(c for c in schema_names if c.startswith("out.params.")) + return annual_lf.select(keep) + + +def join_aggregated_energy_to_annual( + aggregated_energy_lf: pl.LazyFrame, + annual_params_lf: pl.LazyFrame, +) -> pl.LazyFrame: + """Left-join aggregated ``_sb`` metrics onto slim raw annual params/weight/upgrade.""" + return aggregated_energy_lf.join(annual_params_lf, on="bldg_id", how="left") + + +def write_consolidated_annual( + annual_rows: list[pl.DataFrame], + path_annual_raw: Path, + path_output: Path, + state: str, + upgrade: str, +) -> Path | None: + """Concat annual rows, join raw params, write one ResStock-style annual parquet. + + Returns the output path, or ``None`` if there is nothing to write. + """ + if not annual_rows: + return None + + annual_raw_dir = ( + path_annual_raw / f"load_curve_annual/state={state}/upgrade={upgrade}" + ) + if not annual_raw_dir.exists(): + raise FileNotFoundError( + f"Raw annual directory does not exist: {annual_raw_dir}" + ) + annual_raw_files = sorted(annual_raw_dir.glob("*.parquet")) + if not annual_raw_files: + raise FileNotFoundError(f"No annual parquet files in {annual_raw_dir}") + + output_dir = path_output / f"load_curve_annual/state={state}/upgrade={upgrade}" + output_dir.mkdir(parents=True, exist_ok=True) + + aggregated = cast(pl.DataFrame, pl.concat(annual_rows, how="vertical_relaxed")) + annual_params_lf = select_annual_params_weight_upgrade( + pl.scan_parquet(str(annual_raw_dir)) + ) + joined = cast( + pl.DataFrame, + join_aggregated_energy_to_annual(aggregated.lazy(), annual_params_lf).collect(), + ) + + if len(annual_raw_files) == 1: + out_name = annual_raw_files[0].name + else: + out_name = f"{state}_upgrade{upgrade}_metadata_and_annual_results.parquet" + + out_path = output_dir / out_name + joined.write_parquet(out_path) + return out_path + + +# ── Monthly helpers ────────────────────────────────────────────────────────── + + +def monthly_aggregation_exprs(rules: Mapping[str, str]) -> list[pl.Expr]: + """Build polars monthly aggregation expressions from a bsf rule map.""" + exprs: list[pl.Expr] = [] + for col, agg in rules.items(): + if col == "timestamp": + continue + match agg: + case "sum": + exprs.append(pl.col(col).sum()) + case "mean": + exprs.append(pl.col(col).mean()) + case "first": + exprs.append(pl.col(col).first()) + case _: + raise ValueError( + f"Unknown aggregation function '{agg}' for column '{col}'" + ) + return exprs + + +def _write_monthly_from_hourly_df( + hourly_df: pl.DataFrame, + output_path: Path, + monthly_exprs: Sequence[pl.Expr], +) -> None: + """Aggregate an in-memory hourly DataFrame to monthly and write parquet.""" + output_path.parent.mkdir(parents=True, exist_ok=True) + + # bsf's column map covers out.* and bldg_id but not year/day/hour/timestamp + # (those are added by bsf after aggregation). We add year.first() here and + # reconstruct timestamp from the result. + all_rules = [*monthly_exprs, pl.col("year").first()] + + df = hourly_df.group_by("month").agg(all_rules).sort("month") + + year = df["year"][0] + + df = df.with_columns( + pl.datetime(year=year, month=pl.col("month"), day=1).alias("timestamp"), + pl.col("year").cast(pl.Int32), + pl.col("month").cast(pl.Int8), + ) + + out_cols = [c for c in df.columns if c.startswith("out.")] + col_order = ["timestamp", *out_cols, "bldg_id", "year", "month"] + df.select(col_order).write_parquet(output_path) + + +# ── Combined processing ────────────────────────────────────────────────────── + + +def aggregate_one_building( + input_path: Path, + *, + add_monthly: bool, + monthly_output_path: Path | None, + monthly_exprs: Sequence[pl.Expr] | None, + add_annual: bool, + annual_rules: Mapping[str, str] | None, +) -> pl.DataFrame | None: + """Read one hourly parquet once; conditionally write monthly and/or return annual row.""" + hourly_df = pl.read_parquet(input_path) + if add_monthly: + assert monthly_output_path is not None + assert monthly_exprs is not None + _write_monthly_from_hourly_df(hourly_df, monthly_output_path, monthly_exprs) + if add_annual: + assert annual_rules is not None + return aggregate_hourly_df_to_annual_row(hourly_df, annual_rules) + return None + + +def process_upgrade( + path_input: Path, + path_output: Path, + state: str, + upgrade: str, + rules: Mapping[str, str], + workers: int, + *, + add_monthly: bool = True, + add_annual: bool = False, + path_annual_raw: Path | None = None, +) -> None: + """Process all building files for one upgrade. + + Reads each hourly parquet once, then conditionally: + - Writes monthly aggregation (when *add_monthly* is True) + - Accumulates annual rows and writes a consolidated annual parquet + (when *add_annual* is True; requires *path_annual_raw*) + """ + if not add_monthly and not add_annual: + raise ValueError("At least one of add_monthly or add_annual must be True") + if add_annual and path_annual_raw is None: + raise ValueError("path_annual_raw is required when add_annual=True") + + input_dir = path_input / f"load_curve_hourly/state={state}/upgrade={upgrade}" + monthly_dir = path_output / f"load_curve_monthly/state={state}/upgrade={upgrade}" + + if not input_dir.exists(): + print(f" Input directory does not exist, skipping: {input_dir}") + return + + files = sorted(input_dir.glob("*.parquet")) + n_files = len(files) + if n_files == 0: + print(f" No parquet files found in {input_dir}") + return + + monthly_exprs = monthly_aggregation_exprs(rules) if add_monthly else None + + outputs: list[str] = [] + if add_monthly: + outputs.append(f"monthly -> {monthly_dir}") + if add_annual: + annual_out_dir = ( + path_output / f"load_curve_annual/state={state}/upgrade={upgrade}" + ) + outputs.append(f"annual -> {annual_out_dir}") + + print(f" Found {n_files:,} hourly files in {input_dir}") + for o in outputs: + print(f" {o}") + + if add_monthly: + monthly_dir.mkdir(parents=True, exist_ok=True) + + t0 = time.time() + annual_frames: list[pl.DataFrame] = [] + done = 0 + errors = 0 + + def _process(src: Path) -> pl.DataFrame | str | None: + try: + return aggregate_one_building( + src, + add_monthly=add_monthly, + monthly_output_path=monthly_dir / src.name if add_monthly else None, + monthly_exprs=monthly_exprs, + add_annual=add_annual, + annual_rules=rules if add_annual else None, + ) + except Exception as e: + return f"{src.name}: {e}" + + with ThreadPoolExecutor(max_workers=workers) as pool: + futures = {pool.submit(_process, f): f for f in files} + for future in as_completed(futures): + result = future.result() + done += 1 + if isinstance(result, str): + errors += 1 + print(f" ERROR {result}") + elif isinstance(result, pl.DataFrame): + annual_frames.append(result) + if done % 5000 == 0 or done == n_files: + elapsed = time.time() - t0 + rate = done / elapsed if elapsed > 0 else 0 + print( + f" {done:,}/{n_files:,} ({rate:.0f} files/s, {elapsed:.1f}s elapsed)" + ) + + if add_annual: + assert path_annual_raw is not None + out_path = write_consolidated_annual( + annual_frames, path_annual_raw, path_output, state, upgrade + ) + if out_path is not None: + print(f" Wrote annual: {out_path} ({len(annual_frames):,} buildings)") + else: + print(f" WARNING: no annual rows to write for upgrade={upgrade}") + + elapsed = time.time() - t0 + print(f" Done: {done:,} files in {elapsed:.1f}s ({errors} errors)") + + +# ── CLI ────────────────────────────────────────────────────────────────────── + + +def main() -> None: + parser = argparse.ArgumentParser( + description=( + "Aggregate hourly ResStock load curves to coarser time resolutions " + "(monthly and/or annual). Use --add-monthly and/or --add-annual." + ) + ) + parser.add_argument( + "--path-input", + required=True, + help="Root of the ResStock release (local), e.g. .../res_2024_amy2018_2_sb", + ) + parser.add_argument( + "--path-output", + required=True, + help="Root to write monthly/annual load curves into (local)", + ) + parser.add_argument( + "--path-annual-raw", + default=None, + help=( + "Root of the raw NREL release with load_curve_annual. Required when " + "--add-annual is set (provides params/weight/upgrade_name)." + ), + ) + parser.add_argument("--state", required=True, help="Two-letter state code, e.g. NY") + parser.add_argument( + "--upgrade-ids", + required=True, + help='Space-separated upgrade IDs, e.g. "00 02"', + ) + parser.add_argument( + "--bsf-release", + required=True, + help="bsf release key for column aggregation rules, e.g. res_2024_amy2018_2", + ) + parser.add_argument( + "--workers", + type=int, + default=50, + help="Number of parallel workers (default: 50)", + ) + parser.add_argument( + "--add-monthly", + action="store_true", + default=False, + help="Produce monthly load curves (default: False unless neither flag is set).", + ) + parser.add_argument( + "--add-annual", + action="store_true", + default=False, + help="Produce annual load curves (default: False unless neither flag is set).", + ) + args = parser.parse_args() + + add_monthly: bool = args.add_monthly + add_annual: bool = args.add_annual + if not add_monthly and not add_annual: + add_monthly = True + add_annual = True + print("Neither --add-monthly nor --add-annual specified; producing both.") + + path_input = Path(args.path_input) + path_output = Path(args.path_output) + path_annual_raw = ( + Path(args.path_annual_raw) if args.path_annual_raw is not None else None + ) + upgrade_ids = args.upgrade_ids.split() + + if not path_input.exists(): + print(f"Error: input path does not exist: {path_input}", file=sys.stderr) + sys.exit(1) + if add_annual and path_annual_raw is None: + print( + "Error: --path-annual-raw is required when --add-annual is set.", + file=sys.stderr, + ) + sys.exit(1) + if path_annual_raw is not None and not path_annual_raw.exists(): + print( + f"Error: --path-annual-raw does not exist: {path_annual_raw}", + file=sys.stderr, + ) + sys.exit(1) + + print(f"Loading bsf aggregation rules for release '{args.bsf_release}'...") + rules = load_bsf_aggregation_map(args.bsf_release) + annual_n = sum(1 for c in rules if is_annual_metric_column(c)) + print(f" {len(rules)} column rules ({annual_n} used for annual subset)") + print( + f" Producing: {'monthly' if add_monthly else ''}" + f"{'+ ' if add_monthly and add_annual else ''}" + f"{'annual' if add_annual else ''}" + ) + + for upgrade in upgrade_ids: + print(f"\n{'=' * 60}") + print(f"Processing state={args.state}, upgrade={upgrade}") + print(f"{'=' * 60}") + process_upgrade( + path_input=path_input, + path_output=path_output, + state=args.state, + upgrade=upgrade, + rules=rules, + workers=args.workers, + add_monthly=add_monthly, + add_annual=add_annual, + path_annual_raw=path_annual_raw, + ) + + print("\nAll done.") + + +if __name__ == "__main__": + main() diff --git a/data/resstock/main.py b/data/resstock/main.py index 2d7787c5..39c2ab3f 100644 --- a/data/resstock/main.py +++ b/data/resstock/main.py @@ -41,12 +41,11 @@ HEATING_TYPE_COLS, HP_CUSTOMERS_COLS, NATGAS_CONNECTION_COLS, - SB_EXCLUDED_FILE_TYPES, + SB_CLONE_EXCLUDED_FILE_TYPES, VULNERABILITY_COLS, ) from data.resstock.nrel.copy_resstock_data import clone_release -from data.resstock.load_curve.add_monthly_loads import ( - load_aggregation_rules, +from data.resstock.load_curve.aggregate_loads import ( process_upgrade, ) from data.resstock.load_curve.adjust_mf_electricity import ( @@ -91,7 +90,7 @@ validate_metadata_columns, validate_metadata_output, validate_metadata_readable, - validate_no_stale_monthly_loads, + validate_no_stale_aggregate_loads, validate_s3_objects, validate_utility_assignment_args, ) @@ -490,37 +489,52 @@ def _assign_utility( gc.collect() -def _add_monthly_loads( +def _add_aggregate_loads( *, states: list[str], path_sb: Path, + path_raw: Path, upgrade_ids: list[str], release: str, s3_base_sb: str, sample: int, workers: int, + add_monthly: bool, + add_annual: bool, ) -> list[str]: - """Aggregate _sb hourly load curves into monthly load curves and upload them. - - Reads per-building hourly parquets from ``path_sb/load_curve_hourly/`` and - writes one monthly parquet per building to ``path_sb/load_curve_monthly/``. - Aggregation rules (sum vs mean vs first) come from bsf's column-aggregation - CSV for ``release`` (the raw release name, not the _sb variant). + """Aggregate _sb hourly load curves into requested time resolutions and upload. - After each state is processed, the ``load_curve_monthly/state=/`` - directory is synced to S3 via ``aws s3 sync``. + Reads each hourly parquet once per building in a ThreadPoolExecutor, then + conditionally performs: + - Monthly aggregation (when *add_monthly* is True): writes one monthly + parquet per building. + - Annual aggregation (when *add_annual* is True): accumulates one row per + building, then writes one consolidated annual parquet per upgrade (joined + to raw NREL annual params from ``path_raw``). - When ``--sample > 0`` only N hourly files exist locally. The aggregation - proceeds on whatever files are present and N monthly files are written. - This is expected behaviour for development/testing; run without ``--sample`` - for production. + Aggregation rules come from bsf's column-aggregation CSV for ``release``. + After each state is processed, the output directories are synced to S3. - Returns the list of (state, upgrade) pairs that were actually processed, - in ``"state= upgrade="`` format, for manifest recording. + Returns the list of (state, upgrade) pairs processed, as + ``"state= upgrade="``, for manifest recording. """ + from data.resstock.load_curve.aggregate_loads import ( + is_annual_metric_column, + load_bsf_aggregation_map, + ) + + outputs_label = " + ".join( + name + for name, flag in [("monthly", add_monthly), ("annual", add_annual)] + if flag + ) print(f" Loading bsf aggregation rules for release '{release}'...", flush=True) - agg_rules = load_aggregation_rules(release) - print(f" {len(agg_rules)} column rules loaded.", flush=True) + rules = load_bsf_aggregation_map(release) + annual_n = sum(1 for c in rules if is_annual_metric_column(c)) + print( + f" {len(rules)} column rules loaded ({annual_n} used for annual subset).", + flush=True, + ) processed: list[str] = [] @@ -528,8 +542,8 @@ def _add_monthly_loads( if sample > 0: print( f" NOTE: --sample active for state={s}. " - f"Monthly load curves will be generated only for the sampled buildings. " - f"Run without --sample for production.", + f"Aggregated load curves will be generated only for the sampled " + f"buildings. Run without --sample for production.", flush=True, ) @@ -554,7 +568,7 @@ def _add_monthly_loads( continue print( - f" Aggregating {n_files:,} hourly files → monthly for {loc}...", + f" Aggregating {n_files:,} hourly files → {outputs_label} for {loc}...", flush=True, ) process_upgrade( @@ -562,39 +576,45 @@ def _add_monthly_loads( path_output=path_sb, state=s, upgrade=uid, - agg_rules=agg_rules, + rules=rules, workers=workers, + add_monthly=add_monthly, + add_annual=add_annual, + path_annual_raw=path_raw if add_annual else None, ) processed.append(loc) - # Upload the full load_curve_monthly/state=/ tree for this state once - # all upgrades are done. - monthly_state_dir = path_sb / "load_curve_monthly" / f"state={s}" - if monthly_state_dir.exists(): - n_monthly = sum(1 for _ in monthly_state_dir.rglob("*") if _.is_file()) - s3_dest = f"{s3_base_sb.rstrip('/')}/load_curve_monthly/state={s}/" - print( - f" Uploading load_curve_monthly/state={s} " - f"({n_monthly:,} files) → {s3_dest}", - flush=True, - ) - upload_rc = subprocess.run( - ["aws", "s3", "sync", str(monthly_state_dir), s3_dest, "--quiet"], - check=False, - ).returncode - if upload_rc != 0: - raise RuntimeError( - f"aws s3 sync failed (exit {upload_rc}): " - f"load_curve_monthly/state={s}/ → {s3_dest}" + file_types_to_upload: list[str] = [] + if add_monthly: + file_types_to_upload.append("load_curve_monthly") + if add_annual: + file_types_to_upload.append("load_curve_annual") + + for file_type in file_types_to_upload: + state_dir = path_sb / file_type / f"state={s}" + if state_dir.exists(): + n_out = sum(1 for _ in state_dir.rglob("*") if _.is_file()) + s3_dest = f"{s3_base_sb.rstrip('/')}/{file_type}/state={s}/" + print( + f" Uploading {file_type}/state={s} ({n_out:,} files) → {s3_dest}", + flush=True, ) - else: + upload_rc = subprocess.run( + ["aws", "s3", "sync", str(state_dir), s3_dest, "--quiet"], + check=False, + ).returncode + if upload_rc != 0: + raise RuntimeError( + f"aws s3 sync failed (exit {upload_rc}): " + f"{file_type}/state={s}/ → {s3_dest}" + ) print(" Done.", flush=True) - else: - print( - f" WARNING: No monthly output directory found for state={s} — " - f"nothing to upload.", - flush=True, - ) + else: + print( + f" WARNING: No {file_type} output directory for state={s} — " + f"nothing to upload.", + flush=True, + ) return processed @@ -833,7 +853,17 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: default=True, metavar="BOOL", help=( - "Aggregate _sb hourly load curves into monthly load curves and upload to S3 " + "Aggregate _sb hourly load curves into monthly load curves " + "(default: True). Only runs when load_curve_hourly is in --file-types." + ), + ) + parser.add_argument( + "--add-annual-loads", + type=parse_bool, + default=True, + metavar="BOOL", + help=( + "Aggregate _sb hourly load curves into annual load curves " "(default: True). Only runs when load_curve_hourly is in --file-types." ), ) @@ -874,11 +904,11 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: ## Other misc values parser.add_argument( - "--monthly-workers", + "--aggregation-workers", type=int, default=50, metavar="N", - help="Number of parallel workers for monthly aggregation (default: 50).", + help="Number of parallel workers for load curve aggregation (default: 50).", ) return parser.parse_args(argv) @@ -900,7 +930,9 @@ def main(argv: list[str] | None = None) -> None: s3_base_sb = f"{args.path_s3_dir.rstrip('/')}/{release_sb}" # File types that will actually appear in _sb (excludes raw-only types). - sb_file_types = [ft for ft in args.file_types if ft not in SB_EXCLUDED_FILE_TYPES] + sb_file_types = [ + ft for ft in args.file_types if ft not in SB_CLONE_EXCLUDED_FILE_TYPES + ] # ── Manifest: start a run record ────────────────────────────────────────── # Ensure _sb directory and manifest exist before any I/O so that even a @@ -927,7 +959,8 @@ def main(argv: list[str] | None = None) -> None: "adjust_mf_electricity": args.adjust_mf_electricity, "assign_utility": args.assign_utility, "add_monthly_loads": args.add_monthly_loads, - "monthly_workers": args.monthly_workers, + "add_annual_loads": args.add_annual_loads, + "aggregation_workers": args.aggregation_workers, "sample": args.sample, }, ) @@ -957,14 +990,16 @@ def main(argv: list[str] | None = None) -> None: mf_adj_upgrades=_MF_ADJ_UPGRADES, assign_utility=args.assign_utility, add_monthly_loads=args.add_monthly_loads, + add_annual_loads=args.add_annual_loads, ) if run_warnings: run["warnings"] = run_warnings - validate_no_stale_monthly_loads( + validate_no_stale_aggregate_loads( state=args.state, upgrade_ids=args.upgrade_ids, file_types=args.file_types, add_monthly_loads=args.add_monthly_loads, + add_annual_loads=args.add_annual_loads, path_sb=path_sb, ) validate_utility_assignment_args( @@ -1163,25 +1198,39 @@ def main(argv: list[str] | None = None) -> None: upsert_run(path_sb, run) gc.collect() - # ── 2d. Add monthly load curves ──────────────────────────────────────── - if args.add_monthly_loads and "load_curve_hourly" in args.file_types: - print("Adding monthly load curves...", flush=True) + # ── 2d. Aggregate load curves (monthly / annual) ──────────────────── + _wants_aggregation = ( + args.add_monthly_loads or args.add_annual_loads + ) and "load_curve_hourly" in args.file_types + if _wants_aggregation: + agg_label = " + ".join( + name + for name, flag in [ + ("monthly", args.add_monthly_loads), + ("annual", args.add_annual_loads), + ] + if flag + ) + print(f"Aggregating load curves ({agg_label})...", flush=True) try: - processed_monthly = _add_monthly_loads( + processed_agg = _add_aggregate_loads( states=args.state, path_sb=path_sb, + path_raw=path_raw, upgrade_ids=args.upgrade_ids, release=release, s3_base_sb=s3_base_sb, sample=args.sample, - workers=args.monthly_workers, + workers=args.aggregation_workers, + add_monthly=args.add_monthly_loads, + add_annual=args.add_annual_loads, ) - record_step(run, "add_monthly_loads", processed=processed_monthly) + record_step(run, "add_aggregate_loads", processed=processed_agg) upsert_run(path_sb, run) upload_manifest(path_sb, s3_base_sb) except Exception as exc: run.setdefault("warnings", []).append( - f"add_monthly_loads S3 upload failed: {exc}" + f"add_aggregate_loads failed: {exc}" ) upsert_run(path_sb, run) try: diff --git a/data/resstock/manifest.py b/data/resstock/manifest.py index 6fa3f463..d3a44fbf 100644 --- a/data/resstock/manifest.py +++ b/data/resstock/manifest.py @@ -500,6 +500,8 @@ def print_status( "approximate_non_hp_load": frozenset({"load_curve_hourly"}), "adjust_mf_electricity": frozenset({"load_curve_hourly"}), "add_monthly_loads": frozenset({"load_curve_monthly"}), + "add_annual_loads": frozenset({"load_curve_annual"}), + "add_aggregate_loads": frozenset({"load_curve_monthly", "load_curve_annual"}), "upload_raw": frozenset(), # does not create new types "upload_sb": frozenset(), } @@ -590,17 +592,19 @@ def _expected_file_types(runs: list[dict[str, Any]], *, is_sb: bool) -> set[str] **union** across every run that touched this state, not just the latest. For the raw release: union of ``args.file_types`` over all runs. - For the _sb release: the same union minus ``SB_EXCLUDED_FILE_TYPES``, plus + For the _sb release: the same union minus ``SB_CLONE_EXCLUDED_FILE_TYPES``, plus any types produced by completed steps (``metadata_utility``, ``load_curve_monthly``, etc.). """ - from data.resstock.constants import SB_EXCLUDED_FILE_TYPES + from data.resstock.constants import SB_CLONE_EXCLUDED_FILE_TYPES expected: set[str] = set() for run in runs: base_types = set(run.get("args", {}).get("file_types", [])) if is_sb: - expected |= {ft for ft in base_types if ft not in SB_EXCLUDED_FILE_TYPES} + expected |= { + ft for ft in base_types if ft not in SB_CLONE_EXCLUDED_FILE_TYPES + } for step in run.get("steps", []): expected |= set(_STEP_FILE_TYPES.get(step.get("step", ""), frozenset())) else: @@ -612,7 +616,7 @@ def _latest_step_times_by_file_type( runs: list[dict[str, Any]], *, is_sb: bool ) -> dict[str, datetime]: """Map each file type to the newest step ``completed_at`` across all runs.""" - from data.resstock.constants import SB_EXCLUDED_FILE_TYPES + from data.resstock.constants import SB_CLONE_EXCLUDED_FILE_TYPES times: dict[str, datetime] = {} @@ -620,7 +624,9 @@ def _latest_step_times_by_file_type( args = run.get("args", {}) base_types = set(args.get("file_types", [])) if is_sb: - base_types = {ft for ft in base_types if ft not in SB_EXCLUDED_FILE_TYPES} + base_types = { + ft for ft in base_types if ft not in SB_CLONE_EXCLUDED_FILE_TYPES + } for step in run.get("steps", []): name = step.get("step", "") @@ -686,7 +692,7 @@ def check_integrity( 1. **Expected file types** are the **union** across *all* matching manifest runs (not just the latest). Pipeline runs are additive: a later run that only fetches load curves does not erase metadata from - an earlier run. For ``_sb``, ``SB_EXCLUDED_FILE_TYPES`` are dropped + an earlier run. For ``_sb``, ``SB_CLONE_EXCLUDED_FILE_TYPES`` are dropped and step-produced types (``metadata_utility``, ``load_curve_monthly``) are included. 2. **Actual file types** are the top-level directories under the release that diff --git a/data/resstock/validations.py b/data/resstock/validations.py index bda566c5..dca3960c 100644 --- a/data/resstock/validations.py +++ b/data/resstock/validations.py @@ -178,78 +178,75 @@ def validate_utility_assignment_args( ) -def validate_no_stale_monthly_loads( +def validate_no_stale_aggregate_loads( state: list[str], upgrade_ids: list[str], file_types: list[str], add_monthly_loads: bool, + add_annual_loads: bool, path_sb: Path, ) -> None: - """Raise RuntimeError if fetching hourly load curves would leave monthly files stale. + """Raise RuntimeError if fetching hourly would leave monthly/annual files stale. - Monthly load curve files in the _sb release are derived directly from the _sb - hourly load curves (by ``--add-monthly-loads``). If hourly files are re-fetched - without also updating the corresponding monthly files, the two datasets become - inconsistent — the monthly files would still be aggregated from the old hourly data. + ``load_curve_monthly`` and ``_sb`` ``load_curve_annual`` are derived from + ``_sb`` hourly by the aggregation step. If hourly is re-fetched (and then + modified by non-HP approx / MF adj), any existing aggregate outputs that + are NOT being regenerated would become inconsistent. - This check fires when all of the following are true: + Note: having ``load_curve_monthly`` in ``--file-types`` does NOT help — + that clones raw NREL monthly into ``_sb``, which still disagrees with + modified ``_sb`` hourly. The only fix is enabling the aggregation flag. - - ``load_curve_hourly`` is in ``file_types`` (hourly will be overwritten) - - ``load_curve_monthly`` is NOT in ``file_types`` (raw monthly not being re-fetched) - - ``add_monthly_loads`` is False (pipeline will not regenerate monthly from hourly) - - Existing ``load_curve_monthly`` parquet files are found in the _sb release for - at least one (state, upgrade) combination in the current request + Fires when: - The fix is to either pass ``--add-monthly-loads True`` (recommended — regenerates - monthly from the freshly modified _sb hourly) or add ``load_curve_monthly`` to - ``--file-types`` so raw monthly files are re-fetched alongside hourly. + - ``load_curve_hourly`` is in ``file_types`` (hourly will be refreshed) + - Existing monthly or annual parquets are found for a requested + (state, upgrade) whose corresponding aggregation flag is NOT enabled """ if "load_curve_hourly" not in file_types: return - if "load_curve_monthly" in file_types: - return - if add_monthly_loads: + + checks: list[str] = [] + if not add_monthly_loads: + checks.append("load_curve_monthly") + if not add_annual_loads: + checks.append("load_curve_annual") + + if not checks: return stale: list[str] = [] for s in state: for uid in upgrade_ids: upgrade_id_padded = uid.zfill(2) - monthly_dir = ( - path_sb - / "load_curve_monthly" - / f"state={s}" - / f"upgrade={upgrade_id_padded}" - ) - if monthly_dir.is_dir(): - existing = list(monthly_dir.glob("*.parquet")) - if existing: - stale.append( - f" load_curve_monthly/state={s}/upgrade={upgrade_id_padded}:" - f" {len(existing):,} file(s) at {monthly_dir}" - ) + for file_type in checks: + out_dir = ( + path_sb / file_type / f"state={s}" / f"upgrade={upgrade_id_padded}" + ) + if out_dir.is_dir(): + existing = list(out_dir.glob("*.parquet")) + if existing: + stale.append( + f" {file_type}/state={s}/upgrade={upgrade_id_padded}:" + f" {len(existing):,} file(s) at {out_dir}" + ) if not stale: return stale_list = "\n".join(stale) raise RuntimeError( - f"Stale load_curve_monthly conflict detected.\n" + f"Stale aggregate load curve conflict detected.\n" f"\n" - f"You requested load_curve_hourly without updating load_curve_monthly. " - f"The following existing monthly files in the _sb release were built from the " - f"current hourly data and would become inconsistent with the new hourly files:\n" + f"You requested load_curve_hourly without regenerating all existing " + f"aggregate outputs. The following files in the _sb release would become " + f"inconsistent with the new hourly files:\n" f"\n" f"{stale_list}\n" f"\n" - f"Monthly files in the _sb release are derived directly from _sb hourly files " - f"and must be kept in sync. Resubmit with ONE of the following:\n" - f"\n" - f" Option 1 (recommended): enable --add-monthly-loads True so the pipeline " - f"regenerates monthly files from the freshly modified _sb hourly data:\n" - f" --add-monthly-loads True\n" + f"Monthly and annual files in the _sb release are derived from modified " + f"_sb hourly and must be kept in sync. Enable the corresponding " + f"aggregation flags:\n" f"\n" - f" Option 2: add load_curve_monthly to --file-types so monthly files are " - f"re-fetched from the raw NREL release alongside hourly:\n" - f" --file-types ... load_curve_hourly load_curve_monthly ...\n" + f" --add-monthly-loads True --add-annual-loads True\n" ) diff --git a/data/resstock/warnings.py b/data/resstock/warnings.py index 7a8a2d8c..a0f7a1e6 100644 --- a/data/resstock/warnings.py +++ b/data/resstock/warnings.py @@ -8,7 +8,7 @@ from __future__ import annotations -from data.resstock.constants import SB_EXCLUDED_FILE_TYPES +from data.resstock.constants import SB_CLONE_EXCLUDED_FILE_TYPES def collect_run_warnings( @@ -21,6 +21,7 @@ def collect_run_warnings( mf_adj_upgrades: list[str], assign_utility: bool, add_monthly_loads: bool, + add_annual_loads: bool = True, ) -> list[str]: """Run all pre-run argument/file-type mismatch checks. @@ -45,6 +46,8 @@ def collect_run_warnings( Whether the utility assignment step is enabled. add_monthly_loads: Whether the monthly load aggregation step is enabled. + add_annual_loads: + Whether the annual load aggregation step is enabled. """ warnings: list[str] = [] @@ -52,13 +55,13 @@ def _warn(msg: str) -> None: print(f"WARNING: {msg}", flush=True) warnings.append(msg) - for ft in SB_EXCLUDED_FILE_TYPES: + for ft in SB_CLONE_EXCLUDED_FILE_TYPES: if ft in file_types: _warn( f"'{ft}' will be fetched for the raw release but is NOT copied to " - f"the _sb release. The _sb release has no post-modification annual " - f"equivalent; use load_curve_monthly (derived from load_curve_hourly) " - f"for month-level aggregations of the modified _sb data." + f"the _sb release via clone. When --add-annual-loads is enabled, " + f"_sb load_curve_annual is derived from modified _sb hourly; " + f"when --add-monthly-loads is enabled, _sb load_curve_monthly is too." ) if approximate_non_hp_load and "load_curve_hourly" not in file_types: @@ -107,4 +110,11 @@ def _warn(msg: str) -> None: "'load_curve_hourly' to --file-types if you want monthly load curves generated." ) + if add_annual_loads and "load_curve_hourly" not in file_types: + _warn( + "--add-annual-loads is enabled but 'load_curve_hourly' is not in " + "--file-types. The annual aggregation step will be skipped. Add " + "'load_curve_hourly' to --file-types if you want annual load curves generated." + ) + return warnings diff --git a/pyproject.toml b/pyproject.toml index 8b1a4ba3..34764a91 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,7 +18,7 @@ classifiers = [ "Topic :: Office/Business :: Financial", ] dependencies = [ - "buildstock-fetch>=1.6.1", + "buildstock-fetch>=1.6.6", "cairo", "google-auth>=2.0.0", "gspread>=6.0.0", diff --git a/rate_design/hp_rates/Justfile b/rate_design/hp_rates/Justfile index 687b793e..e97ec2cb 100644 --- a/rate_design/hp_rates/Justfile +++ b/rate_design/hp_rates/Justfile @@ -89,8 +89,6 @@ path_supply_energy_mc := env_var_or_default('SUPPLY_ENERGY_MC', "s3://data.sb/sw path_supply_capacity_mc := env_var_or_default('SUPPLY_CAPACITY_MC', "s3://data.sb/switchbox/marginal_costs/" + state + "/supply/capacity/utility=" + utility + "/year=" + mc_year + "/data.parquet") path_supply_ancillary_mc := env_var_or_default('SUPPLY_ANCILLARY_MC', "") path_resstock_release := "/ebs/data/nrel/resstock/res_2024_amy2018_2_sb" -# Raw NREL release (no _sb): full-population load_curve_annual lives here. -path_resstock_raw := "/ebs/data/nrel/resstock/res_2024_amy2018_2" path_resstock_metadata := path_resstock_release + "/metadata" path_utility_assignment := path_resstock_release + "/metadata_utility" path_resstock_loads_00 := path_resstock_release + "/load_curve_hourly/state=" + state_upper + "/upgrade=" + upgrade @@ -355,7 +353,7 @@ create-gas-tariff-map electric_utility upgrade_id output_dir: "{{ electric_utility }}" \ "{{ upgrade_id }}" \ "{{ output_dir }}" \ - "{{ path_resstock_raw }}/load_curve_annual/state={{ state_upper }}/upgrade={{ upgrade_id }}" + "{{ path_resstock_release }}/load_curve_annual/state={{ state_upper }}/upgrade={{ upgrade_id }}" create-gas-tariff-maps-all: just create-gas-tariff-map {{ utility }} "00" {{ path_tariff_maps }}/gas diff --git a/rate_design/hp_rates/md/Justfile b/rate_design/hp_rates/md/Justfile index 8fd44e76..91b99ae1 100644 --- a/rate_design/hp_rates/md/Justfile +++ b/rate_design/hp_rates/md/Justfile @@ -117,8 +117,9 @@ split-chesapeake-gas-tariffs: # munis is dual-fuel and needs its own map file. somerset_rec / berlin_muni / # hagerstown_muni stay outside this list (electric-only for our purposes). # -# Chesapeake-territory RES-1/RES-2 split reads load_curve_annual from the raw -# NREL release (passed automatically by the shared create-gas-tariff-map recipe). +# Chesapeake-territory RES-1/RES-2 split reads generated load_curve_annual from +# the _sb release so annual therms reflect Switchbox's hourly load modifications. +# The path is passed automatically by the shared create-gas-tariff-map recipe. # # Shortcuts (upgrade 00 only — quick per-utility run): # just -f md/Justfile map-gas-bge diff --git a/tests/test_aggregate_loads.py b/tests/test_aggregate_loads.py new file mode 100644 index 00000000..58f25c8a --- /dev/null +++ b/tests/test_aggregate_loads.py @@ -0,0 +1,216 @@ +"""Tests for _sb annual / monthly load aggregation helpers.""" + +from __future__ import annotations + +from pathlib import Path + +import polars as pl + +from data.resstock.load_curve.aggregate_loads import ( + aggregate_hourly_df_to_annual_row, + aggregate_one_building, + annual_column_name, + build_annual_agg_exprs, + is_annual_metric_column, + join_aggregated_energy_to_annual, + monthly_aggregation_exprs, + select_annual_params_weight_upgrade, + write_consolidated_annual, +) + + +def test_is_annual_metric_column() -> None: + assert is_annual_metric_column("out.electricity.heating.energy_consumption") + assert is_annual_metric_column("out.load.heating.energy_delivered.kbtu") + assert not is_annual_metric_column( + "out.electricity.heating.energy_consumption_intensity" + ) + assert not is_annual_metric_column("out.outdoor_air_dryblub_temp.c") + assert not is_annual_metric_column("bldg_id") + + +def test_annual_column_name_adds_kwh_suffix() -> None: + assert ( + annual_column_name("out.electricity.heating.energy_consumption") + == "out.electricity.heating.energy_consumption.kwh" + ) + assert ( + annual_column_name("out.load.heating.energy_delivered.kbtu") + == "out.load.heating.energy_delivered.kbtu" + ) + + +def test_build_annual_agg_exprs_subset_only() -> None: + rules = { + "out.electricity.heating.energy_consumption": "sum", + "out.electricity.heating.energy_consumption_intensity": "sum", + "out.load.heating.energy_delivered.kbtu": "sum", + "out.outdoor_air_dryblub_temp.c": "mean", + "bldg_id": "first", + } + schema = list(rules) + exprs = build_annual_agg_exprs(rules, schema) + # bldg_id + heating energy + heating delivered (not intensity, not temp) + aliases = [] + for e in exprs: + meta = e.meta + # polars Expr alias via meta.output_name() + aliases.append(meta.output_name()) + assert aliases == [ + "bldg_id", + "out.electricity.heating.energy_consumption.kwh", + "out.load.heating.energy_delivered.kbtu", + ] + + +def test_aggregate_hourly_df_to_annual_row_sums() -> None: + rules = { + "out.electricity.heating.energy_consumption": "sum", + "out.load.heating.energy_delivered.kbtu": "sum", + "bldg_id": "first", + } + hourly = pl.DataFrame( + { + "bldg_id": [1, 1, 1, 1], + "out.electricity.heating.energy_consumption": [1.0, 2.0, 3.0, 4.0], + "out.load.heating.energy_delivered.kbtu": [10.0, 20.0, 30.0, 40.0], + "out.electricity.heating.energy_consumption_intensity": [ + 9.0, + 9.0, + 9.0, + 9.0, + ], + } + ) + annual = aggregate_hourly_df_to_annual_row(hourly, rules) + assert annual.height == 1 + assert annual["bldg_id"][0] == 1 + assert annual["out.electricity.heating.energy_consumption.kwh"][0] == 10.0 + assert annual["out.load.heating.energy_delivered.kbtu"][0] == 100.0 + assert "out.electricity.heating.energy_consumption_intensity" not in annual.columns + + +def test_select_annual_params_keeps_upgrade_name() -> None: + annual = pl.DataFrame( + { + "bldg_id": [1], + "upgrade": [2], + "weight": [100.0], + "upgrade_name": ["HP package"], + "out.params.window_area_ft_2": [10.0], + "out.electricity.heating.energy_consumption.kwh": [999.0], + "out.electricity.heating.energy_consumption.kwh.savings": [1.0], + } + ) + slim = select_annual_params_weight_upgrade(annual.lazy()).collect() + assert isinstance(slim, pl.DataFrame) + assert set(slim.columns) == { + "bldg_id", + "upgrade", + "weight", + "upgrade_name", + "out.params.window_area_ft_2", + } + + +def test_aggregate_one_building_monthly_and_annual(tmp_path: Path) -> None: + rules = { + "out.electricity.heating.energy_consumption": "sum", + "out.load.heating.energy_delivered.kbtu": "sum", + "out.outdoor_air_dryblub_temp.c": "mean", + "bldg_id": "first", + } + hourly = pl.DataFrame( + { + "bldg_id": [42] * 4, + "year": [2018] * 4, + "month": [1, 1, 2, 2], + "out.electricity.heating.energy_consumption": [1.0, 1.0, 2.0, 2.0], + "out.load.heating.energy_delivered.kbtu": [4.0, 4.0, 6.0, 6.0], + "out.outdoor_air_dryblub_temp.c": [0.0, 2.0, 4.0, 6.0], + } + ) + src = tmp_path / "42-0.parquet" + hourly.write_parquet(src) + monthly_out = tmp_path / "monthly" / "42-0.parquet" + monthly_exprs = monthly_aggregation_exprs(rules) + + annual_row = aggregate_one_building( + src, + add_monthly=True, + monthly_output_path=monthly_out, + monthly_exprs=monthly_exprs, + add_annual=True, + annual_rules=rules, + ) + + assert monthly_out.exists() + monthly = pl.read_parquet(monthly_out) + assert monthly.height == 2 + assert set(monthly["month"].to_list()) == {1, 2} + # January heating sum + jan = monthly.filter(pl.col("month") == 1) + assert jan["out.electricity.heating.energy_consumption"][0] == 2.0 + assert jan["out.load.heating.energy_delivered.kbtu"][0] == 8.0 + assert jan["out.outdoor_air_dryblub_temp.c"][0] == 1.0 # mean of 0 and 2 + + assert annual_row is not None + assert annual_row.height == 1 + assert annual_row["out.electricity.heating.energy_consumption.kwh"][0] == 6.0 + assert annual_row["out.load.heating.energy_delivered.kbtu"][0] == 20.0 + assert "out.outdoor_air_dryblub_temp.c" not in annual_row.columns + + +def test_write_consolidated_annual(tmp_path: Path) -> None: + raw_dir = tmp_path / "raw" / "load_curve_annual" / "state=CT" / "upgrade=00" + raw_dir.mkdir(parents=True) + pl.DataFrame( + { + "bldg_id": [1, 2], + "upgrade": [0, 0], + "weight": [10.0, 20.0], + "out.params.window_area_ft_2": [1.0, 2.0], + "out.electricity.heating.energy_consumption.kwh": [999.0, 999.0], + } + ).write_parquet(raw_dir / "CT_upgrade00_metadata_and_annual_results.parquet") + + rows = [ + pl.DataFrame( + { + "bldg_id": [1], + "out.electricity.heating.energy_consumption.kwh": [100.0], + "out.load.heating.energy_delivered.kbtu": [50.0], + } + ), + pl.DataFrame( + { + "bldg_id": [2], + "out.electricity.heating.energy_consumption.kwh": [200.0], + "out.load.heating.energy_delivered.kbtu": [75.0], + } + ), + ] + out_root = tmp_path / "sb" + out_path = write_consolidated_annual(rows, tmp_path / "raw", out_root, "CT", "00") + assert out_path is not None + result = pl.read_parquet(out_path) + assert result.height == 2 + assert "out.params.window_area_ft_2" in result.columns + assert "weight" in result.columns + # Rebuilt energy, not the raw 999 placeholder + by_id = {r["bldg_id"]: r for r in result.to_dicts()} + assert by_id[1]["out.electricity.heating.energy_consumption.kwh"] == 100.0 + assert by_id[2]["out.load.heating.energy_delivered.kbtu"] == 75.0 + + +def test_join_left_keeps_energy_when_params_missing() -> None: + energy = pl.DataFrame( + {"bldg_id": [1], "out.electricity.heating.energy_consumption.kwh": [5.0]} + ).lazy() + params = pl.DataFrame({"bldg_id": [2], "upgrade": [0], "weight": [1.0]}).lazy() + joined = join_aggregated_energy_to_annual(energy, params).collect() + assert isinstance(joined, pl.DataFrame) + assert joined.height == 1 + assert joined["bldg_id"][0] == 1 + assert joined["out.electricity.heating.energy_consumption.kwh"][0] == 5.0 + assert joined["weight"][0] is None diff --git a/tests/test_gas_tariff_mapper.py b/tests/test_gas_tariff_mapper.py index 84b5ac2e..9e0b8e1e 100644 --- a/tests/test_gas_tariff_mapper.py +++ b/tests/test_gas_tariff_mapper.py @@ -8,7 +8,11 @@ import pytest from utils.pre.fetch_gas_tariffs_rateacuity import load_config -from utils.pre.gas_tariff_mapper import EXCLUDED_GAS_UTILITIES, map_gas_tariff +from utils.pre.gas_tariff_mapper import ( + EXCLUDED_GAS_UTILITIES, + _default_path_load_curve_annual, + map_gas_tariff, +) def test_excluded_gas_utilities_compiled_from_state_configs(): @@ -494,6 +498,37 @@ def test_load_annual_gas_therms_converts_kwh(tmp_path: Path): assert abs(df.filter(pl.col("bldg_id") == 2)["annual_gas_therms"][0] - 300.0) < 1e-9 +def test_default_annual_load_path_uses_sb_release(tmp_path: Path) -> None: + """The default annual loads come from the same _sb release as metadata.""" + release_sb = tmp_path / "res_2024_amy2018_2_sb" + annual_dir = release_sb / "load_curve_annual" / "state=MD" / "upgrade=02" + annual_dir.mkdir(parents=True) + + result = _default_path_load_curve_annual(release_sb / "metadata", "MD", "02") + + assert result == annual_dir + + +def test_default_annual_load_path_does_not_fall_back_to_raw( + tmp_path: Path, +) -> None: + """Missing _sb annual loads must not silently use stale raw annual loads.""" + release_sb = tmp_path / "res_2024_amy2018_2_sb" + (release_sb / "metadata").mkdir(parents=True) + raw_annual_dir = ( + tmp_path + / "res_2024_amy2018_2" + / "load_curve_annual" + / "state=MD" + / "upgrade=02" + ) + raw_annual_dir.mkdir(parents=True) + + result = _default_path_load_curve_annual(release_sb / "metadata", "MD", "02") + + assert result is None + + def _md_metadata( bldg_ids: list[int], gas_utilities: Sequence[str | None], diff --git a/tests/test_resstock_manifest.py b/tests/test_resstock_manifest.py index ed4d7503..ab0d6bbb 100644 --- a/tests/test_resstock_manifest.py +++ b/tests/test_resstock_manifest.py @@ -398,7 +398,7 @@ def test_check_integrity_file_type_bijection_passes(tmp_path: Path) -> None: def test_check_integrity_flags_unexpected_file_type_on_sb(tmp_path: Path) -> None: - """load_curve_annual on _sb must fail: excluded from _sb and not in expected set.""" + """Raw-cloned load_curve_annual on _sb must fail when add_monthly_loads did not run.""" from data.resstock.manifest import check_integrity ebs_base = str(tmp_path) diff --git a/utils/pre/gas_tariff_mapper.py b/utils/pre/gas_tariff_mapper.py index 9a0613ad..a15c0b6d 100644 --- a/utils/pre/gas_tariff_mapper.py +++ b/utils/pre/gas_tariff_mapper.py @@ -263,8 +263,12 @@ def map_gas_tariff( raise ValueError( "Chesapeake-territory gas utilities " f"({sorted(distinct_gas_vals & CHESAPEAKE_GAS_UTILITIES)}) require " - "annual_gas_therms (from load_curve_annual) for RES-1/RES-2 mapping. " - "Pass --path_load_curve_annual or annual_gas_therms=..." + "annual_gas_therms from the _sb load_curve_annual for RES-1/RES-2 " + "mapping, but no annual loads were found. Re-run the ResStock " + "_sb pipeline with --add-annual-loads True (default) so " + "load_curve_annual is generated from modified hourly, then " + "point --path_load_curve_annual at " + ".../res_*_sb/load_curve_annual/state=/upgrade=/." ) selected = utility_metadata.select( @@ -295,7 +299,10 @@ def map_gas_tariff( if missing > 0: raise ValueError( f"{missing} Chesapeake-territory building(s) lack annual_gas_therms " - "after joining load_curve_annual; check path and upgrade_id." + "after joining load_curve_annual. Confirm --path_load_curve_annual " + "points at the _sb release annual directory for this state/upgrade " + "(not raw NREL). If that directory is missing or incomplete, " + "re-run the ResStock _sb pipeline with --add-annual-loads True." ) gas_tariff_mapping_df = selected.with_columns(_tariff_key_expr()).drop( @@ -311,25 +318,19 @@ def map_gas_tariff( def _default_path_load_curve_annual( metadata_path: str | Path, state: str, upgrade_id: str ) -> Path | None: - """Prefer raw-release annual loads next to an ``*_sb`` metadata root. + """Use annual loads from the same release as the metadata. - ``load_curve_annual`` is often incomplete under ``*_sb`` (sample-sized); - the full population lives under the matching raw release directory. + Gas tariff subclasses must reflect Switchbox load modifications, so an + ``*_sb`` metadata root resolves to that release's generated + ``load_curve_annual`` rather than the raw NREL sibling. """ meta = Path(metadata_path) # metadata_path is typically .../res_..._sb/metadata release = meta.parent if meta.name == "metadata" else meta - candidates: list[Path] = [] - if release.name.endswith("_sb"): - candidates.append(Path(str(release)[: -len("_sb")])) - candidates.append(release) - for root in candidates: - annual_dir = ( - root / "load_curve_annual" / f"state={state}" / f"upgrade={upgrade_id}" - ) - if annual_dir.exists(): - return annual_dir - return None + annual_dir = ( + release / "load_curve_annual" / f"state={state}" / f"upgrade={upgrade_id}" + ) + return annual_dir if annual_dir.exists() else None if __name__ == "__main__": @@ -364,10 +365,12 @@ def _default_path_load_curve_annual( "--path_load_curve_annual", default=None, help=( - "Path to ResStock load_curve_annual parquet file or directory " + "Path to ResStock _sb load_curve_annual parquet file or directory " "(state=/upgrade= hive folder). Required for Chesapeake RES-1/RES-2 " - "mapping when those gas utilities appear. If omitted, tries the raw " - "release sibling of an *_sb metadata root." + "mapping when those gas utilities appear. If omitted, uses " + "load_curve_annual from the same release as --metadata_path " + "(expected to be the _sb release). Missing _sb annual loads raise " + "an error; re-run the ResStock pipeline with --add-annual-loads." ), ) args = parser.parse_args() @@ -533,14 +536,17 @@ def _default_path_load_curve_annual( except ValueError: use_s3_annual = False annual_path_resolved = Path(annual_path_arg) + missing_annual_msg = ( + f"load_curve_annual path {annual_path_resolved} does not exist. " + "Chesapeake RES-1/RES-2 mapping requires the generated _sb annual " + "loads. Re-run the ResStock _sb pipeline with --add-annual-loads " + "True (default), e.g. " + "`just -f data/resstock/Justfile run-pipeline MD`." + ) if use_s3_annual and not annual_path_resolved.exists(): - raise FileNotFoundError( - f"load_curve_annual path {annual_path_resolved} does not exist" - ) + raise FileNotFoundError(missing_annual_msg) if not use_s3_annual and not Path(annual_path_resolved).exists(): - raise FileNotFoundError( - f"load_curve_annual path {annual_path_resolved} does not exist" - ) + raise FileNotFoundError(missing_annual_msg) annual_gas_therms_lf = load_annual_gas_therms( annual_path_resolved, storage_options=STORAGE_OPTIONS if use_s3_annual else None, diff --git a/uv.lock b/uv.lock index 6b8d8cc9..c9d275c1 100644 --- a/uv.lock +++ b/uv.lock @@ -256,7 +256,7 @@ wheels = [ [[package]] name = "buildstock-fetch" -version = "1.6.5" +version = "1.6.6" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "aioboto3" }, @@ -281,9 +281,9 @@ dependencies = [ { name = "useful-types" }, { name = "xmltodict" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/ff/c8/acdc32a67d719d9c15dc4c50ed88a12a51e21a4fc035209f05488b7c0239/buildstock_fetch-1.6.5.tar.gz", hash = "sha256:f5a60c47a8f4681915f81e0fa6ed5bc9dfa25eb6e257b588190c4efd5bb1a143", size = 80073848, upload-time = "2026-02-12T05:08:55.974Z" } +sdist = { url = "https://files.pythonhosted.org/packages/ab/c0/9eb18ea330b0391437a03ff0294b4d3d9adb95390589f057033f69bd4854/buildstock_fetch-1.6.6.tar.gz", hash = "sha256:9c608f37dd9d0a6706b8ced9c3e075dc051060462a2e3cdffe9e1b7089cf4f4f", size = 80080404, upload-time = "2026-07-28T14:24:18.11Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/17/4b/0380c0f4cb9473f35434e5a7a1142fdff6093fac6bb724fdac05a357900f/buildstock_fetch-1.6.5-py3-none-any.whl", hash = "sha256:8840b52452c01314dc15713cd8db93d69fe078ea50dd4afff23fd3bb5a98ae1b", size = 80540260, upload-time = "2026-02-12T05:08:47.513Z" }, + { url = "https://files.pythonhosted.org/packages/69/08/82221b6949d289bf9561583754504c84907b33b4d2d4bd3ee032dbd4e6c3/buildstock_fetch-1.6.6-py3-none-any.whl", hash = "sha256:36eb929db14c89c7a63e4c7655d26cf888ece1f4b74a42fcaa424fe602efc183", size = 80545275, upload-time = "2026-07-28T14:24:11.95Z" }, ] [[package]] @@ -2351,7 +2351,7 @@ dev = [ [package.metadata] requires-dist = [ - { name = "buildstock-fetch", specifier = ">=1.6.1" }, + { name = "buildstock-fetch", specifier = ">=1.6.6" }, { name = "cairo", git = "https://github.com/natlabrockies/cairo?rev=09a4d7f" }, { name = "cloudpathlib", specifier = ">=0.23.0" }, { name = "cloudpathlib", extras = ["s3"], specifier = ">=0.23.0" },