Spaces:
Sleeping
Sleeping
File size: 5,874 Bytes
4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 5b4eb04 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 5b4eb04 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 | 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 | """Dispatch protocol validation.
This module validates whether a dispatcher action is legal given the current state.
"""
from __future__ import annotations
from dataclasses import dataclass
from src.city_schema import CitySchema
from src.models import (
Action,
DispatchAction,
IncidentSeverity,
IncidentStatus,
State,
UnitStatus,
)
@dataclass(frozen=True)
class ValidationResult:
ok: bool
issues: list[str]
class DispatchProtocolValidator:
"""Validates dispatch actions against the environment state."""
@staticmethod
def _severity_rank(sev: IncidentSeverity) -> int:
if sev == IncidentSeverity.PRIORITY_1:
return 3
if sev == IncidentSeverity.PRIORITY_2:
return 2
return 1
def validate(self, schema: CitySchema, state: State, action: Action) -> ValidationResult:
issues: list[str] = []
errors: list[str] = []
def warn(msg: str) -> None:
issues.append(f"warn:{msg}")
def error(msg: str) -> None:
issues.append(msg)
errors.append(msg)
unit = state.units.get(action.unit_id)
if unit is None:
error(f"Unknown unit_id '{action.unit_id}'")
return ValidationResult(ok=False, issues=issues)
incident = state.incidents.get(action.incident_id)
if incident is None:
error(f"Unknown incident_id '{action.incident_id}'")
return ValidationResult(ok=False, issues=issues)
if action.action_type == DispatchAction.DISPATCH:
if unit.status != UnitStatus.AVAILABLE:
error(f"Unit '{unit.unit_id}' not available (status={unit.status})")
if unit.assigned_incident_id is not None:
error(f"Unit '{unit.unit_id}' already assigned to '{unit.assigned_incident_id}'")
if incident.status in {IncidentStatus.RESOLVED}:
error(f"Incident '{incident.incident_id}' already resolved")
# Triage type matching is a soft rule: record warning, do not invalidate.
required = schema.default_required_units.get(incident.incident_type)
if required is not None and required:
if unit.unit_type not in required:
warn(
f"Unit '{unit.unit_id}' type {unit.unit_type} mismatches "
f"recommended {incident.incident_type} types {required}"
)
elif action.action_type == DispatchAction.REASSIGN:
if incident.status in {IncidentStatus.RESOLVED}:
error(f"Incident '{incident.incident_id}' already resolved")
if unit.assigned_incident_id is None:
error(f"Unit '{unit.unit_id}' is not currently assigned")
elif unit.assigned_incident_id == incident.incident_id:
error(f"Unit '{unit.unit_id}' already assigned to incident '{incident.incident_id}'")
if unit.status not in {UnitStatus.DISPATCHED, UnitStatus.ON_SCENE, UnitStatus.TRANSPORTING}:
error(f"Unit '{unit.unit_id}' cannot be reassigned (status={unit.status})")
# Triage type matching is a soft rule: record warning, do not invalidate.
required = schema.default_required_units.get(incident.incident_type)
if required is not None and required:
if unit.unit_type not in required:
warn(
f"Unit '{unit.unit_id}' type {unit.unit_type} mismatches "
f"recommended {incident.incident_type} types {required}"
)
elif action.action_type == DispatchAction.CANCEL:
if unit.assigned_incident_id != incident.incident_id:
error(f"Unit '{unit.unit_id}' not assigned to incident '{incident.incident_id}'")
elif action.action_type == DispatchAction.MUTUAL_AID:
if incident.status == IncidentStatus.RESOLVED:
error(f"Incident '{incident.incident_id}' already resolved")
# Mutual aid only when local same-type units are exhausted.
same_type_available = any(
u.unit_type == unit.unit_type and u.status == UnitStatus.AVAILABLE
for u in state.units.values()
)
if same_type_available:
error(
f"Mutual aid not allowed: local {unit.unit_type} units still AVAILABLE"
)
elif action.action_type in {DispatchAction.UPGRADE, DispatchAction.DOWNGRADE}:
if incident.status == IncidentStatus.RESOLVED:
error(f"Incident '{incident.incident_id}' already resolved")
if action.priority_override is None:
error("priority_override required for severity change")
else:
cur = self._severity_rank(incident.severity)
new = self._severity_rank(action.priority_override)
if action.action_type == DispatchAction.UPGRADE and new <= cur:
error(
f"UPGRADE requires higher severity than {incident.severity}"
)
if action.action_type == DispatchAction.DOWNGRADE and new >= cur:
error(
f"DOWNGRADE requires lower severity than {incident.severity}"
)
elif action.action_type in {
DispatchAction.STAGE,
}:
# Allowed in principle; the state machine may choose to ignore.
if incident.status == IncidentStatus.RESOLVED:
error(f"Incident '{incident.incident_id}' already resolved")
else:
error(f"Unsupported action_type '{action.action_type}'")
return ValidationResult(ok=(len(errors) == 0), issues=issues)
|