nbws/disaster-response-api
0
1"""2Communication tools for disaster response system.3Handles emergency notifications and stakeholder communications.4"""5 6from typing import Dict, List, Optional, Any7import httpx8import asyncio9import logging10from datetime import datetime11import json12import smtplib13from email.mime.text import MIMEText14from email.mime.multipart import MIMEMultipart15 16from utils.config import settings17 18logger = logging.getLogger(__name__)19 20class CommunicationTool:21 """Tool for emergency communications and notifications."""22 23 def __init__(self):24 self.client = None25 self.notification_templates = {}26 self.stakeholder_groups = {}27 28 async def initialize(self):29 """Initialize the communication tool."""30 self.client = httpx.AsyncClient(timeout=settings.external_api_timeout)31 self._load_templates()32 self._load_stakeholder_groups()33 logger.info("Communication tool initialized")34 35 async def cleanup(self):36 """Cleanup resources."""37 if self.client:38 await self.client.aclose()39 40 def _load_templates(self):41 """Load communication templates."""42 self.notification_templates = {43 "emergency_alert": {44 "subject": "🚨 EMERGENCY ALERT: {incident_type} - {location}",45 "template": """46EMERGENCY ALERT47 48Incident Type: {incident_type}49Location: {location}50Severity: {severity}/1051Time: {timestamp}52 53{description}54 55IMMEDIATE ACTIONS REQUIRED:56{actions}57 58EVACUATION INFORMATION:59{evacuation_info}60 61SHELTER LOCATIONS:62{shelter_info}63 64Stay tuned for updates. Follow official emergency channels.65 66Emergency Hotline: {emergency_contact}67"""68 },69 "public_notification": {70 "subject": "Emergency Update: {incident_type} in {location}",71 "template": """72EMERGENCY UPDATE73 74What: {incident_type}75Where: {location}76When: {timestamp}77Risk Level: {risk_level}78 79SITUATION OVERVIEW:80{situation_overview}81 82SAFETY INSTRUCTIONS:83{safety_instructions}84 85RESOURCES AVAILABLE:86{resources}87 88Next update scheduled: {next_update}89 90Stay safe and follow official guidance.91"""92 },93 "responder_brief": {94 "subject": "Emergency Response Deployment: {incident_id}",95 "template": """96EMERGENCY RESPONSE BRIEFING97 98Incident ID: {incident_id}99Type: {incident_type}100Priority: {priority}101Assignment: {assignment}102 103SITUATION:104{situation_summary}105 106DEPLOYMENT INSTRUCTIONS:107{deployment_instructions}108 109RESOURCE ALLOCATION:110{resources_assigned}111 112COORDINATION:113- Incident Commander: {incident_commander}114- Radio Channel: {radio_channel}115- Check-in Time: {checkin_time}116 117SAFETY NOTES:118{safety_notes}119 120Acknowledge receipt and report status upon arrival.121"""122 },123 "social_media": {124 "emergency": "🚨 EMERGENCY ALERT: {incident_type} in {location}. {action_required} Follow @EmergencyServices for updates. #Emergency{hashtags}",125 "update": "📢 UPDATE: {incident_type} situation in {location}. Current status: {status}. {additional_info} #Emergency{hashtags}",126 "all_clear": "✅ ALL CLEAR: {incident_type} emergency in {location} has been resolved. Normal operations resuming. Thank you for your cooperation. #AllClear"127 }128 }129 130 def _load_stakeholder_groups(self):131 """Load stakeholder contact groups."""132 self.stakeholder_groups = {133 "emergency_services": {134 "fire_department": {"priority": 1, "channels": ["radio", "mobile", "email"]},135 "police": {"priority": 1, "channels": ["radio", "mobile", "email"]}, 136 "medical": {"priority": 1, "channels": ["radio", "mobile", "email"]},137 "emergency_management": {"priority": 1, "channels": ["mobile", "email", "sms"]}138 },139 "government": {140 "mayor": {"priority": 2, "channels": ["mobile", "email"]},141 "emergency_director": {"priority": 1, "channels": ["mobile", "email", "sms"]},142 "public_information": {"priority": 2, "channels": ["email", "social"]}143 },144 "media": {145 "local_news": {"priority": 3, "channels": ["email", "social"]},146 "radio_stations": {"priority": 3, "channels": ["email", "phone"]},147 "social_media": {"priority": 3, "channels": ["social"]}148 },149 "organizations": {150 "red_cross": {"priority": 2, "channels": ["email", "mobile"]},151 "salvation_army": {"priority": 3, "channels": ["email"]},152 "volunteer_groups": {"priority": 4, "channels": ["email", "social"]}153 },154 "utilities": {155 "power_company": {"priority": 2, "channels": ["email", "mobile"]},156 "water_department": {"priority": 2, "channels": ["email", "mobile"]},157 "telecom": {"priority": 3, "channels": ["email"]}158 }159 }160 161 async def draft_emergency_alert(162 self,163 incident_data: Dict[str, Any],164 target_audience: str = "public"165 ) -> Dict[str, Any]:166 """Draft emergency alert message."""167 try:168 template_key = "emergency_alert" if target_audience == "public" else "responder_brief"169 template = self.notification_templates.get(template_key, {})170 171 # Extract incident information172 incident_type = incident_data.get("incident_type", "Emergency")173 location = incident_data.get("location_name", "Unknown Location")174 severity = incident_data.get("severity", 5)175 description = incident_data.get("description", "Emergency situation requires immediate attention.")176 177 # Generate content based on audience178 if target_audience == "public":179 alert_content = self._generate_public_alert(incident_data, template)180 elif target_audience == "responders":181 alert_content = self._generate_responder_brief(incident_data, template)182 else:183 alert_content = self._generate_stakeholder_update(incident_data, template)184 185 # Add approval workflow186 alert_content["requires_approval"] = True187 alert_content["approval_level"] = self._determine_approval_level(severity)188 alert_content["estimated_reach"] = self._estimate_message_reach(target_audience)189 190 return alert_content191 192 except Exception as e:193 logger.error(f"Error drafting emergency alert: {e}")194 return {"error": str(e), "source": "alert_drafting"}195 196 async def send_notifications(197 self,198 message_data: Dict[str, Any],199 recipient_groups: List[str],200 channels: List[str] = ["email", "sms"],201 priority: int = 1202 ) -> Dict[str, Any]:203 """Send notifications to specified groups via multiple channels."""204 try:205 results = {206 "total_sent": 0,207 "successful_sends": 0,208 "failed_sends": 0,209 "channel_results": {},210 "recipient_results": {},211 "timestamp": datetime.utcnow().isoformat()212 }213 214 # Process each recipient group215 for group in recipient_groups:216 if group in self.stakeholder_groups:217 group_results = await self._send_to_group(218 message_data, 219 group, 220 channels, 221 priority222 )223 results["recipient_results"][group] = group_results224 results["successful_sends"] += group_results.get("successful", 0)225 results["failed_sends"] += group_results.get("failed", 0)226 227 results["total_sent"] = results["successful_sends"] + results["failed_sends"]228 results["success_rate"] = (229 results["successful_sends"] / results["total_sent"] 230 if results["total_sent"] > 0 else 0231 )232 233 return results234 235 except Exception as e:236 logger.error(f"Error sending notifications: {e}")237 return {"error": str(e), "source": "notification_sending"}238 239 async def create_social_media_posts(240 self,241 incident_data: Dict[str, Any],242 post_type: str = "emergency"243 ) -> List[Dict[str, Any]]:244 """Create social media posts for different platforms."""245 try:246 posts = []247 templates = self.notification_templates["social_media"]248 249 if post_type not in templates:250 post_type = "emergency"251 252 base_template = templates[post_type]253 254 # Generate hashtags255 incident_type = incident_data.get("incident_type", "Emergency")256 location = incident_data.get("location_name", "LocalArea")257 hashtags = self._generate_hashtags(incident_type, location)258 259 # Create platform-specific posts260 platforms = {261 "twitter": {"char_limit": 280, "hashtag_count": 3},262 "facebook": {"char_limit": 500, "hashtag_count": 5},263 "instagram": {"char_limit": 300, "hashtag_count": 10}264 }265 266 for platform, limits in platforms.items():267 post_content = self._create_platform_post(268 base_template,269 incident_data,270 hashtags,271 limits272 )273 274 posts.append({275 "platform": platform,276 "content": post_content,277 "hashtags": hashtags[:limits["hashtag_count"]],278 "character_count": len(post_content),279 "requires_approval": True,280 "urgency_level": self._assess_post_urgency(incident_data),281 "estimated_reach": self._estimate_social_reach(platform, hashtags)282 })283 284 return posts285 286 except Exception as e:287 logger.error(f"Error creating social media posts: {e}")288 return [{"error": str(e), "source": "social_media_creation"}]289 290 async def coordinate_multi_channel_alert(291 self,292 incident_data: Dict[str, Any],293 alert_level: str = "high"294 ) -> Dict[str, Any]:295 """Coordinate alerts across multiple communication channels."""296 try:297 coordination_plan = {298 "alert_id": f"alert_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}",299 "alert_level": alert_level,300 "incident_data": incident_data,301 "channels": {},302 "timeline": {},303 "approval_required": True,304 "estimated_total_reach": 0305 }306 307 # Determine communication strategy based on alert level308 strategy = self._get_communication_strategy(alert_level)309 310 # Plan emergency alerts311 if "emergency_alert" in strategy["channels"]:312 alert_draft = await self.draft_emergency_alert(incident_data, "public")313 coordination_plan["channels"]["emergency_alert"] = alert_draft314 coordination_plan["timeline"]["emergency_alert"] = "immediate"315 316 # Plan responder notifications317 if "responder_brief" in strategy["channels"]:318 responder_brief = await self.draft_emergency_alert(incident_data, "responders")319 coordination_plan["channels"]["responder_brief"] = responder_brief320 coordination_plan["timeline"]["responder_brief"] = "immediate"321 322 # Plan social media323 if "social_media" in strategy["channels"]:324 social_posts = await self.create_social_media_posts(incident_data, "emergency")325 coordination_plan["channels"]["social_media"] = social_posts326 coordination_plan["timeline"]["social_media"] = "5_minutes"327 328 # Plan stakeholder notifications329 if "stakeholder_update" in strategy["channels"]:330 stakeholder_groups = strategy.get("stakeholder_groups", ["government", "organizations"])331 coordination_plan["channels"]["stakeholder_notifications"] = {332 "groups": stakeholder_groups,333 "message": await self.draft_emergency_alert(incident_data, "stakeholders")334 }335 coordination_plan["timeline"]["stakeholder_notifications"] = "10_minutes"336 337 # Calculate total estimated reach338 coordination_plan["estimated_total_reach"] = self._calculate_total_reach(coordination_plan)339 340 return coordination_plan341 342 except Exception as e:343 logger.error(f"Error coordinating multi-channel alert: {e}")344 return {"error": str(e), "source": "multi_channel_coordination"}345 346 def _generate_public_alert(self, incident_data: Dict, template: Dict) -> Dict[str, Any]:347 """Generate public emergency alert."""348 try:349 # Extract and format incident data350 incident_type = incident_data.get("incident_type", "Emergency")351 location = incident_data.get("location_name", "affected area")352 severity = incident_data.get("severity", 5)353 354 # Generate actions based on incident type and severity355 actions = self._generate_safety_actions(incident_type, severity)356 evacuation_info = self._generate_evacuation_info(incident_data)357 shelter_info = self._generate_shelter_info(incident_data)358 359 content = template["template"].format(360 incident_type=incident_type,361 location=location,362 severity=severity,363 timestamp=datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC"),364 description=incident_data.get("description", "Emergency situation in progress"),365 actions=actions,366 evacuation_info=evacuation_info,367 shelter_info=shelter_info,368 emergency_contact="911 or local emergency services"369 )370 371 return {372 "type": "public_alert",373 "subject": template["subject"].format(374 incident_type=incident_type,375 location=location376 ),377 "content": content,378 "priority": "high" if severity >= 7 else "medium",379 "target_audience": "general_public",380 "distribution_channels": ["emergency_broadcast", "social_media", "website", "mobile_alert"]381 }382 383 except Exception as e:384 logger.error(f"Error generating public alert: {e}")385 return {"error": str(e)}386 387 def _generate_responder_brief(self, incident_data: Dict, template: Dict) -> Dict[str, Any]:388 """Generate emergency responder briefing."""389 try:390 incident_id = incident_data.get("id", f"INC_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}")391 incident_type = incident_data.get("incident_type", "Emergency")392 393 # Generate responder-specific information394 deployment_instructions = self._generate_deployment_instructions(incident_data)395 resource_assignments = self._generate_resource_assignments(incident_data)396 safety_notes = self._generate_safety_notes(incident_data)397 398 content = template["template"].format(399 incident_id=incident_id,400 incident_type=incident_type,401 priority="HIGH" if incident_data.get("severity", 5) >= 7 else "MEDIUM",402 assignment=incident_data.get("assignment", "Emergency Response"),403 situation_summary=incident_data.get("description", "Emergency response required"),404 deployment_instructions=deployment_instructions,405 resources_assigned=resource_assignments,406 incident_commander="IC-001", # Would be dynamically assigned407 radio_channel="TAC-1",408 checkin_time="Every 15 minutes",409 safety_notes=safety_notes410 )411 412 return {413 "type": "responder_brief",414 "subject": template["subject"].format(incident_id=incident_id),415 "content": content,416 "priority": "critical",417 "target_audience": "emergency_responders",418 "distribution_channels": ["secure_radio", "mobile_mdt", "email"],419 "security_classification": "operational"420 }421 422 except Exception as e:423 logger.error(f"Error generating responder brief: {e}")424 return {"error": str(e)}425 426 def _generate_stakeholder_update(self, incident_data: Dict, template: Dict) -> Dict[str, Any]:427 """Generate stakeholder update message."""428 return {429 "type": "stakeholder_update",430 "content": f"Incident Update: {incident_data.get('incident_type', 'Emergency')} at {incident_data.get('location_name', 'location')}. Response coordinated, situation being monitored.",431 "priority": "normal",432 "target_audience": "stakeholders"433 }434 435 def _generate_safety_actions(self, incident_type: str, severity: int) -> str:436 """Generate safety actions based on incident type."""437 actions_map = {438 "earthquake": [439 "Drop, Cover, and Hold On if shaking continues",440 "Stay away from windows and heavy objects",441 "Exit building only if safe to do so",442 "Check for injuries and hazards"443 ],444 "flood": [445 "Move to higher ground immediately",446 "Avoid walking or driving through flood water",447 "Stay away from downed power lines",448 "Monitor emergency broadcasts"449 ],450 "fire": [451 "Evacuate immediately if in affected area",452 "Stay low to avoid smoke inhalation",453 "Call 911 if not already reported",454 "Do not return until all-clear given"455 ],456 "storm": [457 "Seek shelter in sturdy building",458 "Stay away from windows",459 "Avoid outdoor activities",460 "Monitor weather updates"461 ]462 }463 464 default_actions = [465 "Follow instructions from emergency personnel",466 "Stay informed through official channels",467 "Avoid affected areas",468 "Check on neighbors if safe to do so"469 ]470 471 actions = actions_map.get(incident_type.lower(), default_actions)472 473 if severity >= 8:474 actions.insert(0, "⚠️ IMMEDIATE ACTION REQUIRED ⚠️")475 476 return "\n".join(f"• {action}" for action in actions)477 478 def _generate_evacuation_info(self, incident_data: Dict) -> str:479 """Generate evacuation information."""480 evacuation_routes = incident_data.get("evacuation_routes", [])481 482 if not evacuation_routes:483 return "Evacuation routes being assessed. Await further instructions."484 485 info = "EVACUATION ROUTES:\n"486 for i, route in enumerate(evacuation_routes[:3], 1):487 route_name = route.get("name", f"Route {i}")488 capacity = route.get("capacity", "Unknown")489 status = route.get("status", "Available")490 info += f"• {route_name}: {status} (Capacity: {capacity})\n"491 492 return info493 494 def _generate_shelter_info(self, incident_data: Dict) -> str:495 """Generate shelter information."""496 shelters = incident_data.get("shelters", [])497 498 if not shelters:499 return "Emergency shelters being prepared. Updates to follow."500 501 info = "EMERGENCY SHELTERS:\n"502 for shelter in shelters[:3]:503 name = shelter.get("name", "Emergency Shelter")504 address = shelter.get("address", "Address TBD")505 capacity = shelter.get("available_capacity", "Unknown")506 info += f"• {name}: {address} (Available: {capacity})\n"507 508 return info509 510 def _generate_deployment_instructions(self, incident_data: Dict) -> str:511 """Generate deployment instructions for responders."""512 instructions = [513 f"Respond to: {incident_data.get('location_name', 'incident location')}",514 f"Approach via: {incident_data.get('approach_route', 'standard access routes')}",515 "Establish command post at designated location",516 "Coordinate with on-scene incident commander"517 ]518 519 if incident_data.get("severity", 5) >= 8:520 instructions.insert(0, "⚠️ HIGH PRIORITY RESPONSE ⚠️")521 522 return "\n".join(f"• {instruction}" for instruction in instructions)523 524 def _generate_resource_assignments(self, incident_data: Dict) -> str:525 """Generate resource assignment information."""526 assignments = incident_data.get("resource_assignments", {})527 528 if not assignments:529 return "Resource assignments pending - check with dispatch"530 531 info = ""532 for resource_type, details in assignments.items():533 info += f"• {resource_type}: {details.get('units', 'TBD')} units\n"534 535 return info536 537 def _generate_safety_notes(self, incident_data: Dict) -> str:538 """Generate safety notes for responders."""539 notes = [540 "Use appropriate PPE for incident type",541 "Maintain situational awareness at all times",542 "Follow established safety protocols"543 ]544 545 incident_type = incident_data.get("incident_type", "").lower()546 547 if "chemical" in incident_type or "hazmat" in incident_type:548 notes.append("HAZMAT protocols in effect - specialized equipment required")549 elif "fire" in incident_type:550 notes.append("Fire conditions present - full protective gear required")551 elif "flood" in incident_type:552 notes.append("Water rescue protocols - swift water training required")553 554 return "\n".join(f"• {note}" for note in notes)555 556 def _generate_hashtags(self, incident_type: str, location: str) -> List[str]:557 """Generate relevant hashtags for social media."""558 base_hashtags = ["Emergency", "Safety", "Alert"]559 560 # Add incident-specific tags561 type_tags = {562 "earthquake": ["Earthquake", "Seismic"],563 "flood": ["Flood", "FloodWarning"],564 "fire": ["Fire", "Evacuation"],565 "storm": ["Storm", "Weather"]566 }567 568 hashtags = base_hashtags.copy()569 hashtags.extend(type_tags.get(incident_type.lower(), [incident_type.title()]))570 571 # Add location-based tags572 location_clean = location.replace(" ", "").replace(",", "")573 hashtags.append(f"{location_clean}Emergency")574 575 return hashtags576 577 def _create_platform_post(578 self, 579 template: str, 580 incident_data: Dict, 581 hashtags: List[str], 582 limits: Dict583 ) -> str:584 """Create platform-specific social media post."""585 try:586 # Format the base template587 formatted_post = template.format(588 incident_type=incident_data.get("incident_type", "Emergency"),589 location=incident_data.get("location_name", "local area"),590 action_required="Seek shelter and follow emergency guidance.",591 status="ongoing monitoring",592 additional_info="Updates will be posted as available.",593 hashtags=" ".join(f"#{tag}" for tag in hashtags[:limits["hashtag_count"]])594 )595 596 # Trim to character limit if necessary597 if len(formatted_post) > limits["char_limit"]:598 trim_length = limits["char_limit"] - 3 # Account for "..."599 formatted_post = formatted_post[:trim_length] + "..."600 601 return formatted_post602 603 except Exception as e:604 logger.error(f"Error creating platform post: {e}")605 return f"Emergency alert: {incident_data.get('incident_type', 'situation')} in {incident_data.get('location_name', 'area')}. Follow official channels for updates."606 607 def _determine_approval_level(self, severity: int) -> str:608 """Determine required approval level based on severity."""609 if severity >= 9:610 return "executive"611 elif severity >= 7:612 return "manager" 613 elif severity >= 5:614 return "supervisor"615 else:616 return "operator"617 618 def _estimate_message_reach(self, target_audience: str) -> Dict[str, int]:619 """Estimate message reach for different audiences."""620 reach_estimates = {621 "public": {"immediate": 10000, "1_hour": 50000, "24_hour": 200000},622 "responders": {"immediate": 200, "1_hour": 500, "24_hour": 800},623 "stakeholders": {"immediate": 50, "1_hour": 100, "24_hour": 150}624 }625 626 return reach_estimates.get(target_audience, {"immediate": 1000, "1_hour": 5000, "24_hour": 20000})627 628 async def _send_to_group(629 self, 630 message_data: Dict, 631 group: str, 632 channels: List[str], 633 priority: int634 ) -> Dict[str, Any]:635 """Send message to a stakeholder group."""636 # Mock implementation - in production, integrate with actual messaging services637 group_data = self.stakeholder_groups.get(group, {})638 639 results = {640 "group": group,641 "channels_attempted": channels,642 "successful": len(channels) * len(group_data), # Mock success643 "failed": 0,644 "timestamp": datetime.utcnow().isoformat()645 }646 647 return results648 649 def _get_communication_strategy(self, alert_level: str) -> Dict[str, Any]:650 """Get communication strategy based on alert level."""651 strategies = {652 "low": {653 "channels": ["stakeholder_update"],654 "stakeholder_groups": ["government"],655 "timeline": "normal"656 },657 "medium": {658 "channels": ["stakeholder_update", "social_media"],659 "stakeholder_groups": ["government", "organizations"],660 "timeline": "expedited"661 },662 "high": {663 "channels": ["emergency_alert", "responder_brief", "social_media", "stakeholder_update"],664 "stakeholder_groups": ["emergency_services", "government", "organizations"],665 "timeline": "immediate"666 },667 "critical": {668 "channels": ["emergency_alert", "responder_brief", "social_media", "stakeholder_update"],669 "stakeholder_groups": ["emergency_services", "government", "organizations", "media", "utilities"],670 "timeline": "immediate"671 }672 }673 674 return strategies.get(alert_level, strategies["medium"])675 676 def _assess_post_urgency(self, incident_data: Dict) -> str:677 """Assess urgency level for social media posts."""678 severity = incident_data.get("severity", 5)679 680 if severity >= 8:681 return "critical"682 elif severity >= 6:683 return "high"684 elif severity >= 4:685 return "medium"686 else:687 return "low"688 689 def _estimate_social_reach(self, platform: str, hashtags: List[str]) -> Dict[str, int]:690 """Estimate social media reach by platform."""691 base_reach = {692 "twitter": {"immediate": 500, "1_hour": 2000, "24_hour": 10000},693 "facebook": {"immediate": 300, "1_hour": 1500, "24_hour": 8000},694 "instagram": {"immediate": 200, "1_hour": 1000, "24_hour": 5000}695 }696 697 # Boost reach for emergency hashtags698 if any(tag in ["Emergency", "Alert", "Evacuation"] for tag in hashtags):699 for timeframe in base_reach.get(platform, {}):700 base_reach[platform][timeframe] = int(base_reach[platform][timeframe] * 1.5)701 702 return base_reach.get(platform, {"immediate": 100, "1_hour": 500, "24_hour": 2000})703 704 def _calculate_total_reach(self, coordination_plan: Dict) -> int:705 """Calculate total estimated reach across all channels."""706 total_reach = 0707 708 for channel, data in coordination_plan.get("channels", {}).items():709 if channel == "emergency_alert":710 total_reach += data.get("estimated_reach", {}).get("immediate", 0)711 elif channel == "social_media":712 for post in data:713 total_reach += post.get("estimated_reach", {}).get("immediate", 0)714 elif channel == "stakeholder_notifications":715 # Estimate based on number of groups716 total_reach += len(data.get("groups", [])) * 20717 718 return total_reach