diff --git a/scripts/us_cdc/cdc500_state/README.md b/scripts/us_cdc/cdc500_state/README.md index 79f5abb00c..e651276c93 100644 --- a/scripts/us_cdc/cdc500_state/README.md +++ b/scripts/us_cdc/cdc500_state/README.md @@ -5,6 +5,8 @@ Author: Padma Gundapaneni @padma-g ## Table of Contents 1. [About the Dataset](#about-the-dataset) 1. [Overview](#overview) + 2. [Data Sources and Tables](#data-sources-and-tables) + 3. [Aggregation Methodology](#aggregation-methodology) 2. [About the Import](#about-the-import) 1. [Artifacts](#artifacts) 2. [Import Procedure](#import-procedure) @@ -12,29 +14,52 @@ Author: Padma Gundapaneni @padma-g ## About the Dataset ### Overview -The state level data is aggragated from city level data coming from CDC500 import. +The state-level dataset calculates aggregated health indicator prevalence estimates for US states from the city-level CDC 500 Cities (`CDC500`) project data, weighted by corresponding Census ACS 5-Year population counts. -To get the data for this import run: -```bash -$ python3 process.py -``` +### Data Sources and Tables + +The aggregation script queries Google Cloud BigQuery graph tables in dataset `datcom-store.spanner_dc_graph_prod_DEFAULT`: + +1. **`TimeSeries`**: + - **CDC 500 Series**: Identifies CDC 500 Statistical Variables (`provenance = 'dc/base/CDC500'` and `variable_measured LIKE 'Percent_%'`) and extracts their measurement methods (`measurement_method`). It maps each percentage health metric to its appropriate denominator demographic cohort StatVar (e.g., `Count_Person_18OrMoreYears`, `Count_Person_Female_50To74Years`, `Count_Person_Female_21To65Years`, `Count_Person_65OrMoreYears`, etc.). + - **Census ACS 5-Year Series**: Filters and joins population counts from Census ACS 5-Year Survey (`provenance = 'dc/base/CensusACS5YearSurvey'`). + +2. **`Observation`**: + - **Health Indicator Percentages**: Fetches city-level percentage values (`value AS percent`), observation dates (`date`), and city geoIds (`entity1 LIKE 'geoId/%'`) for CDC 500 StatVars. + - **City Cohort Populations**: Fetches city-level population counts (`value AS population`) for the corresponding demographic cohort StatVars. + +### Aggregation Methodology + +For each state, indicator StatVar, and observation date: +- City observations are joined with their corresponding demographic population counts. +- City geoIds (`geoId/XXXXXXX`) are mapped to state geoIds (`geoId/XX`) using the first 8 characters (including the prefix). +- State-level prevalence percentages are computed as a population-weighted average: + +$$\text{State Percent} = \frac{\sum (\text{City Population} \times \text{City Percent})}{\sum \text{City Population}}$$ + +The output measurement method is prefixed with `dcAggregate/` (e.g., `dcAggregate/CrudePrevalence`). ## About the Import ### Artifacts #### Scripts -[`process.py`](https://github.com/datacommonsorg/data/blob/master//scripts/us_cdc/cdc500_state/process.py) +[`process.py`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/process.py) +#### Unit Tests +[`process_test.py`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/process_test.py) -#### tMCFs +#### tMCF Template [`cdc500_state.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/cdc500_state.tmcf) +#### Validation Config +[`validation_config.json`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/cdc500_state/validation_config.json) + ### Import Procedure #### Data Download and Processing Steps -To get the data for this import run: +To run the BigQuery aggregation and generate the output CSV: ```bash $ python3 process.py diff --git a/scripts/us_cdc/cdc500_state/cdc500_state.tmcf b/scripts/us_cdc/cdc500_state/cdc500_state.tmcf index 73a294f9cf..b7d6106b85 100644 --- a/scripts/us_cdc/cdc500_state/cdc500_state.tmcf +++ b/scripts/us_cdc/cdc500_state/cdc500_state.tmcf @@ -4,5 +4,7 @@ variableMeasured: C:CDC->statvar observationAbout: C:CDC->observation_about observationDate: C:CDC->observation_date value: C:CDC->percent +unit: Percent +scalingFactor: 100 measurementMethod: C:CDC->measurement_method observationPeriod: "P1Y" \ No newline at end of file diff --git a/scripts/us_cdc/cdc500_state/manifest.json b/scripts/us_cdc/cdc500_state/manifest.json index eae459b1c6..4a2ac714b8 100644 --- a/scripts/us_cdc/cdc500_state/manifest.json +++ b/scripts/us_cdc/cdc500_state/manifest.json @@ -21,10 +21,8 @@ "memory": 64, "disk": 100 }, - "source_files": [ - "CDC500State_Output/CDC500State_Output.csv" - ], - "cron_schedule": "0 1 * * 1" + "cron_schedule": "0 1 * * 1", + "validation_config_file": "validation_config.json" } ] } \ No newline at end of file diff --git a/scripts/us_cdc/cdc500_state/process.py b/scripts/us_cdc/cdc500_state/process.py index 0115b96e43..e20cdf8a11 100644 --- a/scripts/us_cdc/cdc500_state/process.py +++ b/scripts/us_cdc/cdc500_state/process.py @@ -1,4 +1,4 @@ -# Copyright 2021 Google LLC +# 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. @@ -11,75 +11,128 @@ # 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. +"""Processes CDC 500 cities data into aggregated state-level health indicators.""" import os +from absl import app +from absl import flags from absl import logging from google.cloud import bigquery +import pandas as pd +_FLAGS = flags.FLAGS _MODULE_DIR = os.path.dirname(os.path.abspath(__file__)) -_OUTPUT_FILE_PATH = os.path.join(_MODULE_DIR + '/CDC500State_Output') -if not os.path.exists(_OUTPUT_FILE_PATH): - os.mkdir(_OUTPUT_FILE_PATH) +_DEFAULT_OUTPUT_DIR = os.path.join(_MODULE_DIR, 'CDC500State_Output') -query = """ -SELECT distinct * from( -SELECT - statvar, - SUBSTR(observation_about,0,8) as observation_about, - observation_date, - CONCAT('dcAggregate/',measurement_method) as measurement_method, - population_statvar, - SUM(CAST(pop_count AS FLOAT64))*100/SUM(CAST(population AS FLOAT64)) as percent -FROM -( +flags.DEFINE_string('output_dir', _DEFAULT_OUTPUT_DIR, + 'Directory to write output CSV.') + +QUERY = """ +WITH cdc_sv AS ( + SELECT + variable_measured AS cdc500, + CASE + WHEN variable_measured LIKE '%Female_50To74Years%' OR variable_measured LIKE '%50To74Years_Female%' THEN 'Count_Person_Female_50To74Years' + WHEN variable_measured LIKE '%Female_21To65Years%' OR variable_measured LIKE '%21To65Years_Female%' THEN 'Count_Person_Female_21To65Years' + WHEN variable_measured LIKE '%Female_65OrMoreYears%' OR variable_measured LIKE '%65OrMoreYears_Female%' THEN 'Count_Person_Female_65OrMoreYears' + WHEN variable_measured LIKE '%Male_65OrMoreYears%' OR variable_measured LIKE '%65OrMoreYears_Male%' THEN 'Count_Person_Male_65OrMoreYears' + WHEN variable_measured LIKE '%65OrMoreYears%' THEN 'Count_Person_65OrMoreYears' + WHEN variable_measured LIKE '%18To64Years%' THEN 'Count_Person_18To64Years' + WHEN variable_measured LIKE '%18OrMoreYears%' THEN 'Count_Person_18OrMoreYears' + ELSE 'Count_Person' + END AS pop_statvar + FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.TimeSeries` + WHERE provenance = 'dc/base/CDC500' + AND variable_measured LIKE 'Percent_%' + GROUP BY cdc500, pop_statvar +), + +svo_percent AS ( + SELECT + O.variable_measured AS statvar, + O.entity1 AS observation_about, + O.date AS observation_date, + O.value AS percent, + T.measurement_method AS measurement_method, + cdc_sv.pop_statvar + FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.Observation` AS O + INNER JOIN `datcom-store.spanner_dc_graph_prod_DEFAULT.TimeSeries` AS T + ON O.variable_measured = T.variable_measured + AND O.entity1 = T.entity1 + AND O.facet_id = T.facet_id + AND T.provenance = 'dc/base/CDC500' + AND T.variable_measured LIKE 'Percent_%' + INNER JOIN cdc_sv + ON O.variable_measured = cdc_sv.cdc500 + WHERE O.entity1 LIKE 'geoId/%' + AND O.variable_measured LIKE 'Percent_%' +), + +svo_count AS ( SELECT - SVO1.variable_measured as statvar, - SVO1.observation_about as observation_about, - SVO1.observation_date as observation_date, - SVO1.value as percent, - SVO1.measurement_method as measurement_method, - SVO2.variable_measured as population_statvar, - SVO2.value as population, - CAST(SVO2.value AS FLOAT64) * CAST(SVO1.value AS FLOAT64) / 100 as pop_count - FROM `datcom-store.dc_kg_latest.StatVarObservation` as SVO1 - JOIN `datcom-store.dc_kg_latest.StatVarObservation` as SVO2 ON TRUE - JOIN ( - # Get the statvars and corresponding population statvar - # with ‘Percent_’ replaced with ‘Count_’ and - # dropping the non-age, non-gender constraints. - SELECT - SVO.variable_measured as CDC500, - CONCAT('Count_', REGEXP_SUBSTR(SVO.variable_measured, '(Person_.*ale|Person_.*Years|Person)')) as pop_statvar - FROM `datcom-store.dc_kg_latest.StatVarObservation` as SVO - WHERE - SVO.prov_id = 'dc/base/CDC500' - AND SVO.variable_measured like 'Percent_%' - GROUP BY CDC500, pop_statvar - ) AS CDC_SV ON TRUE - WHERE - SVO1.prov_id = 'dc/base/CDC500' - AND SVO1.variable_measured LIKE 'Percent%' - AND SVO1.observation_about = SVO2.observation_about - AND SVO1.observation_date = SVO2.observation_date - AND SVO1.variable_measured = CDC_SV.CDC500 - AND SVO2.variable_measured = CDC_SV.pop_statvar - AND SVO1.observation_about like "geoId/%" -) group by 1,2,3,4,5 + O.variable_measured AS population_statvar, + O.entity1 AS observation_about, + O.date AS observation_date, + O.value AS population + FROM `datcom-store.spanner_dc_graph_prod_DEFAULT.Observation` AS O + INNER JOIN `datcom-store.spanner_dc_graph_prod_DEFAULT.TimeSeries` AS T + ON O.variable_measured = T.variable_measured + AND O.entity1 = T.entity1 + AND O.facet_id = T.facet_id + AND T.provenance = 'dc/base/CensusACS5YearSurvey' + INNER JOIN ( + SELECT DISTINCT pop_statvar + FROM cdc_sv + ) AS pop + ON O.variable_measured = pop.pop_statvar + WHERE O.entity1 LIKE 'geoId/%' ) + +SELECT + p.statvar, + SUBSTR(p.observation_about, 1, 8) AS observation_about, + p.observation_date, + CONCAT('dcAggregate/', p.measurement_method) AS measurement_method, + p.pop_statvar AS population_statvar, + SAFE_DIVIDE( + SUM(CAST(c.population AS FLOAT64) * CAST(p.percent AS FLOAT64)), + SUM(CAST(c.population AS FLOAT64)) + ) AS percent +FROM svo_percent AS p +INNER JOIN svo_count AS c + ON p.observation_about = c.observation_about + AND p.observation_date = c.observation_date + AND p.pop_statvar = c.population_statvar +GROUP BY 1, 2, 3, 4, 5 """ -client = bigquery.Client() -try: - logging.info("Running the query") - query_job = client.query(query) -except Exception as e: - logging.fatal(f"Error faced while running the query {e}") -try: - logging.info("Converting to dataframe") - results = query_job.to_dataframe() -except Exception as e: - logging.info(f"Error faced while fetching results: {e}") +def run_process(client: bigquery.Client, output_file: str) -> None: + """Executes the BigQuery query and writes the resulting DataFrame to output_file.""" + logging.info("Running BigQuery aggregation query...") + try: + query_job = client.query(QUERY) + except Exception as e: + logging.error("Failed to submit BigQuery query: %s", e) + raise + + logging.info("Fetching query results into dataframe...") + try: + df = query_job.to_dataframe() + except Exception as e: + logging.error("Failed to fetch query results into dataframe: %s", e) + raise + + output_dir = os.path.dirname(output_file) + if output_dir: + os.makedirs(output_dir, exist_ok=True) + logging.info("Writing %d rows to %s", len(df), output_file) + df.to_csv(output_file, index=False) + +def main(argv): + del argv # Unused. + client = bigquery.Client() + output_file = os.path.join(_FLAGS.output_dir, 'CDC500State_Output.csv') + run_process(client, output_file) -logging.info("Writing output to CSV") -output_file = os.path.join(_OUTPUT_FILE_PATH + "/CDC500State_Output.csv") -results.to_csv(output_file, index=False) +if __name__ == '__main__': + app.run(main) diff --git a/scripts/us_cdc/cdc500_state/process_test.py b/scripts/us_cdc/cdc500_state/process_test.py new file mode 100644 index 0000000000..eaec37e2f0 --- /dev/null +++ b/scripts/us_cdc/cdc500_state/process_test.py @@ -0,0 +1,66 @@ +# 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 +# +# https://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 CDC 500 State aggregation script.""" + +import os +import tempfile +import unittest +from unittest import mock +import pandas as pd + +from scripts.us_cdc.cdc500_state import process + +class CDC500StateProcessTest(unittest.TestCase): + + def test_query_constants(self): + query = process.QUERY + self.assertIn("spanner_dc_graph_prod_DEFAULT.TimeSeries", query) + self.assertIn("spanner_dc_graph_prod_DEFAULT.Observation", query) + self.assertIn("dc/base/CDC500", query) + self.assertIn("dc/base/CensusACS5YearSurvey", query) + self.assertIn("SAFE_DIVIDE", query) + self.assertIn("SUBSTR(p.observation_about, 1, 8)", query) + + def test_run_process_success(self): + mock_client = mock.MagicMock() + sample_data = pd.DataFrame({ + 'statvar': ['Percent_Person_18OrMoreYears_WithAnyDisability'], + 'observation_about': ['geoId/06'], + 'observation_date': ['2022'], + 'measurement_method': ['dcAggregate/CrudePrevalence'], + 'population_statvar': ['Count_Person_18OrMoreYears'], + 'percent': [29.6479] + }) + mock_client.query.return_value.to_dataframe.return_value = sample_data + + with tempfile.TemporaryDirectory() as tmp_dir: + output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv') + process.run_process(mock_client, output_file) + + mock_client.query.assert_called_once() + self.assertTrue(os.path.exists(output_file)) + saved_df = pd.read_csv(output_file) + self.assertEqual(len(saved_df), 1) + self.assertEqual(saved_df['observation_about'].iloc[0], 'geoId/06') + + def test_run_process_query_error(self): + mock_client = mock.MagicMock() + mock_client.query.side_effect = Exception("BigQuery Access Denied") + with tempfile.TemporaryDirectory() as tmp_dir: + output_file = os.path.join(tmp_dir, 'CDC500State_Output.csv') + with self.assertRaises(Exception): + process.run_process(mock_client, output_file) + +if __name__ == '__main__': + unittest.main() diff --git a/scripts/us_cdc/cdc500_state/validation_config.json b/scripts/us_cdc/cdc500_state/validation_config.json new file mode 100644 index 0000000000..5ac0fba24f --- /dev/null +++ b/scripts/us_cdc/cdc500_state/validation_config.json @@ -0,0 +1,30 @@ +{ + "schema_version": "1.0", + "rules": [ + { + "rule_id": "check_max_value_percentage", + "description": "Checks that all percentage StatVars do not exceed 100%.", + "validator": "MAX_VALUE_CHECK", + "params": { + "maximum": 100 + } + }, + { + "rule_id": "check_min_value_percentage", + "description": "Checks that all percentage StatVars are not below 0%.", + "validator": "MIN_VALUE_CHECK", + "params": { + "minimum": 0 + } + }, + { + "rule_id": "check_num_places_state_count", + "description": "Checks that state-level observations cover all 50-52 US state entities.", + "validator": "NUM_PLACES_COUNT", + "params": { + "minimum": 50, + "maximum": 52 + } + } + ] +}