Update app.py

app.py

@@ -3,216 +3,220 @@ import requests
 import pandas as pd
 from sklearn.linear_model import LogisticRegression
 from fastai.vision.all import *
-from fastai.vision.all import PILImage
 import os
-from datetime import datetime
+from datetime import datetime, timedelta
 
 # Load the trained model for image-based fog classification
 learn = load_learner('fog_classifier.pkl')
 labels = learn.dls.vocab
 
-# Use environment variables for API keys
 API_KEY = os.environ.get("OPENWEATHER_API_KEY")
 BASE_URL = 'https://api.openweathermap.org/data/2.5/'
 
-# Define the prediction function for image-based fog classification
 def predict_image(img):
+    """Predict fog conditions from image and return confidence scores"""
     img = PILImage.create(img)
     img = img.resize((512, 512))
     pred, pred_idx, probs = learn.predict(img)
     return {labels[i]: float(probs[i]) for i in range(len(labels))}
 
-# Function to get weather data
-def get_weather_data(location):
-    current_weather_url = f'{BASE_URL}weather?q={location}&appid={API_KEY}&units=metric'
-    current_response = requests.get(current_weather_url)
-    current_data = current_response.json()
-
-    current_weather = {
-        'temperature': current_data['main'].get('temp', 0),
-        'feels_like': current_data['main'].get('feels_like', 0),
-        'description': current_data['weather'][0].get('description', ''),
-        'wind_speed': current_data['wind'].get('speed', 0),
-        'pressure': current_data['main'].get('pressure', 0),
-        'humidity': current_data['main'].get('humidity', 0),
-        'visibility': current_data.get('visibility', 10000) / 1000,
-        'dew_point': current_data['main'].get('temp', 0) - ((100 - current_data['main'].get('humidity', 0)) / 5.0)
+def calculate_fog_risk_score(weather_data):
+    """Calculate a fog risk score (0-1) based on weather conditions"""
+    # Normalized weights for each factor
+    weights = {
+        'humidity': 0.3,
+        'dew_point_temp_diff': 0.3,
+        'visibility': 0.2,
+        'wind_speed': 0.1,
+        'pressure_change': 0.1
     }
-    return current_weather
-
-# Function to train the fog prediction model
-def train_fog_model():
-    df = pd.read_csv('fog_weather_data.csv')
-    df = pd.get_dummies(df, columns=['Description'], drop_first=True)
-    X = df.drop('Fog', axis=1)
-    y = df['Fog']
-    model = LogisticRegression()
-    model.fit(X, y)
-    return model, X.columns
-
-# Function to predict fog based on weather data
-def predict_fog(model, feature_columns, weather_data):
-    new_data = pd.DataFrame({
-        'Temperature': [weather_data.get('temperature', 0)],
-        'Feels like': [weather_data.get('feels_like', 0)],
-        'Wind speed': [weather_data.get('wind_speed', 0)],
-        'Pressure': [weather_data.get('pressure', 0)],
-        'Humidity': [weather_data.get('humidity', 0)],
-        'Dew point': [weather_data.get('dew_point', 0)],
-        'Visibility': [weather_data.get('visibility', 10)]
-    })
-
-    for col in feature_columns:
-        if col.startswith('Description_'):
-            new_data[col] = 0
+    
+    # Calculate dew point
+    dew_point = weather_data['temperature'] - ((100 - weather_data['humidity']) / 5.0)
+    dew_point_temp_diff = abs(weather_data['temperature'] - dew_point)
+    
+    # Normalize each factor to 0-1 scale
+    humidity_score = min(weather_data['humidity'] / 100, 1)
+    dew_point_score = 1 - min(dew_point_temp_diff / 5, 1)  # Closer to dew point = higher score
+    visibility_score = 1 - min(weather_data['visibility'] / 10, 1)  # Lower visibility = higher score
+    wind_score = 1 - min(weather_data['wind_speed'] / 10, 1)  # Lower wind = higher score
+    pressure_score = min(abs(weather_data['pressure'] - 1013.25) / 50, 1)  # Deviation from normal pressure
+    
+    # Calculate weighted score
+    fog_risk = (
+        weights['humidity'] * humidity_score +
+        weights['dew_point_temp_diff'] * dew_point_score +
+        weights['visibility'] * visibility_score +
+        weights['wind_speed'] * wind_score +
+        weights['pressure_change'] * pressure_score
+    )
+    
+    return fog_risk
 
-    description_column = f"Description_{weather_data.get('description', '').replace(' ', '_')}"
-    if description_column in feature_columns:
-        new_data[description_column] = 1
-
-    prediction = model.predict(new_data)
-    return "Foggy weather" if prediction[0] == 1 else "Clear weather"
-
-# Function to get 5-day weather forecast
-def get_5day_forecast(location):
-    forecast_url = f'{BASE_URL}forecast?q={location}&appid={API_KEY}&units=metric'
-    forecast_response = requests.get(forecast_url)
-    forecast_data = forecast_response.json()
-
-    forecast_list = []
-    for entry in forecast_data['list']:
-        forecast_list.append({
-            'date': entry['dt_txt'],
-            'temperature': entry['main'].get('temp', 0),
-            'humidity': entry['main'].get('humidity', 0),
-            'description': entry['weather'][0].get('description', '')
-        })
-    return forecast_list
+def get_weather_data(location):
+    """Get current weather data with enhanced error handling"""
+    try:
+        current_weather_url = f'{BASE_URL}weather?q={location}&appid={API_KEY}&units=metric'
+        response = requests.get(current_weather_url)
+        response.raise_for_status()
+        data = response.json()
+        
+        return {
+            'temperature': data['main'].get('temp', 0),
+            'feels_like': data['main'].get('feels_like', 0),
+            'description': data['weather'][0].get('description', ''),
+            'wind_speed': data['wind'].get('speed', 0),
+            'pressure': data['main'].get('pressure', 0),
+            'humidity': data['main'].get('humidity', 0),
+            'visibility': data.get('visibility', 10000) / 1000,
+            'timestamp': datetime.fromtimestamp(data['dt'])
+        }
+    except requests.exceptions.RequestException as e:
+        raise Exception(f"Failed to fetch weather data: {str(e)}")
 
-# Load the fog prediction model
-fog_model, feature_columns = train_fog_model()
+def get_forecast_data(location):
+    """Get 5-day forecast with enhanced error handling"""
+    try:
+        forecast_url = f'{BASE_URL}forecast?q={location}&appid={API_KEY}&units=metric'
+        response = requests.get(forecast_url)
+        response.raise_for_status()
+        data = response.json()
+        
+        forecasts = []
+        for item in data['list']:
+            forecasts.append({
+                'temperature': item['main'].get('temp', 0),
+                'humidity': item['main'].get('humidity', 0),
+                'description': item['weather'][0].get('description', ''),
+                'wind_speed': item['wind'].get('speed', 0),
+                'pressure': item['main'].get('pressure', 0),
+                'visibility': item.get('visibility', 10000) / 1000,
+                'timestamp': datetime.fromtimestamp(item['dt'])
+            })
+        return forecasts
+    except requests.exceptions.RequestException as e:
+        raise Exception(f"Failed to fetch forecast data: {str(e)}")
+
+def determine_transmission_power(image_prediction, weather_data, forecast_data=None):
+    """
+    Determine transmission power based on current conditions and forecast
+    Returns: (power_level, duration, explanation)
+    """
+    # Get fog confidence from image
+    image_fog_confidence = max(
+        image_prediction.get('Dense_Fog', 0),
+        image_prediction.get('Moderate_Fog', 0) * 0.6
+    )
+    
+    # Calculate weather-based fog risk
+    current_fog_risk = calculate_fog_risk_score(weather_data)
+    
+    # Combine image and weather predictions with weighted average
+    # Give slightly more weight to image prediction as it's more reliable
+    combined_fog_risk = (image_fog_confidence * 0.6) + (current_fog_risk * 0.4)
+    
+    # Initialize explanation
+    explanation = []
+    
+    # Determine base power level from current conditions
+    if combined_fog_risk > 0.7:
+        power_level = "High"
+        explanation.append(f"High fog risk detected (Risk score: {combined_fog_risk:.2f})")
+    elif combined_fog_risk > 0.4:
+        power_level = "Medium"
+        explanation.append(f"Moderate fog risk detected (Risk score: {combined_fog_risk:.2f})")
+    else:
+        power_level = "Normal"
+        explanation.append(f"Low fog risk detected (Risk score: {combined_fog_risk:.2f})")
+    
+    # Analyze forecast data if available
+    duration = timedelta(hours=1)  # Default duration
+    if forecast_data:
+        future_risks = []
+        for forecast in forecast_data[:40]:  # 5 days of 3-hour forecasts
+            risk = calculate_fog_risk_score(forecast)
+            future_risks.append(risk)
+            
+        # Find periods of high risk
+        high_risk_periods = [risk > 0.6 for risk in future_risks]
+        if any(high_risk_periods):
+            # Find the last high-risk timestamp
+            last_high_risk_idx = len(high_risk_periods) - 1 - high_risk_periods[::-1].index(True)
+            duration = forecast_data[last_high_risk_idx]['timestamp'] - weather_data['timestamp']
+            
+            explanation.append(f"High fog risk predicted to continue for {duration.days} days and {duration.hours} hours")
+            
+            # Adjust power level based on forecast
+            if sum(high_risk_periods) / len(high_risk_periods) > 0.5:
+                power_level = "High"
+                explanation.append("Power level set to High due to sustained fog risk in forecast")
+    
+    return power_level, duration, explanation
 
-# Function to predict current weather and fog
-def predict_current_weather(location):
+def integrated_prediction(image, location):
+    """Main function to process image and weather data"""
     try:
+        # Get image prediction
+        image_prediction = predict_image(image)
+        
+        # Get current weather
         current_weather = get_weather_data(location)
-        fog_prediction = predict_fog(fog_model, feature_columns, current_weather)
-
-        result = f"Current weather in {location}:\n"
-        result += f"Temperature: {current_weather.get('temperature', 0)}°C\n"
-        result += f"Feels like: {current_weather.get('feels_like', 0)}°C\n"
-        result += f"Description: {current_weather.get('description', '')}\n"
-        result += f"Wind speed: {current_weather.get('wind_speed', 0)} m/s\n"
-        result += f"Pressure: {current_weather.get('pressure', 0)} hPa\n"
-        result += f"Humidity: {current_weather.get('humidity', 0)}%\n"
-        result += f"Dew point: {current_weather.get('dew_point', 0)}°C\n"
-        result += f"Visibility: {current_weather.get('visibility', 10)} km\n"
-        result += f"\nFog Prediction: {fog_prediction}"
-
-        return result
+        
+        # Get forecast
+        forecast_data = get_forecast_data(location)
+        
+        # Determine transmission power
+        power_level, duration, explanation = determine_transmission_power(
+            image_prediction,
+            current_weather,
+            forecast_data
+        )
+        
+        # Format result
+        result = [
+            f"Current Conditions ({current_weather['timestamp'].strftime('%Y-%m-%d %H:%M')})",
+            f"Temperature: {current_weather['temperature']:.1f}°C",
+            f"Humidity: {current_weather['humidity']}%",
+            f"Visibility: {current_weather['visibility']:.1f} km",
+            f"Wind Speed: {current_weather['wind_speed']} m/s",
+            "",
+            "Analysis Results:",
+            *explanation,
+            "",
+            f"Recommended Power Level: {power_level}",
+            f"Duration: {duration.days} days and {duration.hours} hours",
+            "",
+            "5-Day Forecast Summary:"
+        ]
+        
+        # Add daily forecast summary
+        current_date = current_weather['timestamp'].date()
+        for day in range(5):
+            forecast_date = current_date + timedelta(days=day)
+            day_forecasts = [f for f in forecast_data if f['timestamp'].date() == forecast_date]
+            
+            if day_forecasts:
+                avg_risk = sum(calculate_fog_risk_score(f) for f in day_forecasts) / len(day_forecasts)
+                result.append(f"{forecast_date.strftime('%Y-%m-%d')}: "
+                            f"Fog Risk: {'High' if avg_risk > 0.6 else 'Moderate' if avg_risk > 0.3 else 'Low'} "
+                            f"({avg_risk:.2f})")
+        
+        return "\n".join(result)
+        
     except Exception as e:
         return f"Error: {str(e)}"
 
-# Function to determine transmission power based on combined results
-def determine_transmission_power(image_prediction, weather_prediction, humidity):
-    image_class = max(image_prediction, key=image_prediction.get)
-    weather_class = "Foggy weather" if weather_prediction == "Foggy weather" else "Clear weather"
-
-    if humidity > 80:  # High humidity increases the likelihood of fog
-        if image_class == "Dense_Fog" or weather_class == "Foggy weather":
-            return "High"
-        elif image_class == "Moderate_Fog":
-            return "Medium"
-        else:
-            return "Normal"
-    else:
-        if image_class == "Dense_Fog":
-            return "Medium"
-        elif image_class == "Moderate_Fog":
-            return "Normal"
-        else:
-            return "Normal"
-
-# Function to format 5-day weather forecast
-def format_5day_forecast(forecast_list):
-    forecast_dict = {}
-    for entry in forecast_list:
-        date = entry['date'].split()[0]
-        if date not in forecast_dict:
-            forecast_dict[date] = {
-                'temperature': entry['temperature'],
-                'humidity': entry['humidity'],
-                'description': entry['description']
-            }
-        else:
-            forecast_dict[date]['temperature'] += entry['temperature']
-            forecast_dict[date]['humidity'] += entry['humidity']
-            forecast_dict[date]['description'] = entry['description']
-
-    formatted_forecast = []
-    for date, data in forecast_dict.items():
-        avg_temperature = data['temperature'] / 8  # 8 entries per day
-        avg_humidity = data['humidity'] / 8
-        formatted_forecast.append(f"Date: {date}\nTemperature: {avg_temperature:.2f}°C\nHumidity: {avg_humidity:.2f}%\nDescription: {data['description']}\n")
-
-    return "\n".join(formatted_forecast)
-
-# Main function to integrate all functionalities
-def integrated_prediction(image, location):
-    # Predict fog from image
-    image_prediction = predict_image(image)
-
-    # Get current weather data and predict fog
-    current_weather = get_weather_data(location)
-    weather_prediction = predict_fog(fog_model, feature_columns, current_weather)
-
-    # Get 5-day weather forecast
-    forecast_list = get_5day_forecast(location)
-    formatted_forecast = format_5day_forecast(forecast_list)
-
-    # Determine transmission power based on current and future fog predictions
-    transmission_power = determine_transmission_power(image_prediction, weather_prediction, current_weather.get('humidity', 0))
-
-    # Check fog predictions for the next 5 days
-    fog_days = []
-    for entry in forecast_list:
-        date = entry['date'].split()[0]
-        weather_data = {
-            'temperature': entry.get('temperature', 0),
-            'humidity': entry.get('humidity', 0),
-            'description': entry.get('description', ''),
-            'visibility': 10,  # Assuming visibility is 10 km for simplicity
-            'dew_point': entry.get('temperature', 0) - ((100 - entry.get('humidity', 0)) / 5.0)
-        }
-        fog_prediction = predict_fog(fog_model, feature_columns, weather_data)
-        if fog_prediction == "Foggy weather":
-            fog_days.append(date)
-
-    # Prepare the result
-    result = f"Current Weather and Fog Prediction:\n{predict_current_weather(location)}\n\n"
-    result += f"5-Day Weather Forecast:\n{formatted_forecast}\n\n"
-    result += f"Transmission Power: {transmission_power}\n"
-
-    if fog_days:
-        result += f"Foggy conditions predicted on: {', '.join(fog_days)}\n"
-        if len(fog_days) == 5:
-            result += "Transmission power will be kept high for the next 5 days.\n"
-        else:
-            result += f"Transmission power will be high on: {fog_days[-1]}\n"
-
-    return result
-
 # Gradio interface
 with gr.Blocks() as demo:
-    gr.Markdown("# Integrated Fog Prediction and Transmission Power Decision")
-
+    gr.Markdown("# Enhanced Fog Prediction and Transmission Power System")
+    
     with gr.Row():
-        image_input = gr.Image(label="Upload Image")
+        image_input = gr.Image(label="Upload Current Conditions Image")
         location_input = gr.Textbox(label="Enter Location")
-
-    predict_button = gr.Button("Predict and Determine Transmission Power")
-    output = gr.Textbox(label="Prediction and Transmission Power Result")
-
+    
+    predict_button = gr.Button("Analyze and Determine Transmission Power")
+    output = gr.Textbox(label="Analysis Results", lines=15)
+    
     predict_button.click(integrated_prediction, inputs=[image_input, location_input], outputs=output)
 
 demo.launch()