#!/usr/bin/env python3 """ ECMWF Open Data Weather Forecast Application Access real ECMWF operational forecast data with coordinate-based lookups. """ import gradio as gr import numpy as np import pandas as pd import matplotlib.pyplot as plt import xarray as xr import requests import tempfile import os from datetime import datetime, timedelta import warnings import time import folium import plotly.graph_objects as go import plotly.express as px from plotly.subplots import make_subplots warnings.filterwarnings('ignore') try: from ecmwf.opendata import Client as OpenDataClient OPENDATA_AVAILABLE = True except ImportError: OPENDATA_AVAILABLE = False class ECMWFDataManager: def __init__(self): self.temp_dir = tempfile.mkdtemp() self.client = None if OPENDATA_AVAILABLE: try: self.client = OpenDataClient() except: self.client = None # AWS S3 direct access URLs self.aws_base_url = "https://ecmwf-forecasts.s3.eu-central-1.amazonaws.com" # ECMWF Open Data parameters - verified available as of 2024/2025 self.parameters = { # Surface level parameters (single level) "2t": {"name": "Temperature (2m)", "units": "°C", "description": "2-meter temperature", "level_type": "sfc"}, "msl": {"name": "Sea Level Pressure", "units": "hPa", "description": "Mean sea level pressure", "level_type": "sfc"}, "sp": {"name": "Surface Pressure", "units": "hPa", "description": "Surface pressure", "level_type": "sfc"}, "10u": {"name": "Wind U (10m)", "units": "m/s", "description": "10-meter U wind component", "level_type": "sfc"}, "10v": {"name": "Wind V (10m)", "units": "m/s", "description": "10-meter V wind component", "level_type": "sfc"}, "tp": {"name": "Precipitation", "units": "mm", "description": "Total precipitation", "level_type": "sfc"}, "tcwv": {"name": "Water Vapor", "units": "kg/m²", "description": "Total column water vapor", "level_type": "sfc"}, "skt": {"name": "Skin Temperature", "units": "°C", "description": "Skin temperature", "level_type": "sfc"}, "ro": {"name": "Runoff", "units": "m", "description": "Runoff", "level_type": "sfc"}, "st": {"name": "Soil Temperature", "units": "°C", "description": "Soil temperature", "level_type": "sfc"}, # Available severe weather parameters - confirmed in ECMWF Open Data # Note: Advanced hazard parameters like MUCAPE, precipitation probabilities, # and gust probabilities are available in ECMWF's full datasets but not in the free Open Data stream # We'll use derived calculations from basic parameters for hazard assessment # Pressure level parameters (add common levels) "t": {"name": "Temperature", "units": "°C", "description": "Temperature at pressure levels", "level_type": "pl", "levels": [850, 500, 200]}, "gh": {"name": "Geopotential Height", "units": "m", "description": "Geopotential height", "level_type": "pl", "levels": [850, 500, 200]}, "u": {"name": "Wind U", "units": "m/s", "description": "U wind component", "level_type": "pl", "levels": [850, 500, 200]}, "v": {"name": "Wind V", "units": "m/s", "description": "V wind component", "level_type": "pl", "levels": [850, 500, 200]}, "q": {"name": "Specific Humidity", "units": "g/kg", "description": "Specific humidity", "level_type": "pl", "levels": [850, 500]}, "r": {"name": "Relative Humidity", "units": "%", "description": "Relative humidity", "level_type": "pl", "levels": [850, 500]}, } self.forecast_cache = {} self.preloaded_data = {} def get_latest_forecast_info(self): """Get the most recent available forecast run""" now = datetime.utcnow() # Check recent 6-hour cycles for hours_back in range(4, 24, 6): test_time = now - timedelta(hours=hours_back) run_hour = (test_time.hour // 6) * 6 run_time = test_time.replace(hour=run_hour, minute=0, second=0, microsecond=0) date_str = run_time.strftime("%Y%m%d") time_str = f"{run_hour:02d}" # Test availability test_url = f"{self.aws_base_url}/{date_str}/{time_str}z/0p25/oper/" try: response = requests.head(test_url, timeout=10) if response.status_code in [200, 403]: return date_str, time_str, run_time except: continue # Fallback return now.strftime("%Y%m%d"), "12", now def download_forecast_data(self, parameter="2t", step=0, level=None, max_retries=3): """Download ECMWF forecast data using multiple methods with rate limiting""" date_str, time_str, run_time = self.get_latest_forecast_info() # Get parameter info param_info = self.parameters.get(parameter, {}) level_type = param_info.get('level_type', 'sfc') for attempt in range(max_retries): try: # Method 1: Official client if OPENDATA_AVAILABLE and self.client: try: cache_suffix = f"_{level}" if level else "" filename = os.path.join(self.temp_dir, f'ecmwf_{parameter}{cache_suffix}_{step}h.grib') # Build retrieval request request = { "type": "fc", "param": parameter, "step": step, "target": filename } # Add pressure level if needed if level_type == 'pl' and level: request["levelist"] = level elif level_type == 'pl': # Use first available level if no specific level requested levels = param_info.get('levels', [850]) request["levelist"] = levels[0] self.client.retrieve(**request) if os.path.exists(filename) and os.path.getsize(filename) > 1000: level_info = f" at {level}hPa" if level else "" return filename, f"Downloaded {parameter}{level_info} +{step}h via ECMWF client" except Exception as e: error_msg = str(e).lower() if "429" in error_msg or "428" in error_msg or "too many requests" in error_msg: if attempt < max_retries - 1: print(f"Rate limited for {parameter} step {step}, retrying in 10 seconds (attempt {attempt + 1}/{max_retries})") time.sleep(10) continue print(f"Client method failed for {parameter}: {e}") # Method 2: AWS S3 direct access (for surface parameters only) if level_type == 'sfc': try: step_str = f"{step:03d}" filename = f"{date_str}{time_str}0000-{step_str}h-oper-fc.grib2" url = f"{self.aws_base_url}/{date_str}/{time_str}z/0p25/oper/{filename}" response = requests.get(url, timeout=120, stream=True) if response.status_code == 429 or response.status_code == 428: if attempt < max_retries - 1: print(f"Rate limited for {parameter} step {step}, retrying in 10 seconds (attempt {attempt + 1}/{max_retries})") time.sleep(10) continue if response.status_code == 200: local_file = os.path.join(self.temp_dir, f'ecmwf_{parameter}_{step}h.grib2') with open(local_file, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): f.write(chunk) if os.path.getsize(local_file) > 1000: return local_file, f"Downloaded {parameter} +{step}h via AWS S3" except Exception as e: error_msg = str(e).lower() if "429" in error_msg or "428" in error_msg or "too many requests" in error_msg: if attempt < max_retries - 1: print(f"Rate limited for {parameter} step {step}, retrying in 10 seconds (attempt {attempt + 1}/{max_retries})") time.sleep(10) continue print(f"AWS method failed for {parameter}: {e}") break # If we get here without rate limiting, don't retry except Exception as e: if attempt < max_retries - 1: print(f"General error for {parameter} step {step}, retrying (attempt {attempt + 1}/{max_retries}): {e}") time.sleep(5) else: print(f"Failed all attempts for {parameter} step {step}: {e}") return None, f"Failed to download {parameter} at +{step}h after {max_retries} attempts" def extract_point_data(self, filename, lat, lon, parameter): """Extract weather data at specific coordinates""" try: ds = xr.open_dataset(filename, engine='cfgrib', backend_kwargs={'indexpath': ''}) data_vars = list(ds.data_vars.keys()) if not data_vars: return None data = ds[data_vars[0]] # Handle coordinates if 'latitude' in ds.coords: lats, lons = ds.latitude, ds.longitude elif 'lat' in ds.coords: lats, lons = ds.lat, ds.longitude else: return None # Select first time if multiple if 'time' in data.dims and len(data.time) > 1: data = data.isel(time=0) elif 'valid_time' in data.dims: data = data.isel(valid_time=0) # Find nearest point try: point_data = data.sel(latitude=lat, longitude=lon, method='nearest') except: try: point_data = data.sel(lat=lat, lon=lon, method='nearest') except: return None value = float(point_data.values) ds.close() return self.convert_units(value, parameter) except Exception as e: print(f"Error extracting point data: {e}") return None def convert_units(self, value, parameter): """Convert values to standard meteorological units""" if parameter in ['2t', 'skt', 't', 'st'] and value > 100: return value - 273.15 # K to °C elif parameter in ['msl', 'sp']: return value / 100 # Pa to hPa elif parameter == 'tp': return value * 1000 # m to mm elif parameter == 'q': return value * 1000 # kg/kg to g/kg elif parameter == 'ro': return value * 1000 # m to mm elif parameter == 'gh': return value / 9.80665 # m²/s² to meters (geopotential to height) return value def preload_all_data(self): """Preload all forecast data for quick access""" # 3-hourly for first 24 hours (ECMWF operational availability), then longer intervals forecast_steps = [0, 3, 6, 9, 12, 15, 18, 21, 24, 30, 36, 42, 48, 60, 72, 96, 120] # Use confirmed available surface parameters only surface_params = ["2t", "msl", "sp", "10u", "10v", "tp", "tcwv", "skt"] total_files = len(surface_params) * len(forecast_steps) loaded_count = 0 for param in surface_params: for step in forecast_steps: try: cache_key = f"{param}_{step}" filename, msg = self.download_forecast_data(param, step) if filename: self.preloaded_data[cache_key] = filename loaded_count += 1 time.sleep(0.5) # Small delay to avoid rate limiting except Exception as e: print(f"Failed to preload {param} at step {step}: {e}") continue return loaded_count, total_files def get_point_forecast(self, latitude, longitude): """Get comprehensive forecast for a specific location""" forecast_data = [] # 3-hourly for first 24 hours (ECMWF operational availability), then longer intervals forecast_steps = [0, 3, 6, 9, 12, 15, 18, 21, 24, 30, 36, 42, 48, 60, 72, 96, 120] # Focus on surface parameters that are reliably available surface_params = ["2t", "msl", "sp", "10u", "10v", "tp", "tcwv", "skt"] for param in surface_params: if param not in self.parameters: continue param_data = [] for step in forecast_steps: try: cache_key = f"{param}_{step}" # Try to use preloaded data first if cache_key in self.preloaded_data: filename = self.preloaded_data[cache_key] else: filename, _ = self.download_forecast_data(param, step) if filename and os.path.exists(filename): value = self.extract_point_data(filename, latitude, longitude, param) if value is not None: param_data.append({ 'step': step, 'value': value, 'datetime': datetime.utcnow() + timedelta(hours=step) }) except Exception as e: print(f"Error getting {param} at step {step}: {e}") continue if param_data: forecast_data.append({ 'parameter': param, 'name': self.parameters[param]['name'], 'units': self.parameters[param]['units'], 'data': param_data }) return forecast_data def generate_weather_narrative(self, forecast_data, latitude, longitude): """Generate a comprehensive weather.gov-style narrative forecast with daily paragraphs""" if not forecast_data: return "No forecast data available for narrative generation." # Extract parameters for narrative temp_data = None pressure_data = None wind_u_data = None wind_v_data = None precip_data = None humidity_data = None # tcwv for atmospheric moisture analysis for param_info in forecast_data: param = param_info['parameter'] if param == '2t': temp_data = param_info['data'] elif param == 'msl': pressure_data = param_info['data'] elif param == '10u': wind_u_data = param_info['data'] elif param == '10v': wind_v_data = param_info['data'] elif param == 'tp': precip_data = param_info['data'] elif param == 'tcwv': humidity_data = param_info['data'] if not temp_data or len(temp_data) < 4: return "Insufficient data for narrative generation." # Generate location description location_desc = f"{abs(latitude):.1f}°{'N' if latitude >= 0 else 'S'}, {abs(longitude):.1f}°{'E' if longitude >= 0 else 'W'}" # Build narrative parts narrative_parts = [] # Current conditions (convert to Fahrenheit) current_temp_f = self._celsius_to_fahrenheit(temp_data[0]['value']) narrative_parts.append(f"Weather forecast for {location_desc}. Current conditions show temperatures near {current_temp_f:.0f}°F.") # Group data by complete days (24-hour periods) with derived hazard analysis daily_forecasts = self._organize_into_daily_forecasts( temp_data, wind_u_data, wind_v_data, precip_data, pressure_data, humidity_data ) # Generate narrative for each day with different detail levels for i, day_forecast in enumerate(daily_forecasts): if i >= 5: # Limit to 5 days break if i < 2: # Detailed descriptions for first 2 days (Today & Tomorrow) detailed_narrative = self._generate_detailed_daily_narrative(day_forecast, i) if detailed_narrative: narrative_parts.append(detailed_narrative) else: # Concise day/night summaries for days 3-5 concise_narrative = self._generate_concise_daily_narrative(day_forecast, i) if concise_narrative: narrative_parts.append(concise_narrative) # Add overall trend analysis if len(daily_forecasts) >= 2: trend_analysis = self._analyze_weekly_trends(daily_forecasts) if trend_analysis: narrative_parts.append(trend_analysis) return "\n\n".join(narrative_parts) def _organize_into_daily_forecasts(self, temp_data, wind_u_data, wind_v_data, precip_data, pressure_data, humidity_data=None): """Organize forecast data into complete daily forecasts (24-hour periods) with derived hazard analysis""" daily_forecasts = [] # Group data into 24-hour chunks starting from current time i = 0 day_count = 0 while i < len(temp_data) and day_count < 5: # Up to 5 days # Determine day name if day_count == 0: day_name = "Today" elif day_count == 1: day_name = "Tomorrow" else: # Generate day names (this is approximate - could use actual dates) day_names = ["Wednesday", "Thursday", "Friday", "Saturday", "Sunday", "Monday", "Tuesday"] day_name = day_names[(day_count - 2) % len(day_names)] # Collect data for this day (up to 8 data points for 24 hours at 3-hour intervals) daily_data = { 'day_name': day_name, 'day_number': day_count, 'temps': [], 'winds': [], 'precip': [], 'pressure': [], 'humidity': [], # For atmospheric moisture analysis 'hazards': { 'derived_instability': [], # Derived from pressure changes and humidity 'wind_speeds': [], # Calculated wind speeds for gust assessment 'precip_intensity': [], # Calculated precipitation rates 'pressure_tendency': [] # Pressure change rates }, 'start_step': temp_data[i]['step'] if i < len(temp_data) else 0 } # Collect up to 8 data points (24 hours worth) points_per_day = 8 # 24 hours / 3 hours = 8 data points for j in range(min(points_per_day, len(temp_data) - i)): if i + j < len(temp_data): daily_data['temps'].append(temp_data[i + j]['value']) # Add wind data if available if wind_u_data and wind_v_data and i + j < len(wind_u_data) and i + j < len(wind_v_data): u = wind_u_data[i + j]['value'] v = wind_v_data[i + j]['value'] speed = np.sqrt(u**2 + v**2) * 2.237 # Convert to mph direction = (270 - np.degrees(np.arctan2(v, u))) % 360 daily_data['winds'].append({'speed': speed, 'direction': direction}) # Add precipitation data if available if precip_data and i + j < len(precip_data): daily_data['precip'].append(precip_data[i + j]['value']) # Add pressure data if available if pressure_data and i + j < len(pressure_data): daily_data['pressure'].append(pressure_data[i + j]['value']) # Add humidity data if available if humidity_data and i + j < len(humidity_data): daily_data['humidity'].append(humidity_data[i + j]['value']) # Calculate derived hazard indicators if wind_u_data and wind_v_data and i + j < len(wind_u_data) and i + j < len(wind_v_data): u = wind_u_data[i + j]['value'] v = wind_v_data[i + j]['value'] wind_speed = np.sqrt(u**2 + v**2) * 2.237 # Convert to mph daily_data['hazards']['wind_speeds'].append(wind_speed) # Calculate precipitation intensity (mm per 3-hour period) if precip_data and i + j < len(precip_data): precip_rate = precip_data[i + j]['value'] / 3.0 # mm per hour daily_data['hazards']['precip_intensity'].append(precip_rate) # Calculate pressure tendency if we have multiple points if pressure_data and len(daily_data['pressure']) > 1: if i + j < len(pressure_data): current_pressure = pressure_data[i + j]['value'] previous_pressure = daily_data['pressure'][-1] pressure_change = current_pressure - previous_pressure daily_data['hazards']['pressure_tendency'].append(pressure_change) # Derive atmospheric instability indicator from humidity and pressure if humidity_data and pressure_data and i + j < len(humidity_data) and i + j < len(pressure_data): moisture = humidity_data[i + j]['value'] pressure = pressure_data[i + j]['value'] # Simple instability index based on high moisture + low pressure instability = moisture / (pressure / 1000.0) # Normalized daily_data['hazards']['derived_instability'].append(instability) if daily_data['temps']: # Only add if we have temperature data daily_forecasts.append(daily_data) day_count += 1 i += points_per_day return daily_forecasts def _generate_detailed_daily_narrative(self, day_forecast, day_index): """Generate detailed comprehensive narrative for first 2 days (Today & Tomorrow)""" if not day_forecast['temps']: return "" temps = day_forecast['temps'] day_name = day_forecast['day_name'] # Calculate daily temperature statistics (convert to Fahrenheit) min_temp_f = self._celsius_to_fahrenheit(min(temps)) max_temp_f = self._celsius_to_fahrenheit(max(temps)) temp_range_f = max_temp_f - min_temp_f # Start building the detailed narrative narrative_parts = [] # Temperature narrative with range (Fahrenheit) if temp_range_f > 15: # 15°F range is significant narrative_parts.append(f"{day_name}: A variable day with temperatures ranging from a low of {min_temp_f:.0f}°F to a high of {max_temp_f:.0f}°F.") else: narrative_parts.append(f"{day_name}: High {max_temp_f:.0f}°F, low {min_temp_f:.0f}°F.") # Detailed wind narrative if day_forecast['winds']: wind_speeds = [w['speed'] for w in day_forecast['winds']] wind_dirs = [w['direction'] for w in day_forecast['winds']] avg_wind = np.mean(wind_speeds) max_wind = max(wind_speeds) min_wind = min(wind_speeds) avg_dir = np.mean(wind_dirs) dir_name = self._get_wind_direction_name(avg_dir) # Determine wind variability wind_variability = max_wind - min_wind if avg_wind < 5: narrative_parts.append("Light and variable winds throughout the day with occasional calm periods.") elif wind_variability > 10: narrative_parts.append(f"{dir_name} winds varying from {min_wind:.0f} to {max_wind:.0f} mph, becoming gusty at times.") elif avg_wind < 15: narrative_parts.append(f"Moderate {dir_name} winds averaging {avg_wind:.0f} mph with gusts to {max_wind:.0f} mph.") else: narrative_parts.append(f"Breezy conditions with {dir_name} winds {avg_wind:.0f} to {max_wind:.0f} mph, gusts up to {max_wind*1.3:.0f} mph.") # Detailed precipitation narrative if day_forecast['precip']: total_precip = sum(day_forecast['precip']) precip_periods = len([p for p in day_forecast['precip'] if p > 0.1]) if total_precip > 0.1: if total_precip < 2.5: narrative_parts.append("Scattered light showers possible with minimal accumulation. Brief periods of light rain expected.") elif total_precip < 10: narrative_parts.append(f"Periods of rain expected with {total_precip:.1f}mm total accumulation. Rain likely during {precip_periods} periods throughout the day.") elif total_precip < 25: narrative_parts.append(f"Significant rainfall likely with {total_precip:.1f}mm total accumulation, heaviest during afternoon and evening hours.") else: narrative_parts.append(f"Heavy rain expected throughout much of the day with {total_precip:.1f}mm total accumulation. Potential for localized flooding.") else: narrative_parts.append("Dry conditions prevail with clear to partly cloudy skies and no significant precipitation expected.") else: narrative_parts.append("Fair weather expected with dry conditions and mostly sunny to partly cloudy skies.") # Severe weather and hazard assessment hazard_warnings = self._assess_daily_hazards(day_forecast['hazards']) if hazard_warnings: narrative_parts.extend(hazard_warnings) # Detailed pressure trend analysis if day_forecast['pressure'] and len(day_forecast['pressure']) > 2: pressure_start = day_forecast['pressure'][0] pressure_end = day_forecast['pressure'][-1] pressure_change = pressure_end - pressure_start pressure_trend = "steady" if pressure_change > 5: pressure_trend = "rapidly rising" narrative_parts.append("Rapidly rising pressure indicates clearing weather and improving conditions through the day.") elif pressure_change > 2: pressure_trend = "rising" narrative_parts.append("Rising pressure suggests gradually improving weather conditions.") elif pressure_change < -5: pressure_trend = "rapidly falling" narrative_parts.append("Rapidly falling pressure indicates an approaching weather system with potential for deteriorating conditions.") elif pressure_change < -2: pressure_trend = "falling" narrative_parts.append("Falling pressure suggests increasing instability and possible weather changes.") else: narrative_parts.append("Steady pressure indicates stable weather patterns continuing.") return " ".join(narrative_parts) def _generate_concise_daily_narrative(self, day_forecast, day_index): """Generate concise day/night summaries for days 3-5 with highs/lows and notable conditions""" if not day_forecast['temps']: return "" temps = day_forecast['temps'] day_name = day_forecast['day_name'] # Split data into day/night periods (approximate) mid_point = len(temps) // 2 day_temps = temps[:mid_point] if len(temps) > 4 else temps[:4] night_temps = temps[mid_point:] if len(temps) > 4 else temps[4:] if len(temps) > 4 else temps[-2:] # Calculate day and night temperatures (convert to Fahrenheit) day_high_f = self._celsius_to_fahrenheit(max(day_temps)) if day_temps else self._celsius_to_fahrenheit(max(temps)) night_low_f = self._celsius_to_fahrenheit(min(night_temps)) if night_temps else self._celsius_to_fahrenheit(min(temps)) # Start building concise narrative narrative_parts = [] # Day period summary day_conditions = [] night_conditions = [] # Analyze wind conditions for notable mentions if day_forecast['winds']: max_wind = max([w['speed'] for w in day_forecast['winds']]) if max_wind > 25: day_conditions.append(f"windy, gusts to {max_wind:.0f} mph") elif max_wind > 15: day_conditions.append("breezy") # Analyze precipitation if day_forecast['precip']: total_precip = sum(day_forecast['precip']) if total_precip > 10: day_conditions.append("rain likely") elif total_precip > 2: day_conditions.append("chance of rain") # Check for notable hazards hazard_warnings = self._assess_daily_hazards(day_forecast['hazards']) if hazard_warnings: if any("SEVERE" in warning or "FLOOD" in warning for warning in hazard_warnings): day_conditions.append("severe weather possible") elif any("THUNDERSTORM" in warning for warning in hazard_warnings): day_conditions.append("thunderstorms possible") # Build day/night summary day_summary = f"High {day_high_f:.0f}°F" if day_conditions: day_summary += f", {', '.join(day_conditions)}" night_summary = f"Low {night_low_f:.0f}°F" # Add any notable night conditions (typically fewer) if day_forecast['precip'] and sum(day_forecast['precip'][-3:]) > 2: # Rain in latter part of day night_summary += ", evening rain possible" narrative_parts.append(f"{day_name}: {day_summary}. {day_name} night: {night_summary}.") return " ".join(narrative_parts) def _celsius_to_fahrenheit(self, celsius): """Convert Celsius to Fahrenheit""" return (celsius * 9/5) + 32 def _assess_daily_hazards(self, hazards): """Assess severe weather hazards using derived calculations from basic parameters""" warnings = [] # Atmospheric instability assessment (derived from humidity and pressure) if hazards['derived_instability']: max_instability = max(hazards['derived_instability']) avg_instability = np.mean(hazards['derived_instability']) if max_instability > 35: # High moisture + low pressure warnings.append("⚠️ MODERATE THUNDERSTORM RISK: Atmospheric conditions favor thunderstorm development with high moisture and low pressure.") elif max_instability > 25: warnings.append("⚠️ Thunderstorm potential exists with elevated atmospheric moisture and unstable conditions.") # Wind hazard assessment (from calculated wind speeds) if hazards['wind_speeds']: max_wind = max(hazards['wind_speeds']) avg_wind = np.mean(hazards['wind_speeds']) # Estimate gust potential (typically 1.3-1.5x sustained winds) estimated_gusts = max_wind * 1.4 if estimated_gusts > 65: warnings.append("⚠️ SEVERE WIND WARNING: Damaging wind gusts possible, potentially exceeding 65 mph.") elif estimated_gusts > 45: warnings.append("⚠️ Strong wind gusts forecast, potentially reaching 45-65 mph.") elif max_wind > 25: warnings.append("Strong winds expected with gusts possible.") # Heavy precipitation/flooding risk assessment (from precipitation rates) if hazards['precip_intensity']: max_rate = max(hazards['precip_intensity']) # mm/hour total_period_precip = sum([rate * 3 for rate in hazards['precip_intensity']]) # Total over day if max_rate > 8: # >8mm/hour is heavy rainfall warnings.append("⚠️ FLOOD RISK: Heavy rainfall rates (>8mm/hour) create potential for localized flooding.") elif max_rate > 4 and total_period_precip > 20: warnings.append("⚠️ Heavy rainfall possible with flooding risk from sustained moderate rates.") elif total_period_precip > 25: warnings.append("⚠️ Significant rainfall accumulation expected, monitor for potential flooding.") # Rapid pressure changes (indicates weather system intensity) if hazards['pressure_tendency']: max_pressure_drop = min(hazards['pressure_tendency']) # Most negative = biggest drop max_pressure_rise = max(hazards['pressure_tendency']) # Most positive = biggest rise if max_pressure_drop < -3: warnings.append("⚠️ Rapidly falling pressure indicates an intensifying weather system approaching.") elif max_pressure_rise > 3: warnings.append("Rapidly rising pressure suggests weather conditions improving quickly.") return warnings def _analyze_weekly_trends(self, daily_forecasts): """Analyze overall weather trends across the 5-day forecast period""" if len(daily_forecasts) < 2: return "" # Extract daily highs and lows daily_highs = [] daily_lows = [] total_precip_by_day = [] for day in daily_forecasts: if day['temps']: daily_highs.append(max(day['temps'])) daily_lows.append(min(day['temps'])) if day['precip']: total_precip_by_day.append(sum(day['precip'])) else: total_precip_by_day.append(0) trend_parts = [] # Temperature trends if len(daily_highs) >= 3: temp_trend = daily_highs[-1] - daily_highs[0] if temp_trend > 8: trend_parts.append("Temperatures trending significantly warmer through the forecast period.") elif temp_trend < -8: trend_parts.append("Temperatures trending notably cooler through the forecast period.") elif temp_trend > 3: trend_parts.append("Gradual warming trend expected.") elif temp_trend < -3: trend_parts.append("Gradual cooling trend anticipated.") # Precipitation patterns total_period_precip = sum(total_precip_by_day) wet_days = sum(1 for p in total_precip_by_day if p > 1.0) if total_period_precip > 50: trend_parts.append("A wet period ahead with frequent rain expected.") elif wet_days >= 3: trend_parts.append("Unsettled weather pattern with multiple days of rain likely.") elif total_period_precip < 5: trend_parts.append("Generally dry conditions expected through the forecast period.") # Overall stability assessment temp_variability = max(daily_highs) - min(daily_highs) if daily_highs else 0 if temp_variability > 15: trend_parts.append("Highly variable weather pattern with significant temperature swings.") elif temp_variability < 5: trend_parts.append("Stable weather pattern with consistent temperatures.") # Severe weather outlook across the period using derived hazard data hazard_days = 0 thunderstorm_days = 0 windy_days = 0 for day in daily_forecasts: hazards = day.get('hazards', {}) # Count days with thunderstorm potential (derived instability) if hazards.get('derived_instability') and max(hazards['derived_instability']) > 25: thunderstorm_days += 1 # Count days with wind hazards (estimated gusts > 45 mph) if hazards.get('wind_speeds') and max(hazards['wind_speeds']) * 1.4 > 45: windy_days += 1 # Count days with significant weather hazards (wind, rain, or instability) if (hazards.get('wind_speeds') and max(hazards['wind_speeds']) > 25) or \ (hazards.get('precip_intensity') and max(hazards['precip_intensity']) > 4) or \ (hazards.get('derived_instability') and max(hazards['derived_instability']) > 25): hazard_days += 1 if thunderstorm_days >= 3: trend_parts.append("⚠️ EXTENDED THUNDERSTORM PERIOD: Multiple days with atmospheric instability and thunderstorm potential.") elif thunderstorm_days >= 2: trend_parts.append("⚠️ Several days with thunderstorm potential from unstable atmospheric conditions.") if windy_days >= 3: trend_parts.append("⚠️ Prolonged windy period with multiple days of strong winds expected.") if hazard_days >= 4: trend_parts.append("⚠️ Multiple days with various severe weather hazards possible.") if trend_parts: return "5-Day Outlook: " + " ".join(trend_parts) return "" def _get_wind_direction_name(self, direction): """Convert wind direction in degrees to cardinal direction name""" directions = ["North", "Northeast", "East", "Southeast", "South", "Southwest", "West", "Northwest"] idx = round(direction / 45) % 8 return directions[idx] def _get_time_description(self, hour): """Convert hour to descriptive time phrase""" if hour == 0: return "currently" elif hour <= 6: return "early morning" elif hour <= 12: return "late morning" elif hour <= 15: return "early afternoon" elif hour <= 18: return "late afternoon" elif hour <= 21: return "evening" else: return "overnight" class WeatherApp: def __init__(self): self.ecmwf = ECMWFDataManager() self.preload_status = {"loaded": False, "count": 0, "total": 0} def create_map(self): """Create interactive map for location selection""" try: m = folium.Map( location=[45.0, 0.0], zoom_start=2, tiles='OpenStreetMap' ) # Add click functionality m.add_child(folium.ClickForMarker(popup="Click for coordinates")) return m._repr_html_() except: return """

🗺️ World Map

Map unavailable - use coordinate inputs below

""" def preload_data(self): """Preload forecast data for faster access""" try: loaded_count, total_files = self.ecmwf.preload_all_data() self.preload_status = {"loaded": True, "count": loaded_count, "total": total_files} return f"""✅ Data Preloaded Successfully! 📊 Status: {loaded_count}/{total_files} files cached 🌍 Coverage: Global forecast data ready ⚡ Ready for instant weather lookups anywhere on Earth! Now you can click on the map or enter coordinates for instant forecasts.""" except Exception as e: return f"❌ Preload failed: {str(e)}" def get_weather_forecast(self, latitude, longitude): """Get weather forecast for specified coordinates""" try: if not (-90 <= latitude <= 90) or not (-180 <= longitude <= 180): return "Invalid coordinates", "", "", "" forecast_data = self.ecmwf.get_point_forecast(latitude, longitude) if not forecast_data: return "No forecast data available", "", "", "" # Generate weather narrative weather_narrative = self.ecmwf.generate_weather_narrative(forecast_data, latitude, longitude) # Create visualization with calculated wind speed fig = go.Figure() colors = ['#e74c3c', '#3498db', '#2ecc71', '#f39c12', '#9b59b6', '#1abc9c', '#e67e22', '#34495e'] color_idx = 0 # Find wind components for speed calculation wind_u_data = None wind_v_data = None for param_info in forecast_data: if param_info['parameter'] == '10u': wind_u_data = param_info['data'] elif param_info['parameter'] == '10v': wind_v_data = param_info['data'] for param_info in forecast_data: if param_info['data']: steps = [d['step'] for d in param_info['data']] values = [d['value'] for d in param_info['data']] fig.add_trace(go.Scatter( x=steps, y=values, mode='lines+markers', name=f"{param_info['name']} ({param_info['units']})", line=dict(color=colors[color_idx % len(colors)], width=2), marker=dict(size=4), connectgaps=True )) color_idx += 1 # Add calculated wind speed if both components available if wind_u_data and wind_v_data and len(wind_u_data) == len(wind_v_data): wind_speeds = [] wind_steps = [] for i, u_data in enumerate(wind_u_data): if i < len(wind_v_data): v_data = wind_v_data[i] if u_data['step'] == v_data['step']: wind_speed = np.sqrt(u_data['value']**2 + v_data['value']**2) wind_speeds.append(wind_speed) wind_steps.append(u_data['step']) if wind_speeds: fig.add_trace(go.Scatter( x=wind_steps, y=wind_speeds, mode='lines+markers', name="Wind Speed (m/s)", line=dict(color=colors[color_idx % len(colors)], width=3, dash='dash'), marker=dict(size=4), connectgaps=True )) fig.update_layout( title=f"🌍 ECMWF 3-Hourly Forecast - {latitude:.3f}°N, {longitude:.3f}°E", xaxis_title="Hours Ahead", yaxis_title="Values", height=700, hovermode='x unified', xaxis=dict( tickmode='array', tickvals=[0, 3, 6, 9, 12, 15, 18, 21, 24, 48, 72, 96, 120], ticktext=['0h', '3h', '6h', '9h', '12h', '15h', '18h', '21h', '1d', '2d', '3d', '4d', '5d'], gridcolor='lightgray', gridwidth=1 ), legend=dict( orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1 ), margin=dict(t=80) ) # Create enhanced data table with calculated parameters table_data = [] for param_info in forecast_data: for data_point in param_info['data']: table_data.append({ 'Parameter': param_info['name'], 'Hours': f"+{data_point['step']}h", 'Value': f"{data_point['value']:.2f} {param_info['units']}", 'Valid Time': data_point['datetime'].strftime('%Y-%m-%d %H:%M UTC') }) # Add calculated wind parameters if wind_u_data and wind_v_data: for i, u_data in enumerate(wind_u_data): if i < len(wind_v_data): v_data = wind_v_data[i] if u_data['step'] == v_data['step']: # Wind speed wind_speed = np.sqrt(u_data['value']**2 + v_data['value']**2) # Wind direction (meteorological convention) wind_dir = (270 - np.degrees(np.arctan2(v_data['value'], u_data['value']))) % 360 table_data.extend([{ 'Parameter': 'Wind Speed (calculated)', 'Hours': f"+{u_data['step']}h", 'Value': f"{wind_speed:.2f} m/s", 'Valid Time': u_data['datetime'].strftime('%Y-%m-%d %H:%M UTC') }, { 'Parameter': 'Wind Direction (calculated)', 'Hours': f"+{u_data['step']}h", 'Value': f"{wind_dir:.0f} degrees", 'Valid Time': u_data['datetime'].strftime('%Y-%m-%d %H:%M UTC') }]) df = pd.DataFrame(table_data) table_html = df.to_html(index=False, classes="table table-striped") status = f"""✅ 3-Hourly Forecast Retrieved! 📍 Location: {latitude:.4f}°N, {longitude:.4f}°E 📊 Parameters: {len(forecast_data)} weather variables ⏰ Forecast range: 3-hourly for first 24h, then extended to 120h 🔄 Data points: {len(table_data)} measurements 🕐 Resolution: 3-hour intervals for first day, then 6-hour+""" return status, fig, table_html, weather_narrative except Exception as e: return f"Error: {str(e)}", None, "", "" # Initialize the application weather_app = WeatherApp() # Gradio interface with gr.Blocks(title="ECMWF Weather Forecast") as app: gr.Markdown(""" # 🌍 ECMWF Global Weather Forecast ## Real-time weather data from ECMWF operational forecasts **Features:** - 🌐 Global coverage at 25km resolution - 🕐 **3-hourly forecasts for first 24 hours** - ⚠️ **Intelligent hazard assessment** - 🌩️ **Derived thunderstorm risk analysis** - 💨 **Wind gust potential estimation** - 🌊 **Flood risk evaluation from rainfall rates** - 📈 **Pressure tendency analysis** - 🔄 Updated every 6 hours - 📊 Professional meteorological data - 🆓 No API keys required """) with gr.Row(): with gr.Column(scale=2): gr.Markdown("### 🗺️ Interactive World Map") map_display = gr.HTML(value=weather_app.create_map()) with gr.Column(scale=1): gr.Markdown("### ⚡ Quick Setup") preload_btn = gr.Button("🚀 Preload Global Data", variant="primary", size="lg") preload_status = gr.Textbox(label="Status", lines=8, interactive=False) gr.Markdown("### 📍 Enter Coordinates") latitude = gr.Number( label="Latitude (-90 to 90)", value=40.7128, minimum=-90, maximum=90, step=0.001 ) longitude = gr.Number( label="Longitude (-180 to 180)", value=-74.0060, minimum=-180, maximum=180, step=0.001 ) get_forecast_btn = gr.Button("🌤️ Get Weather Forecast", variant="secondary", size="lg") with gr.Row(): with gr.Column(): forecast_status = gr.Textbox(label="Forecast Status", lines=6) forecast_plot = gr.Plot(label="Weather Forecast Chart") with gr.Column(): forecast_table = gr.HTML(label="Detailed Forecast Data") with gr.Row(): with gr.Column(): gr.Markdown("### 📝 5-Day Weather Narrative with Hazard Assessment") weather_narrative = gr.Textbox( label="Plain English Forecast (Detailed Days 1-2, Day/Night Summary Days 3-5)", lines=18, interactive=False, placeholder="Enhanced weather narrative with detailed descriptions for Today & Tomorrow, plus day/night summaries for the extended forecast, including severe weather warnings and temperatures in Fahrenheit...", max_lines=35 ) # Event handlers preload_btn.click( weather_app.preload_data, outputs=[preload_status] ) get_forecast_btn.click( weather_app.get_weather_forecast, inputs=[latitude, longitude], outputs=[forecast_status, forecast_plot, forecast_table, weather_narrative] ) if __name__ == "__main__": app.launch(server_name="0.0.0.0", server_port=7860)