Spaces:
Sleeping
Sleeping
File size: 5,202 Bytes
07473e9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 | """Render a DispatchSimulation as a human-readable text view for the LLM agent."""
from __future__ import annotations
from typing import List
from models import EmergencyCall, EmergencyUnit, Hospital, UnitStatus
from simulation import DispatchSimulation
from utils import calculate_distance, calculate_eta
# Maximum number of calls / units / outcomes to show in the text view.
# Truncation prevents context blow-up on the hard task (30 calls).
MAX_PENDING_CALLS = 8
MAX_BUSY_UNITS = 8
MAX_RECENT_OUTCOMES = 3
def _format_call(call: EmergencyCall, sim: DispatchSimulation) -> List[str]:
wait = sim.current_time - call.timestamp
rt = call.reported_type.value if call.reported_type else "unknown"
rs = call.reported_severity.value if call.reported_severity else "?"
return [
f' {call.call_id} [t={call.timestamp}min] "{call.caller_description}"',
(
f" location=({call.location.x}, {call.location.y}) "
f"reported={rt}/sev{rs} waiting={wait}min"
),
]
def _format_unit(unit: EmergencyUnit, sim: DispatchSimulation, pending: list) -> str:
base = (
f" {unit.unit_id:7s} | {unit.unit_type.value:14s} | "
f"pos=({unit.position.x:.1f}, {unit.position.y:.1f})"
)
if pending:
closest = min(pending, key=lambda c: calculate_distance(unit.position, c.location))
eta = calculate_eta(unit, closest.location)
base += f" closest_call_eta={eta:.1f}min ({closest.call_id})"
return base
def _format_busy_unit(unit: EmergencyUnit) -> str:
detail = unit.status.value
if unit.assigned_call_id:
detail += f" -> {unit.assigned_call_id}"
if unit.busy_until is not None and unit.status == UnitStatus.ON_SCENE:
detail += f" (free at t={unit.busy_until}min)"
return f" {unit.unit_id:7s} | {unit.unit_type.value:14s} | {detail}"
def _format_hospital(hosp: Hospital) -> str:
specs = []
if hosp.has_trauma_center:
specs.append("trauma")
if hosp.has_cardiac_unit:
specs.append("cardiac")
if hosp.has_stroke_unit:
specs.append("stroke")
status = "DIVERSION" if hosp.on_diversion else "open"
return (
f" {hosp.hospital_id} {hosp.name} ({hosp.position.x},{hosp.position.y}) "
f"beds={hosp.available_beds}/{hosp.capacity} "
f"specialties=[{','.join(specs) or 'none'}] status={status}"
)
def render_dispatch_center(sim: DispatchSimulation, task_name: str) -> str:
"""Pretty-print the current state for the LLM agent."""
lines: List[str] = []
lines.append("=== DISPATCHPULSE DISPATCH CENTER ===")
lines.append(
f"task={task_name} time={sim.current_time}min/"
f"{sim.config.time_limit_minutes}min "
f"scenario={sim.scenario_name}"
)
lines.append("")
# Pending calls (sorted by reported severity, then arrival time)
pending = sim.get_pending_calls()
pending_sorted = sorted(
pending,
key=lambda c: (
c.reported_severity.value if c.reported_severity else 5,
c.timestamp,
),
)
lines.append(f"PENDING CALLS ({len(pending_sorted)} total):")
if not pending_sorted:
lines.append(" (none)")
for call in pending_sorted[:MAX_PENDING_CALLS]:
lines.extend(_format_call(call, sim))
if len(pending_sorted) > MAX_PENDING_CALLS:
hidden = len(pending_sorted) - MAX_PENDING_CALLS
lines.append(f" ... and {hidden} more lower-priority calls")
lines.append("")
# Available units
available = sim.get_available_units()
lines.append(f"AVAILABLE UNITS ({len(available)} total):")
if not available:
lines.append(" (none — all units busy)")
for unit in available:
lines.append(_format_unit(unit, sim, pending_sorted[:MAX_PENDING_CALLS]))
lines.append("")
# Busy units
busy = [u for u in sim.units.values() if u.status != UnitStatus.AVAILABLE]
if busy:
lines.append(f"BUSY UNITS ({len(busy)} total):")
for unit in busy[:MAX_BUSY_UNITS]:
lines.append(_format_busy_unit(unit))
if len(busy) > MAX_BUSY_UNITS:
lines.append(f" ... and {len(busy) - MAX_BUSY_UNITS} more busy units")
lines.append("")
# Hospitals
lines.append("HOSPITALS:")
for hosp in sim.hospitals.values():
lines.append(_format_hospital(hosp))
lines.append("")
# Recent outcomes
if sim.completed_calls:
recent = sim.completed_calls[-MAX_RECENT_OUTCOMES:]
lines.append("RECENT OUTCOMES:")
for r in recent:
mark = "OK " if r["outcome_score"] >= 0.5 else "BAD"
lines.append(
f" [{mark}] {r['call_id']} {r['true_type']} sev{r['true_severity']} "
f"response={r['response_time']:.1f}min outcome={r['outcome_score']:.2f}"
)
lines.append("")
# Stats footer
lines.append(
f"STATS: total={sim.total_calls()} completed={len(sim.completed_calls)} "
f"timed_out={len(sim.timed_out_calls)} pending={len(pending_sorted)}"
)
if sim.episode_done:
lines.append("EPISODE: DONE")
return "\n".join(lines)
|