""" TriageResult: the single structured object that flows out of the pipeline. Keeping this as one dataclass makes app.py, tests, and the report generator all agree on one contract instead of passing loose dicts around. """ from dataclasses import dataclass, field from typing import List, Dict, Any from .config import risk_level_from_score @dataclass class TriageResult: scene_summary: List[str] = field(default_factory=list) location_type: str = "unknown" water_level: str = "n/a" road_status: str = "unknown" structural_damage: str = "unknown" hazards: List[str] = field(default_factory=list) people_visible_count: int = 0 people_description: str = "" risk_score: int = 0 risk_level: str = "" recommended_actions: List[str] = field(default_factory=list) confidence: float = 0.0 yolo_detections: Dict[str, int] = field(default_factory=dict) raw_model_output: str = "" def __post_init__(self): # Clamp risk score defensively — never trust a free-text model blindly. self.risk_score = max(0, min(100, int(self.risk_score))) if not self.risk_level: self.risk_level, _ = risk_level_from_score(self.risk_score) @property def risk_color(self) -> str: _, color = risk_level_from_score(self.risk_score) return color @classmethod def from_dict(cls, data: Dict[str, Any], yolo_detections: Dict[str, int], raw_model_output: str = "") -> "TriageResult": """Build a TriageResult from the VLM's (possibly messy) JSON dict, filling safe defaults for any missing/malformed field.""" people = data.get("people_status", {}) or {} return cls( scene_summary=list(data.get("scene_summary", []) or []), location_type=str(data.get("location_type", "unknown")), water_level=str(data.get("water_level", "n/a")), road_status=str(data.get("road_status", "unknown")), structural_damage=str(data.get("structural_damage", "unknown")), hazards=list(data.get("hazards", []) or []), people_visible_count=int(people.get("visible_count", 0) or 0), people_description=str(people.get("description", "")), risk_score=int(data.get("risk_score", 0) or 0), risk_level=str(data.get("risk_level", "") or ""), recommended_actions=list(data.get("recommended_actions", []) or []), confidence=float(data.get("confidence", 0.0) or 0.0), yolo_detections=yolo_detections, raw_model_output=raw_model_output, ) def to_markdown(self) -> str: bullets = "\n".join(f"- {s}" for s in self.scene_summary) or "- (no details extracted)" hazards = "\n".join(f"- ⚠️ {h}" for h in self.hazards) or "- None detected" actions = "\n".join(f"- [ ] {a}" for a in self.recommended_actions) or "- None" detections = ", ".join(f"{v}x {k}" for k, v in self.yolo_detections.items()) or "none" return f"""## Scene Summary {bullets} **Location type:** {self.location_type} **Water level:** {self.water_level} **Road status:** {self.road_status} **Structural damage:** {self.structural_damage} **People visible:** {self.people_visible_count} — {self.people_description or "n/a"} ### Hazards {hazards} ### Risk Score: {self.risk_score}/100 ({self.risk_level}) *Model confidence: {self.confidence:.0%}* ### Recommended Response {actions} --- **YOLO grounded detections:** {detections} """