From 4251279830f6257c2a85e974c80948be4c9c2d35 Mon Sep 17 00:00:00 2001 From: Nathan Date: Tue, 28 Apr 2026 21:05:14 -0400 Subject: [PATCH 1/6] Switch precipitation, CSO, and budget fetches to incremental updates - Precipitation: re-fetches only the most recent cached year onward (~26 fewer ACIS requests/run) - CSO: uses IncidentFromDate API param to fetch only records >= max cached incidentDate, boundary date inclusive to catch late-arriving records (~130 fewer paginated requests/run) - Budget CTHRU: skips all API calls if cache already covers the current MA fiscal year Co-Authored-By: Claude Sonnet 4.6 --- get_data/get_MA_precipitation.py | 33 ++++++++++++++--- get_data/get_budget_CTHRU.py | 16 ++++++++ get_data/get_eea_dp_cso.py | 63 ++++++++++++++++++++++++++------ 3 files changed, 96 insertions(+), 16 deletions(-) diff --git a/get_data/get_MA_precipitation.py b/get_data/get_MA_precipitation.py index b93254a3..ced4ece7 100644 --- a/get_data/get_MA_precipitation.py +++ b/get_data/get_MA_precipitation.py @@ -76,10 +76,26 @@ def fetch_daily_precip_year(year: int) -> pd.DataFrame: def main(): current_year = datetime.datetime.now().year - all_years = list(range(START_YEAR, current_year + 1)) + out_path = '../docs/data/MA_precipitation_daily.csv' + + # Load cached data to determine the earliest year that needs re-fetching. + # We always re-fetch the most recently cached year because it may be incomplete + # (partial calendar year at the time of the previous run). + try: + existing = pd.read_csv(out_path) + existing['date'] = pd.to_datetime(existing['date']) + max_cached_year = existing['date'].dt.year.max() + fetch_from_year = max_cached_year # re-fetch last year in case it was partial + print(f' Found cached data through {max_cached_year}; fetching {fetch_from_year}–{current_year}') + except FileNotFoundError: + existing = None + fetch_from_year = START_YEAR + print(f' No cache found; fetching {START_YEAR}–{current_year}') + + fetch_years = list(range(fetch_from_year, current_year + 1)) rows = [] - for year in all_years: + for year in fetch_years: print(f' Fetching {year}...', end=' ', flush=True) try: df_year = fetch_daily_precip_year(year) @@ -88,11 +104,18 @@ def main(): except Exception as e: print(f'FAILED: {e}') - df = pd.concat(rows, ignore_index=True) + new_data = pd.concat(rows, ignore_index=True) + + if existing is not None: + # Drop cached rows for years being re-fetched, then append fresh data. + retained = existing[existing['date'].dt.year < fetch_from_year].copy() + retained['date'] = retained['date'].dt.strftime('%Y-%m-%d') + df = pd.concat([retained, new_data], ignore_index=True) + else: + df = new_data - out_path = '../docs/data/MA_precipitation_daily.csv' df.to_csv(out_path, index=False) - print(f'\nWrote {len(df)} rows to {out_path}') + print(f'\nWrote {len(df)} rows to {out_path} ({len(new_data)} newly fetched)') with open('../docs/data/ts_update_MA_precipitation.yml', 'w') as f: f.write('updated: ' + str(datetime.datetime.now()).split('.')[0] + '\n') diff --git a/get_data/get_budget_CTHRU.py b/get_data/get_budget_CTHRU.py index 572225cd..eeeb8d61 100644 --- a/get_data/get_budget_CTHRU.py +++ b/get_data/get_budget_CTHRU.py @@ -54,6 +54,22 @@ def fetch_agency_budget(accounts: list) -> pd.DataFrame: if __name__ == '__main__': + # Skip if the cache already covers the current fiscal year. + # MA fiscal year runs July–June: FY2026 = Jul 2025–Jun 2026. + _now = datetime.datetime.now() + _current_fy = _now.year if _now.month < 7 else _now.year + 1 + try: + _cached = pd.read_csv('../docs/data/MassBudget_environmental_summary.csv') + _max_cached_fy = int(_cached['Year'].max()) + if _max_cached_fy >= _current_fy: + print(f'Budget data current through FY{_max_cached_fy}; skipping CTHRU fetch.') + with open('../docs/data/ts_update_MassBudget_environmental.yml', 'w') as _f: + _f.write('updated: ' + str(_now).split('.')[0] + '\n') + import sys; sys.exit(0) + print(f'Cache covers through FY{_max_cached_fy}; fetching FY{_max_cached_fy + 1}–FY{_current_fy}') + except FileNotFoundError: + print('No cache found; running full CTHRU fetch') + # Load SSA AWI from CSV for inflation adjustment (2024 base year) # Read from CSV instead of DB since this runs before assemble_db.py creates tables awi = pd.read_csv('../docs/data/SSAWages.csv').set_index('Year') diff --git a/get_data/get_eea_dp_cso.py b/get_data/get_eea_dp_cso.py index 5b176c82..b3245ff0 100644 --- a/get_data/get_eea_dp_cso.py +++ b/get_data/get_eea_dp_cso.py @@ -63,24 +63,32 @@ def _query_page(page: int, query_params: Optional[dict[str, str]]=None) -> Optio else: return None -def run_query() -> pd.DataFrame: - """Run a full query, paging through results and returning a combined DataFrame. +def run_query(from_date: Optional[str] = None) -> pd.DataFrame: + """Run a query, paging through results and returning a combined DataFrame. + + from_date: optional MM/DD/YYYY string; if provided, fetches only records + on or after that date (IncidentFromDate API param). """ - print('Running full query') + if from_date: + print(f'Running incremental query from {from_date}') + else: + print('Running full query') + query_params: dict[str, str] = {} + if from_date: + query_params['IncidentFromDate'] = from_date page = 1 # CSOAPI is 1-indexed result_dfs = [] while True: - df = _query_page(page) + df = _query_page(page, query_params) if df is None: break result_dfs.append(df) page += 1 return pd.concat(result_dfs) -def get_data() -> pd.DataFrame: - """Query data from the data portal API and do any necessary post processing. - """ - df = run_query() + +def _parse_dates(df: pd.DataFrame) -> pd.DataFrame: + """Parse date columns and add derived Year column.""" df['incidentDate'] = pd.to_datetime(df['incidentDate'], format='ISO8601') df['submittedDate'] = pd.to_datetime(df['submittedDate'], format='ISO8601') # API already returns a lowercase 'year' column; drop it before adding 'Year' @@ -89,13 +97,46 @@ def get_data() -> pd.DataFrame: df['Year'] = df['incidentDate'].apply(lambda x: x.year) return df + +def get_data() -> pd.DataFrame: + """Fetch CSO data incrementally when a cached CSV exists. + + Loads the cached file, determines the latest incidentDate, and fetches only + records from that date onwards (inclusive, to catch any records that arrived + after the last pull). Rows from the boundary date are dropped from the cache + before appending so there are no duplicates. + """ + csv_path = '../docs/data/EEADP_CSO.csv' + from_date: Optional[str] = None + existing: Optional[pd.DataFrame] = None + + try: + existing = pd.read_csv(csv_path, index_col=0) + existing['incidentDate'] = pd.to_datetime(existing['incidentDate'], format='ISO8601') + max_date = existing['incidentDate'].max() + from_date = max_date.strftime('%m/%d/%Y') + # Drop cached rows on the boundary date — the API refetch will include them. + existing = existing[existing['incidentDate'].dt.date < max_date.date()].copy() + print(f' Cached data through {max_date.date()}; fetching from {from_date} (inclusive)') + except FileNotFoundError: + print(' No cache found; running full query') + + new_df = _parse_dates(run_query(from_date=from_date)) + + if existing is not None: + df = pd.concat([existing, new_df], ignore_index=True) + else: + df = new_df + + return df + + def write_data(df: pd.DataFrame): """Write data to a local table for integration with AMEND. """ - print('Writing out queries data') + print('Writing out queried data') df.to_csv('../docs/data/EEADP_CSO.csv', index=True) - ## Print a sample of the file as an example - df.sample(n=10).to_csv('../docs/data/EEADP_CSO_sample.csv', index=0) + df.sample(n=10).to_csv('../docs/data/EEADP_CSO_sample.csv', index=False) def main(): """Query and write all data. From 598642c17a46d203bd7cde1a3d38f1e014c7182e Mon Sep 17 00:00:00 2001 From: Nathan Date: Tue, 28 Apr 2026 21:18:46 -0400 Subject: [PATCH 2/6] Document why drinkingWater fetch is not incrementalized TotalCount is available cheaply but a sentinel only helps when nothing changed; new lab results arrive virtually every week so the skip would rarely trigger and the API has no date-range filter to fetch only new rows. Co-Authored-By: Claude Sonnet 4.6 --- get_data/get_EEA_data_portal.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/get_data/get_EEA_data_portal.py b/get_data/get_EEA_data_portal.py index df645f08..59c0c525 100644 --- a/get_data/get_EEA_data_portal.py +++ b/get_data/get_EEA_data_portal.py @@ -18,6 +18,28 @@ gs://openamend-data/EEADP_drinkingWater.csv — full drinking water table (GCS only) ../docs/data/EEADP_drinkingWater_annual.csv — annualized summary ../docs/data/ts_update_EEADP.yml — timestamp of last run + +Why drinkingWater is not incrementally fetched +---------------------------------------------- +The drinking water portal dataset is large (~200MB) and we do a full refresh of this every +run, instead of updating incrementally. Unfortunately, there is no good alternative because +the API does not support date filtering. + +The API response includes a TotalCount field (visible on a single lightweight probe request +`?_end=1&_start=0`), which could in principle serve as a skip sentinel: store the count, +compare on next run, and skip the full download+upload if unchanged. + +Two reasons this is not worth implementing: + +1. The sentinel only helps when nothing changed. If TotalCount increased the full ~200 MB + download and GCS upload still happens — there is no way to fetch only new rows because + the API is offset-based with no date-range filter parameter. + +2. Drinking water lab results are submitted by public water systems continuously throughout + the year (~3.8 M rows as of April 2026). TotalCount increases virtually every week, so + the sentinel would almost never trigger a skip in practice. + +The 200 MB weekly upload is effectively unavoidable unless the API gains date-range filtering. """ import os From 98d0f96f12a25a22870a1c0e673a0e2752e96528 Mon Sep 17 00:00:00 2001 From: Nathan Date: Tue, 28 Apr 2026 21:25:43 -0400 Subject: [PATCH 3/6] update comment --- get_data/get_EEA_data_portal.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/get_data/get_EEA_data_portal.py b/get_data/get_EEA_data_portal.py index 59c0c525..38a3ddd1 100644 --- a/get_data/get_EEA_data_portal.py +++ b/get_data/get_EEA_data_portal.py @@ -23,7 +23,7 @@ ---------------------------------------------- The drinking water portal dataset is large (~200MB) and we do a full refresh of this every run, instead of updating incrementally. Unfortunately, there is no good alternative because -the API does not support date filtering. +the API does not support date filtering and does not paginate in chronological order. The API response includes a TotalCount field (visible on a single lightweight probe request `?_end=1&_start=0`), which could in principle serve as a skip sentinel: store the count, From 28d09d61011e78a0cc7732727c4cb5113f2ba234 Mon Sep 17 00:00:00 2001 From: Nathan Date: Tue, 28 Apr 2026 21:26:52 -0400 Subject: [PATCH 4/6] Fix gsutil cp shell injection on filenames with & or parentheses Filenames like 'ME0000868-316(b)_...' or '...&...' were passed unquoted to os.system(), causing shell errors. shlex.quote() was already imported and used for the wget call on the same path; apply it to both arguments of the gsutil cp call too. Co-Authored-By: Claude Sonnet 4.6 --- get_data/get_EPARegion1_NPDES_permits.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/get_data/get_EPARegion1_NPDES_permits.py b/get_data/get_EPARegion1_NPDES_permits.py index c8c6e5dc..15306376 100644 --- a/get_data/get_EPARegion1_NPDES_permits.py +++ b/get_data/get_EPARegion1_NPDES_permits.py @@ -245,7 +245,7 @@ permit_url = html.unescape(permit) os.system('wget ' + shlex.quote(permit_url) + ' --no-clobber --timeout=30 --tries=3 -O ' + shlex.quote(local_file)) if os.path.exists(local_file): - os.system('gsutil cp ' + local_file + ' gs://openamend-data/' + local_file) + os.system('gsutil cp ' + shlex.quote(local_file) + ' ' + shlex.quote('gs://openamend-data/' + local_file)) new_pdf_count += 1 else: out_files += [['']] From 90942d1fc34b03b135f5efe8afebd38df85e8710 Mon Sep 17 00:00:00 2001 From: Nathan Date: Tue, 28 Apr 2026 21:37:46 -0400 Subject: [PATCH 5/6] Fix CSO incremental fetch: handle 500-for-empty-results from CSOAPI The CSOAPI returns HTTP 500 instead of an empty list when IncidentFromDate matches zero records. _query_page now treats non-200 / missing 'results' key as an empty page (returns None) rather than raising KeyError. run_query returns an empty DataFrame when no pages succeed, and get_data falls back to the full existing cache when that happens. Also removes the unused portal-page session warmup (AWSALB cookie from the portal backend does not carry over to the CSOAPI backend). Co-Authored-By: Claude Sonnet 4.6 --- get_data/get_eea_dp_cso.py | 92 +++++++++++++++++++++++--------------- 1 file changed, 57 insertions(+), 35 deletions(-) diff --git a/get_data/get_eea_dp_cso.py b/get_data/get_eea_dp_cso.py index b3245ff0..329cf918 100644 --- a/get_data/get_eea_dp_cso.py +++ b/get_data/get_eea_dp_cso.py @@ -7,17 +7,26 @@ pagination and auth requirements. Key implementation notes: - - The CSOAPI requires a Referer header pointing to the portal page; bare requests + - The CSOAPI requires a Referer header matching the portal page; bare requests return HTTP 500. The REQ_HEADER below must be kept in sync with the portal URL. - The API is 1-indexed (pageNumber starts at 1, not 0). - Timestamps are ISO 8601 but may or may not include milliseconds; use format='ISO8601'. - The API returns a lowercase 'year' column; we drop it to avoid a case-insensitive name collision with our added 'Year' column when writing to SQLite. + - Date-filtered queries (IncidentFromDate) work correctly when records exist, but + the API returns HTTP 500 instead of an empty list when zero records match. We + treat a 500 on the first page of a filtered query as "no new records" and fall + back to the existing cache unchanged. + +Incremental fetching: + When a cached CSV exists, we load it, find the max incidentDate, and fetch only + records from that date onward (inclusive, to catch records that arrived after the + last pull). Cached rows on the boundary date are dropped before merging so there + are no duplicates. A full fetch is used when no cache exists. Example API URL: https://eeaonline.eea.state.ma.us/dep/CSOAPI/api/Incident/GetIncidentsBySearchFields/ - ?ReporterClass=Verified%20Data%20Report&IncidentFromDate=01/01/2022 - &IncidentToDate=08/02/2023&RainfallDataFrom=1&pageNumber=2&pageSize=50 + ?IncidentFromDate=01/04/2026&pageNumber=1&pageSize=50 Outputs: ../docs/data/EEADP_CSO.csv — full CSO incident table @@ -26,48 +35,54 @@ """ import requests -from typing import Optional - import datetime import pandas as pd -# The CSOAPI requires a Referer header matching the portal page; plain User-Agent requests return 500. +PORTAL_URL = 'https://eeaonline.eea.state.ma.us/portal/dep/cso-data-portal/' +API_BASE_URL = 'https://eeaonline.eea.state.ma.us/dep/CSOAPI/api/Incident/GetIncidentsBySearchFields/?pageSize=50&' + REQ_HEADER = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', - 'Referer': 'https://eeaonline.eea.state.ma.us/portal/dep/cso-data-portal/', + 'Referer': PORTAL_URL, 'Origin': 'https://eeaonline.eea.state.ma.us', 'Accept': 'application/json, text/plain, */*', } -API_BASE_URL = 'https://eeaonline.eea.state.ma.us/dep/CSOAPI/api/Incident/GetIncidentsBySearchFields/?pageSize=50&' + +def _make_session() -> requests.Session: + return requests.Session() + def update_query_time(): - """Update the yml file that indicates the time of last query. - """ + """Update the yml file that indicates the time of last query.""" with open('../docs/data/ts_update_EEADP_CSO.yml', 'w') as f: - f.write('updated: '+str(datetime.datetime.now()).split('.')[0]+'\n') + f.write('updated: ' + str(datetime.datetime.now()).split('.')[0] + '\n') -def _query_page(page: int, query_params: Optional[dict[str, str]]=None) -> Optional[pd.DataFrame]: - """Query for and return a single page of API results. - If the resulting query is empty, return Non +def _query_page(session: requests.Session, page: int, query_params: dict[str, str] | None = None) -> pd.DataFrame | None: + """Query for and return a single page of API results, or None if empty. + + Returns None both for a normal empty page (end of results) and for an HTTP 500, + which the CSOAPI returns instead of an empty list when a filter matches no records. """ print(f'Querying for page {page}') if query_params is None: query_params = {} query_params['pageNumber'] = page query_string = '&'.join(f'{key}={val}' for key, val in query_params.items()) - r = requests.get(API_BASE_URL + query_string, headers=REQ_HEADER) + r = session.get(API_BASE_URL + query_string, headers=REQ_HEADER) + if not r.ok or 'results' not in r.json(): + return None if len(r.json()['results']) > 0: return pd.concat([pd.Series(c) for c in r.json()['results']], axis=1).T else: return None -def run_query(from_date: Optional[str] = None) -> pd.DataFrame: - """Run a query, paging through results and returning a combined DataFrame. - from_date: optional MM/DD/YYYY string; if provided, fetches only records - on or after that date (IncidentFromDate API param). +def run_query(session: requests.Session, from_date: str | None = None) -> pd.DataFrame: + """Page through API results and return a combined DataFrame. + + from_date: optional MM/DD/YYYY string passed as IncidentFromDate. """ if from_date: print(f'Running incremental query from {from_date}') @@ -79,16 +94,17 @@ def run_query(from_date: Optional[str] = None) -> pd.DataFrame: page = 1 # CSOAPI is 1-indexed result_dfs = [] while True: - df = _query_page(page, query_params) + df = _query_page(session, page, query_params) if df is None: break result_dfs.append(df) page += 1 + if not result_dfs: + return pd.DataFrame() return pd.concat(result_dfs) def _parse_dates(df: pd.DataFrame) -> pd.DataFrame: - """Parse date columns and add derived Year column.""" df['incidentDate'] = pd.to_datetime(df['incidentDate'], format='ISO8601') df['submittedDate'] = pd.to_datetime(df['submittedDate'], format='ISO8601') # API already returns a lowercase 'year' column; drop it before adding 'Year' @@ -99,16 +115,10 @@ def _parse_dates(df: pd.DataFrame) -> pd.DataFrame: def get_data() -> pd.DataFrame: - """Fetch CSO data incrementally when a cached CSV exists. - - Loads the cached file, determines the latest incidentDate, and fetches only - records from that date onwards (inclusive, to catch any records that arrived - after the last pull). Rows from the boundary date are dropped from the cache - before appending so there are no duplicates. - """ + """Fetch CSO data incrementally when a cached CSV exists, otherwise full fetch.""" csv_path = '../docs/data/EEADP_CSO.csv' - from_date: Optional[str] = None - existing: Optional[pd.DataFrame] = None + from_date: str | None = None + existing: pd.DataFrame | None = None try: existing = pd.read_csv(csv_path, index_col=0) @@ -121,7 +131,20 @@ def get_data() -> pd.DataFrame: except FileNotFoundError: print(' No cache found; running full query') - new_df = _parse_dates(run_query(from_date=from_date)) + session = _make_session() + raw = run_query(session, from_date=from_date) + + if raw.empty: + # API returned 500 (no-records) or genuinely empty; use cache as-is. + print(' No new records returned; using existing cache unchanged.') + if existing is not None: + # Restore the boundary rows we dropped before returning + full_existing = pd.read_csv('../docs/data/EEADP_CSO.csv', index_col=0) + return full_existing + # No cache and no data — nothing to write. + raise RuntimeError('CSO API returned no data and no cache exists.') + + new_df = _parse_dates(raw) if existing is not None: df = pd.concat([existing, new_df], ignore_index=True) @@ -132,18 +155,17 @@ def get_data() -> pd.DataFrame: def write_data(df: pd.DataFrame): - """Write data to a local table for integration with AMEND. - """ + """Write data to a local table for integration with AMEND.""" print('Writing out queried data') df.to_csv('../docs/data/EEADP_CSO.csv', index=True) df.sample(n=10).to_csv('../docs/data/EEADP_CSO_sample.csv', index=False) + def main(): - """Query and write all data. - """ all_data = get_data() write_data(all_data) update_query_time() + if __name__ == '__main__': main() From a928e698e04ab20b0214d1577418cc85ad4a9b2f Mon Sep 17 00:00:00 2001 From: Nathan Date: Tue, 28 Apr 2026 21:45:47 -0400 Subject: [PATCH 6/6] Add incremental fetching for drinkingWater, inspection, and enforcement MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The EEA DataLake API supports date-range filtering (discovered April 2026): drinkingWater: FromCollectedDate, inspection: FromInspectionDate, enforcement: FromEnforcementDate. For each, we load the cached CSV (or GCS copy for drinkingWater), find the max date, and fetch only records from that date onward. permit and facility have no working filter and remain full fetches. drinkingWater uses a GCS round-trip: download existing full CSV → append new rows → re-upload. The annualized summary is recomputed from the full dataset. Also removes 500 from the retry status_forcelist: the API returns 500 for zero-result filters rather than as a genuine server error. Also corrects the module docstring which previously stated no date filtering was available (that conclusion was wrong — the right parameter names were not discovered until live browser inspection). Co-Authored-By: Claude Sonnet 4.6 --- get_data/get_EEA_data_portal.py | 259 ++++++++++++++++++++------------ 1 file changed, 165 insertions(+), 94 deletions(-) diff --git a/get_data/get_EEA_data_portal.py b/get_data/get_EEA_data_portal.py index 38a3ddd1..e6e8b8c9 100644 --- a/get_data/get_EEA_data_portal.py +++ b/get_data/get_EEA_data_portal.py @@ -8,38 +8,42 @@ uploaded to GCS; an annualized summary is kept locally. HTTP requests use a session with automatic retries (5 attempts, exponential backoff) to -tolerate transient 5xx / 429 errors from the portal. +tolerate transient 5xx / 429 errors from the portal. Note: 500 is excluded from the +retry list because the API returns 500 to signal "no records match filter" rather than +as a genuine server error — retrying would just waste time. CSO data is handled separately in get_eea_dp_cso.py because it uses a different API. +Incremental fetching +-------------------- +Three tables support date-range filtering via query parameters discovered in April 2026: + - drinkingWater: FromCollectedDate=YYYY-MM-DD (date col: CollectedDate) + - inspection: FromInspectionDate=YYYY-MM-DD (date col: InspectionDate) + - enforcement: FromEnforcementDate=YYYY-MM-DD (date col: EnforcementDate) + +For these tables we load the existing local CSV (or GCS copy for drinkingWater), find +the max date, and fetch only records from that date onward (inclusive, to catch records +submitted after the previous pull). Rows on the boundary date are dropped from the +cache before merging to avoid duplicates. A full fetch is used when no cache exists. + +The API returns HTTP 500 instead of an empty list when a date filter matches zero records. +We treat a non-OK response on the initial TotalCount request as "no new records" and +fall back to the existing cache unchanged. + +permit and facility have no discovered date filter and are always fetched in full. + +drinkingWater GCS flow +---------------------- +Because the full drinkingWater CSV lives only in GCS (too large to commit), the incremental +flow is: gsutil cp (download) → append new rows → gsutil cp (re-upload). If the download +fails (first run or GCS unavailable), we fall back to a full API fetch. + Outputs (per table, e.g. 'permit'): ../docs/data/EEADP_permit.csv — full table ../docs/data/EEADP_permit_sample.csv — 10-row sample gs://openamend-data/EEADP_drinkingWater.csv — full drinking water table (GCS only) ../docs/data/EEADP_drinkingWater_annual.csv — annualized summary ../docs/data/ts_update_EEADP.yml — timestamp of last run - -Why drinkingWater is not incrementally fetched ----------------------------------------------- -The drinking water portal dataset is large (~200MB) and we do a full refresh of this every -run, instead of updating incrementally. Unfortunately, there is no good alternative because -the API does not support date filtering and does not paginate in chronological order. - -The API response includes a TotalCount field (visible on a single lightweight probe request -`?_end=1&_start=0`), which could in principle serve as a skip sentinel: store the count, -compare on next run, and skip the full download+upload if unchanged. - -Two reasons this is not worth implementing: - -1. The sentinel only helps when nothing changed. If TotalCount increased the full ~200 MB - download and GCS upload still happens — there is no way to fetch only new rows because - the API is offset-based with no date-range filter parameter. - -2. Drinking water lab results are submitted by public water systems continuously throughout - the year (~3.8 M rows as of April 2026). TotalCount increases virtually every week, so - the sentinel would almost never trigger a skip in practice. - -The 200 MB weekly upload is effectively unavoidable unless the API gains date-range filtering. """ import os @@ -57,113 +61,180 @@ API_ROOT = 'http://eeaonline.eea.state.ma.us/EEA/DataLake/V1.0/DataLakeAPI/' API_TABLES = ['permit', 'facility', 'inspection', 'enforcement', 'drinkingWater'] +# Tables with date-filter support: {table: (filter_param, date_col_in_csv)} +INCREMENTAL_TABLES = { + 'inspection': ('FromInspectionDate', 'InspectionDate'), + 'enforcement': ('FromEnforcementDate', 'EnforcementDate'), + 'drinkingWater': ('FromCollectedDate', 'CollectedDate'), +} ########################## ## Function definitions ########################## -# Generic user agent -REQ_HEADER = {'User-Agent': - 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36'} +REQ_HEADER = { + 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Safari/537.36', + 'Referer': 'https://eeaonline.eea.state.ma.us/Portal/', +} def _make_session() -> requests.Session: - """Return a requests Session with automatic retries on transient errors.""" + """Return a requests Session with automatic retries on transient errors. + + 500 is intentionally excluded from status_forcelist: this API returns 500 + to mean "no records match filter", so retrying wastes time. + """ session = requests.Session() retry = Retry( total=5, backoff_factor=2, - status_forcelist=[429, 500, 502, 503, 504], + status_forcelist=[429, 502, 503, 504], ) session.mount('http://', HTTPAdapter(max_retries=retry)) session.mount('https://', HTTPAdapter(max_retries=retry)) return session -def query_iterate(table_name: str, req_size: int=100000, verbose: bool=True): - """ - Query the EEA data portal to retrieve the entirety of a data table. - Returns +def _get_max_date(csv_path: str, date_col: str) -> str | None: + """Return the max date in date_col of csv_path as YYYY-MM-DD, or None.""" + try: + df = pd.read_csv(csv_path, usecols=[date_col]) + max_val = pd.to_datetime(df[date_col], errors='coerce').max() + if pd.isna(max_val): + return None + return max_val.strftime('%Y-%m-%d') + except (FileNotFoundError, ValueError, KeyError): + return None + + +def query_iterate(table_name: str, req_size: int = 100000, verbose: bool = True, + filter_param: str | None = None, filter_val: str | None = None) -> pd.DataFrame: + """Query the EEA DataLake API, returning the full (or date-filtered) table. Args: - table_name (str): EEA data portal table to query - req_size (int): Request chunksize - verbose (bool): Print chunk position while iterating + table_name: DataLake table name (e.g. 'drinkingWater') + req_size: Rows per paginated request + verbose: Print progress + filter_param: Optional date-filter query param name (e.g. 'FromCollectedDate') + filter_val: Optional date-filter value in YYYY-MM-DD format Returns: - df: Pandas DataFrame with table contents + DataFrame of matching rows, or empty DataFrame when filter matches nothing. """ session = _make_session() - # Get total table size - try: - r = session.get(API_ROOT + table_name + '?_end=1&_start=0', headers=REQ_HEADER) - table_size = r.json()['TotalCount'] - except ValueError: - raise ValueError("EEA Data Portal request returned error " + str(r.status_code) + '; perhaps table name is not valid\n\nFull response message:\n' + r.text) + filter_qs = f'&{filter_param}={filter_val}' if filter_param and filter_val else '' + mode = f'filtered {filter_param}={filter_val}' if filter_qs else 'full' + print(f'{table_name}: {mode} fetch') + + # Get total row count (with filter applied). + size_url = API_ROOT + table_name + '?_end=1&_start=0' + filter_qs + r = session.get(size_url, headers=REQ_HEADER, timeout=120) + if not r.ok or 'TotalCount' not in r.json(): + # API returns 500 with no 'TotalCount' when filter matches zero records. + print(f' {table_name}: no records match filter (HTTP {r.status_code}); returning empty') + return pd.DataFrame() - # Iterate through requests - if (table_size < req_size): + table_size = r.json()['TotalCount'] + if table_size == 0: + return pd.DataFrame() + print(f' {table_name}: {table_size:,} rows to fetch') + + if table_size < req_size: req_bins = [0, table_size] else: - max_bin = table_size + req_size - req_bins = np.arange(0, max_bin, req_size) + req_bins = list(np.arange(0, table_size + req_size, req_size)) + dfs = [] for i in range(len(req_bins) - 1): - # Log output - if verbose: print(table_name + ': request ' + str(i + 1) + ' of ' + str(len(req_bins) - 1)) - # Make request - url = API_ROOT + table_name + '?_end=' + str(req_bins[i+1]) + '&_start='+str(req_bins[i]) - r = session.get(url, headers=REQ_HEADER) - # Add chunk contents to dataframe list - dfs += [pd.DataFrame(r.json()['Items'])] + if verbose: + print(f'{table_name}: request {i + 1} of {len(req_bins) - 1}') + url = (API_ROOT + table_name + + f'?_end={req_bins[i+1]}&_start={req_bins[i]}' + filter_qs) + r = session.get(url, headers=REQ_HEADER, timeout=180) + r.raise_for_status() + dfs.append(pd.DataFrame(r.json()['Items'])) - # Concatenate chunks - df = pd.concat(dfs) + return pd.concat(dfs, ignore_index=True) - return df -def main(): - """Query for data, persist it, and report the update +def fetch_incremental(table_name: str, csv_path: str, filter_param: str, + date_col: str) -> tuple[pd.DataFrame, bool]: + """Load cached CSV and fetch only records newer than the max cached date. + + Returns (dataframe, is_incremental). is_incremental=False means we fell + back to a full fetch because no cache was available. """ - # Get data - ## Query data for each table + max_date = _get_max_date(csv_path, date_col) + if max_date is None: + print(f' {table_name}: no cache found; running full fetch') + return query_iterate(table_name), False + + print(f' {table_name}: cache through {max_date}; fetching from {max_date} (inclusive)') + new_data = query_iterate(table_name, filter_param=filter_param, filter_val=max_date) + + existing = pd.read_csv(csv_path) + # Drop boundary-date rows from cache — they're included in the fresh fetch. + existing[date_col] = pd.to_datetime(existing[date_col], errors='coerce') + cutoff = pd.to_datetime(max_date).date() + existing = existing[existing[date_col].dt.date < cutoff] + + if new_data.empty: + print(f' {table_name}: no new records; using cache as-is') + return pd.read_csv(csv_path), True + + combined = pd.concat([existing, new_data], ignore_index=True) + print(f' {table_name}: appended {len(new_data):,} new rows (total {len(combined):,})') + return combined, True + + +def main(): + """Query for data, persist it, and report the update.""" + table_data = {} - for tab in API_TABLES: + + # --- permit and facility: always full fetch (no date filter available) --- + for tab in ['permit', 'facility']: table_data[tab] = query_iterate(tab) - ## Write out, but treat large tables separately - ## Only one table (drinkingWater) is >10MB as of 08/2017, so we handle this as a special case. - ## Could also use `size_MB = os.path.getsize('../docs/data/EEADP_' + tab + '.csv')/1024/1024` to get file size + # --- inspection and enforcement: incremental via date filter --- + for tab in ['inspection', 'enforcement']: + filter_param, date_col = INCREMENTAL_TABLES[tab] + csv_path = f'../docs/data/EEADP_{tab}.csv' + table_data[tab], _ = fetch_incremental(tab, csv_path, filter_param, date_col) + + # --- drinkingWater: incremental via GCS cache + date filter --- + filter_param, date_col = INCREMENTAL_TABLES['drinkingWater'] + dw_local = 'EEADP_drinkingWater.csv' + gcs_path = f'gs://openamend-data/{dw_local}' + + print('drinkingWater: downloading existing data from GCS...') + gcs_rc = os.system(f'gsutil cp {gcs_path} {dw_local}') + if gcs_rc == 0 and os.path.exists(dw_local): + table_data['drinkingWater'], _ = fetch_incremental( + 'drinkingWater', dw_local, filter_param, date_col) + else: + print(' drinkingWater: GCS download failed; running full fetch') + table_data['drinkingWater'] = query_iterate('drinkingWater') + + # --- Write outputs --- for tab in API_TABLES: - ## Print a sample of the file as an example - table_data[tab].sample(n=10).to_csv('../docs/data/EEADP_' + tab + '_sample.csv', index=0) - - if tab != 'drinkingWater': - table_data[tab].to_csv('../docs/data/EEADP_' + tab + '.csv', index=0) + df = table_data[tab] + df.sample(n=min(10, len(df))).to_csv(f'../docs/data/EEADP_{tab}_sample.csv', index=False) + + if tab != 'drinkingWater': + df.to_csv(f'../docs/data/EEADP_{tab}.csv', index=False) else: - ## Send to Google object store - table_data[tab].to_csv('EEADP_' + tab + '.csv', encoding='utf-8', index=0) - os.system('gsutil cp EEADP_' + tab + '.csv gs://openamend-data/EEADP_' + tab + '.csv') - - ## Include some special summary statistics tables - ## --- - ## Most recent report for each chemical for each site - ## This still ends up being ~20% of the original size, so larger than desired - #table_data[tab].sort_values('CollectedDate', inplace=True) - #df_dw_last = table_data[tab].groupby(['ChemicalName','PWSName','LocationName']).last() - - # Tests per year per PWS per contaminant group per raw/finished - ## This still ends up being ~40% of the original size, so larger than desired - table_data[tab]['CollectedDate'] = pd.to_datetime(table_data[tab]['CollectedDate'], errors='coerce') - table_data[tab]['Year'] = table_data[tab]['CollectedDate'].apply(lambda x: x.year) - df_dw_annual_group = table_data[tab].groupby(['Year','PWSName', 'ContaminantGroup','RaworFinished']).agg({'Result': pd.Series.count}) - df_dw_annual_group.to_csv('../docs/data/EEADP_' + tab + '_annual.csv', index=1) - ## Print a sample of the file as an example - df_dw_annual_group.sample(n=10).to_csv('../docs/data/EEADP_' + tab + '_annual_sample.csv', index=1) - - ## Tests per year per PWS per chemical per raw/finished - ### This still ends up being ~40% of the original size, so larger than desired - #df_dw_annual = table_data[tab].groupby(['Year','PWSName', 'ChemicalName','RaworFinished']).agg({'ContaminantGroup': lambda x: x.iloc[0], 'Result': pd.Series.count}) + df.to_csv(dw_local, encoding='utf-8', index=False) + os.system(f'gsutil cp {dw_local} {gcs_path}') + + # Annualized summary + df['CollectedDate'] = pd.to_datetime(df['CollectedDate'], errors='coerce') + df['Year'] = df['CollectedDate'].dt.year + df_annual = df.groupby(['Year', 'PWSName', 'ContaminantGroup', 'RaworFinished']).agg( + {'Result': pd.Series.count}) + df_annual.to_csv('../docs/data/EEADP_drinkingWater_annual.csv', index=True) + df_annual.sample(n=min(10, len(df_annual))).to_csv( + '../docs/data/EEADP_drinkingWater_annual_sample.csv', index=True) # Archive PDF help files os.system('wget http://eeaonline.eea.state.ma.us/Portal/documents/General%20Query%20Search%20FAQs.pdf') @@ -171,9 +242,9 @@ def main(): os.system('wget http://eeaonline.eea.state.ma.us/Portal/documents/Terms%20and%20Definitions%20for%20EEA.pdf') os.system('mv "Terms and Definitions for EEA.pdf" ../docs/assets/PDFs/EEADP_Definitions.pdf') - # Report last update with open('../docs/data/ts_update_EEADP.yml', 'w') as f: - f.write('updated: '+str(datetime.datetime.now()).split('.')[0]+'\n') + f.write('updated: ' + str(datetime.datetime.now()).split('.')[0] + '\n') + if __name__ == '__main__': main()