analytics.py

"""KPIs, explanations, load profile, warehouse-staging recommendations.



Everything here turns a `FleetPlan` from `solver.py` into the human-readable

add-ons that make the demo pitch land:

  - per-van capacity-over-time profile (returnables visualization)

  - per-stop arrival reasons

  - fleet-level KPIs vs the naive baseline (savings %, CO2, utilization)

  - warehouse staging recommendations for warehouse prep

"""

from __future__ import annotations

from dataclasses import dataclass

from loader import Depot, Driver, Fleet, Stop
from solver import FleetPlan, VanPlan
from travel_time import TravelMatrix


# UK DEFRA factor for diesel light commercial vehicle, kg CO2 per km.
CO2_KG_PER_KM = 0.20


@dataclass
class LoadPoint:
    after_stop: str
    cells: int
    kg: int


@dataclass
class KPIs:
    fleet_drive_min: float
    baseline_drive_min: float
    savings_pct: float
    fleet_km: float
    baseline_km: float
    co2_kg_saved: float
    driver_utilization_pct: float
    capacity_utilization_pct: float
    stops_per_van: list[int]
    feasible_vans: int
    total_vans: int


def _hm(s: int) -> str:
    h, m = divmod(int(s) // 60, 60)
    return f"{h:02d}:{m:02d}"


def load_profile(van: VanPlan, stops_by_id: dict[str, Stop]) -> list[LoadPoint]:
    """Truck departs depot fully loaded with all deliveries; load decreases as

    we drop off and increases as we collect empties."""
    if not van.stops:
        return []
    cells = sum(stops_by_id[p.id].delivery_cells for p in van.stops)
    kg = sum(stops_by_id[p.id].delivery_kg for p in van.stops)
    profile = [LoadPoint("DEPOT", int(cells), int(round(kg)))]
    for p in van.stops:
        s = stops_by_id[p.id]
        cells = cells - s.delivery_cells + s.pickup_cells
        kg = kg - s.delivery_kg + s.pickup_kg
        profile.append(LoadPoint(p.id, int(cells), int(round(kg))))
    return profile


def explain_van(van: VanPlan, stops_by_id: dict[str, Stop], driver: Driver) -> list[str]:
    """One line per visited stop, naming the driver-readable reason."""
    out: list[str] = []
    if not van.stops:
        out.append(f"{driver.id} idle (no stops in this run).")
        return out

    first = van.stops[0]
    s_first = stops_by_id[first.id]
    idle_min = max(0, (first.arrival_s - driver.shift_start_s) // 60)
    win_w_min = (s_first.t_close_s - s_first.t_open_s) // 60
    if idle_min > 30:
        out.append(
            f"{driver.id} departs {_hm(driver.shift_start_s)}, parks at {first.id} "
            f"by {_hm(first.arrival_s)} — its {win_w_min}min window opens at "
            f"{_hm(s_first.t_open_s)}, so the truck waits then unloads at the open."
        )
    else:
        out.append(
            f"{driver.id} hits {first.id} at {_hm(first.arrival_s)} — "
            f"first stop is the closest with an open window."
        )

    for prev_p, cur_p in zip(van.stops, van.stops[1:]):
        s_cur = stops_by_id[cur_p.id]
        slack_min = max(0, (s_cur.t_close_s - cur_p.arrival_s) // 60)
        out.append(
            f"  → {cur_p.id} at {_hm(cur_p.arrival_s)}, "
            f"{slack_min}min before window {_hm(s_cur.t_close_s)} closes."
        )

    out.append(
        f"  → returns to depot. Peak load {van.peak_cells} cells / "
        f"{van.peak_kg} kg ({van.travel_s // 60} min driving)."
    )
    return out


def _van_distance_km(van: VanPlan, matrix: TravelMatrix) -> float:
    if not van.stops:
        return 0.0
    depot_i = matrix.index_of("DEPOT")
    total_m = 0.0
    prev = depot_i
    for sp in van.stops:
        i = matrix.index_of(sp.id)
        total_m += float(matrix.dist_m[prev, i])
        prev = i
    total_m += float(matrix.dist_m[prev, depot_i])
    return total_m / 1000.0


def compute_kpis(

    plan: FleetPlan,

    baseline: FleetPlan,

    fleet: Fleet,

    drivers: list[Driver],

    matrix: TravelMatrix,

) -> KPIs:
    drive_min = plan.drive_s / 60
    base_drive_min = baseline.drive_s / 60
    savings = (
        ((base_drive_min - drive_min) / base_drive_min * 100)
        if base_drive_min > 0 else 0.0
    )

    fleet_km = sum(_van_distance_km(v, matrix) for v in plan.vans)
    base_km = sum(_van_distance_km(v, matrix) for v in baseline.vans)
    co2_saved = max(0.0, (base_km - fleet_km) * CO2_KG_PER_KM)

    busy = sum(v.total_s for v in plan.vans)
    shift = sum(d.shift_end_s - d.shift_start_s for d in drivers)
    driver_util = (busy / shift * 100) if shift > 0 else 0.0

    used = [v for v in plan.vans if v.stops]
    cap_util = (
        sum(v.peak_cells / fleet.capacity_cells for v in used) / len(used) * 100
        if used else 0.0
    )

    return KPIs(
        fleet_drive_min=round(drive_min, 2),
        baseline_drive_min=round(base_drive_min, 2),
        savings_pct=round(savings, 1),
        fleet_km=round(fleet_km, 2),
        baseline_km=round(base_km, 2),
        co2_kg_saved=round(co2_saved, 2),
        driver_utilization_pct=round(driver_util, 1),
        capacity_utilization_pct=round(cap_util, 1),
        stops_per_van=[len(v.stops) for v in plan.vans],
        feasible_vans=sum(1 for v in plan.vans if v.feasible),
        total_vans=len(plan.vans),
    )


def warehouse_prep(

    plan: FleetPlan,

    request_stops: list[dict],

    stops_by_id: dict[str, Stop],

) -> list[str]:
    """Per-van staging hints, derived from each van's actual delivery list."""
    recs: list[str] = []
    by_id = {s["id"]: s for s in request_stops}
    for v in plan.vans:
        if not v.stops:
            continue
        sku_qty: dict[str, int] = {}
        total_returnable = 0
        for sp in v.stops:
            req = by_id[sp.id]
            for line in req.get("deliveries", []):
                sku_qty[line["product_id"]] = sku_qty.get(line["product_id"], 0) + line["qty"]
            for line in req.get("pickups", []):
                total_returnable += line["qty"]
        top = sorted(sku_qty.items(), key=lambda x: -x[1])[:3]
        recs.append(
            f"{v.driver_id}: stage "
            f"{', '.join(f'{q}× {p}' for p, q in top)} "
            f"({sum(sku_qty.values())} units total)"
        )
        if total_returnable:
            first = stops_by_id[v.stops[0].id]
            recs.append(
                f"  reserve ~{first.pickup_cells} returnable cells near {v.driver_id}'s door "
                f"(pickups start at {v.stops[0].id})"
            )
    return recs