Merge pull request #1509 from NREL/sdr_2025_release_1_postprocessing

SDR postprocessing

Author: asparke2

postprocessing/README.md

@@ -35,11 +35,19 @@ To install the package, we recommend using `uv` for Python package management.
    uv run --group dev pre-commit install
    ```
 
-# Usage
-
-
 4. Run the scripts as desired
    ```bash
+   # Output the failure log
    cd path/to/postprocessing
    uv run resstockpostproc/get_failures.py <csv_path> --verbose
+
+   # Export metadata and annual results from files on S3
+   uv run resstockpostproc/process_bsb_results.py "s3://res-sdr/testing-sdr-fy25/a_run" "C:/path/to/bsb/output/a_run_output"
+
+   # Export metdata and annual results from local files
+   # (It is faster to download the /baseline and /upgrades directories from S3 once instead of reading from S3 each time)
+   uv run resstockpostproc/process_bsb_results.py "C:/path/to/bsb/output/a_run" "C:/path/to/bsb/output/a_run_output"
+
+   # Export metdata and annual results to OEDI
+   uv run resstockpostproc/process_bsb_results.py "C:/path/to/bsb/output/a_run" "s3://oedi-data-lake/nrel-pds-building-stock/end-use-load-profiles-for-us-building-stock/2025/resstock_amy2018_release_1"
    ```

postprocessing/resstockpostproc/__init__.py

@@ -1,6 +1 @@
-from resstockpostproc.process_metadata import publish_baseline_annual_results, publish_upgrade_annual_results
-
-__all__ = [
-    "publish_baseline_annual_results",
-    "publish_upgrade_annual_results",
-]
+from .process_metadata import process_simulation_outputs

postprocessing/resstockpostproc/data_dict.py

@@ -0,0 +1,48 @@
+import pandas as pd
+import pathlib
+def data_dictionary(df_sdr):
+    """
+    generate data dictionary based on sdr_column_definitions.csv.
+    """
+    df_sdr_meta = df_sdr[(df_sdr['Publish In Full'] == 'yes') & (df_sdr['Published Annual Name'].notnull())]
+    df_sdr_tsagg = df_sdr[(df_sdr['Timeseries Publish In Full'] == 'yes') & (df_sdr['Published Timeseries Name'].notnull())]
+
+    # metadata_and_annual_results column names, units, and description
+    df_meta = df_sdr_meta[['Published Annual Name',
+                           'Data Type',
+                           'Published Annual Unit',
+                           'Notes']].rename(columns={
+                               'Published Annual Name': 'field_name',
+                               'Data Type': 'data_type',
+                               'Published Annual Unit': 'units',
+                               'Notes': 'field_description' 
+                               })
+    df_meta.insert(loc=0, column='field_location', value='metadata_and_annual')
+
+    # timeseries_aggregates column names, units, and description
+    df_tsagg_sdr = df_sdr_tsagg[['Published Timeseries Name',
+                                 'Data Type',
+                                 'Published Timeseries Unit',
+                                 'Notes']].rename(columns={
+                                     'Published Timeseries Name': 'field_name',
+                                     'Data Type': 'data_type',
+                                     'Published Timeseries Unit': 'units',
+                                     'Notes': 'field_description'
+                                     })
+    df_tsagg_sdr.insert(loc=0, column='field_location', value='timeseries_aggregates')
+
+    #combine metadata_and_annual_results and timeseries_aggregates
+    df_data_dict = pd.concat([df_meta, df_tsagg_sdr], ignore_index=True)
+    df_data_dict['units'] = df_data_dict['units'].fillna('n/a')
+    
+    return df_data_dict
+
+def main():
+    here = pathlib.Path(__file__).resolve().parent
+    df_sdr = pd.read_csv(here / "resources" / "publication" / "sdr_column_definitions.csv")
+    df_data_dict = data_dictionary(df_sdr)
+    df_data_dict.to_csv(here / "resources" / "publication" / "data_dictionary.tsv", sep='\t', index=None)
+
+
+if __name__ == "__main__":
+    main()

postprocessing/resstockpostproc/enum_dict.py

@@ -0,0 +1,84 @@
+import pandas as pd
+import pathlib
+
+
+def enum(df):
+    """
+    enumerations for a dataframe
+    """
+    df_enum = (
+        pd.concat(
+            [pd.DataFrame({'metadata_column': col, 'enumeration': df[col].unique()})
+             for col in df.columns],
+             ignore_index=True
+             )
+             )
+    
+    return df_enum
+
+
+def enum_dict(df_data_dict, df_bs_csv, df_meta_up, up_files):
+    #format buildstock.csv column names
+    df_bs_csv.columns = ['in.' + col.lower().replace(' ', '_') for col in df_bs_csv.columns]
+    df_bs_csv = df_bs_csv.drop('in.building', axis=1)
+    df_bs_csv = df_bs_csv.rename(columns={
+        'in.ashrae_iecc_climate_zone_2004_-_sub-cz_split': 'in.ashrae_iecc_climate_zone_2004_sub_cz_split',
+        'in.income_recs2015': 'in.income_recs_2015',
+        'in.income_recs2020': 'in.income_recs_2020'
+        })
+
+    #enumerations from buildstock.csv
+    df_enum_bs_csv = enum(df_bs_csv)
+
+    df_data_dict_filter = df_data_dict[df_data_dict['field_location'] == 'metadata_and_annual']
+    data_dict_columns = df_data_dict_filter['field_name']
+    data_dict_columns = [x for x in data_dict_columns if not x.startswith(("out.", "calc.weighted", "bldg_id"))]
+
+    bs_csv_columns = df_bs_csv.columns
+    leftover_columns = list(set(data_dict_columns) - set(bs_csv_columns))
+    
+    #enumerations from released data
+    df_enum_meta = pd.DataFrame(columns=['metadata_column', 'enumeration'])
+    for up in up_files:
+        #Do not need the renaming for the released parquet file
+        df_meta_up[up] = df_meta_up[up].rename(columns={
+            'in.sqft': 'in.sqft..ft2',
+            'in.air_leakage_to_outside_ach_50': 'in.air_leakage_to_outside_ach50',
+            'upgrade_name': 'in.upgrade_name',
+            'in.electric_panel_service_rating': 'in.electric_panel_service_rating..a',
+            'in.electric_panel_service_rating_bin': 'in.electric_panel_service_rating_bin..a',
+            'in.air_leakage_to_outside_ach_50': 'in.air_leakage_to_outside_ach50'
+            })
+        existing_cols = [c for c in leftover_columns if c in df_meta_up[up].columns]
+        df_meta_filter = df_meta_up[up][existing_cols]
+        df_meta_filter_enum = enum(df_meta_filter)
+        df_enum_meta = pd.concat([df_enum_meta, df_meta_filter_enum]).drop_duplicates(keep='first')
+    
+    df_enum_dict = pd.concat([df_enum_bs_csv, df_enum_meta]).drop_duplicates(keep='first')
+    df_enum_dict['enumeration'] = df_enum_dict['enumeration'].fillna("None")
+    df_enum_dict = df_enum_dict.sort_values(by=['metadata_column', 'enumeration'])
+
+    df_enum_dict_columns = df_enum_dict['metadata_column'].unique().tolist()
+    missing_cols = [c for c in data_dict_columns if c not in df_enum_dict_columns]
+    print("Missing columns:", missing_cols)
+
+    return df_enum_dict
+
+
+def main():
+    here = pathlib.Path(__file__).resolve().parent
+    test_path = here.parent.parent
+    df_data_dict = pd.read_csv(here / "resources" / "publication" / "data_dictionary.tsv", sep='\t')
+    df_bs_csv = pd.read_csv(test_path / "test" / "base_results" / "baseline"/ "annual"/ "buildstock.csv")
+    df_meta_up = {}
+    up_path = (test_path / "test" / "base_results" / "upgrades"/ "sdr_annual")
+    up_files = [f.name for f in up_path.glob('*.csv')]
+    for up in up_files:
+        df_meta_up[up] = pd.read_csv(test_path / "test" / "base_results" / "upgrades"/ "sdr_annual"/ up)
+
+    df_enum_dict = enum_dict(df_data_dict, df_bs_csv, df_meta_up, up_files)
+    df_enum_dict.to_csv(here / "resources" / "publication" / "enumeration_dictionary.tsv", sep='\t', index=None)
+
+
+if __name__ == "__main__":
+    main()

postprocessing/resstockpostproc/income_mapper.py

@@ -108,7 +108,7 @@ def assign_representative_income(df: pl.LazyFrame | pl.DataFrame, return_map_onl
     check_df = check_df.collect() if lazy else check_df
     assert len(check_df) == 0, f"rep_income could not be mapped for {len(check_df)} rows\n{check_df}"
 
-    print(f"Note: {rep_inc} is not available for vacant units, which have 'Not Available' for in.income")
+    # print(f"Note: {rep_inc} is not available for vacant units, which have 'Not Available' for in.income")
 
     df3 = df2.select([bldg_id, rep_inc])
     if return_map_only:

postprocessing/resstockpostproc/process_bsb_results.py

@@ -4,87 +4,99 @@
 
 Example usage:
 uv run resstockpostproc/process_bsb_results.py /path/to/bsb_raw_results /path/to/output_dir
+uv run resstockpostproc/process_bsb_results.py "C:/Scratch/ResStock/efforts/full_550k_run" "s3://oedi-data-lake/nrel-pds-building-stock/end-use-load-profiles-for-us-building-stock/2025/resstock_amy2018_release_1"
 
 Note: bsb_raw_results folder must contain both baseline and upgrade files. Baseline file should be named
 results_up00.parquet and upgrade files should be named results_upXX.parquet where XX is the upgrade number. The can
 either be in their own folders (baseline and upgrades) or all be in the same folder.
 """
 
-import sys
+import re
 import polars as pl
 from pathlib import Path
 from resstockpostproc.process_metadata import (
-    publish_baseline_annual_results,
-    publish_upgrade_annual_results,
+    get_schema_superset,
+    get_upgrade_rename_dict,
+    get_failed_building_list,
+    process_simulation_outputs,
+    export_metadata_and_annual_results_for_upgrade,
+    cache_simulation_outputs_file
+)
+from resstockpostproc.utils import (
+    setup_fsspec_filesystem
 )
-import re
-
-
-def process_results(raw_results_dir: str, output_dir: str) -> None:
-    output_path = Path(output_dir)
-    output_path.mkdir(parents=True, exist_ok=True)
-    result_files = list(Path(raw_results_dir).rglob("*"))
-    baseline_files = [f for f in result_files if "up00" in f.name.lower()]
-    upgrade_files = [f for f in result_files if "up00" not in f.name.lower()]
-
-    if not baseline_files:
-        print("Error: No baseline or upgrade files found")
-        sys.exit(1)
-    if len(baseline_files) > 1:
-        print("Error: More than one baseline file found")
-        sys.exit(1)
-
-    baseline_file = baseline_files[0]
-    print(f"Processing baseline file: {baseline_file}")
-    baseline_df = read_file(baseline_file)
-
-    failed_bldgs = (
-        baseline_df.filter(pl.col("completed_status") == "Fail")
-        .select(pl.col("building_id"))
-        .collect()["building_id"]
-        .to_list()
-    )
-    print(f"Removing {len(failed_bldgs)} buildings that failed in baseline")
-    bs_pub_df = publish_baseline_annual_results(baseline_df)
-    write_file(bs_pub_df, output_path, upgrade=0)
-
-    for upgrade_file in upgrade_files:
-        up_info = re.search(r"up(\d+)", upgrade_file.name)
-        if up_info is None:
-            continue
-        upgrade_num = int(up_info.group(1))
 
-        print(f"Processing upgrade file: {upgrade_file}, upgrade number: {upgrade_num}")
-        upgrade_df = read_file(upgrade_file)
-        up_up_df = publish_upgrade_annual_results(
-            failed_bldgs, bs_pub_df, upgrade_df, upgrade_num
+def export_metadata_and_annual_results(raw_results_dir: str,
+                                       output_dir: str,
+                                       aws_profile_name = None) -> None:
+    # Set up filesystem objects for raw results and output directories
+    raw_results_dir = setup_fsspec_filesystem(raw_results_dir, aws_profile_name)
+    output_dir = setup_fsspec_filesystem(output_dir, aws_profile_name)
+
+    # Find the raw results files
+    pqt_glob = f'{raw_results_dir["fs_path"]}/**/*.parquet'
+    result_files = raw_results_dir['fs'].glob(pqt_glob)
+    baseline_file = [f for f in result_files if "up00" in Path(f).name.lower()][0]
+    upgrade_ids = [int(re.search(r'up(\d+)', p).group(1)) for p in result_files]
+    upgrade_ids.sort()
+
+    # Information used across upgrades
+    upgrade_renamer = get_upgrade_rename_dict(raw_results_dir)
+    col_schema = get_schema_superset(result_files, raw_results_dir)
+    sim_out_cache_dir = Path(f"{output_dir['fs_path']}/cached_simulation_outputs")
+
+    # Process and cache the simulation outputs, starting with the baseline
+    baseline_df = pl.scan_parquet(baseline_file, storage_options=raw_results_dir['storage_options'])
+    failed_bldgs = get_failed_building_list(baseline_df)
+    processed_baseline_df = None
+    for upgrade_id in upgrade_ids:
+        upgrade_file = f'{raw_results_dir["fs_path"]}/upgrades/upgrade={upgrade_id}/results_up{upgrade_id:02d}.parquet'
+        if upgrade_id == 0:
+            upgrade_file = f'{raw_results_dir["fs_path"]}/baseline/results_up{upgrade_id:02d}.parquet'
+
+        print(f"Processing upgrade file: {upgrade_file}, upgrade number: {upgrade_id} {'*'*100}")
+        raw_upgrade_df = pl.scan_parquet(upgrade_file, storage_options=raw_results_dir['storage_options'])
+        processed_upgrade_df = process_simulation_outputs(
+            failed_bldgs,
+            baseline_df,
+            processed_baseline_df,
+            raw_upgrade_df,
+            upgrade_id,
+            upgrade_renamer,
+            col_schema
         )
-        write_file(up_up_df, output_path, upgrade_num)
-
-
-def read_file(file: Path) -> pl.LazyFrame:
-    match file.suffix:
-        case ".parquet":
-            return pl.scan_parquet(file)
-        case ".csv":
-            return pl.scan_csv(file)
-        case ".gz":
-            assert file.stem.endswith(".csv"), f"gz file is not a csv: {file}"
-            return pl.scan_csv(file)
-        case _:
-            raise ValueError(f"Unsupported file type: {file}")
-
-
-def write_file(df: pl.LazyFrame, output_path: Path, upgrade: int):
-    parquet_file_dir = output_path / "parquet" / f"upgrade={upgrade}"
-    parquet_file_dir.mkdir(parents=True, exist_ok=True)
-    csv_file_dir = output_path / "results_csvs_pub"
-    csv_file_dir.mkdir(parents=True, exist_ok=True)
-    csv_file = csv_file_dir / f"results_up{upgrade:02d}.csv"
-    parquet_file = parquet_file_dir / f"results_up{upgrade:02d}.parquet"
-    df.sink_parquet(parquet_file)
-    df.sink_csv(csv_file)
-    print(f"Wrote {upgrade} to {parquet_file} and {csv_file}")
+        cache_simulation_outputs_file(output_dir, sim_out_cache_dir, upgrade_id, processed_upgrade_df)
+        up_cols = set(sorted(processed_upgrade_df.collect_schema().names()))
+
+        if upgrade_id == 0:
+            processed_baseline_df = processed_upgrade_df
+            base_cols = set(sorted(processed_baseline_df.collect_schema().names()))
+
+        if not base_cols == up_cols:
+            raise ValueError("Column set in baseline and upgrade don't match")
+
+    # Export files to specified geographic partitions
+    geo_exports = [
+    {
+        'geo_top_dir': 'national',
+        'partition_cols': {},
+        'data_types': ['full'],  # TODO add basic
+        'file_types': [ 'csv', 'parquet'],
+    },
+    {
+        'geo_top_dir': 'by_state',
+        'partition_cols': {
+            'in.state': 'state'
+        },
+        'data_types': ['full'],  # TODO add basic
+        'file_types': ['csv', 'parquet'],
+    }
+    ]
+    for upgrade_id in upgrade_ids:
+        export_metadata_and_annual_results_for_upgrade(
+            output_dir,
+            upgrade_id,
+            geo_exports)
 
 
 if __name__ == "__main__":
@@ -104,4 +116,4 @@ def write_file(df: pl.LazyFrame, output_path: Path, upgrade: int):
         help="Directory to write transformed results",
     )
     args = parser.parse_args()
-    process_results(args.raw_results_dir, args.output_dir)
+    export_metadata_and_annual_results(args.raw_results_dir, args.output_dir)