""" Need Profile Engine — 100% deterministic. This module is the boundary between "AI perception" and "AI-free math". Nothing here calls a model. Every function is pure and unit-testable. PRIORITY SCORE FORMULA (documented per project requirement): priority_raw = 0.30 * severity (0-10 scale) + 0.25 * people_affected_norm (min-max normalized across today's scenario, 0-10 scale) + 0.25 * urgency (0-10 scale) + 0.20 * need_avg (mean of medical/rescue/supply need, 0-10 scale) priority_score = round( priority_raw / 10 * 100 ) -> 0-100 Rationale for weights: severity and urgency are weighted highest (0.30 / 0.25) because they represent immediate life-safety risk. Population affected is normalized rather than used raw so that one location with 10x the population of another doesn't mechanically dominate every other signal. Need-type average is weighted lowest because it mostly reflects *what kind* of help is needed, not *how badly*. These weights are a design choice, not a law of nature — they are kept in one place (WEIGHTS below) specifically so they can be inspected, challenged, and changed without touching any other file. """ from __future__ import annotations from core.schemas import VisionAssessment, ReportExtraction, NeedProfile WEIGHTS = { "severity": 0.30, "population": 0.25, "urgency": 0.25, "need_avg": 0.20, } # Deterministic mapping: HF classifier label -> (damage_type, base_severity_at_100pct_confidence) # base_severity is what severity would be if the classifier were 100% confident; # actual severity scales down toward a neutral midpoint as confidence drops, # so a low-confidence classification never produces a falsely extreme score. CLASSIFIER_LABEL_MAP = { "Human_Damage": ("human_casualties", 9.5), "Fire_Disaster": ("fire", 9.0), "Water_Disaster": ("flooding", 7.5), "Land_Disaster": ("road_damage", 6.5), "Damaged_Infrastructure": ("structural_collapse", 8.5), "Non_Damage": ("no_significant_damage", 1.0), } NEUTRAL_SEVERITY = 5.0 # what we fall back toward when the model is unsure def score_vision_severity(classifier_label: str, confidence: float) -> tuple[str, float]: """ Deterministic conversion of a classifier label + confidence into (damage_type, severity_score). The MODEL only supplies label+confidence; THIS FUNCTION decides the number, and it's a fixed, inspectable formula. """ damage_type, base_severity = CLASSIFIER_LABEL_MAP.get( classifier_label, ("structural_collapse", NEUTRAL_SEVERITY) ) confidence = max(0.0, min(1.0, confidence)) # Linear interpolation between neutral (low confidence) and base (high confidence) severity = NEUTRAL_SEVERITY + (base_severity - NEUTRAL_SEVERITY) * confidence return damage_type, round(severity, 2) def _minmax_norm(value: float, all_values: list[float]) -> float: """Normalize `value` against the range of `all_values` onto a 0-10 scale.""" lo, hi = min(all_values), max(all_values) if hi == lo: return 5.0 # everyone equal -> neutral midpoint, avoids div-by-zero return (value - lo) / (hi - lo) * 10.0 def build_need_profiles( vision_by_location: dict[str, VisionAssessment], report_by_location: dict[str, ReportExtraction], display_names: dict[str, str], coordinates: dict[str, tuple[float, float]], ) -> list[NeedProfile]: """ Merge vision + report data per location into NeedProfile objects, with the priority_score computed by the documented formula above. Every location that has EITHER a vision assessment OR a report is included; missing fields fall back to conservative (low-priority) defaults rather than crashing, per the failure-handling requirement. """ location_ids = set(vision_by_location) | set(report_by_location) all_people = [report_by_location[loc].people_affected for loc in location_ids if loc in report_by_location] or [0] profiles = [] for loc in location_ids: vision = vision_by_location.get(loc) report = report_by_location.get(loc) severity = vision.severity_score if vision else NEUTRAL_SEVERITY damage_type = vision.damage_type if vision else "structural_collapse" caption = vision.caption if vision else "No image evidence available." people = report.people_affected if report else 0 urgency_report = report.urgency if report else 5.0 # Blend: if we have both signals, urgency is the average of report-stated # urgency and vision severity (a picture of collapse implies urgency # even if the text report under-states it). If only one exists, use it. urgency = (urgency_report + severity) / 2 if (vision and report) else ( urgency_report if report else severity ) need_types = set(report.need_types) if report else set() medical_need = 8.0 if "medical" in need_types else (severity * 0.3 if damage_type == "human_casualties" else 2.0) rescue_need = 8.0 if "rescue" in need_types else (severity * 0.4 if damage_type in ("structural_collapse", "flooding") else 2.0) supply_need = 6.0 if "supply" in need_types else 2.0 need_avg = (medical_need + rescue_need + supply_need) / 3 pop_norm = _minmax_norm(people, all_people) priority_raw = ( WEIGHTS["severity"] * severity + WEIGHTS["population"] * pop_norm + WEIGHTS["urgency"] * urgency + WEIGHTS["need_avg"] * need_avg ) priority_score = round(priority_raw / 10 * 100, 1) lat, lon = coordinates.get(loc, (0.0, 0.0)) profiles.append(NeedProfile( location_id=loc, display_name=display_names.get(loc, loc), severity=round(severity, 2), people_affected=people, urgency=round(urgency, 2), medical_need=round(medical_need, 2), rescue_need=round(rescue_need, 2), supply_need=round(supply_need, 2), priority_score=priority_score, required_medical_teams=report.requested_medical_teams if report else (2 if medical_need > 5 else 0), required_rescue_teams=report.requested_rescue_teams if report else (2 if rescue_need > 5 else 0), required_supply_trucks=report.requested_supply_trucks if report else (1 if supply_need > 5 else 0), damage_type=damage_type, evidence_caption=caption, lat=lat, lon=lon, )) # Highest priority first — this ordering is used directly by the dashboard's # "Priority Ranking" panel. profiles.sort(key=lambda p: p.priority_score, reverse=True) return profiles