diff --git a/scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution.tmcf b/scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution.tmcf index 26fb09c53c..15750606ac 100644 --- a/scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution.tmcf +++ b/scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution.tmcf @@ -10,7 +10,7 @@ variableMeasured: dcs:Mean_Concentration_AirPollutant_PM2.5 Node: E:PM25CountyPollution->E2 observationAbout: C:PM25CountyPollution->dcid typeOf: dcs:StatVarObservation -observationDate: C:PPM25CountyPollution->date +observationDate: C:PM25CountyPollution->date value: C:PM25CountyPollution->pm25_med_pred observationPeriod: "P24H" unit: MicrogramsPerCubicMeter diff --git a/scripts/us_cdc/environmental_health_toxicology/README.md b/scripts/us_cdc/environmental_health_toxicology/README.md index 0092777035..59c8280498 100644 --- a/scripts/us_cdc/environmental_health_toxicology/README.md +++ b/scripts/us_cdc/environmental_health_toxicology/README.md @@ -69,19 +69,13 @@ These data were collected as part of the [CDC National Environment Public Health [`small_Palmer_expected.csv`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/test_data/small_Palmer_expected.csv) #### tMCFs -[`OzoneCensusTractPollution.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCensusTractPollution.tmcf) - -[`OzoneCountyPollution.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCountyPollution.tmcf) - -[`PalmerDroughtSeverityIndex.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PalmerDroughtSeverityIndex.tmcf) - -[`PM25CensusTractPollution.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CensusTractPollution.tmcf) - -[`PM25CountyPollution.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution.tmcf) - -[`StandardizedPrecipitationEvapotranspirationIndex.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/StandardizedPrecipitationEvapotranspirationIndex.tmcf) - -[`StandardizedPrecipitationIndex.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/StandardizedPrecipitationIndex.tmcf) +* [`OzoneCountyPollution.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCountyPollution.tmcf) +* [`PM25CountyPollution.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CountyPollution.tmcf) +* **Sharded PM2.5 Census Tract tMCFs:** [`PM25CensusTractPollution_part1.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CensusTractPollution_part1.tmcf), [`part2`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CensusTractPollution_part2.tmcf), [`part3`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CensusTractPollution_part3.tmcf), [`part4`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PM25CensusTractPollution_part4.tmcf) +* **Sharded Ozone Census Tract tMCFs:** [`OzoneCensusTractPollution_part1.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCensusTractPollution_part1.tmcf), [`part2`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCensusTractPollution_part2.tmcf), [`part3`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCensusTractPollution_part3.tmcf), [`part4`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/OzoneCensusTractPollution_part4.tmcf) +* [`PalmerDroughtSeverityIndex.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/PalmerDroughtSeverityIndex.tmcf) +* [`StandardizedPrecipitationEvapotranspirationIndex.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/StandardizedPrecipitationEvapotranspirationIndex.tmcf) +* [`StandardizedPrecipitationIndex.tmcf`](https://github.com/datacommonsorg/data/blob/master/scripts/us_cdc/environmental_health_toxicology/StandardizedPrecipitationIndex.tmcf) ### Import Procedure diff --git a/scripts/us_cdc/environmental_health_toxicology/download_files.py b/scripts/us_cdc/environmental_health_toxicology/download_files.py index be61803de4..cff3646a1c 100644 --- a/scripts/us_cdc/environmental_health_toxicology/download_files.py +++ b/scripts/us_cdc/environmental_health_toxicology/download_files.py @@ -36,24 +36,27 @@ def download_files(importname, configs): @retry(tries=3, delay=2, backoff=2) def download_with_retry(url, input_file_name): logging.info(f"Downloading file from URL: {url}") - response = requests.get(url) - response.raise_for_status() - if response.status_code == 200: - if not response.content: - logging.fatal( - f"No data available for URL: {url}. Aborting download.") - return - filename = os.path.join(_INPUT_FILE_PATH, input_file_name) - with file_util.FileIO(filename, 'wb') as f: - f.write(response.content) - else: - logging.error( - f"Failed to download file from URL: {url}. Status code: {response.status_code}" - ) + filename = os.path.join(_INPUT_FILE_PATH, input_file_name) + with requests.get(url, stream=True, timeout=(30, 300)) as response: + response.raise_for_status() + with open(filename, 'wb') as f: + for chunk in response.iter_content(chunk_size=16 * 1024 * 1024): + if chunk: + f.write(chunk) + @retry(tries=3, delay=2, backoff=2) + def get_record_count_with_retry(count_url): + logging.info(f"Querying record count from URL: {count_url}") + resp = requests.get(count_url, timeout=60) + resp.raise_for_status() + return json.loads(resp.text)[0]['COLUMN_ALIAS_GUARD__count'] + + url_new = None + import_found = False try: for config in configs: if config["import_name"] == importname: + import_found = True files = config["files"] for file_info in files: url_new = file_info["url"] @@ -61,35 +64,30 @@ def download_with_retry(url, input_file_name): input_file_name = file_info["input_file_name"] logging.info(f"Input File Name {input_file_name}") - get_record_count = requests.get( - url_new.replace('.csv', record_count_query)) - if get_record_count.status_code == 200: - record_count = json.loads( - get_record_count.text - )[0]['COLUMN_ALIAS_GUARD__count'] - logging.info( - f"Numbers of records found for the URL {url_new} is {record_count}" - ) - url_new = f"{url_new}?$limit={record_count}&$offset=0" - download_with_retry(url_new, input_file_name) - logging.info( - "Successfully downloaded the source data...!!!!") - else: - logging.error( - f"Failed to download files, Status code: {get_record_count.status_code}" - ) + count_url = url_new.replace('.csv', record_count_query) + record_count = get_record_count_with_retry(count_url) + logging.info( + f"Numbers of records found for the URL {url_new} is {record_count}" + ) + url_new = f"{url_new}?$limit={record_count}&$offset=0" + download_with_retry(url_new, input_file_name) + logging.info( + "Successfully downloaded the source data...!!!!") + if not import_found: + raise ValueError( + f"Import name '{importname}' not found in configuration") except Exception as e: - logging.fatal(f"Error downloading URL {url_new} - {e}") + logging.fatal(f"Error downloading URL {url_new or 'unknown'} - {e}") + raise -def main(_): +def main(argv): """Main function to download the csv files.""" global _INPUT_FILE_PATH - _INPUT_FILE_PATH = os.path.join(_FLAGS.input_file_path) _INPUT_FILE_PATH = os.path.join(_MODULE_DIR, _FLAGS.input_file_path) Path(_INPUT_FILE_PATH).mkdir(parents=True, exist_ok=True) - importname = sys.argv[1] + importname = argv[1] logging.info(f'Loading config: {_FLAGS.config_file}') with file_util.FileIO(_FLAGS.config_file, 'r') as f: config = json.load(f) diff --git a/scripts/us_cdc/environmental_health_toxicology/manifest.json b/scripts/us_cdc/environmental_health_toxicology/manifest.json index 526e350993..265c11c316 100644 --- a/scripts/us_cdc/environmental_health_toxicology/manifest.json +++ b/scripts/us_cdc/environmental_health_toxicology/manifest.json @@ -12,12 +12,16 @@ "parse_air_quality.py CDC_PM25CensusTract" ], "source_files": [ - "input/*.gz" + "input_files/*" ], "resource_limits": { - "cpu": 8, - "memory": 64, - "disk": 100 + "cpu": 32, + "memory": 512, + "disk": 2000 + }, + "validation_config_file": "validation_config_census_tract.json", + "config_override": { + "invoke_differ_tool": false }, "import_inputs": [ { @@ -51,12 +55,16 @@ "parse_air_quality.py CDC_OzoneCensusTract" ], "source_files": [ - "input/*.gz" + "input_files/*" ], "resource_limits": { - "cpu": 8, - "memory": 64, - "disk": 100 + "cpu": 32, + "memory": 512, + "disk": 2000 + }, + "validation_config_file": "validation_config_census_tract.json", + "config_override": { + "invoke_differ_tool": false }, "import_inputs": [ { @@ -92,17 +100,30 @@ "source_files": [ "input_files/*" ], + "resource_limits": { + "cpu": 32, + "memory": 512, + "disk": 500 + }, + "validation_config_file": "validation_config_county.json", "import_inputs": [ { "template_mcf": "PM25CountyPollution.tmcf", - "cleaned_csv": "output/PM25county.csv" + "cleaned_csv": "output/PM25county_0.csv" + }, + { + "template_mcf": "PM25CountyPollution.tmcf", + "cleaned_csv": "output/PM25county_1.csv" + }, + { + "template_mcf": "PM25CountyPollution.tmcf", + "cleaned_csv": "output/PM25county_2.csv" + }, + { + "template_mcf": "PM25CountyPollution.tmcf", + "cleaned_csv": "output/PM25county_3.csv" } ], - "resource_limits": { - "cpu": 8, - "memory": 128, - "disk": 200 - }, "cron_schedule": "0 1 4 * *" }, { @@ -119,17 +140,18 @@ "source_files": [ "input_files/*" ], + "resource_limits": { + "cpu": 32, + "memory": 512, + "disk": 500 + }, + "validation_config_file": "validation_config_county.json", "import_inputs": [ { "template_mcf": "OzoneCountyPollution.tmcf", "cleaned_csv": "output/OzoneCounty.csv" } ], - "resource_limits": { - "cpu": 16, - "memory": 512, - "disk": 500 - }, "cron_schedule": "0 1 5 * *" } ] diff --git a/scripts/us_cdc/environmental_health_toxicology/parse_air_quality.py b/scripts/us_cdc/environmental_health_toxicology/parse_air_quality.py index 7b6711997c..b74836ffe9 100644 --- a/scripts/us_cdc/environmental_health_toxicology/parse_air_quality.py +++ b/scripts/us_cdc/environmental_health_toxicology/parse_air_quality.py @@ -29,10 +29,8 @@ 'config_file', 'gs://unresolved_mcf/cdc/environmental/import_configs.json', 'Config file path') flags.DEFINE_string('output_file_path', 'output', 'Output files path') -_MODULE_DIR = os.path.dirname(os.path.abspath(__file__)) _INPUT_FILE_PATH = None _OUTPUT_FILE_PATH = None -record_count_query = '?$query=select%20count(*)%20as%20COLUMN_ALIAS_GUARD__count' # Mapping of column names in file to StatVar names. STATVARS = { @@ -51,27 +49,6 @@ "O3_pop_pred": "PopulationWeighted_Concentration_AirPollutant_Ozone" } -# Mapping of month abbreviations to month numbers. -MONTH_MAP = { - "JAN": 1, - "FEB": 2, - "MAR": 3, - "APR": 4, - "MAY": 5, - "JUN": 6, - "JUL": 7, - "AUG": 8, - "SEP": 9, - "OCT": 10, - "NOV": 11, - "DEC": 12 -} - - -# this method is applicable only for "census tract PM25" -def add_prefix_zero(value, length): - return value.zfill(length) - def clean_air_quality_data(configs, importname, inputpath, outputpath): """ @@ -86,10 +63,11 @@ def clean_air_quality_data(configs, importname, inputpath, outputpath): a cleaned csv file """ try: - global output_file_name logging.info(f"import name from command line {importname}") + import_found = False for config in configs: if config["import_name"] == importname: + import_found = True files = config["files"] for file_info in files: output_file_name = file_info["output_file_name"] @@ -103,90 +81,216 @@ def clean_air_quality_data(configs, importname, inputpath, outputpath): logging.info(f"Cleaning {input_file_name} ....") logging.info(f"Cleaning {input_file_path} ....") try: - data = pd.read_csv(input_file_path) - data["date"] = pd.to_datetime(data["date"], - yearfirst=True) - data["date"] = pd.to_datetime(data["date"], - format="%Y-%m-%d") - - if "PM2.5" in input_file_name: - census_tract = "ds_pm" - elif "Ozone" in input_file_name: - census_tract = "ds_o3" - if "Census" in input_file_name: + if "County" in input_file_name and "PM" in input_file_name: + num_shards = 4 + base_name, ext = os.path.splitext( + output_file_name) + shard_paths = [] + for idx in range(num_shards): + shard_file_name = f"{base_name}_{idx}{ext}" + if not os.path.isabs(shard_file_name): + shard_output_path = os.path.join( + outputpath, shard_file_name) + else: + shard_output_path = shard_file_name + shard_paths.append(shard_output_path) + + with open(input_file_path, 'r') as f: + total_rows = sum(1 for _ in f) - 1 + if total_rows <= 0: + raise ValueError( + f"Input file {input_file_path} contains no data rows (total_rows={total_rows})." + ) + base_size = total_rows // num_shards + rem_size = total_rows % num_shards + shard_sizes = [ + base_size + 1 if i < rem_size else base_size + for i in range(num_shards) + ] + + chunk_size = 500_000 + shard_idx = 0 + shard_written = 0 + first_chunk = True + + for chunk in pd.read_csv(input_file_path, + chunksize=chunk_size): + chunk["date"] = pd.to_datetime( + chunk["date"], + format="%d%b%Y", + errors="raise").dt.strftime("%Y-%m-%d") + chunk["statefips"] = chunk[ + "statefips"].astype(str).str.zfill(2) + chunk["countyfips"] = chunk[ + "countyfips"].astype(str).str.zfill(3) + chunk["dcid"] = "geoId/" + chunk[ + "statefips"] + chunk["countyfips"] + + if first_chunk: + for p in shard_paths: + pd.DataFrame( + columns=chunk.columns).to_csv( + p, index=False) + first_chunk = False + + start_idx = 0 + while start_idx < len(chunk): + if shard_idx < num_shards - 1: + remaining_in_shard = shard_sizes[ + shard_idx] - shard_written + end_idx = min( + start_idx + remaining_in_shard, + len(chunk)) + else: + end_idx = len(chunk) + + sub_chunk = chunk.iloc[ + start_idx:end_idx] + + sub_chunk.to_csv(shard_paths[shard_idx], + mode='a', + header=False, + float_format='%.6f', + index=False) + shard_written += len(sub_chunk) + start_idx = end_idx + + if shard_idx < num_shards - 1 and shard_written >= shard_sizes[ + shard_idx]: + shard_idx += 1 + shard_written = 0 + + for p in shard_paths: + logging.info( + f"Finished cleaning file {os.path.basename(p)}!" + ) + elif "County" in input_file_name and "Ozone" in input_file_name: + chunk_size = 500_000 + first_chunk = True + for chunk in pd.read_csv(input_file_path, + chunksize=chunk_size): + chunk["date"] = pd.to_datetime( + chunk["date"], + format="%d%b%Y", + errors="raise").dt.strftime("%Y-%m-%d") + chunk["statefips"] = chunk[ + "statefips"].astype(str).str.zfill(2) + chunk["countyfips"] = chunk[ + "countyfips"].astype(str).str.zfill(3) + chunk["dcid"] = "geoId/" + chunk[ + "statefips"] + chunk["countyfips"] + if first_chunk: + chunk.to_csv(output_file_path, + float_format='%.6f', + index=False) + first_chunk = False + else: + chunk.to_csv(output_file_path, + mode='a', + header=False, + float_format='%.6f', + index=False) + if first_chunk: + raise ValueError( + f"Input file {input_file_path} contains no data rows." + ) + logging.info( + f"Finished cleaning file {output_file_name}!" + ) + else: + chunk_size = 500_000 + first_chunk = True if "PM2.5" in input_file_name: - data = pd.melt( - data, - id_vars=[ - 'year', 'date', 'statefips', - 'countyfips', 'ctfips', 'latitude', - 'longitude' - ], - value_vars=[ - str(census_tract + '_pred'), - str(census_tract + '_stdd') - ], - var_name='StatisticalVariable', - value_name='Value') + census_tract = "ds_pm" elif "Ozone" in input_file_name: - data = pd.melt( - data, - id_vars=[ - 'year', 'date', 'statefips', - 'countyfips', 'ctfips', 'latitude', - 'longitude', census_tract + '_stdd' - ], - value_vars=[ - str(census_tract + '_pred') - ], - var_name='StatisticalVariable', - value_name='Value') - data.rename( - columns={census_tract + '_stdd': 'Error'}, - inplace=True) - max_length = data['ctfips'].astype( - str).str.len().max() - data['ctfips'] = data['ctfips'].astype( - str).apply(lambda x: add_prefix_zero( - x, max_length)) - data["dcid"] = "geoId/" + data["ctfips"].astype( - str) - data['StatisticalVariable'] = data[ - 'StatisticalVariable'].map(STATVARS) - elif "County" in input_file_name and "PM" in input_file_name: - data["statefips"] = data["statefips"].astype( - str).str.zfill(2) - data["countyfips"] = data["countyfips"].astype( - str).str.zfill(3) - data["dcid"] = "geoId/" + data[ - "statefips"] + data["countyfips"] - elif "County" in input_file_name and "Ozone" in input_file_name: - data["statefips"] = data["statefips"].astype( - str).str.zfill(2) - data["countyfips"] = data["countyfips"].astype( - str).str.zfill(3) - data["dcid"] = "geoId/" + data[ - "statefips"] + data["countyfips"] - data.to_csv(output_file_path, - float_format='%.6f', - index=False) - logging.info( - f"Finished cleaning file {output_file_name}!") - except: - logging.info("Not reading input file...!") + census_tract = "ds_o3" + else: + census_tract = None + + for chunk in pd.read_csv(input_file_path, + chunksize=chunk_size): + chunk["date"] = pd.to_datetime( + chunk["date"], + format="%d%b%Y", + errors="raise").dt.strftime("%Y-%m-%d") + + if "Census" in input_file_name: + if "PM2.5" in input_file_name: + chunk = pd.melt( + chunk, + id_vars=[ + 'year', 'date', 'statefips', + 'countyfips', 'ctfips', + 'latitude', 'longitude' + ], + value_vars=[ + str(census_tract + '_pred'), + str(census_tract + '_stdd') + ], + var_name='StatisticalVariable', + value_name='Value') + elif "Ozone" in input_file_name: + chunk = pd.melt( + chunk, + id_vars=[ + 'year', 'date', 'statefips', + 'countyfips', 'ctfips', + 'latitude', 'longitude', + census_tract + '_stdd' + ], + value_vars=[ + str(census_tract + '_pred') + ], + var_name='StatisticalVariable', + value_name='Value') + chunk.rename(columns={ + census_tract + '_stdd': 'Error' + }, + inplace=True) + chunk['ctfips'] = chunk[ + 'ctfips'].astype(str).str.zfill(11) + chunk["dcid"] = "geoId/" + chunk[ + "ctfips"].astype(str) + chunk['StatisticalVariable'] = chunk[ + 'StatisticalVariable'].map(STATVARS) + if first_chunk: + chunk.to_csv(output_file_path, + float_format='%.6f', + index=False) + first_chunk = False + else: + chunk.to_csv(output_file_path, + mode='a', + header=False, + float_format='%.6f', + index=False) + if first_chunk: + raise ValueError( + f"Input file {input_file_path} contains no data rows." + ) + logging.info( + f"Finished cleaning file {output_file_name}!" + ) + except Exception as e: + logging.error( + f"Error cleaning {input_file_name}: {e}") + raise + if not import_found: + raise ValueError( + f"Import name '{importname}' not found in configuration") except Exception as e: logging.fatal(f"Error while processing the data: {e}") + raise -def main(_): +def main(argv): """Main function to generate the cleaned csv file.""" global _INPUT_FILE_PATH, _OUTPUT_FILE_PATH - _INPUT_FILE_PATH = _FLAGS.input_file_path _INPUT_FILE_PATH = os.path.join(_MODULE_DIR, _FLAGS.input_file_path) Path(_INPUT_FILE_PATH).mkdir(parents=True, exist_ok=True) _OUTPUT_FILE_PATH = os.path.join(_MODULE_DIR, _FLAGS.output_file_path) Path(_OUTPUT_FILE_PATH).mkdir(parents=True, exist_ok=True) - importname = sys.argv[1] + importname = argv[1] logging.info(f'Loading config: {_FLAGS.config_file}') with file_util.FileIO(_FLAGS.config_file, 'r') as f: config = json.load(f) diff --git a/scripts/us_cdc/environmental_health_toxicology/parse_air_quality_test.py b/scripts/us_cdc/environmental_health_toxicology/parse_air_quality_test.py index a5653348ad..c0b17923f6 100644 --- a/scripts/us_cdc/environmental_health_toxicology/parse_air_quality_test.py +++ b/scripts/us_cdc/environmental_health_toxicology/parse_air_quality_test.py @@ -8,14 +8,19 @@ # # 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. -import unittest import os -from .parse_air_quality import clean_air_quality_data +import shutil +import sys +import unittest _MODULE_DIR = os.path.dirname(__file__) +sys.path.insert(0, _MODULE_DIR) + +try: + from .parse_air_quality import clean_air_quality_data +except ImportError: + from parse_air_quality import clean_air_quality_data + TEST_DATA_DIR = os.path.join(_MODULE_DIR, 'test_data') INPUT_DIR = 'input_files' OUTPUT_DIR = 'actual_output_files' @@ -63,15 +68,13 @@ os.path.join(TEST_DATA_DIR, "CDC_PM25County", INPUT_DIR), "output_dir": os.path.join(TEST_DATA_DIR, "CDC_PM25County", OUTPUT_DIR), - "expected_file": + "expected_files": [ os.path.join(TEST_DATA_DIR, "CDC_PM25County", OUTPUT_FILES, - "PM25county.csv"), + f"PM25county_{i}.csv") for i in range(4) + ], "files": [{ - "input_file_name": - "PM2.5County_input_0.csv", - "output_file_name": - os.path.join(TEST_DATA_DIR, "CDC_PM25County", OUTPUT_DIR, - "PM25county.csv") + "input_file_name": "PM2.5County_input_0.csv", + "output_file_name": "PM25county.csv" }] }, { "import_name": @@ -103,19 +106,34 @@ def test_clean_air_quality_data(self): Tests the clean_air_quality_data function for all the 4 imports. """ for data in TEST_DATA: - for data1 in data["files"]: - output_dir = data["output_dir"] - os.makedirs(output_dir, exist_ok=True) - clean_air_quality_data(TEST_DATA, data["import_name"], - data["input_dir"], output_dir) - with open(data1["output_file_name"], - encoding="utf-8") as actual_csv_file: - actual_csv_data = actual_csv_file.read().strip() - with open(data["expected_file"], - encoding="utf-8") as expected_csv_file: - expected_csv_data = expected_csv_file.read().strip() + output_dir = data["output_dir"] + os.makedirs(output_dir, exist_ok=True) + clean_air_quality_data(TEST_DATA, data["import_name"], + data["input_dir"], output_dir) + if "expected_files" in data: + for expected_file in data["expected_files"]: + file_name = os.path.basename(expected_file) + actual_file = os.path.join(output_dir, file_name) + with open(actual_file, encoding="utf-8") as actual_csv_file: + actual_csv_data = actual_csv_file.read().strip() + with open(expected_file, + encoding="utf-8") as expected_csv_file: + expected_csv_data = expected_csv_file.read().strip() + self.assertEqual(expected_csv_data, actual_csv_data) + else: + for data1 in data["files"]: + with open(data1["output_file_name"], + encoding="utf-8") as actual_csv_file: + actual_csv_data = actual_csv_file.read().strip() + with open(data["expected_file"], + encoding="utf-8") as expected_csv_file: + expected_csv_data = expected_csv_file.read().strip() - self.assertEqual(expected_csv_data, actual_csv_data) + self.assertEqual(expected_csv_data, actual_csv_data) + + def tearDown(self): + for data in TEST_DATA: + shutil.rmtree(data["output_dir"], ignore_errors=True) if __name__ == '__main__': diff --git a/scripts/us_cdc/environmental_health_toxicology/parse_precipitation_index_test.py b/scripts/us_cdc/environmental_health_toxicology/parse_precipitation_index_test.py index b69ac7c8b1..8f5e1f4183 100644 --- a/scripts/us_cdc/environmental_health_toxicology/parse_precipitation_index_test.py +++ b/scripts/us_cdc/environmental_health_toxicology/parse_precipitation_index_test.py @@ -20,11 +20,17 @@ python3 parse_precipitation_index_test.py input_file output_file ''' -import unittest import os -from .parse_precipitation_index import clean_precipitation_data +import sys +import unittest module_dir_ = os.path.dirname(__file__) +sys.path.insert(0, module_dir_) + +try: + from .parse_precipitation_index import clean_precipitation_data +except ImportError: + from parse_precipitation_index import clean_precipitation_data class TestParsePrecipitationData(unittest.TestCase): diff --git a/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_OzoneCensusTract/expected_output_files/Census_Tract_Level_Ozone_Concentrations_0.csv b/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_OzoneCensusTract/expected_output_files/Census_Tract_Level_Ozone_Concentrations_0.csv index 9debb88cb2..99ac99313b 100644 --- a/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_OzoneCensusTract/expected_output_files/Census_Tract_Level_Ozone_Concentrations_0.csv +++ b/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_OzoneCensusTract/expected_output_files/Census_Tract_Level_Ozone_Concentrations_0.csv @@ -1,6 +1,6 @@ year,date,statefips,countyfips,ctfips,latitude,longitude,Error,StatisticalVariable,Value,dcid -2001,2001-01-01,1,1,1001020100,32.477180,-86.490010,5.931458,Mean_Concentration_AirPollutant_Ozone,31.659713,geoId/1001020100 -2001,2001-01-01,1,1,1001020200,32.474250,-86.473390,5.703084,Mean_Concentration_AirPollutant_Ozone,31.939058,geoId/1001020200 -2001,2001-01-01,1,1,1001020300,32.475440,-86.460200,5.779281,Mean_Concentration_AirPollutant_Ozone,31.859155,geoId/1001020300 -2001,2001-01-01,1,1,1001020400,32.472040,-86.443700,5.784730,Mean_Concentration_AirPollutant_Ozone,31.681802,geoId/1001020400 -2001,2001-01-01,1,1,1001020500,32.458920,-86.422710,5.797620,Mean_Concentration_AirPollutant_Ozone,31.752820,geoId/1001020500 +2001,2001-01-01,1,1,01001020100,32.477180,-86.490010,5.931458,Mean_Concentration_AirPollutant_Ozone,31.659713,geoId/01001020100 +2001,2001-01-01,1,1,01001020200,32.474250,-86.473390,5.703084,Mean_Concentration_AirPollutant_Ozone,31.939058,geoId/01001020200 +2001,2001-01-01,1,1,01001020300,32.475440,-86.460200,5.779281,Mean_Concentration_AirPollutant_Ozone,31.859155,geoId/01001020300 +2001,2001-01-01,1,1,01001020400,32.472040,-86.443700,5.784730,Mean_Concentration_AirPollutant_Ozone,31.681802,geoId/01001020400 +2001,2001-01-01,1,1,01001020500,32.458920,-86.422710,5.797620,Mean_Concentration_AirPollutant_Ozone,31.752820,geoId/01001020500 diff --git a/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_PM25County/expected_output_files/PM25county_0.csv b/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_PM25County/expected_output_files/PM25county_0.csv new file mode 100644 index 0000000000..fb8733bcb7 --- /dev/null +++ b/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_PM25County/expected_output_files/PM25county_0.csv @@ -0,0 +1,3 @@ +year,date,statefips,countyfips,pm25_max_pred,pm25_med_pred,pm25_mean_pred,pm25_pop_pred,dcid +2016,2016-01-01,01,001,12.035500,11.173300,11.307642,11.313675,geoId/01001 +2016,2016-01-01,01,003,9.117200,8.486000,8.497219,8.483280,geoId/01003 diff --git a/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_PM25County/expected_output_files/PM25county_1.csv b/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_PM25County/expected_output_files/PM25county_1.csv new file mode 100644 index 0000000000..fd76a1b72f --- /dev/null +++ b/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_PM25County/expected_output_files/PM25county_1.csv @@ -0,0 +1,2 @@ +year,date,statefips,countyfips,pm25_max_pred,pm25_med_pred,pm25_mean_pred,pm25_pop_pred,dcid +2016,2016-01-01,01,005,8.898900,8.331700,8.422811,8.435805,geoId/01005 diff --git a/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_PM25County/expected_output_files/PM25county_2.csv b/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_PM25County/expected_output_files/PM25county_2.csv new file mode 100644 index 0000000000..6e848f5a92 --- /dev/null +++ b/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_PM25County/expected_output_files/PM25county_2.csv @@ -0,0 +1,2 @@ +year,date,statefips,countyfips,pm25_max_pred,pm25_med_pred,pm25_mean_pred,pm25_pop_pred,dcid +2016,2016-01-01,01,007,11.730300,10.851400,10.992450,10.864566,geoId/01007 diff --git a/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_PM25County/expected_output_files/PM25county_3.csv b/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_PM25County/expected_output_files/PM25county_3.csv new file mode 100644 index 0000000000..bdae75cb64 --- /dev/null +++ b/scripts/us_cdc/environmental_health_toxicology/test_data/CDC_PM25County/expected_output_files/PM25county_3.csv @@ -0,0 +1,2 @@ +year,date,statefips,countyfips,pm25_max_pred,pm25_med_pred,pm25_mean_pred,pm25_pop_pred,dcid +2016,2016-01-01,01,009,13.224700,12.895900,12.703789,12.725276,geoId/01009 diff --git a/scripts/us_cdc/environmental_health_toxicology/validation_config_census_tract.json b/scripts/us_cdc/environmental_health_toxicology/validation_config_census_tract.json new file mode 100644 index 0000000000..91f4418647 --- /dev/null +++ b/scripts/us_cdc/environmental_health_toxicology/validation_config_census_tract.json @@ -0,0 +1,15 @@ +{ + "schema_version": "1.0", + "rules": [ + { + "rule_id": "check_max_date_consistent", + "description": "Checks if the MaxDate is the same for all StatVars.", + "validator": "MAX_DATE_CONSISTENT" + }, + { + "rule_id": "check_deleted_records_percent", + "description": "Disable deleted records check because differ is decoupled for high-scale Census Tracts.", + "enabled": false + } + ] +} diff --git a/scripts/us_cdc/environmental_health_toxicology/validation_config_county.json b/scripts/us_cdc/environmental_health_toxicology/validation_config_county.json new file mode 100644 index 0000000000..bac5d68f65 --- /dev/null +++ b/scripts/us_cdc/environmental_health_toxicology/validation_config_county.json @@ -0,0 +1,18 @@ +{ + "schema_version": "1.0", + "rules": [ + { + "rule_id": "check_max_date_consistent", + "description": "Checks if the MaxDate is the same for all StatVars.", + "validator": "MAX_DATE_CONSISTENT" + }, + { + "rule_id": "check_deleted_records_percent", + "description": "Checks that the percentage of deleted points is within the threshold.", + "validator": "DELETED_RECORDS_PERCENT", + "params": { + "threshold": 0.05 + } + } + ] +}