diff --git a/scripts/us_census/surface_area/README.md b/scripts/us_census/surface_area/README.md new file mode 100644 index 0000000000..c7de709c52 --- /dev/null +++ b/scripts/us_census/surface_area/README.md @@ -0,0 +1,141 @@ +# US Census Surface Area Import (`USCensusGeos_SurfaceArea`) + +## Overview + +- **Import Type**: Automated (Scheduled) +- **Variable Measured**: `dcs:SurfaceArea` +- **Unit**: `dcs:SquareMile` +- **Formula**: + $$\text{SurfaceArea} = \text{round}\left(\frac{\text{ALAND} + \text{AWATER}}{2589988.11}, 4\right)$$ + (Converts square meters to square miles, rounded to 4 decimal places). + +Historically, this dataset was generated via a manual Google3 internal BigQuery script (`//depot/google3/datacommons/import/mcf/manifest/us_census/USCensusGeos_SurfaceArea.textproto`). This automated pipeline migrates and automates the process to directly fetch public U.S. Census Gazetteer files and State Area Measurements, compute the total surface area (land area + water area), and produce production-ready Data Commons CSV and TMCF files across all available years (2018–2025+). + +## Data Sources + +1. **U.S. Census Bureau Gazetteer Files**: + - Base URL: `https://www2.census.gov/geo/docs/maps-data/data/gazetteer/` + - Files utilized per annual release: + - `counties`: `*Gaz_counties_national.zip` + - `congressional_districts`: `*Gaz_*CDs_national.zip` (e.g. 116CDs, 119CDs) + - `cbsa`: `*Gaz_cbsa_national.zip` + - `place`: `*Gaz_place_national.zip` + - `cousubs`: `*Gaz_cousubs_national.zip` + - `unsd`: `*Gaz_unsd_national.zip` (Unified School Districts) + - `elsd`: `*Gaz_elsd_national.zip` (Elementary School Districts) + - `scsd`: `*Gaz_scsd_national.zip` (Secondary School Districts) + - `tracts`: `*Gaz_tracts_national.zip` + - `state`: `*Gaz_state_national.zip` (for releases >= 2024) + +2. **U.S. Census Bureau State Area Measurements**: + - URL: `https://www.census.gov/geographies/reference-files/2010/geo/state-area.html` + - Used for baseline 50 U.S. States, District of Columbia, and Puerto Rico. + +## Geographic Entities & Coverage + +| Entity Type | DCID Prefix | Example DCID | 2018 Baseline Count | +|---|---|---|---| +| States | `geoId/` | `geoId/01` (Alabama) | 52 | +| Counties | `geoId/` | `geoId/01001` (Autauga County) | 3,220 | +| Congressional Districts | `geoId/` | `geoId/0101` (AL CD 1) | 440 | +| Core Based Statistical Areas (CBSA) | `geoId/C` | `geoId/C10100` | 945 | +| Places | `geoId/` | `geoId/0100124` | 29,574 | +| County Subdivisions (MCDs) | `geoId/` | `geoId/0100190000` | 36,630 | +| Unified School Districts | `geoId/sch` | `geoId/sch0100005` | 10,887 | +| Elementary School Districts | `geoId/sch` | `geoId/sch0100001` | 1,958 | +| Secondary School Districts | `geoId/sch` | `geoId/sch0400004` | 486 | +| Census Tracts | `geoId/` | `geoId/01001020100` | 74,001 | +| **Total** | | | **158,193** | + +## Directory Structure + +``` +surface_area/ +├── README.md # Pipeline documentation and operational guidelines +├── manifest.json # Import automation manifest specification +├── validation_config.json # Data validation threshold configuration +├── preprocess.py # Download and preprocessing script +├── preprocess_test.py # Hermetic unit tests +├── input_files/ # Downloaded raw input files +├── output_files/ # Generated cleaned CSV and TMCF +└── test_data/ # Sample fixtures and expected test outputs +``` + +## Running the Pipeline + +### Dynamic Year Discovery & Full Automation +By default, `--years` is set to `auto`. When executed (e.g. via scheduled monthly cron), the pipeline dynamically queries `https://www2.census.gov/geo/docs/maps-data/data/gazetteer/`, detects all available annual release directories starting from 2018 up to the latest published year (e.g. 2018–2026+), downloads any newly released files, and processes all years. No manual code updates or year bumps are needed for future releases. + +### Download Mode +Fetches the raw gazetteer and reference files from census.gov (skipping already-downloaded historical files): +```bash +# Automatically discovers and downloads all available years >= 2018: +python3 preprocess.py --mode=download + +# Or specify a targeted year / range: +python3 preprocess.py --mode=download --year=latest +python3 preprocess.py --mode=download --years=2018-2025 +``` + +### Process Mode +Processes downloaded raw files into `output_files/surface_area.csv` and `output_files/surface_area.tmcf`: +```bash +python3 preprocess.py --mode=process +``` + +### End-to-End Execution +Downloads any new/missing files and processes all available years in a single step: +```bash +python3 preprocess.py --mode=all +``` + +## Validation & Verification + +### 1. Unit Tests +Run unit tests hermetically: +```bash +python3 -m unittest preprocess_test.py +``` + +### 2. Validation Configuration & Threshold Justifications + +The import configuration in `validation_config.json` enforces 2 critical validation rules: + +1. **`check_deleted_records_percent` (`DELETED_RECORDS_PERCENT`: 0.1%)**: + - **Justification**: A strict 0.1% threshold is maintained because Census geographic boundaries (e.g. Census Designated Places, School Districts, and Tracts) occasionally undergo rare dissolutions, annexations, or boundary consolidations across annual national releases. This threshold accommodates legitimate administrative changes while catching unintended data drops. +2. **`check_max_date_freshness` (`SQL_VALIDATOR`)**: + - **Justification**: Uses DuckDB SQL condition `max_year >= (EXTRACT(YEAR FROM CURRENT_DATE) - 2)` to verify freshness within allowable annual release latency without prematurely failing on calendar year rollovers before the Census Bureau releases the new year's Gazetteer files. + +Run the import validator locally: +```bash +python3 tools/import_validation/runner.py \ + --stats_summary=scripts/us_census/surface_area/output_files/dc_generated/summary_report.csv \ + --lint_report=scripts/us_census/surface_area/output_files/dc_generated/report.json \ + --differ_output=scripts/us_census/surface_area/diff \ + --validation_config=scripts/us_census/surface_area/validation_config.json \ + --validation_output=/tmp/validation_output.csv +``` + +### 3. Differ Comparison against Production Baseline + +To verify backwards compatibility with legacy production data: +```bash +python3 tools/import_differ/import_differ.py \ + --current_data=scripts/us_census/surface_area/output_files/dc_generated/table_mcf_nodes_surface_area.mcf \ + --previous_data=scripts/us_census/surface_area/output_files/prod.mcf \ + --output_location=scripts/us_census/surface_area/diff \ + --file_format=mcf \ + --runner_mode=native +``` + +Result for 2018 baseline comparison: +- **Deleted Observations**: **0** (100% legacy entity coverage) +- **Modified Observations**: **0** (Exact numerical agreement) +- **Added Observations**: New observations covering years 2019 through 2025. + +## Troubleshooting + +- **HTTP 404 on Future Years**: + If `--years` includes an unreleased future year, `preprocess.py` logs a clear warning and stops year discovery early without failing the pipeline. +- **Corrupted or Truncated Downloads**: + Downloads write to temporary files in `input_files/` and atomically replace existing files only after checking that the file is non-empty (`os.path.getsize > 0`). If a download is interrupted, retry with `python3 preprocess.py --mode=download`. diff --git a/scripts/us_census/surface_area/manifest.json b/scripts/us_census/surface_area/manifest.json new file mode 100644 index 0000000000..7adc1f2e0e --- /dev/null +++ b/scripts/us_census/surface_area/manifest.json @@ -0,0 +1,27 @@ +{ + "import_specifications": [ + { + "import_name": "USCensusGeos_SurfaceArea", + "curator_emails": [ + "support@datacommons.org" + ], + "provenance_url": "https://www.census.gov/geographies/reference-files.html", + "provenance_description": "Surface area measurements (land and water) for US geographic entities from Census Gazetteer files.", + "scripts": [ + "preprocess.py" + ], + "source_files": [ + "input_files/*", + "validation_config.json" + ], + "import_inputs": [ + { + "template_mcf": "output_files/surface_area.tmcf", + "cleaned_csv": "output_files/surface_area.csv" + } + ], + "cron_schedule": "0 0 1 * *", + "validation_config_file": "validation_config.json" + } + ] +} diff --git a/scripts/us_census/surface_area/preprocess.py b/scripts/us_census/surface_area/preprocess.py new file mode 100755 index 0000000000..b508c9e2d7 --- /dev/null +++ b/scripts/us_census/surface_area/preprocess.py @@ -0,0 +1,638 @@ +#!/usr/bin/env python3 +# Copyright 2024 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. +"""Download and preprocess US Census Gazetteer files for Surface Area.""" + +from contextlib import contextmanager +import os +import re +import shutil +import tempfile +from typing import Dict, List, Tuple +import zipfile + +from absl import app +from absl import flags +from absl import logging +from bs4 import BeautifulSoup +import pandas as pd +import requests +from requests.adapters import HTTPAdapter +from urllib3.util.retry import Retry + +# ----------------------------------------------------------------------------- +# Configuration and Constants +# ----------------------------------------------------------------------------- + +INPUT_DIR = 'input_files' +OUTPUT_DIR = 'output_files' + +OUTPUT_CSV = 'surface_area.csv' +OUTPUT_TMCF = 'surface_area.tmcf' + +# Conversion factor: 1 Square Mile = 2,589,988.11 Square Meters +SQMI_FACTOR = 2589988.11 + +# URLs +CENSUS_BASE_URL = 'https://www2.census.gov/geo/docs/maps-data/data/gazetteer/' +STATE_AREA_URL = ( + 'https://www.census.gov/geographies/reference-files/2010/geo/state-area.html' +) + +DEFAULT_YEAR = '2018' + +# Gazetteer file specifications: (file_pattern, dcid_prefix, geoid_column, description) +GAZETTEER_FILE_SPECS = [ + (r'.*_Gaz_counties_national\.(zip|txt)', 'geoId/', 'GEOID', 'Counties'), + (r'.*_Gaz_.*CDs_national\.(zip|txt)', 'geoId/', 'GEOID', + 'Congressional Districts'), + (r'.*_Gaz_cbsa_national\.(zip|txt)', 'geoId/C', 'GEOID', + 'Core Based Statistical Areas'), + (r'.*_Gaz_place_national\.(zip|txt)', 'geoId/', 'GEOID', 'Places'), + (r'.*_Gaz_cousubs_national\.(zip|txt)', 'geoId/', 'GEOID', + 'County Subdivisions'), + (r'.*_Gaz_unsd_national\.(zip|txt)', 'geoId/sch', 'GEOID', + 'Unified School Districts'), + (r'.*_Gaz_elsd_national\.(zip|txt)', 'geoId/sch', 'GEOID', + 'Elementary School Districts'), + (r'.*_Gaz_scsd_national\.(zip|txt)', 'geoId/sch', 'GEOID', + 'Secondary School Districts'), + (r'.*_Gaz_tracts_national\.(zip|txt)', 'geoId/', 'GEOID', 'Census Tracts'), +] + +# File pattern for state gazetteer file (available in recent releases >= 2024) +STATE_GAZETTEER_PATTERN = r'.*_Gaz_state_national\.(zip|txt)' + +# State name to 2-digit FIPS code mapping +STATE_FIPS_MAP = { + 'Alabama': '01', + 'Alaska': '02', + 'Arizona': '04', + 'Arkansas': '05', + 'California': '06', + 'Colorado': '08', + 'Connecticut': '09', + 'Delaware': '10', + 'District of Columbia': '11', + 'Florida': '12', + 'Georgia': '13', + 'Hawaii': '15', + 'Idaho': '16', + 'Illinois': '17', + 'Indiana': '18', + 'Iowa': '19', + 'Kansas': '20', + 'Kentucky': '21', + 'Louisiana': '22', + 'Maine': '23', + 'Maryland': '24', + 'Massachusetts': '25', + 'Michigan': '26', + 'Minnesota': '27', + 'Mississippi': '28', + 'Missouri': '29', + 'Montana': '30', + 'Nebraska': '31', + 'Nevada': '32', + 'New Hampshire': '33', + 'New Jersey': '34', + 'New Mexico': '35', + 'New York': '36', + 'North Carolina': '37', + 'North Dakota': '38', + 'Ohio': '39', + 'Oklahoma': '40', + 'Oregon': '41', + 'Pennsylvania': '42', + 'Rhode Island': '44', + 'South Carolina': '45', + 'South Dakota': '46', + 'Tennessee': '47', + 'Texas': '48', + 'Utah': '49', + 'Vermont': '50', + 'Virginia': '51', + 'Washington': '53', + 'West Virginia': '54', + 'Wisconsin': '55', + 'Wyoming': '56', + 'Puerto Rico': '72', +} + +DEFAULT_YEARS = 'auto' + +# TMCF template +TMCF_TEMPLATE = ('Node: E:Data->E0\n' + 'typeOf: schema:StatVarObservation\n' + 'variableMeasured: dcs:SurfaceArea\n' + 'observationAbout: C:Data->dcid\n' + 'observationDate: C:Data->observationDate\n' + 'value: C:Data->SurfaceArea\n' + 'unit: dcs:SquareMile\n') + +_USER_AGENT = ('Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ' + '(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 DataCommons') + +_MODULE_DIR = os.path.dirname(os.path.abspath(__file__)) + +_FLAGS = flags.FLAGS + +flags.DEFINE_enum('mode', 'all', ['download', 'process', 'all'], + 'Mode of operation: download, process, or all.') +flags.DEFINE_string( + 'years', DEFAULT_YEARS, + 'Gazetteer release years to download and process. Defaults to "auto" ' + '(dynamically discovers all available years >= 2018 from census.gov).') +flags.DEFINE_string('year', '', 'Alias for --years.') +flags.DEFINE_string( + 'input_dir', '', + 'Directory to store/read raw input files. Defaults to input_files/ in module dir.' +) +flags.DEFINE_string( + 'output_dir', '', + 'Directory to store generated CSV and TMCF. Defaults to output_files/ in module dir.' +) + + +def get_requests_session() -> requests.Session: + """Creates a requests.Session with connection pooling and exponential backoff retries.""" + session = requests.Session() + session.headers.update({'User-Agent': _USER_AGENT}) + retry_strategy = Retry( + total=3, + backoff_factor=1, + status_forcelist=[429, 500, 502, 503, 504], + raise_on_status=False, + ) + adapter = HTTPAdapter(max_retries=retry_strategy, + pool_connections=10, + pool_maxsize=10) + session.mount('https://', adapter) + session.mount('http://', adapter) + return session + + +def resolve_all_available_years(start_year: int = 2018, + session: requests.Session = None) -> List[str]: + """Discovers all available release years >= start_year from census.gov.""" + if session is None: + session = get_requests_session() + logging.info('Discovering available gazetteer release years from %s...', + CENSUS_BASE_URL) + try: + index_html = fetch_url(CENSUS_BASE_URL, + session).decode('utf-8', errors='replace') + years = sorted( + set(re.findall(r'href="([0-9]{4})_Gazetteer/"', index_html))) + available = [y for y in years if int(y) >= start_year] + if available: + logging.info('Found available gazetteer years: %s', + ', '.join(available)) + return available + except (requests.RequestException, ValueError, re.error) as e: + logging.warning( + 'Could not discover years dynamically (%s); falling back to 2018', + e) + return [str(start_year)] + + +def parse_years(year_spec: str, session: requests.Session = None) -> List[str]: + """Parses year specification (e.g. 'auto', '2018', '2018-2025', '2018-latest').""" + spec = str(year_spec).strip() + if spec.lower() in ('auto', 'all'): + return resolve_all_available_years(start_year=2018, session=session) + if spec.lower() == 'latest': + s = session or get_requests_session() + return [resolve_latest_gazetteer_year(s)] + if 'latest' in spec.lower(): + start = int(spec.lower().split('-', + 1)[0].strip()) if '-' in spec else 2018 + return resolve_all_available_years(start_year=start, session=session) + if '-' in spec: + start, end = spec.split('-', 1) + return [str(y) for y in range(int(start.strip()), int(end.strip()) + 1)] + return [y.strip() for y in spec.split(',') if y.strip()] + + +def calc_surface_area(aland: float, awater: float) -> float: + """Calculates surface area in Square Miles from land and water in Square Meters. + + Args: + aland: Land area in Square Meters. + awater: Water area in Square Meters. + + Returns: + Surface area in Square Miles rounded to 4 decimal places. + """ + return round((float(aland) + float(awater)) / SQMI_FACTOR, 4) + + +def _detect_separator(file_obj) -> str: + """Detects whether a text stream is pipe-separated or tab-separated.""" + pos = file_obj.tell() if hasattr(file_obj, 'tell') else None + first_line = file_obj.readline() + if pos is not None and hasattr(file_obj, 'seek'): + file_obj.seek(pos) + if isinstance(first_line, bytes): + return '|' if b'|' in first_line else '\t' + return '|' if '|' in str(first_line) else '\t' + + +def parse_gazetteer_data(file_obj, + dcid_prefix: str, + geoid_col: str = 'GEOID') -> Dict[str, float]: + """Parses a gazetteer text stream and computes surface area for each entity. + + Args: + file_obj: A file-like object or file path containing tab-separated values. + dcid_prefix: Prefix to prepend to GEOID (e.g. 'geoId/', 'geoId/C', 'geoId/sch'). + geoid_col: Column name containing the GEOID. + + Returns: + Dictionary mapping dcid -> SurfaceArea. + """ + sep = _detect_separator(file_obj) + df = pd.read_csv(file_obj, sep=sep, dtype=str, encoding='latin1') + df.columns = [c.strip() for c in df.columns] + + if geoid_col not in df.columns: + raise ValueError( + f'Column {geoid_col} not found in gazetteer file columns: {df.columns}' + ) + if 'ALAND' not in df.columns or 'AWATER' not in df.columns: + raise ValueError( + f'ALAND or AWATER column missing in gazetteer file columns: {df.columns}' + ) + + aland = pd.to_numeric(df['ALAND'].str.strip(), errors='coerce').fillna(0.0) + awater = pd.to_numeric(df['AWATER'].str.strip(), + errors='coerce').fillna(0.0) + surface_area = ((aland + awater) / SQMI_FACTOR).round(4) + + dcids = dcid_prefix + df[geoid_col].str.strip() + return dict(zip(dcids, surface_area)) + + +def parse_state_area_html(html_content: str) -> Dict[str, float]: + """Parses US Census State Area Measurements HTML table. + + Args: + html_content: HTML content of the state area reference page. + + Returns: + Dictionary mapping state dcid (e.g. 'geoId/01') -> SurfaceArea. + """ + soup = BeautifulSoup(html_content, 'html.parser') + records = {} + + for tr in soup.find_all('tr'): + tds = [td.get_text(strip=True) for td in tr.find_all('td')] + if len(tds) >= 6 and tds[0] in STATE_FIPS_MAP: + state_name = tds[0] + fips = STATE_FIPS_MAP[state_name] + dcid = f'geoId/{fips}' + try: + # Column 3: Land area (sq mi), Column 5: Water area (sq mi) + land_sqmi = float(tds[3].replace(',', '')) + water_sqmi = float(tds[5].replace(',', '')) + records[dcid] = land_sqmi + water_sqmi + except ValueError: + logging.warning('Could not parse area for state %s', state_name) + + return records + + +def parse_state_gazetteer(file_obj) -> Dict[str, float]: + """Parses a modern state gazetteer file (2024+). + + Args: + file_obj: File-like object with state gazetteer table. + + Returns: + Dictionary mapping state dcid -> SurfaceArea. + """ + sep = _detect_separator(file_obj) + df = pd.read_csv(file_obj, sep=sep, dtype=str, encoding='latin1') + df.columns = [c.strip() for c in df.columns] + + geoid_col = 'GEOID' + if geoid_col not in df.columns: + raise ValueError('GEOID column not found in state gazetteer file.') + + aland = pd.to_numeric(df['ALAND'].str.strip(), errors='coerce').fillna(0.0) + awater = pd.to_numeric(df['AWATER'].str.strip(), + errors='coerce').fillna(0.0) + surface_area = ((aland + awater) / SQMI_FACTOR).round(4) + + dcids = 'geoId/' + df[geoid_col].str.strip().str.zfill(2) + return dict(zip(dcids, surface_area)) + + +def fetch_url(url: str, session: requests.Session, timeout: int = 60) -> bytes: + """Fetches URL contents using a shared session with retries.""" + logging.info('HTTP GET %s', url) + response = session.get(url, timeout=timeout) + response.raise_for_status() + logging.info('HTTP GET %s succeeded (%d bytes)', url, len(response.content)) + return response.content + + +def download_file_atomic(url: str, + dest_path: str, + session: requests.Session, + timeout: int = 60) -> None: + """Downloads a file atomically to a temp file, verifies size, and moves.""" + dest_dir = os.path.dirname(dest_path) + os.makedirs(dest_dir, exist_ok=True) + logging.info('HTTP GET %s -> %s', url, dest_path) + + with tempfile.NamedTemporaryFile(dir=dest_dir, delete=False) as tmp_file: + tmp_path = tmp_file.name + try: + with session.get(url, stream=True, timeout=timeout) as response: + response.raise_for_status() + for chunk in response.iter_content(chunk_size=65536): + if chunk: + tmp_file.write(chunk) + tmp_file.flush() + tmp_file.close() + size = os.path.getsize(tmp_path) + if size == 0: + raise RuntimeError(f'Downloaded file {url} is empty (0 bytes)') + shutil.move(tmp_path, dest_path) + logging.info('Successfully downloaded %s (%d bytes)', dest_path, + size) + except Exception: + tmp_file.close() + if os.path.exists(tmp_path): + os.remove(tmp_path) + raise + + +def resolve_latest_gazetteer_year(session: requests.Session) -> str: + """Resolves the latest available gazetteer year from census.gov.""" + logging.info('Resolving latest gazetteer year from census.gov...') + index_html = fetch_url(CENSUS_BASE_URL, session).decode('utf-8', + errors='replace') + years = re.findall(r'href="([0-9]{4})_Gazetteer/"', index_html) + if not years: + logging.warning('Could not discover years; falling back to %s', + DEFAULT_YEAR) + return DEFAULT_YEAR + + latest_year = max(years) + logging.info('Resolved latest gazetteer year: %s', latest_year) + return latest_year + + +def get_gazetteer_year_file_list(year: str, + session: requests.Session) -> List[str]: + """Retrieves file listing for the specified gazetteer year directory.""" + year_url = f'{CENSUS_BASE_URL}{year}_Gazetteer/' + html = fetch_url(year_url, session).decode('utf-8', errors='replace') + files = re.findall(r'href="([^"]*Gaz_[^"]*)"', html) + return files + + +def download_files(input_dir: str, year_spec: str = DEFAULT_YEARS) -> List[str]: + """Downloads required gazetteer files and state area table into input_dir for all years. + + Args: + input_dir: Directory where downloaded files are stored. + year_spec: Year range or specification (e.g. '2018-2025' or 'latest'). + + Returns: + List of paths to downloaded files. + """ + os.makedirs(input_dir, exist_ok=True) + session = get_requests_session() + if str(year_spec).strip().lower() == 'latest': + years = [resolve_latest_gazetteer_year(session)] + else: + years = parse_years(year_spec) + downloaded_paths = [] + + for year in years: + year_url = f'{CENSUS_BASE_URL}{year}_Gazetteer/' + logging.info('Discovering files for year %s from %s...', year, year_url) + try: + available_files = get_gazetteer_year_file_list(year, session) + except requests.exceptions.HTTPError as e: + if e.response is not None and e.response.status_code == 404: + logging.warning( + 'Year %s gazetteer directory not found (HTTP 404). Stopping early.', + year) + break + raise + + for pattern, _, _, desc in GAZETTEER_FILE_SPECS: + matched_filename = None + for filename in available_files: + if re.match(pattern, filename): + matched_filename = filename + break + + if not matched_filename: + logging.warning( + 'No gazetteer file found for %s %s (pattern: %s)', year, + desc, pattern) + continue + + target_url = f'{year_url}{matched_filename}' + dest_path = os.path.join(input_dir, matched_filename) + + if not (os.path.exists(dest_path) and + os.path.getsize(dest_path) > 0): + logging.info('Downloading %s (%s) from %s...', matched_filename, + desc, target_url) + download_file_atomic(target_url, dest_path, session) + downloaded_paths.append(dest_path) + + # State file for modern years (2024+) + state_gaz_file = None + for filename in available_files: + if re.match(STATE_GAZETTEER_PATTERN, filename): + state_gaz_file = filename + break + + if state_gaz_file: + dest_path = os.path.join(input_dir, state_gaz_file) + if not (os.path.exists(dest_path) and + os.path.getsize(dest_path) > 0): + target_url = f'{year_url}{state_gaz_file}' + logging.info('Downloading state gazetteer %s from %s...', + state_gaz_file, target_url) + download_file_atomic(target_url, dest_path, session) + downloaded_paths.append(dest_path) + + # State area reference table (for 2018-2023) + html_dest = os.path.join(input_dir, 'state_area.html') + if not (os.path.exists(html_dest) and os.path.getsize(html_dest) > 0): + logging.info('Downloading state area reference table from %s...', + STATE_AREA_URL) + download_file_atomic(STATE_AREA_URL, html_dest, session) + downloaded_paths.append(html_dest) + + return downloaded_paths + + +@contextmanager +def _open_input_stream(filepath: str): + """Context manager to stream-read a plain text file or the first file in a zip archive.""" + if filepath.endswith('.zip'): + with zipfile.ZipFile(filepath) as zf: + inner_names = [ + n for n in zf.namelist() if not n.startswith('__MACOSX') + ] + with zf.open(inner_names[0]) as f: + yield f + else: + with open(filepath, 'r', encoding='latin1') as f: + yield f + + +def process(input_dir: str, + output_dir: str, + year_spec: str = DEFAULT_YEARS) -> Tuple[str, str]: + """Processes gazetteer raw files into final CSV and TMCF across specified years. + + Args: + input_dir: Directory containing downloaded raw files. + output_dir: Directory where surface_area.csv and surface_area.tmcf will be saved. + year_spec: Year range or specification (e.g. '2018-2025' or '2018'). + + Returns: + Tuple of (csv_path, tmcf_path). + """ + os.makedirs(output_dir, exist_ok=True) + files_in_input = os.listdir(input_dir) if os.path.exists(input_dir) else [] + + if str(year_spec).strip().lower() in ('auto', 'all'): + input_years = set() + for filename in files_in_input: + m = re.match(r'^([0-9]{4})_', filename) + if m and int(m.group(1)) >= 2018: + input_years.add(m.group(1)) + if input_years: + years = sorted(input_years) + else: + years = parse_years(year_spec) + else: + years = parse_years(year_spec) + + all_rows = [] + + for year in years: + logging.info('Processing data for year %s...', year) + year_records: Dict[str, float] = {} + + # 1. Process States + state_gaz_found = False + for filename in files_in_input: + if filename.startswith(f'{year}_') and re.match( + STATE_GAZETTEER_PATTERN, filename): + filepath = os.path.join(input_dir, filename) + logging.info('Processing state gazetteer: %s', filename) + with _open_input_stream(filepath) as f: + year_records.update(parse_state_gazetteer(f)) + state_gaz_found = True + break + + if not state_gaz_found: + html_path = os.path.join(input_dir, 'state_area.html') + if os.path.exists(html_path): + logging.info('Processing state area reference table for %s: %s', + year, html_path) + with open(html_path, 'r', encoding='utf-8', + errors='replace') as f: + year_records.update(parse_state_area_html(f.read())) + else: + logging.warning('No state data file found in %s', input_dir) + + logging.info('Year %s states loaded: %d', year, len(year_records)) + + # 2. Process Gazetteer specs + for pattern, prefix, geoid_col, desc in GAZETTEER_FILE_SPECS: + matched_file = None + for filename in files_in_input: + if filename.startswith(f'{year}_') and re.match( + pattern, filename): + matched_file = filename + break + # Fallback for test datasets without year prefix + if not matched_file and len(years) == 1: + for filename in files_in_input: + if re.match(pattern, filename) and not re.match( + r'^[0-9]{4}_', filename): + matched_file = filename + break + + if not matched_file: + logging.warning('No file found for %s %s (pattern: %s)', year, + desc, pattern) + continue + + filepath = os.path.join(input_dir, matched_file) + logging.info('Processing %s (%s)...', matched_file, desc) + with _open_input_stream(filepath) as f: + records = parse_gazetteer_data(f, prefix, geoid_col) + logging.info(' Loaded %d records for %s %s', len(records), + year, desc) + year_records.update(records) + + logging.info('Year %s total unique entities: %d', year, + len(year_records)) + for dcid, area in year_records.items(): + all_rows.append((dcid, str(year), area)) + + logging.info('Total observations across all years: %d', len(all_rows)) + + # 3. Create sorted DataFrame + all_rows.sort(key=lambda r: (r[0], r[1])) + output_df = pd.DataFrame(all_rows, + columns=['dcid', 'observationDate', 'SurfaceArea']) + + # 4. Write CSV + csv_path = os.path.join(output_dir, OUTPUT_CSV) + output_df.to_csv(csv_path, index=False) + logging.info('Cleaned CSV successfully written to %s (%d rows)', csv_path, + len(output_df)) + + # 5. Write TMCF + tmcf_path = os.path.join(output_dir, OUTPUT_TMCF) + with open(tmcf_path, 'w', encoding='utf-8') as f: + f.write(TMCF_TEMPLATE) + logging.info('Template MCF successfully written to %s', tmcf_path) + + return csv_path, tmcf_path + + +def main(argv): + """Main entry point for command-line execution.""" + del argv # Unused + input_dir = _FLAGS.input_dir or os.path.join(_MODULE_DIR, INPUT_DIR) + output_dir = _FLAGS.output_dir or os.path.join(_MODULE_DIR, OUTPUT_DIR) + year_spec = _FLAGS.year or _FLAGS.years or DEFAULT_YEARS + + if _FLAGS.mode in ('download', 'all'): + logging.info('Running download mode...') + download_files(input_dir, year_spec) + + if _FLAGS.mode in ('process', 'all'): + logging.info('Running process mode...') + process(input_dir, output_dir, year_spec) + + +if __name__ == '__main__': + app.run(main) diff --git a/scripts/us_census/surface_area/preprocess_test.py b/scripts/us_census/surface_area/preprocess_test.py new file mode 100644 index 0000000000..4ce1c7e6be --- /dev/null +++ b/scripts/us_census/surface_area/preprocess_test.py @@ -0,0 +1,175 @@ +# Copyright 2024 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. +"""Unit tests for US Census Surface Area preprocessing module.""" + +import io +import os +import sys +import tempfile +import unittest +from unittest import mock + +import pandas as pd + +_MODULE_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, _MODULE_DIR) + +# pylint: disable=wrong-import-position +from preprocess import (OUTPUT_CSV, OUTPUT_TMCF, calc_surface_area, + download_file_atomic, parse_gazetteer_data, + parse_state_area_html, parse_state_gazetteer, process) +# pylint: enable=wrong-import-position + +_TEST_DATA_DIR = 'test_data' + + +class TestSurfaceAreaPreprocess(unittest.TestCase): + """Tests for US Census Surface Area preprocessing.""" + + def test_calc_surface_area(self): + """Tests the conversion from square meters to square miles.""" + self.assertEqual(calc_surface_area(2589988.11, 0), 1.0) + self.assertEqual(calc_surface_area(0, 2589988.11), 1.0) + # Autauga County, AL (FIPS 01001): ALAND=1539602123, AWATER=25706961 + self.assertEqual(calc_surface_area(1539602123, 25706961), 604.3692) + + def test_parse_gazetteer_data_tab_and_pipe(self): + """Tests parsing gazetteer with both tab and pipe separators.""" + # Tab-separated (legacy 2018 format) + sample_tsv = ( + 'USPS\tGEOID\tANSICODE\tNAME\tALAND\tAWATER\n' + 'AL\t01001\t00161526\tAutauga County\t1539602123\t25706961\n' + 'AL\t01003\t00161527\tBaldwin County\t4117546676\t1133055836\n') + records_tab = parse_gazetteer_data(io.StringIO(sample_tsv), + dcid_prefix='geoId/') + self.assertEqual(len(records_tab), 2) + self.assertEqual(records_tab['geoId/01001'], 604.3692) + self.assertEqual(records_tab['geoId/01003'], 2027.269) + + # Pipe-separated (modern 2025 format) + sample_psv = ( + 'USPS|GEOID|GEOIDFQ|ANSICODE|NAME|ALAND|AWATER|ALAND_SQMI|AWATER_SQMI|' + 'INTPTLAT|INTPTLONG\n' + 'AL|01001|0500000US01001|00161526|Autauga County|1539602123|25706961|' + '594.455|9.914|32.532237|-86.64644\n') + records_pipe = parse_gazetteer_data(io.StringIO(sample_psv), + dcid_prefix='geoId/') + self.assertEqual(len(records_pipe), 1) + self.assertEqual(records_pipe['geoId/01001'], 604.3692) + + def test_parse_state_area_html(self): + """Tests parsing state area reference HTML table.""" + sample_html = ( + '\n' + '' + '\n' + '' + '\n' + '
Alabama52,420135,76750,645131,1711,7754,597
Alaska665,3841,723,337570,6411,477,95394,743245,384
') + records = parse_state_area_html(sample_html) + self.assertEqual(len(records), 2) + self.assertEqual(records['geoId/01'], 52420.0) + self.assertEqual(records['geoId/02'], 665384.0) + + def test_parse_state_gazetteer(self): + """Tests parsing modern state gazetteer table.""" + sample_state_txt = ( + 'USPS|GEOID|GEOIDFQ|NAME|ALAND|AWATER|ALAND_SQMI|AWATER_SQMI|' + 'INTPTLAT|INTPTLONG\n' + 'AL|01|0400000US01|Alabama|131186429591|4580729056|50651.366|1768.629|' + '32.739579|-86.843447\n') + records = parse_state_gazetteer(io.StringIO(sample_state_txt)) + self.assertEqual(len(records), 1) + self.assertIn('geoId/01', records) + self.assertAlmostEqual(records['geoId/01'], 52420.0, places=1) + + def test_process_pipeline_against_expected(self): + """Tests the end-to-end process() pipeline using test data.""" + test_input_dir = os.path.join(_MODULE_DIR, _TEST_DATA_DIR, + 'input_files') + expected_dir = os.path.join(_MODULE_DIR, _TEST_DATA_DIR, + 'expected_files') + + with tempfile.TemporaryDirectory() as temp_out_dir: + csv_path, tmcf_path = process(test_input_dir, temp_out_dir, '2018') + + # Verify CSV exists and matches expected + self.assertTrue(os.path.exists(csv_path)) + actual_df = pd.read_csv(csv_path, dtype={'observationDate': str}) + expected_csv_path = os.path.join(expected_dir, OUTPUT_CSV) + expected_df = pd.read_csv(expected_csv_path, + dtype={'observationDate': str}) + pd.testing.assert_frame_equal(actual_df, expected_df) + + # Verify TMCF exists and matches expected + self.assertTrue(os.path.exists(tmcf_path)) + with open(tmcf_path, 'r', encoding='utf-8') as f: + actual_tmcf = f.read() + expected_tmcf_path = os.path.join(expected_dir, OUTPUT_TMCF) + with open(expected_tmcf_path, 'r', encoding='utf-8') as f: + expected_tmcf = f.read() + self.assertEqual(actual_tmcf.strip(), expected_tmcf.strip()) + + def test_download_file_atomic_success(self): + """Tests atomic download when response is valid.""" + mock_session = mock.MagicMock() + mock_response = mock.MagicMock() + mock_response.iter_content.return_value = [b'test-data'] + mock_response.__enter__.return_value = mock_response + mock_session.get.return_value = mock_response + + with tempfile.TemporaryDirectory() as temp_dir: + dest_file = os.path.join(temp_dir, 'dest.txt') + download_file_atomic('http://example.com/test.txt', dest_file, + mock_session) + self.assertTrue(os.path.exists(dest_file)) + with open(dest_file, 'rb') as f: + self.assertEqual(f.read(), b'test-data') + + def test_download_file_atomic_empty_error(self): + """Tests atomic download raises RuntimeError on 0-byte file.""" + mock_session = mock.MagicMock() + mock_response = mock.MagicMock() + mock_response.iter_content.return_value = [] + mock_response.__enter__.return_value = mock_response + mock_session.get.return_value = mock_response + + with tempfile.TemporaryDirectory() as temp_dir: + dest_file = os.path.join(temp_dir, 'empty.txt') + with self.assertRaises(RuntimeError): + download_file_atomic('http://example.com/empty.txt', dest_file, + mock_session) + self.assertFalse(os.path.exists(dest_file)) + + def test_fallback_matching_excludes_other_year_files(self): + """Tests that fallback gazetteer matching ignores files with other year prefixes.""" + with tempfile.TemporaryDirectory() as temp_in_dir, \ + tempfile.TemporaryDirectory() as temp_out_dir: + # Create a file for 2018 in the input directory + other_year_file = os.path.join(temp_in_dir, + '2018_Gaz_counties_national.txt') + with open(other_year_file, 'w', encoding='utf-8') as f: + f.write( + 'USPS\tGEOID\tANSICODE\tNAME\tALAND\tAWATER\n' + 'AL\t01001\t00161526\tAutauga County\t1539602123\t25706961\n' + ) + + # Process for year 2020: 2018 file should not be matched + csv_path, _ = process(temp_in_dir, temp_out_dir, '2020') + df = pd.read_csv(csv_path) + self.assertEqual(len(df), 0) + + +if __name__ == '__main__': + unittest.main() diff --git a/scripts/us_census/surface_area/test_data/expected_files/surface_area.csv b/scripts/us_census/surface_area/test_data/expected_files/surface_area.csv new file mode 100644 index 0000000000..8009407590 --- /dev/null +++ b/scripts/us_census/surface_area/test_data/expected_files/surface_area.csv @@ -0,0 +1,21 @@ +dcid,observationDate,SurfaceArea +geoId/01,2018,52420.0 +geoId/01001,2018,604.3692 +geoId/0100100,2018,3.0109 +geoId/01001020100,2018,3.8017 +geoId/01001020200,2018,1.2862 +geoId/0100124,2018,15.5842 +geoId/0100190171,2018,189.9814 +geoId/0100190315,2018,149.9778 +geoId/01003,2018,2027.269 +geoId/0101,2018,6945.5427 +geoId/0102,2018,10260.08 +geoId/02,2018,665384.0 +geoId/C10260,2018,67.0959 +geoId/C10300,2018,761.4026 +geoId/sch0100001,2018,91.0187 +geoId/sch0100003,2018,3.5055 +geoId/sch0100195,2018,33.176 +geoId/sch0400004,2018,81.7505 +geoId/sch0400082,2018,1075.6082 +geoId/sch0400450,2018,94.7422 diff --git a/scripts/us_census/surface_area/test_data/expected_files/surface_area.tmcf b/scripts/us_census/surface_area/test_data/expected_files/surface_area.tmcf new file mode 100644 index 0000000000..ed11470eba --- /dev/null +++ b/scripts/us_census/surface_area/test_data/expected_files/surface_area.tmcf @@ -0,0 +1,7 @@ +Node: E:Data->E0 +typeOf: schema:StatVarObservation +variableMeasured: dcs:SurfaceArea +observationAbout: C:Data->dcid +observationDate: C:Data->observationDate +value: C:Data->SurfaceArea +unit: dcs:SquareMile diff --git a/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_116CDs_national.txt b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_116CDs_national.txt new file mode 100644 index 0000000000..21fc505f5a --- /dev/null +++ b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_116CDs_national.txt @@ -0,0 +1,3 @@ +USPS GEOID ALAND AWATER ALAND_SQMI AWATER_SQMI INTPTLAT INTPTLONG +AL 0101 15713197908 2275675091 6066.9 878.643 31.002052 -87.787972 +AL 0102 26269902298 303582941 10142.866 117.214 31.702085 -86.076842 diff --git a/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_cbsa_national.txt b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_cbsa_national.txt new file mode 100644 index 0000000000..6a8194345d --- /dev/null +++ b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_cbsa_national.txt @@ -0,0 +1,3 @@ +CSAFP GEOID NAME CBSA_TYPE ALAND AWATER ALAND_SQMI AWATER_SQMI INTPTLAT INTPTLONG +434 10260 Adjuntas, PR Micro Area 2 172725731 1051789 66.69 0.406 18.181611 -66.758165 +220 10300 Adrian, MI Micro Area 2 1941541486 30482189 749.633 11.769 41.896022 -84.074356 diff --git a/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_counties_national.txt b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_counties_national.txt new file mode 100644 index 0000000000..2e52b25a76 --- /dev/null +++ b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_counties_national.txt @@ -0,0 +1,3 @@ +USPS GEOID ANSICODE NAME ALAND AWATER ALAND_SQMI AWATER_SQMI INTPTLAT INTPTLONG +AL 01001 00161526 Autauga County 1539602123 25706961 594.444 9.926 32.532237 -86.64644 +AL 01003 00161527 Baldwin County 4117546676 1133055836 1589.794 437.475 30.659218 -87.746067 diff --git a/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_cousubs_national.txt b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_cousubs_national.txt new file mode 100644 index 0000000000..649f826ba7 --- /dev/null +++ b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_cousubs_national.txt @@ -0,0 +1,3 @@ +USPS GEOID ANSICODE NAME FUNCSTAT ALAND AWATER ALAND_SQMI AWATER_SQMI INTPTLAT INTPTLONG +AL 0100190171 00161593 Autaugaville CCD S 478151753 13897907 184.615 5.366 32.457413 -86.729003 +AL 0100190315 00165647 Billingsley CCD S 386865281 1575402 149.37 0.608 32.606138 -86.748985 diff --git a/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_elsd_national.txt b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_elsd_national.txt new file mode 100644 index 0000000000..dfa147d409 --- /dev/null +++ b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_elsd_national.txt @@ -0,0 +1,3 @@ +USPS GEOID NAME LOGRADE HIGRADE ALAND AWATER ALAND_SQMI AWATER_SQMI INTPTLAT INTPTLONG +AL 0100195 Pike Road City School District KG 12 85146878 778561 32.875 0.301 32.283188 -85.96568 +AZ 0400004 Clarkdale-Jerome Elementary District PK 08 211351942 380983 81.603 0.147 34.748857 -112.112028 diff --git a/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_place_national.txt b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_place_national.txt new file mode 100644 index 0000000000..d212887488 --- /dev/null +++ b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_place_national.txt @@ -0,0 +1,3 @@ +USPS GEOID ANSICODE NAME LSAD FUNCSTAT ALAND AWATER ALAND_SQMI AWATER_SQMI INTPTLAT INTPTLONG +AL 0100100 02582661 Abanda CDP 57 S 7764034 34284 2.998 0.013 33.091627 -85.527029 +AL 0100124 02403054 Abbeville city 25 A 40255362 107642 15.543 0.042 31.564689 -85.259124 diff --git a/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_scsd_national.txt b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_scsd_national.txt new file mode 100644 index 0000000000..25637e79f9 --- /dev/null +++ b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_scsd_national.txt @@ -0,0 +1,3 @@ +USPS GEOID NAME LOGRADE HIGRADE ALAND AWATER ALAND_SQMI AWATER_SQMI INTPTLAT INTPTLONG +AZ 0400082 Colorado River Union High School District 09 12 2711174269 74638171 1046.79 28.818 35.105214 -114.467834 +AZ 0400450 Agua Fria Union High School District 09 12 244645640 735546 94.458 0.284 33.481427 -112.407871 diff --git a/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_tracts_national.txt b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_tracts_national.txt new file mode 100644 index 0000000000..b07c8071e1 --- /dev/null +++ b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_tracts_national.txt @@ -0,0 +1,3 @@ +USPS GEOID ALAND AWATER ALAND_SQMI AWATER_SQMI INTPTLAT INTPTLONG +AL 01001020100 9817813 28435 3.791 0.011 32.4819591 -86.4913377 +AL 01001020200 3325679 5669 1.284 0.002 32.475758 -86.4724678 diff --git a/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_unsd_national.txt b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_unsd_national.txt new file mode 100644 index 0000000000..9707cf554e --- /dev/null +++ b/scripts/us_census/surface_area/test_data/input_files/2018_Gaz_unsd_national.txt @@ -0,0 +1,3 @@ +USPS GEOID NAME LOGRADE HIGRADE ALAND AWATER ALAND_SQMI AWATER_SQMI INTPTLAT INTPTLONG +AL 0100001 Fort Rucker School District KG 12 232935772 2801707 89.937 1.082 31.409737 -85.745807 +AL 0100003 Maxwell AFB School District KG 12 8706782 372428 3.362 0.144 32.380944 -86.363749 diff --git a/scripts/us_census/surface_area/test_data/input_files/state_area.html b/scripts/us_census/surface_area/test_data/input_files/state_area.html new file mode 100644 index 0000000000..305149bae9 --- /dev/null +++ b/scripts/us_census/surface_area/test_data/input_files/state_area.html @@ -0,0 +1,10 @@ + + + + + + + +
StateTotal Sq MiTotal Sq KmLand Sq MiLand Sq KmWater Sq MiWater Sq Km
Alabama52,420135,76750,645131,1711,7754,597
Alaska665,3841,723,337570,6411,477,95394,743245,384
+ + \ No newline at end of file diff --git a/scripts/us_census/surface_area/validation_config.json b/scripts/us_census/surface_area/validation_config.json new file mode 100644 index 0000000000..b0b5bcbd94 --- /dev/null +++ b/scripts/us_census/surface_area/validation_config.json @@ -0,0 +1,22 @@ +{ + "schema_version": "1.0", + "rules": [ + { + "rule_id": "check_deleted_records_percent", + "description": "Strictly enforce historical deletion threshold of 0.1% for boundary adjustments", + "validator": "DELETED_RECORDS_PERCENT", + "params": { + "threshold": 0.1 + } + }, + { + "rule_id": "check_max_date_freshness", + "description": "Verifies that the dataset's maximum observation date meets freshness requirements (within 2 years of current year)", + "validator": "SQL_VALIDATOR", + "params": { + "query": "SELECT MAX(CAST(LEFT(CAST(MaxDate AS VARCHAR), 4) AS INT)) AS max_year FROM stats", + "condition": "max_year >= (EXTRACT(YEAR FROM CURRENT_DATE) - 2)" + } + } + ] +}