diff --git a/statvar_imports/ntia_internet_use_survey/commerce_ntia/README.md b/statvar_imports/ntia_internet_use_survey/commerce_ntia/README.md index eb6e1e8ea1..da991ab109 100644 --- a/statvar_imports/ntia_internet_use_survey/commerce_ntia/README.md +++ b/statvar_imports/ntia_internet_use_survey/commerce_ntia/README.md @@ -1,24 +1,42 @@ -# Commerce_NTIA +# commerce_ntia -- source: `https://www.ntia.gov/sites/default/files/data_central_downloads/datasets/ntia-analyze-table.csv` +## Import Metadata + +| Field | Value | +|---|---| +| **Import Type** | Automated | +| **Cron schedule** | `0 06 * * 5` (weekly on Friday at 06:00 UTC) | +| **Date coverage notes** | Waves 1994–2023 | +| **Test runner instructions** | `./run_tests.sh -p statvar_imports/ntia_internet_use_survey/commerce_ntia` | +| **Source URL** | `https://www.ntia.gov/sites/default/files/data_central_downloads/datasets/ntia-analyze-table.csv` | +| **Type of place** | Demographics | +| **Statistical Variables** | Demographics | - NTIA programs and policymaking focus largely on expanding broadband Internet access and adoption in America, expanding the use of spectrum by all users. -- how to download data: - To download and process the data, you'll need to run the provided preprocess script, `preprocess.py`. This script will automatically create an "input_files" folder where you should place the file to be processed. - By using this script, we are creating two more columns in the input files such as 'universeAgeResol', 'variableAgeResol'. This columns are created based on the universe and variable columns in the existing data. +### Date Coverage Notes +The dataset encompasses NTIA Internet Use Survey data spanning survey waves from 1994 to 2023. -- type of place: Demographics. +### Testing Instructions +To run hermetic unit tests for this import: +```bash +./run_tests.sh -p statvar_imports/ntia_internet_use_survey/commerce_ntia +``` -- statvars: Demographics +- how to download data: + To download and process the data, you'll need to run the provided preprocess script, `preprocess.py`. This script will automatically create an "input_files" folder and download the file to be processed. + This script organizes the data and splits it into general survey data and age-breakdown data. ### How to run: -``` -python3 stat_var_processor.py ---input_data='../../statvar_imports/ntia_internet_use_survey/Commerce_NTIA/input_files/' ---pv_map='../../statvar_imports/ntia_internet_use_survey/Commerce_NTIA/' --config_file='../../statvar_imports/ntia_internet_use_survey/Commerce_NTIA/' --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf ---output_path='../../statvar_imports/ntia_internet_use_survey/Commerce_NTIA//' +```bash +python3 stat_var_processor.py \ + --input_data='../../statvar_imports/ntia_internet_use_survey/commerce_ntia/input_files/' \ + --pv_map='../../statvar_imports/ntia_internet_use_survey/commerce_ntia/' \ + --config_file='../../statvar_imports/ntia_internet_use_survey/commerce_ntia/' \ + --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf \ + --output_path='../../statvar_imports/ntia_internet_use_survey/commerce_ntia//' \ + --output_counters='../../statvar_imports/ntia_internet_use_survey/commerce_ntia//' ``` #### Download the data: @@ -34,19 +52,23 @@ This preprocess script will split the downloaded input file into two files based Execute the script inside the folder `/data/tools/statvar_importer/` -``` -python3 stat_var_processor.py ---input_data=../../statvar_imports/ntia_internet_use_survey/Commerce_NTIA/input_files/ntia-data.csv ---pv_map=../../statvar_imports/ntia_internet_use_survey/Commerce_NTIA/ntia_pvmap.csv ---config_file=../../statvar_imports/ntia_internet_use_survey/Commerce_NTIA/ntia_metadata.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf ---output_path=../../statvar_imports/ntia_internet_use_survey/Commerce_NTIA/output_files/ntia_output +```bash +python3 stat_var_processor.py \ + --input_data=../../statvar_imports/ntia_internet_use_survey/commerce_ntia/input_files/ntia-data.csv \ + --pv_map=../../statvar_imports/ntia_internet_use_survey/commerce_ntia/ntia_pvmap.csv \ + --config_file=../../statvar_imports/ntia_internet_use_survey/commerce_ntia/ntia_metadata.csv \ + --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf \ + --output_path=../../statvar_imports/ntia_internet_use_survey/commerce_ntia/output_files/ntia_output \ + --output_counters=../../statvar_imports/ntia_internet_use_survey/commerce_ntia/counters/ntia_output_counters.csv ``` -``` -python3 stat_var_processor.py ---input_data=../../statvar_imports/ntia_internet_use_survey/Commerce_NTIA/input_files/ntia-data-age-only.csv ---pv_map=../../statvar_imports/ntia_internet_use_survey/Commerce_NTIA/ntia_age_pvmap.csv ---config_file=../../statvar_imports/ntia_internet_use_survey/Commerce_NTIA/ntia_metadata.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf ---output_path=../../statvar_imports/ntia_internet_use_survey/Commerce_NTIA/output_files/ntia_age_output +```bash +python3 stat_var_processor.py \ + --input_data=../../statvar_imports/ntia_internet_use_survey/commerce_ntia/input_files/ntia-data-age-only.csv \ + --pv_map=../../statvar_imports/ntia_internet_use_survey/commerce_ntia/ntia_age_pvmap.csv \ + --config_file=../../statvar_imports/ntia_internet_use_survey/commerce_ntia/ntia_metadata.csv \ + --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf \ + --output_path=../../statvar_imports/ntia_internet_use_survey/commerce_ntia/output_files/ntia_age_output \ + --output_counters=../../statvar_imports/ntia_internet_use_survey/commerce_ntia/counters/ntia_age_output_counters.csv ``` diff --git a/statvar_imports/ntia_internet_use_survey/commerce_ntia/commerce_ntia_test.py b/statvar_imports/ntia_internet_use_survey/commerce_ntia/commerce_ntia_test.py new file mode 100644 index 0000000000..fe47b15cb7 --- /dev/null +++ b/statvar_imports/ntia_internet_use_survey/commerce_ntia/commerce_ntia_test.py @@ -0,0 +1,294 @@ +# Copyright 2025 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. +"""Hermetic unit tests for commerce_ntia preprocess module.""" + +import os +import sys +import tempfile +import unittest +from unittest import mock +import pandas as pd + +_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, _SCRIPT_DIR) +import preprocess + + +class PreprocessTest(unittest.TestCase): + + def test_move_column_left_success(self): + """Tests that move_column_left places column immediately left of target.""" + df = pd.DataFrame({'a': [1], 'b': [2], 'c': [3], 'd': [4]}) + result = preprocess.move_column_left(df, 'd', 'b') + self.assertEqual(list(result.columns), ['a', 'd', 'b', 'c']) + + def test_move_column_left_missing_cols(self): + """Tests that move_column_left returns original df if columns are not present.""" + df = pd.DataFrame({'a': [1], 'b': [2]}) + result = preprocess.move_column_left(df, 'missing', 'b') + self.assertEqual(list(result.columns), ['a', 'b']) + + def test_move_column_left_same_column(self): + """Tests that move_column_left safely handles column_to_move equal to target_column.""" + df = pd.DataFrame({'a': [1], 'b': [2], 'c': [3]}) + result = preprocess.move_column_left(df, 'b', 'b') + self.assertEqual(list(result.columns), ['a', 'b', 'c']) + + def test_preprocess_data(self): + """Tests data preprocessing and splitting into age-only and general survey CSVs.""" + with tempfile.TemporaryDirectory() as tmp_dir: + input_file = os.path.join(tmp_dir, 'ntia-analyze-table.csv') + output_age = os.path.join(tmp_dir, 'ntia-data-age-only.csv') + output_data = os.path.join(tmp_dir, 'ntia-data.csv') + + raw_data = { + 'dataset': [ + 'Nov 2023', 'Nov 2023', 'Nov 2023', 'Nov 2023', 'Nov 2023' + ], + 'variable': [ + 'Streaming', 'Email', 'Broadband', 'isPerson', 'isAdult' + ], + 'description': [ + 'Desc 1', 'Desc 2', 'Desc 3', 'Desc 4', 'Desc 5' + ], + 'universe': [ + 'isPerson', 'isAdult', 'isHousehold', 'isHousehold', + 'isHousehold' + ], + 'age314Count': [10, 20, 30, 40, 50], + 'age314Prop': [0.1, 0.2, 0.3, 0.4, 0.5], + 'age1524Count': [11, 21, 31, 41, 51], + 'age2544Count': [12, 22, 32, 42, 52], + 'age4564Count': [13, 23, 33, 43, 53], + 'age65pCount': [14, 24, 34, 44, 54], + 'age65pSE': [0.01, 0.02, 0.03, 0.04, 0.05], + 'agencyAccess': [5, 10, 15, 20, 25], + 'totalCount': [100, 200, 300, 400, 500], + 'otherMetric': [1.5, 2.5, 3.5, 4.5, 5.5] + } + pd.DataFrame(raw_data).to_csv(input_file, + index=False, + encoding='utf-8-sig') + + with mock.patch.object(preprocess, 'INPUT_DIR', tmp_dir), \ + mock.patch.object(preprocess, 'INPUT_FILE', input_file), \ + mock.patch.object(preprocess, 'INPUT_FILE_1', output_age), \ + mock.patch.object(preprocess, 'INPUT_FILE_2', output_data): + preprocess.preprocess_data() + + self.assertTrue(os.path.exists(output_age)) + self.assertTrue(os.path.exists(output_data)) + + df_age = pd.read_csv(output_age) + cols_age = list(df_age.columns) + self.assertEqual( + cols_age.index('universe') + 1, cols_age.index('variable')) + for age_col in preprocess.AGE_COLUMNS: + self.assertIn(age_col, cols_age) + self.assertNotIn('age314Prop', cols_age) + self.assertNotIn('age65pSE', cols_age) + self.assertNotIn('agencyAccess', cols_age) + self.assertNotIn('totalCount', cols_age) + self.assertNotIn('otherMetric', cols_age) + self.assertIn('universeAgeResol', cols_age) + self.assertIn('variableAgeResol', cols_age) + self.assertEqual(df_age.loc[0, 'universeAgeResol'], 'CivilPerson') + self.assertEqual(df_age.loc[1, 'universeAgeResol'], 'Adult') + self.assertTrue(pd.isna(df_age.loc[2, 'universeAgeResol'])) + self.assertTrue(pd.isna(df_age.loc[0, 'variableAgeResol'])) + self.assertEqual(df_age.loc[3, 'variableAgeResol'], 'CivilPerson') + self.assertEqual(df_age.loc[4, 'variableAgeResol'], 'Adult') + + df_data = pd.read_csv(output_data) + cols_data = list(df_data.columns) + self.assertEqual( + cols_data.index('universe') + 1, cols_data.index('variable')) + self.assertIn('agencyAccess', cols_data) + self.assertIn('totalCount', cols_data) + self.assertIn('otherMetric', cols_data) + self.assertIn('universeAgeResol', cols_data) + self.assertIn('variableAgeResol', cols_data) + self.assertEqual(df_data.loc[0, 'universeAgeResol'], 'CivilPerson') + self.assertEqual(df_data.loc[1, 'universeAgeResol'], 'Adult') + self.assertTrue(pd.isna(df_data.loc[2, 'universeAgeResol'])) + self.assertTrue(pd.isna(df_data.loc[0, 'variableAgeResol'])) + self.assertEqual(df_data.loc[3, 'variableAgeResol'], 'CivilPerson') + self.assertEqual(df_data.loc[4, 'variableAgeResol'], 'Adult') + for age_col in preprocess.AGE_COLUMNS: + self.assertNotIn(age_col, cols_data) + self.assertNotIn('age314Prop', cols_data) + self.assertNotIn('age65pSE', cols_data) + + def test_write_csv_atomically_success(self): + """Tests that _write_csv_atomically writes CSV and removes .tmp file.""" + with tempfile.TemporaryDirectory() as tmp_dir: + target_path = os.path.join(tmp_dir, 'output.csv') + df = pd.DataFrame({'col': [1, 2, 3]}) + preprocess._write_csv_atomically(df, target_path) + self.assertTrue(os.path.exists(target_path)) + self.assertFalse(os.path.exists(f"{target_path}.tmp")) + self.assertGreater(os.path.getsize(target_path), 0) + + def test_write_csv_atomically_empty_raises(self): + """Tests that _write_csv_atomically raises RuntimeError if output file is empty.""" + with tempfile.TemporaryDirectory() as tmp_dir: + target_path = os.path.join(tmp_dir, 'output.csv') + df = pd.DataFrame() + with mock.patch.object(pd.DataFrame, 'to_csv') as mock_to_csv: + + def create_empty(path, **kwargs): + open(path, 'w').close() + + mock_to_csv.side_effect = create_empty + with self.assertRaises(RuntimeError) as cm: + preprocess._write_csv_atomically(df, target_path) + self.assertIn("empty", str(cm.exception)) + self.assertFalse(os.path.exists(target_path)) + + @mock.patch('preprocess.logging.error') + def test_preprocess_data_file_not_found(self, mock_error): + """Tests that preprocess_data raises RuntimeError and logs error if input + file is missing.""" + with tempfile.TemporaryDirectory() as tmp_dir: + missing_input = os.path.join(tmp_dir, 'nonexistent.csv') + with mock.patch.object(preprocess, 'INPUT_DIR', tmp_dir), \ + mock.patch.object(preprocess, 'INPUT_FILE', missing_input): + with self.assertRaises(RuntimeError) as cm: + preprocess.preprocess_data() + self.assertIn("nonexistent.csv", str(cm.exception)) + self.assertTrue(mock_error.called) + self.assertTrue(mock_error.call_args.kwargs.get('exc_info')) + + @mock.patch('preprocess.logging.error') + def test_preprocess_data_missing_columns(self, mock_error): + """Tests that preprocess_data raises RuntimeError and logs error if + input CSV lacks required columns.""" + with tempfile.TemporaryDirectory() as tmp_dir: + bad_input = os.path.join(tmp_dir, 'bad.csv') + pd.DataFrame({'incomplete': [1, 2]}).to_csv(bad_input, index=False) + with mock.patch.object(preprocess, 'INPUT_DIR', tmp_dir), \ + mock.patch.object(preprocess, 'INPUT_FILE', bad_input): + with self.assertRaises(RuntimeError): + preprocess.preprocess_data() + self.assertTrue(mock_error.called) + self.assertTrue(mock_error.call_args.kwargs.get('exc_info')) + + @mock.patch('preprocess.preprocess_data') + @mock.patch('preprocess.download_file') + def test_main_download_success(self, mock_download, mock_preprocess): + """Tests that main downloads file and executes preprocess_data on success.""" + mock_download.return_value = True + with mock.patch('os.path.exists', return_value=True), \ + mock.patch('os.path.getsize', return_value=1024): + preprocess.main([]) + mock_download.assert_called_once_with( + url=preprocess.COMMERCE_NTIA_URL, + output_folder=preprocess.INPUT_DIR, + unzip=False, + headers=preprocess.HEADERS, + tries=3, + delay=5, + backoff=2, + ) + mock_preprocess.assert_called_once() + + @mock.patch('preprocess.preprocess_data') + @mock.patch('preprocess.download_file') + @mock.patch('preprocess.logging.fatal') + def test_main_download_failure(self, mock_fatal, mock_download, + mock_preprocess): + """Tests that main logs fatal error and exits with code 1 when download returns False.""" + mock_fatal.side_effect = SystemExit(1) + mock_download.return_value = False + with self.assertRaises(SystemExit) as cm: + preprocess.main([]) + self.assertEqual(cm.exception.code, 1) + mock_fatal.assert_called_once_with( + "Failed to download Commerce_NTIA file or file is empty.", + exc_info=True) + mock_preprocess.assert_not_called() + + @mock.patch('preprocess.preprocess_data') + @mock.patch('preprocess.download_file') + @mock.patch('preprocess.logging.fatal') + def test_main_download_success_file_missing(self, mock_fatal, mock_download, + mock_preprocess): + """Tests that main exits with code 1 if download reports success but file is missing.""" + mock_fatal.side_effect = SystemExit(1) + mock_download.return_value = True + with mock.patch('os.path.exists', return_value=False): + with self.assertRaises(SystemExit) as cm: + preprocess.main([]) + self.assertEqual(cm.exception.code, 1) + mock_fatal.assert_called_once_with( + "Failed to download Commerce_NTIA file or file is empty.", + exc_info=True) + mock_preprocess.assert_not_called() + + @mock.patch('preprocess.preprocess_data') + @mock.patch('preprocess.download_file') + @mock.patch('preprocess.logging.fatal') + def test_main_download_success_empty_file(self, mock_fatal, mock_download, + mock_preprocess): + """Tests that main exits with code 1 if downloaded file is 0 bytes.""" + mock_fatal.side_effect = SystemExit(1) + mock_download.return_value = True + with mock.patch('os.path.exists', return_value=True), \ + mock.patch('os.path.getsize', return_value=0): + with self.assertRaises(SystemExit) as cm: + preprocess.main([]) + self.assertEqual(cm.exception.code, 1) + mock_fatal.assert_called_once_with( + "Failed to download Commerce_NTIA file or file is empty.", + exc_info=True) + mock_preprocess.assert_not_called() + + @mock.patch('preprocess.preprocess_data') + @mock.patch('preprocess.download_file') + @mock.patch('preprocess.logging.fatal') + def test_main_download_exception(self, mock_fatal, mock_download, + mock_preprocess): + """Tests that main logs fatal error and exits with code 1 when + download raises an exception.""" + mock_fatal.side_effect = SystemExit(1) + mock_download.side_effect = Exception("Connection timeout") + with self.assertRaises(SystemExit) as cm: + preprocess.main([]) + self.assertEqual(cm.exception.code, 1) + self.assertTrue(mock_fatal.called) + self.assertIn("Connection timeout", str(mock_fatal.call_args)) + self.assertTrue(mock_fatal.call_args.kwargs.get('exc_info')) + mock_preprocess.assert_not_called() + + @mock.patch('preprocess.preprocess_data') + @mock.patch('preprocess.download_file') + @mock.patch('preprocess.logging.fatal') + def test_main_preprocess_exception(self, mock_fatal, mock_download, + mock_preprocess): + """Tests that main logs fatal error and exits when preprocess_data raises RuntimeError.""" + mock_fatal.side_effect = SystemExit(1) + mock_download.return_value = True + mock_preprocess.side_effect = RuntimeError("Preprocessing failure") + with mock.patch('os.path.exists', return_value=True), \ + mock.patch('os.path.getsize', return_value=1024): + with self.assertRaises(SystemExit) as cm: + preprocess.main([]) + self.assertEqual(cm.exception.code, 1) + self.assertTrue(mock_fatal.called) + self.assertIn("Preprocessing failure", str(mock_fatal.call_args)) + self.assertTrue(mock_fatal.call_args.kwargs.get('exc_info')) + + +if __name__ == '__main__': + unittest.main() diff --git a/statvar_imports/ntia_internet_use_survey/commerce_ntia/config.py b/statvar_imports/ntia_internet_use_survey/commerce_ntia/config.py index 7b4f49bda0..74068d9e8e 100644 --- a/statvar_imports/ntia_internet_use_survey/commerce_ntia/config.py +++ b/statvar_imports/ntia_internet_use_survey/commerce_ntia/config.py @@ -1 +1,3 @@ -Commerce_NTIA_URL = "https://www.ntia.gov/sites/default/files/data_central_downloads/datasets/ntia-analyze-table.csv" \ No newline at end of file +COMMERCE_NTIA_URL = ( + "https://www.ntia.gov/sites/default/files/data_central_downloads/" + "datasets/ntia-analyze-table.csv") diff --git a/statvar_imports/ntia_internet_use_survey/commerce_ntia/manifest.json b/statvar_imports/ntia_internet_use_survey/commerce_ntia/manifest.json index 2906cb50c3..2315bdfce8 100644 --- a/statvar_imports/ntia_internet_use_survey/commerce_ntia/manifest.json +++ b/statvar_imports/ntia_internet_use_survey/commerce_ntia/manifest.json @@ -9,23 +9,28 @@ "provenance_description": "NTIA programs and policymaking focus largely on expanding broadband Internet access and adoption in America, expanding the use of spectrum by all users.", "scripts": [ "preprocess.py", - "../../../tools/statvar_importer/stat_var_processor.py --input_data=input_files/ntia-data.csv --pv_map=ntia_pvmap.csv --config_file=ntia_metadata.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path=output_files/ntia_output", - "../../../tools/statvar_importer/stat_var_processor.py --input_data=input_files/ntia-data-age-only.csv --pv_map=ntia_age_pvmap.csv --config_file=ntia_metadata.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path=output_files/ntia_age_output" + "../../../tools/statvar_importer/stat_var_processor.py --input_data=input_files/ntia-data.csv --pv_map=ntia_pvmap.csv --config_file=ntia_metadata.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path=output_files/ntia_output --output_counters=counters/ntia_output_counters.csv", + "../../../tools/statvar_importer/stat_var_processor.py --input_data=input_files/ntia-data-age-only.csv --pv_map=ntia_age_pvmap.csv --config_file=ntia_metadata.csv --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path=output_files/ntia_age_output --output_counters=counters/ntia_age_output_counters.csv" ], "source_files": [ - "input_files/ntia-analyze-table.csv" + "input_files/ntia-analyze-table.csv", + "counters/*.csv", + "validation_config.json" ], "import_inputs": [ { "template_mcf": "output_files/ntia_output.tmcf", - "cleaned_csv": "output_files/ntia_output.csv" + "cleaned_csv": "output_files/ntia_output.csv", + "node_mcf": "output_files/ntia_output*_stat_vars.mcf" }, { "template_mcf": "output_files/ntia_age_output.tmcf", - "cleaned_csv": "output_files/ntia_age_output.csv" + "cleaned_csv": "output_files/ntia_age_output.csv", + "node_mcf": "output_files/ntia_age_output*_stat_vars.mcf" } ], - "cron_schedule": "0 06 * * 5" + "cron_schedule": "0 06 * * 5", + "validation_config_file": "validation_config.json" } ] -} \ No newline at end of file +} diff --git a/statvar_imports/ntia_internet_use_survey/commerce_ntia/preprocess.py b/statvar_imports/ntia_internet_use_survey/commerce_ntia/preprocess.py index 9bcf786a14..96ccb33bf3 100644 --- a/statvar_imports/ntia_internet_use_survey/commerce_ntia/preprocess.py +++ b/statvar_imports/ntia_internet_use_survey/commerce_ntia/preprocess.py @@ -12,10 +12,12 @@ # See the License for the specific language governing permissions and # limitations under the License. -import os, sys -import pandas as pd +import os +import sys + from absl import app, logging -from pathlib import Path +import pandas as pd + import config script_dir = os.path.dirname(os.path.abspath(__file__)) @@ -24,21 +26,40 @@ from download_util_script import download_file -Commerce_NTIA_URL = config.Commerce_NTIA_URL +COMMERCE_NTIA_URL = config.COMMERCE_NTIA_URL INPUT_DIR = os.path.join(script_dir, "input_files") -Path(INPUT_DIR).mkdir(parents=True, exist_ok=True) - COMMON_COLUMNS = ["dataset", "variable", "description", "universe"] -AGE_COLUMNS = ["age314Count", "age1524Count", "age2544Count", "age4564Count", "age65pCount"] +AGE_COLUMNS = [ + "age314Count", "age1524Count", "age2544Count", "age4564Count", "age65pCount" +] +_AGE_RESOL_MAP = {'isPerson': 'CivilPerson', 'isAdult': 'Adult'} INPUT_FILE = os.path.join(INPUT_DIR, "ntia-analyze-table.csv") INPUT_FILE_1 = os.path.join(INPUT_DIR, "ntia-data-age-only.csv") INPUT_FILE_2 = os.path.join(INPUT_DIR, "ntia-data.csv") +HEADERS = { + 'User-Agent': ('Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ' + '(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'), + 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', +} + + +def _write_csv_atomically(df: pd.DataFrame, target_path: str) -> None: + """Writes a DataFrame to a CSV atomically using a temporary file.""" + tmp_path = f"{target_path}.tmp" + df.to_csv(tmp_path, index=False) + if not os.path.exists(tmp_path) or os.path.getsize(tmp_path) == 0: + raise RuntimeError( + f"Failed to write CSV or output file is empty: {tmp_path}") + os.replace(tmp_path, target_path) + def move_column_left(df, column_to_move, target_column): """Moves the universe column to the left of variable column.""" + if column_to_move == target_column: + return df cols = df.columns.tolist() if column_to_move in cols and target_column in cols: cols.remove(column_to_move) @@ -50,45 +71,70 @@ def move_column_left(df, column_to_move, target_column): def preprocess_data(): try: - org_df = pd.read_csv(INPUT_FILE) - + os.makedirs(INPUT_DIR, exist_ok=True) + org_df = pd.read_csv(INPUT_FILE, encoding='utf-8-sig') + + # 1. Process Age-only data df1 = org_df[COMMON_COLUMNS + AGE_COLUMNS].copy() - df1['universeAgeResol'] = df1['universe'].apply( - lambda x: 'CivilPerson' if x == 'isPerson' else ('Adult' if x == 'isAdult' else None) - ) - df1['variableAgeResol'] = df1['variable'].apply( - lambda x: 'CivilPerson' if x == 'isPerson' else ('Adult' if x == 'isAdult' else None) - ) + df1['universeAgeResol'] = df1['universe'].map(_AGE_RESOL_MAP) + df1['variableAgeResol'] = df1['variable'].map(_AGE_RESOL_MAP) df1_moved = move_column_left(df1, 'universe', 'variable') - df1_moved.to_csv(INPUT_FILE_1, index=False) - - df2_cols_to_keep = [col for col in org_df.columns if not col.startswith('age')] + _write_csv_atomically(df1_moved, INPUT_FILE_1) + + # 2. Process General survey data + df2_cols_to_keep = [ + col for col in org_df.columns + if not col.startswith(('age314', 'age1524', 'age2544', 'age4564', + 'age65p')) + ] df2 = org_df[df2_cols_to_keep].copy() - df2['universeAgeResol'] = df2['universe'].apply( - lambda x: 'CivilPerson' if x == 'isPerson' else ('Adult' if x == 'isAdult' else None) - ) - df2['variableAgeResol'] = df2['variable'].apply( - lambda x: 'CivilPerson' if x == 'isPerson' else ('Adult' if x == 'isAdult' else None) - ) + df2['universeAgeResol'] = df2['universe'].map(_AGE_RESOL_MAP) + df2['variableAgeResol'] = df2['variable'].map(_AGE_RESOL_MAP) df2_moved = move_column_left(df2, 'universe', 'variable') - df2_moved.to_csv(INPUT_FILE_2, index=False) + _write_csv_atomically(df2_moved, INPUT_FILE_2) + logging.info( + f"Successfully preprocessed {len(df1_moved)} age-only rows " + f"and {len(df2_moved)} general survey rows.") except Exception as e: - logging.fatal(f"An error occurred while preprocessing the input data: {e}") - return None + logging.error( + f"An error occurred while preprocessing the input data: {e}", + exc_info=True) + raise RuntimeError( + f"An error occurred while preprocessing the input data: {e}") from e + def main(argv): + del argv + try: + logging.info(f"Downloading source data from {COMMERCE_NTIA_URL}") + success = download_file( + url=COMMERCE_NTIA_URL, + output_folder=INPUT_DIR, + unzip=False, + headers=HEADERS, + tries=3, + delay=5, + backoff=2, + ) + if not success or not os.path.exists(INPUT_FILE) or os.path.getsize( + INPUT_FILE) == 0: + logging.fatal( + "Failed to download Commerce_NTIA file or file is empty.", + exc_info=True) + logging.info( + f"Successfully downloaded {INPUT_FILE} ({os.path.getsize(INPUT_FILE)} bytes)." + ) + except Exception as e: + logging.fatal(f"Failed to download Commerce_NTIA file: {e}", + exc_info=True) + try: - download_file(url=Commerce_NTIA_URL, - output_folder=INPUT_DIR, - unzip=False, - headers= None, - tries= 3, - delay= 5, - backoff= 2) + preprocess_data() except Exception as e: - logging.fatal(f"Failed to download Commerce_NTIA file: {e}") - preprocess_data() + logging.fatal(f"Failed to preprocess Commerce_NTIA data: {e}", + exc_info=True) + if __name__ == "__main__": app.run(main) diff --git a/statvar_imports/ntia_internet_use_survey/commerce_ntia/validation_config.json b/statvar_imports/ntia_internet_use_survey/commerce_ntia/validation_config.json new file mode 100644 index 0000000000..5a032b3af3 --- /dev/null +++ b/statvar_imports/ntia_internet_use_survey/commerce_ntia/validation_config.json @@ -0,0 +1,31 @@ +{ + "schema_version": "1.0", + "rules": [ + { + "rule_id": "check_deleted_records_percent", + "description": "Checks that the percentage of deleted records for the entire import does not exceed 0.1%.", + "validator": "DELETED_RECORDS_PERCENT", + "params": { + "threshold": 0.1 + } + }, + { + "rule_id": "check_active_survey_wave_count", + "description": "Checks that all active StatVars (1935 in input0, 227 in input1) have MaxDate >= 2023-11.", + "validator": "SQL_VALIDATOR", + "params": { + "query": "SELECT COUNT(*) AS total_svs, COALESCE(SUM(CASE WHEN MaxDate >= '2023-11' THEN 1 ELSE 0 END), 0) AS active_svs, COALESCE(SUM(CASE WHEN regexp_matches(StatVar, '3To14Years|15To24Years|25To44Years|45To64Years|65OrMoreYears') THEN 1 ELSE 0 END), 0) AS age_svs FROM stats", + "condition": "(age_svs = 0 AND total_svs >= 1935 AND active_svs >= 1935) OR (age_svs = total_svs AND total_svs >= 227 AND active_svs >= 227)" + } + }, + { + "rule_id": "check_statvar_min_max_date", + "description": "Checks that no StatVar regresses below the oldest historical wave (2013-07).", + "validator": "SQL_VALIDATOR", + "params": { + "query": "SELECT StatVar, MaxDate FROM stats", + "condition": "MaxDate IS NOT NULL AND MaxDate >= '2013-07'" + } + } + ] +}