ny_brfss_health_indicators

statvar_imports/us_newyork/ny_brfss_health_indicators/README.md

@@ -0,0 +1,90 @@
+# New York BRFSS Health Indicators Data
+
+## 1. Import Overview
+
+This project processes and imports health indicator prevalence rates across all 62 counties in New York State and New York City, provided by the New York State Department of Health. The dataset provides annual county-level estimates across 5 survey releases (2014, 2016, 2018, 2021, and 2024).
+
+*   **Source URL**: [https://health.data.ny.gov/Health/Behavioral-Risk-Factor-Surveillance-System-BRFSS-H/jsy7-eb4n/about_data](https://health.data.ny.gov/Health/Behavioral-Risk-Factor-Surveillance-System-BRFSS-H/jsy7-eb4n/about_data)
+*   **Import Type**: Automated
+*   **Source Data Availability**: Data is available for 2014, 2016, 2018, 2021, and 2024.
+*   **Release Frequency**: Periodic survey releases (biennial / triennial survey waves).
+*   **Notes**: This dataset provides county-level estimates across 75 health indicators spanning chronic disease, mental health, substance use, disability, immunizations, and social determinants of health. The data originates from the Behavioral Risk Factor Surveillance System (BRFSS).
+
+---
+
+## 2. Preprocessing Steps
+
+The import process involves querying the NYSDOH Socrata API and running a processing script on downloaded source data to generate the final artifacts for ingestion.
+
+*   **Input files**:
+    *   `input_files/`: This directory contains the raw unpivoted data file (`ny_brfss_health_indicators_raw.csv`) containing all 17,700 records across all survey years (2014, 2016, 2018, 2021, 2024).
+    *   `ny_brfss_health_indicators_metadata.csv`: Configuration file for the data processing script specifying column mappings, header row offset, and provenance URL.
+    *   `ny_brfss_health_indicators_pv_map.csv`: Property-value mapping file used by the processor to map indicators and county locations to Data Commons entities.
+    *   `validation_config.json`: Configuration defining historical deletion and date freshness validation rules.
+    *   `test_data/`: Sample input data and expected output files for integration testing.
+
+*   **Transformation pipeline**:
+    1.  The raw data is queried from the NYSDOH Socrata API using deterministic pagination (`$order: ':id'`) via `download.py` and placed in the `input_files/` directory.
+    2.  The `stat_var_processor.py` tool is run on the raw data against `ny_brfss_health_indicators_pv_map.csv` and `ny_brfss_health_indicators_metadata.csv`, referencing canonical schema `gs://unresolved_mcf/scripts/statvar/stat_vars.mcf`.
+    3.  The processor filters non-county regional rows, resolves county FIPS DCIDs, maps indicators to canonical or provisional StatVars, and generates the final `ny_brfss_health_indicators_output.csv`, `ny_brfss_health_indicators_output.tmcf`, and supporting StatVar and schema MCF files in the `output_files/` directory.
+    4.  Processor statistics and metrics are recorded in `counters/ny_brfss_health_indicators_counters.csv`.
+
+*   **Data Quality Checks**:
+    *   The `dc_generated/` directory contains `report.json` and `summary_report.csv`, which provide validation and summary statistics for the generated data.
+    *   Automated validation via `validator.py` evaluates the output against `validation_config.json` enforcing the historical deletion threshold (<= 0.1%) and date freshness (`CAST(MaxDate AS INTEGER) >= (EXTRACT(YEAR FROM CURRENT_DATE) - 3)`).
+
+---
+
+## 3. Automated Import
+
+This import is designed to be fully automated and autorefreshed. Future survey releases are automatically queried, processed, and validated.
+
+### Automated Steps
+1. The automated job triggers annually based on cron schedule `0 0 1 8 *` configured in `manifest.json`.
+2. `download.py` queries the NYSDOH Socrata API endpoint with deterministic pagination (`$order: ':id'`) and updates `input_files/ny_brfss_health_indicators_raw.csv`.
+3. `stat_var_processor.py` executes to regenerate the output CSV, TMCF, and MCF artifacts.
+4. `validator.py` enforces validation rules from `validation_config.json` before publication.
+
+---
+
+## 4. Script Execution Details
+
+To run the import pipeline, execute the processing scripts as detailed below.
+
+### Download the Data
+
+This script downloads the multi-year health indicators dataset from the NYSDOH Socrata API into `input_files/`:
+
+**Usage**:
+```bash
+python3 download.py
+```
+
+### Process the Data
+
+This script processes the raw input file to generate the final `ny_brfss_health_indicators_output.csv` file, `ny_brfss_health_indicators_output.tmcf` template, and supporting MCF files.
+
+**Usage**:
+```bash
+python3 ../../../tools/statvar_importer/stat_var_processor.py \
+  --input_data="input_files/ny_brfss_health_indicators_raw.csv" \
+  --pv_map=ny_brfss_health_indicators_pv_map.csv \
+  --config_file=ny_brfss_health_indicators_metadata.csv \
+  --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf \
+  --output_path=output_files/ny_brfss_health_indicators_output \
+  --output_counters=counters/ny_brfss_health_indicators_counters.csv
+```
+
+### Run Sample Integration Test
+
+This script verifies property-value mapping against the test data fixtures:
+
+**Usage**:
+```bash
+python3 ../../../tools/statvar_importer/stat_var_processor.py \
+  --input_data="test_data/sample_input.csv" \
+  --pv_map=ny_brfss_health_indicators_pv_map.csv \
+  --config_file=ny_brfss_health_indicators_metadata.csv \
+  --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf \
+  --output_path=test_data/sample_expected_output
+```

statvar_imports/us_newyork/ny_brfss_health_indicators/download.py

@@ -0,0 +1,135 @@
+# Copyright 2026 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.
+"""Automated Multi-Year Downloader for NYS BRFSS Health Indicators by County and Region.
+
+Downloads live county and region level health outcome indicators for New York State
+from the NYSDOH Socrata Open Data API (jsy7-eb4n) across all survey years (2014, 2016,
+2018, 2021, 2024), covering all 62 counties, 11 DSRIP regions, NYC, Rest of State,
+and Statewide.
+"""
+
+import os
+from absl import app
+from absl import flags
+from absl import logging
+import pandas as pd
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+FLAGS = flags.FLAGS
+
+flags.DEFINE_string(
+    'output_dir',
+    'input_files',
+    'Directory where downloaded input CSV files will be saved.',
+)
+flags.DEFINE_string(
+    'endpoint',
+    'https://health.data.ny.gov/resource/jsy7-eb4n.json',
+    'Health Data NY BRFSS Socrata API endpoint.',
+)
+
+
+def create_session() -> requests.Session:
+    """Creates a requests session configured with retries and connection pooling."""
+    session = requests.Session()
+    retries = Retry(
+        total=5,
+        backoff_factor=1.5,
+        status_forcelist=[429, 500, 502, 503, 504],
+        raise_on_status=False,
+    )
+    adapter = HTTPAdapter(max_retries=retries)
+    session.mount('https://', adapter)
+    session.mount('http://', adapter)
+    session.headers.update({
+        'User-Agent':
+            'Mozilla/5.0 (DataCommons Ingestion; +https://datacommons.org)',
+        'Accept':
+            'application/json, text/csv, */*',
+    })
+    return session
+
+
+def atomic_to_csv(df: pd.DataFrame, target_path: str) -> None:
+    """Writes a DataFrame to a CSV file atomically via a temporary file."""
+    temp_path = f'{target_path}.tmp'
+    df.to_csv(temp_path, index=False)
+    if not (os.path.exists(temp_path) and os.path.getsize(temp_path) > 0):
+        raise RuntimeError(
+            f'Atomic write failed: {temp_path} is empty or missing.'
+        )
+    os.replace(temp_path, target_path)
+
+
+def download_health_indicators(endpoint: str,
+                               output_dir: str) -> tuple[int, list[str]]:
+    """Downloads all health indicators for all NY counties and regions."""
+    os.makedirs(output_dir, exist_ok=True)
+
+    records = []
+    offset = 0
+    limit = 50000
+
+    with create_session() as session:
+        while True:
+            params = {
+                '$limit': limit,
+                '$offset': offset,
+                '$order': ':id',
+            }
+            logging.info('GET %s with params %s', endpoint, params)
+            resp = session.get(endpoint, params=params, timeout=60)
+            if resp.status_code != 200:
+                logging.error('Health Data NY API returned HTTP %d: %s',
+                              resp.status_code, resp.text[:200])
+                raise RuntimeError(
+                    f'Health Data NY API returned HTTP {resp.status_code}: {resp.text[:200]}'
+                )
+
+            chunk = resp.json()
+            if not chunk:
+                break
+            records.extend(chunk)
+            logging.info('Fetched %d records (offset %d).', len(chunk), offset)
+            offset += len(chunk)
+
+    if not records:
+        logging.error('Health Data NY API returned 0 records.')
+        raise RuntimeError('Health Data NY API returned 0 records.')
+
+    logging.info('Received %d total raw records from Health Data NY API.',
+                 len(records))
+    raw_df = pd.DataFrame(records)
+
+    # Save complete raw unpivoted dataset (all 17,000+ API records)
+    raw_file = os.path.join(output_dir, 'ny_brfss_health_indicators_raw.csv')
+    atomic_to_csv(raw_df, raw_file)
+    saved_files = [raw_file]
+    logging.info('Wrote raw unpivoted dataset (%d records) -> %s', len(raw_df),
+                 raw_file)
+
+    return len(raw_df), saved_files
+
+
+def main(_):
+    script_dir = os.path.dirname(os.path.abspath(__file__))
+    output_dir = (FLAGS.output_dir if os.path.isabs(FLAGS.output_dir) else
+                  os.path.join(script_dir, FLAGS.output_dir))
+    download_health_indicators(FLAGS.endpoint, output_dir)
+
+
+if __name__ == '__main__':
+    app.run(main)

statvar_imports/us_newyork/ny_brfss_health_indicators/manifest.json

@@ -0,0 +1,30 @@
+{
+    "import_specifications": [
+        {
+            "import_name": "NewYork_BRFSS_Health_Indicators",
+            "curator_emails": [
+                "support@datacommons.org"
+            ],
+            "provenance_url": "https://health.data.ny.gov/Health/Behavioral-Risk-Factor-Surveillance-System-BRFSS-H/jsy7-eb4n/about_data",
+            "provenance_description": "The Behavioral Risk Factor Surveillance System (BRFSS) Health Indicators dataset from the New York State Department of Health provides annual estimates of chronic disease prevalence across all counties in New York.",
+            "scripts": [
+                "download.py",
+                "../../../tools/statvar_importer/stat_var_processor.py --input_data=input_files/ny_brfss_health_indicators_raw.csv --pv_map=ny_brfss_health_indicators_pv_map.csv --config_file=ny_brfss_health_indicators_metadata.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path=output_files/ny_brfss_health_indicators_output --output_counters=counters/ny_brfss_health_indicators_counters.csv"
+            ],
+            "import_inputs": [
+                {
+                    "template_mcf": "output_files/ny_brfss_health_indicators_output.tmcf",
+                    "cleaned_csv": "output_files/ny_brfss_health_indicators_output.csv",
+                    "node_mcf": "output_files/*.mcf"
+                }
+            ],
+            "source_files": [
+                "input_files/*.csv",
+                "counters/*.csv",
+                "validation_config.json"
+            ],
+            "cron_schedule": "0 0 1 * *",
+            "validation_config_file": "validation_config.json"
+        }
+    ]
+}

statvar_imports/us_newyork/ny_brfss_health_indicators/ny_brfss_health_indicators_metadata.csv

@@ -0,0 +1,5 @@
+parameter,value
+url,https://health.data.ny.gov/Health/Behavioral-Risk-Factor-Surveillance-System-BRFSS-H/jsy7-eb4n/about_data
+output_columns,"observationAbout,observationDate,value,variableMeasured"
+header_rows,1
+mapped_columns,4