Deploy Sensorflow Studio (part 2)

src/services/vitis.ts

@@ -0,0 +1,122 @@
+/** Typed fetch wrappers for the Vitis acceleration layer API (/api/vitis). */
+import type {
+  AugmentBatch,
+  AugmentationSpec,
+  BackendStatus,
+  HilRun,
+  HilSweep,
+  IspRun,
+  PrdDoc,
+  PrdListEntry,
+  RunListEntry,
+  TemporalRun,
+} from '../types/vitis';
+
+async function handle<T>(res: Response): Promise<T> {
+  if (!res.ok) {
+    let detail = '';
+    try {
+      const body = await res.json();
+      detail = typeof body?.detail === 'string' ? body.detail : JSON.stringify(body);
+    } catch {
+      detail = await res.text().catch(() => '');
+    }
+    const err = new Error(detail || res.statusText) as Error & { status?: number };
+    err.status = res.status;
+    throw err;
+  }
+  return res.json() as Promise<T>;
+}
+
+const JSON_HEADERS = { 'Content-Type': 'application/json' };
+
+const get = <T,>(url: string): Promise<T> => fetch(url).then((r) => handle<T>(r));
+const post = <T,>(url: string, body?: unknown): Promise<T> =>
+  fetch(url, {
+    method: 'POST',
+    headers: JSON_HEADERS,
+    body: body === undefined ? undefined : JSON.stringify(body),
+  }).then((r) => handle<T>(r));
+
+// ------------------------------------------------------------- backends
+
+export const getBackendStatus = () => get<BackendStatus>('/api/vitis/backends/status');
+
+// ------------------------------------------------------------- HIL
+
+export interface HilRunRequest {
+  n_sequences?: number;
+  frames_per_sequence?: number;
+  seed?: number;
+  width_bits?: number;
+  int_bits?: number;
+  max_line_buffer_depth?: number;
+  use_lut_approx?: boolean;
+  lut_bits?: number;
+  device?: string;
+  regression_delta?: number;
+  alpha?: number;
+  run_ablation?: boolean;
+}
+
+export const runHil = (body: HilRunRequest) => post<HilRun>('/api/vitis/hil/run', body);
+
+export const runHilSweep = (body: HilRunRequest & { widths?: number[] }) =>
+  post<HilSweep>('/api/vitis/hil/sweep', body);
+
+export const listHilRuns = () => get<{ runs: RunListEntry[] }>('/api/vitis/hil/runs');
+
+// ------------------------------------------------------------- ISP
+
+export interface IspRunRequest {
+  n_frames?: number;
+  seed?: number;
+  stages?: string[];
+  stage_params?: Record<string, unknown>;
+  width_bits?: number;
+  int_bits?: number;
+  max_line_buffer_depth?: number;
+  use_lut_approx?: boolean;
+  device?: string;
+  include_previews?: boolean;
+}
+
+export const runIsp = (body: IspRunRequest) => post<IspRun>('/api/vitis/isp/run', body);
+
+export const getIspStages = () => get<{ stages: string[] }>('/api/vitis/isp/stages');
+
+// ------------------------------------------------------------- augment
+
+export const getAugmentRecipes = () =>
+  get<{ augmentations: AugmentationSpec[] }>('/api/vitis/augment/recipes');
+
+export const generateAugmentBatch = (body: {
+  recipes?: { aug: string; params?: Record<string, number> }[];
+  n_variants?: number;
+  seed?: number;
+  backend?: string;
+  width_bits?: number;
+  device?: string;
+  include_thumbnails?: boolean;
+}) => post<AugmentBatch>('/api/vitis/augment/generate', body);
+
+// ------------------------------------------------------------- temporal
+
+export const getTemporalEngines = () =>
+  get<{ engines: string[] }>('/api/vitis/temporal/engines');
+
+export const runTemporal = (body: {
+  engines?: string[];
+  n_sequences?: number;
+  frames_per_sequence?: number;
+  seed?: number;
+  width_bits?: number;
+  device?: string;
+}) => post<TemporalRun>('/api/vitis/temporal/run', body);
+
+// ------------------------------------------------------------- PRD
+
+export const listPrds = () => get<{ prds: PrdListEntry[] }>('/api/vitis/prd');
+
+export const getPrd = (id: string) =>
+  get<PrdDoc>(`/api/vitis/prd/${encodeURIComponent(id)}`);

src/types/vitis.ts

@@ -0,0 +1,253 @@
+/** Types mirroring the /api/vitis JSON payloads (snake_case from FastAPI). */
+
+// ------------------------------------------------------------- backends
+
+export interface BackendInfo {
+  name: string;
+  available: boolean;
+  emulated: boolean;
+  description: string;
+}
+
+export interface BackendStatus {
+  backends: BackendInfo[];
+  devices: Record<string, { clock_mhz: number; has_aie: boolean; description: string }>;
+  hardware_present: boolean;
+  note: string;
+}
+
+// ------------------------------------------------------------- HIL
+
+export interface HilVerdict {
+  decision: 'REGRESSION' | 'PASS' | 'INSUFFICIENT_EVIDENCE';
+  mean_delta: number;
+  ci: [number, number];
+  n: number;
+  method: string;
+}
+
+export interface HilComparison {
+  config: Record<string, unknown>;
+  totals: {
+    gt_objects: number;
+    ref_detected: number;
+    vitis_detected: number;
+    paired_detections: number;
+    dropped_by_vitis: number;
+    spurious_in_vitis: number;
+    class_flips: number;
+  };
+  drift: {
+    mean_confidence_drift: number;
+    mean_abs_confidence_drift: number;
+    mean_position_drift_px: number;
+    mean_pair_iou: number;
+  };
+  gap_score: number;
+  cohort_delta: Record<string, number>;
+}
+
+export interface HilAblation {
+  legs: Record<string, { gap_score: number }>;
+  attribution: Record<string, number>;
+  note: string;
+}
+
+export interface HilRun {
+  run_id: string;
+  created_at: string;
+  params: Record<string, unknown>;
+  comparison: HilComparison;
+  verdict: HilVerdict;
+  ablation: HilAblation | null;
+  elapsed_s: number;
+  emulation_note: string;
+}
+
+export interface HilSweepPoint {
+  width_bits: number;
+  int_bits: number;
+  gap_score: number;
+  dropped: number;
+  class_flips: number;
+  mean_abs_confidence_drift: number;
+  mean_position_drift_px: number;
+  mean_pair_iou: number;
+  decision: HilVerdict['decision'];
+  mean_delta: number;
+}
+
+export interface HilSweep {
+  run_id: string;
+  created_at: string;
+  params: Record<string, unknown>;
+  points: HilSweepPoint[];
+  minimal_passing_config: {
+    width_bits: number;
+    int_bits: number;
+    decision: string;
+    note?: string;
+  } | null;
+  elapsed_s: number;
+  emulation_note: string;
+}
+
+// ------------------------------------------------------------- ISP
+
+export interface IspStageEntry {
+  stage: string;
+  psnr_db: number;
+  ssim: number;
+  measured_cpu_ms: number;
+  modeled_fpga_ms?: number;
+  modeled_placement?: string;
+  modeled_speedup_x?: number;
+  modeled_not_measured?: boolean;
+}
+
+export interface IspThroughput {
+  measured_cpu_ms_per_frame: number;
+  measured_cpu_fps: number;
+  modeled_fpga_ms_per_frame_serial: number;
+  modeled_fpga_fps_pipelined: number;
+  modeled_speedup_x_serial: number;
+  modeled_not_measured: boolean;
+  note: string;
+}
+
+export interface IspPreviewStage {
+  stage: string;
+  reference_png: string;
+  vitis_png: string;
+  diff_png: string;
+}
+
+export interface IspRun {
+  run_id: string;
+  created_at: string;
+  params: Record<string, unknown>;
+  stage_report: IspStageEntry[];
+  throughput: IspThroughput;
+  previews: { frame_id: string; cohort: string; input_png?: string; stages: IspPreviewStage[] }[];
+  elapsed_s: number;
+  emulation_note: string;
+}
+
+// ------------------------------------------------------------- augment
+
+export interface AugmentationSpec {
+  name: string;
+  description: string;
+  defaults: Record<string, number>;
+}
+
+export interface AugmentVariant {
+  variant_id: string;
+  frame_id: string;
+  dataset_id: string;
+  scene_id: string;
+  sequence_id: string;
+  weather: string;
+  time_of_day: string;
+  scenario_tags: string[];
+  evaluation_only: boolean;
+  training_eligible: boolean;
+  protected_evaluation: boolean;
+  recommended_dataset_destination: string;
+  lineage: {
+    batch_id: string;
+    source_frame_id: string;
+    source_sequence_id: string;
+    recipe: { aug: string; params: Record<string, number> }[];
+    seed: number;
+    backend: string;
+    backend_emulated: boolean;
+  };
+  recipe: { aug: string; params: Record<string, number> }[];
+  thumbnail_png?: string;
+}
+
+export interface AugmentBatch {
+  run_id: string;
+  created_at: string;
+  params: Record<string, unknown>;
+  variants: AugmentVariant[];
+  raremine_hook: { available: boolean; routed_candidates: number; note?: string };
+  elapsed_s: number;
+  emulation_note: string;
+}
+
+// ------------------------------------------------------------- temporal
+
+export interface EngineStability {
+  flicker_rate: number;
+  mean_jitter: number;
+  fragmentation_per_track: number;
+  id_switches: number;
+  id_switches_at_flow_break: number;
+  unexcused_id_switch_fraction: number;
+  stability_score: number;
+  cohorts?: Record<string, EngineStability>;
+}
+
+export interface StereoReport {
+  objects_checked: number;
+  median_abs_depth_error_m: number | null;
+  median_rel_depth_error: number | null;
+  median_abs_depth_error_near_m: number | null;
+  p90_abs_depth_error_m: number | null;
+  median_abs_disparity_error_px: number | null;
+  note?: string;
+}
+
+export interface TimelineFrame {
+  frame_index: number;
+  detected: boolean;
+  track_id: string | number | null;
+  flow_continuous: boolean;
+  flow_residual_m: number | null;
+  occluded: boolean;
+}
+
+export interface TemporalRun {
+  run_id: string;
+  created_at: string;
+  params: Record<string, unknown>;
+  results: Record<string, { engines: Record<string, EngineStability>; stereo: StereoReport }>;
+  backend_agreement: {
+    ranking_agrees: boolean;
+    ranking_reference: string[];
+    ranking_vitis_emulated: string[];
+    stability_score_delta_by_engine: Record<string, number>;
+    max_abs_score_delta: number;
+    note: string;
+  };
+  timeline_sample: {
+    instance_id: string;
+    engine: string;
+    sequence_id: string;
+    frames: TimelineFrame[];
+  } | null;
+  elapsed_s: number;
+  emulation_note: string;
+}
+
+// ------------------------------------------------------------- PRD / listings
+
+export interface PrdListEntry {
+  id: string;
+  file: string;
+  available: boolean;
+}
+
+export interface PrdDoc {
+  id: string;
+  file: string;
+  markdown: string;
+}
+
+export interface RunListEntry {
+  run_id: string;
+  created_at: string | null;
+  summary: Record<string, unknown>;
+}

tests/test_vitis/conftest.py

@@ -0,0 +1,21 @@
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from sensorflow.vitis import store
+
+
+@pytest.fixture()
+def vitis_root(tmp_path):
+    store.set_vitis_root(str(tmp_path / "vitis"))
+    yield tmp_path / "vitis"
+    store.set_vitis_root("runs/vitis")
+
+
+@pytest.fixture()
+def vitis_client(vitis_root):
+    from sensorflow.vitis import api as vitis_api
+    app = FastAPI()
+    app.include_router(vitis_api.router)
+    with TestClient(app) as client:
+        yield client

tests/test_vitis/test_api.py

@@ -0,0 +1,106 @@
+"""API lifecycle across every endpoint group."""
+
+
+class TestBackendsApi:
+    def test_status(self, vitis_client):
+        r = vitis_client.get("/api/vitis/backends/status")
+        assert r.status_code == 200
+        body = r.json()
+        assert body["hardware_present"] is False
+        names = {b["name"]: b for b in body["backends"]}
+        assert names["vitis_emulated"]["emulated"] is True
+        assert names["vitis_hw"]["available"] is False
+
+    def test_devices(self, vitis_client):
+        r = vitis_client.get("/api/vitis/backends/devices")
+        assert {d["name"] for d in r.json()["devices"]} == \
+            {"versal-ai-edge", "zynq-ultrascale"}
+
+
+class TestHilApi:
+    def test_run_report_lifecycle(self, vitis_client):
+        r = vitis_client.post("/api/vitis/hil/run", json={
+            "n_sequences": 1, "frames_per_sequence": 6, "width_bits": 10,
+            "run_ablation": True})
+        assert r.status_code == 200
+        run = r.json()
+        assert run["verdict"]["decision"] in (
+            "REGRESSION", "PASS", "INSUFFICIENT_EVIDENCE")
+        assert run["ablation"]["attribution"]
+
+        rid = run["run_id"]
+        assert vitis_client.get(f"/api/vitis/hil/runs/{rid}").status_code == 200
+        listing = vitis_client.get("/api/vitis/hil/runs").json()["runs"]
+        assert any(x["run_id"] == rid for x in listing)
+        assert vitis_client.get("/api/vitis/hil/runs/nope").status_code == 404
+
+    def test_sweep(self, vitis_client):
+        r = vitis_client.post("/api/vitis/hil/sweep", json={
+            "n_sequences": 1, "frames_per_sequence": 6, "widths": [8, 12]})
+        assert r.status_code == 200
+        assert len(r.json()["points"]) == 2
+
+    def test_validation(self, vitis_client):
+        assert vitis_client.post("/api/vitis/hil/run", json={
+            "width_bits": 4, "int_bits": 4}).status_code == 400
+        assert vitis_client.post("/api/vitis/hil/run", json={
+            "device": "cray-1"}).status_code == 400
+
+
+class TestIspAugmentApi:
+    def test_isp_lifecycle(self, vitis_client):
+        r = vitis_client.post("/api/vitis/isp/run", json={
+            "n_frames": 1, "include_previews": False})
+        assert r.status_code == 200
+        run = r.json()
+        assert run["throughput"]["modeled_not_measured"] is True
+        rid = run["run_id"]
+        assert vitis_client.get(f"/api/vitis/isp/runs/{rid}").status_code == 200
+        assert vitis_client.post("/api/vitis/isp/run", json={
+            "stages": ["defrag"]}).status_code == 400
+        assert vitis_client.get("/api/vitis/isp/stages").json()["stages"]
+
+    def test_augment_lifecycle(self, vitis_client):
+        r = vitis_client.get("/api/vitis/augment/recipes")
+        assert len(r.json()["augmentations"]) == 7
+        r = vitis_client.post("/api/vitis/augment/generate", json={
+            "n_variants": 2, "include_thumbnails": False})
+        assert r.status_code == 200
+        batch = r.json()
+        assert all(v["evaluation_only"] for v in batch["variants"])
+        bid = batch["run_id"]
+        assert vitis_client.get(
+            f"/api/vitis/augment/batches/{bid}").status_code == 200
+        variants = vitis_client.get("/api/vitis/augment/variants").json()
+        assert len(variants["variants"]) >= 2
+        assert vitis_client.post("/api/vitis/augment/generate", json={
+            "backend": "vitis_hw"}).status_code == 400
+
+
+class TestTemporalApi:
+    def test_temporal_lifecycle(self, vitis_client):
+        engines = vitis_client.get("/api/vitis/temporal/engines").json()["engines"]
+        assert len(engines) == 2
+        r = vitis_client.post("/api/vitis/temporal/run", json={
+            "n_sequences": 1, "frames_per_sequence": 10})
+        assert r.status_code == 200
+        run = r.json()
+        assert set(run["results"]) == {"reference", "vitis_emulated"}
+        assert "ranking_agrees" in run["backend_agreement"]
+        rid = run["run_id"]
+        assert vitis_client.get(
+            f"/api/vitis/temporal/runs/{rid}").status_code == 200
+        assert vitis_client.post("/api/vitis/temporal/run", json={
+            "engines": ["skynet"]}).status_code == 400
+
+
+class TestPrdApi:
+    def test_prd_listing_and_content(self, vitis_client):
+        listing = vitis_client.get("/api/vitis/prd").json()["prds"]
+        assert {p["id"] for p in listing} == {
+            "vitis-hil-regression", "vitis-isp-preprocessing",
+            "vitis-temporal-stability"}
+        assert all(p["available"] for p in listing)
+        doc = vitis_client.get("/api/vitis/prd/vitis-hil-regression").json()
+        assert "Value Proposition" in doc["markdown"]
+        assert vitis_client.get("/api/vitis/prd/nope").status_code == 404

tests/test_vitis/test_backend.py

@@ -0,0 +1,156 @@
+"""Backend abstraction: parity, quantization monotonicity, XFCVDEPTH,
+LUT approximations, latency model, registry."""
+
+import numpy as np
+import pytest
+
+from sensorflow.vitis.backend import (
+    BACKENDS, DeviceConfig, PipelineConfig, backend_status, get_backend,
+    make_bayer_mosaic,
+)
+
+
+@pytest.fixture(scope="module")
+def img():
+    rng = np.random.default_rng(3)
+    base = rng.random((96, 128, 3)).astype(np.float32)
+    return base
+
+
+@pytest.fixture(scope="module")
+def gray(img):
+    return img.mean(axis=2).astype(np.float32)
+
+
+def _cfg(w, i=6, **kw):
+    return PipelineConfig(precision={"default": (w, i)}, **kw)
+
+
+class TestParityAndMonotonicity:
+    def test_high_bitwidth_parity_within_quantization_bound(self, img):
+        ref = get_backend("reference")
+        vit = get_backend("vitis_emulated", _cfg(16))
+        r = ref.gaussian_filter(img, 1.5)
+        v = vit.gaussian_filter(img, 1.5)
+        # ap_fixed<16,6> has 10 fractional bits -> lsb ~ 1e-3; allow a few lsb
+        # of accumulated error through the filter.
+        assert float(np.abs(r - v).mean()) < 4 * 2.0 ** -10
+
+    def test_divergence_grows_monotonically_as_bits_shrink(self, img):
+        ref = get_backend("reference")
+        target = ref.gaussian_filter(img, 1.5)
+        errs = []
+        for w in (16, 12, 10, 8):
+            vit = get_backend("vitis_emulated", _cfg(w))
+            errs.append(float(np.abs(vit.gaussian_filter(img, 1.5) - target).mean()))
+        assert errs == sorted(errs), f"not monotone: {errs}"
+        assert errs[-1] > 10 * errs[0]
+
+    def test_ops_run_on_both_backends(self, img, gray):
+        raw = make_bayer_mosaic(img)
+        for name in ("reference", "vitis_emulated"):
+            be = get_backend(name, _cfg(16))
+            assert be.demosaic(raw).shape == img.shape
+            assert be.resize(img, 48, 64).shape == (48, 64, 3)
+            assert be.resize(gray, 48, 64, "area").shape == (48, 64)
+            assert be.crop(img, 10, 10, 20, 30).shape == (20, 30, 3)
+            yuv = be.rgb_to_yuv(img)
+            back = be.yuv_to_rgb(yuv)
+            assert float(np.abs(back - img).mean()) < 0.05
+            assert be.bad_pixel_correction(gray).shape == gray.shape
+            assert be.hdr_tone_map(img).max() <= 1.0
+            assert be.gain_exposure(img, 2.0).max() <= 1.0
+            assert be.lens_distortion(gray, k1=0.2).shape == gray.shape
+            assert be.median_filter(gray, 3).shape == gray.shape
+            merged = be.hdr_merge([img * 0.3, img], [0.5, 1.0])
+            assert merged.shape == img.shape
+
+
+class TestStreamingDepthConstraint:
+    def test_no_artifacts_when_depth_sufficient(self, gray):
+        wide = get_backend("vitis_emulated", _cfg(16, max_line_buffer_depth=4096))
+        exact = get_backend("vitis_emulated",
+                            _cfg(16, max_line_buffer_depth=gray.shape[1]))
+        a = wide.gaussian_filter(gray, 2.0)
+        b = exact.gaussian_filter(gray, 2.0)
+        assert np.array_equal(a, b)
+
+    def test_localized_boundary_artifacts_only_when_exceeded(self, gray):
+        depth = 48
+        full = get_backend("vitis_emulated", _cfg(16, max_line_buffer_depth=4096))
+        narrow = get_backend("vitis_emulated",
+                             _cfg(16, max_line_buffer_depth=depth))
+        a = full.gaussian_filter(gray, 2.0)
+        b = narrow.gaussian_filter(gray, 2.0)
+        col_err = np.abs(a - b).mean(axis=0)
+        radius = 8  # scipy gaussian kernel truncates at 4*sigma = 8 px
+        seam_cols = [c for s in range(depth, gray.shape[1], depth)
+                     for c in range(max(0, s - radius),
+                                    min(gray.shape[1], s + radius))]
+        interior = [c for c in range(gray.shape[1]) if c not in set(seam_cols)]
+        assert col_err[seam_cols].max() > 1e-3, "expected seam artifacts"
+        assert col_err[interior].max() < 1e-6, "artifacts leaked beyond seams"
+
+
+class TestHlsApproximations:
+    def test_lut_recip_bounded_relative_error(self):
+        cfg = _cfg(24, i=8, lut_bits=8)
+        from sensorflow.vitis.backend import _FixedNumerics
+        num = _FixedNumerics(cfg)
+        x = np.linspace(0.01, 50.0, 5000).astype(np.float32)
+        rel = np.abs(num.recip(x) - 1.0 / x) * x
+        assert float(rel.max()) < 2.0 ** -7  # ~2x lsb of a 256-entry LUT
+
+    def test_lut_sqrt_bounded_relative_error(self):
+        cfg = _cfg(24, i=8, lut_bits=8)
+        from sensorflow.vitis.backend import _FixedNumerics
+        num = _FixedNumerics(cfg)
+        x = np.linspace(0.01, 50.0, 5000).astype(np.float32)
+        rel = np.abs(num.sqrt(x) - np.sqrt(x)) / np.sqrt(x)
+        assert float(rel.max()) < 2.0 ** -8
+
+    def test_lut_toggle_changes_output(self, img):
+        on = get_backend("vitis_emulated", _cfg(20, i=6, use_lut_approx=True))
+        off = get_backend("vitis_emulated", _cfg(20, i=6, use_lut_approx=False))
+        a = on.hdr_tone_map(img)
+        b = off.hdr_tone_map(img)
+        assert not np.array_equal(a, b)
+
+
+class TestLatencyModelAndRegistry:
+    def test_modeled_latency_deterministic_and_labeled(self, gray):
+        be = get_backend("vitis_emulated", _cfg(12))
+        be.gaussian_filter(gray, 1.0)
+        rep = be.profile_report()[-1]
+        assert rep["modeled"]["modeled_not_measured"] is True
+        assert rep["modeled"]["latency_ms"] > 0
+        be2 = get_backend("vitis_emulated", _cfg(12))
+        be2.gaussian_filter(gray, 1.0)
+        assert be2.profile_report()[-1]["modeled"]["latency_ms"] == \
+            rep["modeled"]["latency_ms"]
+
+    def test_aie_placement_depends_on_device(self, gray):
+        versal = get_backend("vitis_emulated", PipelineConfig(
+            device=DeviceConfig(name="versal-ai-edge")))
+        zynq = get_backend("vitis_emulated", PipelineConfig(
+            device=DeviceConfig(name="zynq-ultrascale")))
+        versal.gaussian_filter(gray, 1.0)
+        zynq.gaussian_filter(gray, 1.0)
+        assert versal.profile_report()[-1]["modeled"]["placement"] == "AIE"
+        assert zynq.profile_report()[-1]["modeled"]["placement"] == "PL"
+
+    def test_reference_backend_reports_no_modeled_numbers(self, gray):
+        ref = get_backend("reference")
+        ref.gaussian_filter(gray, 1.0)
+        assert "modeled" not in ref.profile_report()[-1]
+
+    def test_registry_and_hw_stub(self):
+        assert set(BACKENDS) == {"reference", "vitis_emulated", "vitis_hw"}
+        with pytest.raises(NotImplementedError):
+            get_backend("vitis_hw")
+        with pytest.raises(ValueError):
+            get_backend("nonsense")
+        status = backend_status()
+        assert status["hardware_present"] is False
+        emu = next(b for b in status["backends"] if b["name"] == "vitis_emulated")
+        assert emu["emulated"] is True and "modeled" in emu["description"]

tests/test_vitis/test_hil.py

@@ -0,0 +1,74 @@
+"""HIL feature: ablation attribution, sequential/fallback verdicts, sweep."""
+
+import numpy as np
+
+from sensorflow.vitis.hil import (
+    _fallback_verdict, run_bitwidth_sweep, run_hil, sequential_verdict,
+)
+
+
+class TestAblationAttribution:
+    def test_planted_precision_regression_attributed_to_precision(self, vitis_root):
+        # W=8 with generous depth and LUTs on: precision is the only real
+        # constraint that bites at this image size.
+        run = run_hil(n_sequences=2, frames_per_sequence=8, seed=7,
+                      width_bits=8, int_bits=4,
+                      max_line_buffer_depth=2048, use_lut_approx=True)
+        attr = run["ablation"]["attribution"]
+        assert attr["precision_only"] > 0.8
+        assert attr["precision_only"] > attr["streaming_only"]
+        assert attr["precision_only"] > attr["hls_approx_only"]
+
+    def test_high_precision_run_has_negligible_gap(self, vitis_root):
+        run = run_hil(n_sequences=2, frames_per_sequence=8, seed=7,
+                      width_bits=16, int_bits=4, run_ablation=False)
+        assert run["comparison"]["gap_score"] < 0.01
+        assert run["comparison"]["totals"]["dropped_by_vitis"] == 0
+
+
+class TestSequentialVerdicts:
+    def test_fallback_regression_on_planted_regression(self):
+        v = _fallback_verdict([-0.25 + 0.01 * (i % 3) for i in range(40)],
+                              delta=0.02, alpha=0.05)
+        assert v["decision"] == "REGRESSION"
+
+    def test_fallback_pass_on_no_regression(self):
+        v = _fallback_verdict([0.002 - 0.001 * (i % 2) for i in range(40)],
+                              delta=0.02, alpha=0.05)
+        assert v["decision"] == "PASS"
+
+    def test_fallback_insufficient_on_tiny_sample(self):
+        assert _fallback_verdict([0.0], 0.02, 0.05)["decision"] == \
+            "INSUFFICIENT_EVIDENCE"
+
+    def test_sequential_verdict_detects_planted_regression(self):
+        # Strong planted regression: vitis path loses ~half its objects.
+        rng = np.random.default_rng(5)
+        deltas = list(np.clip(rng.normal(-0.5, 0.05, 120), -1, 1))
+        pairs = [(True, bool(rng.random() > 0.5)) for _ in range(600)]
+        v = sequential_verdict(deltas, pairs, delta=0.02, alpha=0.05)
+        assert v["decision"] == "REGRESSION"
+        assert v["method"] in ("seqeval_anytime_valid", "paired_t_fallback")
+
+    def test_sequential_verdict_no_false_regression(self):
+        deltas = [0.0] * 120
+        pairs = [(True, True)] * 600
+        v = sequential_verdict(deltas, pairs, delta=0.02, alpha=0.05)
+        assert v["decision"] != "REGRESSION"
+
+
+class TestSweep:
+    def test_sweep_monotone_gap_and_minimal_config(self, vitis_root):
+        sw = run_bitwidth_sweep(n_sequences=2, frames_per_sequence=8, seed=7,
+                                widths=[6, 8, 12, 16])
+        pts = sw["points"]  # sorted ascending by width
+        gaps = [p["gap_score"] for p in pts]
+        assert gaps == sorted(gaps, reverse=True), f"gap not monotone: {gaps}"
+        assert pts[0]["decision"] == "REGRESSION"  # W=6 is a planted disaster
+        mp = sw["minimal_passing_config"]
+        assert mp is not None and mp["width_bits"] > 6
+
+    def test_run_persisted(self, vitis_root):
+        run = run_hil(n_sequences=1, frames_per_sequence=6, width_bits=12,
+                      run_ablation=False)
+        assert (vitis_root / "hil" / f"{run['run_id']}.json").exists()

tests/test_vitis/test_isp_augment.py

@@ -0,0 +1,80 @@
+"""ISP per-stage quality/throughput + augmentation lineage & leakage guard."""
+
+from sensorflow.vitis.augment import generate_batch, list_augmentations
+from sensorflow.vitis.isp import run_isp
+
+
+class TestIsp:
+    def test_per_stage_psnr_sanity(self, vitis_root):
+        run = run_isp(n_frames=2, width_bits=12, include_previews=False)
+        assert [e["stage"] for e in run["stage_report"]] == \
+            ["bad_pixel_correction", "demosaic", "hdr_tone_map", "gain",
+             "denoise", "resize"]
+        for e in run["stage_report"]:
+            assert 20.0 < e["psnr_db"] <= 99.0, e
+            assert 0.5 < e["ssim"] <= 1.0, e
+        # Higher precision must not be worse than lower precision.
+        lo = run_isp(n_frames=2, width_bits=8, include_previews=False)
+        assert min(e["psnr_db"] for e in run["stage_report"]) > \
+            min(e["psnr_db"] for e in lo["stage_report"])
+
+    def test_throughput_labeled_modeled(self, vitis_root):
+        run = run_isp(n_frames=1, include_previews=False)
+        t = run["throughput"]
+        assert t["modeled_not_measured"] is True
+        assert t["measured_cpu_ms_per_frame"] > 0
+        assert t["modeled_fpga_fps_pipelined"] > 0
+        for e in run["stage_report"]:
+            assert e["modeled_not_measured"] is True
+
+    def test_previews_and_persistence(self, vitis_root):
+        run = run_isp(n_frames=1, include_previews=True)
+        assert run["previews"][0]["stages"][0]["reference_png"].startswith(
+            "data:image/png;base64,")
+        assert (vitis_root / "isp" / f"{run['run_id']}.json").exists()
+
+
+class TestAugment:
+    def test_lineage_completeness_and_eval_only_flag(self, vitis_root):
+        batch = generate_batch(n_variants=5, include_thumbnails=False)
+        assert len(batch["variants"]) == 5
+        for v in batch["variants"]:
+            lin = v["lineage"]
+            for key in ("batch_id", "source_frame_id", "source_sequence_id",
+                        "recipe", "seed", "backend", "backend_config"):
+                assert key in lin, f"lineage missing {key}"
+            assert lin["recipe"] and all("aug" in s and "params" in s
+                                         for s in lin["recipe"])
+            # Leakage guard: evaluation-only by default, never training data.
+            assert v["evaluation_only"] is True
+            assert v["training_eligible"] is False
+            assert v["recommended_dataset_destination"] == \
+                "REGRESSION_EVALUATION_SET"
+            assert v["gt_boxes"], "variants must inherit GT"
+
+    def test_deterministic_given_seed(self, vitis_root):
+        a = generate_batch(n_variants=3, seed=99, include_thumbnails=False)
+        b = generate_batch(n_variants=3, seed=99, include_thumbnails=False)
+        assert [v["lineage"]["seed"] for v in a["variants"]] == \
+            [v["lineage"]["seed"] for v in b["variants"]]
+
+    def test_unknown_recipe_rejected(self, vitis_root):
+        import pytest
+        with pytest.raises(ValueError):
+            generate_batch(recipes=[{"aug": "sharknado"}], n_variants=1)
+
+    def test_raremine_hook_skipped_for_test_batches(self, vitis_root):
+        # Test batches run under a redirected vitis root; routing into
+        # raremine's real candidate store must be skipped so tests never
+        # touch shared state. Real persisted batches route when raremine
+        # is importable (verified manually via the live API).
+        batch = generate_batch(n_variants=4, include_thumbnails=False)
+        hook = batch["raremine_hook"]
+        assert "available" in hook and "routed_candidates" in hook
+        assert hook["routed_candidates"] == 0
+        assert "ephemeral/test" in hook["note"]
+
+    def test_registry_listing(self):
+        names = {a["name"] for a in list_augmentations()}
+        assert {"sensor_noise", "low_light", "hdr_extreme", "lens_distortion",
+                "chromatic_aberration", "motion_blur", "glare"} <= names

tests/test_vitis/test_temporal.py

@@ -0,0 +1,130 @@
+"""Temporal profiling: flow recovery, engine ranking, stereo geometry."""
+
+import numpy as np
+
+from sensorflow.bevfusion.scenes import generate_sequences
+from sensorflow.vitis.backend import get_backend
+from sensorflow.vitis.render import (
+    BASELINE_M, FOCAL_PX, render_stereo_pair,
+)
+from sensorflow.vitis.temporal import (
+    _aggregate_engine, _engine_track_metrics, _match_engine_to_gt,
+    compute_flow_baseline, run_temporal_profile,
+)
+
+
+class TestOpticalFlowRecovery:
+    def test_recovers_planted_constant_velocity(self):
+        """A textured square moving at constant (dx, dy) px/frame must be
+        recovered by the pyramidal LK flow within tolerance."""
+        rng = np.random.default_rng(2)
+        h, w = 96, 128
+        tex = rng.uniform(0.4, 1.0, (18, 18)).astype(np.float32)
+        dx, dy = 3.0, 1.0
+
+        def frame(t):
+            img = np.full((h, w), 0.05, dtype=np.float32)
+            y0, x0 = int(30 + dy * t), int(20 + dx * t)
+            img[y0:y0 + 18, x0:x0 + 18] = tex
+            return img
+
+        for name in ("reference", "vitis_emulated"):
+            be = get_backend(name)
+            flow = be.optical_flow(frame(0), frame(1), levels=3, window=9,
+                                   iterations=3)
+            patch = flow[32:46, 22:36]
+            assert abs(float(np.median(patch[..., 0])) - dx) < 0.5, name
+            assert abs(float(np.median(patch[..., 1])) - dy) < 0.5, name
+
+    def test_flow_baseline_marks_tracks_continuous(self):
+        seqs = generate_sequences(1, 10, seed=7)
+        be = get_backend("reference")
+        records = compute_flow_baseline(seqs[0], be, seed=7)
+        continuous = [r["continuous"] for per in records.values()
+                      for r in per.values()]
+        assert sum(continuous) > 0.5 * len(continuous)
+
+
+class TestEngineRanking:
+    def test_unstable_engine_ranks_below_stable(self, vitis_root):
+        """The naive camera-only engine (frame-to-frame NN tracker, misses
+        under occlusion/night) must score below the fused engine."""
+        run = run_temporal_profile(n_sequences=2, frames_per_sequence=12,
+                                   seed=7)
+        for be_name in ("reference", "vitis_emulated"):
+            eng = run["results"][be_name]["engines"]
+            cam = eng["perception-v1-camera"]
+            fused = eng["perception-v3-bevfusion"]
+            assert fused["stability_score"] > cam["stability_score"], be_name
+            assert cam["flicker_rate"] >= fused["flicker_rate"], be_name
+            assert cam["fragmentation_per_track"] > \
+                fused["fragmentation_per_track"], be_name
+
+    def test_synthetic_flicker_engine_penalized(self):
+        """Deliberately dropping every third detection of a stable engine
+        must raise flicker and lower the aggregate score."""
+        seqs = generate_sequences(1, 14, seed=9)
+        seq = seqs[0]
+        be = get_backend("reference")
+        flow_records = compute_flow_baseline(seq, be, seed=9)
+
+        stable_out = {f.frame_id: [{"bbox_3d": g.bbox_3d,
+                                    "class_name": g.class_name,
+                                    "confidence": 0.9,
+                                    "track_id": g.instance_id}
+                                   for g in f.gt]
+                      for f in seq.frames}
+        flicker_out = {fid: (dets if i % 3 else dets[:max(0, len(dets) - 3)])
+                       for i, (fid, dets) in enumerate(stable_out.items())}
+
+        m_stable = _aggregate_engine([_engine_track_metrics(
+            seq, flow_records, _match_engine_to_gt(seq, stable_out))])
+        m_flicker = _aggregate_engine([_engine_track_metrics(
+            seq, flow_records, _match_engine_to_gt(seq, flicker_out))])
+        assert m_flicker["flicker_rate"] > m_stable["flicker_rate"]
+        assert m_flicker["stability_score"] < m_stable["stability_score"]
+
+    def test_backend_agreement_metacheck(self, vitis_root):
+        run = run_temporal_profile(n_sequences=2, frames_per_sequence=12,
+                                   seed=7, width_bits=12)
+        meta = run["backend_agreement"]
+        assert meta["ranking_agrees"] is True
+        assert meta["max_abs_score_delta"] < 5.0
+
+    def test_cohort_breakdown_present(self, vitis_root):
+        run = run_temporal_profile(n_sequences=3, frames_per_sequence=12,
+                                   seed=7)
+        cohorts = run["results"]["reference"]["engines"][
+            "perception-v3-bevfusion"]["cohorts"]
+        assert "night/clear" in cohorts and "occluded" in cohorts
+
+
+class TestStereoConsistency:
+    def test_disparity_depth_within_tolerance_of_geometry(self):
+        seqs = generate_sequences(1, 6, seed=7)
+        seq = seqs[0]
+        be = get_backend("reference")
+        rel_errors = []
+        for frame in seq.frames[:3]:
+            left, right, objs = render_stereo_pair(frame, seq, seed=7)
+            disp = be.stereo_block_match(left, right, max_disparity=48, block=9)
+            for o in objs:
+                u, v = int(round(o["u"])), int(round(o["v"]))
+                if not (4 <= v < disp.shape[0] - 4 and 4 <= u < disp.shape[1] - 4):
+                    continue
+                d = float(np.median(disp[v - 2:v + 3, u - 2:u + 3]))
+                if d < 1.0 or o["depth_m"] > 30.0:
+                    continue  # near field is the geometry-limited regime
+                depth = FOCAL_PX * BASELINE_M / d
+                rel_errors.append(abs(depth - o["depth_m"]) / o["depth_m"])
+        # Median over objects: individual matches can be corrupted by
+        # object overlap, but the population must track scene geometry.
+        assert len(rel_errors) >= 3
+        assert float(np.median(rel_errors)) < 0.15, rel_errors
+
+    def test_stereo_report_in_profile(self, vitis_root):
+        run = run_temporal_profile(n_sequences=1, frames_per_sequence=10,
+                                   seed=7)
+        stereo = run["results"]["reference"]["stereo"]
+        assert stereo["objects_checked"] > 0
+        assert stereo["median_abs_disparity_error_px"] < 2.0