Fix for epa-airpollutantemission-level1

[shourya116]

Fixes an issue where 767,314 observations across the 2017 and 2020 observation periods (specifically nonpoint/area sources) were silently dropped during the Cloud Batch import validation pipeline for EPA_AirPollutantEmission_Level1.

Root Cause

1. Pandas 2.x Type Mismatch: The raw 2017 and 2020 nonpoint CSVs contained empty emissions type code columns, which Pandas loaded as float64. The check 'point' in file_path evaluated to True for ..._nonpoint/... files, triggering df.loc[:, 'emissions type code'] = ''. In Pandas 2.x, assigning a string to a float64 series via .loc raises TypeError: Invalid value '' for dtype 'float64'.
2. Broad Exception Swallowing & Unconsumed Thread Pool: _national_emissions() caught the TypeError and returned an empty DataFrame, silently dropping the entire nonpoint dataset (767k rows) during consolidation.

Key Changes

• process.py:
  • Corrected file routing to distinguish point vs. nonpoint files using 'point_' in os.path.basename(file_path) and direct column assignment (df['emissions type code'] = '').
  • Added support for both 2014 tribal data schemas in _regularize_columns.
  • Ensured worker threads (_national_emissions, _process_file) log via logging.exception and re-raise so exceptions bubble cleanly to list(executor.map(...)).
  • Removed unreachable code (raise, sys.exit(1)) after terminating logging.fatal calls, and removed commented-out imports.
• process_test.py & __init__.py:
  • Added module initialization and path resolution for CI (./run_tests.sh -p scripts/us_epa/national_emissions_inventory).
  • Added 4 targeted regression tests in RegularizeColumnsTest covering 2017/2020 nonpoint, point, and 2014 tribal schema variants (5 unit tests passing).
• manifest.json:
  • Scaled VM limits to 32 CPUs / 512 GiB RAM / 300 GB disk (n2-highmem-64) to resolve historical OOM failures.
  • Registered node_mcf (resolving 8 missing reference warnings).
  • Added validation_config.json to source_files and removed trailing blank lines.
• validation_config.json:
  • Configured check_deleted_records_percent with a 0.1% threshold and explicit deletion rationale.
  • Added active check_date_freshness (SQL_VALIDATOR: MAX(MaxDate) >= 2020 AND MIN(MaxDate) >= 2008).

Verification & Artifacts

• Latest Cloud Batch Run: epa-airpollutantemission-level1-shouryasingh-20260908-184507 (SUCCEEDED, all 3,041,170 observations processed, all validations passed).
• GCS Output: gs://datcom-import-test/scripts/us_epa/national_emissions_inventory/EPA_AirPollutantEmission_Level1/2026_09_08T11_47_48_764425_07_00
  • Cloud Storage Console
• Review & Postmortem:
  • CRA Document: https://paste.googleplex.com/6620650563829760
  • Postmortem Report: https://paste.googleplex.com/5957668808818688

[google-cla]

Thanks for your pull request! It looks like this may be your first contribution to a Google open source project. Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

View this failed invocation of the CLA check for more information.

For the most up to date status, view the checks section at the bottom of the pull request.

[gemini-code-assist]

Code Review

This pull request increases the resource limits (CPU, memory, and disk) in the manifest configuration and refactors pandas DataFrame operations in process.py to avoid using .loc for simple column assignments and to replace deprecated inplace=True usage. Feedback is provided to remove a redundant .replace('', np.nan) call on the observation column, as empty strings are already converted to NaN earlier in the processing pipeline.

[abhishekjaisw]

Review scope

• Target: PR #2148 (4fc4609467faf6425589b232ce6e674c6677ec2d)
• Reviewed:
  • scripts/us_epa/national_emissions_inventory/manifest.json
  • scripts/us_epa/national_emissions_inventory/process.py
  • scripts/us_epa/national_emissions_inventory/validation_config.json
• Skipped: None

Summary of Findings

1. [P1] CI Test Suite Failure (data-pull-request-py) & Missing Regression Tests for 2017/2020 Nonpoint & Tribal Fixes (process.py:144, process_test.py)
   • Running ./run_tests.sh -p scripts/us_epa/national_emissions_inventory (executed by GitHub CI data-pull-request-py) fails with ImportError: attempted relative import with no known parent package at process_test.py:20 (from .process import *) because scripts/us_epa/national_emissions_inventory/ lacks an __init__.py.
   • Furthermore, PR #2148 modifies _regularize_columns to fix the silent drop of 767,314 observations across 2017/2020 nonpoint, point_, and 2014 tribes files, but adds zero unit tests or test fixtures for these branches.
2. [P2] Missing Date Freshness Validation Rule in validation_config.json (Outdated Branch / Already Fixed DuckDB Differ Issue) (validation_config.json:9)
   • validation_config.json omits date freshness validation because SQL_VALIDATOR previously failed on empty differ_df (Need a DataFrame with at least one column). However, this framework bug was already fixed on master in commits 3315fc4b (Aug 14, 2026) and 4d856505 (Aug 31, 2026) in tools/import_validation/validator.py:62-63 and runner.py:194. Rebasing PR #2148 onto master enables re-adding the SQL_VALIDATOR date freshness check (MAX(MaxDate) >= 2020 and MIN(MaxDate) >= 2014).
3. [P2] Exception Tracebacks Discarded in logging.fatal Catch Blocks (process.py:239)
   • Passing only {e} to logging.fatal(...) discards the Python stack trace (exc_info), obscuring line numbers and call stacks during Cloud Batch failures.
4. [P3] Operational Configs in source_files & Google CLA Check (manifest.json:25)
   • Consider adding "validation_config.json" to "source_files" in manifest.json so validation configs are archived in GCS. Additionally, please resolve the failing cla/google check on GitHub before merge.

Positive findings

• scripts/us_epa/national_emissions_inventory/process.py:139-155 - Precise filename/path disambiguation and direct column assignment ✓
  • Finding: Good - Replaced broad "point" in file_path checks with "point_" in os.path.basename(file_path) or "facility_process" in file_path and explicit "nonpoint" branches, avoiding substring collisions with nonpoint files, and replaced .loc[:, col] = "" with direct column assignment (df["emissions type code"] = "") to prevent Pandas 2.x float64 dtype errors.
• scripts/us_epa/national_emissions_inventory/process.py:314 - Eager evaluation of ThreadPoolExecutor.map ✓
  • Finding: Good - Wrapped executor.map(...) in list(...) so worker thread exceptions and SystemExit from logging.fatal propagate immediately to the main thread rather than silently dropping failed files.
• scripts/us_epa/national_emissions_inventory/manifest.json:21-29 - Scaled Cloud Batch compute resources and registered node_mcf ✓
  • Finding: Good - Increased memory to 512 GiB and CPU to 32 to eliminate OOM kills (exit code 50002) during multi-year DataFrame concatenation, and explicitly wired node_mcf and validation_config_file.

Coverage

File	Status	Result
scripts/us_epa/national_emissions_inventory/manifest.json	Reviewed	One P3 finding
scripts/us_epa/national_emissions_inventory/process.py	Reviewed	One P1 finding, one P2 finding
scripts/us_epa/national_emissions_inventory/validation_config.json	Reviewed	One P2 finding

Verification and limitations

• Checks run:
  • ./run_tests.sh -p scripts/us_epa/national_emissions_inventory (Reproduced CI failure: ImportError: attempted relative import with no known parent package in process_test.py)
  • PYTHONPATH=. .env/bin/python3 -m unittest scripts/us_epa/national_emissions_inventory/process_test.py (Passed: 1 test in 0.575s)
  • Inspected GCS prod (2025_12_31T16_03_29_040514_08_00) and dev (2026_09_06T07_23_45_163580_07_00) summary_report.csv and validation_output.csv.
• Checks not run: None
• Limitations: None

[shourya116]

Changed Implemented as per feedback

[abhishekjaisw]

Code Review Summary for PR #2148 (EPA_AirPollutantEmission_Level1)

Root Cause & Data Transformation Fixes (Verified):

• Fix in _regularize_columns ('point_' in os.path.basename(file_path) and direct column assignment df['emissions type code'] = '') resolves the Pandas 2.x float64 TypeError that dropped 2017/2020 nonpoint observations.
• RegularizeColumnsTest in process_test.py (5 unit tests passing) covers nonpoint, point, and 2014 tribal schema variants.
• "node_mcf": "gcs_output/output_files/national_emissions.mcf" in manifest.json is valid and should be kept (unlike PR Added validation configuration for EPA_airqualityindex #2147), as it resolves 8 Existence_MissingReference_variableMeasured warnings present in the baseline run (report.json).
• Cloud Batch run (2026_09_07T06_25_07_719529_07_00) ran after the latest commit (15b2958b) and confirmed check_date_freshness passed.

Action Items Before Merge:

1. [P2] Sanitize PR Description URLs (from PR Added validation configuration for EPA_airqualityindex #2147 review): Replace internal https://pantheon.corp.google.com/... links (with corp query params) in the PR description with gs://datcom-import-test/scripts/us_epa/national_emissions_inventory/EPA_AirPollutantEmission_Level1/2026_09_07T06_25_07_719529_07_00 or clean https://console.cloud.google.com/... links.
2. [P2] process.py (logging.fatal + raise): Remove unreachable raise / sys.exit(1) after logging.fatal() across 6 exception blocks (lines 242–246, 263–267, 337–341, 344–345, 432–436, 458–461), avoid calling logging.fatal() inside ThreadPoolExecutor worker threads (_national_emissions, _process_file) so exceptions propagate cleanly via list(executor.map(...)), and remove commented-out imports (# import shutil, # import tempfile).
3. [P2] validation_config.json (DELETED_RECORDS_PERCENT): Update the description of check_deleted_records_percent to explicitly state the 0.1% threshold and deletion rationale (and fix the typo in Postmortem Section 5.3 from (10%) to (0.1%)).
4. [P2] CRA Paste & Postmortem Sync: Update both documents to reference head commit 15b2958b68aa936fe242d820ee50f3f01661d8d1, reflect that check_date_freshness (SQL_VALIDATOR) is active and passing, and update the unit test count in Section 5.4 to 5 tests.
5. [P3] Formatting & CI: Remove the 3 extra trailing blank lines at EOF in manifest.json, resolve the cla/google check, and trigger /gcbrun.

[shourya116]

All the latest review comments have been resolved and verified. Both the CRA review document and Postmortem report have also been updated and synced live.

Summary of Actions Taken

1. [P2] process.py Exception Handling & Cleanup:

   • Worker Thread Exception Propagation: Avoided calling logging.fatal() inside ThreadPoolExecutor worker threads (_national_emissions and _process_file). Errors are logged using logging.exception() and re-raised so exceptions propagate cleanly to list(executor.map(...)) on the main thread.
   • Unreachable Code Removal: Removed unreachable raise and sys.exit(1) statements after terminating logging.fatal() calls across 6 exception blocks (intermediate file reading loop, empty DataFrame check, input file discovery, and main execution handler).
   • Import Cleanup: Removed commented-out imports (# import shutil, # import tempfile).

2. [P2] validation_config.json Description Update:

   • Updated check_deleted_records_percent description to explicitly state the 0.1% threshold and deletion rationale:

     {
         "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
         }
     }

   • Corrected the typo in Postmortem Section 5.3 from (10%) to (0.1%).

3. [P3] manifest.json Formatting:

   • Removed the 3 extra trailing blank lines at EOF.

4. [P2] CRA Paste & Postmortem Sync:

   • Head Commit Sync: Updated both documents to reference head commit 242f22800dcf60581dc309111b5388ddb77f974e.
   • Unit Tests: Updated unit test counts in both documents to reflect 5 passing tests (0.635s, OK).
   • Date Freshness Validation: Confirmed that check_date_freshness (SQL_VALIDATOR) is active and passing following the clean rebase with master's DuckDB empty-differ fix.
   • Updated Paste Links:
     • CRA Paste: https://paste.googleplex.com/6620650563829760
     • Postmortem Report: https://paste.googleplex.com/5957668808818688

5. Cloud Batch Verification Run (SUCCEEDED):

   • Ran end-to-end verification job epa-airpollutantemission-level1-shouryasingh-20260908-184507 on us-west4 using n2-highmem-64 (32 CPUs / 512 GiB RAM / 300 GB disk).
   • Status: SUCCEEDED (execution duration: 7,932s / ~2.2 hrs).
   • All 3,041,170 rows were successfully processed (all 767,314 missing nonpoint observations recovered).
   • All 5 validations passed:
     • check_deleted_records_percent: PASSED (0 deleted records, 0.0% vs 0.1% threshold)
     • check_empty_import: PASSED (3,041,634 nodes, 3,041,170 rows)
     • check_missing_refs_count: PASSED (0 missing references)
     • check_lint_error_count: PASSED (0 lint errors)
     • check_date_freshness: PASSED (MAX(MaxDate) >= 2020 AND MIN(MaxDate) >= 2008)
   • Cloud Batch Job: https://console.cloud.google.com/batch/jobsDetail/regions/us-west4/jobs/epa-airpollutantemission-level1-shouryasingh-20260908-184507/details?project=datcom-infosys-dev
   • GCS Output: gs://datcom-import-test/scripts/us_epa/national_emissions_inventory/EPA_AirPollutantEmission_Level1/2026_09_08T11_47_48_764425_07_00
   • GCS Console: https://console.cloud.google.com/storage/browser/datcom-import-test/scripts/us_epa/national_emissions_inventory/EPA_AirPollutantEmission_Level1/2026_09_08T11_47_48_764425_07_00?project=datcom-infosys-dev

6. Sanitized PR Description