sensorflow/studio2/hardware.py

"""Hardware/domain dimensions as first-class gate strata.

A release that passes globally can still be unshippable on one platform in
one city with one sensor generation. This module builds a gate matrix whose
rows are combinations of:

    compute_platform   real device configs from sensorflow.vitis.backend
                       (KNOWN_DEVICES) + the reference CPU harness
    sensor_config      the megaeval 'sensor' cube dimension (lidar/camera/fused)
    sensor_generation  config-mapped generation label per sensor_config
    calibration_version latest sensorflow.safety.calibration status
    firmware           config-mapped per compute_platform
    region             domain stratum proxied from the megaeval 'road_type'
                       cube dimension (the derivation is recorded per row)

Metrics per combination come from REAL per-cohort data (megaeval metric cube
of the latest published run). Platform-specific evidence exists only where a
vitis HIL run exists for that device; combinations without evidence are
reported INSUFFICIENT — never assumed passing.

evaluate_matrix() is a pure function of (rows, policy) so the gate logic is
testable on planted inputs, separate from data synthesis.

Minimum-evidence math is delegated to hardening/seqeval power utilities when
importable; the fallback is a Wilson-interval width bound.
"""

from __future__ import annotations

import math
import os
from typing import Dict, List, Optional

from sensorflow.studio2 import store
from sensorflow.studio2.registry import content_hash

# ------------------------------------------------------------------ policy

DEFAULT_HARDWARE_POLICY: Dict = {
    "policy_name": "studio2-hardware-v1",
    "metrics": {
        "recall": {"min": 0.80},
        "precision": {"min": 0.90},
    },
    # evidence adequacy: a combination needs enough support that a Wilson CI
    # on its pass proportion is narrower than this (min-n delegated to
    # hardening/seqeval when importable)
    "max_ci_width": 0.10,
    # combinations matching any of these dicts are launch-critical: a failure
    # there blocks even when the global aggregate passes
    "critical_combinations": [
        {"region": "san_francisco", "sensor_config": "lidar"},
        {"sensor_config": "fused"},
    ],
}

# region strata proxied from the megaeval road_type dimension; the mapping is
# reported on every row so nobody mistakes it for geo-tagged data
REGION_PROFILES = {
    "san_francisco": {"road_type": "urban"},
    "phoenix": {"road_type": "highway"},
    "rural_pilot": {"road_type": "rural"},
}

SENSOR_GENERATIONS = {"lidar": "LiDAR-Gen2", "camera": "Camera-Gen3",
                      "fused": "Fusion-Stack-v1"}

FIRMWARE = {"cpu-reference": "host-fp32", "versal-ai-edge": "vitis-2024.2",
            "zynq-ultrascale": "vitis-2024.2"}


# ------------------------------------------------------------------ min-n


def min_support(max_ci_width: float, p: float = 0.85) -> Dict:
    """Minimum per-combination sample size for adequate evidence.

    Delegates to hardening.power when that package is importable (n to detect
    a proportion drop of max_ci_width/2 below the threshold, one-sided);
    otherwise a Wilson/normal-approximation bound: the CI full width at
    proportion p is ~ 2*z*sqrt(p(1-p)/n).
    """
    try:
        from sensorflow.hardening import power as hardening_power  # type: ignore
        n = int(hardening_power.required_n_two_proportions(
            baseline_rate=p, mde=max_ci_width / 2.0, two_sided=False))
        return {"n": n, "method": "hardening.power.required_n_two_proportions "
                f"(one-sided, p={p}, mde={max_ci_width / 2.0})"}
    except Exception:
        pass
    z = 1.959964  # 95%
    n = math.ceil((2.0 * z / max_ci_width) ** 2 * p * (1.0 - p))
    return {"n": int(n), "method": "wilson/normal-approx fallback "
            f"(z=1.96, p={p}, full width<={max_ci_width})"}


# ------------------------------------------------------------------ synthesis


def _latest_published_mega_run():
    """Returns (mega_store, run_object) for the newest published run."""
    from sensorflow.megaeval.runs import get_mega_store
    store_ = get_mega_store()
    published = [r for r in store_.runs.values()
                 if getattr(r, "status", None) == "published"]
    if not published:
        raise RuntimeError("no published megaeval runs")
    published.sort(key=lambda r: getattr(r, "published_at", None)
                   or getattr(r, "created_at", "") or "")
    return store_, published[-1]


def _cohort_metrics(mega_store, run_id: str, sensor: str, road_type: str) -> Optional[Dict]:
    from sensorflow.megaeval import cube as cube_mod
    cube_df = mega_store.artifacts(run_id)["cube"]
    rows, _ = cube_mod.aggregate(cube_df,
                                 {"sensor": [sensor], "road_type": [road_type]},
                                 None, ["n", "recall", "precision"], 1)
    return rows[0] if rows else None


def _vitis_platforms() -> Dict[str, Dict]:
    """Real device configs + whether any HIL evidence exists for them."""
    platforms: Dict[str, Dict] = {
        "cpu-reference": {"description": "reference CPU harness (fp32)",
                          "hil_evidence": True,
                          "evidence_note": "metrics measured directly on this harness"}}
    try:
        from sensorflow.vitis.backend import KNOWN_DEVICES
        hil_dir = os.path.join("runs", "vitis", "hil")
        has_hil = os.path.isdir(hil_dir) and any(
            n.startswith(("hilrun-", "hilsweep-")) for n in os.listdir(hil_dir))
        for name, spec in KNOWN_DEVICES.items():
            platforms[name] = {
                "description": spec.get("description", name),
                "hil_evidence": bool(has_hil),
                "evidence_note": ("vitis HIL quantization-gap runs present"
                                  if has_hil else
                                  "no vitis HIL run for this device"),
            }
    except Exception:
        pass
    return platforms


def synthesize_combinations() -> Dict:
    """Build combination rows from the real stores. Raises RuntimeError when
    the megaeval store has no published run (callers surface that as an
    unavailable matrix, they do not invent one)."""
    mega_store, run = _latest_published_mega_run()
    run_id = run.run_id
    calibration = None
    try:
        from sensorflow.safety import calibration as calib_mod
        cal = calib_mod.latest_status()
        calibration = (cal or {}).get("status")
    except Exception:
        pass
    calibration_version = f"calib-{(calibration or 'UNKNOWN').lower()}"

    platforms = _vitis_platforms()
    sensors = ["lidar", "camera", "fused"]

    rows: List[Dict] = []
    for platform, pinfo in platforms.items():
        for sensor in sensors:
            for region, proxy in REGION_PROFILES.items():
                combo = {
                    "compute_platform": platform,
                    "sensor_config": sensor,
                    "sensor_generation": SENSOR_GENERATIONS.get(sensor, "unknown"),
                    "calibration_version": calibration_version,
                    "firmware": FIRMWARE.get(platform, "unknown"),
                    "region": region,
                }
                row: Dict = {
                    "combination": combo,
                    "combination_label": f"{region} × {platform} × "
                                         f"{SENSOR_GENERATIONS.get(sensor, sensor)}",
                    "derivation": {
                        "metrics_source": f"megaeval cube of run {run_id}, "
                                          f"filter sensor={sensor}, "
                                          f"road_type={proxy['road_type']} "
                                          f"(region proxied from road_type)",
                        "platform_evidence": pinfo["evidence_note"],
                    },
                }
                if not pinfo["hil_evidence"]:
                    row.update({"metrics": None, "n": 0,
                                "evidence": "MISSING",
                                "evidence_reason": pinfo["evidence_note"]})
                else:
                    m = _cohort_metrics(mega_store, run_id, sensor,
                                        proxy["road_type"])
                    if m is None:
                        row.update({"metrics": None, "n": 0,
                                    "evidence": "MISSING",
                                    "evidence_reason": "no cube cells for this cohort"})
                    else:
                        row.update({"metrics": {"recall": m.get("recall"),
                                                "precision": m.get("precision")},
                                    "n": int(m.get("n") or 0),
                                    "evidence": "PRESENT"})
                rows.append(row)

    # global aggregate for the "global pass but critical combo fails" check
    headline = getattr(run, "headline", None) or {}
    global_metrics = {"recall": headline.get("recall"),
                      "precision": headline.get("precision")}
    return {"rows": rows, "global_metrics": global_metrics,
            "source_run_id": run_id}


# ------------------------------------------------------------------ evaluation


def _check_metrics(metrics: Optional[Dict], thresholds: Dict) -> List[str]:
    failed = []
    for metric, spec in thresholds.items():
        val = (metrics or {}).get(metric)
        if val is None:
            failed.append(f"{metric} missing")
        elif "min" in spec and val < spec["min"]:
            failed.append(f"{metric} {val:.3f} < {spec['min']}")
        elif "max" in spec and val > spec["max"]:
            failed.append(f"{metric} {val:.3f} > {spec['max']}")
    return failed


def _matches(combo: Dict, pattern: Dict) -> bool:
    return all(combo.get(k) == v for k, v in pattern.items())


def evaluate_matrix(rows: List[Dict], policy: Optional[Dict] = None,
                    global_metrics: Optional[Dict] = None,
                    source_run_id: Optional[str] = None,
                    persist: bool = True) -> Dict:
    """Pure gate logic over combination rows.

    Each row: {combination: {...}, combination_label, metrics|None, n,
    evidence: PRESENT|MISSING, [evidence_reason]}.
    """
    pol = {**DEFAULT_HARDWARE_POLICY, **(policy or {})}
    support = min_support(pol["max_ci_width"])
    thresholds = pol["metrics"]

    evaluated: List[Dict] = []
    insufficient: List[Dict] = []
    failures: List[Dict] = []
    critical_failures: List[Dict] = []

    for row in rows:
        out = dict(row)
        critical = any(_matches(row["combination"], pat)
                       for pat in pol["critical_combinations"])
        out["critical"] = critical
        if row.get("evidence") != "PRESENT" or row.get("metrics") is None:
            out["status"] = "INSUFFICIENT"
            out["reason"] = row.get("evidence_reason") or "no evidence"
            insufficient.append(out)
        elif int(row.get("n") or 0) < support["n"]:
            out["status"] = "INSUFFICIENT"
            out["reason"] = (f"support n={row.get('n')} below minimum "
                             f"{support['n']} ({support['method']})")
            insufficient.append(out)
        else:
            failed = _check_metrics(row["metrics"], thresholds)
            if failed:
                out["status"] = "FAIL"
                out["failed_checks"] = failed
                failures.append(out)
                if critical:
                    critical_failures.append(out)
            else:
                out["status"] = "PASS"
        evaluated.append(out)

    global_failed = _check_metrics(global_metrics, thresholds) if global_metrics else None
    global_pass = (global_failed == []) if global_failed is not None else None

    if critical_failures:
        status = "FAIL_CRITICAL"
    elif failures:
        status = "FAIL"
    elif not any(r["status"] == "PASS" for r in evaluated):
        status = "INSUFFICIENT"
    else:
        status = "PASS"

    report = {
        "matrix_id": f"hwm-{content_hash({'rows': [r.get('combination') for r in rows], 'policy': pol}, exclude=())}",
        "status": status,
        "global_pass": global_pass,
        "global_metrics": global_metrics,
        "global_vs_matrix_note": (
            "global aggregate passes but a launch-critical combination fails — "
            "the aggregate is hiding a stratum failure (Simpson-style); the "
            "matrix verdict wins"
            if global_pass and critical_failures else None),
        "n_combinations": len(evaluated),
        "n_pass": sum(1 for r in evaluated if r["status"] == "PASS"),
        "n_fail": len(failures),
        "n_insufficient": len(insufficient),
        "min_support": support,
        "critical_failures": critical_failures,
        "failures": [f for f in failures if f not in critical_failures],
        "insufficient": [{"combination_label": r["combination_label"],
                          "combination": r["combination"],
                          "critical": r["critical"],
                          "reason": r["reason"]} for r in insufficient],
        "rows": evaluated,
        "policy": pol,
        "source_run_id": source_run_id,
        "evaluated_at": store.now_iso(),
    }
    if persist:
        store.write_json(report, "hardware", "latest.json")
    return report


def gate_matrix(policy: Optional[Dict] = None, persist: bool = True) -> Dict:
    """Synthesize combinations from the real stores and evaluate them."""
    data = synthesize_combinations()
    return evaluate_matrix(data["rows"], policy=policy,
                           global_metrics=data["global_metrics"],
                           source_run_id=data["source_run_id"],
                           persist=persist)


def latest_matrix() -> Optional[Dict]:
    return store.read_json("hardware", "latest.json")