AI-Powered Emergency Response: How Robots Became Better at Crisis Management Than Your Panic-Prone Coworkers

import asyncio
import json
import logging
from datetime import datetime
from typing import Dict, List, Optional

class EmergencyAI:
    """
    AI-powered emergency response system that monitors sensors
    and automatically triggers crisis management protocols
    """
    
    def __init__(self):
        self.sensors: Dict[str, float] = {}
        self.emergency_protocols: Dict[str, callable] = {}
        self.alert_history: List[Dict] = []
        self.is_monitoring = False
        
        # Configure logging for emergency events
        logging.basicConfig(level=logging.INFO)
        self.logger = logging.getLogger("EmergencyAI")
        
    def register_sensor(self, sensor_id: str, initial_value: float = 0.0):
        """Register a new sensor for monitoring"""
        self.sensors[sensor_id] = initial_value
        self.logger.info(f"Registered sensor: {sensor_id}")
        
    def add_protocol(self, emergency_type: str, protocol_function: callable):
        """Add emergency response protocol for specific crisis types"""
        self.emergency_protocols[emergency_type] = protocol_function
        self.logger.info(f"Added protocol for: {emergency_type}")
    
    async def monitor_conditions(self):
        """Continuously monitor all registered sensors"""
        self.is_monitoring = True
        self.logger.info("Emergency monitoring started")
        
        while self.is_monitoring:
            # Simulate reading sensor data (replace with actual sensor APIs)
            current_conditions = await self.read_all_sensors()
            
            # Analyze conditions using ML model (simplified for demo)
            threat_assessment = self.analyze_threats(current_conditions)
            
            # Execute emergency protocols if threats detected
            if threat_assessment["severity"] > 0.7:
                await self.execute_emergency_response(threat_assessment)
            
            # Wait before next monitoring cycle
            await asyncio.sleep(1)
    
    async def read_all_sensors(self) -> Dict[str, float]:
        """Read current values from all registered sensors"""
        # In real implementation, this would query actual IoT sensors
        import random
        
        current_readings = {}
        for sensor_id in self.sensors:
            # Simulate sensor readings with some randomness
            base_value = self.sensors[sensor_id]
            current_readings[sensor_id] = base_value + random.uniform(-0.1, 0.1)
            
        return current_readings
    
    def analyze_threats(self, sensor_data: Dict[str, float]) -> Dict[str, any]:
        """
        AI threat analysis - in production, this would use trained ML models
        For demo, we use simple threshold-based detection
        """
        threats = []
        max_severity = 0.0
        
        # Fire detection logic
        if sensor_data.get("temperature", 0) > 80:  # Celsius
            severity = min(1.0, (sensor_data["temperature"] - 80) / 50)
            threats.append({"type": "fire", "severity": severity})
            max_severity = max(max_severity, severity)
        
        # Smoke detection
        if sensor_data.get("smoke_level", 0) > 0.3:
            severity = min(1.0, sensor_data["smoke_level"])
            threats.append({"type": "fire", "severity": severity})
            max_severity = max(max_severity, severity)
        
        # Gas leak detection
        if sensor_data.get("gas_concentration", 0) > 0.1:
            severity = min(1.0, sensor_data["gas_concentration"] * 2)
            threats.append({"type": "gas_leak", "severity": severity})
            max_severity = max(max_severity, severity)
        
        return {
            "threats": threats,
            "severity": max_severity,
            "timestamp": datetime.now().isoformat(),
            "sensor_data": sensor_data
        }
    
    async def execute_emergency_response(self, threat_assessment: Dict):
        """Execute automated emergency response protocols"""
        self.logger.warning(f"EMERGENCY DETECTED: {threat_assessment}")
        
        # Log the emergency
        self.alert_history.append(threat_assessment)
        
        # Execute protocols for each detected threat
        for threat in threat_assessment["threats"]:
            threat_type = threat["type"]
            
            if threat_type in self.emergency_protocols:
                await self.emergency_protocols[threat_type](threat, threat_assessment)
                self.logger.info(f"Executed protocol for {threat_type}")
            else:
                await self.default_emergency_protocol(threat, threat_assessment)
    
    async def default_emergency_protocol(self, threat: Dict, assessment: Dict):
        """Default emergency response when no specific protocol exists"""
        actions = [
            "Send alert to emergency services",
            "Notify building security",
            "Activate emergency lighting",
            "Begin evacuation announcement"
        ]
        
        for action in actions:
            self.logger.info(f"EMERGENCY ACTION: {action}")
            await asyncio.sleep(0.1)  # Simulate action execution time

# Example emergency protocols
async def fire_response_protocol(threat: Dict, assessment: Dict):
    """Automated fire response protocol"""
    severity = threat["severity"]
    
    actions = [
        "Activate fire suppression system",
        "Sound fire alarm",
        "Call fire department",
        "Unlock emergency exits",
        "Activate emergency lighting",
        "Begin evacuation procedures"
    ]
    
    # Execute actions based on severity
    actions_to_execute = int(len(actions) * severity)
    for action in actions[:actions_to_execute]:
        print(f"🔥 FIRE PROTOCOL: {action}")
        await asyncio.sleep(0.2)

async def gas_leak_protocol(threat: Dict, assessment: Dict):
    """Automated gas leak response protocol"""
    actions = [
        "Shut off main gas valve",
        "Activate ventilation systems",
        "Evacuate affected areas",
        "Call hazmat team",
        "Disable electrical systems in affected zones"
    ]
    
    for action in actions:
        print(f"⚠️ GAS LEAK PROTOCOL: {action}")
        await asyncio.sleep(0.2)

# Usage example
async def main():
    """Demonstrate AI-powered emergency response system"""
    
    # Initialize emergency AI system
    emergency_ai = EmergencyAI()
    
    # Register sensors (in real system, these would be IoT devices)
    emergency_ai.register_sensor("temperature", 22.0)  # Celsius
    emergency_ai.register_sensor("smoke_level", 0.0)   # 0-1 scale
    emergency_ai.register_sensor("gas_concentration", 0.0)  # ppm
    
    # Add emergency protocols
    emergency_ai.add_protocol("fire", fire_response_protocol)
    emergency_ai.add_protocol("gas_leak", gas_leak_protocol)
    
    print("🤖 Emergency AI System Online")
    print("Monitoring building conditions...")
    
    # Simulate emergency condition after 5 seconds
    async def simulate_emergency():
        await asyncio.sleep(5)
        print("\n🚨 SIMULATING FIRE EMERGENCY")
        emergency_ai.sensors["temperature"] = 85.0  # High temperature
        emergency_ai.sensors["smoke_level"] = 0.8   # High smoke
        
        await asyncio.sleep(10)  # Let it respond
        emergency_ai.is_monitoring = False
    
    # Run monitoring and simulation concurrently
    await asyncio.gather(
        emergency_ai.monitor_conditions(),
        simulate_emergency()
    )

# Run the demo
if __name__ == "__main__":
    asyncio.run(main())

class CityFireResponseAI {
    constructor(cityConfig) {
        this.fireStations = cityConfig.fireStations;
        this.trafficSystems = cityConfig.trafficSystems;
        this.emergencyServices = cityConfig.emergencyServices;
    }
    
    async detectAndRespond(fireLocation, severity) {
        // AI calculates optimal response strategy
        const response = await this.optimizeResponse({
            location: fireLocation,
            severity: severity,
            availableUnits: this.getAvailableUnits(),
            trafficConditions: await this.getTrafficData(),
            weatherConditions: await this.getWeatherData()
        });
        
        // Execute coordinated response
        return await this.executeCoordinatedResponse(response);
    }
    
    async executeCoordinatedResponse(responseplan) {
        const actions = await Promise.all([
            this.dispatchFireTrucks(responsePlan.units),
            this.clearTrafficRoutes(responsePlan.routes),
            this.notifyHospitals(responsePlan.casualties),
            this.coordinateEvacuation(responsePlan.evacuationZones)
        ]);
        
        return {
            responseTime: responsePlan.estimatedArrival,
            unitsDispatched: responsePlan.units.length,
            evacuationStatus: actions[3].status
        };
    }
}

class HospitalEmergencyAI:
    """
    AI system for hospital emergency management
    Handles patient flow, resource allocation, and crisis response
    """
    
    def __init__(self):
        self.patient_queue = []
        self.available_resources = {
            "emergency_rooms": 12,
            "icu_beds": 8,
            "operating_rooms": 6,
            "medical_staff": 24
        }
    
    def triage_patient(self, patient_data):
        """AI-powered patient triage using severity prediction"""
        
        # Machine learning model predicts severity (simplified for demo)
        severity_score = self.predict_severity(patient_data)
        
        # Assign priority based on AI assessment
        if severity_score > 0.9:
            priority = "CRITICAL"
            estimated_wait = 0
        elif severity_score > 0.7:
            priority = "URGENT"
            estimated_wait = self.calculate_wait_time("urgent")
        else:
            priority = "ROUTINE"
            estimated_wait = self.calculate_wait_time("routine")
            
        return {
            "patient_id": patient_data["id"],
            "priority": priority,
            "severity_score": severity_score,
            "estimated_wait": estimated_wait,
            "recommended_department": self.recommend_department(patient_data)
        }
    
    def predict_severity(self, patient_data):
        """Simplified ML model for patient severity prediction"""
        # In production, this would use trained neural networks
        # analyzing vital signs, symptoms, medical history, etc.
        
        factors = {
            "vital_signs_abnormal": 0.3,
            "critical_symptoms": 0.4,
            "age_risk": 0.1,
            "medical_history": 0.2
        }
        
        # Simulate ML prediction
        severity = 0.0
        for factor, weight in factors.items():
            if patient_data.get(factor, False):
                severity += weight
                
        return min(1.0, severity + patient_data.get("pain_level", 0) * 0.1)

import numpy as np
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler

class CrisisPredictionAI:
    """
    Predictive AI system that forecasts potential emergencies
    Uses historical data and current conditions to prevent crises
    """
    
    def __init__(self):
        self.model = RandomForestClassifier(n_estimators=100, random_state=42)
        self.scaler = StandardScaler()
        self.is_trained = False
        self.feature_columns = [
            'temperature', 'humidity', 'pressure', 'wind_speed',
            'electrical_load', 'structural_stress', 'human_traffic',
            'time_of_day', 'day_of_week', 'seasonal_factor'
        ]
    
    def train_prediction_model(self, historical_data, emergency_labels):
        """Train ML model on historical emergency data"""
        
        # Prepare features
        features = historical_data[self.feature_columns]
        scaled_features = self.scaler.fit_transform(features)
        
        # Train the model
        self.model.fit(scaled_features, emergency_labels)
        self.is_trained = True
        
        # Calculate accuracy
        accuracy = self.model.score(scaled_features, emergency_labels)
        print(f"🎯 Model trained with {accuracy:.2%} accuracy")
        
        return accuracy
    
    def predict_emergency_risk(self, current_conditions):
        """Predict likelihood of emergency in next time window"""
        
        if not self.is_trained:
            raise ValueError("Model must be trained before making predictions")
        
        # Prepare current conditions for prediction
        features = np.array([[
            current_conditions.get(col, 0) for col in self.feature_columns
        ]])
        
        scaled_features = self.scaler.transform(features)
        
        # Get prediction probability
        risk_probability = self.model.predict_proba(scaled_features)[0][1]
        
        # Get feature importance for explanation
        feature_importance = dict(zip(
            self.feature_columns, 
            self.model.feature_importances_
        ))
        
        return {
            "emergency_risk": risk_probability,
            "risk_level": self.categorize_risk(risk_probability),
            "key_factors": sorted(
                feature_importance.items(), 
                key=lambda x: x[1], 
                reverse=True
            )[:3],
            "recommendation": self.get_prevention_recommendation(
                risk_probability, current_conditions
            )
        }
    
    def categorize_risk(self, probability):
        """Convert probability to risk category"""
        if probability > 0.8:
            return "CRITICAL"
        elif probability > 0.6:
            return "HIGH"
        elif probability > 0.4:
            return "MODERATE"
        else:
            return "LOW"
    
    def get_prevention_recommendation(self, risk_prob, conditions):
        """AI-generated recommendations to prevent predicted emergencies"""
        
        if risk_prob > 0.7:
            return [
                "Increase monitoring frequency to every 30 seconds",
                "Deploy additional response units to standby",
                "Notify emergency services of elevated risk",
                "Consider temporary evacuation of high-risk areas"
            ]
        elif risk_prob > 0.5:
            return [
                "Increase sensor monitoring",
                "Alert maintenance teams",
                "Review emergency protocols with staff"
            ]
        else:
            return ["Continue normal monitoring"]

# Example usage
def demonstrate_crisis_prediction():
    """Show how AI predicts and prevents emergencies"""
    
    # Initialize prediction system
    crisis_ai = CrisisPredictionAI()
    
    # Generate sample training data (in real system, use historical records)
    np.random.seed(42)
    n_samples = 1000
    
    # Create synthetic historical data
    historical_data = pd.DataFrame({
        'temperature': np.random.normal(25, 10, n_samples),
        'humidity': np.random.normal(60, 20, n_samples),
        'pressure': np.random.normal(1013, 10, n_samples),
        'wind_speed': np.random.exponential(5, n_samples),
        'electrical_load': np.random.normal(75, 15, n_samples),
        'structural_stress': np.random.normal(30, 10, n_samples),
        'human_traffic': np.random.poisson(100, n_samples),
        'time_of_day': np.random.randint(0, 24, n_samples),
        'day_of_week': np.random.randint(0, 7, n_samples),
        'seasonal_factor': np.random.normal(0.5, 0.2, n_samples)
    })
    
    # Create emergency labels (simplified logic)
    emergency_labels = (
        (historical_data['temperature'] > 35) |
        (historical_data['electrical_load'] > 90) |
        (historical_data['structural_stress'] > 45)
    ).astype(int)
    
    # Train the model
    accuracy = crisis_ai.train_prediction_model(historical_data, emergency_labels)
    
    # Test prediction with current conditions
    current_conditions = {
        'temperature': 38,  # High temperature
        'humidity': 45,
        'pressure': 1010,
        'wind_speed': 12,
        'electrical_load': 88,  # High electrical load
        'structural_stress': 42,
        'human_traffic': 150,
        'time_of_day': 14,
        'day_of_week': 3,
        'seasonal_factor': 0.7
    }
    
    prediction = crisis_ai.predict_emergency_risk(current_conditions)
    
    print(f"\n🔮 CRISIS PREDICTION RESULTS:")
    print(f"Emergency Risk: {prediction['emergency_risk']:.2%}")
    print(f"Risk Level: {prediction['risk_level']}")
    print(f"Key Risk Factors:")
    for factor, importance in prediction['key_factors']:
        print(f"  - {factor}: {importance:.3f}")
    
    print(f"\n💡 AI Recommendations:")
    for rec in prediction['recommendation']:
        print(f"  • {rec}")

# Run the demonstration
demonstrate_crisis_prediction()

import asyncio
import json
from typing import Dict, List
import paho.mqtt.client as mqtt

class IoTEmergencyNetwork:
    """
    IoT sensor network integration for emergency response
    Handles real-time data from multiple sensor types
    """
    
    def __init__(self, mqtt_broker="localhost", mqtt_port=1883):
        self.mqtt_client = mqtt.Client()
        self.mqtt_broker = mqtt_broker
        self.mqtt_port = mqtt_port
        self.sensors = {}
        self.emergency_thresholds = {}
        self.response_callbacks = {}
        
        # Setup MQTT callbacks
        self.mqtt_client.on_connect = self.on_mqtt_connect
        self.mqtt_client.on_message = self.on_mqtt_message
    
    def register_sensor_type(self, sensor_type: str, emergency_threshold: float, 
                           response_callback: callable):
        """Register new sensor type with emergency thresholds"""
        self.emergency_thresholds[sensor_type] = emergency_threshold
        self.response_callbacks[sensor_type] = response_callback
        print(f"📡 Registered sensor type: {sensor_type}")
    
    def on_mqtt_connect(self, client, userdata, flags, rc):
        """Callback for MQTT connection"""
        if rc == 0:
            print("🌐 Connected to IoT sensor network")
            # Subscribe to all sensor topics
            client.subscribe("sensors/+/+")  # sensors/{building}/{sensor_type}
        else:
            print(f"❌ Failed to connect to MQTT broker: {rc}")
    
    def on_mqtt_message(self, client, userdata, msg):
        """Process incoming sensor data"""
        try:
            # Parse topic: sensors/building_id/sensor_type
            topic_parts = msg.topic.split('/')
            building_id = topic_parts[1]
            sensor_type = topic_parts[2]
            
            # Parse sensor data
            sensor_data = json.loads(msg.payload.decode())
            
            # Store sensor reading
            sensor_key = f"{building_id}_{sensor_type}"
            self.sensors[sensor_key] = {
                "building": building_id,
                "type": sensor_type,
                "value": sensor_data["value"],
                "timestamp": sensor_data["timestamp"],
                "location": sensor_data.get("location", "unknown")
            }
            
            # Check for emergency conditions
            await self.check_emergency_conditions(sensor_key)
            
        except Exception as e:
            print(f"⚠️ Error processing sensor data: {e}")
    
    async def check_emergency_conditions(self, sensor_key: str):
        """Check if sensor reading indicates emergency"""
        sensor_data = self.sensors[sensor_key]
        sensor_type = sensor_data["type"]
        sensor_value = sensor_data["value"]
        
        # Check against emergency threshold
        if sensor_type in self.emergency_thresholds:
            threshold = self.emergency_thresholds[sensor_type]
            
            if sensor_value > threshold:
                print(f"🚨 EMERGENCY: {sensor_type} reading {sensor_value} exceeds threshold {threshold}")
                
                # Execute emergency response callback
                if sensor_type in self.response_callbacks:
                    await self.response_callbacks[sensor_type](sensor_data)
    
    async def start_monitoring(self):
        """Start IoT network monitoring"""
        self.mqtt_client.connect(self.mqtt_broker, self.mqtt_port, 60)
        self.mqtt_client.loop_start()
        print("🔍 IoT emergency monitoring started")

# Emergency response callbacks for different sensor types
async def temperature_emergency_response(sensor_data):
    """Handle high temperature emergency"""
    building = sensor_data["building"]
    location = sensor_data["location"]
    temperature = sensor_data["value"]
    
    actions = [
        f"🔥 Fire alert in {building} at {location}",
        f"🚨 Temperature: {temperature}°C detected",
        "📞 Alerting fire department",
        "🔊 Activating evacuation alarms",
        "🚪 Unlocking emergency exits"
    ]
    
    for action in actions:
        print(f"FIRE RESPONSE: {action}")
        await asyncio.sleep(0.5)

async def gas_emergency_response(sensor_data):
    """Handle gas leak emergency"""
    building = sensor_data["building"]
    concentration = sensor_data["value"]
    
    actions = [
        f"⚠️ Gas leak detected in {building}",
        f"📊 Concentration: {concentration} ppm",
        "🔧 Shutting off gas valves",
        "💨 Activating ventilation systems",
        "📱 Notifying hazmat team"
    ]
    
    for action in actions:
        print(f"GAS RESPONSE: {action}")
        await asyncio.sleep(0.3)

# Example usage
async def demo_iot_emergency_network():
    """Demonstrate IoT-integrated emergency response"""
    
    # Initialize IoT network
    iot_network = IoTEmergencyNetwork()
    
    # Register sensor types with emergency thresholds
    iot_network.register_sensor_type("temperature", 60.0, temperature_emergency_response)
    iot_network.register_sensor_type("gas", 100.0, gas_emergency_response)  # ppm
    iot_network.register_sensor_type("smoke", 0.5, temperature_emergency_response)
    
    # Start monitoring
    await iot_network.start_monitoring()
    
    # Keep running
    while True:
        await asyncio.sleep(1)

# Run the IoT emergency network demo
# asyncio.run(demo_iot_emergency_network())

import time
from dataclasses import dataclass
from typing import List, Dict
import matplotlib.pyplot as plt

@dataclass
class EmergencyResponse:
    """Data structure for tracking emergency response performance"""
    emergency_id: str
    detection_time: float
    response_time: float
    resolution_time: float
    severity_level: int
    response_accuracy: float
    false_positive: bool

class EmergencyMetricsAnalyzer:
    """
    Performance analysis for AI emergency response systems
    Tracks response times, accuracy, and optimization opportunities
    """
    
    def __init__(self):
        self.response_history: List[EmergencyResponse] = []
        self.performance_benchmarks = {
            "detection_time": 2.0,  # seconds
            "response_time": 30.0,  # seconds
            "accuracy": 0.95,       # 95%
            "false_positive_rate": 0.05  # 5%
        }
    
    def record_emergency_response(self, response: EmergencyResponse):
        """Record emergency response for analysis"""
        self.response_history.append(response)
        
        # Real-time performance check
        self.check_performance_alerts(response)
    
    def check_performance_alerts(self, response: EmergencyResponse):
        """Check if response meets performance benchmarks"""
        alerts = []
        
        if response.detection_time > self.performance_benchmarks["detection_time"]:
            alerts.append(f"⚠️ Slow detection: {response.detection_time}s")
        
        if response.response_time > self.performance_benchmarks["response_time"]:
            alerts.append(f"⚠️ Slow response: {response.response_time}s")
        
        if response.response_accuracy < self.performance_benchmarks["accuracy"]:
            alerts.append(f"⚠️ Low accuracy: {response.response_accuracy:.2%}")
        
        if alerts:
            print(f"🚨 Performance Alert for {response.emergency_id}:")
            for alert in alerts:
                print(f"  {alert}")
    
    def generate_performance_report(self) -> Dict:
        """Generate comprehensive performance analysis report"""
        if not self.response_history:
            return {"error": "No response data available"}
        
        # Calculate metrics
        total_responses = len(self.response_history)
        false_positives = sum(1 for r in self.response_history if r.false_positive)
        
        avg_detection_time = sum(r.detection_time for r in self.response_history) / total_responses
        avg_response_time = sum(r.response_time for r in self.response_history) / total_responses
        avg_accuracy = sum(r.response_accuracy for r in self.response_history) / total_responses
        false_positive_rate = false_positives / total_responses
        
        # Performance grade
        grade = self.calculate_performance_grade(
            avg_detection_time, avg_response_time, avg_accuracy, false_positive_rate
        )
        
        report = {
            "total_emergencies": total_responses,
            "average_detection_time": round(avg_detection_time, 2),
            "average_response_time": round(avg_response_time, 2),
            "average_accuracy": round(avg_accuracy, 3),
            "false_positive_rate": round(false_positive_rate, 3),
            "performance_grade": grade,
            "benchmark_comparison": self.compare_to_benchmarks(),
            "optimization_recommendations": self.generate_optimization_recommendations()
        }
        
        return report
    
    def calculate_performance_grade(self, detection_time, response_time, accuracy, fp_rate):
        """Calculate overall performance grade A-F"""
        score = 0
        
        # Detection time score (30% weight)
        if detection_time <= 1.0:
            score += 30
        elif detection_time <= 2.0:
            score += 25
        elif detection_time <= 5.0:
            score += 20
        else:
            score += 10
        
        # Response time score (25% weight)
        if response_time <= 15.0:
            score += 25
        elif response_time <= 30.0:
            score += 20
        elif response_time <= 60.0:
            score += 15
        else:
            score += 5
        
        # Accuracy score (35% weight)
        score += int(accuracy * 35)
        
        # False positive penalty (10% weight)
        score += int((1 - fp_rate) * 10)
        
        # Convert to letter grade
        if score >= 90:
            return "A"
        elif score >= 80:
            return "B"
        elif score >= 70:
            return "C"
        elif score >= 60:
            return "D"
        else:
            return "F"
    
    def generate_optimization_recommendations(self) -> List[str]:
        """AI-generated recommendations for system optimization"""
        recommendations = []
        
        if not self.response_history:
            return recommendations
        
        avg_detection = sum(r.detection_time for r in self.response_history) / len(self.response_history)
        avg_response = sum(r.response_time for r in self.response_history) / len(self.response_history)
        avg_accuracy = sum(r.response_accuracy for r in self.response_history) / len(self.response_history)
        
        if avg_detection > 2.0:
            recommendations.append(
                "Optimize sensor polling frequency and ML model inference speed"
            )
        
        if avg_response > 30.0:
            recommendations.append(
                "Implement pre-positioned response units and automated protocols"
            )
        
        if avg_accuracy < 0.9:
            recommendations.append(
                "Retrain ML models with recent data and add sensor redundancy"
            )
        
        false_positives = sum(1 for r in self.response_history if r.false_positive)
        if false_positives > len(self.response_history) * 0.1:
            recommendations.append(
                "Implement multi-sensor confirmation and adjust detection thresholds"
            )
        
        return recommendations

# Usage example
def demo_performance_monitoring():
    """Demonstrate emergency response performance monitoring"""
    
    analyzer = EmergencyMetricsAnalyzer()
    
    # Simulate emergency responses
    sample_responses = [
        EmergencyResponse("FIRE_001", 1.2, 25.0, 180.0, 8, 0.95, False),
        EmergencyResponse("GAS_002", 0.8, 15.0, 120.0, 6, 0.98, False),
        EmergencyResponse("FIRE_003", 3.5, 45.0, 300.0, 9, 0.85, False),
        EmergencyResponse("FALSE_004", 1.0, 20.0, 0.0, 2, 0.60, True),
        EmergencyResponse("MEDICAL_005", 0.5, 12.0, 90.0, 7, 0.99, False),
    ]
    
    # Record responses
    for response in sample_responses:
        analyzer.record_emergency_response(response)
    
    # Generate performance report
    report = analyzer.generate_performance_report()
    
    print("📊 EMERGENCY RESPONSE PERFORMANCE REPORT")
    print("=" * 50)
    print(f"Total Emergencies Handled: {report['total_emergencies']}")
    print(f"Average Detection Time: {report['average_detection_time']}s")
    print(f"Average Response Time: {report['average_response_time']}s")
    print(f"Average Accuracy: {report['average_accuracy']:.2%}")
    print(f"False Positive Rate: {report['false_positive_rate']:.1%}")
    print(f"Performance Grade: {report['performance_grade']}")
    
    print("\n🎯 OPTIMIZATION RECOMMENDATIONS:")
    for i, rec in enumerate(report['optimization_recommendations'], 1):
        print(f"{i}. {rec}")

# Run performance monitoring demo
demo_performance_monitoring()

# docker-compose.yml for emergency response system
version: '3.8'

services:
  emergency-ai:
    image: emergency-ai:latest
    ports:
      - "8080:8080"
    environment:
      - REDIS_URL=redis://redis:6379
      - POSTGRES_URL=postgresql://postgres:password@postgres:5432/emergencydb
      - MQTT_BROKER=mosquitto:1883
    depends_on:
      - redis
      - postgres
      - mosquitto
    restart: unless-stopped
    
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    restart: unless-stopped
    
  postgres:
    image: postgres:15
    environment:
      POSTGRES_DB: emergencydb
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: password
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: unless-stopped
    
  mosquitto:
    image: eclipse-mosquitto:2
    ports:
      - "1883:1883"
      - "9001:9001"
    volumes:
      - ./mosquitto.conf:/mosquitto/config/mosquitto.conf
    restart: unless-stopped
    
  prometheus:
    image: prom/prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
    restart: unless-stopped
    
  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin
    volumes:
      - grafana_data:/var/lib/grafana
    restart: unless-stopped

volumes:
  postgres_data:
  grafana_data:

import asyncio
from typing import List
import consul
import redis
from kubernetes import client, config

class HighAvailabilityEmergencyAI:
    """
    High-availability emergency response system with:
    - Multiple AI processing nodes
    - Automatic failover
    - Load balancing
    - Geographic redundancy
    """
    
    def __init__(self):
        self.consul_client = consul.Consul()
        self.redis_client = redis.Redis(host='redis-cluster')
        self.active_nodes = []
        self.current_node_id = self.generate_node_id()
        
        # Kubernetes client for scaling
        config.load_incluster_config()  # When running in K8s
        self.k8s_client = client.AppsV1Api()
    
    async def register_node(self):
        """Register this AI node with service discovery"""
        node_info = {
            'id': self.current_node_id,
            'address': self.get_local_ip(),
            'port': 8080,
            'health_check_url': f'http://{self.get_local_ip()}:8080/health',
            'tags': ['emergency-ai', 'primary'],
            'meta': {
                'region': 'us-east-1',
                'processing_capacity': '1000',
                'specialization': 'fire,medical,gas'
            }
        }
        
        self.consul_client.agent.service.register(
            name='emergency-ai',
            service_id=self.current_node_id,
            **node_info
        )
        
        print(f"🔗 Registered AI node {self.current_node_id}")
    
    async def monitor_cluster_health(self):
        """Monitor health of all AI processing nodes"""
        while True:
            try:
                # Get all healthy emergency-ai services
                healthy_nodes = self.consul_client.health.service(
                    'emergency-ai', passing=True
                )[1]
                
                self.active_nodes = [
                    node['Service']['ID'] for node in healthy_nodes
                ]
                
                # Auto-scale if needed
                await self.check_scaling_requirements()
                
                # Rebalance load if topology changed
                await self.rebalance_emergency_load()
                
            except Exception as e:
                print(f"❌ Cluster monitoring error: {e}")
            
            await asyncio.sleep(30)  # Check every 30 seconds
    
    async def check_scaling_requirements(self):
        """Automatically scale AI nodes based on emergency load"""
        
        # Get current emergency queue size
        queue_size = self.redis_client.llen('emergency_queue')
        current_nodes = len(self.active_nodes)
        
        # Scale up if queue is backing up
        if queue_size > current_nodes * 10:  # >10 emergencies per node
            await self.scale_up()
        
        # Scale down if overcapacity (but keep minimum 2 nodes)
        elif queue_size < current_nodes * 2 and current_nodes > 2:
            await self.scale_down()
    
    async def scale_up(self):
        """Add more AI processing nodes"""
        try:
            # Update Kubernetes deployment
            deployment = self.k8s_client.read_namespaced_deployment(
                name='emergency-ai', namespace='default'
            )
            
            current_replicas = deployment.spec.replicas
            new_replicas = min(current_replicas + 2, 10)  # Max 10 nodes
            
            deployment.spec.replicas = new_replicas
            
            self.k8s_client.patch_namespaced_deployment(
                name='emergency-ai',
                namespace='default',
                body=deployment
            )
            
            print(f"📈 Scaled up to {new_replicas} AI nodes")
            
        except Exception as e:
            print(f"❌ Scale up failed: {e}")
    
    async def handle_node_failure(self, failed_node_id: str):
        """Handle failure of an AI processing node"""
        
        print(f"💀 Node failure detected: {failed_node_id}")
        
        # Redistribute emergencies from failed node
        failed_emergencies = self.redis_client.lrange(
            f'node:{failed_node_id}:emergencies', 0, -1
        )
        
        for emergency in failed_emergencies:
            # Requeue to main emergency queue
            self.redis_client.rpush('emergency_queue', emergency)
        
        # Clear failed node's queue
        self.redis_client.delete(f'node:{failed_node_id}:emergencies')
        
        # Update node list
        if failed_node_id in self.active_nodes:
            self.active_nodes.remove(failed_node_id)
        
        print(f"🔄 Redistributed {len(failed_emergencies)} emergencies")

# Load balancer for emergency requests
class EmergencyLoadBalancer:
    """
    Intelligent load balancer that routes emergencies to optimal AI nodes
    based on node capacity, specialization, and geographic proximity
    """
    
    def __init__(self):
        self.consul_client = consul.Consul()
        self.routing_strategy = 'intelligent'  # or 'round_robin', 'least_connections'
    
    async def route_emergency(self, emergency_data):
        """Route emergency to optimal AI node"""
        
        # Get available nodes
        available_nodes = self.consul_client.health.service(
            'emergency-ai', passing=True
        )[1]
        
        if not available_nodes:
            raise Exception("No healthy AI nodes available!")
        
        # Select optimal node based on strategy
        if self.routing_strategy == 'intelligent':
            selected_node = await self.intelligent_routing(emergency_data, available_nodes)
        else:
            selected_node = self.round_robin_routing(available_nodes)
        
        # Route emergency to selected node
        return await self.send_to_node(emergency_data, selected_node)
    
    async def intelligent_routing(self, emergency_data, available_nodes):
        """AI-powered routing based on emergency type and node capabilities"""
        
        emergency_type = emergency_data.get('type', 'unknown')
        emergency_location = emergency_data.get('location', {})
        
        best_node = None
        best_score = -1
        
        for node_info in available_nodes:
            node = node_info['Service']
            node_meta = node.get('Meta', {})
            
            score = 0
            
            # Specialization match
            specializations = node_meta.get('specialization', '').split(',')
            if emergency_type in specializations:
                score += 50
            
            # Geographic proximity
            node_region = node_meta.get('region', '')
            if emergency_location.get('region') == node_region:
                score += 30
            
            # Current load
            current_load = int(node_meta.get('current_load', '0'))
            capacity = int(node_meta.get('processing_capacity', '100'))
            load_ratio = current_load / capacity
            score += int((1 - load_ratio) * 20)  # Lower load = higher score
            
            if score > best_score:
                best_score = score
                best_node = node
        
        return best_node