Disaster Response with AI Drones
When disasters strike, time is critical. AI-powered drones can rapidly map affected areas, assess damage, locate survivors, and deliver supplies—providing crucial intelligence for rescue operations when every minute counts.
The Disaster Response System
Real-World Analogy
Think of disaster response drones as "emergency reconnaissance units." Just as military scouts gather intelligence before an advance, disaster drones provide real-time situational awareness to guide rescue teams and resource allocation.
SVG: Disaster Response Workflow and Damage Classification
Rapid Mapping System
from typing import List, Dict, Tuple
from dataclasses import dataclass
from enum import Enum
from datetime import datetime
class DamageLevel(Enum):
INTACT = 1
MINOR = 2
MODERATE = 3
MAJOR = 4
SEVERE = 5
DESTROYED = 6
@dataclass
class BuildingDamage:
building_id: str
location: np.ndarray
damage_level: DamageLevel
confidence: float
roof_damage: float
wall_damage: float
debris_present: bool
accessible: bool
class RapidMapper:
def __init__(self, grid_resolution: float = 10.0):
self.grid_resolution = grid_resolution
self.damage_grid = None
self.buildings: List[BuildingDamage] = []
def create_damage_map(self, rgb_frames: List[np.ndarray],
thermal_frames: List[np.ndarray],
positions: List[np.ndarray]) -> np.ndarray:
"""Create comprehensive damage assessment map"""
h, w = rgb_frames[0].shape[:2]
damage_map = np.zeros((h, w), dtype=int)
for rgb, thermal, pos in zip(rgb_frames, thermal_frames, positions):
# Analyze visual damage
visual_damage = self._analyze_visual_damage(rgb)
# Analyze thermal anomalies
thermal_damage = self._analyze_thermal_damage(thermal)
# Combine assessments
combined = np.maximum(visual_damage, thermal_damage)
# Update global map
x, y = int(pos[0]), int(pos[1])
if 0 <= x < h and 0 <= y < w:
damage_map[max(0,x-5):min(h,x+5),
max(0,y-5):min(w,y+5)] = combined
self.damage_grid = damage_map
return damage_map
def _analyze_visual_damage(self, rgb: np.ndarray) -> np.ndarray:
"""Analyze visual indicators of damage"""
gray = np.mean(rgb, axis=2)
# Edge detection for structural damage
from scipy import ndimage
edges = ndimage.sobel(gray)
edge_density = ndimage.uniform_filter(edges, size=10)
# Color analysis for fire/smoke damage
hsv = self._rgb_to_hsv(rgb)
fire_damage = (hsv[:,:,0] < 30) & (hsv[:,:,2] > 100)
# Debris detection (irregular textures)
texture = np.std(ndimage.uniform_filter(gray, size=5), axis=0)
# Combine indicators
damage = np.zeros_like(gray, dtype=int)
damage[edge_density > 30] = DamageLevel.MODERATE.value
damage[fire_damage] = DamageLevel.MAJOR.value
damage[texture > 20] = DamageLevel.MINOR.value
return damage
def _analyze_thermal_damage(self, thermal: np.ndarray) -> np.ndarray:
"""Analyze thermal indicators"""
damage = np.zeros_like(thermal, dtype=int)
# Hot spots (fires, structural failures)
hot_threshold = np.percentile(thermal, 95)
damage[thermal > hot_threshold] = DamageLevel.MAJOR.value
# Cold spots (exposed interiors, water damage)
cold_threshold = np.percentile(thermal, 5)
damage[thermal < cold_threshold] = DamageLevel.MODERATE.value
return damage
def _rgb_to_hsv(self, rgb: np.ndarray) -> np.ndarray:
"""Convert RGB to HSV"""
r, g, b = rgb[:,:,0]/255.0, rgb[:,:,1]/255.0, rgb[:,:,2]/255.0
max_c = np.maximum(np.maximum(r, g), b)
min_c = np.minimum(np.minimum(r, g), b)
v = max_c
s = np.zeros_like(v)
mask = max_c > 0
s[mask] = (max_c[mask] - min_c[mask]) / max_c[mask]
h = np.zeros_like(v)
return np.stack([h * 180, s * 255, v * 255], axis=2)
def identify_casualty_zones(self, damage_map: np.ndarray,
thermal_data: np.ndarray) -> List[np.ndarray]:
"""Identify areas likely to have casualties"""
zones = []
# High damage areas
severe_mask = damage_map >= DamageLevel.MAJOR.value
# With thermal signatures (survivors)
thermal_mask = thermal_data > 30 # Body temperature
# Combine
casualty_risk = severe_mask & thermal_mask
from scipy import ndimage
labeled, num_features = ndimage.label(casualty_risk)
for i in range(1, num_features + 1):
centroid = ndimage.center_of_mass(labeled == i)
zones.append(np.array(centroid))
return zones
def prioritize_search_areas(self, damage_map: np.ndarray,
thermal_data: np.ndarray) -> List[Dict]:
"""Prioritize areas for search and rescue"""
priorities = []
# Calculate priority score for each grid cell
h, w = damage_map.shape
cell_size = 50
for y in range(0, h, cell_size):
for x in range(0, w, cell_size):
cell_damage = damage_map[y:y+cell_size, x:x+cell_size]
cell_thermal = thermal_data[y:y+cell_size, x:x+cell_size]
# Priority factors
damage_score = np.mean(cell_damage) / 6
heat_score = np.mean(cell_thermal > 30) if cell_thermal.size > 0 else 0
accessibility = 1 - damage_score * 0.5
priority_score = (damage_score * 0.4 +
heat_score * 0.4 +
accessibility * 0.2)
if priority_score > 0.3:
priorities.append({
'location': np.array([x + cell_size//2, y + cell_size//2]),
'priority_score': priority_score,
'damage_level': int(np.max(cell_damage)),
'heat_signatures': heat_score > 0.1
})
# Sort by priority
priorities.sort(key=lambda p: p['priority_score'], reverse=True)
return priorities
def generate_situation_report(self) -> Dict:
"""Generate comprehensive situation report"""
if self.damage_grid is None:
return {'error': 'No damage map available'}
total_cells = self.damage_grid.size
damage_counts = {}
for level in DamageLevel:
count = np.sum(self.damage_grid == level.value)
damage_counts[level.name] = {
'count': int(count),
'percentage': count / total_cells * 100
}
return {
'total_area_assessed': total_cells * (self.grid_resolution ** 2) / 10000,
'damage_summary': damage_counts,
'critical_areas': int(np.sum(self.damage_grid >= DamageLevel.MAJOR.value)),
'estimated_displaced': self._estimate_displaced(),
'priority_response_needed': self._estimate_response_needs()
}
def _estimate_displaced(self) -> int:
"""Estimate number of displaced people"""
major_damage = np.sum(self.damage_grid >= DamageLevel.MAJOR.value)
return int(major_damage * 10) # ~10 people per major damage unit
def _estimate_response_needs(self) -> Dict:
"""Estimate response resource needs"""
severe = np.sum(self.damage_grid >= DamageLevel.SEVERE.value)
major = np.sum(self.damage_grid == DamageLevel.MAJOR.value)
return {
'rescue_teams': int(severe / 10) + 1,
'medical_units': int(major / 20) + 1,
'shelter_capacity': int(major * 50),
'supply_trucks': int(major / 15) + 1
}
Damage Assessment AI
from typing import List, Dict, Tuple
from dataclasses import dataclass
from enum import Enum
class StructureType(Enum):
RESIDENTIAL = "residential"
COMMERCIAL = "commercial"
INDUSTRIAL = "industrial"
INFRASTRUCTURE = "infrastructure"
@dataclass
class StructureAssessment:
structure_id: str
structure_type: StructureType
location: np.ndarray
damage_percentage: float
structural_integrity: float
fire_damage: float
water_damage: float
roof_condition: float
wall_condition: float
accessible: bool
search_priority: int
class DamageAssessor:
def __init__(self):
self.integrity_thresholds = {
'safe': 0.8,
'caution': 0.5,
'dangerous': 0.2
}
def assess_structure(self, rgb: np.ndarray,
thermal: np.ndarray,
lidar: np.ndarray = None) -> StructureAssessment:
"""Comprehensive structure assessment"""
# Visual damage analysis
visual_damage = self._analyze_visual_damage(rgb)
# Thermal damage analysis
thermal_damage = self._analyze_thermal_damage(thermal)
# Structural integrity (if LiDAR available)
integrity = self._assess_integrity(lidar) if lidar is not None else 0.5
# Calculate overall damage
overall_damage = (visual_damage * 0.5 +
thermal_damage * 0.3 +
(1 - integrity) * 0.2)
return StructureAssessment(
structure_id=f"STRUCT-{np.random.randint(1000, 9999)}",
structure_type=self._classify_structure(rgb),
location=np.array([rgb.shape[0]//2, rgb.shape[1]//2]),
damage_percentage=overall_damage * 100,
structural_integrity=integrity,
fire_damage=thermal_damage,
water_damage=self._detect_water_damage(rgb),
roof_condition=self._assess_roof(rgb),
wall_condition=self._assess_walls(rgb),
accessible=overall_damage < 0.7,
search_priority=self._calculate_priority(overall_damage, thermal_damage)
)
def _analyze_visual_damage(self, rgb: np.ndarray) -> float:
"""Analyze visual damage indicators"""
gray = np.mean(rgb, axis=2)
# Edge density indicates structural damage
from scipy import ndimage
edges = ndimage.sobel(gray)
edge_score = np.mean(edges > 50)
# Color irregularities
color_std = np.std(rgb, axis=(0,1))
color_anomaly = np.mean(color_std) / 100
return min(1.0, edge_score * 0.6 + color_anomaly * 0.4)
def _analyze_thermal_damage(self, thermal: np.ndarray) -> float:
"""Analyze thermal damage indicators"""
# Extreme temperatures indicate fire damage
hot_ratio = np.mean(thermal > 60)
cold_ratio = np.mean(thermal < 5)
return min(1.0, hot_ratio * 2 + cold_ratio)
def _assess_integrity(self, lidar: np.ndarray) -> float:
"""Assess structural integrity from LiDAR data"""
# Simple planarity check
h, w = lidar.shape
x, y = np.meshgrid(np.arange(w), np.arange(h))
A = np.column_stack([x.ravel(), y.ravel(), np.ones_like(x.ravel())])
result = np.linalg.lstsq(A, lidar.ravel(), rcond=None)
plane = (result[0][0] * x + result[0][1] * y + result[0][2])
deviation = np.std(lidar - plane)
integrity = max(0, 1 - deviation / 0.5)
return integrity
def _detect_water_damage(self, rgb: np.ndarray) -> float:
"""Detect water damage indicators"""
# Blue tint and stains
b_channel = rgb[:,:,2]
water_indicator = np.mean(b_channel > 150)
return min(1.0, water_indicator * 2)
def _assess_roof(self, rgb: np.ndarray) -> float:
"""Assess roof condition"""
# Top portion of image
top_third = rgb[:rgb.shape[0]//3, :]
damage = self._analyze_visual_damage(top_third)
return max(0, 1 - damage)
def _assess_walls(self, rgb: np.ndarray) -> float:
"""Assess wall condition"""
middle = rgb[rgb.shape[0]//3:2*rgb.shape[0]//3, :]
damage = self._analyze_visual_damage(middle)
return max(0, 1 - damage)
def _classify_structure(self, rgb: np.ndarray) -> StructureType:
"""Classify structure type"""
# Simplified - would use trained model
return StructureType.RESIDENTIAL
def _calculate_priority(self, damage: float,
thermal_damage: float) -> int:
"""Calculate search priority (1=highest)"""
if damage > 0.8 and thermal_damage > 0.3:
return 1
elif damage > 0.6:
return 2
elif damage > 0.4:
return 3
return 4
Communication Restoration
from typing import List, Dict, Tuple
from dataclasses import dataclass
from enum import Enum
class RelayStatus(Enum):
ACTIVE = "active"
LOW_BATTERY = "low_battery"
OFFLINE = "offline"
@dataclass
class CommunicationRelay:
relay_id: str
position: np.ndarray
status: RelayStatus
battery_level: float
coverage_radius: float
connected_users: int
class CommRestorationSystem:
def __init__(self, disaster_area: Tuple[float, float]):
self.area = disaster_area
self.relays: List[CommunicationRelay] = []
self.base_station = np.array([0, 0])
def deploy_relays(self, positions: List[np.ndarray],
battery_levels: List[float] = None):
"""Deploy communication relays"""
if battery_levels is None:
battery_levels = [100.0] * len(positions)
for i, (pos, batt) in enumerate(zip(positions, battery_levels)):
self.relays.append(CommunicationRelay(
relay_id=f"RELAY-{i+1:03d}",
position=pos,
status=RelayStatus.ACTIVE,
battery_level=batt,
coverage_radius=200,
connected_users=0
))
def optimize_relay_positions(self, priority_zones: List[np.ndarray],
num_relays: int = 8) -> List[np.ndarray]:
"""Optimize relay positions for maximum coverage"""
# K-means like placement
if len(priority_zones) <= num_relays:
return priority_zones[:num_relays]
# Initialize with priority zones
positions = priority_zones[:num_relays].copy()
# Iterative optimization
for _ in range(10):
# Assign zones to nearest relay
assignments = [[] for _ in range(num_relays)]
for zone in priority_zones:
distances = [np.linalg.norm(zone - pos) for pos in positions]
nearest = np.argmin(distances)
assignments[nearest].append(zone)
# Update relay positions to center of assigned zones
for i in range(num_relays):
if assignments[i]:
positions[i] = np.mean(assignments[i], axis=0)
return positions
def calculate_coverage(self) -> float:
"""Calculate percentage of area covered"""
total_area = self.area[0] * self.area[1]
covered_area = 0
for relay in self.relays:
if relay.status != RelayStatus.OFFLINE:
# Approximate coverage as circle
coverage = np.pi * relay.coverage_radius ** 2
covered_area += coverage
# Remove overlaps (simplified)
coverage_ratio = min(1.0, covered_area / total_area * 0.7)
return coverage_ratio * 100
def monitor_network_health(self) -> Dict:
"""Monitor communication network health"""
active = sum(1 for r in self.relays if r.status == RelayStatus.ACTIVE)
low_battery = sum(1 for r in self.relays if r.status == RelayStatus.LOW_BATTERY)
offline = sum(1 for r in self.relays if r.status == RelayStatus.OFFLINE)
total_users = sum(r.connected_users for r in self.relays)
return {
'active_relays': active,
'low_battery_relays': low_battery,
'offline_relays': offline,
'total_relays': len(self.relays),
'coverage_percentage': self.calculate_coverage(),
'connected_users': total_users,
'network_resilience': active / max(1, len(self.relays))
}
def simulate_battery_drain(self, hours: float = 24):
"""Simulate battery drain over time"""
drain_rate = 2.0 # % per hour
for relay in self.relays:
relay.battery_level -= drain_rate * hours
relay.battery_level = max(0, relay.battery_level)
if relay.battery_level < 20:
relay.status = RelayStatus.LOW_BATTERY
if relay.battery_level < 5:
relay.status = RelayStatus.OFFLINE
Supply Delivery System
from typing import List, Dict, Tuple
from dataclasses import dataclass
from enum import Enum
class SupplyType(Enum):
MEDICAL = "medical"
WATER = "water"
FOOD = "food"
SHELTER = "shelter"
COMMUNICATION = "communication"
class Priority(Enum):
CRITICAL = 1
HIGH = 2
MEDIUM = 3
LOW = 4
@dataclass
class SupplyRequest:
request_id: str
supply_type: SupplyType
location: np.ndarray
quantity: int
priority: Priority
timestamp: float
class SupplyDeliverySystem:
def __init__(self, depot_location: np.ndarray):
self.depot = depot_location
self.pending_requests: List[SupplyRequest] = []
self.active_deliveries: Dict[str, SupplyRequest] = {}
self.completed_deliveries: List[str] = []
def add_request(self, request: SupplyRequest):
"""Add supply request"""
self.pending_requests.append(request)
self._prioritize_requests()
def _prioritize_requests(self):
"""Sort requests by priority and urgency"""
self.pending_requests.sort(
key=lambda r: (r.priority.value, -r.timestamp))
def plan_delivery_batch(self, max_deliveries: int = 5) -> List[Dict]:
"""Plan batch of deliveries"""
batch = []
for request in self.pending_requests[:max_deliveries]:
route = self._plan_route(request)
batch.append({
'request': request,
'route': route,
'estimated_time': self._estimate_time(route),
'drone_requirements': self._get_drone_requirements(request)
})
return batch
def _plan_route(self, request: SupplyRequest) -> List[np.ndarray]:
"""Plan delivery route"""
waypoints = [
self.depot.copy(),
request.location.copy()
]
# Add intermediate waypoints for obstacle avoidance
mid_point = (self.depot + request.location) / 2
mid_point[2] = 50 # Altitude
waypoints.insert(1, mid_point)
return waypoints
def _estimate_time(self, route: List[np.ndarray]) -> float:
"""Estimate delivery time in minutes"""
total_distance = sum(
np.linalg.norm(route[i+1] - route[i])
for i in range(len(route)-1)
)
speed = 10 # m/s
return total_distance / speed / 60
def _get_drone_requirements(self, request: SupplyRequest) -> Dict:
"""Get drone requirements for delivery"""
weight_map = {
SupplyType.MEDICAL: 2.0,
SupplyType.WATER: 5.0,
SupplyType.FOOD: 3.0,
SupplyType.SHELTER: 8.0,
SupplyType.COMMUNICATION: 1.5
}
weight = weight_map.get(request.supply_type, 3.0)
volume = weight * 0.001 # Approximate
return {
'min_payload_kg': weight,
'min_volume_m3': volume,
'special_equipment': request.supply_type == SupplyType.MEDICAL,
'priority': request.priority.name
}
def track_delivery(self, request_id: str) -> Dict:
"""Track delivery status"""
if request_id in self.active_deliveries:
request = self.active_deliveries[request_id]
return {
'status': 'in_transit',
'location': request.location.tolist(),
'supply_type': request.supply_type.value
}
elif request_id in self.completed_deliveries:
return {'status': 'delivered'}
return {'status': 'pending'}
def generate_delivery_report(self) -> str:
"""Generate delivery status report"""
report = "=" * 50 + "\n"
report += "SUPPLY DELIVERY REPORT\n"
report += "=" * 50 + "\n\n"
report += f"Pending Requests: {len(self.pending_requests)}\n"
report += f"Active Deliveries: {len(self.active_deliveries)}\n"
report += f"Completed: {len(self.completed_deliveries)}\n\n"
# Breakdown by type
type_counts = {}
for req in self.pending_requests:
stype = req.supply_type.value
type_counts[stype] = type_counts.get(stype, 0) + 1
report += "Pending by Type:\n"
for stype, count in type_counts.items():
report += f" {stype}: {count}\n"
return report
Hands-On Project: Complete Disaster Response System
# disaster_response_system.py
from typing import List, Dict
from dataclasses import dataclass
from enum import Enum
class ResponsePhase(Enum):
INITIAL_ASSESSMENT = "initial_assessment"
SEARCH_RESCUE = "search_rescue"
RECOVERY = "recovery"
REHABILITATION = "rehabilitation"
class DisasterResponseSystem:
def __init__(self, disaster_type: str, affected_area: tuple):
self.disaster_type = disaster_type
self.affected_area = affected_area
self.mapper = RapidMapper()
self.assessor = DamageAssessor()
self.comm_system = CommRestorationSystem(affected_area)
self.supply_system = SupplyDeliverySystem(np.array([0, 0]))
self.phase = ResponsePhase.INITIAL_ASSESSMENT
self.situation_report = {}
def initial_assessment(self, sensor_data: Dict) -> Dict:
"""Perform initial disaster assessment"""
print("PHASE 1: Initial Assessment")
# Create damage map
damage_map = self.mapper.create_damage_map(
sensor_data['rgb_frames'],
sensor_data['thermal_frames'],
sensor_data['positions']
)
# Identify priority areas
priorities = self.mapper.prioritize_search_areas(
damage_map, sensor_data['thermal_frames'][0])
# Generate report
self.situation_report = self.mapper.generate_situation_report()
# Deploy communication relays
relay_positions = self.comm_system.optimize_relay_positions(
[p['location'] for p in priorities[:8]])
self.comm_system.deploy_relays(relay_positions)
self.phase = ResponsePhase.SEARCH_RESCUE
return {
'damage_map': damage_map,
'priority_areas': priorities[:10],
'situation_report': self.situation_report,
'communication_status': self.comm_system.monitor_network_health()
}
def search_rescue_operations(self, survivor_detections: List[Dict]) -> Dict:
"""Coordinate search and rescue"""
print("PHASE 2: Search & Rescue")
# Prioritize survivors
survivor_priorities = self._prioritize_survivors(survivor_detections)
# Deploy supply deliveries for critical survivors
critical_survivors = [s for s in survivor_priorities if s['priority'] == 1]
for survivor in critical_survivors:
request = SupplyRequest(
request_id=f"SUPPLY-{np.random.randint(1000, 9999)}",
supply_type=SupplyType.MEDICAL,
location=survivor['location'],
quantity=1,
priority=Priority.CRITICAL,
timestamp=0
)
self.supply_system.add_request(request)
# Plan delivery batch
delivery_batch = self.supply_system.plan_delivery_batch()
return {
'survivor_count': len(survivor_priorities),
'critical_survivors': len(critical_survivors),
'deliveries_planned': len(delivery_batch),
'communication_status': self.comm_system.monitor_network_health()
}
def _prioritize_survivors(self, detections: List[Dict]) -> List[Dict]:
"""Prioritize survivors for rescue"""
prioritized = []
for det in detections:
# Priority factors
heat_score = det.get('heat_signature', 0)
movement_score = det.get('movement', 0)
accessibility = det.get('accessibility', 0.5)
priority_score = (heat_score * 0.4 +
movement_score * 0.3 +
accessibility * 0.3)
priority = 1 if priority_score > 0.7 else \
2 if priority_score > 0.4 else 3
prioritized.append({
'location': det['location'],
'priority': priority,
'priority_score': priority_score
})
return sorted(prioritized, key=lambda x: x['priority_score'], reverse=True)
def recovery_operations(self) -> Dict:
"""Coordinate recovery operations"""
print("PHASE 3: Recovery")
# Assess infrastructure
infrastructure_status = {
'roads': self._assess_roads(),
'utilities': self._assess_utilities(),
'buildings': self._assess_buildings()
}
# Resource needs
resource_needs = self.situation_report.get('priority_response_needed', {})
return {
'infrastructure_status': infrastructure_status,
'resource_needs': resource_needs,
'timeline_estimate': self._estimate_recovery_timeline()
}
def _assess_roads(self) -> Dict:
return {'functional': 45, 'blocked': 35, 'damaged': 20}
def _assess_utilities(self) -> Dict:
return {'power': 30, 'water': 50, 'comms': 70}
def _assess_buildings(self) -> Dict:
return self.situation_report.get('damage_summary', {})
def _estimate_recovery_timeline(self) -> Dict:
return {
'critical_repairs': '1-2 weeks',
'infrastructure': '1-3 months',
'full_recovery': '6-12 months'
}
def generate_mission_report(self) -> str:
"""Generate comprehensive mission report"""
report = "=" * 60 + "\n"
report += "DISASTER RESPONSE MISSION REPORT\n"
report += "=" * 60 + "\n\n"
report += f"Disaster Type: {self.disaster_type}\n"
report += f"Affected Area: {self.affected_area[0]} km²\n"
report += f"Current Phase: {self.phase.value}\n\n"
report += "SITUATION SUMMARY:\n"
for key, value in self.situation_report.items():
report += f" {key}: {value}\n"
report += f"\nCOMMUNICATION STATUS:\n"
comm_status = self.comm_system.monitor_network_health()
for key, value in comm_status.items():
report += f" {key}: {value}\n"
return report
# Usage
system = DisasterResponseSystem(
disaster_type="Earthquake",
affected_area=(5.0, 5.0) # 5km x 5km
)
# Simulate sensor data
sensor_data = {
'rgb_frames': [np.random.randint(50, 200, (100, 100, 3), dtype=np.uint8)],
'thermal_frames': [np.random.uniform(15, 45, (100, 100))],
'positions': [np.array([50, 50, 0])]
}
# Run initial assessment
results = system.initial_assessment(sensor_data)
# Run search and rescue
survivor_detections = [
{'location': np.array([30, 40, 0]), 'heat_signature': 0.9, 'movement': 0.7},
{'location': np.array([70, 60, 0]), 'heat_signature': 0.8, 'movement': 0.3}
]
sar_results = system.search_rescue_operations(survivor_detections)
print(system.generate_mission_report())
Exercise
- Implement real-time path planning for obstacle avoidance in debris fields
- Create a multi-drone coordination system for large-scale operations
- Build a supply inventory management system for tracking resources
- Design an integration layer with existing emergency management systems
⚠️
Disaster response drone operations require special authorizations and coordination with emergency services. Always follow established incident command systems and local regulations.