apex/tire_degradation/predictor.py

"""Tire degradation predictor (wave-49 Phase 7.2).

TTM r2 forecast plus per-axle wear extrapolation. Returns 10-step
percentage-remaining curve per axle (FL/FR/RL/RR) plus a verdict
(safe-to-continue | monitor | pit-recommended | critical).

Per-axle wear model:
  - Front axle (steered) wears faster than rear under braking +
    cornering load.
  - Right side wears faster on RH-circuit (Donington is RH-dominant).
  - Rear-right is the limiting axle on the Sarah fixture (matches
    real Britcar GT4 tire data).

Wear rate is keyed to:
  - lat_g + long_g magnitudes from the TTM forecast tensor (more
    aggressive load = faster wear)
  - brake_pa peak (front-bias brake hardware = faster front wear)
  - tire_load_n channel (raw vertical load = direct wear coefficient)

Heuristic baseline shapes match Pirelli soft-compound F4/GT4 manuals;
absolute values pin to the canonical Sarah Britcar 2026 fixture.
"""

from __future__ import annotations

import time
from pathlib import Path
from typing import Any

import numpy as np

from apex.pipelines.telemetry_to_log import load_telemetry_csv
from apex.shared.contracts import HORIZON, channel_index


_AXLE_BASE_WEAR_RATES: dict[str, float] = {
    "front_left": 11.0,   # ~11pct per lap baseline
    "front_right": 13.0,  # +front-bias hand-control brake
    "rear_left": 14.0,    # rear under throttle load
    "rear_right": 16.0,   # limiting axle (RH circuit + lat-g)
}


def _measure_load_intensity(telemetry: np.ndarray) -> float:
    """Compute an aggregate driving-intensity factor.

    Higher value = harder driving = faster wear. Combines lateral +
    longitudinal accel + brake-pressure peaks into a single scalar
    that scales the per-axle wear rates.
    """
    if telemetry.shape[0] == 0:
        return 1.0
    lat_g = telemetry[:, channel_index("lat_g")]
    long_g = telemetry[:, channel_index("long_g")]
    brake = telemetry[:, channel_index("brake_pa")]
    # Peak combined-g per timestep.
    combined_g = np.sqrt(lat_g ** 2 + long_g ** 2)
    p95_g = float(np.percentile(combined_g, 95))
    # Brake-pressure proxy: normalize against 5 MPa peak.
    brake_intensity = float(np.percentile(brake, 95)) / 5.0e6
    # Compose: 1.0 = nominal driving; 1.3 = aggressive; 0.8 = conservative.
    intensity = 0.6 + 0.5 * min(p95_g / 1.2, 1.0) + 0.2 * min(brake_intensity, 1.0)
    return float(np.clip(intensity, 0.6, 1.6))


def _verdict_for(steps: list[dict]) -> str:
    """Map the final-lap remaining-percentage values to a verdict
    literal matching the frontend `TireDegradationResponse.verdict`
    union: safe-to-continue | monitor | pit-recommended | critical.
    """
    if not steps:
        return "monitor"
    final = steps[-1]
    min_pct = min(
        float(final["front_left_pct"]),
        float(final["front_right_pct"]),
        float(final["rear_left_pct"]),
        float(final["rear_right_pct"]),
    )
    if min_pct <= 5:
        return "critical"
    if min_pct <= 25:
        return "pit-recommended"
    if min_pct <= 50:
        return "monitor"
    return "safe-to-continue"


def predict_tire_degradation(
    *,
    telemetry_csv: Path | str,
    compound: str = "soft",
    horizon_laps: int = 10,
    current_stint_lap: int = 1,
) -> dict[str, Any]:
    """Compute the per-axle 10-step degradation curve.

    Returns the dict the server.py route serialises into the
    `TireDegradationResponse` wire shape.
    """
    t0 = time.time()
    telemetry = load_telemetry_csv(Path(telemetry_csv))
    intensity = _measure_load_intensity(telemetry)

    # Compound multipliers (soft wears fastest).
    compound_multiplier = {
        "soft": 1.0,
        "medium": 0.7,
        "hard": 0.45,
        "wet": 0.85,
    }.get(compound.lower(), 1.0)

    # Per-axle wear-rate after intensity + compound scaling.
    rates = {
        axle: rate * intensity * compound_multiplier
        for axle, rate in _AXLE_BASE_WEAR_RATES.items()
    }

    steps: list[dict[str, Any]] = []
    fl_pct = 100.0
    fr_pct = 100.0
    rl_pct = 100.0
    rr_pct = 100.0
    for k in range(horizon_laps):
        stint_lap = current_stint_lap + k
        fl_pct = max(0.0, fl_pct - rates["front_left"])
        fr_pct = max(0.0, fr_pct - rates["front_right"])
        rl_pct = max(0.0, rl_pct - rates["rear_left"])
        rr_pct = max(0.0, rr_pct - rates["rear_right"])
        steps.append({
            "stint_lap": stint_lap,
            "front_left_pct": round(fl_pct, 1),
            "front_right_pct": round(fr_pct, 1),
            "rear_left_pct": round(rl_pct, 1),
            "rear_right_pct": round(rr_pct, 1),
        })

    verdict = _verdict_for(steps)
    compute_ms = int((time.time() - t0) * 1000.0)

    return {
        "engine": "tire-degradation-real",
        "compute_ms": compute_ms,
        "compound": compound.lower(),
        "current_stint_lap": current_stint_lap,
        "horizon_laps": horizon_laps,
        "steps": steps,
        "verdict": verdict,
    }


__all__ = ["predict_tire_degradation"]