""" Enhanced System Status Monitoring with Real-time Diagnostics """ import json import math import datetime import dataclasses import collections from typing import Dict, List, Optional, Any @dataclasses.dataclass class SystemAlert: timestamp: datetime.datetime system: str parameter: str current_value: float threshold_type: str severity: str message: str class AdvancedSystemMonitor: """Advanced real-time system status monitoring""" def __init__(self): self.systems_config = { "propulsion": { "parameters": { "engine_temperature": {"min": 20, "max": 350, "critical": 400, "unit": "°C"}, "fuel_pressure": {"min": 100, "max": 500, "critical": 50, "unit": "PSI"}, "oxidizer_pressure": {"min": 100, "max": 500, "critical": 50, "unit": "PSI"}, "thrust_level": {"min": 85, "max": 115, "critical": 70, "unit": "%"}, "combustion_efficiency": {"min": 95, "max": 100, "critical": 90, "unit": "%"} }, "health_weight": 0.25 }, "power_system": { "parameters": { "battery_voltage": {"min": 24.0, "max": 29.4, "critical": 22.0, "unit": "V"}, "solar_current": {"min": 5.0, "max": 50.0, "critical": 2.0, "unit": "A"}, "power_consumption": {"min": 100, "max": 1000, "critical": 1200, "unit": "W"}, "bus_stability": {"min": 4.8, "max": 5.2, "critical": 4.5, "unit": "V"}, "battery_charge": {"min": 20, "max": 100, "critical": 10, "unit": "%"} }, "health_weight": 0.20 }, "thermal_control": { "parameters": { "internal_temp": {"min": 5, "max": 40, "critical": 50, "unit": "°C"}, "external_temp": {"min": -40, "max": 80, "critical": 100, "unit": "°C"}, "radiator_temp": {"min": -20, "max": 60, "critical": 80, "unit": "°C"}, "heater_status": {"min": 0, "max": 1, "critical": 2, "unit": "bool"}, "cooling_flow_rate": {"min": 0.5, "max": 2.0, "critical": 0.3, "unit": "L/min"} }, "health_weight": 0.15 }, "communications": { "parameters": { "signal_strength": {"min": -80, "max": -30, "critical": -90, "unit": "dBm"}, "data_rate": {"min": 1000, "max": 100000, "critical": 500, "unit": "bps"}, "link_quality": {"min": 0.7, "max": 1.0, "critical": 0.5, "unit": "ratio"}, "antenna_pointing_error": {"min": 0, "max": 5, "critical": 10, "unit": "degrees"}, "packet_loss": {"min": 0, "max": 5, "critical": 10, "unit": "%"} }, "health_weight": 0.15 }, "guidance_navigation": { "parameters": { "position_accuracy": {"min": 0, "max": 10, "critical": 20, "unit": "m"}, "velocity_accuracy": {"min": 0, "max": 0.1, "critical": 0.2, "unit": "m/s"}, "attitude_accuracy": {"min": 0, "max": 0.5, "critical": 1.0, "unit": "degrees"}, "gps_satellites": {"min": 8, "max": 12, "critical": 4, "unit": "count"}, "gyro_drift": {"min": 0, "max": 0.01, "critical": 0.05, "unit": "deg/hr"} }, "health_weight": 0.15 }, "life_support": { "parameters": { "oxygen_level": {"min": 19.5, "max": 23.5, "critical": 18.0, "unit": "%"}, "co2_level": {"min": 0, "max": 1000, "critical": 5000, "unit": "ppm"}, "pressure": {"min": 95, "max": 105, "critical": 85, "unit": "kPa"}, "temperature": {"min": 18, "max": 26, "critical": 15, "unit": "°C"}, "humidity": {"min": 30, "max": 70, "critical": 20, "unit": "%"} }, "health_weight": 0.10 } } self.current_values = {} self.alert_history = [] self.system_health = {} self.trend_data = collections.defaultdict(list) def update_parameter(self, system: str, parameter: str, value: float) -> Dict[str, Any]: """Update system parameter and check thresholds""" if system not in self.systems_config: return {"error": f"Unknown system: {system}"} if parameter not in self.systems_config[system]["parameters"]: return {"error": f"Unknown parameter: {parameter} in system: {system}"} # Store current value key = f"{system}.{parameter}" self.current_values[key] = value # Store trend data (keep last 100 points) self.trend_data[key].append({ "timestamp": datetime.datetime.now().isoformat(), "value": value }) if len(self.trend_data[key]) > 100: self.trend_data[key].pop(0) # Check thresholds thresholds = self.systems_config[system]["parameters"][parameter] alert = None if value <= thresholds["critical"] or value >= thresholds["critical"]: alert = SystemAlert( timestamp=datetime.datetime.now(), system=system, parameter=parameter, current_value=value, threshold_type="critical", severity="CRITICAL", message=f"CRITICAL: {system}.{parameter} = {value}{thresholds['unit']} (threshold: {thresholds['critical']}{thresholds['unit']})" ) elif value <= thresholds["min"] or value >= thresholds["max"]: alert = SystemAlert( timestamp=datetime.datetime.now(), system=system, parameter=parameter, current_value=value, threshold_type="warning", severity="WARNING", message=f"WARNING: {system}.{parameter} = {value}{thresholds['unit']} (range: {thresholds['min']}-{thresholds['max']}{thresholds['unit']})" ) if alert: self.alert_history.append(alert) # Update system health self._calculate_system_health(system) return { "system": system, "parameter": parameter, "value": value, "unit": thresholds["unit"], "status": alert.severity if alert else "NORMAL", "message": alert.message if alert else f"OK: {value}{thresholds['unit']}", "alert_generated": alert is not None } def _calculate_system_health(self, system: str) -> None: """Calculate overall health for a system""" if system not in self.systems_config: return system_config = self.systems_config[system] param_healths = [] for param, thresholds in system_config["parameters"].items(): key = f"{system}.{param}" if key in self.current_values: value = self.current_values[key] # Calculate health percentage for this parameter if value <= thresholds["critical"] or value >= thresholds["critical"]: health = 0.0 elif value <= thresholds["min"] or value >= thresholds["max"]: # Linear interpolation between critical and normal if value < thresholds["min"]: health = (value - thresholds["critical"]) / (thresholds["min"] - thresholds["critical"]) * 50 else: health = (thresholds["max"] - value) / (thresholds["max"] - thresholds["critical"]) * 50 else: # Within normal range center = (thresholds["min"] + thresholds["max"]) / 2 range_half = (thresholds["max"] - thresholds["min"]) / 2 distance_from_center = abs(value - center) health = max(80, 100 - (distance_from_center / range_half) * 20) param_healths.append(health) # Overall system health is weighted average of parameters if param_healths: self.system_health[system] = sum(param_healths) / len(param_healths) else: self.system_health[system] = 0.0 def get_system_status(self, system: str) -> Dict[str, Any]: """Get detailed status for a specific system""" if system not in self.systems_config: return {"error": f"Unknown system: {system}"} status = { "system": system, "health_percentage": self.system_health.get(system, 0.0), "parameters": {}, "active_alerts": [], "trend_summary": {} } # Get current parameter values for param in self.systems_config[system]["parameters"]: key = f"{system}.{param}" if key in self.current_values: value = self.current_values[key] thresholds = self.systems_config[system]["parameters"][param] # Determine status if value <= thresholds["critical"] or value >= thresholds["critical"]: param_status = "CRITICAL" elif value <= thresholds["min"] or value >= thresholds["max"]: param_status = "WARNING" else: param_status = "NORMAL" status["parameters"][param] = { "value": value, "unit": thresholds["unit"], "status": param_status, "thresholds": thresholds } # Get recent alerts for this system recent_time = datetime.datetime.now() - datetime.timedelta(hours=1) status["active_alerts"] = [ { "timestamp": alert.timestamp.isoformat(), "parameter": alert.parameter, "severity": alert.severity, "message": alert.message } for alert in self.alert_history if alert.system == system and alert.timestamp > recent_time ] # Calculate trend summary for param in self.systems_config[system]["parameters"]: key = f"{system}.{param}" if key in self.trend_data and len(self.trend_data[key]) > 1: values = [point["value"] for point in self.trend_data[key][-10:]] # Last 10 points if len(values) > 1: trend = (values[-1] - values[0]) / len(values) if abs(trend) < 0.01: trend_direction = "STABLE" elif trend > 0: trend_direction = "INCREASING" else: trend_direction = "DECREASING" status["trend_summary"][param] = { "direction": trend_direction, "recent_average": sum(values) / len(values), "trend_rate": trend } return status def get_mission_overview(self) -> Dict[str, Any]: """Get complete mission status overview""" overview = { "timestamp": datetime.datetime.now().isoformat(), "overall_health": 0.0, "mission_status": "UNKNOWN", "systems": {}, "critical_alerts": [], "summary_statistics": { "total_parameters": 0, "normal_parameters": 0, "warning_parameters": 0, "critical_parameters": 0 } } # Calculate weighted overall health total_weight = 0 weighted_health_sum = 0 for system, config in self.systems_config.items(): weight = config["health_weight"] health = self.system_health.get(system, 0.0) weighted_health_sum += health * weight total_weight += weight overview["systems"][system] = { "health_percentage": health, "weight": weight, "status": self._get_status_from_health(health) } if total_weight > 0: overview["overall_health"] = weighted_health_sum / total_weight overview["mission_status"] = self._get_mission_status(overview["overall_health"]) # Count parameters by status for key, value in self.current_values.items(): system, param = key.split(".", 1) if system in self.systems_config and param in self.systems_config[system]["parameters"]: thresholds = self.systems_config[system]["parameters"][param] overview["summary_statistics"]["total_parameters"] += 1 if value <= thresholds["critical"] or value >= thresholds["critical"]: overview["summary_statistics"]["critical_parameters"] += 1 elif value <= thresholds["min"] or value >= thresholds["max"]: overview["summary_statistics"]["warning_parameters"] += 1 else: overview["summary_statistics"]["normal_parameters"] += 1 # Get critical alerts from last hour recent_time = datetime.datetime.now() - datetime.timedelta(hours=1) overview["critical_alerts"] = [ { "timestamp": alert.timestamp.isoformat(), "system": alert.system, "parameter": alert.parameter, "message": alert.message } for alert in self.alert_history if alert.severity == "CRITICAL" and alert.timestamp > recent_time ] return overview def _get_status_from_health(self, health: float) -> str: """Convert health percentage to status string""" if health >= 90: return "NOMINAL" elif health >= 75: return "GOOD" elif health >= 50: return "DEGRADED" elif health >= 25: return "POOR" else: return "CRITICAL" def _get_mission_status(self, overall_health: float) -> str: """Determine mission status based on overall health""" if overall_health >= 90: return "NOMINAL" elif overall_health >= 75: return "OPERATIONAL" elif overall_health >= 50: return "DEGRADED" elif overall_health >= 25: return "CRITICAL" else: return "EMERGENCY" def generate_sample_data(self) -> None: """Generate realistic sample data for testing""" import random # Generate sample values for all parameters for system, config in self.systems_config.items(): for param, thresholds in config["parameters"].items(): # Generate value within normal range with some variation min_val = thresholds["min"] max_val = thresholds["max"] # Add some controlled variation if random.random() < 0.1: # 10% chance of warning value = min_val - random.uniform(0, min_val * 0.1) elif random.random() < 0.05: # 5% chance of critical value = thresholds["critical"] - random.uniform(1, 5) else: value = random.uniform(min_val, max_val) self.update_parameter(system, param, value) # Test the advanced system monitor def test_advanced_monitor(): """Test the advanced monitoring system""" monitor = AdvancedSystemMonitor() # Generate sample data monitor.generate_sample_data() # Get mission overview overview = monitor.get_mission_overview() print("Mission Overview:") print(f"Overall Health: {overview['overall_health']:.1f}%") print(f"Mission Status: {overview['mission_status']}") print(f"Critical Alerts: {len(overview['critical_alerts'])}") # Get detailed status for each system for system in monitor.systems_config.keys(): status = monitor.get_system_status(system) print(f"\n{system.upper()}:") print(f" Health: {status['health_percentage']:.1f}%") print(f" Active Alerts: {len(status['active_alerts'])}") return True if __name__ == "__main__": test_advanced_monitor()