IQ.Lvbs/IQ.Pilot
8
7
Fork 3
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

IQ.Pilot Release Commit @ bec7652

Browse Source
This commit is contained in:
IQ.Lvbs CI [bot]
2026-08-22 21:29:55 -05:00
parent 11cefcb266
commit 7786d94c33
58 changed files with 578 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
iqpilot/selfdrive/car/gap_button_actions.py
View File

@@ -1,9 +1,5 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
Maps the distance/gap steering-wheel button to an IQ.Pilot action: holding it for
long enough toggles Experimental mode exactly once per hold. Only active when
IQ.Pilot owns longitudinal control and cruise is available.
"""
from iqpilot.cereal import car, custom
from iqdbc.car import structs

14
iqpilot/selfdrive/car/vehicle_catalog.json
View File

@@ -114,6 +114,20 @@
],
"req": "Adaptive Cruise Control (ACC) & Lane Assist"
},
"AUDI_A4_MK4|2013-2016": {
"label": "Audi A4 2013-16",
"id": "AUDI_A4_MK4",
"mk": "Audi",
"grp": "volkswagen",
"mdl": "A4",
"yrs": [
"2013",
"2014",
"2015",
"2016"
],
"req": "Cruise Control"
},
"AUDI_Q2_MK1|2018": {
"label": "Audi Q2 2018",
"id": "AUDI_Q2_MK1",

14
iqpilot/selfdrive/iqmodeld/daemon.py
View File

@@ -63,6 +63,9 @@ IQP_NAV_MODEL_INFLUENCE_ENABLED = False
TurnDirection = custom.IQTurnSignalDirection
IQMODEL_EVAL_WARN_US = int(DT_MDL * 1_000_000)
IQMODEL_EVAL_ERROR_US = IQMODEL_EVAL_WARN_US * 2
_FRAME_STARVED_BACKOFF_POLLS = 5
_FRAME_STARVED_BACKOFF_SECONDS = 0.005
_FRAME_STARVED_LOG_EVERY = 200
def _plan_y_std_1s(outputs: dict[str, np.ndarray]) -> float:
@@ -605,12 +608,21 @@ class InferenceDaemon:
def serve(self) -> None:
tick = 0
starved_polls = 0
while True:
frame_pair = self._cameras.pull()
if frame_pair is None:
cloudlog.debug("visionipc frame missing")
starved_polls += 1
if starved_polls >= _FRAME_STARVED_BACKOFF_POLLS:
time.sleep(_FRAME_STARVED_BACKOFF_SECONDS)
if starved_polls % _FRAME_STARVED_LOG_EVERY == 0:
cloudlog.error(f"visionipc delivered no frames for {starved_polls} polls; model is not running")
continue
if starved_polls:
cloudlog.warning(f"visionipc recovered after {starved_polls} frameless polls")
starved_polls = 0
main_buf, extra_buf, main_stamp, extra_stamp = frame_pair
self._sub.update(0)
self._refresh_tunables(tick)

2
iqpilot/selfdrive/iqmodeld/models/__init__.py
View File

@@ -1,3 +1,3 @@
"""
IQ model selection and runner support that is actively used by iqmodeld.
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""

7
iqpilot/selfdrive/iqmodeld/models/fetcher.py
View File

@@ -1,11 +1,8 @@
#!/usr/bin/env python3
"""
Copyright (c) IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
Public entry point for the model-manifest fetcher: prefers the compiled private
bundle, falling back to the in-tree source. The default-runner fallback lives in
ManifestDecoder now, so no post-import patching is needed.
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
from iqpilot._proprietary_loader import ProprietaryModuleMissing, load_private_module
try:

6
iqpilot/selfdrive/iqmodeld/models/helpers.py
View File

@@ -40,7 +40,6 @@ _DEFAULT_BUNDLE_REF = "default"
def get_default_model_bundle(_bundles):
"""Legacy compatibility hook: stock default is preinstalled, not a manifest bundle."""
return None
@@ -239,10 +238,13 @@ def select_default_model(params: Params = None) -> None:
def seed_default_bundle_if_unset(params: Params = None) -> None:
params = Params() if params is None else params
if params.get(_ACTIVE_BUNDLE_KEY) or params.get(_DOWNLOAD_INDEX_KEY) is not None:
if params.get(_ACTIVE_BUNDLE_KEY):
return
queued_download = params.get(_DOWNLOAD_INDEX_KEY)
try:
select_default_model(params)
if queued_download is not None:
params.put(_DOWNLOAD_INDEX_KEY, queued_download)
cloudlog.warning("default_model: seeded Default (CD210) as active bundle")
except Exception as e:
cloudlog.exception(f"default_model: failed to seed default bundle: {e}")

4
iqpilot/selfdrive/iqmodeld/models/inference_state.py
View File

@@ -1,9 +1,5 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
Common base for the per-process inference/runtime states. It seeds the lateral
steer delay from the cached learned value so every subclass starts with a usable
number before its first lateralDelay message arrives.
"""
from iqpilot.common.steer_delay import cached_steer_delay

2
iqpilot/selfdrive/iqmodeld/models/runners/__init__.py
View File

@@ -1,3 +1,3 @@
"""
Runner interfaces used by iqmodeld model execution.
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""

4
iqpilot/selfdrive/iqmodeld/models/runners/tinygrad/__init__.py
View File

@@ -1,3 +1,3 @@
"""
Tinygrad runner support for iqmodeld.
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""

7
iqpilot/selfdrive/iqmodeld/models/runners/tinygrad/fused_runner.py
View File

@@ -27,8 +27,6 @@ WARP_DEV = os.getenv('WARP_DEV')
class TinygradFusedRunner(ModelRunner):
"""Runs a fused warp+vision+policy pkl. Bundle ships one `driving_fused_*` artifact."""
uses_opencl_warp: bool = False
def __init__(self):
@@ -110,7 +108,6 @@ class TinygradFusedRunner(ModelRunner):
'feat_q': zeros_f32((self._frame_skip * (fb[1] - 1) + 1, fb[0], fb[2])),
'desire_q': zeros_f32((self._frame_skip * dp[1], dp[0], dp[2])),
}
# shapes must match the captured run_policy JIT inputs
on_shapes = self._on_meta['input_shapes']
captured = self._run_policy.captured
jit_shapes = {
@@ -135,7 +132,6 @@ class TinygradFusedRunner(ModelRunner):
self._cam_resolution = (cam_w, cam_h)
def run_fused(self, bufs: dict, transforms: dict[str, np.ndarray], numpy_inputs: NumpyDict) -> NumpyDict:
"""warp + vision + policy in one pass from raw NV12 bufs + transform matrices."""
Tensor, Device = _tinygrad_imports()
main_buf = bufs['img']
@@ -154,7 +150,6 @@ class TinygradFusedRunner(ModelRunner):
npy = lambda key: Tensor(self._npy_buffers[key], device='NPY')
# frames go on the compute device to match the captured warp JIT
frame = self._frame_tensor('img', bufs['img'])
big_frame = self._frame_tensor('big_img', bufs['big_img'])
@@ -169,8 +164,6 @@ class TinygradFusedRunner(ModelRunner):
policy_inputs['action_t'] = npy('action_t')
vision_out_t, on_out_t, off_out_t = self._run_policy(**policy_inputs)
# parse each model's output on its own sliced dict; parsing a merged dict
# would run parse_dynamic_outputs twice and double-parse plan/lead
def _slice(tensor_out, meta) -> NumpyDict:
flat = tensor_out.numpy().flatten()
return {k: flat[np.newaxis, sl] for k, sl in meta['output_slices'].items() if k != 'pad'}

7
iqpilot/selfdrive/iqmodeld/models/runners/tinygrad/supercombo_runner.py
View File

@@ -68,8 +68,6 @@ def _is_jit_arg_mismatch(err: BaseException) -> bool:
class TinygradSupercomboRunner(ModelRunner):
"""Runs a single combined supercombo pkl. Bundle ships one `driving_supercombo_*` artifact."""
uses_opencl_warp: bool = False
def __init__(self):
@@ -282,7 +280,6 @@ class TinygradSupercomboRunner(ModelRunner):
zeros_u8 = lambda s: Tensor(np.zeros(s, dtype=np.uint8), device=Device.DEFAULT).contiguous().realize()
zeros_f32 = lambda s: Tensor(np.zeros(s, dtype=np.float32), device=Device.DEFAULT).contiguous().realize()
# packed npy block (single NPY tensor, mutated in place via views): order matches run_policy.split
shapes = {'desire': (dp[2],), 'traffic_convention': tuple(tc), 'action_t': tuple(at), 'prev_feat': (fb[0], fb[2])}
sizes = [math.prod(s) for s in shapes.values()]
packed = np.zeros(sum(sizes), dtype=np.float32)
@@ -318,7 +315,6 @@ class TinygradSupercomboRunner(ModelRunner):
self._npy['traffic_convention'][:] = numpy_inputs['traffic_convention']
if 'action_t' in numpy_inputs:
self._npy['action_t'][:] = numpy_inputs['action_t']
# self._npy['prev_feat'] holds last frame's hidden_state (zeros on the first frame)
frame = self._frame_tensor('img', bufs['img'])
big_frame = self._frame_tensor('big_img', bufs['big_img'])
@@ -334,11 +330,10 @@ class TinygradSupercomboRunner(ModelRunner):
raise
flat = out.numpy().flatten()
# feed hidden_state back as prev_feat for the next frame
self._npy['prev_feat'][:] = flat[self._hidden_slice].reshape(self._npy['prev_feat'].shape)
sliced = {k: flat[np.newaxis, sl] for k, sl in self._slices.items()}
return self._parser.parse_vision_outputs(sliced) # single-pass; parse_outputs double-parses a combined dict
return self._parser.parse_vision_outputs(sliced)
def _run_model(self) -> NumpyDict:
raise RuntimeError("supercombo path goes through run_fused(), not _run_model()")

2
iqpilot/selfdrive/iqmodeld/models/runners/tinygrad/tinygrad_runner.py
View File

@@ -85,8 +85,6 @@ class TinygradRunner(ModelRunner, SupercomboTinygrad, PolicyTinygrad, VisionTiny
self.model_run = _load_program_blob(asset_name)
self._input_plan = _compile_input_plan(self.model_run.captured)
# the warp pipeline hands the runner raw uint8 YUV; a float image interface
# would silently reinterpret those bytes and drive on garbage vision
for name, spec in self._input_plan.items():
if "img" in name and spec.dtype is not dtypes.uint8:
raise ValueError(f"{asset_name}: image input {name} expects {spec.dtype}, incompatible with uint8 warp buffer")

7
iqpilot/selfdrive/iqmodeld/models/split_model_constants.py
View File

@@ -1,4 +1,3 @@
# openpilot model I/O constants (comma.ai, MIT — see LICENSE)
import numpy as np
@@ -7,7 +6,6 @@ def index_function(idx, max_val=192, max_idx=32):
class SplitModelConstants:
# time and distance indices
IDX_N = 33
T_IDXS = [index_function(idx, max_val=10.0) for idx in range(IDX_N)]
X_IDXS = [index_function(idx, max_val=192.0) for idx in range(IDX_N)]
@@ -15,7 +13,6 @@ class SplitModelConstants:
LEAD_T_OFFSETS = [0., 2., 4.]
META_T_IDXS = [2., 4., 6., 8., 10.]
# split-model temporal / history run parameters
MODEL_FREQ = 20
HISTORY_FREQ = 5
HISTORY_LEN_SECONDS = 5
@@ -31,7 +28,6 @@ class SplitModelConstants:
LATERAL_CONTROL_PARAMS_LEN = 2
PREV_DESIRED_CURV_LEN = 1
# model outputs constants
FCW_THRESHOLDS_5MS2 = np.array([.05, .05, .15, .15, .15], dtype=np.float32)
FCW_THRESHOLDS_3MS2 = np.array([.7, .7], dtype=np.float32)
FCW_5MS2_PROBS_WIDTH = 5
@@ -71,7 +67,6 @@ class SplitModelConstants:
POLY_PATH_DEGREE = 4
# model outputs slices
class Plan:
POSITION = slice(0, 3)
VELOCITY = slice(3, 6)
@@ -82,14 +77,12 @@ class Plan:
class Meta:
ENGAGED = slice(0, 1)
# next 2, 4, 6, 8, 10 seconds
GAS_DISENGAGE = slice(1, 31, 6)
BRAKE_DISENGAGE = slice(2, 31, 6)
STEER_OVERRIDE = slice(3, 31, 6)
HARD_BRAKE_3 = slice(4, 31, 6)
HARD_BRAKE_4 = slice(5, 31, 6)
HARD_BRAKE_5 = slice(6, 31, 6)
# next 0, 2, 4, 6, 8, 10 seconds
GAS_PRESS = slice(31, 55, 4)
BRAKE_PRESS = slice(32, 55, 4)
LEFT_BLINKER = slice(33, 55, 4)

3
iqpilot/selfdrive/iqmodeld/tests/__init__.py
View File

@@ -0,0 +1,3 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""

1
iqpilot/selfdrive/iqmodeld/tests/dmon_lag/repro.cc
View File

@@ -1,5 +1,4 @@
// 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>

3
iqpilot/selfdrive/iqmodeld/tests/test_action_dispatch.py
View File

@@ -1,3 +1,6 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
from __future__ import annotations
from types import SimpleNamespace

3
iqpilot/selfdrive/iqmodeld/tests/test_combined_split_runner.py
View File

@@ -1,3 +1,6 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
from __future__ import annotations
from dataclasses import dataclass

3
iqpilot/selfdrive/iqmodeld/tests/test_compile_supercombo_guards.py
View File

@@ -1,3 +1,6 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
from __future__ import annotations
import numpy as np

3
iqpilot/selfdrive/iqmodeld/tests/test_iqmodeld_contracts.py
View File

@@ -1,3 +1,6 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
from __future__ import annotations
import copy

3
iqpilot/selfdrive/iqmodeld/tests/test_model_runner_smoke.py
View File

@@ -1,3 +1,6 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
from __future__ import annotations
from dataclasses import dataclass

3
iqpilot/selfdrive/iqmodeld/tests/test_public_surface.py
View File

@@ -1,3 +1,6 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
from iqpilot.selfdrive.iqmodeld import metadata, messaging, parser
from iqpilot.selfdrive.iqmodeld.daemon import CaptureStamp, NeuralEngineState

16
iqpilot/selfdrive/iqmodeld/tests/test_selector_share_smoke.py
View File

@@ -1,3 +1,6 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
from __future__ import annotations
import os
@@ -5,7 +8,6 @@ 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
@@ -123,8 +125,9 @@ def _run_onnx_bundle(bundle_dir: Path):
)
@pytest.mark.skipif(not SHARE_ROOT.is_dir(), reason="selector model share is not mounted")
def test_three_selector_models_parse_via_share_onnx():
if not SHARE_ROOT.is_dir():
return
selector_dirs = _find_selector_dirs(limit=3, require_onnx=True)
assert len(selector_dirs) >= 3
@@ -134,8 +137,9 @@ def test_three_selector_models_parse_via_share_onnx():
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):
if not SHARE_ROOT.is_dir():
return
selector_dirs = _find_selector_dirs(limit=10)
attempted = 0
executed = 0
@@ -152,6 +156,10 @@ def test_selector_tinygrad_pkls_execute_when_host_compatible(monkeypatch):
if "/dev/kgsl-3d0" in str(exc):
continue
raise
except TypeError as exc:
if "DType.__init__()" in str(exc):
continue
raise
assert "pose" in vision_outputs
assert "plan" in policy_outputs
@@ -160,4 +168,4 @@ def test_selector_tinygrad_pkls_execute_when_host_compatible(monkeypatch):
break
if executed == 0:
pytest.skip(f"share tinygrad pkls are QCOM-only on this host; inspected {attempted} bundles")
assert attempted > 0, "no selector bundles were inspected on the share"

28
iqpilot/selfdrive/iqmodeld/tests/test_supercombo_runner_guards.py
View File

@@ -1,3 +1,6 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
from __future__ import annotations
import hashlib
@@ -153,6 +156,31 @@ def test_select_default_model_clears_custom_download_state(tmp_path: Path, monke
assert not pending_restore.exists()
def test_seed_default_bundle_runs_while_a_download_is_queued(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(model_helpers, "ensure_default_model_files", lambda *a, **k: None)
params = _FakeParams()
params.put("ModelManager_DownloadIndex", "81")
model_helpers.seed_default_bundle_if_unset(params)
active = params.get("ModelManager_ActiveBundle")
assert active is not None and active.get("ref") == "default"
assert params.get("ModelManager_DownloadIndex") == "81"
def test_seed_default_bundle_leaves_an_existing_active_bundle_alone(monkeypatch: pytest.MonkeyPatch):
monkeypatch.setattr(model_helpers, "ensure_default_model_files", lambda *a, **k: None)
params = _FakeParams({"index": 81, "ref": "pop"})
params.put("ModelManager_DownloadIndex", "81")
model_helpers.seed_default_bundle_if_unset(params)
assert params.get("ModelManager_ActiveBundle").get("ref") == "pop"
assert params.get("ModelManager_DownloadIndex") == "81"
def test_default_model_is_not_resolved_to_manifest_pop_bundle():
pop_bundle = type("Bundle", (), {"internalName": "Pop (Default)", "displayName": "Pop (Default)"})()

3
iqpilot/selfdrive/iqmodeld/tests/test_temporal_replay.py
View File

@@ -1,3 +1,6 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
from __future__ import annotations
from dataclasses import dataclass

3
iqpilot/selfdrive/iqmodeld/tools/compile_model.py
View File

@@ -1,3 +1,6 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos/
"""
import os
import pickle
import re

5
iqpilot/selfdrive/iqmodeld/tools/compile_supercombo.py
View File

@@ -59,7 +59,6 @@ def warp_perspective_tinygrad(src_flat, M_inv, dst_shape, src_shape, stride_pad,
x = Tensor.arange(w_dst).reshape(1, w_dst).expand(h_dst, w_dst).reshape(-1)
y = Tensor.arange(h_dst).reshape(h_dst, 1).expand(h_dst, w_dst).reshape(-1)
# inline 3x3 matmul as elementwise to avoid reduce op (enables fusion with gather)
src_x = M_inv[0, 0] * x + M_inv[0, 1] * y + M_inv[0, 2]
src_y = M_inv[1, 0] * x + M_inv[1, 1] * y + M_inv[1, 2]
src_w = M_inv[2, 0] * x + M_inv[2, 1] * y + M_inv[2, 2]
@@ -100,9 +99,7 @@ def make_frame_prepare(nv12: NV12Frame, model_w, model_h):
stride_pad = stride - cam_w
def frame_prepare_tinygrad(input_frame, M_inv):
# UV_SCALE @ M_inv @ UV_SCALE_INV simplifies to elementwise scaling
M_inv_uv = M_inv * Tensor([[1.0, 1.0, 0.5], [1.0, 1.0, 0.5], [2.0, 2.0, 1.0]], device=WARP_DEV)
# deinterleave NV12 UV plane (UVUV... -> separate U, V)
uv = input_frame[uv_offset:uv_offset + uv_height * stride].reshape(uv_height, stride)
with Context(SPLIT_REDUCEOP=0):
y = warp_perspective_tinygrad(input_frame[:cam_h*stride],
@@ -142,7 +139,6 @@ def get_policy_npy_shapes(input_shapes):
tc = input_shapes['traffic_convention'] # (1, 2)
at = input_shapes['action_t'] # (1, 2)
fb = input_shapes['features_buffer'] # (1, 24, 512)
# TODO prev_feat shouldn't exist and be handled inside the JIT, but corrupt on QCOM for now
shapes = {'desire': (dp[2],), 'traffic_convention': tuple(tc), 'action_t': tuple(at), 'prev_feat': (fb[0], fb[2])}
return shapes, [math.prod(s) for s in shapes.values()]
@@ -155,7 +151,6 @@ def make_input_queues(input_shapes, frame_skip, device):
shapes, sizes = get_policy_npy_shapes(input_shapes)
packed_npy_inputs = np.zeros(sum(sizes), dtype=np.float32)
# views into the packed inputs, to be refilled at runtime
npy.update({k: v.reshape(s) for (k, s), v in zip(shapes.items(), np.split(packed_npy_inputs, np.cumsum(sizes[:-1])), strict=True)})
input_queues.update({
'feat_q': Tensor(np.zeros((frame_skip * fb[1], fb[0], fb[2]), dtype=np.float32), device=device).contiguous().realize(),

12
iqpilot/selfdrive/selfdrived/events.py
View File

@@ -648,8 +648,16 @@ EVENTS: dict[int, dict[str, Alert | AlertCallbackType]] = {
},
EventName.wrongGear: {
ET.SOFT_DISABLE: user_soft_disable_alert("Gear not D"),
ET.NO_ENTRY: NoEntryAlert("Gear not D"),
ET.SOFT_DISABLE: Alert(
"",
"",
AlertStatus.normal, AlertSize.none,
Priority.LOWEST, VisualAlert.none, AudibleAlert.none, 0.),
ET.NO_ENTRY: Alert(
"",
"",
AlertStatus.normal, AlertSize.none,
Priority.LOWEST, VisualAlert.none, AudibleAlert.none, 0.),
},
# This alert is thrown when the calibration angles are outside of the acceptable range.

13
iqpilot/selfdrive/selfdrived/iq_events.py
View File

@@ -269,7 +269,7 @@ _ENGAGE_EVENTS: EVENTS_IQ_TYPE = {
EventNameIQ.steeringOverrideReengageAlc: {
ET.WARNING: Alert(
"Steering Overridden By Driver",
"Re-Engage ALC",
"Double Tap SET or Cycle the Cruise Main to Re-Engage ALC",
AlertStatus.userPrompt, AlertSize.mid,
Priority.MID, VisualAlert.none, AudibleAlert.prompt, 2.0),
},
@@ -294,9 +294,9 @@ _CABIN_BLOCK_EVENTS: EVENTS_IQ_TYPE = {
AlertStatus.normal, AlertSize.none,
Priority.LOWEST, VisualAlert.none, AudibleAlert.none, 0.),
ET.NO_ENTRY: Alert(
"Not in Drive",
"IQ.Pilot Unavailable",
AlertStatus.normal, AlertSize.mid,
"",
"",
AlertStatus.normal, AlertSize.none,
Priority.LOW, VisualAlert.none, AudibleAlert.none, 0.),
},
@@ -344,11 +344,11 @@ _NOTICE_EVENTS: EVENTS_IQ_TYPE = {
},
EventNameIQ.pedalHeldNotice: {
ET.WARNING: NoEntryAlert("Pedal Held")
ET.WARNING: NoEntryAlert("Brake Pedal Held")
},
EventNameIQ.experimentalToggled: {
ET.WARNING: NormalPermanentAlert("Experimental Mode Switched", duration=1.5)
ET.WARNING: NormalPermanentAlert("Switched to IQ.Pilot End to End Control", duration=1.5)
},
EventNameIQ.e2eChime: {
@@ -365,7 +365,6 @@ _NOTICE_EVENTS: EVENTS_IQ_TYPE = {
priority=Priority.LOW),
},
# outranks the generic processNotRunning alert so the driver sees why engagement is blocked
EventNameIQ.modelUpdating: {
ET.NO_ENTRY: NoEntryAlert("Update finishes while parked with internet",
alert_text_1="Driving Model Updating",

14
iqpilot/selfdrive/selfdrived/tests/test_alerts.py
View File

@@ -4,11 +4,12 @@ import os
import random
from PIL import Image, ImageDraw, ImageFont
from iqpilot.cereal import log, car
from iqpilot.cereal import log, car, custom
from iqpilot.cereal.messaging import SubMaster
from iqpilot.common.basedir import BASEDIR
from iqpilot.common.params import Params
from iqpilot.selfdrive.selfdrived.events import Alert, EVENTS, ET
from iqpilot.selfdrive.selfdrived.iq_events import EVENTS_IQ
from iqpilot.selfdrive.selfdrived.events import invalid_lkas_setting_alert, invalid_lkas_setting_no_entry_alert
from iqpilot.selfdrive.selfdrived.alertmanager import set_offroad_alert
from iqpilot.selfdrive.test.process_replay.process_replay import CONFIGS
@@ -26,6 +27,17 @@ for event_types in EVENTS.values():
class TestAlerts:
def test_wrong_gear_alerts_are_silent_and_invisible(self):
wrong_gear_alerts = (EVENTS[log.OnroadEvent.EventName.wrongGear][ET.SOFT_DISABLE],
EVENTS[log.OnroadEvent.EventName.wrongGear][ET.NO_ENTRY],
EVENTS_IQ[custom.IQOnroadEvent.EventName.gearNotDriveSilent][ET.NO_ENTRY])
for alert in wrong_gear_alerts:
assert alert.alert_size == AlertSize.none
assert alert.audible_alert == car.CarControl.HUDControl.AudibleAlert.none
assert alert.alert_text_1 == ""
assert alert.alert_text_2 == ""
@classmethod
def setup_class(cls):
with open(OFFROAD_ALERTS_PATH) as f: