IQ.Pilot Release Commit @ da36e21

iqpilot/selfdrive/iqmodeld/tests/dmon_lag/repro.cc

@@ -0,0 +1,96 @@
+// Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
+// clang++ -O2 repro.cc && ./a.out
+
+#include <sys/types.h>
+#include <unistd.h>
+
+#include <cstdint>
+#include <cstdio>
+#include <cstdlib>
+#include <ctime>
+#include <vector>
+
+namespace {
+
+constexpr int kModelWidth = 320;
+constexpr int kModelHeight = 640;
+constexpr int kRetryWindow = 20;
+constexpr int kSlowThresholdMs = 10;
+
+double millis_since_boot() {
+  timespec stamp{};
+#ifdef CLOCK_BOOTTIME
+  clock_gettime(CLOCK_BOOTTIME, &stamp);
+#else
+  clock_gettime(CLOCK_MONOTONIC, &stamp);
+#endif
+  return stamp.tv_sec * 1000.0 + stamp.tv_nsec * 1e-6;
+}
+
+inline float identity_input(uint8_t value) {
+  return value;
+}
+
+void pack_monitoring_tensor(uint8_t *nv12_frame, float *tensor_out) {
+  const int half_h = kModelHeight / 2;
+  const int half_w = kModelWidth / 2;
+  const int plane_area = half_w * half_h;
+  const int uv_base = kModelWidth * kModelHeight;
+
+  for (int row = 0; row < half_h; ++row) {
+    for (int col = 0; col < half_w; ++col) {
+      const int slot = col * half_h + row;
+      const int y_row = row * 2;
+      const int y_col = col * 2;
+
+      tensor_out[slot] = identity_input(nv12_frame[(y_row * kModelWidth) + y_col]);
+      tensor_out[slot + plane_area] = identity_input(nv12_frame[((y_row + 1) * kModelWidth) + y_col]);
+      tensor_out[slot + (plane_area * 2)] = identity_input(nv12_frame[(y_row * kModelWidth) + y_col + 1]);
+      tensor_out[slot + (plane_area * 3)] = identity_input(nv12_frame[((y_row + 1) * kModelWidth) + y_col + 1]);
+      tensor_out[slot + (plane_area * 4)] = identity_input(nv12_frame[uv_base + (row * half_w) + col]);
+      tensor_out[slot + (plane_area * 5)] = identity_input(nv12_frame[uv_base + plane_area + (row * half_w) + col]);
+    }
+  }
+}
+
+double average_runtime_ms(uint8_t *nv12_frame, float *tensor_out) {
+  double total_ms = 0.0;
+  for (int i = 0; i < kRetryWindow; ++i) {
+    const double start_ms = millis_since_boot();
+    pack_monitoring_tensor(nv12_frame, tensor_out);
+    total_ms += millis_since_boot() - start_ms;
+  }
+  return total_ms / static_cast<double>(kRetryWindow);
+}
+
+void dump_stall_trace(uint8_t *nv12_frame, float *tensor_out) {
+  for (int i = 0; i < 200; ++i) {
+    const double start_ms = millis_since_boot();
+    pack_monitoring_tensor(nv12_frame, tensor_out);
+    printf("%.2f   ", millis_since_boot() - start_ms);
+  }
+  printf("\n");
+}
+
+}  // namespace
+
+int main() {
+  const size_t nv12_bytes = kModelWidth * kModelHeight * 3 / 2;
+  const size_t tensor_floats = (kModelWidth / 2) * (kModelHeight / 2) * 6;
+
+  while (true) {
+    auto *nv12_frame = static_cast<uint8_t *>(malloc(nv12_bytes));
+    auto *tensor_out = static_cast<float *>(malloc(tensor_floats * sizeof(float)));
+    printf("allocate -- %p 0x%zx -- %p 0x%zx\n", nv12_frame, nv12_bytes, tensor_out, tensor_floats * sizeof(float));
+
+    const double mean_ms = average_runtime_ms(nv12_frame, tensor_out);
+    if (mean_ms > kSlowThresholdMs) {
+      printf("HIT %.2f\n", mean_ms);
+      printf("BAD\n");
+      dump_stall_trace(nv12_frame, tensor_out);
+      return 0;
+    }
+
+    printf("got %.2f\n", mean_ms);
+  }
+}

iqpilot/selfdrive/iqmodeld/tests/test_action_dispatch.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+
+from iqpilot.cereal import log
+
+from iqpilot.selfdrive.iqmodeld.config import Plan
+from iqpilot.selfdrive.iqmodeld.daemon import NeuralEngineState, _merged_plan
+import iqpilot.selfdrive.iqmodeld.daemon as iqmodeld_daemon
+from iqpilot.selfdrive.controls.lib.drive_helpers import smooth_value
+
+
+def _fake_state(**overrides):
+  base = dict(
+    PLANPLUS_CONTROL=1.0,
+    LONG_SMOOTH_SECONDS=0.3,
+    LAT_SMOOTH_SECONDS=0.1,
+    MIN_LAT_CONTROL_SPEED=0.3,
+    mlsim=True,
+    generation=12,
+    constants=SimpleNamespace(T_IDXS=np.arange(100), DESIRE_LEN=8),
+  )
+  base.update(overrides)
+  return SimpleNamespace(**base)
+
+
+@pytest.mark.parametrize(
+  ("control", "vego", "factor"),
+  [
+    (0.55, 20.0, 1.0),
+    (1.0, 25.0, 0.75),
+    (1.5, 25.1, 0.75),
+    (2.0, 20.0, 1.0),
+  ],
+)
+def test_planplus_merge_matches_speed_gate(control: float, vego: float, factor: float):
+  state = _fake_state(PLANPLUS_CONTROL=control)
+  base = np.random.rand(1, 100, 15).astype(np.float32)
+  extra = np.random.rand(1, 100, 15).astype(np.float32)
+  merged = _merged_plan(state, {"plan": base, "planplus": extra}, vego)
+  expected = base[0] + (control * factor) * extra[0]
+  np.testing.assert_allclose(merged, expected, rtol=1e-6, atol=1e-6)
+
+
+def test_action_dispatch_uses_merged_plan_for_longitudinal_choice(monkeypatch: pytest.MonkeyPatch):
+  state = _fake_state()
+  previous = log.ModelDataV2.Action()
+  recorded_velocity: list[np.ndarray] = []
+
+  def fake_accel(plan_vel, plan_accel, t_idxs, action_t=0.0):
+    recorded_velocity.append(plan_vel.copy())
+    return 0.0, False
+
+  monkeypatch.setattr(iqmodeld_daemon, "get_accel_from_plan", fake_accel)
+  monkeypatch.setattr(iqmodeld_daemon, "pick_curvature", lambda *args: 0.0)
+
+  plan = np.random.rand(1, 100, 15).astype(np.float32)
+  planplus = np.random.rand(1, 100, 15).astype(np.float32)
+  outputs = {"plan": plan.copy(), "planplus": planplus.copy()}
+
+  NeuralEngineState.get_action_from_model(state, outputs, previous, 0.0, 0.0, 25.0)
+  expected = plan[0, :, Plan.VELOCITY][:, 0] + 0.75 * planplus[0, :, Plan.VELOCITY][:, 0]
+  np.testing.assert_allclose(recorded_velocity[0], expected, rtol=1e-5, atol=1e-6)
+
+
+def test_action_dispatch_honors_direct_action_outputs():
+  state = _fake_state(mlsim=False, generation=9)
+  previous = log.ModelDataV2.Action(desiredCurvature=0.0, desiredAcceleration=0.0, shouldStop=False)
+  outputs = {"action": np.array([[4.0, -0.25]], dtype=np.float32)}
+  action = NeuralEngineState.get_action_from_model(state, outputs, previous, 0.0, 0.0, 10.0)
+  expected_accel = smooth_value(-0.25, previous.desiredAcceleration, state.LONG_SMOOTH_SECONDS)
+  expected_curvature = smooth_value(0.04, previous.desiredCurvature, state.LAT_SMOOTH_SECONDS)
+  assert action.desiredAcceleration == pytest.approx(expected_accel)
+  assert action.desiredCurvature == pytest.approx(expected_curvature)
+  assert action.shouldStop is False

iqpilot/selfdrive/iqmodeld/tests/test_combined_split_runner.py

@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+import numpy as np
+
+from iqpilot.selfdrive.iqmodeld.models.combined_artifact import resolve_combined_split_artifact
+import iqpilot.selfdrive.iqmodeld.models.runners.model_runner as runner_helpers
+from iqpilot.selfdrive.iqmodeld.models.runners.tinygrad import tinygrad_runner as tinygrad_runner_mod
+from iqpilot.selfdrive.iqmodeld.models.runners.tinygrad.combined_split_runner import TinygradCombinedSplitRunner
+from iqpilot.selfdrive.iqmodeld.models.runners.tinygrad import combined_split_runner as combined_runner_mod
+from iqpilot.selfdrive.iqmodeld.models.runners.model_runner import ModelType
+from iqpilot.selfdrive.iqmodeld.parser import PhaseParser
+from iqpilot.selfdrive.iqmodeld.tests.test_iqmodeld_contracts import _phase_sample
+
+
+@dataclass
+class _TypeWrap:
+  raw: int
+
+
+@dataclass
+class _Artifact:
+  fileName: str
+
+
+class _Model:
+  def __init__(self, model_type: int, artifact_name: str):
+    self.type = _TypeWrap(model_type)
+    self.artifact = _Artifact(artifact_name)
+
+
+class _Override:
+  def __init__(self, key: str, value: str):
+    self.key = key
+    self.value = value
+
+
+class _Bundle:
+  def __init__(self, models: list[_Model], overrides: list[_Override] | None = None, generation: int = 10):
+    self.models = models
+    self.overrides = overrides or []
+    self.generation = generation
+
+
+class _FakeTensor:
+  def __init__(self, values):
+    self._values = np.asarray(values, dtype=np.float32)
+
+  def numpy(self):
+    return self._values
+
+
+class _FakeVisionBuf:
+  width = 1928
+  height = 1208
+  data = memoryview(b"\x00" * 64)
+
+
+def _slice_pack(outputs: dict[str, np.ndarray]) -> tuple[np.ndarray, dict[str, slice]]:
+  chunks = []
+  slices: dict[str, slice] = {}
+  cursor = 0
+  for name, value in outputs.items():
+    flat = value.reshape(-1)
+    slices[name] = slice(cursor, cursor + flat.size)
+    chunks.append(flat)
+    cursor += flat.size
+  return np.concatenate(chunks).astype(np.float32), slices
+
+
+def test_resolve_combined_split_artifact_prefers_override(tmp_path: Path, monkeypatch):
+  bundle = _Bundle(
+    [_Model(ModelType.vision, "driving_vision_demo_tinygrad.pkl"), _Model(ModelType.policy, "driving_policy_demo_tinygrad.pkl")],
+    overrides=[_Override("combinedRuntimeArtifact", "driving_combined_demo.pkl")],
+  )
+  expected = tmp_path / "driving_combined_demo.pkl"
+  expected.write_bytes(b"iq")
+
+  monkeypatch.setattr("iqpilot.selfdrive.iqmodeld.models.combined_artifact._MODEL_ROOT", tmp_path)
+
+  assert resolve_combined_split_artifact(bundle) == expected
+
+
+def test_get_model_runner_prefers_combined_split_artifact(monkeypatch):
+  bundle = _Bundle([
+    _Model(ModelType.vision, "driving_vision_demo_tinygrad.pkl"),
+    _Model(ModelType.policy, "driving_policy_demo_tinygrad.pkl"),
+  ], generation=11)
+
+  marker = object()
+  monkeypatch.setattr(runner_helpers, "_fetch_bundle", lambda: bundle)
+  monkeypatch.setattr(runner_helpers, "has_combined_split_artifact", lambda _: True)
+  monkeypatch.setattr(combined_runner_mod, "TinygradCombinedSplitRunner", lambda: marker)
+
+  assert runner_helpers.get_model_runner() is marker
+
+
+def test_get_model_runner_keeps_split_bundle_on_existing_runner_without_combined_artifact(monkeypatch):
+  bundle = _Bundle([
+    _Model(ModelType.vision, "driving_vision_demo_tinygrad.pkl"),
+    _Model(ModelType.policy, "driving_policy_demo_tinygrad.pkl"),
+  ], generation=12)
+
+  marker = object()
+  monkeypatch.setattr(runner_helpers, "_fetch_bundle", lambda: bundle)
+  monkeypatch.setattr(runner_helpers, "has_combined_split_artifact", lambda _: False)
+  monkeypatch.setattr(tinygrad_runner_mod, "TinygradSplitRunner", lambda: marker)
+
+  assert runner_helpers.get_model_runner() is marker
+
+
+def test_combined_split_runner_parses_single_policy_payload(monkeypatch):
+  vision_raw = _phase_sample(np.random.default_rng(11))
+  policy_raw = _phase_sample(np.random.default_rng(17))
+  vision_blob, vision_slices = _slice_pack(vision_raw)
+  policy_blob, policy_slices = _slice_pack(policy_raw)
+
+  runner = TinygradCombinedSplitRunner.__new__(TinygradCombinedSplitRunner)
+  runner._vision_meta = {
+    "input_shapes": {"img": (1, 12, 128, 256), "big_img": (1, 12, 128, 256)},
+    "output_slices": vision_slices,
+  }
+  runner._meta_by_role = {
+    "vision": runner._vision_meta,
+    "policy": {
+      "input_shapes": {
+        "features_buffer": (1, 25, 512),
+        "desire_pulse": (1, 25, 8),
+        "traffic_convention": (1, 2),
+        "action_t": (1, 2),
+      },
+      "output_slices": policy_slices,
+    },
+  }
+  runner._policy_roles = ["policy"]
+  runner._desired_key = "desire_pulse"
+  runner._road_key = "img"
+  runner._wide_key = "big_img"
+  runner._extra_policy_keys = []
+  runner._queue_tensors = {
+    "img_q": object(),
+    "big_img_q": object(),
+    "feat_q": object(),
+    "desire_q": object(),
+    "tfm": object(),
+    "big_tfm": object(),
+    "desire": object(),
+    "traffic_convention": object(),
+    "action_t": object(),
+  }
+  runner._numpy_state = {
+    "tfm": np.zeros((3, 3), dtype=np.float32),
+    "big_tfm": np.zeros((3, 3), dtype=np.float32),
+    "desire": np.zeros(8, dtype=np.float32),
+    "traffic_convention": np.zeros((1, 2), dtype=np.float32),
+    "action_t": np.zeros((1, 2), dtype=np.float32),
+  }
+  runner._camera_shape = (1928, 1208)
+  runner._camera_programs = {
+    (1928, 1208): {"stage_inputs": lambda **kwargs: ("road", "wide")},
+  }
+  runner._execute_bundle = lambda **kwargs: (_FakeTensor(vision_blob), _FakeTensor(policy_blob))
+  runner._parser = PhaseParser()
+  runner._last_desire = np.zeros(8, dtype=np.float32)
+  runner._blob_cache = {}
+
+  monkeypatch.setattr(TinygradCombinedSplitRunner, "_allocate_runtime_state", lambda self, w, h: None)
+  monkeypatch.setattr(TinygradCombinedSplitRunner, "_frame_blob", lambda self, name, buf: object())
+
+  outputs = runner.run_fused(
+    {"img": _FakeVisionBuf(), "big_img": _FakeVisionBuf()},
+    {"img": np.eye(3, dtype=np.float32), "big_img": np.eye(3, dtype=np.float32)},
+    {
+      "desire_pulse": np.array([1, 0, 0, 0, 0, 0, 0, 0], dtype=np.float32),
+      "traffic_convention": np.zeros((1, 2), dtype=np.float32),
+      "action_t": np.zeros((1, 2), dtype=np.float32),
+    },
+  )
+
+  assert "pose" in outputs
+  assert "plan" in outputs
+  assert outputs["plan"].shape == (1, 33, 15)
+  assert outputs["action"].shape == (1, 2)

iqpilot/selfdrive/iqmodeld/tests/test_compile_supercombo_guards.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+
+import numpy as np
+
+from iqpilot.selfdrive.iqmodeld.tools.compile_supercombo import (
+  _captured_devices,
+  _validate_pose_outputs,
+)
+from iqpilot.selfdrive.iqmodeld.models.runners.tinygrad.supercombo_runner import _captured_queue_depth
+
+
+class _Captured:
+  def __init__(self, expected_input_info):
+    self.expected_input_info = expected_input_info
+
+
+class _FakeJit:
+  def __init__(self, expected_input_info):
+    self.captured = _Captured(expected_input_info)
+
+
+def test_captured_queue_helpers_extract_depth_and_device():
+  infos = [
+    ("noop", (), "uchar", "QCOM"),
+    ("reshape(arg=None, src=(noop, stack(arg=None, src=(const(arg=5), const(arg=6), const(arg=128), const(arg=256)))))", (), "uchar", "QCOM"),
+    ("reshape(arg=None, src=(noop, const(arg=3)))", (), "float", "NPY"),
+  ]
+  fake_jit = _FakeJit(infos)
+
+  assert _captured_queue_depth(fake_jit) == 5
+  assert _captured_devices(fake_jit) == {"QCOM", "NPY"}
+
+
+def test_validate_pose_outputs_accepts_sane_odometry_payload():
+  outputs = {
+    "pose": np.array([[1.0, 0.5, 0.25, 0.1, 0.2, 0.3]], dtype=np.float32),
+    "pose_stds": np.array([[0.5, 0.4, 0.3, 0.2, 0.2, 0.2]], dtype=np.float32),
+    "wide_from_device_euler": np.array([[0.1, 0.2, 0.3]], dtype=np.float32),
+    "wide_from_device_euler_stds": np.array([[0.2, 0.2, 0.2]], dtype=np.float32),
+    "road_transform": np.array([[0.5, 0.4, 0.3, 0.2, 0.1, 0.0]], dtype=np.float32),
+    "road_transform_stds": np.array([[0.3, 0.3, 0.3, 0.2, 0.2, 0.2]], dtype=np.float32),
+  }
+
+  _validate_pose_outputs(outputs)

iqpilot/selfdrive/iqmodeld/tests/test_iqmodeld_contracts.py

@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+import copy
+
+import iqpilot.cereal.messaging as messaging
+import numpy as np
+from iqpilot.cereal import log
+
+from iqpilot.selfdrive.iqmodeld.config import Meta, ModelConstants
+from iqpilot.selfdrive.iqmodeld.messaging import (
+  DrivePacketMemory,
+  pick_curvature,
+  populate_drive_messages,
+  populate_odometry_message,
+)
+from iqpilot.selfdrive.iqmodeld.models.split_model_constants import SplitModelConstants
+from iqpilot.selfdrive.iqmodeld.parser import ArchiveParser, PhaseParser
+
+
+def _archive_sample(rng: np.random.Generator) -> dict[str, np.ndarray]:
+  return {
+    "plan": rng.standard_normal((1, ModelConstants.PLAN_MHP_N * (2 * ModelConstants.IDX_N * ModelConstants.PLAN_WIDTH + ModelConstants.PLAN_MHP_SELECTION)), dtype=np.float32),
+    "lane_lines": rng.standard_normal((1, 2 * ModelConstants.NUM_LANE_LINES * ModelConstants.IDX_N * ModelConstants.LANE_LINES_WIDTH), dtype=np.float32),
+    "road_edges": rng.standard_normal((1, 2 * ModelConstants.NUM_ROAD_EDGES * ModelConstants.IDX_N * ModelConstants.LANE_LINES_WIDTH), dtype=np.float32),
+    "pose": rng.standard_normal((1, 2 * ModelConstants.POSE_WIDTH), dtype=np.float32),
+    "road_transform": rng.standard_normal((1, 2 * ModelConstants.POSE_WIDTH), dtype=np.float32),
+    "sim_pose": rng.standard_normal((1, 2 * ModelConstants.POSE_WIDTH), dtype=np.float32),
+    "wide_from_device_euler": rng.standard_normal((1, 2 * ModelConstants.WIDE_FROM_DEVICE_WIDTH), dtype=np.float32),
+    "lead": rng.standard_normal((1, ModelConstants.LEAD_MHP_N * (2 * ModelConstants.LEAD_TRAJ_LEN * ModelConstants.LEAD_WIDTH + ModelConstants.LEAD_MHP_SELECTION)), dtype=np.float32),
+    "lat_planner_solution": rng.standard_normal((1, 2 * ModelConstants.IDX_N * ModelConstants.LAT_PLANNER_SOLUTION_WIDTH), dtype=np.float32),
+    "desired_curvature": rng.standard_normal((1, 2 * ModelConstants.DESIRED_CURV_WIDTH), dtype=np.float32),
+    "lead_prob": rng.standard_normal((1, ModelConstants.LEAD_MHP_SELECTION), dtype=np.float32),
+    "lane_lines_prob": rng.standard_normal((1, ModelConstants.NUM_LANE_LINES * 2), dtype=np.float32),
+    "meta": rng.standard_normal((1, 55), dtype=np.float32),
+    "desire_state": rng.standard_normal((1, ModelConstants.DESIRE_PRED_WIDTH), dtype=np.float32),
+    "desire_pred": rng.standard_normal((1, ModelConstants.DESIRE_PRED_LEN * ModelConstants.DESIRE_PRED_WIDTH), dtype=np.float32),
+  }
+
+
+def _phase_sample(rng: np.random.Generator) -> dict[str, np.ndarray]:
+  c = SplitModelConstants
+  return {
+    "pose": rng.standard_normal((1, 2 * c.POSE_WIDTH), dtype=np.float32),
+    "wide_from_device_euler": rng.standard_normal((1, 2 * c.WIDE_FROM_DEVICE_WIDTH), dtype=np.float32),
+    "road_transform": rng.standard_normal((1, 2 * c.POSE_WIDTH), dtype=np.float32),
+    "lead": rng.standard_normal((1, c.LEAD_MHP_N * (2 * c.LEAD_TRAJ_LEN * c.LEAD_WIDTH + c.LEAD_MHP_SELECTION)), dtype=np.float32),
+    "plan": rng.standard_normal((1, c.PLAN_MHP_N * (2 * c.IDX_N * c.PLAN_WIDTH + c.PLAN_MHP_SELECTION)), dtype=np.float32),
+    "planplus": rng.standard_normal((1, 2 * c.IDX_N * c.PLAN_WIDTH), dtype=np.float32),
+    "action": rng.standard_normal((1, 2 * c.ACTION_WIDTH), dtype=np.float32),
+    "desired_curvature": rng.standard_normal((1, 2 * c.DESIRED_CURV_WIDTH), dtype=np.float32),
+    "desire_pred": rng.standard_normal((1, c.DESIRE_PRED_LEN * c.DESIRE_PRED_WIDTH), dtype=np.float32),
+    "desire_state": rng.standard_normal((1, c.DESIRE_PRED_WIDTH), dtype=np.float32),
+    "lane_lines": rng.standard_normal((1, 2 * c.NUM_LANE_LINES * c.IDX_N * c.LANE_LINES_WIDTH), dtype=np.float32),
+    "lane_lines_prob": rng.standard_normal((1, c.NUM_LANE_LINES * 2), dtype=np.float32),
+    "lead_prob": rng.standard_normal((1, c.LEAD_MHP_SELECTION), dtype=np.float32),
+    "lat_planner_solution": rng.standard_normal((1, 2 * c.IDX_N * c.LAT_PLANNER_SOLUTION_WIDTH), dtype=np.float32),
+    "meta": rng.standard_normal((1, 55), dtype=np.float32),
+    "road_edges": rng.standard_normal((1, 2 * c.NUM_ROAD_EDGES * c.IDX_N * c.LANE_LINES_WIDTH), dtype=np.float32),
+    "sim_pose": rng.standard_normal((1, 2 * c.POSE_WIDTH), dtype=np.float32),
+  }
+
+
+def test_archive_parser_contract_snapshot():
+  outputs = ArchiveParser().parse_outputs(copy.deepcopy(_archive_sample(np.random.default_rng(7))))
+
+  assert outputs["plan"].shape == (1, 33, 15)
+  assert outputs["lane_lines"].shape == (1, 4, 33, 2)
+  assert outputs["road_edges"].shape == (1, 2, 33, 2)
+  assert outputs["desire_pred"].shape == (1, 4, 8)
+
+  np.testing.assert_allclose(outputs["pose"][0, 0], 0.45617363, rtol=1e-6, atol=1e-6)
+  np.testing.assert_allclose(outputs["lane_lines_prob"][0, 2], 0.85733712, rtol=1e-6, atol=1e-6)
+  np.testing.assert_allclose(outputs["desire_state"][0, 0], 0.44964141, rtol=1e-6, atol=1e-6)
+  np.testing.assert_allclose(outputs["lead_prob"][0, 0], 0.21613698, rtol=1e-6, atol=1e-6)
+
+
+def test_phase_parser_contract_snapshot():
+  raw = _phase_sample(np.random.default_rng(23))
+  outputs = {**PhaseParser().parse_vision_outputs(copy.deepcopy(raw)), **PhaseParser().parse_policy_outputs(copy.deepcopy(raw))}
+
+  assert outputs["plan"].shape == (1, 33, 15)
+  assert outputs["action"].shape == (1, 2)
+  assert outputs["desired_curvature"].shape == (1, 1)
+  assert outputs["road_edges"].shape == (1, 2, 33, 2)
+
+  np.testing.assert_allclose(outputs["plan"][0, 0, 0], 0.09684439, rtol=1e-6, atol=1e-6)
+  np.testing.assert_allclose(outputs["action"][0, 0], 0.25458091, rtol=1e-6, atol=1e-6)
+  np.testing.assert_allclose(outputs["desired_curvature"][0, 0], -0.97072351, rtol=1e-6, atol=1e-6)
+  np.testing.assert_allclose(outputs["lane_lines_prob"][0, 0], 0.20073657, rtol=1e-6, atol=1e-6)
+
+
+def test_message_population_contract_snapshot():
+  raw = _phase_sample(np.random.default_rng(23))
+  outputs = {**PhaseParser().parse_vision_outputs(copy.deepcopy(raw)), **PhaseParser().parse_policy_outputs(copy.deepcopy(raw))}
+  action = log.ModelDataV2.Action(desiredCurvature=0.031, desiredAcceleration=-0.12, shouldStop=False)
+
+  driving_msg = messaging.new_message("drivingModelData")
+  model_msg = messaging.new_message("modelV2")
+  odometry_msg = messaging.new_message("cameraOdometry")
+  memory = DrivePacketMemory()
+
+  populate_drive_messages(
+    driving_msg, model_msg, outputs, action, memory,
+    2468, 2470, 2480, 0.05, 123456789, 0.014, True, Meta,
+  )
+  populate_odometry_message(odometry_msg, outputs, 2468, 0, 123456789, True)
+
+  np.testing.assert_allclose(driving_msg.drivingModelData.laneLineMeta.leftY, -0.21672775, rtol=1e-6, atol=1e-6)
+  np.testing.assert_allclose(model_msg.modelV2.meta.engagedProb, 0.64853197, rtol=1e-6, atol=1e-6)
+  np.testing.assert_allclose(odometry_msg.cameraOdometry.trans[0], 0.0360266, rtol=1e-6, atol=1e-6)
+  assert int(model_msg.modelV2.confidence.raw) == 2
+
+
+def test_curvature_selection_contract_snapshot():
+  raw = _phase_sample(np.random.default_rng(23))
+  outputs = {**PhaseParser().parse_vision_outputs(copy.deepcopy(raw)), **PhaseParser().parse_policy_outputs(copy.deepcopy(raw))}
+  plan_rows = outputs["plan"][0]
+
+  direct = pick_curvature(outputs, plan_rows, 27.5, 0.8, synthetic_lane_logic=False)
+  fallback = pick_curvature(outputs, plan_rows, 27.5, 0.8, synthetic_lane_logic=True)
+
+  np.testing.assert_allclose(direct, -0.97072351, rtol=1e-6, atol=1e-6)
+  np.testing.assert_allclose(fallback, -0.0689389, rtol=1e-6, atol=1e-6)

iqpilot/selfdrive/iqmodeld/tests/test_model_runner_smoke.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+import numpy as np
+import pytest
+from tinygrad.tensor import Tensor
+
+import iqpilot.selfdrive.iqmodeld.models.helpers as bundle_helpers
+import iqpilot.selfdrive.iqmodeld.models.runners.model_runner as model_runner_mod
+import iqpilot.selfdrive.iqmodeld.models.runners.tinygrad.tinygrad_runner as tinygrad_runner_mod
+from iqpilot.selfdrive.iqmodeld.models.runners.model_runner import ModelType
+from iqpilot.selfdrive.iqmodeld.models.runners.tinygrad.tinygrad_runner import TinygradRunner
+
+
+LOCAL_MODEL_DIR = Path(__file__).resolve().parents[1] / "default_model"
+
+
+@dataclass
+class _TypeWrap:
+  raw: int
+
+
+@dataclass
+class _Artifact:
+  fileName: str
+
+
+class _Model:
+  def __init__(self, model_type: int, artifact_name: str, metadata_name: str):
+    self.type = _TypeWrap(model_type)
+    self.artifact = _Artifact(artifact_name)
+    self.metadata = _Artifact(metadata_name)
+
+
+class _Bundle:
+  def __init__(self, models: list[_Model], is_20hz: bool = False):
+    self.models = models
+    self.is20hz = is_20hz
+
+
+def _seed_runner_inputs(runner: TinygradRunner) -> None:
+  for name, shape in runner.input_shapes.items():
+    runner.inputs[name] = Tensor(
+      np.zeros(shape, dtype=np.float32),
+      device=runner.input_to_device[name],
+      dtype=runner.input_to_dtype[name],
+    ).realize()
+
+
+@pytest.mark.tici
+def test_local_tinygrad_models_execute(monkeypatch):
+  bundle = _Bundle([
+    _Model(ModelType.vision, "driving_vision_c210m_tinygrad.pkl", "driving_vision_c210m_metadata.pkl"),
+    _Model(ModelType.policy, "driving_policy_c210m_tinygrad.pkl", "driving_policy_c210m_metadata.pkl"),
+  ])
+
+  monkeypatch.setattr(bundle_helpers, "get_active_bundle", lambda params=None: bundle, raising=False)
+  monkeypatch.setattr(model_runner_mod, "_fetch_bundle", lambda params=None: bundle)
+  monkeypatch.setattr(tinygrad_runner_mod, "CUSTOM_MODEL_PATH", str(LOCAL_MODEL_DIR), raising=False)
+  monkeypatch.setattr(model_runner_mod, "CUSTOM_MODEL_PATH", str(LOCAL_MODEL_DIR), raising=False)
+
+  vision_runner = TinygradRunner(ModelType.vision)
+  _seed_runner_inputs(vision_runner)
+  vision_outputs = vision_runner.run_model()
+  assert "pose" in vision_outputs
+  assert "lane_lines" in vision_outputs
+
+  policy_runner = TinygradRunner(ModelType.policy)
+  _seed_runner_inputs(policy_runner)
+  policy_outputs = policy_runner.run_model()
+  assert "plan" in policy_outputs
+  assert "desire_state" in policy_outputs

iqpilot/selfdrive/iqmodeld/tests/test_public_surface.py

@@ -0,0 +1,18 @@
+from iqpilot.selfdrive.iqmodeld import metadata, messaging, parser
+from iqpilot.selfdrive.iqmodeld.daemon import CaptureStamp, NeuralEngineState
+
+
+def test_public_module_surface():
+  assert hasattr(messaging, "DrivePacketMemory")
+  assert hasattr(messaging, "pick_curvature")
+  assert hasattr(messaging, "populate_drive_messages")
+  assert hasattr(messaging, "populate_odometry_message")
+
+  assert hasattr(parser, "ArchiveParser")
+  assert hasattr(parser, "PhaseParser")
+
+  assert hasattr(metadata, "select_meta_layout")
+  assert hasattr(metadata, "build_metadata_record")
+
+  assert CaptureStamp.__name__ == "CaptureStamp"
+  assert NeuralEngineState.__name__ == "NeuralEngineState"

iqpilot/selfdrive/iqmodeld/tests/test_selector_share_smoke.py

@@ -0,0 +1,163 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+import numpy as np
+import pytest
+from tinygrad.nn.onnx import OnnxRunner
+from tinygrad.tensor import Tensor
+
+import iqpilot.selfdrive.iqmodeld.models.helpers as bundle_helpers
+import iqpilot.selfdrive.iqmodeld.models.runners.model_runner as model_runner_mod
+import iqpilot.selfdrive.iqmodeld.models.runners.tinygrad.tinygrad_runner as tinygrad_runner_mod
+from iqpilot.selfdrive.iqmodeld.models.runners.model_runner import ModelType
+from iqpilot.selfdrive.iqmodeld.models.runners.tinygrad.tinygrad_runner import TinygradRunner
+
+
+SHARE_ROOT = Path(os.getenv("IQPILOT_SELECTOR_SHARE", "/Volumes/New New Vault/IQModels/models/recompiled16"))
+
+
+@dataclass
+class _TypeWrap:
+  raw: int
+
+
+@dataclass
+class _Artifact:
+  fileName: str
+
+
+class _Model:
+  def __init__(self, model_type: int, artifact_name: str, metadata_name: str):
+    self.type = _TypeWrap(model_type)
+    self.artifact = _Artifact(artifact_name)
+    self.metadata = _Artifact(metadata_name)
+
+
+class _Bundle:
+  def __init__(self, models: list[_Model], is_20hz: bool = False):
+    self.models = models
+    self.is20hz = is_20hz
+
+
+def _find_selector_dirs(limit: int = 3, require_onnx: bool = False) -> list[Path]:
+  found: list[Path] = []
+  if not SHARE_ROOT.is_dir():
+    return found
+
+  for bundle_dir in sorted(SHARE_ROOT.iterdir()):
+    if not bundle_dir.is_dir():
+      continue
+    vision = next(bundle_dir.glob("driving_vision*_tinygrad.pkl"), None)
+    policy = next(bundle_dir.glob("driving_policy*_tinygrad.pkl"), None)
+    vision_meta = next(bundle_dir.glob("driving_vision*_metadata.pkl"), None)
+    policy_meta = next(bundle_dir.glob("driving_policy*_metadata.pkl"), None)
+    has_onnx = (bundle_dir / "driving_vision.onnx").is_file() and (bundle_dir / "driving_policy.onnx").is_file()
+    if vision and policy and vision_meta and policy_meta and (has_onnx or not require_onnx):
+      found.append(bundle_dir)
+    if len(found) >= limit:
+      break
+  return found
+
+
+def _seed_runner_inputs(runner: TinygradRunner) -> None:
+  for name, shape in runner.input_shapes.items():
+    runner.inputs[name] = Tensor(
+      np.zeros(shape, dtype=np.float32),
+      device=runner.input_to_device[name],
+      dtype=runner.input_to_dtype[name],
+    ).realize()
+
+
+def _bundle_for_dir(bundle_dir: Path) -> _Bundle:
+  vision = next(bundle_dir.glob("driving_vision*_tinygrad.pkl"))
+  policy = next(bundle_dir.glob("driving_policy*_tinygrad.pkl"))
+  vision_meta = next(bundle_dir.glob("driving_vision*_metadata.pkl"))
+  policy_meta = next(bundle_dir.glob("driving_policy*_metadata.pkl"))
+  return _Bundle([
+    _Model(ModelType.vision, vision.name, vision_meta.name),
+    _Model(ModelType.policy, policy.name, policy_meta.name),
+  ])
+
+
+def _run_tinygrad_bundle(bundle_dir: Path, monkeypatch):
+  bundle = _bundle_for_dir(bundle_dir)
+  monkeypatch.setattr(bundle_helpers, "get_active_bundle", lambda params=None: bundle, raising=False)
+  monkeypatch.setattr(model_runner_mod, "get_active_bundle", lambda params=None: bundle, raising=False)
+  monkeypatch.setattr(model_runner_mod, "_fetch_bundle", lambda: bundle)
+  monkeypatch.setattr(tinygrad_runner_mod, "CUSTOM_MODEL_PATH", str(bundle_dir), raising=False)
+  monkeypatch.setattr(model_runner_mod, "CUSTOM_MODEL_PATH", str(bundle_dir), raising=False)
+
+  vision_runner = TinygradRunner(ModelType.vision)
+  _seed_runner_inputs(vision_runner)
+  vision_outputs = vision_runner.run_model()
+
+  policy_runner = TinygradRunner(ModelType.policy)
+  _seed_runner_inputs(policy_runner)
+  policy_outputs = policy_runner.run_model()
+
+  return vision_outputs, policy_outputs
+
+
+def _run_onnx_bundle(bundle_dir: Path):
+  vision_session = OnnxRunner(bundle_dir / "driving_vision.onnx")
+  policy_session = OnnxRunner(bundle_dir / "driving_policy.onnx")
+
+  def seed_inputs(session):
+    seeded = {}
+    for name, spec in session.graph_inputs.items():
+      dtype_text = str(spec.dtype).lower()
+      if "uchar" in dtype_text or "uint8" in dtype_text:
+        seeded[name] = Tensor(np.zeros(spec.shape, dtype=np.uint8))
+      elif "half" in dtype_text or "float16" in dtype_text:
+        seeded[name] = Tensor(np.zeros(spec.shape, dtype=np.float16))
+      else:
+        seeded[name] = Tensor(np.zeros(spec.shape, dtype=np.float32))
+    return seeded
+
+  return (
+    vision_session(seed_inputs(vision_session))["outputs"].numpy().flatten(),
+    policy_session(seed_inputs(policy_session))["outputs"].numpy().flatten(),
+  )
+
+
+@pytest.mark.skipif(not SHARE_ROOT.is_dir(), reason="selector model share is not mounted")
+def test_three_selector_models_parse_via_share_onnx():
+  selector_dirs = _find_selector_dirs(limit=3, require_onnx=True)
+  assert len(selector_dirs) >= 3
+
+  for bundle_dir in selector_dirs:
+    vision_raw, policy_raw = _run_onnx_bundle(bundle_dir)
+    assert vision_raw.size > 0
+    assert policy_raw.size > 0
+
+
+@pytest.mark.skipif(not SHARE_ROOT.is_dir(), reason="selector model share is not mounted")
+def test_selector_tinygrad_pkls_execute_when_host_compatible(monkeypatch):
+  selector_dirs = _find_selector_dirs(limit=10)
+  attempted = 0
+  executed = 0
+
+  for bundle_dir in selector_dirs:
+    attempted += 1
+    try:
+      vision_outputs, policy_outputs = _run_tinygrad_bundle(bundle_dir, monkeypatch)
+    except AssertionError as exc:
+      if "Model was built on C3 or C3X" in str(exc):
+        continue
+      raise
+    except FileNotFoundError as exc:
+      if "/dev/kgsl-3d0" in str(exc):
+        continue
+      raise
+
+    assert "pose" in vision_outputs
+    assert "plan" in policy_outputs
+    executed += 1
+    if executed >= 3:
+      break
+
+  if executed == 0:
+    pytest.skip(f"share tinygrad pkls are QCOM-only on this host; inspected {attempted} bundles")

iqpilot/selfdrive/iqmodeld/tests/test_supercombo_runner_guards.py

@@ -0,0 +1,226 @@
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+
+import pytest
+
+from iqpilot.cereal import custom
+from iqpilot.selfdrive.iqmodeld.models import helpers as model_helpers
+from iqpilot.selfdrive.iqmodeld.models.runners.tinygrad import supercombo_runner as supercombo_runner_mod
+from iqpilot.selfdrive.iqmodeld.models.runners.tinygrad.supercombo_runner import (
+  TinygradSupercomboRunner,
+)
+
+
+class _Captured:
+  def __init__(self, expected_names):
+    self.expected_names = expected_names
+
+
+class _FakeJit:
+  def __init__(self, expected_names):
+    self.captured = _Captured(expected_names)
+
+
+class _Boom:
+  def __init__(self, err: Exception):
+    self.err = err
+
+  def __call__(self, *args, **kwargs):
+    raise self.err
+
+
+class _FakeParams:
+  def __init__(self, active_bundle=None):
+    self.store = {}
+    if active_bundle is not None:
+      self.store["ModelManager_ActiveBundle"] = active_bundle
+
+  def get(self, key):
+    return self.store.get(key)
+
+  def put(self, key, value):
+    self.store[key] = value
+
+  def remove(self, key):
+    self.store.pop(key, None)
+
+
+def test_verify_artifact_file_deletes_stale_cached_pkl(tmp_path: Path):
+  pkl_path = tmp_path / "driving_supercombo_guard.pkl"
+  pkl_path.write_bytes(b"stale-pkl")
+
+  runner = TinygradSupercomboRunner.__new__(TinygradSupercomboRunner)
+  runner._pkl_path = str(pkl_path)
+  runner._expected_sha256 = hashlib.sha256(b"fresh-pkl").hexdigest()
+
+  with pytest.raises(RuntimeError, match="SHA mismatch"):
+    runner._verify_artifact_file()
+
+  assert not pkl_path.exists()
+
+
+def test_validate_jit_names_accepts_current_runtime_contract():
+  runner = TinygradSupercomboRunner.__new__(TinygradSupercomboRunner)
+  runner._pkl_path = "/tmp/does-not-matter.pkl"
+  runner._expected_sha256 = ""
+  runner._run_policy = _FakeJit(['warped', 'img_q', 'big_img_q', 'feat_q', 'desire_q', 'packed_npy_inputs'])
+  runner._warp_jits = {
+    (1344, 760): _FakeJit(['tfm', 'big_tfm', 'frame', 'big_frame']),
+  }
+
+  runner._validate_jit_names()
+
+
+def test_validate_jit_names_raises_clear_error_for_contract_mismatch():
+  runner = TinygradSupercomboRunner.__new__(TinygradSupercomboRunner)
+  runner._pkl_path = "/tmp/does-not-matter.pkl"
+  runner._expected_sha256 = ""
+  runner._run_policy = _FakeJit(['img', 'big_img', 'feat_q', 'desire_q', 'desire', 'traffic_convention', 'action_t'])
+  runner._warp_jits = {
+    (1344, 760): _FakeJit(['img_q', 'big_img_q', 'tfm', 'big_tfm', 'frame', 'big_frame']),
+  }
+
+  with pytest.raises(RuntimeError, match="JIT argument mismatch"):
+    runner._validate_jit_names()
+
+
+def test_handle_runtime_jit_mismatch_deletes_stale_cached_pkl(tmp_path: Path):
+  pkl_path = tmp_path / "driving_supercombo_guard.pkl"
+  pkl_path.write_bytes(b"stale-pkl")
+
+  runner = TinygradSupercomboRunner.__new__(TinygradSupercomboRunner)
+  runner._pkl_path = str(pkl_path)
+  runner._expected_sha256 = hashlib.sha256(b"fresh-pkl").hexdigest()
+
+  with pytest.raises(RuntimeError, match="runtime JIT mismatch with stale cached SHA"):
+    runner._handle_runtime_jit_mismatch(RuntimeError("args mismatch in JIT: stale bundle"))
+
+  assert not pkl_path.exists()
+
+
+def test_handle_runtime_jit_mismatch_raises_clear_error_without_sha_mismatch(tmp_path: Path):
+  pkl_path = tmp_path / "driving_supercombo_guard.pkl"
+  pkl_path.write_bytes(b"fresh-pkl")
+
+  runner = TinygradSupercomboRunner.__new__(TinygradSupercomboRunner)
+  runner._pkl_path = str(pkl_path)
+  runner._expected_sha256 = hashlib.sha256(b"fresh-pkl").hexdigest()
+
+  with pytest.raises(RuntimeError, match="runtime JIT mismatch"):
+    runner._handle_runtime_jit_mismatch(RuntimeError("args mismatch in JIT: wrong contract"))
+
+
+def test_schedule_active_bundle_redownload_sets_download_index(monkeypatch: pytest.MonkeyPatch):
+  params = _FakeParams({"index": 81})
+  monkeypatch.setattr(supercombo_runner_mod, "Params", lambda: params)
+
+  runner = TinygradSupercomboRunner.__new__(TinygradSupercomboRunner)
+  msg = runner._schedule_active_bundle_redownload()
+
+  assert params.get("ModelManager_DownloadIndex") == "81"
+  assert msg == "; scheduled automatic re-download of the active model"
+
+
+def test_no_active_bundle_seeds_default_tinygrad(monkeypatch: pytest.MonkeyPatch):
+  monkeypatch.setattr(model_helpers, "ensure_default_model_files", lambda *a, **k: None)
+  params = _FakeParams()
+
+  runner = model_helpers.get_active_model_runner(params)
+
+  assert runner == custom.IQModelManager.Runner.tinygrad
+  active = params.get("ModelManager_ActiveBundle")
+  assert active is not None and active.get("ref") == "default"
+
+
+def test_select_default_model_clears_custom_download_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+  pending_restore = tmp_path / "pending_model_restore"
+  pending_restore.write_text("Pop")
+  monkeypatch.setattr(model_helpers, "_PENDING_MODEL_RESTORE_FILE", str(pending_restore))
+  monkeypatch.setattr(model_helpers, "ensure_default_model_files", lambda *a, **k: None)
+
+  params = _FakeParams({"index": 81, "ref": "pop"})
+  params.put("ModelManager_DownloadIndex", "81")
+  params.put("ModelRunnerTypeCache", int(custom.IQModelManager.Runner.tinygrad))
+
+  model_helpers.select_default_model(params)
+
+  assert params.get("ModelManager_DownloadIndex") is None
+  active = params.get("ModelManager_ActiveBundle")
+  assert active is not None and active.get("ref") == "default"
+  assert int(params.get("ModelRunnerTypeCache")) == int(custom.IQModelManager.Runner.tinygrad)
+  assert not pending_restore.exists()
+
+
+def test_default_model_is_not_resolved_to_manifest_pop_bundle():
+  pop_bundle = type("Bundle", (), {"internalName": "Pop (Default)", "displayName": "Pop (Default)"})()
+
+  assert model_helpers.get_default_model_bundle([pop_bundle]) is None
+
+
+def test_verify_artifact_file_schedules_redownload_for_stale_cached_pkl(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+  params = _FakeParams({"index": 81})
+  monkeypatch.setattr(supercombo_runner_mod, "Params", lambda: params)
+
+  pkl_path = tmp_path / "driving_supercombo_guard.pkl"
+  pkl_path.write_bytes(b"stale-pkl")
+
+  runner = TinygradSupercomboRunner.__new__(TinygradSupercomboRunner)
+  runner._pkl_path = str(pkl_path)
+  runner._expected_sha256 = hashlib.sha256(b"fresh-pkl").hexdigest()
+
+  with pytest.raises(RuntimeError, match="scheduled automatic re-download"):
+    runner._verify_artifact_file()
+
+  assert params.get("ModelManager_DownloadIndex") == "81"
+  assert not pkl_path.exists()
+
+
+def test_run_fused_converts_raw_warp_jit_mismatch_to_runtime_error(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+  pkl_path = tmp_path / "driving_supercombo_guard.pkl"
+  pkl_path.write_bytes(b"fresh-pkl")
+
+  runner = TinygradSupercomboRunner.__new__(TinygradSupercomboRunner)
+  runner._pkl_path = str(pkl_path)
+  runner._expected_sha256 = hashlib.sha256(b"fresh-pkl").hexdigest()
+  runner._frame_skip = 4
+  runner._cam = (1344, 760)
+  runner._queues = {
+    "tfm": object(),
+    "big_tfm": object(),
+    "img_q": object(),
+    "big_img_q": object(),
+    "feat_q": object(),
+    "desire_q": object(),
+    "packed_npy_inputs": object(),
+  }
+  runner._npy = {
+    "tfm": [0.0],
+    "big_tfm": [0.0],
+    "desire": [0.0],
+    "prev_feat": [0.0],
+  }
+  runner._prev_desire = [0.0]
+  runner._warp_jits = {
+    (1344, 760): _Boom(RuntimeError("args mismatch in JIT: self.captured.expected_names=['big_frame'] != ['frame']")),
+  }
+  runner._run_policy = _FakeJit(["warped", "img_q", "big_img_q", "feat_q", "desire_q", "packed_npy_inputs"])
+  runner._hidden_slice = slice(0, 1)
+  runner._slices = {"out": slice(0, 1)}
+  runner._parser = type("P", (), {"parse_vision_outputs": staticmethod(lambda sliced: sliced)})()
+  runner._frame_tensor = lambda *args, **kwargs: object()
+
+  monkeypatch.setattr(TinygradSupercomboRunner, "_ensure_queues", lambda self, cam_w, cam_h: None)
+
+  class _Buf:
+    width = 1344
+    height = 760
+    data = memoryview(b"\x00")
+
+  with pytest.raises(RuntimeError, match="runtime JIT mismatch"):
+    runner.run_fused(
+      {"img": _Buf(), "big_img": _Buf()},
+      {"img": [0.0], "big_img": [0.0]},
+      {},
+    )

iqpilot/selfdrive/iqmodeld/tests/test_temporal_replay.py

@@ -0,0 +1,152 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from types import SimpleNamespace
+
+import numpy as np
+import pytest
+
+import iqpilot.selfdrive.iqmodeld.models.helpers as bundle_helpers
+import iqpilot.selfdrive.iqmodeld.models.runners.model_runner as runner_helpers
+import iqpilot.selfdrive.iqmodeld.daemon as iqmodeld_daemon
+
+
+@dataclass
+class StubOverride:
+  key: str
+  value: str
+
+
+class StubBundle:
+  def __init__(self, generation: int = 10):
+    self.overrides = [StubOverride("lat", ".1"), StubOverride("long", ".3")]
+    self.generation = generation
+
+
+class StubRunner:
+  def __init__(self, input_shapes: dict[str, tuple[int, ...]]) -> None:
+    self.input_shapes = input_shapes
+    self.constants = SimpleNamespace(
+      FULL_HISTORY_BUFFER_LEN=100,
+      FEATURE_LEN=512,
+      DESIRE_LEN=8,
+      PREV_DESIRED_CURV_LEN=1,
+      INPUT_HISTORY_BUFFER_LEN=25,
+      TEMPORAL_SKIP=4,
+    )
+    self.vision_input_names: list[str] = []
+    self.is_20hz = input_shapes.get(next(iter(input_shapes)), (1, 0, 0))[1] == 25
+
+  def prepare_inputs(self, imgs_cl, numpy_inputs, frames):
+    return None
+
+  def run_model(self):
+    return {
+      "hidden_state": np.zeros((1, self.constants.FEATURE_LEN), dtype=np.float32),
+      "desired_curvature": np.zeros((1, 1), dtype=np.float32),
+    }
+
+
+def _install_runtime(monkeypatch: pytest.MonkeyPatch, shapes: dict[str, tuple[int, ...]], generation: int = 10):
+  bundle = StubBundle(generation=generation)
+  runner = StubRunner(shapes)
+  monkeypatch.setattr(bundle_helpers, "get_active_bundle", lambda params=None: bundle, raising=False)
+  monkeypatch.setattr(runner_helpers, "get_model_runner", lambda: runner, raising=False)
+  monkeypatch.setattr(iqmodeld_daemon, "get_active_bundle", lambda params=None: bundle, raising=False)
+  monkeypatch.setattr(iqmodeld_daemon, "get_model_runner", lambda: runner, raising=False)
+  return iqmodeld_daemon.NeuralEngineState(None), runner
+
+
+def _expected_selector_indices(shape: tuple[int, ...], mode: str) -> np.ndarray | None:
+  if mode == "split":
+    full = 100
+    return np.arange(full)[-1 - (4 * (25 - 1))::4]
+  if mode == "20hz":
+    step = int(-100 / shape[1])
+    return np.arange(step, step * (shape[1] + 1), step)[::-1]
+  if mode == "dense":
+    return np.arange(shape[1])
+  return None
+
+
+@pytest.mark.parametrize(
+  ("shapes", "mode"),
+  [
+    ({"desire": (1, 100, 8), "features_buffer": (1, 99, 512), "prev_desired_curv": (1, 100, 1)}, "dense"),
+    ({"desire": (1, 25, 8), "features_buffer": (1, 24, 512)}, "20hz"),
+    ({"desire_pulse": (1, 25, 8), "features_buffer": (1, 25, 512)}, "split"),
+  ],
+)
+def test_replay_ledger_layout_matches_expected_history(monkeypatch: pytest.MonkeyPatch,
+                                                       shapes: dict[str, tuple[int, ...]],
+                                                       mode: str):
+  state, _runner = _install_runtime(monkeypatch, shapes)
+
+  for tensor_name, tensor_shape in shapes.items():
+    history = state.temporal_buffers.get(tensor_name)
+    selector = state.temporal_idxs_map.get(tensor_name)
+    if history is None:
+      continue
+
+    if mode == "dense":
+      expected_shape = (1, tensor_shape[1], tensor_shape[2])
+    else:
+      expected_shape = (1, 100, tensor_shape[2])
+
+    assert history.shape == expected_shape
+    expected_selector = _expected_selector_indices(tensor_shape, mode)
+    if expected_selector is None:
+      assert selector is None or selector.size == 0
+    else:
+      assert np.array_equal(selector, expected_selector)
+
+
+def test_replay_ledger_rising_edge_and_hidden_state_updates(monkeypatch: pytest.MonkeyPatch):
+  state, runner = _install_runtime(monkeypatch, {
+    "desire": (1, 100, 8),
+    "features_buffer": (1, 99, 512),
+    "prev_desired_curv": (1, 100, 1),
+  })
+
+  pulse = np.zeros(8, dtype=np.float32)
+  pulse[3] = 1.0
+  state.run({}, {}, {"desire": pulse})
+  first_export = state.numpy_inputs["desire"].copy()
+  assert np.count_nonzero(first_export) == 1
+
+  state.run({}, {}, {"desire": pulse})
+  second_export = state.numpy_inputs["desire"].copy()
+  assert np.count_nonzero(second_export) == 1
+  assert second_export[0, -1, 3] == 0.0
+
+  hidden_value = np.arange(runner.constants.FEATURE_LEN, dtype=np.float32)
+
+  def hidden_state_run():
+    return {
+      "hidden_state": hidden_value.reshape(1, -1),
+      "desired_curvature": np.array([[0.25]], dtype=np.float32),
+    }
+
+  state.model_runner.run_model = hidden_state_run
+  state.run({}, {}, {"desire": np.zeros(8, dtype=np.float32)})
+
+  np.testing.assert_allclose(state.numpy_inputs["features_buffer"][0, -1], hidden_value, rtol=0, atol=0)
+  assert state.numpy_inputs["prev_desired_curv"][0, -1, 0] == pytest.approx(0.25)
+
+
+def test_replay_ledger_zeroes_feedback_for_mlsim_generation(monkeypatch: pytest.MonkeyPatch):
+  state, _runner = _install_runtime(monkeypatch, {
+    "desire": (1, 100, 8),
+    "features_buffer": (1, 99, 512),
+    "prev_desired_curv": (1, 100, 1),
+  }, generation=11)
+
+  def ml_run():
+    return {
+      "hidden_state": np.zeros((1, 512), dtype=np.float32),
+      "desired_curvature": np.array([[1.5]], dtype=np.float32),
+    }
+
+  state.model_runner.run_model = ml_run
+  state.run({}, {}, {"desire": np.zeros(8, dtype=np.float32)})
+  assert np.count_nonzero(state.numpy_inputs["prev_desired_curv"]) == 0

iqpilot/selfdrive/iqmodeld/tests/tf_test/build.sh

@@ -0,0 +1,17 @@
+#!/usr/bin/env bash
+# Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
+
+set -euo pipefail
+
+TF_ROOT="${TF_ROOT:-/home/batman/one/external/tensorflow}"
+TF_INCLUDE_DIR="${TF_INCLUDE_DIR:-$TF_ROOT/include}"
+TF_LIB_DIR="${TF_LIB_DIR:-$TF_ROOT/lib}"
+CXX="${CXX:-clang++}"
+
+exec "$CXX" \
+  -std=c++17 \
+  -I "$TF_INCLUDE_DIR" \
+  -L "$TF_LIB_DIR" \
+  -Wl,-rpath="$TF_LIB_DIR" \
+  main.cc \
+  -ltensorflow

iqpilot/selfdrive/iqmodeld/tests/tf_test/main.cc

@@ -0,0 +1,92 @@
+// Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
+
+#include <cassert>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <filesystem>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "tensorflow/c/c_api.h"
+
+namespace {
+
+struct FileBlob {
+  std::vector<uint8_t> bytes;
+};
+
+FileBlob read_blob(const std::filesystem::path &path) {
+  FILE *handle = fopen(path.c_str(), "rb");
+  if (handle == nullptr) {
+    return {};
+  }
+
+  fseek(handle, 0, SEEK_END);
+  const long byte_count = ftell(handle);
+  rewind(handle);
+
+  FileBlob blob;
+  blob.bytes.resize(byte_count);
+  const size_t read_count = fread(blob.bytes.data(), static_cast<size_t>(byte_count), 1, handle);
+  fclose(handle);
+
+  if (read_count != 1) {
+    blob.bytes.clear();
+  }
+  return blob;
+}
+
+void free_tf_buffer(void *data, size_t) {
+  free(data);
+}
+
+TF_Buffer *make_tf_buffer(FileBlob &&blob) {
+  auto *buffer = TF_NewBuffer();
+  auto *payload = static_cast<uint8_t *>(malloc(blob.bytes.size()));
+  assert(payload != nullptr);
+  memcpy(payload, blob.bytes.data(), blob.bytes.size());
+  buffer->data = payload;
+  buffer->length = blob.bytes.size();
+  buffer->data_deallocator = free_tf_buffer;
+  return buffer;
+}
+
+std::string pb_path_from_prefix(const char *prefix) {
+  return std::string(prefix) + ".pb";
+}
+
+}  // namespace
+
+int main(int argc, char *argv[]) {
+  if (argc < 2) {
+    printf("usage: %s <graph-prefix>\n", argv[0]);
+    return 1;
+  }
+
+  const std::string pb_path = pb_path_from_prefix(argv[1]);
+  printf("loading model %s\n", pb_path.c_str());
+
+  FileBlob blob = read_blob(pb_path);
+  if (blob.bytes.empty()) {
+    printf("FAIL: unable to read graph bytes\n");
+    return 1;
+  }
+  printf("loaded model of size %zu\n", blob.bytes.size());
+
+  std::unique_ptr<TF_Status, decltype(&TF_DeleteStatus)> status(TF_NewStatus(), TF_DeleteStatus);
+  std::unique_ptr<TF_Graph, decltype(&TF_DeleteGraph)> graph(TF_NewGraph(), TF_DeleteGraph);
+  std::unique_ptr<TF_ImportGraphDefOptions, decltype(&TF_DeleteImportGraphDefOptions)> options(
+    TF_NewImportGraphDefOptions(), TF_DeleteImportGraphDefOptions);
+  std::unique_ptr<TF_Buffer, decltype(&TF_DeleteBuffer)> buffer(make_tf_buffer(std::move(blob)), TF_DeleteBuffer);
+
+  TF_GraphImportGraphDef(graph.get(), buffer.get(), options.get(), status.get());
+  if (TF_GetCode(status.get()) != TF_OK) {
+    printf("FAIL: %s\n", TF_Message(status.get()));
+    return 1;
+  }
+
+  printf("SUCCESS\n");
+  return 0;
+}

iqpilot/selfdrive/iqmodeld/tests/tf_test/pb_loader.py

@@ -0,0 +1,32 @@
+"""
+Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import tensorflow as tf
+
+
+def _load_graph_bytes(graph_path: Path) -> bytes:
+  return graph_path.read_bytes()
+
+
+def _parse_graph(graph_path: Path) -> tf.compat.v1.GraphDef:
+  graph = tf.compat.v1.GraphDef()
+  graph.ParseFromString(_load_graph_bytes(graph_path))
+  return graph
+
+
+def main(argv: list[str]) -> int:
+  if len(argv) < 2:
+    print("Usage: pb_loader.py <graph.pb>")
+    return 1
+  _parse_graph(Path(argv[1]))
+  return 0
+
+
+if __name__ == "__main__":
+  raise SystemExit(main(sys.argv))

iqpilot/selfdrive/iqmodeld/tests/timing/benchmark.py

@@ -0,0 +1,54 @@
+"""
+Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
+"""
+
+from __future__ import annotations
+
+import os
+import time
+
+import numpy as np
+
+import iqpilot.cereal.messaging as messaging
+from iqpilot.system.manager.process_config import managed_processes
+
+RUN_COUNT = int(os.getenv("N", "5"))
+WINDOW_SECONDS = int(os.getenv("TIME", "30"))
+WARMUP_MESSAGES = 10
+
+
+def _collect_execution_samples(sock, duration_s: int) -> np.ndarray:
+  samples: list[float] = []
+  deadline = time.monotonic() + duration_s
+  while time.monotonic() < deadline:
+    for message in messaging.drain_sock(sock, wait_for_one=True):
+      samples.append(message.modelV2.modelExecutionTime)
+  return np.array(samples[WARMUP_MESSAGES:]) * 1000.0
+
+
+def _single_benchmark_pass(sock) -> np.ndarray:
+  os.environ["LOGPRINT"] = "debug"
+  managed_processes["modeld"].start()
+  time.sleep(5)
+  try:
+    return _collect_execution_samples(sock, WINDOW_SECONDS)
+  finally:
+    managed_processes["modeld"].stop()
+
+
+def _report_run(index: int, values_ms: np.ndarray) -> None:
+  print(
+    f"run {index}: avg={values_ms.mean():0.2f}ms "
+    f"min={values_ms.min():0.2f}ms max={values_ms.max():0.2f}ms"
+  )
+
+
+if __name__ == "__main__":
+  subscriber = messaging.sub_sock("modelV2", conflate=False, timeout=1000)
+  all_runs = [_single_benchmark_pass(subscriber) for _ in range(RUN_COUNT)]
+
+  print("\n")
+  print(f"ran modeld {RUN_COUNT} times for {WINDOW_SECONDS}s each")
+  for index, values_ms in enumerate(all_runs, start=1):
+    _report_run(index, values_ms)
+  print("\n")