Fix for epa-airpollutantemission-level1

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.

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
             }
         }
     ]
 }
-
-

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,16 +237,16 @@ 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:
         """
         Process a single file and save the intermediate result.
         """
         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(_):

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()