Fleet Management & Staged Rollout

"""Fleet health monitoring after OTA deployment."""
from dataclasses import dataclass
from typing import Dict

@dataclass
class UpdateCampaignMetrics:
    campaign_id: str
    software_version: str
    vehicles_targeted: int
    vehicles_updated: int
    vehicles_rolled_back: int
    vehicles_failed: int
    dtcs_new_post_update: Dict[str, int]  # DTC code -> count
    crash_reports_per_1k: float

def evaluate_campaign_health(metrics: UpdateCampaignMetrics) -> str:
    """Return GO / HOLD / ROLLBACK decision for campaign progression."""
    rollback_rate = (
        metrics.vehicles_rolled_back / metrics.vehicles_updated
        if metrics.vehicles_updated > 0 else 0
    )
    failure_rate = (
        metrics.vehicles_failed / metrics.vehicles_targeted
        if metrics.vehicles_targeted > 0 else 0
    )

    # Safety-critical DTC threshold
    safety_dtcs = {
        k: v for k, v in metrics.dtcs_new_post_update.items()
        if k.startswith(("C", "U"))  # chassis / network DTCs
    }

    if safety_dtcs or failure_rate > 0.01:
        return "ROLLBACK"   # immediate full campaign rollback
    elif rollback_rate > 0.02 or metrics.crash_reports_per_1k > 5:
        return "HOLD"       # pause; investigate
    else:
        return "GO"         # proceed to next stage

# Example usage
metrics = UpdateCampaignMetrics(
    campaign_id="FW_2025_Q1_001",
    software_version="2.4.1",
    vehicles_targeted=5000,
    vehicles_updated=4850,
    vehicles_rolled_back=12,
    vehicles_failed=3,
    dtcs_new_post_update={"P0300": 2},  # misfire (powertrain)
    crash_reports_per_1k=0.8
)
decision = evaluate_campaign_health(metrics)
print(f"Campaign decision: {decision}")  # GO