From 4a2201ad3a7e24bbdf70086a16215fdd6fb877fd Mon Sep 17 00:00:00 2001 From: goutamabrol Date: Sat, 17 Oct 2026 03:49:37 +0000 Subject: [PATCH 1/7] S2303 --- .../acs5yr/subject_tables/S2303/README.md | 82 ++- .../subject_tables/S2303/data_loader.py | 505 ++++++++++++++++++ .../acs5yr/subject_tables/S2303/process.py | 256 ++++----- 3 files changed, 690 insertions(+), 153 deletions(-) create mode 100644 scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/README.md b/scripts/us_census/acs5yr/subject_tables/S2303/README.md index a0cd358eda..f5f73260a5 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/README.md +++ b/scripts/us_census/acs5yr/subject_tables/S2303/README.md @@ -1,12 +1,82 @@ -# US census ACS5YR S2303 Subject Table import +# US Census ACS5YR S2303 Subject Table Import This import gives the count of people in the age range 16-64 years who have worked 1 or more weeks in the past 12 months. It provides a further breakdown of the data on the gender, number of working weeks and the amount of hours worked per week. -Years: 2010-2019 -Geo : Country, State, County and Place +- **Years**: 2010-2024 +- **Geo**: Country, State, County, County Subdivision, Census Tract, Place/City, Congressional District, ZCTA, School District, CBSA, Census Division -Important Notes : +## Important Notes 1. Median age of workers is not available from 2010-2014. 2. Data for "Workers 16 to 64 years who worked full-time, year-round" is not available from 2010-2014. -3. The breakdown of usual hours worked from 2010 - 2014 is a subset of the ones available from 2015 - 2019. -4. The values are published as counts from 2015 - 2019 and as percentages from 2010 - 2014. +3. The breakdown of usual hours worked from 2010 - 2014 is a subset of the ones available from 2015 - 2024. +4. The values are published as counts from 2015 - 2024 and as percentages from 2010 - 2014. + +## Download Step + +Run `census_api_data_downloader.py` from `scripts/us_census/api_utils` for each 3-year range using `--start_year` / `--end_year` and copy each generated `S2303.zip` into `input_files/`: + +```bash +cd scripts/us_census/api_utils +mkdir -p ../acs5yr/subject_tables/S2303/input_files + +for range in 2010_2012 2013_2015 2016_2018 2019_2021 2022_2024; do + start_yr=${range%_*} + end_yr=${range#*_} + python3 census_api_data_downloader.py \ + --dataset=acs/acs5/subject \ + --table_id=S2303 \ + --start_year=$start_yr \ + --end_year=$end_yr \ + --summary_levels=010,030,040,050,060,140,160,310,500,860,950,960,970 \ + --output_path=~/us_census \ + --api_key= + cp ~/us_census/acs/acs5/subject/S2303/S2303.zip \ + ../acs5yr/subject_tables/S2303/input_files/s2303_${range}.zip +done +``` + +## Process Step + +Run `process.py` from `scripts/us_census/acs5yr/subject_tables/S2303`: + +```bash +cd ../acs5yr/subject_tables/S2303 + +for b in s2303_2010_2012 s2303_2013_2015 s2303_2016_2018 s2303_2019_2021 s2303_2022_2024; do + python3 process.py \ + --option=all \ + --table_prefix=$b \ + --has_percent=True \ + --debug=False \ + --spec_path=./S2303_spec.json \ + --input_path=input_files/${b}.zip \ + --output_dir=./ + mv column_map.json ${b}_column_map.json + mv ${b}_output.tmcf s2303_output.tmcf +done +``` + +- `--option=all`: Runs both column map generation (`column_map.json`) and observation processing (`_cleaned.csv`, `_output.mcf`, `_output.tmcf`, `_summary.json`). +- `--has_percent=True`: Converts percentage estimates (`2010`–`2014`) into counts using the mapped denominator keys in `S2303_spec.json` (`data_loader.py` restricts conversion to `percent_years = ['2014', '2013', '2012', '2011', '2010']`). + +## Batch Splitting & Runtime Optimization + +Processing all 15 years (`2010`–`2024`, ~80 GiB intermediate CSV / ~459M rows) as a single `.zip` requires ~155 GiB RAM during the final `pd.read_csv()` cleanup step and spills to swap. Splitting into **5 3-year `.zip` batches** keeps each run in RAM with zero swap thrashing: + +| Batch / `--table_prefix` | CSV Files Inside `.zip` | Output Files Generated | +| :--- | :--- | :--- | +| `s2303_2010_2012` | `ACSST5Y2010*.csv`, `ACSST5Y2011*.csv`, `ACSST5Y2012*.csv` | `s2303_2010_2012_{cleaned.csv, output.mcf, summary.json}` | +| `s2303_2013_2015` | `ACSST5Y2013*.csv`, `ACSST5Y2014*.csv`, `ACSST5Y2015*.csv` | `s2303_2013_2015_{cleaned.csv, output.mcf, summary.json}` | +| `s2303_2016_2018` | `ACSST5Y2016*.csv`, `ACSST5Y2017*.csv`, `ACSST5Y2018*.csv` | `s2303_2016_2018_{cleaned.csv, output.mcf, summary.json}` | +| `s2303_2019_2021` | `ACSST5Y2019*.csv`, `ACSST5Y2020*.csv`, `ACSST5Y2021*.csv` | `s2303_2019_2021_{cleaned.csv, output.mcf, summary.json}` | +| `s2303_2022_2024` | `ACSST5Y2022*.csv`, `ACSST5Y2023*.csv`, `ACSST5Y2024*.csv` | `s2303_2022_2024_{cleaned.csv, output.mcf, summary.json}` | + +### Runtime Comparison +| Method | Total Wall-Clock Time | Time Saved | Notes | +| :--- | :---: | :---: | :--- | +| **1 Single `.zip` (`2010`–`2024`)** | ~110 – 120 mins | — | Spills to swap during final `pd.read_csv()` rewrite. | +| **5 Split `.zip`s (Sequential)** | ~45 – 55 mins | ~60 – 65 mins (~2x faster) | Each batch (~16 GiB CSV) fits in RAM with zero swap. | +| **5 Split `.zip`s (Parallel, 2–3 jobs)** | ~18 – 22 mins | ~90 – 100 mins (~5x faster) | Utilizes multiple cores while staying under system RAM. | + + + diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py b/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py new file mode 100644 index 0000000000..2eab35ffbb --- /dev/null +++ b/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py @@ -0,0 +1,505 @@ +# Copyright 2021 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. +"""Common data processing module for ACS Subject Tables""" +import os +import sys +import json +from zipfile import ZipFile +from absl import logging +import numpy as np +import pandas as pd + +# Allows the following module imports to work when running as a script +_SCRIPT_PATH = os.path.dirname(os.path.abspath(__file__)) + +sys.path.append(os.path.join(_SCRIPT_PATH, '../common')) # for resolve_geo_id +from resolve_geo_id import convert_to_place_dcid + +_IGNORED_VALUES = set(['-', '*', '**', '***', '*****', 'N', '(X)', 'null']) + +# TODO: Consider using logs for logging warning messages for debug at a later +# date + + +def process_subject_tables(table_prefix='', + input_path='./', + output_dir='./', + column_map_path=None, + spec_path=None, + debug=False, + delimiter='!!', + has_percent=False, + decimal_places=3, + header_row=1, + estimate_period=5): + """ + Wrapper method for invoking the data processing module, maps files with the + corresponding processing function based on input format (*.zip/*.csv), + output files are saved in the specified output directory. + + NOTE: This module does not generate the column_map, you will need to run the + `generate_col_map.py` module before running this module in standalone mode. + + NOTE: If the path for column_map is not specified this module will exit with + an error, stating that the column_map is a required input. + + Args: + table_prefix: string prefix that will be added to differentiate outputs + input_path: path to the input file(s). The module supports inputs as either + a zip file or a csv file or a directory with zip/ csv files + output_dir: path to the output directory + column_map_path: path to the column_map generated with `generate_col_map.py` + spec_path: path to the JSON specification for the subject table + debug: if set, adds additional columns to output csv for debugging (default: False) + delimiter: String denoting the delimiter character in dataset (default: '!!') + has_percent: if set, converts percent to count values (default: False) + decimal_places: specifies the number of digits after decimal place (default: 3) + estimate_period: specifies the duration of the subject period estimate (default: 5) + : since we use the 5 year subject table estimate + header_row: specifies the row index to be considered as the dataframe's + header for column name assignment. For subject tables, the second row (row index = 1) + has a human readable column header. (default: 1) + header_row: specifies the row index to be considered as the dataframe's header + for column name assignment. For subject tables, the second row (row index = 1) + has a human readable column header. (default: 1) + + Outputs: + The module will write the following files to the specified output directory + _cleaned.csv: + csv file with StatVarObservations i.e. quantity measured for a stat_var at a particular place and date. + It also has additional columns like units, and scalingFactor if they are specified in the JSON spec. + _output.mcf: + file in mcf format with all the StatVars generated in the column map + _output.tmcf: + template mcf file used in combination with the csv file for data import + _summary.json: + summary stats of data processing year-wise + + If debug=True, additional output files are written to the output directory + column_to_dcid.csv: + maps the stat var dcid associated to each column of the data file + column_to_statvar_map.json: + a json file with column mapped to the complete stat-var node + _cleaned.csv will have an additional column containing the name of column in the dataset for which each StatVarObservation was recorded + """ + ## create a data loader object with base parameters + data_loader = SubjectTableDataLoaderBase(table_id=table_prefix, + col_delimiter=delimiter, + has_percent=has_percent, + debug=debug, + output_path_dir=output_dir, + json_spec=spec_path, + column_map_path=column_map_path, + decimal_places=decimal_places, + estimate_period=estimate_period, + header_row=header_row) + + ## if input_path is a file, select csv/zip processing methods + _, file_extension = os.path.splitext(input_path) + if file_extension == '.zip': + data_loader._process_zip_file(input_path) + + elif file_extension == '.csv': + data_loader._process_csv_file(input_path) + + else: + data_loader._process_dir(input_path) + + +class SubjectTableDataLoaderBase: + """ + A common module that has a collection of methods for subject table data processing. + + + Attributes: + table_prefix: string prefix that will be added to differentiate outputs + input_path: path to the input file(s). The module supports inputs as either + a zip file or a csv file or a directory with zip/ csv files + output_dir: path to the output directory + column_map_path: path to the column_map generated with `generate_col_map.py` + spec_path: path to the JSON specification for the subject table + debug: if set, adds additional columns to output csv for debugging (default: False) + col_delimiter: String denoting the delimiter character in dataset (default: '!!') + has_percent: if set, converts percent to count values (default: False) + decimal_places: specifies the number of digits after decimal place (default: 3) + estimate_period: specifies the duration of the subject period estimate (default: 5) + : since we use the 5 year subject table estimate + header_row: specifies the row index to be considered as the dataframe's header + for column name assignment. For subject tables, the second row (row index = 1) + has a human readable column header. (default: 1) + + """ + + def __init__(self, + table_id='', + col_delimiter='!!', + has_percent=False, + debug=False, + output_path_dir='./', + json_spec='./', + column_map_path='./', + decimal_places=3, + estimate_period=5, + header_row=1): + """module init""" + ## static inputs + self.table_id = table_id + self.has_percent = has_percent + self.debug = debug + self.decimal_places = decimal_places + self.delimiter = col_delimiter + self.estimate_period = estimate_period + self.header_row = header_row + + ## set the output path, if not exists, make directory + if not os.path.exists(output_path_dir): + os.mkdir(output_path_dir) + self.output_path_dir = output_path_dir + self.base_tmcf_path = os.path.join(_SCRIPT_PATH, + '../common/subject_table_base.tmcf') + + ## get the content of JSON spec + try: + f = open(json_spec, 'r') + self.json_spec = json.load(f) + f.close() + except: + logging.fatal( + "Please ensure that the path to the JSON spec is correct.") + + ## get the content from Column map + try: + f = open(column_map_path, 'r') + self.column_map = json.load(f) + f.close() + except: + # TODO: If the module is called from `process.py`, the message needs + # to be to run with --option=colmap. So rephrasing this message will help. + logging.fatal( + """To run this module in standalone mode, ensure the column map + is generated by running `generate_col_map` module.""") + + ## initialize other class variables + self.mcf_dict = {} + + # initialize a dictionary of counters + # TODO: Add a means to fill the git commit SHA + self.counter_dict = { + "summary": { + "column map generator commit SHA": "", + "data processing script commit SHA": "", + "total StatVarObservations": 0, + }, + } + + # initialize the StatVarObservation csv + self.csv_columns = [ + 'Year', 'Place', 'StatVar', 'Quantity', 'Unit', 'ScalingFactor', + 'Column' + ] #optional: Unit, ScalingFactor + self.clean_csv_path = os.path.join(output_path_dir, + f'./{table_id}_cleaned.csv') + self.year_count = 0 + if os.path.exists(self.clean_csv_path): + os.remove(self.clean_csv_path) + + def _convert_percent_to_numbers(self, df): + """utility to convert percent to numbers""" + # TODO: Currently this function expects full names for function, which + # makes the spec long, perhaps we can add support for tokens + df_columns = df.columns.tolist() + for count_col, percent_col_list in self.json_spec['denominators'].items( + ): + # TODO: add a warning message + if count_col in df_columns: + for col in percent_col_list: + if col in df_columns: + df[col] = df[col].astype(float) + df[count_col] = df[count_col].astype(float) + df[col] = (df[col] / 100) * df[count_col] + df[col] = df[col].round( + 3 + ) # TODO: replace with parameter self.decimal_places + return df + + def _get_summary(self): + """ + method returns the total number of columns, stat vars and other stats + from the column_map and csv. + """ + for key, count_dict in self.counter_dict.items(): + if key != 'summary': + self.counter_dict['summary'][ + "total StatVarObservations"] += count_dict[ + "number of observations"] + return self.counter_dict + + def _replace_ignore_values_with_nan(self, df): + """ + Replaces cells of the dataframe by column with nan if cell has values in + _IGNORED_VALUES. + """ + for col in df.columns.tolist(): + df[col] = df[col].apply(lambda row: np.nan + if row in _IGNORED_VALUES else row) + return df + + def _process_dataframe(self, df, filename): + """processes a dataframe read from a csv file""" + df = self._replace_ignore_values_with_nan( + df) #handle the values to be ignored + year = filename.split(f'ACSST{self.estimate_period}Y')[1][:4] + logging.info(f"Processing: {filename}") + # if has_percent is set, convert percentages to counts. Requires the + # 'denominators' key to be specified in the spec + percent_years = self.json_spec.get( + 'percent_years', ['2014', '2013', '2012', '2011', '2010']) + if self.has_percent and year in percent_years: + df = self._convert_percent_to_numbers(df) + + # if column map is not available generate, will rarely be False + column_map = self.column_map[year] + + # add stats to dict_counter for current year + self.counter_dict[year] = { + "filename": + filename, + "number of columns in dataset": + df.shape[1], + "number of rows in dataset": + df.shape[0], + "number of statVars generated for columns": + len(list(column_map.keys())), + "number of observations": + 0, + "number of unique StatVars with observations": + 0, + } + + csv_file = open(self.clean_csv_path, 'a') + if 'id' in df.columns: + place_geoIds = df['id'].apply(convert_to_place_dcid) + else: + place_geoIds = df['Geography'].apply(convert_to_place_dcid) + + # update the clean csv + for column in df.columns.tolist(): + if column in column_map: + obs_df = pd.DataFrame(columns=self.csv_columns) + obs_df['Place'] = place_geoIds + obs_df['StatVar'] = column_map[column]['Node'] + obs_df['Quantity'] = df[column].values.tolist() + # add unit to the csv + try: + unit = column_map[column]['unit'] + del column_map[column]['unit'] + except KeyError: + unit = np.nan + obs_df['Unit'] = unit + + # add scaling factor to the csv + try: + scalingFactor = column_map[column]['scalingFactor'] + del column_map[column]['scalingFactor'] + except KeyError: + scalingFactor = np.nan + obs_df['ScalingFactor'] = scalingFactor + + # if StatVar not in mcf_dict, add dcid + dcid = column_map[column]['Node'] + + if dcid not in self.mcf_dict: + ## key --> node dcid + self.mcf_dict[dcid] = {} + ## add pvs to dict + for key, value in column_map[column].items(): + if key != 'Node': + self.mcf_dict[dcid][key] = value + + obs_df['Year'] = year + obs_df['Column'] = column + + # Replace empty places (unresolved geoIds) as null values + obs_df['Place'] = obs_df['Place'].replace('', np.nan) + + # Drop rows with observations for empty (null) values + obs_df.dropna(subset=['Place', 'Quantity'], + axis=0, + inplace=True) + + # Write the processed observations to the clean_csv + if self.year_count == 0: + obs_df.to_csv(csv_file, header=True, index=False, mode='w') + else: + obs_df.to_csv(csv_file, header=False, index=False, mode='a') + self.year_count += 1 + + # update stats for the year: + self.counter_dict[year]["number of unique geos"] = len( + obs_df['Place'].unique()) + self.counter_dict[year][ + "number of observations"] += obs_df.shape[0] + self.counter_dict[year][ + "number of unique StatVars with observations"] += len( + obs_df['StatVar'].unique()) + self.counter_dict[year]["number of StatVars in mcf_dict"] = len( + list(self.mcf_dict.keys())) + csv_file.close() + logging.info( + f"""Completed with { self.counter_dict[year]['number of observations'] } + observation for { self.counter_dict[year]['number of unique StatVars with observations'] } + StatVars at { self.counter_dict[year]['number of unique geos'] } places. + """) + + def _generate_mcf_from_column_map(self): + """ + uses column_map to generate a mcf file with all the stat_vars nodes that + are created. + """ + + #get cleaned column map + final_mcf = "" + + for column, stat_var in self.mcf_dict.items(): + col_stat_var = [] + # TODO: Add 'Node:' as the first element of col_stat_var, and append + # all the stat-var node information to col_stat_var + dcid = 'Node: ' + column + '\n' + + for p, v in stat_var.items(): + if p != 'unit' or p != 'scalingFactor': + col_stat_var.append(f"{p}: {v}") + # TODO: Replace the next two lines with final_mcf = final_mcf + '\n'.join(col_stat_var) + '\n\n' + col_stat_var = dcid + '\n'.join(col_stat_var) + final_mcf = final_mcf + col_stat_var + "\n\n" + return final_mcf + + def _generate_tmcf_from_clean_csv(self): + """update the base tmcf template based on the clean_csv""" + clean_csv = pd.read_csv(self.clean_csv_path, low_memory=False) + empty_cols = [ + col for col in clean_csv.columns if clean_csv[col].isnull().all() + ] + + f = open(self.base_tmcf_path, 'r') + tmcf_contents = f.read().splitlines() + f.close() + + ## modify the tmcf file content + for col in empty_cols: + for index, line in enumerate(tmcf_contents): + if col in line: + del tmcf_contents[index] + + ## write into the output tmcf + tmcf_contents = '\n'.join(tmcf_contents) + if not self.debug: + clean_csv.drop(columns=['Column'], inplace=True) + + ## update the csv by dropping empty column + clean_csv.drop(columns=empty_cols, inplace=True) + clean_csv.to_csv(self.clean_csv_path, index=False) + return tmcf_contents + + def _get_outputs(self): + """aggregated method which stores tmcf and clean_csv to output directory""" + ### write to mcf + final_mcf = self._generate_mcf_from_column_map() + f = open( + os.path.join(self.output_path_dir, f'./{self.table_id}_output.mcf'), + 'w') + f.write(final_mcf) + f.close() + + ## write to tmcf + final_tmcf = self._generate_tmcf_from_clean_csv() + f = open( + os.path.join(self.output_path_dir, + f'./{self.table_id}_output.tmcf'), 'w') + f.write(final_tmcf) + f.close() + + ## write all stat_vars from column map to file for debugging + if self.debug: + debug_df = pd.DataFrame(columns=['Year', 'Column', 'StatVarID']) + for year, val in self.column_map.items(): + for column, stat_var in val.items(): + debug_df = debug_df.append( + { + 'Year': year, + 'Column': column, + 'StatVarID': stat_var['Node'] + }, + ignore_index=True) + debug_df.to_csv(os.path.join(self.output_path_dir, + './column_to_dcid.csv'), + index=False) + f = open( + os.path.join(self.output_path_dir, + './column_to_statvar_map.json'), 'w') + json.dump(self.column_map, f, indent=4) + f.close() + + ## write summary dict of counters + counter_dict = self._get_summary() + summary_path = os.path.join(self.output_path_dir, + f'{self.table_id}_summary.json') + f = open(summary_path, 'w') + json.dump(counter_dict, f, indent=4) + f.close() + logging.info( + f"""processing of subject table is complete: the summary of the + process is available at {summary_path} along with the generated + output files.""") + + def _process_zip_file(self, zip_file_path): + """processes each dataset in a zip file to make the clean csv of StatVarObs""" + zf = ZipFile(zip_file_path) + for filename in zf.namelist(): + if 'data_with_overlays' in filename: + df = pd.read_csv(zf.open(filename, 'r'), + header=self.header_row, + low_memory=False) + self._process_dataframe(df, filename) + zf.close() + self._get_outputs() + + def _process_dir(self, input_dir_path): + """specify the folder_path with data files to generate StatVarObs csv""" + try: + for filename in os.listdir(input_dir_path): + ## if zip file is present in directory, try _process_zip_file() + if os.path.splitext(filename)[1] == 'zip': + self._process_zip_file( + os.path.join(input_dir_path, filename)) + ## if input_directory has csv files, process them + if 'data_with_overlays' in filename: + input_path = os.path.join(input_dir_path, filename) + df = pd.read_csv(input_path, + header=self.header_row, + low_memory=False) + self._process_dataframe(df, filename) + self._get_outputs() + except: + logging.fatal( + """ensure input path is that of directory with dataset files, + a zip file or a csv file""") + + def _process_csv_file(self, input_path): + df = pd.read_csv(input_path, header=self.header_row, low_memory=False) + self._process_dataframe(df, input_path) + self._get_outputs() + + +S2303SubjectTableDataLoader = SubjectTableDataLoaderBase diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/process.py b/scripts/us_census/acs5yr/subject_tables/S2303/process.py index 963ad6c9a3..a36a053055 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/process.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/process.py @@ -1,149 +1,111 @@ -"""Process data from Census ACS5Year Subject Table S2303.""" +# Copyright 2021 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. +"""Generic proces module to generate the csv/tmcf and csv""" +# TODO: Add unit tests import os import sys -import pandas as pd -import numpy as np - -# pylint: disable=missing-module-docstring -# pylint: disable=missing-class-docstring -# pylint: disable=missing-function-docstring - -#pylint: disable=wrong-import-position -#pylint: disable=import-error -_CODEDIR = os.path.dirname(os.path.realpath(__file__)) -sys.path.insert(1, os.path.join(_CODEDIR, '..', 'common')) - -from data_loader import SubjectTableDataLoaderBase # commit hash - 4bc5102 -from resolve_geo_id import convert_to_place_dcid -#pylint: enable=wrong-import-position -#pylint: enable=import-error - - -class S2303SubjectTableDataLoader(SubjectTableDataLoaderBase): - """ - A child class for the S2303 import. - """ - - def _convert_percent_to_numbers(self, df): - df_cols = df.columns.tolist() - for count_col, percent_cols in self.json_spec['denominators'].items(): - if count_col in df_cols: - for col in percent_cols: - if col in df_cols: - df[col] = df[col].astype(float) - df[count_col] = df[count_col].astype(float) - df[col] = (df[col] / 100) * df[count_col] - df[col] = df[col].round(3) - return df - - def _process_dataframe(self, df, filename): - """processes a dataframe read from a csv file""" - df = self._replace_ignore_values_with_nan(df) - year = filename.split(f'ACSST{self.estimate_period}Y')[1][:4] - print(f"Processing: {filename}", end=" | ", flush=True) - - # Process only at a country, state, county and place level - df = df[~((df['id'].str.startswith('950')) | - (df['id'].str.startswith('960')) | - (df['id'].str.startswith('970')))] - - # Only calculating percentages for selected years - percent_years = ['2014', '2013', '2012', '2011', '2010'] - if self.has_percent and year in percent_years: - print("Converting percent to numbers", end=" | ", flush=True) - df = self._convert_percent_to_numbers(df) - - # if column map is not available generate, will rarely be False - column_map = self.column_map[year] - - # add stats to dict_counter for current year - self.counter_dict[year] = { - "filename": - filename, - "number of columns in dataset": - df.shape[1], - "number of rows in dataset": - df.shape[0], - "number of statVars generated for columns": - len(list(column_map.keys())), - "number of observations": - 0, - "number of unique StatVars with observations": - 0, - } - - csv_file = open(self.clean_csv_path, 'a') - place_geo_ids = df['id'].apply(convert_to_place_dcid) - - # update the clean csv - for column in df.columns.tolist(): - if column in column_map: - obs_df = pd.DataFrame(columns=self.csv_columns) - obs_df['Place'] = place_geo_ids - obs_df['StatVar'] = column_map[column]['Node'] - obs_df['Quantity'] = df[column].values.tolist() - # add unit to the csv - try: - unit = column_map[column]['unit'] - del column_map[column]['unit'] - except KeyError: - unit = np.nan - obs_df['Unit'] = unit - - # add scaling factor to the csv - try: - scaling_factor = column_map[column]['scalingFactor'] - del column_map[column]['scalingFactor'] - except KeyError: - unit = np.nan - scaling_factor = np.nan - - # if StatVar not in mcf_dict, add dcid - dcid = column_map[column]['Node'] - - if dcid not in self.mcf_dict: - ## key --> node dcid - self.mcf_dict[dcid] = {} - ## add pvs to dict - for key, value in column_map[column].items(): - if key != 'Node': - self.mcf_dict[dcid][key] = value - - obs_df['ScalingFactor'] = scaling_factor - obs_df['Year'] = year - obs_df['Column'] = column - - # Replace empty places (unresolved geoIds) as null values - obs_df['Place'].replace('', np.nan, inplace=True) - - # Drop rows with observations for empty (null) values - obs_df.dropna(subset=['Place', 'Quantity'], - axis=0, - inplace=True) - - # Write the processed observations to the clean_csv - if self.year_count == 0: - obs_df.to_csv(csv_file, header=True, index=False, mode='w') - else: - obs_df.to_csv(csv_file, header=False, index=False, mode='a') - self.year_count += 1 - - # update stats for the year: - self.counter_dict[year]["number of unique geos"] = len( - obs_df['Place'].unique()) - self.counter_dict[year][ - "number of observations"] += obs_df.shape[0] - self.counter_dict[year][ - "number of unique StatVars with observations"] += len( - obs_df['StatVar'].unique()) - self.counter_dict[year]["number of StatVars in mcf_dict"] = len( - list(self.mcf_dict.keys())) - csv_file.close() - n_obs = self.counter_dict[year]['number of observations'] - n_sv = self.counter_dict[year][ - 'number of unique StatVars with observations'] - n_geos = self.counter_dict[year]['number of unique geos'] - print( - f" Completed with {n_obs} observation for {n_sv} StatVars at " - f"{n_geos} places. ", - flush=True) +import json + +from absl import app, flags +# TODO: logs from the column map step is empty when invoked from here, needs to be checked + +# Allows the following module imports to work when running as a script +_SCRIPT_PATH = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, _SCRIPT_PATH) # for local S2303 data_loader +sys.path.append(os.path.join( + _SCRIPT_PATH, '../common')) # for generate_col_map, resolve_geo_id + +from generate_col_map import generate_stat_var_map, process_zip_file +from data_loader import process_subject_tables, S2303SubjectTableDataLoader + +FLAGS = flags.FLAGS +flags.DEFINE_string( + 'option', 'all', + 'Specify how to run the process, colmap -- generates column map, process -- runs processing, all -- runs colmap first and then proessing' +) +flags.DEFINE_string( + 'table_prefix', None, + '[for processing]Subject Table ID as a prefix for output files, eg: S2702') +flags.DEFINE_string('spec_path', None, 'Path to the JSON spec [mandatory]') +flags.DEFINE_string( + 'input_path', None, + 'Path to input directory with (current support only for zip files)') +flags.DEFINE_string('output_dir', './', 'Path to the output directory') +flags.DEFINE_boolean( + 'has_percent', False, + '[for processing]Specify the datasets has percentage values that needs to be convered to counts' +) +flags.DEFINE_boolean( + 'debug', False, + '[for processing]set the flag to add additional columns to debug') + + +def set_column_map(input_path, spec_path, output_dir): + generated_col_map = process_zip_file(input_path, + spec_path, + write_output=False) + f = open(os.path.join(output_dir, 'column_map.json'), 'w') + json.dump(generated_col_map, f, indent=4) + f.close() + + +def main(argv): + ## get inputs from flags + option = FLAGS.option.lower() + table_prefix = FLAGS.table_prefix + spec_path = FLAGS.spec_path + input_path = FLAGS.input_path + output_dir = FLAGS.output_dir + has_percent = FLAGS.has_percent + debug = FLAGS.debug + + # TODO: remove the constraint of inputs being only zip file + # context: the current implementation of the column map generator accepts + # only zip files as input and we will need to add new methods to handle inputs + # as a directory of files or a single csv file. + _, file_extension = os.path.splitext(input_path) + + if file_extension == '.zip': + if option == 'colmap': + set_column_map(input_path, spec_path, output_dir) + if option == 'process': + process_subject_tables(table_prefix=table_prefix, + input_path=input_path, + output_dir=output_dir, + column_map_path=os.path.join( + output_dir, 'column_map.json'), + spec_path=spec_path, + debug=debug, + delimiter='!!', + has_percent=has_percent) + + if option == 'all': + set_column_map(input_path, spec_path, output_dir) + process_subject_tables(table_prefix=table_prefix, + input_path=input_path, + output_dir=output_dir, + column_map_path=os.path.join( + output_dir, 'column_map.json'), + spec_path=spec_path, + debug=debug, + delimiter='!!', + has_percent=has_percent) + else: + print("At the moment, we support only zip files.") + + +if __name__ == '__main__': + flags.mark_flags_as_required( + ['table_prefix', 'spec_path', 'input_path', 'output_dir']) + app.run(main) From 1a0c4bd4f3a02bccaaa33855930558ab6b75e401 Mon Sep 17 00:00:00 2001 From: goutamabrol Date: Sat, 17 Oct 2026 04:15:00 +0000 Subject: [PATCH 2/7] code review --- .../subject_tables/S2303/data_loader.py | 79 +++++++++---------- .../acs5yr/subject_tables/S2303/process.py | 12 ++- .../subject_tables/S2303/process_test.py | 6 +- 3 files changed, 49 insertions(+), 48 deletions(-) diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py b/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py index 2eab35ffbb..2c3fafe1d9 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py @@ -109,12 +109,11 @@ def process_subject_tables(table_prefix='', _, file_extension = os.path.splitext(input_path) if file_extension == '.zip': data_loader._process_zip_file(input_path) - elif file_extension == '.csv': data_loader._process_csv_file(input_path) - else: data_loader._process_dir(input_path) + data_loader._get_outputs() class SubjectTableDataLoaderBase: @@ -171,24 +170,24 @@ def __init__(self, ## get the content of JSON spec try: - f = open(json_spec, 'r') - self.json_spec = json.load(f) - f.close() - except: + with open(json_spec, 'r') as f: + self.json_spec = json.load(f) + except Exception as e: logging.fatal( - "Please ensure that the path to the JSON spec is correct.") + f"Please ensure that the path to the JSON spec is correct. Error: {e}" + ) ## get the content from Column map try: - f = open(column_map_path, 'r') - self.column_map = json.load(f) - f.close() - except: + with open(column_map_path, 'r') as f: + self.column_map = json.load(f) + except Exception as e: # TODO: If the module is called from `process.py`, the message needs # to be to run with --option=colmap. So rephrasing this message will help. logging.fatal( - """To run this module in standalone mode, ensure the column map - is generated by running `generate_col_map` module.""") + f"To run this module in standalone mode, ensure the column map " + f"is generated by running `generate_col_map` module. Error: {e}" + ) ## initialize other class variables self.mcf_dict = {} @@ -392,9 +391,8 @@ def _generate_tmcf_from_clean_csv(self): col for col in clean_csv.columns if clean_csv[col].isnull().all() ] - f = open(self.base_tmcf_path, 'r') - tmcf_contents = f.read().splitlines() - f.close() + with open(self.base_tmcf_path, 'r') as f: + tmcf_contents = f.read().splitlines() ## modify the tmcf file content for col in empty_cols: @@ -416,19 +414,17 @@ def _get_outputs(self): """aggregated method which stores tmcf and clean_csv to output directory""" ### write to mcf final_mcf = self._generate_mcf_from_column_map() - f = open( - os.path.join(self.output_path_dir, f'./{self.table_id}_output.mcf'), - 'w') - f.write(final_mcf) - f.close() + mcf_path = os.path.join(self.output_path_dir, + f'{self.table_id}_output.mcf') + with open(mcf_path, 'w') as f: + f.write(final_mcf) ## write to tmcf final_tmcf = self._generate_tmcf_from_clean_csv() - f = open( - os.path.join(self.output_path_dir, - f'./{self.table_id}_output.tmcf'), 'w') - f.write(final_tmcf) - f.close() + tmcf_path = os.path.join(self.output_path_dir, + f'{self.table_id}_output.tmcf') + with open(tmcf_path, 'w') as f: + f.write(final_tmcf) ## write all stat_vars from column map to file for debugging if self.debug: @@ -445,19 +441,17 @@ def _get_outputs(self): debug_df.to_csv(os.path.join(self.output_path_dir, './column_to_dcid.csv'), index=False) - f = open( - os.path.join(self.output_path_dir, - './column_to_statvar_map.json'), 'w') - json.dump(self.column_map, f, indent=4) - f.close() + map_path = os.path.join(self.output_path_dir, + 'column_to_statvar_map.json') + with open(map_path, 'w') as f: + json.dump(self.column_map, f, indent=4) ## write summary dict of counters counter_dict = self._get_summary() summary_path = os.path.join(self.output_path_dir, f'{self.table_id}_summary.json') - f = open(summary_path, 'w') - json.dump(counter_dict, f, indent=4) - f.close() + with open(summary_path, 'w') as f: + json.dump(counter_dict, f, indent=4) logging.info( f"""processing of subject table is complete: the summary of the process is available at {summary_path} along with the generated @@ -473,14 +467,18 @@ def _process_zip_file(self, zip_file_path): low_memory=False) self._process_dataframe(df, filename) zf.close() - self._get_outputs() def _process_dir(self, input_dir_path): """specify the folder_path with data files to generate StatVarObs csv""" + if not os.path.exists(input_dir_path): + raise FileNotFoundError( + f"Directory does not exist: {input_dir_path}") + if not os.listdir(input_dir_path): + raise FileNotFoundError(f"Directory is empty: {input_dir_path}") try: for filename in os.listdir(input_dir_path): ## if zip file is present in directory, try _process_zip_file() - if os.path.splitext(filename)[1] == 'zip': + if os.path.splitext(filename)[1] == '.zip': self._process_zip_file( os.path.join(input_dir_path, filename)) ## if input_directory has csv files, process them @@ -490,16 +488,15 @@ def _process_dir(self, input_dir_path): header=self.header_row, low_memory=False) self._process_dataframe(df, filename) - self._get_outputs() - except: + except Exception: logging.fatal( - """ensure input path is that of directory with dataset files, - a zip file or a csv file""") + "Ensure input path is that of directory with dataset files, " + "a zip file or a csv file") + raise def _process_csv_file(self, input_path): df = pd.read_csv(input_path, header=self.header_row, low_memory=False) self._process_dataframe(df, input_path) - self._get_outputs() S2303SubjectTableDataLoader = SubjectTableDataLoaderBase diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/process.py b/scripts/us_census/acs5yr/subject_tables/S2303/process.py index a36a053055..2a2cdcce6d 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/process.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/process.py @@ -55,14 +55,18 @@ def set_column_map(input_path, spec_path, output_dir): generated_col_map = process_zip_file(input_path, spec_path, write_output=False) - f = open(os.path.join(output_dir, 'column_map.json'), 'w') - json.dump(generated_col_map, f, indent=4) - f.close() + col_map_path = os.path.join(output_dir, 'column_map.json') + with open(col_map_path, 'w') as f: + json.dump(generated_col_map, f, indent=4) def main(argv): ## get inputs from flags option = FLAGS.option.lower() + if option not in ['colmap', 'process', 'all']: + sys.exit( + f"Error: Invalid option '{option}'. Must be one of 'colmap', 'process', 'all'." + ) table_prefix = FLAGS.table_prefix spec_path = FLAGS.spec_path input_path = FLAGS.input_path @@ -102,7 +106,7 @@ def main(argv): delimiter='!!', has_percent=has_percent) else: - print("At the moment, we support only zip files.") + sys.exit("Error: At the moment, we support only zip files.") if __name__ == '__main__': diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py b/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py index 2faf5ba82f..1fb9f9b85c 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py @@ -32,9 +32,8 @@ def set_column_map(input_path, spec_path, output_dir): generated_col_map = process_zip_file(input_path, spec_path, write_output=False) - f = open(os.path.join(output_dir, 'column_map.json'), 'w') - json.dump(generated_col_map, f, indent=4) - f.close() + with open(os.path.join(output_dir, 'column_map.json'), 'w') as f: + json.dump(generated_col_map, f, indent=4) class ProcessTest(unittest.TestCase): @@ -55,6 +54,7 @@ def test_create_csv_mcf_tmcf(self): estimate_period=5, header_row=1) data_loader._process_zip_file(_TEST_DATA_ZIP) + data_loader._get_outputs() mcf_file_path = os.path.join(tmp_dir, 'S2303_output.mcf') tmcf_file_path = os.path.join(tmp_dir, 'S2303_output.tmcf') From 7d149a439d016930475b11f0624832423ebeda99 Mon Sep 17 00:00:00 2001 From: goutamabrol Date: Sat, 17 Oct 2026 06:36:23 +0000 Subject: [PATCH 3/7] Revert "code review" This reverts commit 1a0c4bd4f3a02bccaaa33855930558ab6b75e401. --- .../subject_tables/S2303/data_loader.py | 79 ++++++++++--------- .../acs5yr/subject_tables/S2303/process.py | 12 +-- .../subject_tables/S2303/process_test.py | 6 +- 3 files changed, 48 insertions(+), 49 deletions(-) diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py b/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py index 2c3fafe1d9..2eab35ffbb 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py @@ -109,11 +109,12 @@ def process_subject_tables(table_prefix='', _, file_extension = os.path.splitext(input_path) if file_extension == '.zip': data_loader._process_zip_file(input_path) + elif file_extension == '.csv': data_loader._process_csv_file(input_path) + else: data_loader._process_dir(input_path) - data_loader._get_outputs() class SubjectTableDataLoaderBase: @@ -170,24 +171,24 @@ def __init__(self, ## get the content of JSON spec try: - with open(json_spec, 'r') as f: - self.json_spec = json.load(f) - except Exception as e: + f = open(json_spec, 'r') + self.json_spec = json.load(f) + f.close() + except: logging.fatal( - f"Please ensure that the path to the JSON spec is correct. Error: {e}" - ) + "Please ensure that the path to the JSON spec is correct.") ## get the content from Column map try: - with open(column_map_path, 'r') as f: - self.column_map = json.load(f) - except Exception as e: + f = open(column_map_path, 'r') + self.column_map = json.load(f) + f.close() + except: # TODO: If the module is called from `process.py`, the message needs # to be to run with --option=colmap. So rephrasing this message will help. logging.fatal( - f"To run this module in standalone mode, ensure the column map " - f"is generated by running `generate_col_map` module. Error: {e}" - ) + """To run this module in standalone mode, ensure the column map + is generated by running `generate_col_map` module.""") ## initialize other class variables self.mcf_dict = {} @@ -391,8 +392,9 @@ def _generate_tmcf_from_clean_csv(self): col for col in clean_csv.columns if clean_csv[col].isnull().all() ] - with open(self.base_tmcf_path, 'r') as f: - tmcf_contents = f.read().splitlines() + f = open(self.base_tmcf_path, 'r') + tmcf_contents = f.read().splitlines() + f.close() ## modify the tmcf file content for col in empty_cols: @@ -414,17 +416,19 @@ def _get_outputs(self): """aggregated method which stores tmcf and clean_csv to output directory""" ### write to mcf final_mcf = self._generate_mcf_from_column_map() - mcf_path = os.path.join(self.output_path_dir, - f'{self.table_id}_output.mcf') - with open(mcf_path, 'w') as f: - f.write(final_mcf) + f = open( + os.path.join(self.output_path_dir, f'./{self.table_id}_output.mcf'), + 'w') + f.write(final_mcf) + f.close() ## write to tmcf final_tmcf = self._generate_tmcf_from_clean_csv() - tmcf_path = os.path.join(self.output_path_dir, - f'{self.table_id}_output.tmcf') - with open(tmcf_path, 'w') as f: - f.write(final_tmcf) + f = open( + os.path.join(self.output_path_dir, + f'./{self.table_id}_output.tmcf'), 'w') + f.write(final_tmcf) + f.close() ## write all stat_vars from column map to file for debugging if self.debug: @@ -441,17 +445,19 @@ def _get_outputs(self): debug_df.to_csv(os.path.join(self.output_path_dir, './column_to_dcid.csv'), index=False) - map_path = os.path.join(self.output_path_dir, - 'column_to_statvar_map.json') - with open(map_path, 'w') as f: - json.dump(self.column_map, f, indent=4) + f = open( + os.path.join(self.output_path_dir, + './column_to_statvar_map.json'), 'w') + json.dump(self.column_map, f, indent=4) + f.close() ## write summary dict of counters counter_dict = self._get_summary() summary_path = os.path.join(self.output_path_dir, f'{self.table_id}_summary.json') - with open(summary_path, 'w') as f: - json.dump(counter_dict, f, indent=4) + f = open(summary_path, 'w') + json.dump(counter_dict, f, indent=4) + f.close() logging.info( f"""processing of subject table is complete: the summary of the process is available at {summary_path} along with the generated @@ -467,18 +473,14 @@ def _process_zip_file(self, zip_file_path): low_memory=False) self._process_dataframe(df, filename) zf.close() + self._get_outputs() def _process_dir(self, input_dir_path): """specify the folder_path with data files to generate StatVarObs csv""" - if not os.path.exists(input_dir_path): - raise FileNotFoundError( - f"Directory does not exist: {input_dir_path}") - if not os.listdir(input_dir_path): - raise FileNotFoundError(f"Directory is empty: {input_dir_path}") try: for filename in os.listdir(input_dir_path): ## if zip file is present in directory, try _process_zip_file() - if os.path.splitext(filename)[1] == '.zip': + if os.path.splitext(filename)[1] == 'zip': self._process_zip_file( os.path.join(input_dir_path, filename)) ## if input_directory has csv files, process them @@ -488,15 +490,16 @@ def _process_dir(self, input_dir_path): header=self.header_row, low_memory=False) self._process_dataframe(df, filename) - except Exception: + self._get_outputs() + except: logging.fatal( - "Ensure input path is that of directory with dataset files, " - "a zip file or a csv file") - raise + """ensure input path is that of directory with dataset files, + a zip file or a csv file""") def _process_csv_file(self, input_path): df = pd.read_csv(input_path, header=self.header_row, low_memory=False) self._process_dataframe(df, input_path) + self._get_outputs() S2303SubjectTableDataLoader = SubjectTableDataLoaderBase diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/process.py b/scripts/us_census/acs5yr/subject_tables/S2303/process.py index 2a2cdcce6d..a36a053055 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/process.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/process.py @@ -55,18 +55,14 @@ def set_column_map(input_path, spec_path, output_dir): generated_col_map = process_zip_file(input_path, spec_path, write_output=False) - col_map_path = os.path.join(output_dir, 'column_map.json') - with open(col_map_path, 'w') as f: - json.dump(generated_col_map, f, indent=4) + f = open(os.path.join(output_dir, 'column_map.json'), 'w') + json.dump(generated_col_map, f, indent=4) + f.close() def main(argv): ## get inputs from flags option = FLAGS.option.lower() - if option not in ['colmap', 'process', 'all']: - sys.exit( - f"Error: Invalid option '{option}'. Must be one of 'colmap', 'process', 'all'." - ) table_prefix = FLAGS.table_prefix spec_path = FLAGS.spec_path input_path = FLAGS.input_path @@ -106,7 +102,7 @@ def main(argv): delimiter='!!', has_percent=has_percent) else: - sys.exit("Error: At the moment, we support only zip files.") + print("At the moment, we support only zip files.") if __name__ == '__main__': diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py b/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py index 1fb9f9b85c..2faf5ba82f 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py @@ -32,8 +32,9 @@ def set_column_map(input_path, spec_path, output_dir): generated_col_map = process_zip_file(input_path, spec_path, write_output=False) - with open(os.path.join(output_dir, 'column_map.json'), 'w') as f: - json.dump(generated_col_map, f, indent=4) + f = open(os.path.join(output_dir, 'column_map.json'), 'w') + json.dump(generated_col_map, f, indent=4) + f.close() class ProcessTest(unittest.TestCase): @@ -54,7 +55,6 @@ def test_create_csv_mcf_tmcf(self): estimate_period=5, header_row=1) data_loader._process_zip_file(_TEST_DATA_ZIP) - data_loader._get_outputs() mcf_file_path = os.path.join(tmp_dir, 'S2303_output.mcf') tmcf_file_path = os.path.join(tmp_dir, 'S2303_output.tmcf') From 7466aaaf377172b0ccb7042bae939c4b226ade0a Mon Sep 17 00:00:00 2001 From: goutamabrol Date: Sat, 17 Oct 2026 08:24:26 +0000 Subject: [PATCH 4/7] Refactor S2303 process.py to subclass SubjectTableDataLoaderBase and fix test discovery --- .../acs5yr/subject_tables/S2303/README.md | 3 +- .../acs5yr/subject_tables/S2303/__init__.py | 0 .../subject_tables/S2303/data_loader.py | 505 ------------------ .../acs5yr/subject_tables/S2303/process.py | 242 +++++++-- .../subject_tables/S2303/process_test.py | 11 +- 5 files changed, 197 insertions(+), 564 deletions(-) create mode 100644 scripts/us_census/acs5yr/subject_tables/S2303/__init__.py delete mode 100644 scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/README.md b/scripts/us_census/acs5yr/subject_tables/S2303/README.md index f5f73260a5..9623f45da8 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/README.md +++ b/scripts/us_census/acs5yr/subject_tables/S2303/README.md @@ -57,7 +57,8 @@ done ``` - `--option=all`: Runs both column map generation (`column_map.json`) and observation processing (`_cleaned.csv`, `_output.mcf`, `_output.tmcf`, `_summary.json`). -- `--has_percent=True`: Converts percentage estimates (`2010`–`2014`) into counts using the mapped denominator keys in `S2303_spec.json` (`data_loader.py` restricts conversion to `percent_years = ['2014', '2013', '2012', '2011', '2010']`). +- `--has_percent=True`: Converts percentage estimates (`2010`–`2014`) into counts using the mapped denominator keys in `S2303_spec.json` (`S2303SubjectTableDataLoader` in `process.py` restricts conversion to `percent_years = ['2014', '2013', '2012', '2011', '2010']`). +- Only a single shared `s2303_output.tmcf` and `s2303_2013_2015_output.mcf` (which spans both the `2010`–`2014` and `2015`–`2024` column layouts and contains all 210 unique StatVar nodes) need to be retained in the final import folder alongside the 5 `*_cleaned.csv` batch files. ## Batch Splitting & Runtime Optimization diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/__init__.py b/scripts/us_census/acs5yr/subject_tables/S2303/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py b/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py deleted file mode 100644 index 2eab35ffbb..0000000000 --- a/scripts/us_census/acs5yr/subject_tables/S2303/data_loader.py +++ /dev/null @@ -1,505 +0,0 @@ -# Copyright 2021 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. -"""Common data processing module for ACS Subject Tables""" -import os -import sys -import json -from zipfile import ZipFile -from absl import logging -import numpy as np -import pandas as pd - -# Allows the following module imports to work when running as a script -_SCRIPT_PATH = os.path.dirname(os.path.abspath(__file__)) - -sys.path.append(os.path.join(_SCRIPT_PATH, '../common')) # for resolve_geo_id -from resolve_geo_id import convert_to_place_dcid - -_IGNORED_VALUES = set(['-', '*', '**', '***', '*****', 'N', '(X)', 'null']) - -# TODO: Consider using logs for logging warning messages for debug at a later -# date - - -def process_subject_tables(table_prefix='', - input_path='./', - output_dir='./', - column_map_path=None, - spec_path=None, - debug=False, - delimiter='!!', - has_percent=False, - decimal_places=3, - header_row=1, - estimate_period=5): - """ - Wrapper method for invoking the data processing module, maps files with the - corresponding processing function based on input format (*.zip/*.csv), - output files are saved in the specified output directory. - - NOTE: This module does not generate the column_map, you will need to run the - `generate_col_map.py` module before running this module in standalone mode. - - NOTE: If the path for column_map is not specified this module will exit with - an error, stating that the column_map is a required input. - - Args: - table_prefix: string prefix that will be added to differentiate outputs - input_path: path to the input file(s). The module supports inputs as either - a zip file or a csv file or a directory with zip/ csv files - output_dir: path to the output directory - column_map_path: path to the column_map generated with `generate_col_map.py` - spec_path: path to the JSON specification for the subject table - debug: if set, adds additional columns to output csv for debugging (default: False) - delimiter: String denoting the delimiter character in dataset (default: '!!') - has_percent: if set, converts percent to count values (default: False) - decimal_places: specifies the number of digits after decimal place (default: 3) - estimate_period: specifies the duration of the subject period estimate (default: 5) - : since we use the 5 year subject table estimate - header_row: specifies the row index to be considered as the dataframe's - header for column name assignment. For subject tables, the second row (row index = 1) - has a human readable column header. (default: 1) - header_row: specifies the row index to be considered as the dataframe's header - for column name assignment. For subject tables, the second row (row index = 1) - has a human readable column header. (default: 1) - - Outputs: - The module will write the following files to the specified output directory - _cleaned.csv: - csv file with StatVarObservations i.e. quantity measured for a stat_var at a particular place and date. - It also has additional columns like units, and scalingFactor if they are specified in the JSON spec. - _output.mcf: - file in mcf format with all the StatVars generated in the column map - _output.tmcf: - template mcf file used in combination with the csv file for data import - _summary.json: - summary stats of data processing year-wise - - If debug=True, additional output files are written to the output directory - column_to_dcid.csv: - maps the stat var dcid associated to each column of the data file - column_to_statvar_map.json: - a json file with column mapped to the complete stat-var node - _cleaned.csv will have an additional column containing the name of column in the dataset for which each StatVarObservation was recorded - """ - ## create a data loader object with base parameters - data_loader = SubjectTableDataLoaderBase(table_id=table_prefix, - col_delimiter=delimiter, - has_percent=has_percent, - debug=debug, - output_path_dir=output_dir, - json_spec=spec_path, - column_map_path=column_map_path, - decimal_places=decimal_places, - estimate_period=estimate_period, - header_row=header_row) - - ## if input_path is a file, select csv/zip processing methods - _, file_extension = os.path.splitext(input_path) - if file_extension == '.zip': - data_loader._process_zip_file(input_path) - - elif file_extension == '.csv': - data_loader._process_csv_file(input_path) - - else: - data_loader._process_dir(input_path) - - -class SubjectTableDataLoaderBase: - """ - A common module that has a collection of methods for subject table data processing. - - - Attributes: - table_prefix: string prefix that will be added to differentiate outputs - input_path: path to the input file(s). The module supports inputs as either - a zip file or a csv file or a directory with zip/ csv files - output_dir: path to the output directory - column_map_path: path to the column_map generated with `generate_col_map.py` - spec_path: path to the JSON specification for the subject table - debug: if set, adds additional columns to output csv for debugging (default: False) - col_delimiter: String denoting the delimiter character in dataset (default: '!!') - has_percent: if set, converts percent to count values (default: False) - decimal_places: specifies the number of digits after decimal place (default: 3) - estimate_period: specifies the duration of the subject period estimate (default: 5) - : since we use the 5 year subject table estimate - header_row: specifies the row index to be considered as the dataframe's header - for column name assignment. For subject tables, the second row (row index = 1) - has a human readable column header. (default: 1) - - """ - - def __init__(self, - table_id='', - col_delimiter='!!', - has_percent=False, - debug=False, - output_path_dir='./', - json_spec='./', - column_map_path='./', - decimal_places=3, - estimate_period=5, - header_row=1): - """module init""" - ## static inputs - self.table_id = table_id - self.has_percent = has_percent - self.debug = debug - self.decimal_places = decimal_places - self.delimiter = col_delimiter - self.estimate_period = estimate_period - self.header_row = header_row - - ## set the output path, if not exists, make directory - if not os.path.exists(output_path_dir): - os.mkdir(output_path_dir) - self.output_path_dir = output_path_dir - self.base_tmcf_path = os.path.join(_SCRIPT_PATH, - '../common/subject_table_base.tmcf') - - ## get the content of JSON spec - try: - f = open(json_spec, 'r') - self.json_spec = json.load(f) - f.close() - except: - logging.fatal( - "Please ensure that the path to the JSON spec is correct.") - - ## get the content from Column map - try: - f = open(column_map_path, 'r') - self.column_map = json.load(f) - f.close() - except: - # TODO: If the module is called from `process.py`, the message needs - # to be to run with --option=colmap. So rephrasing this message will help. - logging.fatal( - """To run this module in standalone mode, ensure the column map - is generated by running `generate_col_map` module.""") - - ## initialize other class variables - self.mcf_dict = {} - - # initialize a dictionary of counters - # TODO: Add a means to fill the git commit SHA - self.counter_dict = { - "summary": { - "column map generator commit SHA": "", - "data processing script commit SHA": "", - "total StatVarObservations": 0, - }, - } - - # initialize the StatVarObservation csv - self.csv_columns = [ - 'Year', 'Place', 'StatVar', 'Quantity', 'Unit', 'ScalingFactor', - 'Column' - ] #optional: Unit, ScalingFactor - self.clean_csv_path = os.path.join(output_path_dir, - f'./{table_id}_cleaned.csv') - self.year_count = 0 - if os.path.exists(self.clean_csv_path): - os.remove(self.clean_csv_path) - - def _convert_percent_to_numbers(self, df): - """utility to convert percent to numbers""" - # TODO: Currently this function expects full names for function, which - # makes the spec long, perhaps we can add support for tokens - df_columns = df.columns.tolist() - for count_col, percent_col_list in self.json_spec['denominators'].items( - ): - # TODO: add a warning message - if count_col in df_columns: - for col in percent_col_list: - if col in df_columns: - df[col] = df[col].astype(float) - df[count_col] = df[count_col].astype(float) - df[col] = (df[col] / 100) * df[count_col] - df[col] = df[col].round( - 3 - ) # TODO: replace with parameter self.decimal_places - return df - - def _get_summary(self): - """ - method returns the total number of columns, stat vars and other stats - from the column_map and csv. - """ - for key, count_dict in self.counter_dict.items(): - if key != 'summary': - self.counter_dict['summary'][ - "total StatVarObservations"] += count_dict[ - "number of observations"] - return self.counter_dict - - def _replace_ignore_values_with_nan(self, df): - """ - Replaces cells of the dataframe by column with nan if cell has values in - _IGNORED_VALUES. - """ - for col in df.columns.tolist(): - df[col] = df[col].apply(lambda row: np.nan - if row in _IGNORED_VALUES else row) - return df - - def _process_dataframe(self, df, filename): - """processes a dataframe read from a csv file""" - df = self._replace_ignore_values_with_nan( - df) #handle the values to be ignored - year = filename.split(f'ACSST{self.estimate_period}Y')[1][:4] - logging.info(f"Processing: {filename}") - # if has_percent is set, convert percentages to counts. Requires the - # 'denominators' key to be specified in the spec - percent_years = self.json_spec.get( - 'percent_years', ['2014', '2013', '2012', '2011', '2010']) - if self.has_percent and year in percent_years: - df = self._convert_percent_to_numbers(df) - - # if column map is not available generate, will rarely be False - column_map = self.column_map[year] - - # add stats to dict_counter for current year - self.counter_dict[year] = { - "filename": - filename, - "number of columns in dataset": - df.shape[1], - "number of rows in dataset": - df.shape[0], - "number of statVars generated for columns": - len(list(column_map.keys())), - "number of observations": - 0, - "number of unique StatVars with observations": - 0, - } - - csv_file = open(self.clean_csv_path, 'a') - if 'id' in df.columns: - place_geoIds = df['id'].apply(convert_to_place_dcid) - else: - place_geoIds = df['Geography'].apply(convert_to_place_dcid) - - # update the clean csv - for column in df.columns.tolist(): - if column in column_map: - obs_df = pd.DataFrame(columns=self.csv_columns) - obs_df['Place'] = place_geoIds - obs_df['StatVar'] = column_map[column]['Node'] - obs_df['Quantity'] = df[column].values.tolist() - # add unit to the csv - try: - unit = column_map[column]['unit'] - del column_map[column]['unit'] - except KeyError: - unit = np.nan - obs_df['Unit'] = unit - - # add scaling factor to the csv - try: - scalingFactor = column_map[column]['scalingFactor'] - del column_map[column]['scalingFactor'] - except KeyError: - scalingFactor = np.nan - obs_df['ScalingFactor'] = scalingFactor - - # if StatVar not in mcf_dict, add dcid - dcid = column_map[column]['Node'] - - if dcid not in self.mcf_dict: - ## key --> node dcid - self.mcf_dict[dcid] = {} - ## add pvs to dict - for key, value in column_map[column].items(): - if key != 'Node': - self.mcf_dict[dcid][key] = value - - obs_df['Year'] = year - obs_df['Column'] = column - - # Replace empty places (unresolved geoIds) as null values - obs_df['Place'] = obs_df['Place'].replace('', np.nan) - - # Drop rows with observations for empty (null) values - obs_df.dropna(subset=['Place', 'Quantity'], - axis=0, - inplace=True) - - # Write the processed observations to the clean_csv - if self.year_count == 0: - obs_df.to_csv(csv_file, header=True, index=False, mode='w') - else: - obs_df.to_csv(csv_file, header=False, index=False, mode='a') - self.year_count += 1 - - # update stats for the year: - self.counter_dict[year]["number of unique geos"] = len( - obs_df['Place'].unique()) - self.counter_dict[year][ - "number of observations"] += obs_df.shape[0] - self.counter_dict[year][ - "number of unique StatVars with observations"] += len( - obs_df['StatVar'].unique()) - self.counter_dict[year]["number of StatVars in mcf_dict"] = len( - list(self.mcf_dict.keys())) - csv_file.close() - logging.info( - f"""Completed with { self.counter_dict[year]['number of observations'] } - observation for { self.counter_dict[year]['number of unique StatVars with observations'] } - StatVars at { self.counter_dict[year]['number of unique geos'] } places. - """) - - def _generate_mcf_from_column_map(self): - """ - uses column_map to generate a mcf file with all the stat_vars nodes that - are created. - """ - - #get cleaned column map - final_mcf = "" - - for column, stat_var in self.mcf_dict.items(): - col_stat_var = [] - # TODO: Add 'Node:' as the first element of col_stat_var, and append - # all the stat-var node information to col_stat_var - dcid = 'Node: ' + column + '\n' - - for p, v in stat_var.items(): - if p != 'unit' or p != 'scalingFactor': - col_stat_var.append(f"{p}: {v}") - # TODO: Replace the next two lines with final_mcf = final_mcf + '\n'.join(col_stat_var) + '\n\n' - col_stat_var = dcid + '\n'.join(col_stat_var) - final_mcf = final_mcf + col_stat_var + "\n\n" - return final_mcf - - def _generate_tmcf_from_clean_csv(self): - """update the base tmcf template based on the clean_csv""" - clean_csv = pd.read_csv(self.clean_csv_path, low_memory=False) - empty_cols = [ - col for col in clean_csv.columns if clean_csv[col].isnull().all() - ] - - f = open(self.base_tmcf_path, 'r') - tmcf_contents = f.read().splitlines() - f.close() - - ## modify the tmcf file content - for col in empty_cols: - for index, line in enumerate(tmcf_contents): - if col in line: - del tmcf_contents[index] - - ## write into the output tmcf - tmcf_contents = '\n'.join(tmcf_contents) - if not self.debug: - clean_csv.drop(columns=['Column'], inplace=True) - - ## update the csv by dropping empty column - clean_csv.drop(columns=empty_cols, inplace=True) - clean_csv.to_csv(self.clean_csv_path, index=False) - return tmcf_contents - - def _get_outputs(self): - """aggregated method which stores tmcf and clean_csv to output directory""" - ### write to mcf - final_mcf = self._generate_mcf_from_column_map() - f = open( - os.path.join(self.output_path_dir, f'./{self.table_id}_output.mcf'), - 'w') - f.write(final_mcf) - f.close() - - ## write to tmcf - final_tmcf = self._generate_tmcf_from_clean_csv() - f = open( - os.path.join(self.output_path_dir, - f'./{self.table_id}_output.tmcf'), 'w') - f.write(final_tmcf) - f.close() - - ## write all stat_vars from column map to file for debugging - if self.debug: - debug_df = pd.DataFrame(columns=['Year', 'Column', 'StatVarID']) - for year, val in self.column_map.items(): - for column, stat_var in val.items(): - debug_df = debug_df.append( - { - 'Year': year, - 'Column': column, - 'StatVarID': stat_var['Node'] - }, - ignore_index=True) - debug_df.to_csv(os.path.join(self.output_path_dir, - './column_to_dcid.csv'), - index=False) - f = open( - os.path.join(self.output_path_dir, - './column_to_statvar_map.json'), 'w') - json.dump(self.column_map, f, indent=4) - f.close() - - ## write summary dict of counters - counter_dict = self._get_summary() - summary_path = os.path.join(self.output_path_dir, - f'{self.table_id}_summary.json') - f = open(summary_path, 'w') - json.dump(counter_dict, f, indent=4) - f.close() - logging.info( - f"""processing of subject table is complete: the summary of the - process is available at {summary_path} along with the generated - output files.""") - - def _process_zip_file(self, zip_file_path): - """processes each dataset in a zip file to make the clean csv of StatVarObs""" - zf = ZipFile(zip_file_path) - for filename in zf.namelist(): - if 'data_with_overlays' in filename: - df = pd.read_csv(zf.open(filename, 'r'), - header=self.header_row, - low_memory=False) - self._process_dataframe(df, filename) - zf.close() - self._get_outputs() - - def _process_dir(self, input_dir_path): - """specify the folder_path with data files to generate StatVarObs csv""" - try: - for filename in os.listdir(input_dir_path): - ## if zip file is present in directory, try _process_zip_file() - if os.path.splitext(filename)[1] == 'zip': - self._process_zip_file( - os.path.join(input_dir_path, filename)) - ## if input_directory has csv files, process them - if 'data_with_overlays' in filename: - input_path = os.path.join(input_dir_path, filename) - df = pd.read_csv(input_path, - header=self.header_row, - low_memory=False) - self._process_dataframe(df, filename) - self._get_outputs() - except: - logging.fatal( - """ensure input path is that of directory with dataset files, - a zip file or a csv file""") - - def _process_csv_file(self, input_path): - df = pd.read_csv(input_path, header=self.header_row, low_memory=False) - self._process_dataframe(df, input_path) - self._get_outputs() - - -S2303SubjectTableDataLoader = SubjectTableDataLoaderBase diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/process.py b/scripts/us_census/acs5yr/subject_tables/S2303/process.py index a36a053055..57be14c31e 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/process.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/process.py @@ -11,57 +11,186 @@ # 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. -"""Generic proces module to generate the csv/tmcf and csv""" -# TODO: Add unit tests +"""Process data from Census ACS 5-Year Subject Table S2303.""" + +import json import os import sys -import json -from absl import app, flags -# TODO: logs from the column map step is empty when invoked from here, needs to be checked +from absl import app +from absl import flags +from absl import logging +import numpy as np +import pandas as pd -# Allows the following module imports to work when running as a script -_SCRIPT_PATH = os.path.dirname(os.path.abspath(__file__)) -sys.path.insert(0, _SCRIPT_PATH) # for local S2303 data_loader -sys.path.append(os.path.join( - _SCRIPT_PATH, '../common')) # for generate_col_map, resolve_geo_id +# pylint: disable=wrong-import-position +# pylint: disable=import-error +_CODEDIR = os.path.dirname(os.path.realpath(__file__)) +sys.path.insert(1, os.path.join(_CODEDIR, '..', 'common')) -from generate_col_map import generate_stat_var_map, process_zip_file -from data_loader import process_subject_tables, S2303SubjectTableDataLoader +from data_loader import SubjectTableDataLoaderBase +from generate_col_map import process_zip_file +from resolve_geo_id import convert_to_place_dcid +# pylint: enable=wrong-import-position +# pylint: enable=import-error FLAGS = flags.FLAGS -flags.DEFINE_string( - 'option', 'all', - 'Specify how to run the process, colmap -- generates column map, process -- runs processing, all -- runs colmap first and then proessing' -) -flags.DEFINE_string( - 'table_prefix', None, - '[for processing]Subject Table ID as a prefix for output files, eg: S2702') -flags.DEFINE_string('spec_path', None, 'Path to the JSON spec [mandatory]') -flags.DEFINE_string( - 'input_path', None, - 'Path to input directory with (current support only for zip files)') -flags.DEFINE_string('output_dir', './', 'Path to the output directory') -flags.DEFINE_boolean( - 'has_percent', False, - '[for processing]Specify the datasets has percentage values that needs to be convered to counts' -) -flags.DEFINE_boolean( - 'debug', False, - '[for processing]set the flag to add additional columns to debug') + + +class S2303SubjectTableDataLoader(SubjectTableDataLoaderBase): + """Data loader for Census ACS 5-Year Subject Table S2303.""" + + def _process_dataframe(self, df, filename): + """Processes a dataframe read from a CSV file.""" + df = self._replace_ignore_values_with_nan(df) + year = filename.split(f'ACSST{self.estimate_period}Y')[1][:4] + logging.info(f'Processing: {filename}') + + # Only convert percentages to counts for 2010-2014 (2015+ report counts directly). + percent_years = self.json_spec.get( + 'percent_years', ['2014', '2013', '2012', '2011', '2010']) + if self.has_percent and year in percent_years: + logging.info(f'Converting percent to numbers for {year}') + df = self._convert_percent_to_numbers(df) + + column_map = self.column_map[year] + + # Add stats to counter_dict for current year. + self.counter_dict[year] = { + 'filename': + filename, + 'number of columns in dataset': + df.shape[1], + 'number of rows in dataset': + df.shape[0], + 'number of statVars generated for columns': + len(list(column_map.keys())), + 'number of observations': + 0, + 'number of unique StatVars with observations': + 0, + } + + if 'id' in df.columns: + place_geo_ids = df['id'].apply(convert_to_place_dcid) + else: + place_geo_ids = df['Geography'].apply(convert_to_place_dcid) + + with open(self.clean_csv_path, 'a') as csv_file: + for column in df.columns.tolist(): + if column in column_map: + obs_df = pd.DataFrame(columns=self.csv_columns) + obs_df['Place'] = place_geo_ids + obs_df['StatVar'] = column_map[column]['Node'] + obs_df['Quantity'] = df[column].values.tolist() + + try: + unit = column_map[column]['unit'] + del column_map[column]['unit'] + except KeyError: + unit = np.nan + obs_df['Unit'] = unit + + try: + scaling_factor = column_map[column]['scalingFactor'] + del column_map[column]['scalingFactor'] + except KeyError: + scaling_factor = np.nan + obs_df['ScalingFactor'] = scaling_factor + + dcid = column_map[column]['Node'] + if dcid not in self.mcf_dict: + self.mcf_dict[dcid] = {} + for key, value in column_map[column].items(): + if key != 'Node': + self.mcf_dict[dcid][key] = value + + obs_df['Year'] = year + obs_df['Column'] = column + + # Replace empty places (unresolved geoIds) as null values. + obs_df['Place'] = obs_df['Place'].replace('', np.nan) + + # Drop rows with observations for empty (null) values. + obs_df.dropna(subset=['Place', 'Quantity'], + axis=0, + inplace=True) + + if self.year_count == 0: + obs_df.to_csv(csv_file, + header=True, + index=False, + mode='w') + else: + obs_df.to_csv(csv_file, + header=False, + index=False, + mode='a') + self.year_count += 1 + + self.counter_dict[year]['number of unique geos'] = len( + obs_df['Place'].unique()) + self.counter_dict[year][ + 'number of observations'] += obs_df.shape[0] + self.counter_dict[year][ + 'number of unique StatVars with observations'] += len( + obs_df['StatVar'].unique()) + self.counter_dict[year][ + 'number of StatVars in mcf_dict'] = len( + list(self.mcf_dict.keys())) + + n_obs = self.counter_dict[year]['number of observations'] + n_sv = self.counter_dict[year][ + 'number of unique StatVars with observations'] + n_geos = self.counter_dict[year]['number of unique geos'] + logging.info( + f'Completed with {n_obs} observations for {n_sv} StatVars at ' + f'{n_geos} places.') + + +def process_subject_tables(table_prefix, + input_path, + column_map_path, + spec_path, + output_dir='./', + has_percent=False, + debug=False, + estimate_period=5, + delimiter='!!', + decimal_places=3, + header_row=1): + """Processes S2303 zip/directory/csv inputs into cleaned CSV, MCF, and TMCF.""" + data_loader = S2303SubjectTableDataLoader(table_id=table_prefix, + col_delimiter=delimiter, + has_percent=has_percent, + debug=debug, + output_path_dir=output_dir, + json_spec=spec_path, + column_map_path=column_map_path, + decimal_places=decimal_places, + estimate_period=estimate_period, + header_row=header_row) + + if os.path.isdir(input_path): + data_loader._process_dir(input_path) + else: + _, file_extension = os.path.splitext(input_path) + if file_extension == '.zip': + data_loader._process_zip_file(input_path) + elif file_extension == '.csv': + data_loader._process_csv_file(input_path) def set_column_map(input_path, spec_path, output_dir): generated_col_map = process_zip_file(input_path, spec_path, write_output=False) - f = open(os.path.join(output_dir, 'column_map.json'), 'w') - json.dump(generated_col_map, f, indent=4) - f.close() + with open(os.path.join(output_dir, 'column_map.json'), 'w') as f: + json.dump(generated_col_map, f, indent=4) def main(argv): - ## get inputs from flags + del argv option = FLAGS.option.lower() table_prefix = FLAGS.table_prefix spec_path = FLAGS.spec_path @@ -70,28 +199,12 @@ def main(argv): has_percent = FLAGS.has_percent debug = FLAGS.debug - # TODO: remove the constraint of inputs being only zip file - # context: the current implementation of the column map generator accepts - # only zip files as input and we will need to add new methods to handle inputs - # as a directory of files or a single csv file. _, file_extension = os.path.splitext(input_path) if file_extension == '.zip': - if option == 'colmap': - set_column_map(input_path, spec_path, output_dir) - if option == 'process': - process_subject_tables(table_prefix=table_prefix, - input_path=input_path, - output_dir=output_dir, - column_map_path=os.path.join( - output_dir, 'column_map.json'), - spec_path=spec_path, - debug=debug, - delimiter='!!', - has_percent=has_percent) - - if option == 'all': + if option in ('colmap', 'all'): set_column_map(input_path, spec_path, output_dir) + if option in ('process', 'all'): process_subject_tables(table_prefix=table_prefix, input_path=input_path, output_dir=output_dir, @@ -102,10 +215,31 @@ def main(argv): delimiter='!!', has_percent=has_percent) else: - print("At the moment, we support only zip files.") + print('At the moment, we support only zip files.') if __name__ == '__main__': + flags.DEFINE_string( + 'option', 'all', + 'Specify how to run the process, colmap -- generates column map, ' + 'process -- runs processing, all -- runs colmap first and then ' + 'processing') + flags.DEFINE_string( + 'table_prefix', None, + '[for processing] Subject Table ID as a prefix for output files, ' + 'eg: S2303') + flags.DEFINE_string('spec_path', None, 'Path to the JSON spec [mandatory]') + flags.DEFINE_string( + 'input_path', None, + 'Path to input directory with (current support only for zip files)') + flags.DEFINE_string('output_dir', './', 'Path to the output directory') + flags.DEFINE_boolean( + 'has_percent', False, + '[for processing] Specify if the dataset has percentage values that ' + 'need to be converted to counts') + flags.DEFINE_boolean( + 'debug', False, + '[for processing] Set the flag to add additional columns to debug') flags.mark_flags_as_required( ['table_prefix', 'spec_path', 'input_path', 'output_dir']) app.run(main) diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py b/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py index 2faf5ba82f..3812cde7a9 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py @@ -14,9 +14,13 @@ #pylint: disable=import-error _CODEDIR = os.path.dirname(os.path.realpath(__file__)) +sys.path.insert(1, _CODEDIR) sys.path.insert(1, os.path.join(_CODEDIR, '..', 'common')) from generate_col_map import process_zip_file -from .process import S2303SubjectTableDataLoader +try: + from .process import S2303SubjectTableDataLoader +except ImportError: + from process import S2303SubjectTableDataLoader #pylint: enable=wrong-import-position #pylint: enable=import-error @@ -32,9 +36,8 @@ def set_column_map(input_path, spec_path, output_dir): generated_col_map = process_zip_file(input_path, spec_path, write_output=False) - f = open(os.path.join(output_dir, 'column_map.json'), 'w') - json.dump(generated_col_map, f, indent=4) - f.close() + with open(os.path.join(output_dir, 'column_map.json'), 'w') as f: + json.dump(generated_col_map, f, indent=4) class ProcessTest(unittest.TestCase): From 9b82c826ac643d982bd6c97e8a0510d6c8385512 Mon Sep 17 00:00:00 2001 From: goutamabrol Date: Sat, 17 Oct 2026 09:41:40 +0000 Subject: [PATCH 5/7] Use logging.fatal for invalid CLI option or non-zip input in S2303/process.py --- scripts/us_census/acs5yr/subject_tables/S2303/process.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/process.py b/scripts/us_census/acs5yr/subject_tables/S2303/process.py index 57be14c31e..3d34786596 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/process.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/process.py @@ -192,6 +192,10 @@ def set_column_map(input_path, spec_path, output_dir): def main(argv): del argv option = FLAGS.option.lower() + if option not in ('colmap', 'process', 'all'): + logging.fatal( + f"Invalid --option '{FLAGS.option}'. Must be one of: 'colmap', " + "'process', 'all'.") table_prefix = FLAGS.table_prefix spec_path = FLAGS.spec_path input_path = FLAGS.input_path @@ -215,7 +219,7 @@ def main(argv): delimiter='!!', has_percent=has_percent) else: - print('At the moment, we support only zip files.') + logging.fatal('At the moment, we support only zip files.') if __name__ == '__main__': From 0710d3eb2e0090671f2b602be3c23b9dd008c590 Mon Sep 17 00:00:00 2001 From: goutamabrol Date: Mon, 19 Oct 2026 05:53:09 +0000 Subject: [PATCH 6/7] Remove empty S2303/__init__.py --- scripts/us_census/acs5yr/subject_tables/S2303/__init__.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) delete mode 100644 scripts/us_census/acs5yr/subject_tables/S2303/__init__.py diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/__init__.py b/scripts/us_census/acs5yr/subject_tables/S2303/__init__.py deleted file mode 100644 index e69de29bb2..0000000000 From 2fe89aa7462cecb129fe355916e57741e3e57335 Mon Sep 17 00:00:00 2001 From: goutamabrol Date: Thu, 22 Oct 2026 08:21:42 +0000 Subject: [PATCH 7/7] Address reviewer feedback on S2303/process.py, process_test.py, and README.md --- .../acs5yr/subject_tables/S2303/README.md | 5 +- .../acs5yr/subject_tables/S2303/process.py | 80 ++++++++++--------- .../subject_tables/S2303/process_test.py | 41 ++++------ 3 files changed, 61 insertions(+), 65 deletions(-) diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/README.md b/scripts/us_census/acs5yr/subject_tables/S2303/README.md index 9623f45da8..ac4958e8f3 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/README.md +++ b/scripts/us_census/acs5yr/subject_tables/S2303/README.md @@ -51,14 +51,13 @@ for b in s2303_2010_2012 s2303_2013_2015 s2303_2016_2018 s2303_2019_2021 s2303_2 --spec_path=./S2303_spec.json \ --input_path=input_files/${b}.zip \ --output_dir=./ - mv column_map.json ${b}_column_map.json mv ${b}_output.tmcf s2303_output.tmcf done ``` -- `--option=all`: Runs both column map generation (`column_map.json`) and observation processing (`_cleaned.csv`, `_output.mcf`, `_output.tmcf`, `_summary.json`). +- `--option=all`: Runs both column map generation (`_column_map.json`) and observation processing (`_cleaned.csv`, `_output.mcf`, `_output.tmcf`, `_summary.json`). - `--has_percent=True`: Converts percentage estimates (`2010`–`2014`) into counts using the mapped denominator keys in `S2303_spec.json` (`S2303SubjectTableDataLoader` in `process.py` restricts conversion to `percent_years = ['2014', '2013', '2012', '2011', '2010']`). -- Only a single shared `s2303_output.tmcf` and `s2303_2013_2015_output.mcf` (which spans both the `2010`–`2014` and `2015`–`2024` column layouts and contains all 210 unique StatVar nodes) need to be retained in the final import folder alongside the 5 `*_cleaned.csv` batch files. +- Only a single shared `s2303_output.tmcf` needs to be retained in the final import folder alongside the 5 `*_output.mcf` and 5 `*_cleaned.csv` batch files. ## Batch Splitting & Runtime Optimization diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/process.py b/scripts/us_census/acs5yr/subject_tables/S2303/process.py index 3d34786596..27064bef09 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/process.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/process.py @@ -65,6 +65,8 @@ def _process_dataframe(self, df, filename): df.shape[0], 'number of statVars generated for columns': len(list(column_map.keys())), + 'number of unique geos': + 0, 'number of observations': 0, 'number of unique StatVars with observations': @@ -76,6 +78,9 @@ def _process_dataframe(self, df, filename): else: place_geo_ids = df['Geography'].apply(convert_to_place_dcid) + unique_geos = set() + unique_statvars = set() + with open(self.clean_csv_path, 'a') as csv_file: for column in df.columns.tolist(): if column in column_map: @@ -84,25 +89,15 @@ def _process_dataframe(self, df, filename): obs_df['StatVar'] = column_map[column]['Node'] obs_df['Quantity'] = df[column].values.tolist() - try: - unit = column_map[column]['unit'] - del column_map[column]['unit'] - except KeyError: - unit = np.nan - obs_df['Unit'] = unit - - try: - scaling_factor = column_map[column]['scalingFactor'] - del column_map[column]['scalingFactor'] - except KeyError: - scaling_factor = np.nan - obs_df['ScalingFactor'] = scaling_factor + obs_df['Unit'] = column_map[column].get('unit', np.nan) + obs_df['ScalingFactor'] = column_map[column].get( + 'scalingFactor', np.nan) dcid = column_map[column]['Node'] if dcid not in self.mcf_dict: self.mcf_dict[dcid] = {} for key, value in column_map[column].items(): - if key != 'Node': + if key not in ('Node', 'unit', 'scalingFactor'): self.mcf_dict[dcid][key] = value obs_df['Year'] = year @@ -116,28 +111,22 @@ def _process_dataframe(self, df, filename): axis=0, inplace=True) - if self.year_count == 0: - obs_df.to_csv(csv_file, - header=True, - index=False, - mode='w') - else: - obs_df.to_csv(csv_file, - header=False, - index=False, - mode='a') + obs_df.to_csv(csv_file, + header=(self.year_count == 0), + index=False) self.year_count += 1 + unique_geos.update(obs_df['Place'].unique()) + unique_statvars.update(obs_df['StatVar'].unique()) self.counter_dict[year]['number of unique geos'] = len( - obs_df['Place'].unique()) + unique_geos) self.counter_dict[year][ 'number of observations'] += obs_df.shape[0] self.counter_dict[year][ - 'number of unique StatVars with observations'] += len( - obs_df['StatVar'].unique()) + 'number of unique StatVars with observations'] = len( + unique_statvars) self.counter_dict[year][ - 'number of StatVars in mcf_dict'] = len( - list(self.mcf_dict.keys())) + 'number of StatVars in mcf_dict'] = len(self.mcf_dict) n_obs = self.counter_dict[year]['number of observations'] n_sv = self.counter_dict[year][ @@ -175,18 +164,26 @@ def process_subject_tables(table_prefix, data_loader._process_dir(input_path) else: _, file_extension = os.path.splitext(input_path) - if file_extension == '.zip': + if file_extension == '.zip' and os.path.isfile(input_path): data_loader._process_zip_file(input_path) - elif file_extension == '.csv': + elif file_extension == '.csv' and os.path.isfile(input_path): data_loader._process_csv_file(input_path) + else: + logging.fatal( + f'Unsupported or non-existent input_path: {input_path}') -def set_column_map(input_path, spec_path, output_dir): +def set_column_map(input_path, spec_path, output_dir, table_prefix=None): + os.makedirs(output_dir, exist_ok=True) generated_col_map = process_zip_file(input_path, spec_path, write_output=False) - with open(os.path.join(output_dir, 'column_map.json'), 'w') as f: + col_map_filename = (f'{table_prefix}_column_map.json' + if table_prefix else 'column_map.json') + col_map_path = os.path.join(output_dir, col_map_filename) + with open(col_map_path, 'w') as f: json.dump(generated_col_map, f, indent=4) + return col_map_path def main(argv): @@ -206,14 +203,22 @@ def main(argv): _, file_extension = os.path.splitext(input_path) if file_extension == '.zip': + col_map_path = os.path.join(output_dir, + f'{table_prefix}_column_map.json') if option in ('colmap', 'all'): - set_column_map(input_path, spec_path, output_dir) + col_map_path = set_column_map(input_path, + spec_path, + output_dir, + table_prefix=table_prefix) if option in ('process', 'all'): + if not os.path.exists(col_map_path): + fallback_path = os.path.join(output_dir, 'column_map.json') + if os.path.exists(fallback_path): + col_map_path = fallback_path process_subject_tables(table_prefix=table_prefix, input_path=input_path, output_dir=output_dir, - column_map_path=os.path.join( - output_dir, 'column_map.json'), + column_map_path=col_map_path, spec_path=spec_path, debug=debug, delimiter='!!', @@ -244,6 +249,5 @@ def main(argv): flags.DEFINE_boolean( 'debug', False, '[for processing] Set the flag to add additional columns to debug') - flags.mark_flags_as_required( - ['table_prefix', 'spec_path', 'input_path', 'output_dir']) + flags.mark_flags_as_required(['table_prefix', 'spec_path', 'input_path']) app.run(main) diff --git a/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py b/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py index 3812cde7a9..9e5af17ba3 100644 --- a/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py +++ b/scripts/us_census/acs5yr/subject_tables/S2303/process_test.py @@ -18,9 +18,9 @@ sys.path.insert(1, os.path.join(_CODEDIR, '..', 'common')) from generate_col_map import process_zip_file try: - from .process import S2303SubjectTableDataLoader + from .process import S2303SubjectTableDataLoader, process_subject_tables, set_column_map except ImportError: - from process import S2303SubjectTableDataLoader + from process import S2303SubjectTableDataLoader, process_subject_tables, set_column_map #pylint: enable=wrong-import-position #pylint: enable=import-error @@ -32,32 +32,25 @@ _EXPECTED_CSV = os.path.join(_TEST_DIR, 'expected.csv') -def set_column_map(input_path, spec_path, output_dir): - generated_col_map = process_zip_file(input_path, - spec_path, - write_output=False) - with open(os.path.join(output_dir, 'column_map.json'), 'w') as f: - json.dump(generated_col_map, f, indent=4) - - class ProcessTest(unittest.TestCase): def test_create_csv_mcf_tmcf(self): with tempfile.TemporaryDirectory() as tmp_dir: - set_column_map(_TEST_DATA_ZIP, _CONFIG_PATH, tmp_dir) - column_map_path = os.path.join(tmp_dir, 'column_map.json') - data_loader = S2303SubjectTableDataLoader( - table_id='S2303', - col_delimiter='!!', - has_percent=True, - debug=False, - output_path_dir=tmp_dir, - json_spec=_CONFIG_PATH, - column_map_path=column_map_path, - decimal_places=3, - estimate_period=5, - header_row=1) - data_loader._process_zip_file(_TEST_DATA_ZIP) + column_map_path = set_column_map(_TEST_DATA_ZIP, + _CONFIG_PATH, + tmp_dir, + table_prefix='S2303') + process_subject_tables(table_prefix='S2303', + input_path=_TEST_DATA_ZIP, + output_dir=tmp_dir, + column_map_path=column_map_path, + spec_path=_CONFIG_PATH, + debug=False, + delimiter='!!', + has_percent=True, + decimal_places=3, + estimate_period=5, + header_row=1) mcf_file_path = os.path.join(tmp_dir, 'S2303_output.mcf') tmcf_file_path = os.path.join(tmp_dir, 'S2303_output.tmcf')