""" Open-Meteo Integration Module ============================== Fetch weather data from Open-Meteo API for agricultural purposes. https://open-meteo.com/ """ import requests from typing import Dict, Optional, Tuple def geocode_location(location: str) -> Optional[Tuple[float, float, str]]: """ Convert location name to coordinates using Open-Meteo Geocoding API. Args: location: City, region, or country name Returns: Tuple of (latitude, longitude, full_location_name) or None if not found """ try: url = "https://geocoding-api.open-meteo.com/v1/search" params = { "name": location, "count": 1, "language": "en", "format": "json" } response = requests.get(url, params=params, timeout=10) response.raise_for_status() data = response.json() if "results" in data and len(data["results"]) > 0: result = data["results"][0] lat = result["latitude"] lon = result["longitude"] # Build full location name name_parts = [result["name"]] if "admin1" in result and result["admin1"]: name_parts.append(result["admin1"]) if "country" in result and result["country"]: name_parts.append(result["country"]) full_name = ", ".join(name_parts) return (lat, lon, full_name) return None except Exception as e: print(f"Geocoding error: {e}") return None def get_weather_data(latitude: float, longitude: float) -> Optional[Dict]: """ Get current and forecast weather data from Open-Meteo API. Args: latitude: Location latitude longitude: Location longitude Returns: Dictionary with weather data or None if error """ try: url = "https://api.open-meteo.com/v1/forecast" params = { "latitude": latitude, "longitude": longitude, "current": [ "temperature_2m", "relative_humidity_2m", "precipitation", "weather_code", "wind_speed_10m", "wind_direction_10m" ], "hourly": [ "temperature_2m", "precipitation_probability", "precipitation", "weather_code" ], "daily": [ "temperature_2m_max", "temperature_2m_min", "precipitation_sum", "precipitation_probability_max", "wind_speed_10m_max" ], "timezone": "auto", "forecast_days": 7 } response = requests.get(url, params=params, timeout=10) response.raise_for_status() return response.json() except Exception as e: print(f"Weather API error: {e}") return None def interpret_weather_code(code: int) -> Tuple[str, str]: """ Interpret WMO Weather interpretation codes. Args: code: WMO weather code Returns: Tuple of (emoji, description) """ weather_codes = { 0: ("☀️", "Clear sky"), 1: ("🌤️", "Mainly clear"), 2: ("⛅", "Partly cloudy"), 3: ("☁️", "Overcast"), 45: ("🌫️", "Foggy"), 48: ("🌫️", "Depositing rime fog"), 51: ("🌦️", "Light drizzle"), 53: ("🌦️", "Moderate drizzle"), 55: ("🌧️", "Dense drizzle"), 61: ("🌧️", "Slight rain"), 63: ("🌧️", "Moderate rain"), 65: ("⛈️", "Heavy rain"), 71: ("🌨️", "Slight snow"), 73: ("🌨️", "Moderate snow"), 75: ("❄️", "Heavy snow"), 77: ("🌨️", "Snow grains"), 80: ("🌦️", "Slight rain showers"), 81: ("⛈️", "Moderate rain showers"), 82: ("⛈️", "Violent rain showers"), 85: ("🌨️", "Slight snow showers"), 86: ("❄️", "Heavy snow showers"), 95: ("⛈️", "Thunderstorm"), 96: ("⛈️", "Thunderstorm with slight hail"), 99: ("⛈️", "Thunderstorm with heavy hail") } return weather_codes.get(code, ("🌈", "Unknown")) def get_agricultural_recommendations(weather_data: Dict) -> list: """ Generate agricultural recommendations based on weather data. Args: weather_data: Weather data from Open-Meteo API Returns: List of recommendation strings """ recommendations = [] if not weather_data: return recommendations current = weather_data.get("current", {}) daily = weather_data.get("daily", {}) # Temperature recommendations temp = current.get("temperature_2m") if temp is not None: if temp < 5: recommendations.append("❄️ **Frost Risk**: Protect sensitive crops from cold damage") elif temp > 35: recommendations.append("🌡️ **Heat Warning**: Increase irrigation and provide shade for sensitive plants") elif temp > 30: recommendations.append("☀️ **High Temperature**: Monitor water needs closely") # Precipitation recommendations precip = current.get("precipitation", 0) if precip > 5: recommendations.append("🌧️ **Heavy Rain**: Delay irrigation and check drainage systems") elif precip > 0: recommendations.append("🌦️ **Light Rain**: Adjust irrigation schedule accordingly") # Wind recommendations wind_speed = current.get("wind_speed_10m", 0) if wind_speed > 50: recommendations.append("💨 **Strong Winds**: Secure equipment and check plant supports") elif wind_speed > 30: recommendations.append("🍃 **Moderate Winds**: Monitor young plants and structures") # Humidity recommendations humidity = current.get("relative_humidity_2m") if humidity is not None: if humidity > 85: recommendations.append("💧 **High Humidity**: Watch for fungal diseases and improve ventilation") elif humidity < 30: recommendations.append("🏜️ **Low Humidity**: Increase watering frequency") # Forecast-based recommendations if daily: max_precip_prob = max(daily.get("precipitation_probability_max", [0])) if max_precip_prob > 70: recommendations.append("☔ **Rain Expected**: Plan indoor activities for the coming days") temps_max = daily.get("temperature_2m_max", []) temps_min = daily.get("temperature_2m_min", []) if temps_min and min(temps_min[:3]) < 0: recommendations.append("⚠️ **Frost Warning**: Frost expected in the next 3 days") if not recommendations: recommendations.append("✅ **Favorable Conditions**: Current weather is good for normal agricultural activities") return recommendations def get_weather_for_location(location: str) -> Optional[Dict]: """ Complete workflow: geocode location and get weather data. Args: location: Location name (city, region, country) Returns: Dictionary with coordinates, location name, weather data, and recommendations """ # Geocode location geocode_result = geocode_location(location) if not geocode_result: return None lat, lon, full_name = geocode_result # Get weather data weather_data = get_weather_data(lat, lon) if not weather_data: return None # Get recommendations recommendations = get_agricultural_recommendations(weather_data) return { "latitude": lat, "longitude": lon, "location_name": full_name, "weather_data": weather_data, "recommendations": recommendations }