Merge pull request #17 from OR-Dept-Environmental-Quality/feature/ryan-add-envista-staging

Feature/ryan-add-envista-staging

Author: DEQabiberi

pipelines/envista/run_env_service.py

@@ -30,7 +30,7 @@
 from envista.extractors.measurements import get_envista_hourly, get_envista_daily
 
 
-ENV_TEST_MODE = config.ENV_TEST_MODE
+ENV_TEST_MODE = str(config.ENV_TEST_MODE).lower() in ("1", "true", "yes")
 ENV_MONITOR_DIR = config.RAW_ENV_MONITORS
 ENV_SAMPLE_DIR = config.RAW_ENV_SAMPLE
 ENV_DAILY_DIR = config.RAW_ENV_DAILY

pipelines/envista/run_env_transform.py

@@ -31,6 +31,7 @@ def run():
     raw_monitors_dir = config.RAW_ENV_MONITORS
     raw_daily_dir = config.RAW_ENV_DAILY
     trans_daily_dir = config.TRANS_DAILY
+    trans_aqi_dir = config.TRANS_AQI
 
     if not raw_monitors_dir.exists():
         print(f"Raw monitors directory not found: {raw_monitors_dir}")
@@ -64,11 +65,10 @@ def run():
             print(f"No data for year {year_str}, skipping")
             continue
 
-        # Write to transform layer
-        output_path = trans_daily_dir / f"envista_daily_{year_str}.csv"
-        write_csv(transform_daily_df, output_path)
-
-        print(f"Wrote {len(transform_daily_df)} AQI records to {output_path}")
+        # Write to AQI transform layer
+        aqi_output_path = trans_aqi_dir / f"envista_aqi_{year_str}.csv"
+        write_csv(transform_daily_df, aqi_output_path)
+        print(f"Wrote {len(transform_daily_df)} AQI records to {aqi_output_path}")
 
         years_processed += 1
         total_records += len(transform_daily_df)

src/stage/consolidate_fct_criteria_daily.py

@@ -36,20 +36,35 @@ def consolidate_criteria_daily_for_year(year: str, transform_dir: Path) -> pd.Da
     Returns:
         Consolidated DataFrame with criteria daily fact data
     """
-    # Read the transformed data for this year
-    input_file = transform_dir / f"aqi_aqs_daily_{year}.csv"
-    if not input_file.exists():
-        print(f"⚠️  No AQI daily file found for year {year}: {input_file}")
+    # Find all transformed data files matching the pattern for this year
+    pattern = f"*aqi*{year}.csv"
+    matching_files = list(transform_dir.glob(pattern))
+    
+    if not matching_files:
+        print(f"⚠️  No AQI daily files found for year {year} matching pattern: {pattern}")
         return pd.DataFrame()
 
-    try:
-        df = pd.read_csv(input_file)
-    except Exception as e:
-        print(f"❌ Error reading {input_file}: {e}")
+    print(f"   📂 Found {len(matching_files)} file(s) matching pattern for {year}")
+    
+    dfs = []
+    for input_file in matching_files:
+        try:
+            file_df = pd.read_csv(input_file)
+            print(f"   ✓ Read {len(file_df)} records from {input_file.name}")
+            dfs.append(file_df)
+        except Exception as e:
+            print(f"❌ Error reading {input_file}: {e}")
+            continue
+    
+    if not dfs:
+        print(f"⚠️  No files were successfully read for year {year}")
         return pd.DataFrame()
-
+    
+    # Combine all dataframes
+    df = pd.concat(dfs, ignore_index=True)
+    
     if df.empty:
-        print(f"⚠️  Empty AQI daily file for year {year}")
+        print(f"⚠️  All combined files are empty for year {year}")
         return pd.DataFrame()
 
     # Define the required columns for fct_criteria_daily fact table