diff --git a/scripts/us_epa/national_emissions_inventory/__init__.py b/scripts/us_epa/national_emissions_inventory/__init__.py new file mode 100644 index 0000000000..f8cb6d9d20 --- /dev/null +++ b/scripts/us_epa/national_emissions_inventory/__init__.py @@ -0,0 +1,13 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. diff --git a/scripts/us_epa/national_emissions_inventory/manifest.json b/scripts/us_epa/national_emissions_inventory/manifest.json index aa997fecf3..32ba242ffd 100644 --- a/scripts/us_epa/national_emissions_inventory/manifest.json +++ b/scripts/us_epa/national_emissions_inventory/manifest.json @@ -12,22 +12,23 @@ "process.py" ], "source_files": [ - "gcs_output/input_files/*/*.csv" + "gcs_output/input_files/*/*.csv", + "validation_config.json" ], "import_inputs": [ { "template_mcf": "gcs_output/output_files/national_emissions.tmcf", - "cleaned_csv": "gcs_output/output_files/national_emissions.csv" + "cleaned_csv": "gcs_output/output_files/national_emissions.csv", + "node_mcf": "gcs_output/output_files/national_emissions.mcf" } ], "cron_schedule": "0 0 1 1-12/3 *", + "validation_config_file": "validation_config.json", "resource_limits": { - "cpu": 8, - "memory": 128, - "disk": 100 + "cpu": 32, + "memory": 512, + "disk": 300 } } ] } - - diff --git a/scripts/us_epa/national_emissions_inventory/process.py b/scripts/us_epa/national_emissions_inventory/process.py index 3a14191517..d493408b2d 100644 --- a/scripts/us_epa/national_emissions_inventory/process.py +++ b/scripts/us_epa/national_emissions_inventory/process.py @@ -19,8 +19,7 @@ import os import sys import time -# import shutil -# import tempfile +import traceback import concurrent.futures from absl import app, flags, logging import pandas as pd @@ -125,39 +124,50 @@ def _regularize_columns(self, df: pd.DataFrame, df.rename(columns=replacement_08_11, inplace=True) df['pollutant type(s)'] = 'nan' if 'event' in file_path: - df.loc[:, 'emissions type code'] = '' + df['emissions type code'] = '' elif 'process' in file_path: df = df.dropna(subset=['fips code']) - df.loc[:, 'emissions type code'] = '' + df['emissions type code'] = '' if '2008' in file_path: - df.loc[:, 'year'] = '2008' + df['year'] = '2008' else: - df.loc[:, 'year'] = '2011' + df['year'] = '2011' elif '2017' in file_path: if 'Event' in file_path: df['pollutant type(s)'] = 'nan' - elif 'point' in file_path: + elif 'point_' in os.path.basename( + file_path) or 'facility_process' in file_path: if 'unknown' in file_path or '678910' in file_path: df.rename(columns=replacement_point_17, inplace=True) - df.loc[:, 'emissions type code'] = '' + df['emissions type code'] = '' + elif 'nonpoint' in file_path: + df['emissions type code'] = '' df['year'] = '2017' elif '2020' in file_path: if 'Event' in file_path: df['pollutant type(s)'] = 'nan' - elif 'point' in file_path: + elif 'point_' in os.path.basename( + file_path) or 'facility_process' in file_path: if 'unknown' in file_path: df.rename(columns=replacement_20, inplace=True) - df.loc[:, 'emissions type code'] = '' + df['emissions type code'] = '' + elif 'nonpoint' in file_path: + df['emissions type code'] = '' df['year'] = '2020' elif 'tribes' in file_path: - df.rename(columns=replacement_tribes, inplace=True) - df = self._data_standardize(df, 'fips code') + if 'fips code' not in df.columns and 'tribal name' in df.columns: + df.rename(columns=replacement_tribes, inplace=True) + df = self._data_standardize(df, 'fips code') + else: + df.rename(columns=replacement_14, inplace=True) + if 'event' in file_path or 'process' in file_path: + df['emissions type code'] = '' df['pollutant type(s)'] = 'nan' df['year'] = '2014' else: df.rename(columns=replacement_14, inplace=True) if 'event' in file_path or 'process' in file_path: - df.loc[:, 'emissions type code'] = '' + df['emissions type code'] = '' df['pollutant type(s)'] = 'nan' df['year'] = '2014' @@ -227,8 +237,8 @@ def _national_emissions(self, file_path: str) -> pd.DataFrame: errors='coerce') return df except Exception as e: - logging.error(f"Error processing file {file_path}: {e}") - return pd.DataFrame() + logging.exception(f"Error processing file {file_path}: {e}") + raise def _process_file(self, file_path: str) -> None: """ @@ -236,7 +246,7 @@ def _process_file(self, file_path: str) -> None: """ try: df = self._national_emissions(file_path) - if not df.empty: + if df is not None and not df.empty: intermediate_file_path = os.path.join( self.temp_dir, f"{str(datetime.now().timestamp()).replace('.', '_')}_{os.path.basename(file_path)}" @@ -245,7 +255,8 @@ def _process_file(self, file_path: str) -> None: logging.info( f"Saved intermediate file at : {intermediate_file_path}") except Exception as e: - logging.error(f"Error processing file {file_path}: {e}") + logging.exception(f"Error processing file {file_path}: {e}") + raise def _mcf_property_generator(self) -> None: """ @@ -303,7 +314,7 @@ def _process(self): logging.info("Starting data processing across all input files.") with concurrent.futures.ThreadPoolExecutor( max_workers=MAX_WORKERS) as executor: - executor.map(self._process_file, self._input_files) + list(executor.map(self._process_file, self._input_files)) logging.info("Consolidating intermediate files.") intermediate_files = [ @@ -315,17 +326,18 @@ def _process(self): dfs.append(pd.read_csv(f, low_memory=False)) logging.info(f"Appending {f}") except Exception as e: - logging.error(f"Error reading intermediate file {f}: {e}") + logging.exception(f"Error reading intermediate file {f}: {e}") + logging.fatal( + f"Error reading intermediate file {f}: {e}\n{traceback.format_exc()}" + ) if not dfs: - logging.error("No dataframes to concatenate. Exiting.") - return + logging.fatal("No dataframes to concatenate. Exiting.") self.final_df = pd.concat(dfs, ignore_index=True) self.final_df = self.final_df.sort_values( by=['geo_Id', 'year', 'SV', 'Measurement_Method', 'observation']) - self.final_df['observation'].replace('', np.nan, inplace=True) self.final_df.dropna(subset=['observation'], inplace=True) self.final_df['observation'] = np.where( self.final_df['unit'] == 'Pound', @@ -407,9 +419,10 @@ def process_files(input_path: str, output_file_path: str, if file.lower().endswith('.csv') ] except Exception as e: + logging.exception(f"Error finding input files: {e}") logging.fatal( - f"Error finding input files: {e}. Run the download script first.\n") - sys.exit(1) + f"Error finding input files: {e}. Run the download script first.\n{traceback.format_exc()}" + ) # Defining Output Files logging.info( @@ -431,7 +444,9 @@ def process_files(input_path: str, output_file_path: str, loader.generate_mcf() loader.generate_tmcf() except Exception as e: - logging.error(f"An unexpected error occurred: {e}") + logging.exception(f"An unexpected error occurred: {e}") + logging.fatal( + f"An unexpected error occurred: {e}\n{traceback.format_exc()}") def main(_): diff --git a/scripts/us_epa/national_emissions_inventory/process_test.py b/scripts/us_epa/national_emissions_inventory/process_test.py index 7b70884dd4..c04aabf69b 100644 --- a/scripts/us_epa/national_emissions_inventory/process_test.py +++ b/scripts/us_epa/national_emissions_inventory/process_test.py @@ -12,12 +12,21 @@ # See the License for the specific language governing permissions and # limitations under the License. -import unittest -import os -import tempfile import filecmp +import os import shutil -from .process import * +import sys +import tempfile +import unittest + +import numpy as np +import pandas as pd + +_MODULE_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, _MODULE_DIR) + +from config import df_columns +from process import USAirEmissionTrends, process_files class ProcessEnhancedTest(unittest.TestCase): @@ -30,6 +39,8 @@ def setUp(self): def tearDown(self): shutil.rmtree(self.temp_dir) + if os.path.exists(self.intermediate_path): + shutil.rmtree(self.intermediate_path) def test_script(self): input_path = os.path.join(self.test_data_dir, 'input') @@ -55,5 +66,157 @@ def test_script(self): f"File content mismatch: {expected_file} and {generated_file}") +class RegularizeColumnsTest(unittest.TestCase): + + def setUp(self): + self.loader = USAirEmissionTrends([], '', '', '', '') + + def test_regularize_columns_2017_nonpoint_float64_nan(self): + # 2017 nonpoint file with float64 NaN emissions type code + df = pd.DataFrame({ + 'fips code': [1001, 1003], + 'scc': [10100101, 10100201], + 'pollutant code': ['CO', 'NOX'], + 'total emissions': [12.5, 34.2], + 'emissions uom': ['TON', 'TON'], + 'emissions type code': [np.nan, np.nan], + }) + self.assertEqual(df['emissions type code'].dtype, np.float64) + + result = self.loader._regularize_columns(df, + '/path/to/2017_nonpoint.csv') + + self.assertEqual(list(result.columns), df_columns) + self.assertTrue((result['year'] == '2017').all()) + self.assertTrue((result['emissions type code'] == '').all()) + self.assertEqual(result['fips code'].tolist(), [1001, 1003]) + + def test_regularize_columns_2020_nonpoint_float64_nan(self): + # 2020 nonpoint file with float64 NaN emissions type code + df = pd.DataFrame({ + 'fips code': [2013, 2016], + 'scc': [20100101, 20100201], + 'pollutant code': ['SO2', 'VOC'], + 'total emissions': [5.1, 8.7], + 'emissions uom': ['TON', 'TON'], + 'emissions type code': [np.nan, np.nan], + }) + self.assertEqual(df['emissions type code'].dtype, np.float64) + + result = self.loader._regularize_columns( + df, '/path/to/2020_nonpoint_data.csv') + + self.assertEqual(list(result.columns), df_columns) + self.assertTrue((result['year'] == '2020').all()) + self.assertTrue((result['emissions type code'] == '').all()) + self.assertEqual(result['fips code'].tolist(), [2013, 2016]) + + def test_regularize_columns_point_files(self): + # 2017 point_ file with unknown + df_pt17_unk = pd.DataFrame({ + 'fips': [1001], + 'pollutant_code': ['CO'], + 'total_emissions': [15.0], + 'emissions_uom': ['TON'], + 'scc': [10100101], + }) + res_pt17_unk = self.loader._regularize_columns( + df_pt17_unk, '/path/to/2017_point_unknown.csv') + self.assertEqual(list(res_pt17_unk.columns), df_columns) + self.assertEqual(res_pt17_unk['year'].iloc[0], '2017') + self.assertEqual(res_pt17_unk['emissions type code'].iloc[0], '') + self.assertEqual(res_pt17_unk['fips code'].iloc[0], 1001) + + # 2017 point_ file with 678910 + df_pt17_num = pd.DataFrame({ + 'fips': [1002], + 'pollutant_code': ['NOX'], + 'total_emissions': [22.0], + 'emissions_uom': ['TON'], + 'scc': [10100102], + }) + res_pt17_num = self.loader._regularize_columns( + df_pt17_num, '/path/to/2017_point_678910.csv') + self.assertEqual(list(res_pt17_num.columns), df_columns) + self.assertEqual(res_pt17_num['year'].iloc[0], '2017') + self.assertEqual(res_pt17_num['emissions type code'].iloc[0], '') + self.assertEqual(res_pt17_num['fips code'].iloc[0], 1002) + + # 2020 point_ file with unknown + df_pt20_unk = pd.DataFrame({ + 'fips state/county code': [1003], + 'pollutant code': ['SO2'], + 'total emissions': [30.0], + 'uom': ['TON'], + 'scc': [10100103], + }) + res_pt20_unk = self.loader._regularize_columns( + df_pt20_unk, '/path/to/2020_point_unknown.csv') + self.assertEqual(list(res_pt20_unk.columns), df_columns) + self.assertEqual(res_pt20_unk['year'].iloc[0], '2020') + self.assertEqual(res_pt20_unk['emissions type code'].iloc[0], '') + self.assertEqual(res_pt20_unk['fips code'].iloc[0], 1003) + + # facility_process file + df_fac = pd.DataFrame({ + 'fips code': [1004], + 'pollutant code': ['PM10-PRI'], + 'total emissions': [4.5], + 'emissions uom': ['TON'], + 'scc': [10100104], + }) + res_fac = self.loader._regularize_columns( + df_fac, '/path/to/2017_facility_process.csv') + self.assertEqual(list(res_fac.columns), df_columns) + self.assertEqual(res_fac['year'].iloc[0], '2017') + self.assertEqual(res_fac['emissions type code'].iloc[0], '') + + def test_regularize_columns_2014_tribes(self): + # Tribal file with 'tribal name' and without 'fips code' + df_tribes = pd.DataFrame({ + 'tribal name': ['Navajo Nation'], + 'scc': [10100101], + 'pollutant code': ['VOC'], + 'total emissions': [10.0], + 'emissions uom': ['TON'], + }) + res_tribes = self.loader._regularize_columns( + df_tribes, '/path/to/2014_tribes_data.csv') + self.assertEqual(list(res_tribes.columns), df_columns) + self.assertEqual(res_tribes['year'].iloc[0], '2014') + self.assertEqual(res_tribes['fips code'].iloc[0], 'Navajo Nation') + self.assertEqual(res_tribes['pollutant type(s)'].iloc[0], 'nan') + + # Tribal process file process_tribes.csv with existing 'fips' column + df_proc_tribes = pd.DataFrame({ + 'fips': [1005], + 'scc': [10100105], + 'pollutant_cd': ['CO'], + 'total_emissions': [18.0], + 'uom': ['TON'], + }) + res_proc_tribes = self.loader._regularize_columns( + df_proc_tribes, '/path/to/process_tribes.csv') + self.assertEqual(list(res_proc_tribes.columns), df_columns) + self.assertEqual(res_proc_tribes['year'].iloc[0], '2014') + self.assertEqual(res_proc_tribes['emissions type code'].iloc[0], '') + self.assertEqual(res_proc_tribes['fips code'].iloc[0], 1005) + self.assertEqual(res_proc_tribes['pollutant type(s)'].iloc[0], 'nan') + + # Tribal event file event_tribes.csv + df_event_tribes = pd.DataFrame({ + 'fips': [1006], + 'scc': [10100106], + 'pollutant_cd': ['PM25-PRI'], + 'total_emissions': [2.5], + 'uom': ['TON'], + }) + res_event_tribes = self.loader._regularize_columns( + df_event_tribes, '/path/to/event_tribes.csv') + self.assertEqual(list(res_event_tribes.columns), df_columns) + self.assertEqual(res_event_tribes['year'].iloc[0], '2014') + self.assertEqual(res_event_tribes['emissions type code'].iloc[0], '') + + if __name__ == '__main__': unittest.main() diff --git a/scripts/us_epa/national_emissions_inventory/validation_config.json b/scripts/us_epa/national_emissions_inventory/validation_config.json new file mode 100644 index 0000000000..8f304120f6 --- /dev/null +++ b/scripts/us_epa/national_emissions_inventory/validation_config.json @@ -0,0 +1,22 @@ +{ + "schema_version": "1.0", + "rules": [ + { + "rule_id": "check_deleted_records_percent", + "description": "Verifies that the percentage of deleted records does not exceed the 0.1% threshold, ensuring unintended observation drops (such as missing nonpoint sources which drop >20% of records) are caught while accommodating minor upstream EPA revisions.", + "validator": "DELETED_RECORDS_PERCENT", + "params": { + "threshold": 0.1 + } + }, + { + "rule_id": "check_date_freshness", + "description": "Verifies that the dataset's observation dates meet freshness and temporal coverage requirements (MAX(MaxDate) >= 2020 and MIN(MaxDate) >= 2008)", + "validator": "SQL_VALIDATOR", + "params": { + "query": "SELECT MAX(MaxDate) AS max_date, MIN(MaxDate) AS min_date FROM stats", + "condition": "max_date >= '2020' AND min_date >= '2008'" + } + } + ] +}