superrob/IQ.Pilot
1
0
Fork 0
forked from IQ.Lvbs/IQ.Pilot
Code Pull Requests Activity

IQ.Pilot Release Commit @ f2a861c

Browse Source
This commit is contained in:
IQ.Lvbs CI [bot]
2026-09-02 15:07:09 -05:00
parent b42569dbca
commit e8748fd704
5497 changed files with 316070 additions and 179848 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
iqpilot/sab/tests/test_sab_state_machine.py
View File

@@ -1,17 +1,13 @@
"""
Copyright © IQ.Lvbs, apart of Project Teal Lvbs, All Rights Reserved, licensed under https://konn3kt.com/tos
Table-driven checks for GuidanceStateMachine: every transition is one row of
(start state, signals present, expected state), and the side effects (queued
alert types, soft-disable timer arming) are asserted separately.
"""
import pytest
from cereal import custom
from openpilot.common.realtime import DT_CTRL
from openpilot.selfdrive.selfdrived.events import ET
from openpilot.selfdrive.selfdrived.state import SOFT_DISABLE_TIME
from openpilot.iqpilot.sab.behavior import (GuidanceStateMachine, PAUSE_WITH_IQ_EVENTS,
from iqpilot.cereal import custom
from iqpilot.common.realtime import DT_CTRL
from iqpilot.selfdrive.selfdrived.events import ET
from iqpilot.selfdrive.selfdrived.state import SOFT_DISABLE_TIME
from iqpilot.sab.behavior import (GuidanceStateMachine, PAUSE_WITH_IQ_EVENTS,
PAUSE_WITH_STOCK_EVENTS)
State = custom.AlwaysOnLateral.AlwaysOnLateralState
@@ -19,20 +15,17 @@ EventNameIQ = custom.IQOnroadEvent.EventName
SOFT_DISABLE_FRAMES = int(SOFT_DISABLE_TIME / DT_CTRL)
# signal aliases used in the table rows
ENABLE = ET.ENABLE
NO_ENTRY = ET.NO_ENTRY
SOFT = ET.SOFT_DISABLE
USER = ET.USER_DISABLE
IMMEDIATE = ET.IMMEDIATE_DISABLE
OVERRIDE = ET.OVERRIDE_LATERAL
SILENT = "silent-disable" # silentLkasDisable present in the IQ event bag
PAUSE_OK = "pause-eligible" # a gear/door/belt event from the pause lists is present
SILENT = "silent-disable"
PAUSE_OK = "pause-eligible"
class SignalBag:
"""Stands in for both event buckets; the machine only probes membership."""
def __init__(self, types=(), names=()):
self._types = set(types)
self._names = set(names)
@@ -48,8 +41,6 @@ class SignalBag:
class Host:
"""Minimal stand-in for the sab/selfdrive plumbing the machine touches."""
class _SSM:
def __init__(self):
self.current_alert_types = []
@@ -70,27 +61,24 @@ class Host:
class Sab:
def __init__(self, host):
def __init__(self, host, pause_on_lateral_override=False):
self.selfdrive = host
self.pause_on_lateral_override = pause_on_lateral_override
def machine_at(state, signals, selfdrive_enabled=False):
def machine_at(state, signals, selfdrive_enabled=False, pause_on_lateral_override=False):
host = Host(signals, selfdrive_enabled)
m = GuidanceStateMachine(Sab(host))
m = GuidanceStateMachine(Sab(host, pause_on_lateral_override))
m.state = state
return m, host
# (id, start state, signals, expected state)
TRANSITIONS = [
# from disabled
("idle stays idle", State.disabled, (), State.disabled),
("engage", State.disabled, (ENABLE,), State.enabled),
("engage while overriding", State.disabled, (ENABLE, OVERRIDE), State.overriding),
("blocked entry", State.disabled, (ENABLE, NO_ENTRY), State.disabled),
("blocked entry parks when pause-eligible", State.disabled, (ENABLE, NO_ENTRY, PAUSE_OK), State.paused),
# from enabled
("cruise steady", State.enabled, (), State.enabled),
("driver off switch", State.enabled, (USER,), State.disabled),
("driver off switch, silent -> pause", State.enabled, (USER, SILENT), State.paused),
@@ -99,13 +87,9 @@ TRANSITIONS = [
("hands on wheel", State.enabled, (OVERRIDE,), State.overriding),
("user beats soft", State.enabled, (USER, SOFT), State.disabled),
("hard beats soft", State.enabled, (IMMEDIATE, SOFT), State.disabled),
# from softDisabling (timer still armed -> stays; see timer tests for expiry)
("condition cleared", State.softDisabling, (), State.enabled),
("user during grace", State.softDisabling, (USER,), State.disabled),
("hard during grace", State.softDisabling, (IMMEDIATE,), State.disabled),
# from paused
("stays parked", State.paused, (), State.paused),
("blocked resume", State.paused, (ENABLE, NO_ENTRY), State.paused),
("resume", State.paused, (ENABLE,), State.enabled),
@@ -113,8 +97,6 @@ TRANSITIONS = [
("user kill while parked", State.paused, (USER,), State.disabled),
("silent user kill re-parks", State.paused, (USER, SILENT), State.paused),
("hard fault while parked", State.paused, (IMMEDIATE,), State.disabled),
# from overriding
("override released", State.overriding, (), State.enabled),
("override held", State.overriding, (OVERRIDE,), State.overriding),
("override to grace", State.overriding, (SOFT,), State.softDisabling),
@@ -122,7 +104,6 @@ TRANSITIONS = [
("override hard fault", State.overriding, (IMMEDIATE,), State.disabled),
]
@pytest.mark.parametrize("label,start,signals,expected", TRANSITIONS, ids=[t[0] for t in TRANSITIONS])
def test_transition(label, start, signals, expected):
m, _ = machine_at(start, signals)
@@ -133,7 +114,7 @@ def test_transition(label, start, signals, expected):
@pytest.mark.parametrize("start,signals,expected_enabled,expected_active", [
(State.disabled, (), False, False),
(State.disabled, (ENABLE,), True, True),
(State.disabled, (ENABLE, NO_ENTRY, PAUSE_OK), True, False), # paused: guidance armed, torque off
(State.disabled, (ENABLE, NO_ENTRY, PAUSE_OK), True, False),
(State.enabled, (), True, True),
(State.enabled, (SOFT,), True, True),
(State.enabled, (USER,), False, False),
@@ -144,7 +125,6 @@ def test_update_outputs(start, signals, expected_enabled, expected_active):
enabled, active = m.update()
assert (enabled, active) == (expected_enabled, expected_active)
class TestSoftDisableTimer:
def test_grace_period_arms_timer_when_solo(self):
m, host = machine_at(State.enabled, (SOFT,))
@@ -171,7 +151,6 @@ class TestSoftDisableTimer:
m.update()
assert m.state == State.softDisabling
class TestAlertQueueing:
def test_alerts_only_queued_when_solo(self):
m, host = machine_at(State.disabled, (ENABLE,), selfdrive_enabled=True)
@@ -182,7 +161,7 @@ class TestAlertQueueing:
m, host = machine_at(State.disabled, (ENABLE,))
m.update()
assert ET.ENABLE in host.state_machine.current_alert_types
assert ET.WARNING in host.state_machine.current_alert_types # active -> warning channel open
assert ET.WARNING in host.state_machine.current_alert_types
def test_no_entry_alert_queued(self):
m, host = machine_at(State.disabled, (ENABLE, NO_ENTRY))
@@ -190,7 +169,6 @@ class TestAlertQueueing:
assert ET.NO_ENTRY in host.state_machine.current_alert_types
def test_user_disable_alert_always_queued(self):
# user disable bypasses the solo gate — the driver asked, the driver hears back
m, host = machine_at(State.enabled, (USER,), selfdrive_enabled=True)
m.update()
assert ET.USER_DISABLE in host.state_machine.current_alert_types
@@ -200,7 +178,6 @@ class TestAlertQueueing:
m.update()
assert ET.OVERRIDE_LATERAL in host.state_machine.current_alert_types
class TestPauseEligibility:
@pytest.mark.parametrize("event_name", PAUSE_WITH_IQ_EVENTS)
def test_each_iq_pause_event_parks(self, event_name):
@@ -219,3 +196,39 @@ class TestPauseEligibility:
m.state = State.disabled
m.update()
assert m.state == State.paused
OVERRIDE_PAUSE_TRANSITIONS = [
("hands on wheel parks", State.enabled, (OVERRIDE,), State.paused),
("override held stays parked", State.overriding, (OVERRIDE,), State.paused),
("engage while overriding parks", State.disabled, (ENABLE, OVERRIDE), State.paused),
("parked resume waits for release", State.paused, (ENABLE, OVERRIDE), State.paused),
("release resumes", State.paused, (ENABLE,), State.enabled),
("hands off keeps steering", State.enabled, (), State.enabled),
("user kill still kills", State.enabled, (OVERRIDE, USER), State.disabled),
("hard fault still faults", State.enabled, (OVERRIDE, IMMEDIATE), State.disabled),
("grace beats override", State.enabled, (OVERRIDE, SOFT), State.softDisabling),
]
@pytest.mark.parametrize("label,start,signals,expected", OVERRIDE_PAUSE_TRANSITIONS,
ids=[t[0] for t in OVERRIDE_PAUSE_TRANSITIONS])
def test_override_pause_transition(label, start, signals, expected):
m, _ = machine_at(start, signals, pause_on_lateral_override=True)
m.update()
assert m.state == expected
class TestOverridePauseOutputs:
def test_torque_stops_while_overriding(self):
m, _ = machine_at(State.enabled, (OVERRIDE,), pause_on_lateral_override=True)
enabled, active = m.update()
assert (enabled, active) == (True, False)
def test_torque_returns_on_release(self):
m, _ = machine_at(State.paused, (ENABLE,), pause_on_lateral_override=True)
enabled, active = m.update()
assert (enabled, active) == (True, True)
def test_override_alert_not_queued_while_parked(self):
m, host = machine_at(State.enabled, (OVERRIDE,), pause_on_lateral_override=True)
m.update()
assert ET.OVERRIDE_LATERAL not in host.state_machine.current_alert_types