Recipe 14.8: Python Implementation Example

pip install boto3 ortools

import json
import math
import logging
import datetime
from datetime import timezone
from decimal import Decimal
from typing import Optional

import boto3
from botocore.config import Config
from ortools.linear_solver import pywraplp

logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

BOTO3_RETRY_CONFIG = Config(retries={"max_attempts": 3, "mode": "adaptive"})
dynamodb = boto3.resource("dynamodb", config=BOTO3_RETRY_CONFIG)

# Table where we store dispatch decisions for audit trail.
DISPATCH_TABLE_NAME = "ems-dispatch-decisions"

# Response time targets (seconds). These come from NFPA 1710 and local
# medical direction protocols. Priority 1 = life-threatening emergency.
RESPONSE_TIME_TARGETS = {
    1: 480,   # 8 minutes for Priority 1 (cardiac arrest, major trauma)
    2: 600,   # 10 minutes for Priority 2 (chest pain, difficulty breathing)
    3: 900,   # 15 minutes for Priority 3 (falls, minor injuries)
    4: 1200,  # 20 minutes for Priority 4 (non-emergency transports)
    5: 1800,  # 30 minutes for Priority 5 (scheduled transfers)
}

# Unit capability levels. ALS units carry paramedics with cardiac monitors,
# intubation equipment, and IV medications. BLS units have EMTs with basic
# interventions. You cannot send a BLS unit to an ALS-required call.
CAPABILITY_LEVELS = {
    "ALS": 2,  # Advanced Life Support (paramedic)
    "BLS": 1,  # Basic Life Support (EMT)
}

# Scoring weights for the dispatch optimizer. These control the trade-off
# between "send the fastest unit" and "preserve system coverage."
# Tuned per system based on historical performance analysis.
DISPATCH_WEIGHTS = {
    "response_time": 0.60,   # dominant factor: get there fast
    "coverage_impact": 0.25, # don't leave zones uncovered
    "fatigue": 0.10,         # crew welfare (hours on shift)
    "workload": 0.05,        # balance calls across crews
}

# For Priority 1 calls, we override the weights to prioritize pure speed.
# Coverage concerns take a back seat when someone is dying.
PRIORITY_1_WEIGHTS = {
    "response_time": 0.90,
    "coverage_impact": 0.10,
    "fatigue": 0.0,
    "workload": 0.0,
}

# Coverage threshold: a zone is "covered" if at least one available unit
# can reach its centroid within this many seconds.
COVERAGE_THRESHOLD_SECONDS = 480  # 8 minutes

# Maximum repositioning moves per optimization cycle. Moving too many units
# at once creates chaos and dispatcher confusion.
MAX_REPOSITION_MOVES = 3

# Average ambulance speed for travel time estimation (km/h).
# Emergency response (lights and sirens) vs. non-emergency.
SPEED_EMERGENCY_KMH = 55   # urban average with L&S
SPEED_NORMAL_KMH = 35      # urban average without L&S

def build_initial_fleet():
    """
    Create a synthetic fleet for demonstration. In production, this state
    lives in DynamoDB and is updated by the GPS ingestion pipeline.

    Each unit has:
    - unit_id: unique identifier (matches the MDT/radio callsign)
    - capability: ALS or BLS
    - latitude/longitude: current GPS position
    - status: AVAILABLE, DISPATCHED, EN_ROUTE, ON_SCENE, TRANSPORTING, AT_HOSPITAL
    - zone: which coverage zone the unit is currently in
    - shift_start: when this crew's shift began (for fatigue scoring)
    - calls_today: how many calls this crew has run (for workload balancing)
    """
    fleet = [
        {
            "unit_id": "MEDIC-1",
            "capability": "ALS",
            "latitude": 38.9072,
            "longitude": -77.0369,
            "status": "AVAILABLE",
            "zone": "ZONE-A",
            "shift_start": "2026-06-01T06:00:00Z",
            "calls_today": 3,
        },
        {
            "unit_id": "MEDIC-3",
            "capability": "ALS",
            "latitude": 38.9200,
            "longitude": -77.0200,
            "status": "AVAILABLE",
            "zone": "ZONE-B",
            "shift_start": "2026-06-01T06:00:00Z",
            "calls_today": 5,
        },
        {
            "unit_id": "MEDIC-5",
            "capability": "ALS",
            "latitude": 38.8950,
            "longitude": -77.0500,
            "status": "EN_ROUTE",
            "zone": "ZONE-C",
            "shift_start": "2026-06-01T06:00:00Z",
            "calls_today": 4,
        },
        {
            "unit_id": "BLS-2",
            "capability": "BLS",
            "latitude": 38.9100,
            "longitude": -77.0450,
            "status": "AVAILABLE",
            "zone": "ZONE-A",
            "shift_start": "2026-06-01T18:00:00Z",
            "calls_today": 1,
        },
        {
            "unit_id": "BLS-4",
            "capability": "BLS",
            "latitude": 38.9300,
            "longitude": -77.0100,
            "status": "AVAILABLE",
            "zone": "ZONE-D",
            "shift_start": "2026-06-01T18:00:00Z",
            "calls_today": 2,
        },
    ]
    return fleet

def get_available_units(fleet, required_capability="BLS"):
    """
    Filter fleet to units that are available and meet the capability requirement.

    ALS units can handle both ALS and BLS calls (paramedics can do everything
    EMTs can do, plus more). BLS units can only handle BLS calls.
    """
    available = []
    for unit in fleet:
        if unit["status"] != "AVAILABLE":
            continue
        # ALS units satisfy any capability requirement.
        # BLS units only satisfy BLS requirements.
        unit_level = CAPABILITY_LEVELS[unit["capability"]]
        required_level = CAPABILITY_LEVELS[required_capability]
        if unit_level >= required_level:
            available.append(unit)
    return available

def validate_gps_fix(gps_event, previous_fix=None, service_area_bounds=None):
    """
    Validate a GPS position fix before accepting it into fleet state.

    In production, these checks run in the Kinesis consumer Lambda before
    the DynamoDB write. Invalid fixes are logged to a GPS anomalies table
    for fleet maintenance review but do not update the unit's position.

    Args:
        gps_event: dict with unit_id, latitude, longitude, timestamp, speed
        previous_fix: the last accepted fix for this unit (for movement checks)
        service_area_bounds: dict with min_lat, max_lat, min_lon, max_lon

    Returns:
        (is_valid, reason) tuple. If invalid, reason explains why.
    """
    lat = gps_event.get("latitude", 0)
    lon = gps_event.get("longitude", 0)

    # Check for GPS default/null coordinates (common when device loses lock)
    if lat == 0 and lon == 0:
        return False, "null_coordinates"

    # Bounds check: must be within service area (with buffer for mutual aid)
    if service_area_bounds:
        buffer = 0.5  # ~55 km buffer for mutual aid runs
        if not (service_area_bounds["min_lat"] - buffer <= lat
                <= service_area_bounds["max_lat"] + buffer):
            return False, "latitude_out_of_bounds"
        if not (service_area_bounds["min_lon"] - buffer <= lon
                <= service_area_bounds["max_lon"] + buffer):
            return False, "longitude_out_of_bounds"

    # Speed plausibility: ambulances do not exceed 200 km/h
    if gps_event.get("speed", 0) > 200:
        return False, "speed_implausible"

    # Movement plausibility: check against previous fix
    if previous_fix:
        distance_km = haversine_distance_km(
            previous_fix["latitude"], previous_fix["longitude"], lat, lon
        )
        time_delta_s = (
            gps_event["timestamp"] - previous_fix["timestamp"]
        ).total_seconds()
        if time_delta_s > 0 and distance_km / (time_delta_s / 3600) > 200:
            return False, "impossible_movement"

    return True, "ok"

def haversine_distance_km(lat1, lon1, lat2, lon2):
    """
    Calculate straight-line distance between two GPS coordinates in kilometers.
    This is the "as the crow flies" distance. Real road distance is typically
    1.3x to 1.6x longer in urban areas (the "circuity factor").
    """
    R = 6371  # Earth's radius in km
    dlat = math.radians(lat2 - lat1)
    dlon = math.radians(lon2 - lon1)
    a = (
        math.sin(dlat / 2) ** 2
        + math.cos(math.radians(lat1))
        * math.cos(math.radians(lat2))
        * math.sin(dlon / 2) ** 2
    )
    c = 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a))
    return R * c

def estimate_travel_time_seconds(origin_lat, origin_lon, dest_lat, dest_lon,
                                  emergency=True):
    """
    Estimate travel time in seconds between two points.

    In production, replace this with a call to Amazon Location Service:
        client.calculate_route(
            CalculatorName="ems-router",
            DeparturePosition=[origin_lon, origin_lat],
            DestinationPosition=[dest_lon, dest_lat],
            TravelMode="Car",
            DepartNow=True
        )

    The Location Service response includes DurationSeconds accounting for
    real-time traffic. You'd then apply an emergency vehicle multiplier
    (typically 0.7x to 0.8x of civilian travel time for L&S response).

    For this example, we use Haversine distance with a circuity factor
    and speed estimate. It's wrong, but it's wrong in a predictable way
    that lets you see the optimization logic clearly.
    """
    straight_line_km = haversine_distance_km(
        origin_lat, origin_lon, dest_lat, dest_lon
    )

    # Urban circuity factor: roads aren't straight lines.
    # 1.4 is typical for grid-pattern cities. Winding suburbs might be 1.6+.
    road_distance_km = straight_line_km * 1.4

    speed_kmh = SPEED_EMERGENCY_KMH if emergency else SPEED_NORMAL_KMH
    travel_time_hours = road_distance_km / speed_kmh
    travel_time_seconds = travel_time_hours * 3600

    return travel_time_seconds

def compute_coverage_impact(unit, fleet, demand_zones):
    """
    Estimate how much system coverage degrades if we dispatch this unit.

    The idea: if this unit is the only one covering its zone, removing it
    creates a critical gap. If three other units can also cover that zone,
    removing this one barely matters.

    Returns a score from 0.0 (no impact) to 1.0 (critical gap created).
    """
    zone = unit["zone"]

    # Count how many other available units can cover this zone
    other_units_in_zone = 0
    for other in fleet:
        if other["unit_id"] == unit["unit_id"]:
            continue
        if other["status"] != "AVAILABLE":
            continue
        if other["zone"] == zone:
            other_units_in_zone += 1

    # Also check adjacent units that could reach this zone within threshold
    # (simplified: in production, use actual travel times to zone centroid)
    zone_centroid = demand_zones.get(zone, {}).get("centroid")
    if zone_centroid:
        for other in fleet:
            if other["unit_id"] == unit["unit_id"]:
                continue
            if other["status"] != "AVAILABLE":
                continue
            if other["zone"] == zone:
                continue  # already counted
            travel = estimate_travel_time_seconds(
                other["latitude"], other["longitude"],
                zone_centroid[0], zone_centroid[1],
                emergency=True,
            )
            if travel <= COVERAGE_THRESHOLD_SECONDS:
                other_units_in_zone += 0.5  # partial credit for adjacent coverage

    # Convert to impact score: fewer backup units = higher impact
    if other_units_in_zone >= 2:
        return 0.1  # well-covered zone, minimal impact
    elif other_units_in_zone >= 1:
        return 0.4  # one backup, moderate concern
    else:
        return 0.9  # no backup, critical gap if we send this unit

def normalize_value(value, max_value):
    """Normalize a value to 0.0-1.0 range. Clamps at 1.0."""
    return min(value / max_value, 1.0)

def score_candidates(call, fleet, demand_zones):
    """
    Score all available, capable units for a given call.

    Args:
        call: dict with location, priority, required_capability, nature_code
        fleet: list of all unit dicts (current fleet state)
        demand_zones: dict of zone_id -> {centroid, demand_weight}

    Returns:
        List of candidate dicts sorted by score (ascending = best first).
        Each candidate includes unit_id, travel_time, score, and coverage_impact.
    """
    required_cap = call.get("required_capability", "BLS")
    available = get_available_units(fleet, required_cap)

    if not available:
        logger.warning("No available units meet capability requirement: %s", required_cap)
        return []

    # Select weight profile based on call priority
    weights = PRIORITY_1_WEIGHTS if call["priority"] == 1 else DISPATCH_WEIGHTS

    candidates = []
    now = datetime.datetime.now(timezone.utc)

    for unit in available:
        # Travel time from unit's current position to call location
        travel_time = estimate_travel_time_seconds(
            unit["latitude"], unit["longitude"],
            call["location"]["lat"], call["location"]["lng"],
            emergency=True,
        )

        # Coverage impact: what happens to system coverage if we send this unit?
        coverage_impact = compute_coverage_impact(unit, fleet, demand_zones)

        # Fatigue: hours on shift. Slight penalty after 10 hours.
        shift_start = datetime.datetime.fromisoformat(
            unit["shift_start"].replace("Z", "+00:00")
        )
        hours_on_shift = (now - shift_start).total_seconds() / 3600
        fatigue_penalty = max(0.0, (hours_on_shift - 10) * 0.1)

        # Workload: calls run today, normalized against a reasonable max
        workload_score = normalize_value(unit["calls_today"], 10)

        # Composite score (lower is better)
        score = (
            weights["response_time"] * normalize_value(travel_time, 1200)
            + weights["coverage_impact"] * coverage_impact
            + weights["fatigue"] * normalize_value(fatigue_penalty, 1.0)
            + weights["workload"] * workload_score
        )

        candidates.append({
            "unit_id": unit["unit_id"],
            "capability": unit["capability"],
            "travel_time_seconds": round(travel_time),
            "coverage_impact": round(coverage_impact, 2),
            "score": round(score, 4),
        })

    # Sort by score ascending (best candidate first)
    candidates.sort(key=lambda c: c["score"])
    return candidates

def build_hospital_status():
    """
    Synthetic hospital status data. In production, this comes from hospital
    systems via EventBridge (diversion status, ED census, bed availability)
    and is stored in DynamoDB for fast lookup.
    """
    hospitals = [
        {
            "hospital_id": "HOSP-GWU",
            "name": "GW University Hospital",
            "location": {"lat": 38.9007, "lng": -77.0508},
            "capabilities": ["cath_lab", "interventional_cardiology",
                             "level_1_trauma", "stroke_center"],
            "ed_capacity": 45,
            "current_ed_census": 24,
            "diversion_status": "OPEN",  # OPEN, CONDITIONAL, FULL_DIVERSION
        },
        {
            "hospital_id": "HOSP-MEDSTAR",
            "name": "MedStar Washington Hospital Center",
            "location": {"lat": 38.9296, "lng": -77.0164},
            "capabilities": ["cath_lab", "interventional_cardiology",
                             "level_1_trauma", "stroke_center", "burn_center"],
            "ed_capacity": 60,
            "current_ed_census": 52,
            "diversion_status": "CONDITIONAL",
        },
        {
            "hospital_id": "HOSP-SIBLEY",
            "name": "Sibley Memorial Hospital",
            "location": {"lat": 38.9378, "lng": -77.1052},
            "capabilities": ["stroke_center", "general_surgery"],
            "ed_capacity": 30,
            "current_ed_census": 12,
            "diversion_status": "OPEN",
        },
    ]
    return hospitals

def select_hospital(patient_needs, unit_location, hospitals):
    """
    Recommend the best destination hospital for a patient.

    Args:
        patient_needs: dict with required_capabilities (list), acuity_level (1-5)
        unit_location: dict with lat, lng (ambulance's current position)
        hospitals: list of hospital status dicts

    Returns:
        Best hospital dict with transport_time and score, or None if no
        eligible hospital found (which would be a serious system failure
        in a real deployment).
    """
    eligible = []

    for hospital in hospitals:
        # Hard filter: must have all required capabilities
        has_all_caps = all(
            cap in hospital["capabilities"]
            for cap in patient_needs["required_capabilities"]
        )
        if not has_all_caps:
            continue

        # Hard filter: diversion status
        if hospital["diversion_status"] == "FULL_DIVERSION":
            continue
        if (hospital["diversion_status"] == "CONDITIONAL"
                and patient_needs["acuity_level"] > 2):
            # Conditional diversion: only accepting critical patients (acuity 1-2)
            continue

        # Transport time from unit's current location to hospital
        transport_seconds = estimate_travel_time_seconds(
            unit_location["lat"], unit_location["lng"],
            hospital["location"]["lat"], hospital["location"]["lng"],
            emergency=True,
        )

        # Capacity score: how full is the ED? (0.0 = empty, 1.0 = at capacity)
        capacity_ratio = hospital["current_ed_census"] / hospital["ed_capacity"]

        # Composite destination score (lower is better)
        dest_score = (
            0.50 * normalize_value(transport_seconds, 1800)  # transport time
            + 0.35 * capacity_ratio                           # ED crowding
            + 0.15 * (0.0 if len(hospital["capabilities"]) > 3 else 0.3)
            # slight preference for higher-capability hospitals
        )

        eligible.append({
            **hospital,
            "transport_time_seconds": round(transport_seconds),
            "transport_time_minutes": round(transport_seconds / 60, 1),
            "capacity_ratio": round(capacity_ratio, 2),
            "score": round(dest_score, 4),
        })

    if not eligible:
        logger.error("No eligible hospitals found for patient needs: %s",
                     patient_needs)
        return None

    # Sort by score ascending (best destination first)
    eligible.sort(key=lambda h: h["score"])
    return eligible[0]

def build_demand_zones():
    """
    Define coverage zones with centroids and demand weights.
    In production, these come from the demand forecast model (SageMaker endpoint)
    which predicts call probability per zone for the next 30 minutes.
    """
    zones = {
        "ZONE-A": {"centroid": (38.9072, -77.0369), "demand_weight": 0.30},
        "ZONE-B": {"centroid": (38.9200, -77.0200), "demand_weight": 0.25},
        "ZONE-C": {"centroid": (38.8950, -77.0500), "demand_weight": 0.20},
        "ZONE-D": {"centroid": (38.9300, -77.0100), "demand_weight": 0.15},
        "ZONE-E": {"centroid": (38.8800, -77.0300), "demand_weight": 0.10},
    }
    return zones

def identify_coverage_gaps(fleet, demand_zones):
    """
    Find zones where no available unit can respond within the coverage threshold.

    Returns a list of gap dicts: zone_id, centroid, best_current_time, demand_weight.
    These are the zones the repositioning optimizer needs to fix.
    """
    available_units = [u for u in fleet if u["status"] == "AVAILABLE"]
    gaps = []

    for zone_id, zone_info in demand_zones.items():
        # Skip very low demand zones (not worth repositioning for)
        if zone_info["demand_weight"] < 0.05:
            continue

        centroid = zone_info["centroid"]

        # Find the fastest available unit that could reach this zone
        best_time = float("inf")
        for unit in available_units:
            travel = estimate_travel_time_seconds(
                unit["latitude"], unit["longitude"],
                centroid[0], centroid[1],
                emergency=True,
            )
            best_time = min(best_time, travel)

        if best_time > COVERAGE_THRESHOLD_SECONDS:
            gaps.append({
                "zone_id": zone_id,
                "centroid": centroid,
                "best_current_time": round(best_time),
                "demand_weight": zone_info["demand_weight"],
            })

    return gaps

def optimize_repositioning(fleet, demand_zones):
    """
    Solve the repositioning problem: which idle units should move where
    to close coverage gaps?

    This is formulated as a weighted set-covering assignment problem:
    - Decision variables: binary (assign unit i to gap j, yes/no)
    - Objective: minimize total repositioning cost weighted by coverage priority
    - Constraints: each gap covered by at most one unit, total moves <= MAX

    Uses OR-Tools' linear solver (SCIP backend) which handles this scale
    (5-50 units, 5-200 zones) in milliseconds.
    """
    gaps = identify_coverage_gaps(fleet, demand_zones)

    if not gaps:
        logger.info("Coverage is adequate. No repositioning needed.")
        return []

    available_units = [u for u in fleet if u["status"] == "AVAILABLE"]

    if not available_units:
        logger.warning("No available units for repositioning.")
        return []

    logger.info("Found %d coverage gaps. %d units available for repositioning.",
                len(gaps), len(available_units))

    # Create the solver
    solver = pywraplp.Solver.CreateSolver("SCIP")
    if not solver:
        logger.error("Could not create SCIP solver. Is OR-Tools installed correctly?")
        return []

    # Decision variables: x[i][j] = 1 if unit i is assigned to cover gap j
    x = {}
    for i, unit in enumerate(available_units):
        for j, gap in enumerate(gaps):
            x[i, j] = solver.IntVar(0, 1, f"x_{i}_{j}")

    # Constraint: each unit can be assigned to at most one gap
    for i in range(len(available_units)):
        solver.Add(
            sum(x[i, j] for j in range(len(gaps))) <= 1
        )

    # Constraint: each gap gets at most one unit assigned
    for j in range(len(gaps)):
        solver.Add(
            sum(x[i, j] for i in range(len(available_units))) <= 1
        )

    # Constraint: total moves limited
    solver.Add(
        sum(x[i, j]
            for i in range(len(available_units))
            for j in range(len(gaps)))
        <= MAX_REPOSITION_MOVES
    )

    # Objective: minimize repositioning cost, weighted by gap priority.
    # Cost = travel time to reach the gap's centroid.
    # Priority = demand weight of the uncovered zone (higher demand = more important to cover).
    objective = solver.Objective()
    for i, unit in enumerate(available_units):
        for j, gap in enumerate(gaps):
            travel = estimate_travel_time_seconds(
                unit["latitude"], unit["longitude"],
                gap["centroid"][0], gap["centroid"][1],
                emergency=False,  # repositioning is non-emergency driving
            )
            # Cost coefficient: travel time penalized, but reduced for high-demand gaps
            # (we're willing to drive farther to cover a high-demand zone)
            cost = travel * (1.0 - 0.5 * gap["demand_weight"])
            objective.SetCoefficient(x[i, j], cost)

    objective.SetMinimization()

    # Solve
    status = solver.Solve()

    if status not in (pywraplp.Solver.OPTIMAL, pywraplp.Solver.FEASIBLE):
        logger.warning("Repositioning solver did not find a feasible solution.")
        return []

    # Extract solution: which units move where?
    moves = []
    for i, unit in enumerate(available_units):
        for j, gap in enumerate(gaps):
            if x[i, j].solution_value() > 0.5:
                travel = estimate_travel_time_seconds(
                    unit["latitude"], unit["longitude"],
                    gap["centroid"][0], gap["centroid"][1],
                    emergency=False,
                )
                moves.append({
                    "unit_id": unit["unit_id"],
                    "from_zone": unit["zone"],
                    "to_zone": gap["zone_id"],
                    "target_position": {
                        "lat": gap["centroid"][0],
                        "lng": gap["centroid"][1],
                    },
                    "repositioning_time_seconds": round(travel),
                    "gap_demand_weight": gap["demand_weight"],
                    "reason": f"Coverage gap in {gap['zone_id']} "
                              f"(best current response: {gap['best_current_time']}s, "
                              f"threshold: {COVERAGE_THRESHOLD_SECONDS}s)",
                })

    logger.info("Repositioning solution: %d moves.", len(moves))
    return moves

def store_dispatch_decision(decision):
    """
    Write a dispatch decision record to DynamoDB.

    This is the audit trail. Every field matters for post-incident review:
    - Which unit was assigned and why?
    - What were the alternatives?
    - What was the estimated response time?
    - What hospital was recommended?
    - Did the dispatcher accept or override the recommendation?

    In production, this table uses an IAM policy that denies UpdateItem and
    DeleteItem for all principals except a break-glass role. Records are
    immutable once written. Retention: 10 years minimum (state EMS record
    retention requirements vary; default to the most stringent). After 12
    months, records export to S3 with Object Lock in Compliance mode for
    tamper-evident long-term archival.
    """
    table = dynamodb.Table(DISPATCH_TABLE_NAME)

    # DynamoDB requires Decimal for numeric values, not float.
    # This is a known boto3 gotcha that will raise TypeError if you forget.
    record = json.loads(json.dumps(decision), parse_float=Decimal)

    table.put_item(Item=record)
    logger.info("Stored dispatch decision: %s", decision.get("call_id", "unknown"))
    return record

def dispatch_ambulance(call):
    """
    Run the full dispatch optimization for a single incoming call.

    Args:
        call: dict with:
            - call_id: unique identifier from CAD system
            - location: {lat, lng}
            - priority: 1-5 (1 = life-threatening)
            - required_capability: "ALS" or "BLS"
            - nature_code: e.g., "CHEST_PAIN", "FALL", "MVA"
            - patient_age: (optional) for clinical context

    Returns:
        Complete dispatch decision dict with assigned unit, recommended hospital,
        coverage impact assessment, and all candidates considered.
    """
    logger.info("=" * 60)
    logger.info("DISPATCH REQUEST: %s", call["call_id"])
    logger.info("  Priority: %d | Nature: %s | Capability: %s",
                call["priority"], call["nature_code"], call["required_capability"])
    logger.info("  Location: (%.4f, %.4f)",
                call["location"]["lat"], call["location"]["lng"])

    # Load current fleet state (in production: read from DynamoDB)
    fleet = build_initial_fleet()
    demand_zones = build_demand_zones()

    # Score all available, capable units
    candidates = score_candidates(call, fleet, demand_zones)

    if not candidates:
        logger.error("NO AVAILABLE UNITS. Call %s cannot be assigned.", call["call_id"])
        return {"call_id": call["call_id"], "status": "UNASSIGNED", "reason": "no_units"}

    # Best candidate is first in the sorted list
    assigned = candidates[0]
    logger.info("  Assigned: %s (travel: %ds, score: %.4f)",
                assigned["unit_id"], assigned["travel_time_seconds"], assigned["score"])

    # Select destination hospital based on call nature
    # (In reality, hospital selection happens after on-scene assessment,
    # but we compute a preliminary recommendation at dispatch time)
    patient_needs = _infer_patient_needs(call)
    hospitals = build_hospital_status()

    # Use call location as proxy for unit location (unit hasn't arrived yet)
    recommended_hospital = select_hospital(
        patient_needs, call["location"], hospitals
    )

    if recommended_hospital:
        logger.info("  Hospital: %s (transport: %.1f min, capacity: %d%%)",
                    recommended_hospital["name"],
                    recommended_hospital["transport_time_minutes"],
                    int(recommended_hospital["capacity_ratio"] * 100))

    # Build the complete dispatch decision record
    decision = {
        "call_id": call["call_id"],
        "call_priority": call["priority"],
        "call_location": call["location"],
        "nature_code": call["nature_code"],
        "assigned_unit": {
            "unit_id": assigned["unit_id"],
            "capability": assigned["capability"],
            "estimated_response_time_seconds": assigned["travel_time_seconds"],
        },
        "recommended_hospital": {
            "hospital_id": recommended_hospital["hospital_id"],
            "name": recommended_hospital["name"],
            "transport_time_minutes": recommended_hospital["transport_time_minutes"],
            "capacity_ratio": recommended_hospital["capacity_ratio"],
        } if recommended_hospital else None,
        "coverage_impact": {
            "assigned_unit_impact": assigned["coverage_impact"],
            "backup_available": assigned["coverage_impact"] < 0.5,
        },
        "candidates_evaluated": len(candidates),
        "all_candidates": candidates[:5],  # top 5 for audit trail
        "decision_timestamp": datetime.datetime.now(timezone.utc).isoformat(),
        "target_response_time_seconds": RESPONSE_TIME_TARGETS[call["priority"]],
        "meets_target": (
            assigned["travel_time_seconds"]
            <= RESPONSE_TIME_TARGETS[call["priority"]]
        ),
        # Dispatcher-in-the-loop fields (populated by the dispatcher console
        # in production; here we simulate auto-accept for Priority 1):
        "dispatcher_action": "auto_dispatched" if call["priority"] == 1 else "pending",
        "dispatcher_id": None,  # filled by dispatcher console on accept/override
        "override_reason_code": None,
    }

    logger.info("  Meets target: %s (%ds vs %ds threshold)",
                decision["meets_target"],
                assigned["travel_time_seconds"],
                RESPONSE_TIME_TARGETS[call["priority"]])
    logger.info("=" * 60)

    # In production, persist the decision for the audit trail:
    # store_dispatch_decision(decision)

    return decision

def _infer_patient_needs(call):
    """
    Map call nature codes to hospital capability requirements.
    In production, this mapping comes from medical direction protocols
    and is maintained by the EMS medical director.
    """
    nature_to_capabilities = {
        "CHEST_PAIN": {"required_capabilities": ["cath_lab"], "acuity_level": 1},
        "STEMI": {"required_capabilities": ["cath_lab", "interventional_cardiology"],
                  "acuity_level": 1},
        "STROKE": {"required_capabilities": ["stroke_center"], "acuity_level": 1},
        "TRAUMA_MAJOR": {"required_capabilities": ["level_1_trauma"], "acuity_level": 1},
        "FALL": {"required_capabilities": [], "acuity_level": 3},
        "MVA": {"required_capabilities": ["general_surgery"], "acuity_level": 2},
        "DIFFICULTY_BREATHING": {"required_capabilities": [], "acuity_level": 2},
    }
    return nature_to_capabilities.get(
        call["nature_code"],
        {"required_capabilities": [], "acuity_level": 3},
    )

# --- Run the full example ---
if __name__ == "__main__":
    # Simulate an incoming Priority 1 cardiac call
    incoming_call = {
        "call_id": "CAD-2026-0601-1847",
        "location": {"lat": 38.9100, "lng": -77.0350},
        "priority": 1,
        "required_capability": "ALS",
        "nature_code": "CHEST_PAIN",
        "patient_age": 67,
    }

    # Run dispatch optimization
    decision = dispatch_ambulance(incoming_call)
    print("\n--- DISPATCH DECISION ---")
    print(json.dumps(decision, indent=2, default=str))

    # Run background repositioning optimization
    print("\n--- REPOSITIONING OPTIMIZATION ---")
    fleet = build_initial_fleet()
    demand_zones = build_demand_zones()
    moves = optimize_repositioning(fleet, demand_zones)
    for move in moves:
        print(f"  MOVE: {move['unit_id']} -> {move['to_zone']} "
              f"({move['repositioning_time_seconds']}s) | {move['reason']}")
    if not moves:
        print("  No repositioning needed. Coverage is adequate.")