diff --git a/scripts/us_epa/national_emissions_inventory/manifest.json b/scripts/us_epa/national_emissions_inventory/manifest.json index aa997fecf3..48212799ca 100644 --- a/scripts/us_epa/national_emissions_inventory/manifest.json +++ b/scripts/us_epa/national_emissions_inventory/manifest.json @@ -17,14 +17,16 @@ "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..ee74849f11 100644 --- a/scripts/us_epa/national_emissions_inventory/process.py +++ b/scripts/us_epa/national_emissions_inventory/process.py @@ -125,29 +125,33 @@ 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) @@ -157,7 +161,7 @@ def _regularize_columns(self, df: pd.DataFrame, 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,7 +231,7 @@ 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}") + logging.fatal(f"Error processing file {file_path}: {e}") return pd.DataFrame() def _process_file(self, file_path: str) -> None: @@ -245,7 +249,7 @@ 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.fatal(f"Error processing file {file_path}: {e}") def _mcf_property_generator(self) -> None: """ @@ -303,7 +307,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 +319,15 @@ 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.fatal(f"Error reading intermediate file {f}: {e}") 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', @@ -431,7 +433,7 @@ 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.fatal(f"An unexpected error occurred: {e}") def main(_): @@ -447,4 +449,4 @@ def main(_): if __name__ == "__main__": - app.run(main) + app.run(main) \ No newline at end of file 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..66014b3369 --- /dev/null +++ b/scripts/us_epa/national_emissions_inventory/validation_config.json @@ -0,0 +1,13 @@ +{ + "schema_version": "1.0", + "rules": [ + { + "rule_id": "check_deleted_records_percent", + "description": "Checks that the percentage of deleted points is within the threshold.", + "validator": "DELETED_RECORDS_PERCENT", + "params": { + "threshold": 0.1 + } + } + ] +}