import requests import json from datetime import datetime, timedelta from typing import Dict, List, Optional import logging from models import WeatherAlert, Farm, db # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class WeatherAlertService: """Service for generating weather-based alerts for farms""" def __init__(self, weather_api_key: str): self.weather_api_key = weather_api_key self.base_url = "http://api.openweathermap.org/data/2.5" def check_and_generate_alerts(self, farm_id: int) -> List[Dict]: """Check weather conditions for a farm and generate alerts if needed""" try: farm = Farm.query.get(farm_id) if not farm or not farm.latitude or not farm.longitude: logger.warning(f"Farm {farm_id} not found or coordinates missing") return [] # Get current weather and forecast current_weather = self._get_current_weather(farm.latitude, farm.longitude) forecast = self._get_weather_forecast(farm.latitude, farm.longitude) alerts = [] # Check for various weather conditions alerts.extend(self._check_rain_alerts(farm, current_weather, forecast)) alerts.extend(self._check_temperature_alerts(farm, current_weather, forecast)) alerts.extend(self._check_wind_alerts(farm, current_weather, forecast)) alerts.extend(self._check_humidity_alerts(farm, current_weather, forecast)) # Save alerts to database for alert_data in alerts: existing = WeatherAlert.query.filter_by( farm_id=farm_id, alert_type=alert_data['alert_type'], is_active=True ).first() if not existing: alert = WeatherAlert( farm_id=farm_id, alert_type=alert_data['alert_type'], severity=alert_data['severity'], title=alert_data['title'], message=alert_data['message'], recommended_action=alert_data['recommended_action'], weather_data=json.dumps(alert_data.get('weather_data', {})), expires_at=datetime.now() + timedelta(hours=24) ) db.session.add(alert) db.session.commit() return alerts except Exception as e: logger.error(f"Error generating weather alerts for farm {farm_id}: {str(e)}") return [] def _get_current_weather(self, lat: float, lon: float) -> Dict: """Get current weather data""" try: url = f"{self.base_url}/weather" params = { 'lat': lat, 'lon': lon, 'appid': self.weather_api_key, 'units': 'metric' } response = requests.get(url, params=params, timeout=10) response.raise_for_status() return response.json() except Exception as e: logger.error(f"Error fetching current weather: {str(e)}") return {} def _get_weather_forecast(self, lat: float, lon: float) -> Dict: """Get 5-day weather forecast""" try: url = f"{self.base_url}/forecast" params = { 'lat': lat, 'lon': lon, 'appid': self.weather_api_key, 'units': 'metric' } response = requests.get(url, params=params, timeout=10) response.raise_for_status() return response.json() except Exception as e: logger.error(f"Error fetching weather forecast: {str(e)}") return {} def _check_rain_alerts(self, farm: Farm, current: Dict, forecast: Dict) -> List[Dict]: """Check for rain-related alerts""" alerts = [] # Heavy rain alert if current.get('rain', {}).get('1h', 0) > 10: # More than 10mm/hour alerts.append({ 'alert_type': 'heavy_rain', 'severity': 'high', 'title': 'Heavy Rain Alert', 'message': f'Heavy rainfall detected at {farm.farm_name}. Current rate: {current["rain"]["1h"]:.1f}mm/hour', 'recommended_action': 'Ensure proper drainage, protect harvested crops, avoid field operations', 'weather_data': current }) # Forecast rain alert forecast_list = forecast.get('list', []) for item in forecast_list[:8]: # Next 24 hours if item.get('rain', {}).get('3h', 0) > 15: # More than 15mm in 3 hours alerts.append({ 'alert_type': 'rain_forecast', 'severity': 'medium', 'title': 'Heavy Rain Expected', 'message': f'Heavy rain expected at {farm.farm_name} in the next 24 hours', 'recommended_action': 'Prepare drainage, postpone irrigation, cover sensitive crops', 'weather_data': item }) break return alerts def _check_temperature_alerts(self, farm: Farm, current: Dict, forecast: Dict) -> List[Dict]: """Check for temperature-related alerts""" alerts = [] temp = current.get('main', {}).get('temp', 0) # Heat wave alert if temp > 40: alerts.append({ 'alert_type': 'heat_wave', 'severity': 'high', 'title': 'Extreme Heat Alert', 'message': f'Very high temperature at {farm.farm_name}: {temp:.1f}°C', 'recommended_action': 'Increase irrigation frequency, provide shade for livestock, avoid midday work', 'weather_data': current }) # Cold wave alert elif temp < 5: alerts.append({ 'alert_type': 'cold_wave', 'severity': 'high', 'title': 'Cold Wave Alert', 'message': f'Very low temperature at {farm.farm_name}: {temp:.1f}°C', 'recommended_action': 'Protect crops from frost, cover sensitive plants, check irrigation pipes', 'weather_data': current }) return alerts def _check_wind_alerts(self, farm: Farm, current: Dict, forecast: Dict) -> List[Dict]: """Check for wind-related alerts""" alerts = [] wind_speed = current.get('wind', {}).get('speed', 0) * 3.6 # Convert m/s to km/h if wind_speed > 50: # Strong winds alerts.append({ 'alert_type': 'strong_wind', 'severity': 'medium', 'title': 'Strong Wind Alert', 'message': f'Strong winds at {farm.farm_name}: {wind_speed:.1f} km/h', 'recommended_action': 'Secure loose structures, check support for tall crops, avoid spraying', 'weather_data': current }) return alerts def _check_humidity_alerts(self, farm: Farm, current: Dict, forecast: Dict) -> List[Dict]: """Check for humidity-related alerts""" alerts = [] humidity = current.get('main', {}).get('humidity', 0) # High humidity - disease risk if humidity > 85: alerts.append({ 'alert_type': 'high_humidity', 'severity': 'medium', 'title': 'Disease Risk Alert', 'message': f'High humidity at {farm.farm_name}: {humidity}% - Increased disease risk', 'recommended_action': 'Monitor crops for fungal diseases, improve ventilation, reduce watering', 'weather_data': current }) return alerts def get_weather_alerts(self, latitude: float, longitude: float) -> List[Dict]: """ Get weather alerts for a location Args: latitude: Farm latitude longitude: Farm longitude Returns: List of weather alert dictionaries """ try: # Get current weather and forecast for this location current_weather = self._get_current_weather(latitude, longitude) forecast = self._get_weather_forecast(latitude, longitude) alerts = [] # Generate alerts based on current conditions if current_weather: # Temperature alerts temp = current_weather.get('main', {}).get('temp', 0) if temp > 40: alerts.append({ 'alert_type': 'extreme_heat', 'severity': 'high', 'title': 'Extreme Heat Warning', 'message': f'Very high temperature: {temp}°C. Protect crops from heat stress.', 'recommendations': 'Increase irrigation, provide shade, harvest early morning', 'created_at': datetime.now().isoformat(), 'is_active': True }) elif temp < 5: alerts.append({ 'alert_type': 'frost_warning', 'severity': 'high', 'title': 'Frost Warning', 'message': f'Low temperature: {temp}°C. Risk of frost damage.', 'recommendations': 'Cover sensitive crops, use frost protection methods', 'created_at': datetime.now().isoformat(), 'is_active': True }) # Rain alerts rain = current_weather.get('rain', {}).get('1h', 0) if rain > 10: alerts.append({ 'alert_type': 'heavy_rain', 'severity': 'medium', 'title': 'Heavy Rain Alert', 'message': f'Heavy rainfall: {rain}mm/hr. Check drainage systems.', 'recommendations': 'Ensure proper drainage, avoid field operations', 'created_at': datetime.now().isoformat(), 'is_active': True }) # Wind alerts wind_speed = current_weather.get('wind', {}).get('speed', 0) * 3.6 # Convert m/s to km/h if wind_speed > 50: alerts.append({ 'alert_type': 'strong_wind', 'severity': 'medium', 'title': 'Strong Wind Warning', 'message': f'Strong winds: {wind_speed:.1f} km/h. Risk of crop damage.', 'recommendations': 'Secure equipment, check for crop lodging', 'created_at': datetime.now().isoformat(), 'is_active': True }) # Humidity alerts humidity = current_weather.get('main', {}).get('humidity', 0) if humidity > 85: alerts.append({ 'alert_type': 'high_humidity', 'severity': 'medium', 'title': 'High Humidity Alert', 'message': f'High humidity: {humidity}%. Increased disease risk.', 'recommendations': 'Monitor for fungal diseases, improve ventilation', 'created_at': datetime.now().isoformat(), 'is_active': True }) # Check forecast for upcoming conditions if forecast and 'list' in forecast: for item in forecast['list'][:8]: # Check next 24 hours forecast_temp = item.get('main', {}).get('temp', 0) forecast_rain = item.get('rain', {}).get('3h', 0) if forecast_temp > 42: alerts.append({ 'alert_type': 'upcoming_heat', 'severity': 'medium', 'title': 'Upcoming Heat Wave', 'message': f'High temperature expected: {forecast_temp}°C', 'recommendations': 'Prepare heat protection measures, plan irrigation', 'created_at': datetime.now().isoformat(), 'is_active': True }) break # Only add one forecast alert if forecast_rain > 15: alerts.append({ 'alert_type': 'upcoming_rain', 'severity': 'medium', 'title': 'Heavy Rain Expected', 'message': f'Heavy rain forecast: {forecast_rain}mm', 'recommendations': 'Postpone spraying, check drainage systems', 'created_at': datetime.now().isoformat(), 'is_active': True }) break # Only add one forecast alert return alerts except Exception as e: logger.error(f"Error getting weather alerts: {str(e)}") return [] def get_active_alerts(self, farm_id: int) -> List[WeatherAlert]: """Get all active alerts for a farm""" return WeatherAlert.query.filter_by( farm_id=farm_id, is_active=True ).filter( WeatherAlert.expires_at > datetime.now() ).order_by(WeatherAlert.created_at.desc()).all() def mark_alert_sent(self, alert_id: int): """Mark an alert as sent""" alert = WeatherAlert.query.get(alert_id) if alert: alert.is_sent = True db.session.commit()