Spaces:
Sleeping
Sleeping
File size: 3,653 Bytes
6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 4904e85 6172160 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 | """Renderer/judge for normalized dispatch phraseology.
This repository originally shipped an ATC phraseology module. After the domain
pivot to 911 dispatch, we keep the same public symbols (`PhraseologyRenderer`,
`PhraseologyJudge`) but implement dispatch-oriented rendering and scoring.
"""
from __future__ import annotations
from pydantic import BaseModel
from src.models import Action, DispatchAction
class PhraseologyRenderer(BaseModel):
"""Converts structured dispatch actions to standardized strings."""
def render(self, action: Action) -> str:
if action.action_type == DispatchAction.DISPATCH:
return f"DISPATCH {action.unit_id} -> {action.incident_id}"
if action.action_type == DispatchAction.REASSIGN:
return f"REASSIGN {action.unit_id} -> {action.incident_id}"
if action.action_type == DispatchAction.CANCEL:
return f"CANCEL {action.unit_id} FROM {action.incident_id}"
if action.action_type == DispatchAction.STAGE:
return f"STAGE {action.unit_id} FOR {action.incident_id}"
if action.action_type == DispatchAction.MUTUAL_AID:
return f"MUTUAL_AID {action.unit_id} -> {action.incident_id}"
if action.action_type in {DispatchAction.UPGRADE, DispatchAction.DOWNGRADE}:
if action.priority_override is None:
return f"INVALID: {action.action_type.value} requires priority_override"
return (
f"{action.action_type.value} {action.incident_id} "
f"TO {action.priority_override.value}"
)
return f"INVALID: unknown action_type {action.action_type}"
class PhraseologyJudge(BaseModel):
"""Scores candidate phraseology against structured action truth."""
def _tokenize(self, text: str) -> set[str]:
import re
# Treat common dispatch IDs like "MED-1" and "INC-001" as single tokens.
return set(re.findall(r"[a-z0-9]+(?:-[a-z0-9]+)*", text.lower()))
def score(self, ground_truth_action: Action, candidate_text: str) -> float:
canonical = PhraseologyRenderer().render(ground_truth_action)
norm_candidate = candidate_text.strip().lower()
norm_canonical = canonical.strip().lower()
if norm_candidate == norm_canonical:
return 1.0
canonical_tokens = self._tokenize(canonical)
candidate_tokens = self._tokenize(candidate_text)
overlap = canonical_tokens & candidate_tokens
if not overlap:
return 0.0
return 0.5 if (len(overlap) / max(len(canonical_tokens), 1)) >= 0.5 else 0.0
def check_readback(self, candidate_text: str, ground_truth_action: Action) -> bool:
tokens = self._tokenize(candidate_text)
if ground_truth_action.action_type in {DispatchAction.DISPATCH, DispatchAction.REASSIGN}:
return (
ground_truth_action.unit_id.lower() in tokens
and ground_truth_action.incident_id.lower() in tokens
)
if ground_truth_action.action_type == DispatchAction.CANCEL:
return (
ground_truth_action.unit_id.lower() in tokens
and ground_truth_action.incident_id.lower() in tokens
)
if ground_truth_action.action_type in {DispatchAction.UPGRADE, DispatchAction.DOWNGRADE}:
if ground_truth_action.priority_override is None:
return False
return (
ground_truth_action.incident_id.lower() in tokens
and ground_truth_action.priority_override.value.lower() in tokens
)
return True
|