diff --git a/statvar_imports/ipeds/student_to_faculty_ratio/README.md b/statvar_imports/ipeds/student_to_faculty_ratio/README.md index 60698a1e90..d84e616695 100644 --- a/statvar_imports/ipeds/student_to_faculty_ratio/README.md +++ b/statvar_imports/ipeds/student_to_faculty_ratio/README.md @@ -4,7 +4,7 @@ This project processes and imports the Student to Faculty Ratio data from the In # Import Name: IPEDS_StudentToFacultyRatio -Source URL: ``https://nces.ed.gov/ipeds/datacenter/DataFiles.aspx?gotoReportId=7&fromIpeds=true&sid=859539d0-db09-4d03-a36a-eabf93d07355&rtid=7` +Source URL: https://nces.ed.gov/ipeds/datacenter/DataFiles.aspx?gotoReportId=7&fromIpeds=true&sid=859539d0-db09-4d03-a36a-eabf93d07355&rtid=7 Provenance Description: Integrated Postsecondary Education Data System (IPEDS) is the official online home for the primary federal source of data on U.S. colleges, universities, and technical/vocational institutions. This import focuses on Student to Faculty Ratio data. @@ -23,11 +23,11 @@ The import process is divided into two main stages: downloading the raw data, pr Transformation pipeline: -download.py downloads the yearly data releases, unzips them, filters for relevant files, renames them, and saves them as CSV files in the input_files/ directory. +download.py downloads the yearly data releases, unzips them, filters for relevant files, and saves them as CSV files in the input_files/ directory. -preprocess.py is executed for additional cleaning/transformation. +preprocess.py is executed for additional cleaning/transformation (renaming to standard filenames, adding 'Year' column, and tagging provisional estimates). -After the download is complete, the stat_var_processor.py tool is run on the cleaned CSV files in the input_files directory using the shell script run.sh. +After the download and preprocessing are complete, the stat_var_processor.py tool is run on the cleaned CSV files in the input_files directory using the shell script run.sh. The processor uses `metadata and pv_map` files (not explicitly named for IPEDS here) to generate the final .csv and .tmcf files, placing them in the processed_output/ directory. @@ -46,7 +46,7 @@ This import is designed to be autorefreshed via a Cloud Scheduler job. # Scripts Executed: `download.py, preprocess.py, run.sh` -Schedule: 0 0 15 7 * (Runs at 00:00 on day 15 of July, i.e., annually on July 15th). +Schedule: 0 0 1,15 * * (Runs at 00:00 on the 1st and 15th of every month). ## Steps: @@ -58,22 +58,30 @@ The shell script run.sh then runs the `stat_var_processor.py` tool to process th The final, validated output files are uploaded to a GCS bucket for ingestion into the Data Commons Knowledge Graph. -# pre Script Execution Details +## Script Execution Details To run the import manually, follow these steps in order. -**Step 1:** Download and Preprocess Raw Data -This script downloads all available data files, unzips them, filters for relevant files, renames them, and adds 'year'. +**Step 1:** Download Raw Data +This script downloads all available data files, unzips them, and filters for relevant CSV files. # Usage: ```Bash - python3 download.py +``` + +The raw source files will be located in `input_files/`. + +**Step 2:** Preprocess Raw Data +This script standardizes filenames, injects the 'Year' column, and tags provisional estimates. + +# Usage: -The processed source files will be located in input_files/. +```Bash +python3 preprocess.py ``` -**Step 2:** +**Step 3:** # Process the Data This script processes all cleaned input files to generate the final CSV and TMCF files. @@ -83,26 +91,23 @@ The shell script run.sh runs the stat_var_processor.py tool on all the files in # Usage: ```Bash - -sh run.sh +bash run.sh ``` A generic command for the processor looks like: ```Bash - -python3 ../../../../tools/statvar_importer/stat_var_processor.py --input_data="input/student_faculty_ratio_data_.csv" --pv_map="student_faculty_ratio_pvmap.csv" --config_file="student_faculty_ratio_metadata.csv" --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path="output/student_to_faculty_ratio_" +python3 ../../../tools/statvar_importer/stat_var_processor.py --input_data="input_files/student_faculty_ratio_data_.csv" --pv_map="student_faculty_ratio_pvmap.csv" --config_file="student_faculty_ratio_metadata.csv" --existing_statvar_mcf=gs://unresolved_mcf/scripts/statvar/stat_vars.mcf --output_path="processed_output/student_faculty_ratio_" ``` -**Step 3:** +**Step 4:** # Validate the Output Files This command validates the generated files for formatting and semantic consistency before ingestion. # Usage: ```Bash - -java -jar /path/to/datacommons-import-tool.jar lint -d 'output/' +java -jar /path/to/datacommons-import-tool.jar lint -d 'processed_output/' ``` -This step ensures that the generated artifacts are ready for ingestion into Data Commons. \ No newline at end of file +This step ensures that the generated artifacts are ready for ingestion into Data Commons. diff --git a/statvar_imports/ipeds/student_to_faculty_ratio/download.py b/statvar_imports/ipeds/student_to_faculty_ratio/download.py index 57a4117d05..4202038c24 100644 --- a/statvar_imports/ipeds/student_to_faculty_ratio/download.py +++ b/statvar_imports/ipeds/student_to_faculty_ratio/download.py @@ -17,152 +17,223 @@ import re import zipfile import time -from datetime import date -from absl import logging +import requests +from datetime import date +from urllib.parse import urlparse +from absl import app, logging # --- Configuration --- START_YEAR = 2009 # Set END_YEAR dynamically to the current calendar year END_YEAR = date.today().year -BASE_URL = "https://nces.ed.gov/ipeds/datacenter/data/EF{}D.zip" -DOWNLOAD_DIR = "input_files" -# Pattern to match files ending in '_rv' followed by a file extension -# The pattern should match '_rv.txt', '_rv.csv', etc. -RV_PATTERN = re.compile(r'_rv\.[a-z0-9]+$', re.IGNORECASE) -# --------------------- + +BASE_URL_RV = "https://nces.ed.gov/ipeds/data-generator?year={year}&tableName=EF{year}D&HasRV=1&type=csv" +BASE_URL_LEGACY = "https://nces.ed.gov/ipeds/datacenter/data/EF{year}D.zip" +BASE_URL_COMPLETE = "https://nces.ed.gov/ipeds/complete-data-files/EF{year}D.zip" +BASE_URL_PROV = "https://nces.ed.gov/ipeds/data-generator?year={year}&tableName=EF{year}D&HasRV=0&type=csv" + +#the order of the templates is important. It is the order in which the script will try to download the data. +BASE_URL_TEMPLATES = [ + BASE_URL_COMPLETE, + BASE_URL_LEGACY, + BASE_URL_RV, + BASE_URL_PROV, +] # --- Path Adjustment for Utility Import --- _SCRIPT_PATH = os.path.dirname(os.path.abspath(__file__)) -# Correct path to the directory containing download_util_script.py: sys.path.append(os.path.join(_SCRIPT_PATH, '../../../util/')) try: - # IMPORT ONLY THE FUNCTION THAT EXISTS in the utility script from download_util_script import download_file except ImportError as e: - # Use logging.fatal for critical import errors and exit logging.fatal("Could not import 'download_file'. Please ensure the utility script is accessible. Original error: %s", e) - raise RuntimeError(f"FATAL: Missing utility script dependency: {e}") + +DOWNLOAD_DIR = os.path.join(_SCRIPT_PATH, "input_files") +# Pattern to match files ending in '_rv' followed by a file extension +RV_PATTERN = re.compile(r'_rv\.[a-z0-9]+$', re.IGNORECASE) +# Pattern to match provisional files (e.g. ef2023d.csv, ef2024d.csv) +PROVISIONAL_PATTERN = re.compile(r'^ef\d{4}d\.[a-z0-9]+$', re.IGNORECASE) +REQUIRED_COLUMNS = {"UNITID", "STUFACR"} +# --------------------- -def process_and_filter_zip(zip_path: str, output_dir: str, filter_pattern: re.Pattern) -> bool: +def is_valid_csv(file_path: str) -> bool: """ - Handles the custom unzipping, RV-pattern filtering, and cleanup. - If extraction fails unexpectedly, it is treated as a fatal error. + Validates that file_path is a valid CSV containing required IPEDS headers (UNITID, STUFACR). """ - zip_filename = os.path.basename(zip_path) - logging.info(" Unzipping and filtering contents (keeping only files matching pattern: %s)...", filter_pattern.pattern) + if not os.path.exists(file_path) or os.path.getsize(file_path) == 0: + return False + try: + with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: + first_line = f.readline() + if not first_line or " bool: + """ + Unzips and filters contents of zip_path. + Extracts files matching RV_PATTERN if present; otherwise matches PROVISIONAL_PATTERN. + Returns True if matching file(s) were found and extracted, False otherwise. + """ + zip_filename = os.path.basename(zip_path) try: - # 1. Unzip and filter with zipfile.ZipFile(zip_path, 'r') as zip_ref: all_files = zip_ref.namelist() - files_to_extract = [] - - for file_name in all_files: - base_name = os.path.basename(file_name) - # Check if the file is at the root or within a single folder. - if filter_pattern.search(base_name): - files_to_extract.append(file_name) + + # Prefer revised files first + files_to_extract = [ + f for f in all_files if RV_PATTERN.search(os.path.basename(f)) + ] + # Fall back to provisional files if no revised file is in the zip + if not files_to_extract: + files_to_extract = [ + f for f in all_files if PROVISIONAL_PATTERN.search(os.path.basename(f)) + ] if not files_to_extract: - # This is a warning, not a failure, as the process was clean. - logging.fatal(" Warning: No files matching the pattern found in %s. Skipping extraction.", zip_filename) - extraction_successful = True + return False - # Extract the filtered files for file_name in files_to_extract: zip_ref.extract(file_name, output_dir) logging.info(" Extracted: %s", file_name) - if files_to_extract: - logging.info(" Extraction successful.") - extraction_successful = True + return True except zipfile.BadZipFile: - # FATAL: Corrupted zip file means data is unavailable, and we must stop processing this file. - logging.fatal(" FATAL ERROR: %s is a corrupted or empty zip file. Cannot proceed.", zip_filename) - raise RuntimeError(f"Corrupted or empty zip file encountered: {zip_filename}") + logging.warning(" %s is a corrupted or empty zip file. Trying next URL...", zip_filename) + return False except Exception as e: - # FATAL: Any unexpected extraction error means partial data, which must be avoided. - logging.fatal(" FATAL ERROR: An unexpected error occurred during unzipping/extraction of %s: %s", zip_filename, e) - # Raise an error to stop the script from proceeding with potentially partial unzipped files - raise RuntimeError(f"Extraction failed for {zip_filename}: {e}") + logging.warning(" An unexpected error occurred during unzipping/extraction of %s: %s", zip_filename, e) + return False finally: - # 2. Clean up the downloaded zip file manually, regardless of success or failure if os.path.exists(zip_path): try: os.remove(zip_path) - logging.info(" Removed zip file: %s", zip_path) - except OSError as e: - # Use info for non-critical file removal errors - logging.info(" Warning: Failed to remove zip file %s: %s", zip_path, e) - - return extraction_successful + except OSError: + pass -def main(): +def download_for_year(year: int) -> bool: """ - Downloads IPEDS zip files using the utility, then handles custom unzipping/filtering locally. - Download failure is logged as a warning, allowing the script to proceed to the next year. - Extraction failure is logged as FATAL, which will stop the script for that year's file. + Downloads data for a given year by trying candidate URLs in order of preference. + Extracts and saves the first successful dataset found for that year. """ + logging.info("\nProcessing year %d...", year) - # 1. Create the target directory if it doesn't exist - if not os.path.exists(DOWNLOAD_DIR): - try: - os.makedirs(DOWNLOAD_DIR) - logging.info("Created directory: %s", DOWNLOAD_DIR) - except OSError as e: - # Use logging.fatal for critical directory creation errors - logging.fatal("FATAL ERROR: Could not create directory %s: %s", DOWNLOAD_DIR, e) - raise RuntimeError(f"FATAL: Directory creation failed for {DOWNLOAD_DIR}: {e}") - - # 2. Iterate through the required year range - for year in range(START_YEAR, END_YEAR + 1): - url = BASE_URL.format(year) - zip_filename = f"EF{year}D.zip" - download_path = os.path.join(DOWNLOAD_DIR, zip_filename) - - logging.info("\nProcessing year %d...", year) + for url_template in BASE_URL_TEMPLATES: + if "{year}" in url_template: + url = url_template.format(year=year) + else: + url = url_template.format(year) try: - # 3. Call the utility function to DOWNLOAD ONLY (unzip=False) download_success = download_file( url=url, output_folder=DOWNLOAD_DIR, - unzip=False, # <-- CRITICAL: Do not let the utility unzip the file - tries=3, - delay=5, + unzip=False, + tries=4, + delay=2, backoff=2 ) if download_success: - # 4. Handle custom processing (unzip, filter, and cleanup) locally - # If process_and_filter_zip encounters a fatal error, it will raise an exception - process_and_filter_zip(download_path, DOWNLOAD_DIR, RV_PATTERN) + # Determine the filename inferred by download_file + parsed_url = urlparse(url) + file_name = os.path.basename(parsed_url.path) + if not file_name: + file_name = "downloaded_file" + elif '.' not in file_name: + file_name = file_name + '.xlsx' + + download_path = os.path.join(DOWNLOAD_DIR, file_name) + default_zip_path = os.path.join(DOWNLOAD_DIR, f"EF{year}D.zip") + + actual_download_path = download_path if os.path.exists(download_path) else default_zip_path + + if os.path.exists(actual_download_path): + if zipfile.is_zipfile(actual_download_path): + if process_and_filter_zip(actual_download_path, DOWNLOAD_DIR): + logging.info(" Successfully fetched and extracted dataset for year %d.", year) + return True + else: + # Direct CSV download (e.g. from data-generator) + if is_valid_csv(actual_download_path): + target_name = f"ef{year}d_rv.csv" if "HasRV=1" in url else f"ef{year}d.csv" + target_path = os.path.join(DOWNLOAD_DIR, target_name) + if os.path.exists(target_path): + os.remove(target_path) + os.rename(actual_download_path, target_path) + logging.info(" Successfully fetched direct CSV dataset for year %d.", year) + return True + else: + logging.warning(" Downloaded file from %s is not a valid CSV dataset with expected columns.", url) + if os.path.exists(actual_download_path): + try: + os.remove(actual_download_path) + except OSError: + pass + except requests.exceptions.RequestException as e: + logging.warning(" Network error for candidate URL %s: %s", url, e) + except Exception as e: + logging.warning(" Candidate URL %s failed: %s", url, e) - else: - # Not available/download failed: Log as a warning and skip, as requested - logging.info("Warning: Download failed for year %d. Skipping processing for this year.", year) + logging.warning("Warning: No dataset found for year %d across candidate URLs.", year) + return False - except Exception as e: - # Catch errors that process_and_filter_zip explicitly raises (BadZipFile, Extraction Error) - # or any other unexpected error during the loop. The error is already logged as FATAL. - logging.info("Execution failed for year %d, continuing to next year (if possible). Error: %s", year, e) - # The script will now proceed to the next iteration unless the error is outside the loop - # Optional: Add a pause between years to respect NCES server requests - time.sleep(5) +def main(_): + """ + Iterates through year range downloading datasets (preferring revised, falling back to provisional). + Tracks successful downloads across all iterations and verifies that at least one dataset was fetched. + """ + + # 1. Create target directory if it doesn't exist + if not os.path.exists(DOWNLOAD_DIR): + try: + os.makedirs(DOWNLOAD_DIR) + logging.info("Created directory: %s", DOWNLOAD_DIR) + except OSError as e: + logging.fatal("FATAL ERROR: Could not create directory %s: %s", DOWNLOAD_DIR, e) + + # 2. Iterate through required year range and track results + successful_years = [] + failed_years = [] + + for year in range(START_YEAR, END_YEAR + 1): + if download_for_year(year): + successful_years.append(year) + else: + failed_years.append(year) + time.sleep(1) + + # 3. Fail fast if zero datasets were fetched across all years + if not successful_years: + logging.fatal( + "FATAL ERROR: Zero datasets were successfully downloaded across years %d-%d.", + START_YEAR, END_YEAR + ) + + if failed_years: + logging.info( + "No datasets downloaded for year(s): %s", + failed_years + ) + + logging.info( + "\nDownload summary: Successfully fetched %d dataset(s) for years: %s", + len(successful_years), + successful_years + ) + logging.info("\nScript finished. Filtered files extracted to the '%s' folder.", DOWNLOAD_DIR) if __name__ == "__main__": - logging.set_verbosity(logging.INFO) - try: - main() - logging.info("\nScript finished. Filtered files extracted to the '%s' folder.", DOWNLOAD_DIR) - except Exception as e: - # Catch errors that prevent main from starting or critical errors like directory creation - logging.fatal("\nFATAL ERROR in main execution: %s", e) \ No newline at end of file + app.run(main) \ No newline at end of file diff --git a/statvar_imports/ipeds/student_to_faculty_ratio/golden_data/golden_summary_report.csv b/statvar_imports/ipeds/student_to_faculty_ratio/golden_data/golden_summary_report.csv new file mode 100644 index 0000000000..51f05797e9 --- /dev/null +++ b/statvar_imports/ipeds/student_to_faculty_ratio/golden_data/golden_summary_report.csv @@ -0,0 +1,2 @@ +"Units","StatVar","MinDate","ScalingFactors","NumPlaces","observationPeriods","MeasurementMethods" +"[]","Percent_Student_AsAFractionOf_Count_Teacher","2009","[]","8714","[]","[, NCES_ProvisionalEstimate]" diff --git a/statvar_imports/ipeds/student_to_faculty_ratio/manifest.json b/statvar_imports/ipeds/student_to_faculty_ratio/manifest.json index 072dc0ca2d..7cb8961b75 100644 --- a/statvar_imports/ipeds/student_to_faculty_ratio/manifest.json +++ b/statvar_imports/ipeds/student_to_faculty_ratio/manifest.json @@ -8,18 +8,23 @@ "provenance_url": "https://nces.ed.gov/ipeds/datacenter/DataFiles.aspx?gotoReportId=7&fromIpeds=true&sid=859539d0-db09-4d03-a36a-eabf93d07355&rtid=7", "provenance_description": "Integrated Postsecondary Education Data System (IPEDS) is the official online home for the primary federal source of data on U.S. colleges, universities, and technical/vocational institutions.. This import focuses on Student to Faculty Ratio data.", "scripts": [ - "download.py","preprocess.py","run.sh" - ], + "download.py", + "preprocess.py", + "run.sh" + ], "import_inputs": [ { "template_mcf": "processed_output/student_faculty_ratio.tmcf", "cleaned_csv": "processed_output/*.csv" } - ], + ], "source_files": [ - "input_files/*.csv" + "input_files/*.csv", + "counters/*.csv", + "golden_data/*.csv" ], - "cron_schedule": "0 0 15 7 *" + "cron_schedule": "0 0 1,15 * *", + "validation_config_file": "validation_config.json" } ] -} \ No newline at end of file +} diff --git a/statvar_imports/ipeds/student_to_faculty_ratio/preprocess.py b/statvar_imports/ipeds/student_to_faculty_ratio/preprocess.py index f295ece380..d257c6c153 100644 --- a/statvar_imports/ipeds/student_to_faculty_ratio/preprocess.py +++ b/statvar_imports/ipeds/student_to_faculty_ratio/preprocess.py @@ -15,65 +15,108 @@ import os import pandas as pd import re -from absl import logging +from absl import app, logging -# Set verbosity level to 2 -logging.set_verbosity(2) +# --- Path Configuration --- +_SCRIPT_PATH = os.path.dirname(os.path.abspath(__file__)) +input_folder = os.path.join(_SCRIPT_PATH, "input_files") -# Folder containing input files -input_folder = "input_files" - -# Regex pattern to extract year from filenames like ef2010d_rv.csv -pattern = re.compile(r"ef(\d{4})d_rv", re.IGNORECASE) +# Regex pattern to extract year from filenames like ef2010d_rv.csv or ef2024d.csv +pattern = re.compile(r"ef(\d{4})d(?:_rv)?", re.IGNORECASE) def process_files(): try: + if not os.path.exists(input_folder): + logging.fatal("FATAL ERROR: Input folder '%s' does not exist.", input_folder) + + # Group files by extracted year + year_files = {} for filename in os.listdir(input_folder): + # Skip hidden files and lock files (e.g. .~lock.ef2024d.csv#) + if filename.startswith(".") or filename.startswith("~"): + continue + file_path = os.path.join(input_folder, filename) # Skip if not a file if not os.path.isfile(file_path): - logging.info(f"Skipping (not a file): {filename}") + logging.info("Skipping (not a file): %s", filename) continue - try: - # Extract year - match = pattern.search(filename) - if not match: - logging.info(f"Skipping {filename}: No year found.") - continue + match = pattern.search(filename) + if not match: + logging.info("Skipping %s: No year found.", filename) + continue + + year = match.group(1) + year_files.setdefault(year, []).append((filename, file_path)) + + if not year_files: + logging.fatal("FATAL ERROR: No valid input files found in '%s'.", input_folder) - year = match.group(1) + # Process each year + for year, files in year_files.items(): + # Prefer _rv file if available + selected_filename, selected_file_path = None, None + for fname, fpath in files: + if "_rv" in fname.lower(): + selected_filename, selected_file_path = fname, fpath + break + if not selected_file_path: + selected_filename, selected_file_path = files[0] - # New filename - new_filename = f"student_faculty_ratio_data_{year}.csv" - new_file_path = os.path.join(input_folder, new_filename) + # Remove any redundant files for the same year + for fname, fpath in files: + if fpath != selected_file_path and os.path.exists(fpath): + os.remove(fpath) + logging.info("Removed redundant file for year %s: %s", year, fname) - # Rename file - os.rename(file_path, new_file_path) - logging.info(f"Renamed: {filename} → {new_filename}") + new_filename = f"student_faculty_ratio_data_{year}.csv" + new_file_path = os.path.join(input_folder, new_filename) + + try: + if selected_file_path != new_file_path: + os.rename(selected_file_path, new_file_path) + logging.info("Renamed: %s → %s", selected_filename, new_filename) except Exception as e: - logging.fatal(f"Error renaming file {filename}: {e}") - continue + logging.fatal("Error renaming file %s: %s", selected_filename, e) try: # Load CSV df = pd.read_csv(new_file_path) - # Add Year column in the second position - df.insert(1, "Year", int(year)) + # Clean column headers + df.columns = df.columns.str.strip() + + # Add or update Year column in the second position + if "Year" not in df.columns: + df.insert(1, "Year", int(year)) + else: + df["Year"] = int(year) + + # Check if the source dataset is provisional (lacks '_rv' suffix) + is_provisional = "_rv" not in selected_filename.lower() + if is_provisional: + df["measurementMethod"] = "NCES_ProvisionalEstimate" + logging.info("Added 'measurementMethod' = 'NCES_ProvisionalEstimate' to provisional file %s", new_filename) + else: + df["measurementMethod"] = "" + logging.info("Set empty 'measurementMethod' for revised file %s", new_filename) # Save updated CSV df.to_csv(new_file_path, index=False) - logging.info(f"Updated: Added 'Year' column to {new_filename}") + logging.info("Updated: Successfully preprocessed %s", new_filename) except Exception as e: - logging.fatal(f"Error processing CSV {new_filename}: {e}") + logging.fatal("Error processing CSV %s: %s", new_filename, e) except Exception as e: - logging.fatal(f"Unexpected error: {e}") + logging.fatal("Unexpected error: %s", e) -if __name__ == "__main__": +def main(_): process_files() + +if __name__ == "__main__": + app.run(main) diff --git a/statvar_imports/ipeds/student_to_faculty_ratio/run.sh b/statvar_imports/ipeds/student_to_faculty_ratio/run.sh index 9260c53051..f11a60400c 100755 --- a/statvar_imports/ipeds/student_to_faculty_ratio/run.sh +++ b/statvar_imports/ipeds/student_to_faculty_ratio/run.sh @@ -1,3 +1,4 @@ +#!/bin/bash # Copyright 2025 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); @@ -12,8 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -#!/bin/bash - # ================================================================= # CONFIGURATION SECTION # ================================================================= @@ -24,7 +23,7 @@ OUTPUT_DIR="processed_output" EXISTING_MCF="gs://unresolved_mcf/scripts/statvar/stat_vars.mcf" PYTHON_SCRIPT="../../../tools/statvar_importer/stat_var_processor.py" -# TODO: Limit the concurrency to a defined number of jobs +# Concurrency limit for background processing jobs MAX_CONCURRENT_JOBS=4 # ================================================================= @@ -32,6 +31,7 @@ MAX_CONCURRENT_JOBS=4 # ================================================================= mkdir -p "$OUTPUT_DIR" +mkdir -p counters if [ ! -d "$INPUT_DIR" ]; then echo "Error: Input directory '$INPUT_DIR' not found." @@ -40,15 +40,17 @@ fi echo "Starting parallel processing..." -PIDS=() JOB_COUNT=0 +EXIT_STATUS=0 for input_file in "$INPUT_DIR"/*; do if [ -f "$input_file" ]; then - # TODO: Concurrency control logic to prevent system overload + # Concurrency control logic to prevent system overload while [ "$(jobs -rp | wc -l)" -ge "$MAX_CONCURRENT_JOBS" ]; do - wait -n + if ! wait -n; then + EXIT_STATUS=1 + fi done base_name=$(basename "$input_file") @@ -69,10 +71,10 @@ for input_file in "$INPUT_DIR"/*; do --config_file="$CONFIG_FILE" \ --existing_statvar_mcf="$EXISTING_MCF" \ --output_path="$output_base_path" \ + --output_counters="counters/${clean_base}_counters.csv" \ --log_level=-2 \ --log_every_n=1000 & - PIDS+=($!) JOB_COUNT=$((JOB_COUNT + 1)) fi done @@ -80,13 +82,9 @@ done echo "---" echo "Waiting for background jobs to complete..." -# TODO: Check for exit codes of background processes -EXIT_STATUS=0 -for pid in "${PIDS[@]}"; do - wait "$pid" - STATUS=$? - if [ $STATUS -ne 0 ]; then - echo "Error: Job PID $pid failed (Exit Code: $STATUS)" +# Wait for all remaining background jobs to finish and capture any failures +while [ "$(jobs -p | wc -l)" -gt 0 ]; do + if ! wait -n; then EXIT_STATUS=1 fi done @@ -97,27 +95,30 @@ done if [ $EXIT_STATUS -eq 0 ]; then echo "Processing successful. Finalizing file names..." - # 1. Delete all .tmcf files EXCEPT the 2009 one - find "$OUTPUT_DIR" -type f -name "*.tmcf" ! -name "*2009*" -delete - - # 2. Rename the 2009 tmcf file to exactly student_faculty_ratio.tmcf - # Note: If the script appended .csv.tmcf, this finds and fixes it - TMCF_2009=$(find "$OUTPUT_DIR" -type f -name "*2009*.tmcf" | head -n 1) - if [ -n "$TMCF_2009" ]; then - mv "$TMCF_2009" "$OUTPUT_DIR/student_faculty_ratio.tmcf" + # 1. Dynamically pick the first available .tmcf file and rename it to student_faculty_ratio.tmcf + FIRST_TMCF=$(find "$OUTPUT_DIR" -type f -name "*.tmcf" | sort | head -n 1) + if [ -n "$FIRST_TMCF" ]; then + if [ "$FIRST_TMCF" != "$OUTPUT_DIR/student_faculty_ratio.tmcf" ]; then + mv "$FIRST_TMCF" "$OUTPUT_DIR/student_faculty_ratio.tmcf" + fi + # Delete any remaining duplicate .tmcf files + find "$OUTPUT_DIR" -type f -name "*.tmcf" ! -name "student_faculty_ratio.tmcf" -delete + else + echo "Error: No .tmcf files found in $OUTPUT_DIR" + EXIT_STATUS=1 fi - # 3. Ensure CSV files are named correctly (removing any double extensions like .csv.csv) - # This specifically looks for the 2009 csv output - CSV_2009=$(find "$OUTPUT_DIR" -type f -name "*2009*.csv" | head -n 1) - if [ -n "$CSV_2009" ]; then - mv "$CSV_2009" "$OUTPUT_DIR/student_faculty_ratio_2009.csv" - fi + # 2. Normalize CSV file names (e.g. fix any double extensions like .csv.csv) for all years + for csv_file in "$OUTPUT_DIR"/*.csv.csv; do + if [ -f "$csv_file" ]; then + mv "$csv_file" "${csv_file%.csv}" + fi + done echo "Cleanup complete." echo "Results: " echo " - $OUTPUT_DIR/student_faculty_ratio.tmcf" - echo " - $OUTPUT_DIR/student_faculty_ratio_2009.csv" + echo " - $OUTPUT_DIR/*.csv" else echo "Cleanup skipped due to job failures." fi diff --git a/statvar_imports/ipeds/student_to_faculty_ratio/student_faculty_ratio_metadata.csv b/statvar_imports/ipeds/student_to_faculty_ratio/student_faculty_ratio_metadata.csv index 18d0e57384..a4a6383de7 100644 --- a/statvar_imports/ipeds/student_to_faculty_ratio/student_faculty_ratio_metadata.csv +++ b/statvar_imports/ipeds/student_to_faculty_ratio/student_faculty_ratio_metadata.csv @@ -7,7 +7,7 @@ start_date,2009 end_date, release_frequency,1 comments, -output_columns,"observationAbout,observationDate,value,variableMeasured" +output_columns,"observationAbout,observationDate,value,variableMeasured,measurementMethod" header_rows,1 drop_statvars_without_svobs,0 mapped_rows,1 diff --git a/statvar_imports/ipeds/student_to_faculty_ratio/student_faculty_ratio_pvmap.csv b/statvar_imports/ipeds/student_to_faculty_ratio/student_faculty_ratio_pvmap.csv index ad920f469d..d8a03fc8af 100644 --- a/statvar_imports/ipeds/student_to_faculty_ratio/student_faculty_ratio_pvmap.csv +++ b/statvar_imports/ipeds/student_to_faculty_ratio/student_faculty_ratio_pvmap.csv @@ -1,4 +1,5 @@ key,property,,,,,,, UNITID,#Format,observationAbout=ipedsId/{Data},,,,,, Year,observationDate,{Number},,,,,, -STUFACR,populationType,Student,measurementDenominator,Count_Teacher,measuredProperty,count,value,{Number} \ No newline at end of file +STUFACR,populationType,Student,measurementDenominator,Count_Teacher,measuredProperty,count,value,{Number} +measurementMethod,measurementMethod,{Data} \ No newline at end of file diff --git a/statvar_imports/ipeds/student_to_faculty_ratio/validation_config.json b/statvar_imports/ipeds/student_to_faculty_ratio/validation_config.json new file mode 100644 index 0000000000..241970276d --- /dev/null +++ b/statvar_imports/ipeds/student_to_faculty_ratio/validation_config.json @@ -0,0 +1,22 @@ +{ + "schema_version": "1.0", + "rules": [ + { + "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.1 + } + }, + { + "rule_id": "check_goldens_summary_report", + "description": "Validates summary report against golden summary report.", + "validator": "GOLDENS_CHECK", + "params": { + "golden_files": "../../../../golden_data/golden_summary_report.csv", + "input_files": "../genmcf/summary_report.csv" + } + } + ] +}