Spaces:
Sleeping
Sleeping
File size: 2,422 Bytes
4904e85 14170d7 4904e85 14170d7 4904e85 14170d7 4904e85 14170d7 | 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 | """Task: Single Incident Response (Easy)."""
from __future__ import annotations
from pydantic import BaseModel, ConfigDict, Field
from src.city_schema import CitySchema
from src.models import Action, DispatchAction, IncidentSeverity, IncidentStatus, IncidentType, State, UnitType
from src.rewards import RewardCalculator
from src.state_machine import DispatchStateMachine
class SingleIncidentTask(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")
city_schema: CitySchema
seed: int | None = None
state_machine: DispatchStateMachine = Field(default=None, exclude=True)
def __init__(self, **data) -> None:
super().__init__(**data)
object.__setattr__(
self,
"state_machine",
DispatchStateMachine(schema=self.city_schema, seed=self.seed),
)
def reset(self, episode_id: str) -> State:
return self.state_machine.reset(task_id="single_incident", episode_id=episode_id)
def step(self, state: State, action: Action) -> tuple[State, object]:
return self.state_machine.step(state, action)
def is_terminal(self, state: State) -> bool:
return self.state_machine.is_terminal(state)
class SingleIncidentGrader:
def __init__(self) -> None:
self.reward_calculator = RewardCalculator()
def grade(self, state: State, rewards: list[float]) -> float:
"""Grade based on: correct unit dispatched, fast response, incident resolved."""
if not rewards:
return 0.0
incident = state.incidents.get("INC-001")
if incident is None:
return 0.0
score = 0.0
# Component 1: Was the incident resolved? (50% weight)
if incident.status.value == "RESOLVED":
score += 0.50
# Component 2: Correct unit type dispatched? (30% weight)
medic_dispatched = any(
u.unit_type.value == "MEDIC"
and (
u.assigned_incident_id == "INC-001"
or u.status.value in {"ON_SCENE", "DISPATCHED"}
)
for u in state.units.values()
)
if medic_dispatched:
score += 0.30
# Component 3: Speed — resolved within first 10 steps (20% weight)
if incident.status.value == "RESOLVED" and state.step_count <= 10:
score += 0.20
return max(0.0, min(1.0, score))
|