Update app.py

app.py

@@ -1,4 +1,3 @@
-
 import os
 os.environ["CUDA_VISIBLE_DEVICES"] = "-1"
 os.environ['JAX_PLATFORMS'] = 'cpu'
@@ -134,13 +133,17 @@ def calculate_overall_aqi(row, aqi_breakpoints):
 def get_latest_data_sequence(sequence_length: int, latitude: float, longitude: float):
     print(f"Attempting to retrieve data for the last {sequence_length} hours from Open-Meteo for Lat: {latitude}, Lon: {longitude}")
 
-    fetch_hours = sequence_length + 5
+    current_utc_time = datetime.now(pytz.utc)
+    print(f"Current UTC time on server for API calls: {current_utc_time.strftime('%Y-%m-%d %H:%M:%S UTC')}")
+
+    # Define a window to fetch from APIs, slightly larger than sequence_length to allow for finding complete data
+    # e.g., if sequence_length is 24, fetch last 48 hours to have a good buffer
+    api_fetch_past_hours = sequence_length + 24 # Fetch a wider window, e.g., 48 hours for a 24-hour sequence
     
-    end_time_for_temp = datetime.now(pytz.utc)
-    start_time_for_temp = end_time_for_temp - timedelta(hours=fetch_hours)
+    # This window will be used to filter the processed data before dropna and tail
+    processing_window_hours = sequence_length + 24 # e.g., 48 hours
 
-    print(f"Requesting data for the past {fetch_hours} hours for air quality.")
-    print(f"Requesting temperature data from {start_time_for_temp.strftime('%Y-%m-%d %H:%M:%S UTC')} to {end_time_for_temp.strftime('%Y-%m-%d %H:%M:%S UTC')}")
+    print(f"Requesting data for the past {api_fetch_past_hours} hours for air quality and temperature from APIs.")
 
     air_quality_url = "https://air-quality-api.open-meteo.com/v1/air-quality"
     air_quality_params = {
@@ -148,19 +151,20 @@ def get_latest_data_sequence(sequence_length: int, latitude: float, longitude: f
         "longitude": longitude,
         "hourly": ["pm2_5", "pm10", "carbon_monoxide"],
         "timezone": "UTC",
-        "past_hours": fetch_hours
+        "past_hours": api_fetch_past_hours
     }
-    print(f"Air quality API params: {air_quality_params}")
+    # print(f"Air quality API params: {air_quality_params}")
 
-    weather_url = "https://api.open-meteo.com/v1/forecast"
+    # Using forecast API for temperature as per user's finding that it works better
+    weather_url = "https://api.open-meteo.com/v1/forecast" 
     weather_params = {
         "latitude": latitude,
         "longitude": longitude,
         "hourly": ["temperature_2m"],
         "timezone": "UTC",
-        "past_hours": fetch_hours
+        "past_hours": api_fetch_past_hours # Fetch same window for temperature
     }
-    print(f"Temperature API params: {weather_params}")
+    # print(f"Temperature API params: {weather_params}")
 
     try:
         print(f"Fetching air quality data from: {air_quality_url}")
@@ -175,34 +179,42 @@ def get_latest_data_sequence(sequence_length: int, latitude: float, longitude: f
         weather_data = weather_response.json()
         print("Temperature data retrieved.")
 
-        print("Data fetched successfully.")
+        print("Data fetched successfully from APIs.")
 
         if 'hourly' not in air_quality_data or 'time' not in air_quality_data['hourly']:
             print("Error: 'hourly' or 'time' key not found in air quality response.")
             return None, "Error: Invalid air quality data format from API."
         df_aq = pd.DataFrame(air_quality_data['hourly'])
-        if df_aq.empty or not all(col in df_aq.columns for col in ['time', 'pm2_5', 'pm10', 'carbon_monoxide']):
-            print("Warning: Air quality data is empty or missing required columns ('time', 'pm2_5', 'pm10', 'carbon_monoxide') after fetching.")
-            # Depending on how critical each pollutant is, you might allow continuation or return error
-            # For now, let's be strict if key columns are missing from the structure
-            if 'time' not in df_aq.columns:
-                 return None, "Error: 'time' column missing in air quality data."
-        df_aq['time'] = pd.to_datetime(df_aq['time'])
-        df_aq.set_index('time', inplace=True)
-        print(f"Processed df_aq. Shape: {df_aq.shape}. Columns: {df_aq.columns.tolist()}")
-
+        if df_aq.empty:
+            print("Warning: Air quality data DataFrame is empty after fetching.")
+        # Continue if not empty, but columns might be missing
+        if not df_aq.empty and not all(col in df_aq.columns for col in ['time', 'pm2_5', 'pm10', 'carbon_monoxide']):
+            print("Warning: Air quality data is missing some expected columns ('time', 'pm2_5', 'pm10', 'carbon_monoxide') after fetching.")
+        if 'time' not in df_aq.columns and not df_aq.empty:
+             return None, "Error: 'time' column missing in air quality data."
+        if not df_aq.empty:
+            df_aq['time'] = pd.to_datetime(df_aq['time'])
+            df_aq.set_index('time', inplace=True)
+        print(f"Processed df_aq. Shape: {df_aq.shape}. Columns: {df_aq.columns.tolist() if not df_aq.empty else 'N/A'}")
 
         if 'hourly' not in weather_data or 'time' not in weather_data['hourly']:
             print("Error: 'hourly' or 'time' key not found in weather response.")
             return None, "Error: Invalid weather data format from API."
         df_temp = pd.DataFrame(weather_data['hourly'])
-        if df_temp.empty or not all(col in df_temp.columns for col in ['time', 'temperature_2m']):
-            print("Warning: Temperature data is empty or missing required columns ('time', 'temperature_2m') after fetching.")
-            if 'time' not in df_temp.columns:
-                return None, "Error: 'time' column missing in temperature data."
-        df_temp['time'] = pd.to_datetime(df_temp['time'])
-        df_temp.set_index('time', inplace=True)
-        print(f"Processed df_temp. Shape: {df_temp.shape}. Columns: {df_temp.columns.tolist()}")
+        if df_temp.empty:
+            print("Warning: Temperature data DataFrame is empty after fetching.")
+        if not df_temp.empty and not all(col in df_temp.columns for col in ['time', 'temperature_2m']):
+            print("Warning: Temperature data is missing some expected columns ('time', 'temperature_2m') after fetching.")
+        if 'time' not in df_temp.columns and not df_temp.empty:
+            return None, "Error: 'time' column missing in temperature data."
+        if not df_temp.empty:
+            df_temp['time'] = pd.to_datetime(df_temp['time'])
+            df_temp.set_index('time', inplace=True)
+        print(f"Processed df_temp. Shape: {df_temp.shape}. Columns: {df_temp.columns.tolist() if not df_temp.empty else 'N/A'}")
+        
+        if df_aq.empty or df_temp.empty:
+            print("Error: One or both dataframes (AQ, Temp) are empty before merge. Cannot proceed.")
+            return None, "Error: Insufficient data from APIs (AQ or Temp empty)."
 
         df_merged = df_aq.merge(df_temp, left_index=True, right_index=True, how='inner')
         print(f"DataFrames merged (inner). Initial merged shape: {df_merged.shape}")
@@ -210,55 +222,71 @@ def get_latest_data_sequence(sequence_length: int, latitude: float, longitude: f
             print("Error: Inner merge of AQ and Temperature data resulted in an empty DataFrame. No overlapping timestamps with data.")
             return None, "Error: No overlapping AQ and Temperature data available for the period."
 
-        df_processed = df_merged.resample('h').ffill().bfill()
-        print(f"DataFrame resampled to hourly. Shape: {df_processed.shape}")
+        # Resample to ensure consistent hourly frequency and fill missing data
+        df_processed = df_merged.resample('h').mean() # Use mean for resampling to handle potential duplicates at same hour
+        df_processed = df_processed.ffill().bfill() # Then fill
+        print(f"DataFrame resampled to hourly, filled NaNs. Shape: {df_processed.shape}")
+        # print(f"df_processed head after resample/ffill/bfill:\n{df_processed.head().to_string()}")
+        # print(f"df_processed NaNs after resample/ffill/bfill:\n{df_processed.isna().sum().to_string()}")
 
         df_processed.rename(columns={'pm2_5': 'pm25', 'carbon_monoxide': 'co', 'temperature_2m': 'temp'}, inplace=True)
         print(f"Renamed columns. Current columns: {df_processed.columns.tolist()}")
 
-        # Ensure all expected columns exist after rename, before calculating AQI
-        expected_cols_for_aqi = ['pm25', 'pm10', 'co'] # temp is also in df_processed
-        missing_for_aqi = [col for col in expected_cols_for_aqi if col not in df_processed.columns]
-        if missing_for_aqi:
-            print(f"Warning: Missing columns required for AQI calculation after rename: {missing_for_aqi}. AQI might be NaN.")
-            # Add missing columns with NaNs if they don't exist, so calculate_overall_aqi doesn't fail
-            for col in missing_for_aqi:
+        expected_cols_for_aqi = ['pm25', 'pm10', 'co']
+        for col in expected_cols_for_aqi:
+            if col not in df_processed.columns:
+                print(f"Warning: Column '{col}' for AQI calculation is missing after rename. Adding as NaN.")
                 df_processed[col] = np.nan
-
-
+        
         df_processed['calculated_aqi'] = df_processed.apply(lambda row: calculate_overall_aqi(row, aqi_breakpoints), axis=1)
         print("Calculated AQI.")
+        # print(f"df_processed head after AQI calculation:\n{df_processed.head().to_string()}")
+        # print(f"df_processed NaNs after AQI calculation:\n{df_processed.isna().sum().to_string()}")
 
         required_columns = ['calculated_aqi', 'temp', 'pm25', 'pm10', 'co']
-        # Ensure all required columns exist before selecting, add if missing to prevent KeyError
         for col in required_columns:
             if col not in df_processed.columns:
                 print(f"Warning: Column '{col}' is missing before final selection. Adding it as NaN.")
                 df_processed[col] = np.nan
         
         df_processed = df_processed[required_columns].copy()
-        print(f"Selected and reordered columns. Current shape: {df_processed.shape}. Columns: {df_processed.columns.tolist()}")
+        # print(f"Selected and reordered columns. Shape before windowing: {df_processed.shape}. Columns: {df_processed.columns.tolist()}")
 
-        initial_rows = len(df_processed)
-        df_processed.dropna(inplace=True)
-        if len(df_processed) < initial_rows:
-             print(f"Warning: Dropped {initial_rows - len(df_processed)} rows with remaining NaNs after all processing.")
+        # Filter to the defined processing window relative to current time
+        # Ensure we only consider data up to the current hour and back by processing_window_hours
+        window_start_time = current_utc_time.replace(minute=0, second=0, microsecond=0) - timedelta(hours=processing_window_hours - 1)
+        window_end_time = current_utc_time.replace(minute=0, second=0, microsecond=0)
+        
+        df_recent_processed = df_processed[(df_processed.index >= window_start_time) & (df_processed.index <= window_end_time)].copy()
+        print(f"Filtered to recent processing window ({processing_window_hours}hrs). Shape: {df_recent_processed.shape}")
+        # print(f"df_recent_processed head:\n{df_recent_processed.head().to_string()}")
+        # print(f"df_recent_processed NaNs before dropna:\n{df_recent_processed.isna().sum().to_string()}")
+
+
+        initial_rows_recent = len(df_recent_processed)
+        df_recent_processed.dropna(inplace=True)
+        if len(df_recent_processed) < initial_rows_recent:
+             print(f"Warning: Dropped {initial_rows_recent - len(df_recent_processed)} rows with NaNs from the recent processing window.")
+        print(f"Shape after dropna on recent window: {df_recent_processed.shape}")
+
+        if len(df_recent_processed) < sequence_length:
+            print(f"Error: Only {len(df_recent_processed)} valid data points remain in the recent window after processing, but {sequence_length} are required.")
+            return None, f"Error: Insufficient historical data in the recent window ({len(df_recent_processed)} points available, {sequence_length} required)."
 
-        if len(df_processed) < sequence_length:
-            print(f"Error: Only {len(df_processed)} valid data points remain after processing, but {sequence_length} are required.")
-            return None, f"Error: Insufficient historical data ({len(df_processed)} points available, {sequence_length} required)."
+        latest_data_sequence_df = df_recent_processed.tail(sequence_length).copy()
+        print(f"Selected last {sequence_length} data points for model input. Shape: {latest_data_sequence_df.shape}")
+        # print(f"Final sequence data:\n{latest_data_sequence_df.to_string()}")
 
-        latest_data_sequence_df = df_processed.tail(sequence_length).copy()
-        print(f"Selected last {sequence_length} data points for model input.")
 
         latest_data_sequence = latest_data_sequence_df.values.reshape(1, sequence_length, len(required_columns))
         timestamps = latest_data_sequence_df.index.tolist()
-        print(f"Prepared input sequence with shape: {latest_data_sequence.shape}")
+        # print(f"Prepared input sequence with shape: {latest_data_sequence.shape}")
 
         return latest_data_sequence, timestamps
 
     except requests.exceptions.RequestException as e:
         print(f"API Request Error: {e}")
+        traceback.print_exc()
         return None, f"API Request Error: {e}"
     except Exception as e:
         print(f"An unexpected error occurred during data retrieval and processing: {e}")
@@ -384,15 +412,18 @@ async def predict_aqi_endpoint(request: PredictionRequest):
 
         current_aqi = calculate_overall_aqi({'pm25': request.pm25, 'pm10': request.pm10, 'co': request.co, 'temp': request.temp}, aqi_breakpoints)
 
-        if not pd.isna(current_aqi):
+        if not pd.isna(current_aqi) and latest_data_sequence_unscaled.shape[1] == SEQUENCE_LENGTH : # Ensure sequence is correctly shaped
             latest_data_sequence_unscaled[0, -1, 0] = current_aqi
             latest_data_sequence_unscaled[0, -1, 1] = request.temp
             latest_data_sequence_unscaled[0, -1, 2] = request.pm25
             latest_data_sequence_unscaled[0, -1, 3] = request.pm10
             latest_data_sequence_unscaled[0, -1, 4] = request.co
             print("Updated last timestep of input sequence with current user inputs.")
-        else:
+        elif pd.isna(current_aqi):
              print("Warning: Could not calculate AQI for current inputs. Last timestep remains historical.")
+        else:
+            print("Warning: Sequence not correctly shaped to update with current user inputs, or current_aqi is NaN.")
+
 
     try:
         X_scaled = input_scaler.transform(latest_data_sequence_unscaled)
@@ -449,4 +480,4 @@ async def predict_aqi_endpoint(request: PredictionRequest):
 
 @app.get("/")
 async def read_root():
-    return {"message": "AQI Prediction API is running."}   
+    return {"message": "AQI Prediction API is running."}