diff --git a/scripts/us_epa/air_emissions_inventory/national_state/README.md b/scripts/us_epa/air_emissions_inventory/national_state/README.md index ad1d6353a6..deb9299c40 100644 --- a/scripts/us_epa/air_emissions_inventory/national_state/README.md +++ b/scripts/us_epa/air_emissions_inventory/national_state/README.md @@ -52,31 +52,29 @@ or ### Import Procedure -The below script will download the data. +The below script will download the data: `python3 download.py --config_path=unresolved_mcf/us_epa/config.json` -Note on Configuration Management for Data Source URLs +#### Configuration Management & Automatic Fallback for Data Source URLs -To ensure consistent access to the most current data, a dedicated configuration file, config.json, will be maintained within our Google Cloud Storage (GCS) environment. This approach addresses the challenge posed by source URLs that contain dynamic and unpredictable year and date components, precluding their programmatic generation. +To access the latest data, a configuration file (`config.json`) is stored in Google Cloud Storage (GCS) at `gs://unresolved_mcf/us_epa/config.json`. Because the EPA periodically publishes revised National and State `.xlsx` workbooks with new release-date suffixes (retiring the older URLs with HTTP 404), `download.py` uses a two-tier resolution strategy: -Content Structure: -The config.json file will adhere to a JSON array format, with each object in the array defining a download_path attribute. This attribute will contain the fully qualified URL for the latest available National and State-level data files. +1. **Primary (GCS Config)**: Reads `download_path` entries from `--config_path` (passed as `bucket_name/path/to/config.json` without the `gs://` prefix) and verifies that all configured URLs return HTTP 200. +2. **Automatic Fallback (Live EPA Trends Page Discovery)**: If the GCS config cannot be read or any configured URL returns a non-200 status (such as HTTP 404 after an EPA release update), `download.py` automatically scrapes `https://www.epa.gov/air-emissions-inventories/air-pollutant-emissions-trends-data` to discover and validate the current `national_tier1_caps*.xlsx` and `state_tier1*ktons.xlsx` URLs. -The config_path parameter designates a Google Cloud Storage (GCS) location. It should be provided as the bucket and object path (e.g., my-bucket/path/to/config.json), without the gs:// prefix. The system will automatically prepend the gs:// scheme internally when accessing the resource." - -Example config.json Content: - -JSON +Current `config.json` Content (`gs://unresolved_mcf/us_epa/config.json`): +```json [ - {"download_path": "https://www.epa.gov/system/files/other-files/2025-04/national_tier1_caps_21feb2025.xlsx"}, - {"download_path": "https://www.epa.gov/system/files/other-files/2025-04/state_tier1_21feb2005_ktons.xlsx"} + {"download_path": "https://www.epa.gov/system/files/other-files/2026-09/national_tier1_caps_04sep2026.xlsx"}, + {"download_path": "https://www.epa.gov/system/files/other-files/2026-09/state_tier1_1990_2025_04sep2026_ktons.xlsx"} ] +``` +*(When the EPA publishes a newer release on the [Air Pollutant Emissions Trends Data](https://www.epa.gov/air-emissions-inventories/air-pollutant-emissions-trends-data) page, `download.py` automatically discovers the updated URLs if `config.json` has not yet been updated.)* -Maintenance: -This configuration mechanism requires manual updates to the config.json file whenever the source URLs for the National or State data undergo revision, thereby ensuring the application always references the latest datasets. +#### Processing -The below script will clean the data, Also generate final csv, mcf and tmcf files. +The below script cleans the downloaded workbooks and generates the final CSV, MCF, and TMCF files: -`python3 process.py` \ No newline at end of file +`python3 process.py` diff --git a/scripts/us_epa/air_emissions_inventory/national_state/download.py b/scripts/us_epa/air_emissions_inventory/national_state/download.py index 203d5c10f5..3f6b8e5d4e 100644 --- a/scripts/us_epa/air_emissions_inventory/national_state/download.py +++ b/scripts/us_epa/air_emissions_inventory/national_state/download.py @@ -18,7 +18,11 @@ import os import json -import urllib.request +import re +import urllib.parse +import requests +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry from google.cloud import storage # For GCS integration from absl import flags, app, logging # Use absl for flag handling @@ -27,31 +31,128 @@ flags.DEFINE_string('config_path', '', 'Path to the configuration file in the GCS bucket.') -# Hardcoded GCS bucket name -GCS_BUCKET_NAME = "unresolved_mcf" +# EPA Air Pollutant Emissions Trends Data provenance page +EPA_TRENDS_PAGE_URL = ("https://www.epa.gov/air-emissions-inventories/" + "air-pollutant-emissions-trends-data") # Download path for files -_DOWNLOAD_PATH = os.path.join(os.path.dirname(__file__), 'input_files') - - -def check_url_status(url: str) -> bool: +_DOWNLOAD_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), + 'input_files') + +_HTTP_TIMEOUT_SECONDS = 60 +_DEFAULT_HEADERS = { + "User-Agent": ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36") +} + + +def _create_http_session() -> requests.Session: + """Creates an HTTP session with retry and exponential backoff.""" + session = requests.Session() + session.headers.update(_DEFAULT_HEADERS) + retry_strategy = Retry( + total=5, + backoff_factor=2, + status_forcelist=[429, 500, 502, 503, 504], + allowed_methods=["HEAD", "GET", "OPTIONS"], + ) + adapter = HTTPAdapter(max_retries=retry_strategy) + session.mount("https://", adapter) + session.mount("http://", adapter) + return session + + +def check_url_status(url: str, session: requests.Session = None) -> bool: """ Check if the given URL is reachable and log URL info. Args: url (str): The URL to check. + session (requests.Session, optional): HTTP session to reuse. Returns: bool: True if the URL is reachable (status code 200), False otherwise. """ + own_session = session is None + if own_session: + session = _create_http_session() try: - with urllib.request.urlopen(url) as response: - status_code = response.status + logging.info(f"Checking URL status: {url}") + response = session.head(url, + allow_redirects=True, + timeout=_HTTP_TIMEOUT_SECONDS) + status_code = response.status_code + response.close() + if status_code in (403, 405): + with session.get(url, stream=True, + timeout=_HTTP_TIMEOUT_SECONDS) as get_response: + status_code = get_response.status_code + if status_code == 200: logging.info(f"URL {url} is reachable. Status code: {status_code}") - return status_code == 200 + return True + logging.warning( + f"URL {url} returned non-200 status code: {status_code}") + return False except Exception as e: - logging.fatal(f"Failed to reach URL {url}. Details: {e}") + logging.warning(f"Failed to reach URL {url}. Details: {e}") return False + finally: + if own_session: + session.close() + + +def scrape_epa_trends_urls(session: requests.Session = None) -> list: + """ + Scrapes the EPA Air Pollutant Emissions Trends Data page for the current + National Tier 1 CAPS and State Tier 1 CAPS .xlsx download URLs. + + Args: + session (requests.Session, optional): HTTP session to reuse. + + Returns: + list: Discovered [national_xlsx_url, state_xlsx_url], or empty list. + """ + own_session = session is None + if own_session: + session = _create_http_session() + try: + logging.info( + f"Discovering latest .xlsx URLs via GET: {EPA_TRENDS_PAGE_URL}") + response = session.get(EPA_TRENDS_PAGE_URL, + timeout=_HTTP_TIMEOUT_SECONDS) + response.raise_for_status() + html = response.text + + hrefs = re.findall(r'href=["\']([^"\']+\.xlsx(?:\?[^"\']*)?)["\']', + html, re.IGNORECASE) + national_url = None + state_url = None + for href in hrefs: + full_url = urllib.parse.urljoin(EPA_TRENDS_PAGE_URL, href) + filename = os.path.basename( + urllib.parse.urlsplit(full_url).path).lower() + if not national_url and re.match(r"^national_tier1_caps.*\.xlsx$", + filename): + national_url = full_url + elif not state_url and re.match(r"^state_tier1.*ktons\.xlsx$", + filename): + state_url = full_url + + discovered = [u for u in (national_url, state_url) if u] + if len(discovered) == 2: + logging.info(f"Discovered current EPA trend URLs: {discovered}") + return discovered + logging.warning(f"Incomplete URL discovery from {EPA_TRENDS_PAGE_URL}; " + f"expected 2 URLs, found: {discovered}") + return [] + except Exception as e: + logging.warning( + f"Failed to scrape EPA trends page {EPA_TRENDS_PAGE_URL}. " + f"Details: {e}") + return [] + finally: + if own_session: + session.close() def parse_gcs_path(full_path: str) -> tuple: @@ -95,13 +196,50 @@ def read_config_from_gcs(bucket_name: str, config_path: str) -> list: ] return urls except Exception as e: - logging.fatal( + logging.warning( f"Failed to read configuration file from GCS. Details: {e}") return [] - except Exception as e: - logging.fatal( - f"Failed to read configuration file from GCS. Details: {e}") + + +def resolve_download_urls(config_urls: list, + session: requests.Session = None) -> list: + """ + Verifies configured URLs and falls back to scraping the EPA trends page + if any configured URL is missing or returns non-200 (e.g., HTTP 404 after + an EPA release update). + """ + own_session = session is None + if own_session: + session = _create_http_session() + try: + if (config_urls and len(config_urls) >= 2 and + all(check_url_status(u, session=session) for u in config_urls)): + return config_urls + + logging.warning( + "Configured GCS URLs are missing or unreachable; falling back to " + "live EPA trends page discovery.") + scraped_urls = scrape_epa_trends_urls(session=session) + if (len(scraped_urls) == 2 and all( + check_url_status(u, session=session) for u in scraped_urls)): + return scraped_urls + return [] + finally: + if own_session: + session.close() + + +def _clean_download_directory(download_dir: str) -> None: + """Removes stale Excel or temporary files from previous runs.""" + if not os.path.exists(download_dir): + return + for existing_file in sorted(os.listdir(download_dir)): + if existing_file.lower().endswith(('.xlsx', '.xls', '.tmp')): + stale_path = os.path.join(download_dir, existing_file) + if os.path.isfile(stale_path): + os.remove(stale_path) + logging.info(f"Removed stale file: {stale_path}") def download_files(urls: list) -> None: @@ -114,49 +252,65 @@ def download_files(urls: list) -> None: Returns: None """ - if not os.path.exists(_DOWNLOAD_PATH): - os.mkdir(_DOWNLOAD_PATH) - os.chdir(_DOWNLOAD_PATH) + os.makedirs(_DOWNLOAD_PATH, exist_ok=True) - for file_url in urls: - # Check URL status before proceeding - if not check_url_status(file_url): - logging.fatal(f"URL not reachable: {file_url}. Skipping download.") + session = _create_http_session() + try: + resolved_urls = resolve_download_urls(urls, session=session) + if not resolved_urls: + raise ValueError("No reachable URLs found to download.") - file_name = file_url.split("/")[-1] + _clean_download_directory(_DOWNLOAD_PATH) - try: + for file_url in resolved_urls: + file_name = os.path.basename(urllib.parse.urlsplit(file_url).path) file_path = os.path.join(_DOWNLOAD_PATH, file_name) - urllib.request.urlretrieve(file_url, file_path) - logging.info(f"Successfully downloaded: {file_name} to {file_path}") - except Exception as e: - logging.fatal( - f"Failed to download {file_name} from {file_url}. Details: {e}") + tmp_path = f"{file_path}.tmp" + + try: + logging.info(f"Downloading GET {file_url} to {file_path}") + with session.get(file_url, + stream=True, + timeout=_HTTP_TIMEOUT_SECONDS) as response: + response.raise_for_status() + with open(tmp_path, "wb") as f_out: + for chunk in response.iter_content(chunk_size=65536): + if chunk: + f_out.write(chunk) + file_size = os.path.getsize(tmp_path) + if file_size == 0: + raise IOError(f"Downloaded file is empty: {tmp_path}") + os.replace(tmp_path, file_path) + logging.info( + f"Successfully downloaded: {file_name} ({file_size} bytes) " + f"to {file_path}") + except Exception as e: + if os.path.exists(tmp_path): + os.remove(tmp_path) + logging.fatal( + f"Failed to download {file_name} from {file_url}. " + f"Details: {e}") + finally: + session.close() def main(argv): + del argv # Get the config path from the flags config_path_flag = _FLAGS.config_path if not config_path_flag: - logging.fatal( + raise ValueError( "Please provide the --config_path flag with a valid GCS path.") - return try: bucket_name, config_path = parse_gcs_path(config_path_flag) except ValueError as e: logging.fatal(f"Invalid GCS path: {e}") - return - # Read URLs from the GCS-hosted config file + # Read URLs from the GCS-hosted config file (with live EPA page fallback) urls = read_config_from_gcs(bucket_name, config_path) - - # Proceed to download files - if urls: - download_files(urls) - else: - logging.fatal("No URLs to download. Exiting program.") + download_files(urls) if __name__ == '__main__': diff --git a/scripts/us_epa/air_emissions_inventory/national_state/manifest.json b/scripts/us_epa/air_emissions_inventory/national_state/manifest.json index 110e3a3785..0f8bc50816 100644 --- a/scripts/us_epa/air_emissions_inventory/national_state/manifest.json +++ b/scripts/us_epa/air_emissions_inventory/national_state/manifest.json @@ -18,9 +18,11 @@ } ], "cron_schedule": "0 10 1,15 * *", + "validation_config_file": "validation_config.json", "source_files": [ - "input_files/*" + "input_files/*", + "validation_config.json" ] } ] -} \ No newline at end of file +} diff --git a/scripts/us_epa/air_emissions_inventory/national_state/process.py b/scripts/us_epa/air_emissions_inventory/national_state/process.py index fc78e92d60..7e0e09a402 100644 --- a/scripts/us_epa/air_emissions_inventory/national_state/process.py +++ b/scripts/us_epa/air_emissions_inventory/national_state/process.py @@ -30,8 +30,8 @@ # Add the path to the 'util' directory to sys.path sys.path.insert( 0, - os.path.abspath( - os.path.join(os.path.dirname(__file__), '../../../../../data/util'))) + os.path.abspath(os.path.join(os.path.dirname(__file__), + '../../../../util'))) # importing from util from statvar_dcid_generator import get_statvar_dcid @@ -106,9 +106,7 @@ def _add_sv_and_mcf_column_to_final_df(self): } sv_list = pd.unique(self._final_df['SV']) - sv_list = [ - item for item in sv_list if not (pd.isnull(item)) == True - ] + sv_list = [item for item in sv_list if not pd.isnull(item)] logging.info("===========sv List==========" + str(sv_list)) sv_list.sort() @@ -131,9 +129,9 @@ def _add_sv_and_mcf_column_to_final_df(self): # Reconstruct the generated_sv without 'Total' if it was removed cleaned_generated_sv = '_'.join(generated_sv_parts) - # Skip emissionSource in the mcf if 'Total' is in the generated_sv + # Skip emissionSource in the mcf if 'Total' is in emissionSource if 'Total' in sv_property[0]: - source = '' # Don't add emissionSource if 'Total' is part of emissionSource + source = '' else: source = '\nemissionSource: dcs:' + sv_property[0] @@ -166,7 +164,21 @@ def _state_emissions(self, file_path: str) -> pd.DataFrame: df (pd.DataFrame): provides the cleaned df as output """ try: - df = pd.read_excel(file_path, SHEET_STATE, skiprows=1, header=0) + with pd.ExcelFile(file_path) as excel_file: + sheet_name = SHEET_STATE + if sheet_name not in excel_file.sheet_names: + for candidate in excel_file.sheet_names: + if (candidate.replace( + '_', + ' ').strip().lower() == SHEET_STATE.lower()): + sheet_name = candidate + break + df = pd.read_excel(excel_file, sheet_name, skiprows=0, header=0) + if 'State FIPS' not in df.columns: + df = pd.read_excel(excel_file, + sheet_name, + skiprows=1, + header=0) # Adding geoId/ and making the State FIPS code of 2 numbers # Eg - 1 -> geoId/01 df['geo_Id'] = [f'{x:02}' for x in df['State FIPS']] @@ -184,10 +196,18 @@ def _state_emissions(self, file_path: str) -> pd.DataFrame: df = df.melt(id_vars=['SV_TEMP', 'geo_Id'], var_name='year', value_name='observation') - df['year'] = (df['year'].str[-2:]).astype(int) + years = pd.to_numeric(df['year'].astype(str).str.replace( + 'emissions', '', case=False).str.strip(), + errors='coerce') + if years.isna().any(): + invalid_cols = df.loc[years.isna(), 'year'].unique().tolist() + raise ValueError( + f"Unparseable year columns in state file: {invalid_cols}") + years = years.astype(int) # As 2001 is available as 01 and 1999 is available as 99, # Putting a logic to change it to proper year - df['year'] = df['year'] + np.where(df['year'] >= 90, 1900, 2000) + df['year'] = np.where((years >= 90) & (years < 100), years + 1900, + np.where(years < 90, years + 2000, years)) # Making copy and using group by to get Aggregated Values. df_agg = self.aggregate_columns( df, ['PrescribedFire', 'Wildfire', 'Miscellaneous'], @@ -195,7 +215,7 @@ def _state_emissions(self, file_path: str) -> pd.DataFrame: df['Measurement_Method'] = 'EPA_NationalEmissionInventory' df = pd.concat([df, df_agg]) return df - except FileNotFoundError as e: + except FileNotFoundError: logging.fatal(f"File not found: {file_path}") except Exception as e: logging.fatal(f"Missing column in state emissions file: {e}") @@ -215,8 +235,9 @@ def _national_emissions(self, file_path: str) -> pd.DataFrame: final_df = pd.DataFrame() # Reading different sheets of Excel one at a time for sheet in SHEETS_NATIONAL: - # Number of rows to skip vary by sheet, inserting if-else for the same. - skiphead = SKIPHEAD_AMMONIA_NATIONAL if sheet == 'NH3' else SKIPHEAD_OTHERS_NATIONAL + # Number of rows to skip vary by sheet + skiphead = (SKIPHEAD_AMMONIA_NATIONAL + if sheet == 'NH3' else SKIPHEAD_OTHERS_NATIONAL) skipfoot = 1 df = pd.read_excel(file_path, sheet, @@ -225,12 +246,12 @@ def _national_emissions(self, file_path: str) -> pd.DataFrame: header=0) # NA values are present as 'NA ' and also as '', removing 'NA ' # to process the column. - df = df.replace('NA ', np.nan) + df = df.replace(['NA ', 'NA', 'N/A'], np.nan) # Dropping rows with only Null Values. df = df.dropna(how='all') - # Clean the 'Source Category' column to handle any formatting issues + # Clean the 'Source Category' column to handle formatting issues df['Source Category'] = df['Source Category'].str.strip() - # Drop only specific unwanted rows, ensuring important ones are retained + # Drop only specific unwanted rows df = df.drop( df[(df['Source Category'] == 'Miscellaneous') | (df['Source Category'] == 'Total without wildfires') | @@ -255,7 +276,7 @@ def _national_emissions(self, file_path: str) -> pd.DataFrame: final_df['geo_Id'] = 'country/USA' final_df['Measurement_Method'] = 'EPA_NationalEmissionInventory' return final_df - except FileNotFoundError as e: + except FileNotFoundError: logging.fatal(f"File not found: {file_path}") except Exception as e: logging.fatal(f"Missing column in national emissions file: {e}") @@ -283,21 +304,20 @@ def _parse_source_files(self): } for file_path in self._input_files: # Extract filename without extension - file_name = file_path.split("/")[-1][:-5] + file_name = os.path.splitext(os.path.basename(file_path))[0] logging.info(f"Processing file: {file_name}") # Initialize a variable to store the matched method matched_method = None # Check if the filename matches any pattern for pattern, method in source_file_to_method_mapping.items(): - if re.match(pattern, file_name): + if re.match(pattern, file_name, re.IGNORECASE): matched_method = method break # Exit the loop as soon as a match is found - # If no match is found, log the error and break the loop + # If no match is found, log the fatal error if not matched_method: logging.fatal( - f"No parsing method found for file: {file_name}. Stopping execution." - ) - break + f"No parsing method found for file: {file_name}. " + "Stopping execution.") # Process the file using the matched method df = matched_method(file_path) @@ -306,25 +326,38 @@ def _parse_source_files(self): self._final_df = self._final_df.sort_values( by=['geo_Id', 'year', 'SV_TEMP', 'observation']) - self._final_df['observation'].replace('', np.nan, inplace=True) - self._final_df.dropna(subset=['observation'], inplace=True) - self._final_df.dropna(subset=['SV_TEMP'], inplace=True) + self._final_df['observation'] = self._final_df[ + 'observation'].replace('', np.nan) + self._final_df['observation'] = pd.to_numeric( + self._final_df['observation'], errors='coerce') + self._final_df = self._final_df.dropna(subset=['observation']) + self._final_df = self._final_df.dropna(subset=['SV_TEMP']) self._final_df['observation'] = self._final_df[ 'observation'].astype(float) * _SCALING_FACTOR self._add_sv_and_mcf_column_to_final_df() except Exception as e: logging.fatal(f"Error parsing source files: {e}") - raise def main(_): input_path = FLAGS.input_path logging.info(f"Input path: {input_path}") try: - ip_files = os.listdir(input_path) + if not os.path.exists(input_path): + raise FileNotFoundError(f"Input path does not exist: {input_path}") + if not os.listdir(input_path): + raise FileNotFoundError(f"Input directory is empty: {input_path}") + ip_files = [ + file for file in sorted(os.listdir(input_path)) + if file.lower().endswith(('.xlsx', '.xls')) + ] logging.info(f"Found {len(ip_files)} files in input path.") + if not ip_files: + raise FileNotFoundError( + f"No Excel files (.xlsx/.xls) found in input path: {input_path}" + ) - ip_files = [input_path + os.sep + file for file in ip_files] + ip_files = [os.path.join(input_path, file) for file in ip_files] output_file_path = os.path.join( os.path.dirname(os.path.abspath(__file__)), "output") # Defining Output Files @@ -357,12 +390,10 @@ def main(_): logging.info(f"TMCF file generated: {tmcf_path}") except Exception as e: logging.fatal(f"Error generating outputs: {e}") - raise - except Exception: + except Exception as e: logging.fatal( - f"There are no files in the input path: {input_path} to download.") - raise + f"Failed to read input files from {input_path}. Details: {e}") if __name__ == "__main__": diff --git a/scripts/us_epa/air_emissions_inventory/national_state/process_test.py b/scripts/us_epa/air_emissions_inventory/national_state/process_test.py index 6f4d8833ad..9eb1a33a69 100644 --- a/scripts/us_epa/air_emissions_inventory/national_state/process_test.py +++ b/scripts/us_epa/air_emissions_inventory/national_state/process_test.py @@ -12,16 +12,24 @@ # See the License for the specific language governing permissions and # limitations under the License. """ -Script to automate the testing for EuroStat BMI (Body Mass Index) process script. +Script to automate the testing for US EPA Air Pollutant Emission Trends +(National and State) process script. """ import os import sys +import tempfile +import unittest +from unittest import mock +import pandas as pd -_COMMON_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__), '...')) +_MODULE_DIR = os.path.dirname(os.path.abspath(__file__)) +_COMMON_PATH = os.path.abspath(os.path.join(_MODULE_DIR, '..')) +sys.path.insert(1, _MODULE_DIR) sys.path.insert(1, _COMMON_PATH) # pylint: disable=wrong-import-position +import download from process import USAirPollutionEmissionTrendsNationalAndState -from common.unitest_common_methods import CommonTestClass +from common.unitest_common_methods_test import CommonTestClass # pylint: enable=wrong-import-position @@ -29,3 +37,82 @@ class USAirPollutionEmissionTrendsNationalAndStateTest( CommonTestClass.CommonTestCases): _import_class = USAirPollutionEmissionTrendsNationalAndState _test_module_directory = os.path.dirname(__file__) + + def test_state_emissions_skiprows_zero_and_year_parsing(self): + """Verifies _state_emissions handles skiprows=0 ('State Trends') and 2/4-digit years.""" + with tempfile.TemporaryDirectory() as tmp_dir: + xlsx_path = os.path.join(tmp_dir, "state_tier1_test.xlsx") + sample_df = pd.DataFrame({ + "State FIPS": [1], + "State": ["AL"], + "Tier 1 Code": [1], + "Tier 1 Description": ["Fuel Comb. Elec. Util."], + "Pollutant": ["CO"], + "emissions99": [1.5], + "emissions01": [2.0], + "emissions2025": [3.25], + }) + with pd.ExcelWriter(xlsx_path) as writer: + sample_df.to_excel(writer, + sheet_name="State Trends", + index=False) + + loader = USAirPollutionEmissionTrendsNationalAndState([xlsx_path]) + result_df = loader._state_emissions(xlsx_path) + raw_df = result_df[result_df["Measurement_Method"] == + "EPA_NationalEmissionInventory"] + years_map = dict(zip(raw_df["year"], raw_df["observation"])) + self.assertEqual(years_map, {1999: 1.5, 2001: 2.0, 2025: 3.25}) + self.assertTrue((raw_df["geo_Id"] == "geoId/01").all()) + + def test_scrape_and_resolve_download_urls(self): + """Verifies scrape_epa_trends_urls and resolve_download_urls fallback behavior.""" + sample_html = ''' + + National + State + + ''' + mock_session = mock.MagicMock() + mock_get_resp = mock.MagicMock() + mock_get_resp.status_code = 200 + mock_get_resp.text = sample_html + mock_session.get.return_value = mock_get_resp + + discovered = download.scrape_epa_trends_urls(session=mock_session) + self.assertEqual( + discovered, + [ + "https://www.epa.gov/system/files/other-files/2026-09/national_tier1_caps_04sep2026.xlsx?ver=1", + "https://www.epa.gov/system/files/other-files/2026-09/state_tier1_1990_2025_04sep2026_ktons.xlsx", + ], + ) + + # Simulate stale GCS URLs (404) falling back to scraped URLs (200) + def side_effect_head(url, **_kwargs): + resp = mock.MagicMock() + resp.status_code = 404 if "2025-04" in url else 200 + return resp + + mock_session.head.side_effect = side_effect_head + stale_urls = [ + "https://www.epa.gov/system/files/other-files/2025-04/national_tier1_caps_21feb2025.xlsx", + "https://www.epa.gov/system/files/other-files/2025-04/state_tier1_21feb2005_ktons.xlsx", + ] + resolved = download.resolve_download_urls(stale_urls, + session=mock_session) + self.assertEqual(resolved, discovered) + + def test_clean_download_directory(self): + """Verifies _clean_download_directory removes stale .xlsx/.tmp files only.""" + with tempfile.TemporaryDirectory() as tmp_dir: + stale_xlsx = os.path.join(tmp_dir, "old.xlsx") + stale_tmp = os.path.join(tmp_dir, "partial.xlsx.tmp") + keep_txt = os.path.join(tmp_dir, "notes.txt") + for p in (stale_xlsx, stale_tmp, keep_txt): + with open(p, "w", encoding="utf-8") as f: + f.write("test") + download._clean_download_directory(tmp_dir) + self.assertFalse(os.path.exists(stale_xlsx)) + self.assertFalse(os.path.exists(stale_tmp)) + self.assertTrue(os.path.exists(keep_txt)) diff --git a/scripts/us_epa/air_emissions_inventory/national_state/validation_config.json b/scripts/us_epa/air_emissions_inventory/national_state/validation_config.json new file mode 100644 index 0000000000..f7e10f1c7e --- /dev/null +++ b/scripts/us_epa/air_emissions_inventory/national_state/validation_config.json @@ -0,0 +1,28 @@ +{ + "schema_version": "1.0", + "rules": [ + { + "rule_id": "check_deleted_records_percent", + "description": "Checks that the percentage of deleted records does not exceed the 0.1% threshold.", + "validator": "DELETED_RECORDS_PERCENT", + "params": { + "threshold": 0.1 + } + }, + { + "rule_id": "check_max_date_consistent", + "description": "Checks that the latest date (MaxDate) is consistent across all StatVars.", + "validator": "MAX_DATE_CONSISTENT", + "params": {} + }, + { + "rule_id": "check_max_date_freshness", + "description": "Verifies that MaxDate meets the expected freshness threshold (>= 2025) across all StatVars.", + "validator": "SQL_VALIDATOR", + "params": { + "query": "SELECT StatVar, MaxDate FROM stats", + "condition": "MaxDate >= '2025'" + } + } + ] +}