""" Post-Launch Monitoring System Executive Summary: Complete monitoring framework for system launch verification with automated checks, alerting, and contingency procedures. """ ## Implementation Code ### Core Monitoring System import json import math import datetime import hashlib from typing import Dict, List, Optional, Any from dataclasses import dataclass from enum import Enum class AlertLevel(Enum): INFO = "info" WARNING = "warning" CRITICAL = "critical" EMERGENCY = "emergency" class SystemStatus(Enum): NOMINAL = "nominal" DEGRADED = "degraded" CRITICAL = "critical" OFFLINE = "offline" @dataclass class MonitoringCheck: name: str check_type: str threshold: float current_value: float status: str timestamp: datetime.datetime alert_level: AlertLevel class PostLaunchMonitor: def __init__(self): self.checks = [] self.system_status = SystemStatus.NOMINAL self.abort_conditions = [] self.contingency_plans = {} def trajectory_verification(self, position: Dict, velocity: Dict, expected_trajectory: Dict) -> Dict: """Verify trajectory against expected parameters.""" # Calculate position deviation pos_deviation = math.sqrt( (position['x'] - expected_trajectory['x'])**2 + (position['y'] - expected_trajectory['y'])**2 + (position['z'] - expected_trajectory['z'])**2 ) # Calculate velocity deviation vel_deviation = math.sqrt( (velocity['vx'] - expected_trajectory['vx'])**2 + (velocity['vy'] - expected_trajectory['vy'])**2 + (velocity['vz'] - expected_trajectory['vz'])**2 ) # Determine status pos_threshold = 100.0 # meters vel_threshold = 10.0 # m/s if pos_deviation > pos_threshold * 3 or vel_deviation > vel_threshold * 3: status = "CRITICAL_DEVIATION" alert_level = AlertLevel.EMERGENCY elif pos_deviation > pos_threshold or vel_deviation > vel_threshold: status = "MINOR_DEVIATION" alert_level = AlertLevel.WARNING else: status = "NOMINAL" alert_level = AlertLevel.INFO check = MonitoringCheck( name="Trajectory Verification", check_type="trajectory", threshold=pos_threshold, current_value=pos_deviation, status=status, timestamp=datetime.datetime.now(), alert_level=alert_level ) self.checks.append(check) return { "status": status, "position_deviation": pos_deviation, "velocity_deviation": vel_deviation, "within_tolerance": pos_deviation <= pos_threshold and vel_deviation <= vel_threshold, "recommendation": self._get_trajectory_recommendation(status) } def system_health_check(self, metrics: Dict) -> Dict: """Monitor critical system parameters.""" health_metrics = {} overall_status = SystemStatus.NOMINAL critical_issues = [] # Check CPU usage cpu_usage = metrics.get('cpu_usage', 0) if cpu_usage > 90: health_metrics['cpu'] = {"status": "CRITICAL", "value": cpu_usage} critical_issues.append("CPU overload") overall_status = SystemStatus.CRITICAL elif cpu_usage > 75: health_metrics['cpu'] = {"status": "WARNING", "value": cpu_usage} overall_status = SystemStatus.DEGRADED else: health_metrics['cpu'] = {"status": "OK", "value": cpu_usage} # Check memory usage memory_usage = metrics.get('memory_usage', 0) if memory_usage > 90: health_metrics['memory'] = {"status": "CRITICAL", "value": memory_usage} critical_issues.append("Memory overflow") overall_status = SystemStatus.CRITICAL elif memory_usage > 80: health_metrics['memory'] = {"status": "WARNING", "value": memory_usage} overall_status = SystemStatus.DEGRADED else: health_metrics['memory'] = {"status": "OK", "value": memory_usage} # Check temperature temperature = metrics.get('temperature', 20) if temperature > 85: health_metrics['temperature'] = {"status": "CRITICAL", "value": temperature} critical_issues.append("Overheating") overall_status = SystemStatus.CRITICAL elif temperature > 75: health_metrics['temperature'] = {"status": "WARNING", "value": temperature} overall_status = SystemStatus.DEGRADED else: health_metrics['temperature'] = {"status": "OK", "value": temperature} # Check connection status connectivity = metrics.get('connectivity', True) health_metrics['connectivity'] = {"status": "OK" if connectivity else "CRITICAL"} if not connectivity: critical_issues.append("Connection lost") overall_status = SystemStatus.CRITICAL check = MonitoringCheck( name="System Health Check", check_type="health", threshold=0, current_value=len(critical_issues), status=overall_status.value, timestamp=datetime.datetime.now(), alert_level=AlertLevel.CRITICAL if critical_issues else AlertLevel.INFO ) self.checks.append(check) return { "overall_status": overall_status.value, "metrics": health_metrics, "critical_issues": critical_issues, "system_stable": len(critical_issues) == 0 } def abort_procedure_check(self, conditions: Dict) -> Dict: """Evaluate abort conditions and trigger procedures if needed.""" abort_triggered = False abort_reasons = [] # Check critical abort conditions if conditions.get('trajectory_deviation_critical', False): abort_triggered = True abort_reasons.append("Critical trajectory deviation") if conditions.get('system_failure', False): abort_triggered = True abort_reasons.append("System failure") if conditions.get('communication_lost', False): abort_triggered = True abort_reasons.append("Communication lost") if conditions.get('safety_violation', False): abort_triggered = True abort_reasons.append("Safety boundary violation") abort_status = { "abort_triggered": abort_triggered, "abort_reasons": abort_reasons, "abort_time": datetime.datetime.now().isoformat() if abort_triggered else None, "procedure": self._get_abort_procedure(abort_reasons) if abort_triggered else None } check = MonitoringCheck( name="Abort Procedure Check", check_type="abort", threshold=0, current_value=len(abort_reasons), status="ABORT" if abort_triggered else "CONTINUE", timestamp=datetime.datetime.now(), alert_level=AlertLevel.EMERGENCY if abort_triggered else AlertLevel.INFO ) self.checks.append(check) return abort_status def contingency_planner(self, failure_scenarios: List[str]) -> Dict: """Generate contingency plans for various failure scenarios.""" plans = {} for scenario in failure_scenarios: plan = self._generate_contingency_plan(scenario) plans[scenario] = plan return { "scenarios_analyzed": len(failure_scenarios), "contingency_plans": plans, "readiness_score": self._calculate_readiness_score(plans) } def _get_trajectory_recommendation(self, status: str) -> str: recommendations = { "NOMINAL": "Continue monitoring. Trajectory within acceptable limits.", "MINOR_DEVIATION": "Execute course correction procedures. Increase monitoring frequency.", "CRITICAL_DEVIATION": "IMMEDIATE ACTION REQUIRED. Execute emergency trajectory correction protocol." } return recommendations.get(status, "Unknown status - investigate immediately.") def _get_abort_procedure(self, reasons: List[str]) -> Dict: return { "procedure_id": f"ABORT-{hashlib.md5(','.join(reasons).encode()).hexdigest()[:8]}", "steps": [ "1. Secure all critical systems", "2. Notify mission control", "3. Execute safe shutdown sequence", "4. Activate recovery protocols", "5. Log all events and conditions" ], "estimated_time": "30 seconds", "safety_checks": ["System isolation", "Power cutoff", "Data backup"] } def _generate_contingency_plan(self, scenario: str) -> Dict: base_template = { "scenario": scenario, "severity": "medium", "response_time": "5 minutes", "required_actions": [], "fallback_options": [] } if "communication" in scenario.lower(): base_template.update({ "severity": "high", "response_time": "2 minutes", "required_actions": [ "Switch to backup communication channel", "Initiate autonomous operation mode", "Buffer critical data" ], "fallback_options": [ "Activate emergency beacon", "Execute pre-programmed mission profile" ] }) elif "power" in scenario.lower(): base_template.update({ "severity": "critical", "response_time": "1 minute", "required_actions": [ "Switch to backup power systems", "Shed non-essential loads", "Conserve battery power" ], "fallback_options": [ "Enter safe mode", "Activate solar panels if available" ] }) elif "navigation" in scenario.lower(): base_template.update({ "severity": "high", "response_time": "3 minutes", "required_actions": [ "Switch to backup navigation system", "Use stellar navigation", "Maintain stable orientation" ], "fallback_options": [ "Hold position", "Wait for ground intervention" ] }) return base_template def _calculate_readiness_score(self, plans: Dict) -> float: if not plans: return 0.0 total_score = 0 for plan in plans.values(): severity_weights = {"low": 1.0, "medium": 0.8, "high": 0.6, "critical": 0.4} severity = plan.get("severity", "medium") total_score += severity_weights.get(severity, 0.5) return (total_score / len(plans)) * 100 def generate_monitoring_report(self) -> Dict: """Generate comprehensive monitoring status report.""" now = datetime.datetime.now() recent_checks = [c for c in self.checks if (now - c.timestamp).total_seconds() < 300] critical_alerts = [c for c in recent_checks if c.alert_level in [AlertLevel.CRITICAL, AlertLevel.EMERGENCY]] warnings = [c for c in recent_checks if c.alert_level == AlertLevel.WARNING] report = { "report_timestamp": now.isoformat(), "system_status": self.system_status.value, "total_checks_performed": len(self.checks), "recent_checks": len(recent_checks), "critical_alerts": len(critical_alerts), "warnings": len(warnings), "overall_health": "HEALTHY" if len(critical_alerts) == 0 else "CRITICAL", "recommendations": self._generate_recommendations(critical_alerts, warnings) } return report def _generate_recommendations(self, critical: List, warnings: List) -> List[str]: recommendations = [] if critical: recommendations.append("IMMEDIATE: Address all critical alerts before proceeding") recommendations.append("Execute emergency protocols for critical systems") if warnings: recommendations.append("Monitor warning conditions closely") recommendations.append("Prepare contingency procedures for warning scenarios") if not critical and not warnings: recommendations.append("System operating nominally - continue standard monitoring") return recommendations # Usage Example and Test def test_monitoring_system(): """Test the complete monitoring system.""" monitor = PostLaunchMonitor() # Test trajectory verification position = {"x": 100050, "y": 200010, "z": 300020} velocity = {"vx": 7500, "vy": 8200, "vz": 9100} expected = {"x": 100000, "y": 200000, "z": 300000, "vx": 7500, "vy": 8200, "vz": 9100} trajectory_result = monitor.trajectory_verification(position, velocity, expected) print("Trajectory Check:", trajectory_result) # Test system health system_metrics = { "cpu_usage": 45, "memory_usage": 67, "temperature": 65, "connectivity": True } health_result = monitor.system_health_check(system_metrics) print("Health Check:", health_result) # Test abort procedures abort_conditions = { "trajectory_deviation_critical": False, "system_failure": False, "communication_lost": False, "safety_violation": False } abort_result = monitor.abort_procedure_check(abort_conditions) print("Abort Check:", abort_result) # Test contingency planning failure_scenarios = [ "Communication system failure", "Power system malfunction", "Navigation system error", "Thermal control issue" ] contingency_result = monitor.contingency_planner(failure_scenarios) print("Contingency Plans Generated:", contingency_result["scenarios_analyzed"]) # Generate final report report = monitor.generate_monitoring_report() print("Final Report:", json.dumps(report, indent=2)) return { "test_passed": True, "trajectory_status": trajectory_result["within_tolerance"], "system_healthy": health_result["system_stable"], "abort_triggered": abort_result["abort_triggered"], "contingency_readiness": contingency_result["readiness_score"] } if __name__ == "__main__": test_result = test_monitoring_system() print("\n=== TEST RESULTS ===") print(json.dumps(test_result, indent=2))