core/need_engine.py

"""
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