#!/usr/bin/env python3 """ Weather Monitoring System for Launch Day Real-time weather assessment with automated alerts and forecasting. Skills: general engineering Type: implementation Version: 1.0 Author: Starlight Launch Systems """ import json import math import datetime import hashlib from typing import Dict, List, Optional, Any, Union from dataclasses import dataclass, asdict from enum import Enum class WeatherAlertLevel(Enum): NORMAL = "NORMAL" ADVISORY = "ADVISORY" WARNING = "WARNING" CRITICAL = "CRITICAL" @dataclass class WeatherAlert: alert_id: str level: WeatherAlertLevel parameter: str current_value: float threshold_value: float message: str timestamp: str active: bool = True @dataclass class ForecastData: time_hours_ahead: int expected_condition: str confidence: float # 0-100% parameters: Dict[str, float] class WeatherMonitoringSystem: def __init__(self): self.alerts = [] self.current_conditions = None self.forecast_history = [] self.monitoring_active = False # Weather thresholds for launch self.launch_thresholds = { 'wind_speed': {'min': 0, 'max': 15.0, 'unit': 'm/s'}, 'temperature': {'min': -20.0, 'max': 40.0, 'unit': '°C'}, 'humidity': {'max': 95.0, 'unit': '%'}, 'visibility': {'min': 5.0, 'unit': 'km'}, 'cloud_ceiling': {'min': 1500.0, 'unit': 'm'}, 'pressure': {'min': 980.0, 'max': 1040.0, 'unit': 'hPa'}, 'lightning_distance': {'min': 10.0, 'unit': 'km'}, 'precipitation': {'max': 2.0, 'unit': 'mm/hr'} } # Forecast parameters for next 24 hours self.forecast_hours = [1, 2, 3, 4, 6, 8, 12, 18, 24] def start_monitoring(self): """Start continuous weather monitoring.""" self.monitoring_active = True return {"status": "monitoring_started", "timestamp": datetime.datetime.now().isoformat()} def stop_monitoring(self): """Stop weather monitoring.""" self.monitoring_active = False return {"status": "monitoring_stopped", "timestamp": datetime.datetime.now().isoformat()} def update_conditions(self, conditions: Dict[str, Any]) -> Dict[str, Any]: """Update current weather conditions and check against thresholds.""" timestamp = datetime.datetime.now().isoformat() conditions['timestamp'] = timestamp self.current_conditions = conditions # Check all parameters against thresholds new_alerts = self._check_thresholds(conditions) # Add new alerts for alert in new_alerts: self.alerts.append(alert) return { "conditions_updated": True, "timestamp": timestamp, "new_alerts": len(new_alerts), "active_alerts": len([a for a in self.alerts if a.active]) } def _check_thresholds(self, conditions: Dict[str, Any]) -> List[WeatherAlert]: """Check conditions against all thresholds and generate alerts.""" alerts = [] for param, thresholds in self.launch_thresholds.items(): if param not in conditions: continue current_value = conditions[param] # Check minimum threshold if 'min' in thresholds and current_value < thresholds['min']: alert_id = hashlib.md5(f"{param}{current_value}{datetime.datetime.now().isoformat()}".encode()).hexdigest()[:8] alerts.append(WeatherAlert( alert_id=alert_id, level=WeatherAlertLevel.WARNING, parameter=param, current_value=current_value, threshold_value=thresholds['min'], message=f"{param.replace('_', ' ').title()} below minimum threshold: {current_value:.1f}{thresholds['unit']} < {thresholds['min']:.1f}{thresholds['unit']}", timestamp=datetime.datetime.now().isoformat() )) # Check maximum threshold if 'max' in thresholds and current_value > thresholds['max']: alert_id = hashlib.md5(f"{param}{current_value}{datetime.datetime.now().isoformat()}".encode()).hexdigest()[:8] alerts.append(WeatherAlert( alert_id=alert_id, level=WeatherAlertLevel.WARNING, parameter=param, current_value=current_value, threshold_value=thresholds['max'], message=f"{param.replace('_', ' ').title()} above maximum threshold: {current_value:.1f}{thresholds['unit']} > {thresholds['max']:.1f}{thresholds['unit']}", timestamp=datetime.datetime.now().isoformat() )) return alerts def generate_forecast(self, forecast_data: List[Dict[str, Any]]) -> Dict[str, Any]: """Process and store weather forecast data.""" forecasts = [] for data in forecast_data: hours_ahead = data.get('hours_ahead', 0) if hours_ahead not in self.forecast_hours: continue forecast = ForecastData( time_hours_ahead=hours_ahead, expected_condition=data.get('condition', 'unknown'), confidence=data.get('confidence', 50.0), parameters=data.get('parameters', {}) ) forecasts.append(forecast) # Store forecast history self.forecast_history.append({ 'timestamp': datetime.datetime.now().isoformat(), 'forecasts': [asdict(f) for f in forecasts] }) return { "forecast_generated": True, "forecasts_count": len(forecasts), "timestamp": datetime.datetime.now().isoformat() } def assess_launch_weather(self) -> Dict[str, Any]: """Assess current and forecasted weather for launch decision.""" if not self.current_conditions: return {"error": "No current weather data available"} # Check current conditions current_go = self._evaluate_conditions(self.current_conditions) # Check forecast conditions forecast_go = True critical_issues = [] for forecast_entry in self.forecast_history[-1]['forecasts'] if self.forecast_history else []: hours_ahead = forecast_entry['time_hours_ahead'] if hours_ahead <= 6: # Check next 6 hours conditions = forecast_entry['parameters'] if not self._evaluate_conditions(conditions): forecast_go = False critical_issues.append(f"Weather issues expected in {hours_ahead} hours") # Overall assessment if current_go and forecast_go: status = "GO" color = "green" elif current_go and not forecast_go: status = "CAUTION" color = "yellow" critical_issues.append("Forecast conditions deteriorating") else: status = "NO_GO" color = "red" return { "overall_status": status, "status_color": color, "current_conditions_go": current_go, "forecast_go": forecast_go, "active_alerts": len([a for a in self.alerts if a.active]), "critical_issues": critical_issues, "assessment_time": datetime.datetime.now().isoformat() } def _evaluate_conditions(self, conditions: Dict[str, Any]) -> bool: """Evaluate if conditions meet launch criteria.""" for param, thresholds in self.launch_thresholds.items(): if param not in conditions: continue value = conditions[param] if 'min' in thresholds and value < thresholds['min']: return False if 'max' in thresholds and value > thresholds['max']: return False return True def clear_alert(self, alert_id: str) -> bool: """Clear a specific weather alert.""" for alert in self.alerts: if alert.alert_id == alert_id: alert.active = False return True return False def clear_all_alerts(self): """Clear all weather alerts.""" for alert in self.alerts: alert.active = False return {"cleared_count": len(self.alerts)} def get_weather_summary(self) -> Dict[str, Any]: """Get comprehensive weather monitoring summary.""" active_alerts = [a for a in self.alerts if a.active] summary = { "monitoring_active": self.monitoring_active, "current_conditions": self.current_conditions, "active_alerts": { "total": len(active_alerts), "critical": len([a for a in active_alerts if a.level == WeatherAlertLevel.CRITICAL]), "warning": len([a for a in active_alerts if a.level == WeatherAlertLevel.WARNING]), "advisory": len([a for a in active_alerts if a.level == WeatherAlertLevel.ADVISORY]) }, "latest_forecast": self.forecast_history[-1] if self.forecast_history else None, "alerts_history": len(self.alerts), "forecast_updates": len(self.forecast_history) } return summary def main(): """Main execution function for weather monitoring system.""" print("🌤️ Initializing Starlight Weather Monitoring System") # Initialize weather monitoring weather_sys = WeatherMonitoringSystem() # Start monitoring weather_sys.start_monitoring() print("✅ Weather monitoring started") # Simulate updating current conditions current_weather = { 'wind_speed': 12.5, 'wind_direction': 270.0, 'temperature': 18.0, 'humidity': 75.0, 'visibility': 8.0, 'cloud_ceiling': 1800.0, 'pressure': 1015.0, 'lightning_distance': 25.0, 'precipitation': 0.0 } result = weather_sys.update_conditions(current_weather) print(f"📊 Weather conditions updated: {result['new_alerts']} new alerts") # Simulate forecast data forecast_data = [ {'hours_ahead': 1, 'condition': 'clear', 'confidence': 85.0, 'parameters': {'wind_speed': 11.0, 'visibility': 10.0}}, {'hours_ahead': 2, 'condition': 'partly_cloudy', 'confidence': 75.0, 'parameters': {'wind_speed': 13.0, 'visibility': 8.0}}, {'hours_ahead': 4, 'condition': 'cloudy', 'confidence': 60.0, 'parameters': {'wind_speed': 16.0, 'visibility': 6.0}}, {'hours_ahead': 6, 'condition': 'clear', 'confidence': 80.0, 'parameters': {'wind_speed': 9.0, 'visibility': 12.0}} ] forecast_result = weather_sys.generate_forecast(forecast_data) print(f"🔮 Forecast generated: {forecast_result['forecasts_count']} predictions") # Assess launch weather assessment = weather_sys.assess_launch_weather() print(f"🚀 Launch Weather Assessment: {assessment['overall_status']} ({assessment['status_color']})") # Get summary summary = weather_sys.get_weather_summary() print(f"📋 Active Alerts: {summary['active_alerts']['total']} total") # Save monitoring data monitoring_data = { "summary": summary, "assessment": assessment, "timestamp": datetime.datetime.now().isoformat() } with open("weather_monitoring_data.json", "w") as f: json.dump(monitoring_data, f, indent=2) print("💾 Weather monitoring data saved to weather_monitoring_data.json") print("🎯 Weather Monitoring System Complete!") if __name__ == "__main__": main()