From 9084495242b2a19e8e566ad95813aeea0436a017 Mon Sep 17 00:00:00 2001 From: Zeph Leggett Date: Sun, 30 Aug 2026 21:12:30 -0500 Subject: [PATCH 1/2] mazda: stop claiming a radar the 2016-20 CX-9 does not have MazdaPlatformConfig's default dbc_dict gained Bus.radar for the CX-5 2022 radar track work, which handed every Mazda a radar bus and left radarUnavailable False on all of them. The 2016-20 CX-9 is the one platform that never puts the 0x361-0x366 tracks on bus 0, so its RadarInterface built a parser that never went valid and test_radar_interface failed on every segment of route 10b5a4b380434151. Give that platform its own dbc_dict without the radar bus. Verified against all six Mazda platforms on segments 0-2: only the CX-9 lacks the tracks. Also declare CAM_PEDESTRIAN in the cam parser at nan frequency, next to the three siblings already there. It is read through vl, so it was registering lazily with liveness checks on, and the 2016-20 CX-9 camera never sends it: canValid stayed false for a whole route (5745 frames on segment 0). --- opendbc/car/mazda/carstate.py | 1 + opendbc/car/mazda/values.py | 3 +++ 2 files changed, 4 insertions(+) diff --git a/opendbc/car/mazda/carstate.py b/opendbc/car/mazda/carstate.py index 727afb96ec0..41e3966a4b5 100644 --- a/opendbc/car/mazda/carstate.py +++ b/opendbc/car/mazda/carstate.py @@ -320,6 +320,7 @@ def get_can_parsers(CP, CP_SP): ("CAM_LANEINFO", float("nan")), ("CAM_TRAFFIC_SIGNS", float("nan")), ("CAM_EMPTY", float("nan")), + ("CAM_PEDESTRIAN", float("nan")), ] return { Bus.pt: CANParser(DBC[CP.carFingerprint][Bus.pt], pt_messages, 0), diff --git a/opendbc/car/mazda/values.py b/opendbc/car/mazda/values.py index 5e6ff89aae5..619939ce769 100644 --- a/opendbc/car/mazda/values.py +++ b/opendbc/car/mazda/values.py @@ -256,6 +256,9 @@ class CAR(Platforms): MAZDA_CX9 = MazdaPlatformConfig( [MazdaCarDocs("Mazda CX-9 2016-20")], MazdaCarSpecs(mass=4217 * CV.LB_TO_KG, wheelbase=2.93, steerRatio=17.6), + # no radar bus: this is the one Mazda whose radar does not put the 0x361-0x366 tracks on + # bus 0, so claiming one would leave radard waiting on a parser that never goes valid + dbc_dict={Bus.pt: 'mazda_2017'}, wmis={WMI.JAPAN_CROSSOVER}, chassis_codes={'TC'}, years={'G', 'H', 'J', 'K', 'L'}, # 2016-20 ) MAZDA_3 = MazdaPlatformConfig( From 54bfae971e45e4bff0a3fc9f8c0360692e8224c6 Mon Sep 17 00:00:00 2001 From: Zeph Leggett Date: Sun, 30 Aug 2026 21:12:41 -0500 Subject: [PATCH 2/2] mazda: convert the fork's tests from pytest to unittest opendbc has no pytest: the suite runs under unittest-parallel, the safety tests under unittest discover, and pytest appears in neither pyproject.toml nor uv.lock. These five files were the only ones importing it, so discovery raised ModuleNotFoundError on each and ./test.sh failed on both runners. The 146 tests had never executed here; they only ever ran in the parent openpilot repo, whose venv has pytest. Convert them in place: fixtures become module-level factories, parametrize becomes a class-level case table walked with subTest, and approx becomes assertAlmostEqual or a small _Approx helper where the tolerance is compared inside an expression. Case coverage is unchanged at 166. TestStandstillHold and TestAdvertisedLead had a run() helper, which shadows TestCase.run and would have swallowed every test in those classes; it is drive() now. Discovery goes from 9529 tests with 5 errors to 9656 with none. --- .../car/mazda/tests/test_mazda_carstate.py | 183 +-- .../car/mazda/tests/test_mazda_controller.py | 1082 +++++++++-------- .../car/mazda/tests/test_mazda_interface.py | 47 +- opendbc/car/mazda/tests/test_mazda_radar.py | 35 +- .../car/tests/test_speed_dep_config.py | 37 +- 5 files changed, 746 insertions(+), 638 deletions(-) diff --git a/opendbc/car/mazda/tests/test_mazda_carstate.py b/opendbc/car/mazda/tests/test_mazda_carstate.py index dd109c82503..6b39f4cc4dc 100644 --- a/opendbc/car/mazda/tests/test_mazda_carstate.py +++ b/opendbc/car/mazda/tests/test_mazda_carstate.py @@ -1,4 +1,4 @@ -import pytest +import unittest from opendbc.car import DT_CTRL, gen_empty_fingerprint from opendbc.car.common.conversions import Conversions as CV @@ -42,18 +42,20 @@ def _feed(CI, payload, seconds): return CI.CS.fsc_settled -@pytest.mark.parametrize("alpha_long", [False, True]) -def test_carstate_runs_with_real_parsers(alpha_long): - # vl_all, unlike vl, has no lazy message registration: every message read through it - # must be listed in get_can_parsers. The op-long FSC settle gate crashed card on its - # first update when CAM_LANEINFO was missing from the cam parser (KeyError, 2026-07-29). - CI = _interface(alpha_long) - assert CI.CP.openpilotLongitudinalControl == alpha_long - for _ in range(10): - CI.update([]) +class TestCarStateParsers(unittest.TestCase): + def test_carstate_runs_with_real_parsers(self): + # vl_all, unlike vl, has no lazy message registration: every message read through it + # must be listed in get_can_parsers. The op-long FSC settle gate crashed card on its + # first update when CAM_LANEINFO was missing from the cam parser (KeyError, 2026-07-29). + for alpha_long in (False, True): + with self.subTest(alpha_long=alpha_long): + CI = _interface(alpha_long) + self.assertEqual(CI.CP.openpilotLongitudinalControl, alpha_long) + for _ in range(10): + CI.update([]) -class TestFscSettleGate: +class TestFscSettleGate(unittest.TestCase): """The gate that defers the radar teardown past the FSC's cold-boot radar-presence check. It must hold while the camera is booting or faulted, and must not be vetoed indefinitely @@ -62,30 +64,30 @@ class TestFscSettleGate: def test_never_settles_while_boot_marker_is_set(self): settle = CarControllerParams.FSC_SETTLE_T - assert not _feed(_interface(), BOOTING, settle * 2) + self.assertFalse(_feed(_interface(), BOOTING, settle * 2)) def test_never_settles_while_err_bit_is_set(self): # a latched i-ACTIVSENSE fault shows the boot markers clear, so ERR_BIT must veto on its own settle = CarControllerParams.FSC_SETTLE_T - assert not _feed(_interface(), FAULTED, settle * 2) + self.assertFalse(_feed(_interface(), FAULTED, settle * 2)) def test_settles_once_the_boot_marker_clears(self): CI = _interface() - assert not _feed(CI, BOOTING, 3.0) - assert not _feed(CI, SETTLED, CarControllerParams.FSC_SETTLE_T - 1.0) - assert _feed(CI, SETTLED, 1.5) + self.assertFalse(_feed(CI, BOOTING, 3.0)) + self.assertFalse(_feed(CI, SETTLED, CarControllerParams.FSC_SETTLE_T - 1.0)) + self.assertTrue(_feed(CI, SETTLED, 1.5)) def test_a_latched_bit2_does_not_block_the_teardown_forever(self): # One CX-5 2022 cold-booted with BIT2 high and NO_ERR_BIT clear for an entire ignition # cycle (36.5 s, route 7c735af5fce56485|00000011). BIT2 was in the gate, so the radar was # never silenced and the two-master guard held accFaulted for the whole drive. - assert _feed(_interface(), BIT2_LATCHED, CarControllerParams.FSC_SETTLE_T * 1.5) + self.assertTrue(_feed(_interface(), BIT2_LATCHED, CarControllerParams.FSC_SETTLE_T * 1.5)) def test_settles_at_the_longest_observed_camera_period(self): # _feed runs at the longest measured period, the worst case the freshness window has to # ride through: a shorter window zeroes the settle counter on every gap and the gate # never opens (the regression that shipped in baf0f383c3 with CAM_LANEINFO_FRESH_T = 0.5) - assert _feed(_interface(), SETTLED, CarControllerParams.FSC_SETTLE_T * 1.5) + self.assertTrue(_feed(_interface(), SETTLED, CarControllerParams.FSC_SETTLE_T * 1.5)) def test_camera_dropout_resets_the_settle_timer(self): # the window is a freshness gate, not decoration: a genuine dropout, well past any real @@ -93,18 +95,18 @@ def test_camera_dropout_resets_the_settle_timer(self): CI = _interface() _feed(CI, SETTLED, CarControllerParams.FSC_SETTLE_T * 0.8) _feed(CI, None, CarControllerParams.CAM_LANEINFO_FRESH_T + 0.5) - assert not _feed(CI, SETTLED, CarControllerParams.FSC_SETTLE_T * 0.5) - assert _feed(CI, SETTLED, CarControllerParams.FSC_SETTLE_T * 0.6) + self.assertFalse(_feed(CI, SETTLED, CarControllerParams.FSC_SETTLE_T * 0.5)) + self.assertTrue(_feed(CI, SETTLED, CarControllerParams.FSC_SETTLE_T * 0.6)) def test_gate_starts_closed_before_any_camera_frame(self): # the parser reads all-zero before the first frame, which would otherwise look settled CI = _interface() for i in range(int(CarControllerParams.FSC_SETTLE_T * 2 / DT_CTRL)): CI.update([(int(i * DT_CTRL * 1e9), [])]) - assert not CI.CS.fsc_settled + self.assertFalse(CI.CS.fsc_settled) -class TestStockFcw: +class TestStockFcw(unittest.TestCase): """0x21d (CAM_EMPTY) idles at STATUS 0x7f and leaves it only while the camera actively shows its SCBS collision display (route 0000004d t+213). The payloads are the captured idle and active frames from that route.""" @@ -118,74 +120,77 @@ def _feed_21d(self, CI, payload, i=0): def test_display_active_sets_fcw(self): CI = _interface() - assert self._feed_21d(CI, self.IDLE).stockFcw is False - assert self._feed_21d(CI, self.ACTIVE, 1).stockFcw is True - assert self._feed_21d(CI, self.IDLE, 2).stockFcw is False + self.assertIs(self._feed_21d(CI, self.IDLE).stockFcw, False) + self.assertIs(self._feed_21d(CI, self.ACTIVE, 1).stockFcw, True) + self.assertIs(self._feed_21d(CI, self.IDLE, 2).stockFcw, False) def test_no_fcw_before_first_camera_frame(self): # the parser reads STATUS as 0 before the first frame, which is != 0x7f CI = _interface() ret, _ = CI.update([(0, [])]) - assert ret.stockFcw is False + self.assertIs(ret.stockFcw, False) def test_ped_warning_bit_sets_fcw(self): # never observed in 1.57M corpus frames, wired for coverage: PED_WARNING is bit 9 CI = _interface() self._feed_21d(CI, self.IDLE) ret, _ = CI.update([(int(1 * DT_CTRL * 1e9), [(CAM_PEDESTRIAN, bytes.fromhex("07fa3c0000000000"), 2), (CAM_EMPTY, self.IDLE, 2)])]) - assert ret.stockFcw is True + self.assertIs(ret.stockFcw, True) -class TestRadarSessionResponse: +class TestRadarSessionResponse(unittest.TestCase): """The radar answers session requests within ~10 ms (route 000000fe t+15.0), and the session manager consumes the flag on the same control frame it is set.""" def test_negative_response_sets_refused(self): CI = _interface() - assert not CI.CS.radar_session_refused + self.assertFalse(CI.CS.radar_session_refused) # 03 7F 10 22: conditionsNotCorrect to a session-control request CI.update([(0, [(RADAR_UDS_RESP, bytes.fromhex("037f102200000000"), 0)])]) - assert CI.CS.radar_session_refused + self.assertTrue(CI.CS.radar_session_refused) CI.update([(int(DT_CTRL * 1e9), [])]) - assert not CI.CS.radar_session_refused, "the flag is same-frame, not latched" + self.assertFalse(CI.CS.radar_session_refused, "the flag is same-frame, not latched") def test_positive_response_is_not_a_refusal(self): CI = _interface() # the real capture: 06 50 02 with the session parameter record (P2*=5.0 s) CI.update([(0, [(RADAR_UDS_RESP, bytes.fromhex("065002001901f400"), 0)])]) - assert not CI.CS.radar_session_refused + self.assertFalse(CI.CS.radar_session_refused) def test_response_pending_is_not_a_refusal(self): CI = _interface() # 03 7F 10 78: requestCorrectlyReceived-ResponsePending; UDS clients wait through it CI.update([(0, [(RADAR_UDS_RESP, bytes.fromhex("037f107800000000"), 0)])]) - assert not CI.CS.radar_session_refused + self.assertFalse(CI.CS.radar_session_refused) -class TestBrakeHold: +class TestBrakeHold(unittest.TestCase): """GEAR.BRAKE_HOLD is the body ECU reporting that it owns the standstill hold. Stock relaxes its own command the instant this sets, so the payloads below come straight off the two logs that pinned the signal down: a hold that latched (route caace206f6 seg 8, 0x17 at 1157.34 s) and one that never did (route 00000065 seg 4, stuck at 0x07 while the car crept).""" - @pytest.mark.parametrize(("payload", "expected"), [ + HOLD_BIT_CASES = [ ("142007ff02f00000", False), # hold not taken over: keep braking ("142017ff02f00000", True), # body has the brakes ("14200fff02f00000", False), # released again at the resume - ]) - def test_decodes_the_hold_bit(self, payload, expected): - CI = _interface() - # CANParser registers a message lazily on first access, so the first frame only arms it - for i in range(2): - CI.update([(int(i * DT_CTRL * 1e9), [(0x228, bytes.fromhex(payload), 0)])]) - assert CI.CS.brake_hold is expected + ] + + def test_decodes_the_hold_bit(self): + for payload, expected in self.HOLD_BIT_CASES: + with self.subTest(payload=payload): + CI = _interface() + # CANParser registers a message lazily on first access, so the first frame only arms it + for i in range(2): + CI.update([(int(i * DT_CTRL * 1e9), [(0x228, bytes.fromhex(payload), 0)])]) + self.assertIs(CI.CS.brake_hold, expected) def test_defaults_to_not_held(self): # nothing parsed yet must read as "the car is not holding", the direction that keeps braking - assert not _interface().CS.brake_hold + self.assertFalse(_interface().CS.brake_hold) -class TestTwoMasterGuard: +class TestTwoMasterGuard(unittest.TestCase): """The stock-radar guard wears two hats: before the first teardown it is the expected boot phase and must only hold availability low (no fault alert); once the radar has been silenced, hearing it again is a genuine two-master conflict and must raise accFaulted.""" @@ -208,16 +213,16 @@ def test_boot_phase_is_not_a_fault(self): # radar broadcasting, teardown not started: engagement blocked quietly, no Cruise Fault CI = _interface() ret, _ = self._feed_guard(CI, 5.0, radar_alive=True) - assert not ret.accFaulted - assert not ret.cruiseState.available + self.assertFalse(ret.accFaulted) + self.assertFalse(ret.cruiseState.available) def test_availability_arrives_with_radar_silence(self): CI = _interface() ret, n = self._feed_guard(CI, 5.0, radar_alive=True) ret, n = self._feed_guard(CI, CarControllerParams.STOCK_RADAR_GUARD_T + 0.5, radar_alive=False, start_frame=n) - assert not ret.accFaulted - assert ret.cruiseState.available + self.assertFalse(ret.accFaulted) + self.assertTrue(ret.cruiseState.available) def test_radar_return_after_teardown_is_a_fault(self): CI = _interface() @@ -225,15 +230,15 @@ def test_radar_return_after_teardown_is_a_fault(self): ret, n = self._feed_guard(CI, CarControllerParams.STOCK_RADAR_GUARD_T + 0.5, radar_alive=False, start_frame=n) ret, n = self._feed_guard(CI, 0.5, radar_alive=True, start_frame=n) - assert ret.accFaulted + self.assertTrue(ret.accFaulted) # availability keys on the latched "was silenced", so a transient return does not # yank lateral out from under MADS on top of the fault - assert ret.cruiseState.available + self.assertTrue(ret.cruiseState.available) # silence restores the clean state ret, n = self._feed_guard(CI, CarControllerParams.STOCK_RADAR_GUARD_T + 0.5, radar_alive=False, start_frame=n) - assert not ret.accFaulted - assert ret.cruiseState.available + self.assertFalse(ret.accFaulted) + self.assertTrue(ret.cruiseState.available) def test_stock_engagement_inside_the_guard_is_not_reported(self): # The radar is still master during the boot phase, so a stock MRCC engage is not ours to @@ -241,8 +246,8 @@ def test_stock_engagement_inside_the_guard_is_not_reported(self): # every off-switch shut (route 00000057). CI = _interface() ret, _ = self._feed_guard(CI, 5.0, radar_alive=True, acc_active=True) - assert not ret.cruiseState.available - assert not ret.cruiseState.enabled + self.assertFalse(ret.cruiseState.available) + self.assertFalse(ret.cruiseState.enabled) def test_engagement_still_live_when_the_guard_lifts_is_not_adopted(self): # Silence alone must not turn a pre-existing stock engagement into an openpilot engage: @@ -251,8 +256,8 @@ def test_engagement_still_live_when_the_guard_lifts_is_not_adopted(self): ret, n = self._feed_guard(CI, 5.0, radar_alive=True, acc_active=True) ret, n = self._feed_guard(CI, CarControllerParams.STOCK_RADAR_GUARD_T + 0.5, radar_alive=False, acc_active=True, start_frame=n) - assert ret.cruiseState.available - assert not ret.cruiseState.enabled + self.assertTrue(ret.cruiseState.available) + self.assertFalse(ret.cruiseState.enabled) def test_engagement_after_an_idle_sample_is_adopted(self): CI = _interface() @@ -260,12 +265,12 @@ def test_engagement_after_an_idle_sample_is_adopted(self): ret, n = self._feed_guard(CI, CarControllerParams.STOCK_RADAR_GUARD_T + 0.5, radar_alive=False, acc_active=True, start_frame=n) ret, n = self._feed_guard(CI, 0.2, radar_alive=False, acc_active=False, start_frame=n) - assert not ret.cruiseState.enabled + self.assertFalse(ret.cruiseState.enabled) ret, n = self._feed_guard(CI, 0.2, radar_alive=False, acc_active=True, start_frame=n) - assert ret.cruiseState.enabled + self.assertTrue(ret.cruiseState.enabled) -class TestSpeedSignLimit: +class TestSpeedSignLimit(unittest.TestCase): """CAM_TRAFFIC_SIGNS.SPEED_SIGN_ON is a 2-bit field carrying the display unit, not a 1-bit on-flag: 1 = limit displayed in mph, 2 = displayed in km/h, 0 = none. Which value an FSC emits tracks its market, not the cluster's unit setting. Payloads are real captures: mph @@ -273,7 +278,7 @@ class TestSpeedSignLimit: ded445e51c0e1830|00000007--4b5a89a1ce) where the old 1-bit decode at bit 12 read 0 and SLA never saw a limit.""" - @pytest.mark.parametrize(("payload", "expected_ms"), [ + UNIT_CASES = [ ("0000000002005300", 0.0), # no limit displayed ("0650000002005300", 25 * CV.MPH_TO_MS), # US 25 mph ("0a10000003001300", 40 * CV.MPH_TO_MS), # US 40 mph @@ -281,32 +286,38 @@ class TestSpeedSignLimit: ("0a20000002000900", 40 * CV.KPH_TO_MS), # NZ 40 km/h ("0ca0000002000900", 50 * CV.KPH_TO_MS), # NZ 50 km/h ("1920000002010900", 100 * CV.KPH_TO_MS), # NZ 100 km/h - ]) - def test_unit_comes_from_the_frame(self, payload, expected_ms): - CI = _interface() - ret_sp = None - for i in range(2): - _, ret_sp = CI.update([(int(i * DT_CTRL * 1e9), [(0x35F, bytes.fromhex(payload), 2)])]) - assert ret_sp.speedLimit == pytest.approx(expected_ms) - - @pytest.mark.parametrize(("sign_on", "speed_sign"), [ + ] + + def test_unit_comes_from_the_frame(self): + for payload, expected_ms in self.UNIT_CASES: + with self.subTest(payload=payload): + CI = _interface() + ret_sp = None + for i in range(2): + _, ret_sp = CI.update([(int(i * DT_CTRL * 1e9), [(0x35F, bytes.fromhex(payload), 2)])]) + self.assertAlmostEqual(ret_sp.speedLimit, expected_ms, delta=max(abs(expected_ms) * 1e-6, 1e-12)) + + IMPLAUSIBLE_CASES = [ (1, 120), # above any real mph posting (2, 127), # all-ones: invalid sentinel (3, 50), # undefined state (1, 0), # displayed-but-zero - ]) - def test_implausible_frames_read_as_no_limit(self, sign_on, speed_sign): + ] + + def test_implausible_frames_read_as_no_limit(self): from opendbc.can import CANPacker packer = CANPacker("mazda_2017") - msg = packer.make_can_msg("CAM_TRAFFIC_SIGNS", 2, {"SPEED_SIGN_ON": sign_on, "SPEED_SIGN": speed_sign}) - CI = _interface() - ret_sp = None - for i in range(2): - _, ret_sp = CI.update([(int(i * DT_CTRL * 1e9), [(msg[0], msg[1], msg[2])])]) - assert ret_sp.speedLimit == 0.0 + for sign_on, speed_sign in self.IMPLAUSIBLE_CASES: + with self.subTest(sign_on=sign_on, speed_sign=speed_sign): + msg = packer.make_can_msg("CAM_TRAFFIC_SIGNS", 2, {"SPEED_SIGN_ON": sign_on, "SPEED_SIGN": speed_sign}) + CI = _interface() + ret_sp = None + for i in range(2): + _, ret_sp = CI.update([(int(i * DT_CTRL * 1e9), [(msg[0], msg[1], msg[2])])]) + self.assertEqual(ret_sp.speedLimit, 0.0) -class TestCancelUnderBraking: +class TestCancelUnderBraking(unittest.TestCase): """The availability brake-hold exists for brake-only PEDALS samples that arrive with both bits low mid-press. A wheel CANCEL turns the MRCC main state off for real and must land even with the brake down (route 7f9e3ff336 t+484-488: cancel mashed under braking was @@ -320,7 +331,7 @@ def _armed_and_silent(self, CI): for i in range(int(guard / DT_CTRL)): msgs = [packer.make_can_msg("PEDALS", 0, {"ACC_OFF": 1})] ret, _ = CI.update([(int(i * DT_CTRL * 1e9), [(m[0], m[1], m[2]) for m in msgs])]) - assert ret.cruiseState.available + self.assertTrue(ret.cruiseState.available) return packer, int(guard / DT_CTRL) def _feed(self, CI, packer, n0, seconds, brake, cancel): @@ -336,13 +347,13 @@ def test_brake_only_dropout_is_held(self): CI = _interface() packer, n = self._armed_and_silent(CI) ret, n = self._feed(CI, packer, n, 1.0, brake=True, cancel=False) - assert ret.cruiseState.available + self.assertTrue(ret.cruiseState.available) def test_cancel_lands_through_the_brake(self): CI = _interface() packer, n = self._armed_and_silent(CI) ret, n = self._feed(CI, packer, n, 0.3, brake=True, cancel=True) - assert not ret.cruiseState.available + self.assertFalse(ret.cruiseState.available) def test_cancel_context_outlives_the_press(self): # the PEDALS reaction can trail the button: press-and-release while still armed, then the @@ -354,12 +365,12 @@ def test_cancel_context_outlives_the_press(self): msgs = [packer.make_can_msg("PEDALS", 0, {"ACC_OFF": 1}), packer.make_can_msg("CRZ_BTNS", 0, {"CAN_OFF": 1})] ret, _ = CI.update([(int(i * DT_CTRL * 1e9), [(m[0], m[1], m[2]) for m in msgs])]) - assert ret.cruiseState.available + self.assertTrue(ret.cruiseState.available) ret, n = self._feed(CI, packer, n + 5, 0.2, brake=True, cancel=False) - assert not ret.cruiseState.available + self.assertFalse(ret.cruiseState.available) -class TestCruiseStandstill: +class TestCruiseStandstill(unittest.TestCase): """PEDALS.STANDSTILL is the PCM's wheel-speed "stopped" bit, not a stock-ACC hold state. LongControl's starting_condition is `not should_stop and not cruise_standstill and not @@ -384,8 +395,8 @@ def _standstill(self, alpha_long): def test_not_reported_under_openpilot_longitudinal(self): # the stock MRCC is not in the loop at all here: its radar is silenced and we synthesize # its frames, so there is no stock standstill state to report - assert not self._standstill(alpha_long=True) + self.assertFalse(self._standstill(alpha_long=True)) def test_still_reported_with_stock_longitudinal(self): # stock long still needs it: it is what drives CC.cruiseControl.resume in controlsd - assert self._standstill(alpha_long=False) + self.assertTrue(self._standstill(alpha_long=False)) diff --git a/opendbc/car/mazda/tests/test_mazda_controller.py b/opendbc/car/mazda/tests/test_mazda_controller.py index 51d29b13cb9..2bb37a9aa4b 100644 --- a/opendbc/car/mazda/tests/test_mazda_controller.py +++ b/opendbc/car/mazda/tests/test_mazda_controller.py @@ -5,7 +5,7 @@ from types import SimpleNamespace import numpy as np -import pytest +import unittest from opendbc.can import CANPacker, CANParser from opendbc.car import Bus, DT_CTRL, structs @@ -18,24 +18,32 @@ from opendbc.car.mazda.values import CAR, CarControllerParams -class TestCarControllerParams: +def _cx5_2022_params(): + class FakeCP: + carFingerprint = CAR.MAZDA_CX5_2022 + minSteerSpeed = 0.0 # steer_to_zero -> CX-5 2022+ EPS present + return CarControllerParams(FakeCP()) - @pytest.fixture - def cx5_2022_params(self): - class FakeCP: - carFingerprint = CAR.MAZDA_CX5_2022 - minSteerSpeed = 0.0 # steer_to_zero -> CX-5 2022+ EPS present - return CarControllerParams(FakeCP()) - @pytest.fixture - def eps_swap_params(self): - # A CX-5 2022+ EPS swapped into (or shared by) another Mazda: different model, same EPS. - class FakeCP: - carFingerprint = CAR.MAZDA_CX9_2021 - minSteerSpeed = 0.0 - return CarControllerParams(FakeCP()) +def _eps_swap_params(): + # A CX-5 2022+ EPS swapped into (or shared by) another Mazda: different model, same EPS. + class FakeCP: + carFingerprint = CAR.MAZDA_CX9_2021 + minSteerSpeed = 0.0 + return CarControllerParams(FakeCP()) - def test_eps_ceiling_never_exceeds_steer_max_scale(self, cx5_2022_params): + +def _pre_2022_params(): + class FakeCP: + carFingerprint = CAR.MAZDA_CX5 + minSteerSpeed = 12.5 # no CX-5 EPS -> low-speed lockout, minSteerSpeed > 0 + return CarControllerParams(FakeCP()) + + +class TestCarControllerParams(unittest.TestCase): + + def test_eps_ceiling_never_exceeds_steer_max_scale(self): + cx5_2022_params = _cx5_2022_params() # The ceiling is a clamp on delivered-torque counts; the scale is STEER_MAX. The clamp is # only meaningful if it sits at or below the scale at every speed. bp, vals = cx5_2022_params.EPS_CEILING_LOOKUP @@ -43,74 +51,91 @@ def test_eps_ceiling_never_exceeds_steer_max_scale(self, cx5_2022_params): ceiling = np.interp(v, bp, vals) steer_max = np.interp(v, cx5_2022_params.STEER_MAX_LOOKUP[0], cx5_2022_params.STEER_MAX_LOOKUP[1]) - assert 0 < ceiling <= steer_max, f"ceiling {ceiling} vs steer_max {steer_max} at {v} m/s" + self.assertLessEqual(0 < ceiling, steer_max, msg=f"ceiling {ceiling} vs steer_max {steer_max} at {v} m/s") - def test_eps_ceiling_is_monotone_and_matches_the_measured_rails(self, cx5_2022_params): + def test_eps_ceiling_is_monotone_and_matches_the_measured_rails(self): + cx5_2022_params = _cx5_2022_params() # Measured over 11.4M clean frames: 1148 below 18 mph, a monotone rolloff, hard 620 from # 32.5 mph up (docs/mazda-lkas-camera-tx-census.md). Nothing above 620 was ever delivered # above 32.5 mph in 7.5M frames, so the high-speed leg must not drift back up. bp, vals = cx5_2022_params.EPS_CEILING_LOOKUP - assert list(vals) == sorted(vals, reverse=True), "ceiling must fall monotonically with speed" - assert np.interp(5.0, bp, vals) == 1148 - assert np.interp(14.5, bp, vals) == 620 - assert np.interp(35.0, bp, vals) == 620 + self.assertEqual(list(vals), sorted(vals, reverse=True), msg="ceiling must fall monotonically with speed") + self.assertEqual(np.interp(5.0, bp, vals), 1148) + self.assertEqual(np.interp(14.5, bp, vals), 620) + self.assertEqual(np.interp(35.0, bp, vals), 620) - def test_steer_delta_matches_the_eps_rate_limit_at_this_steer_step(self, cx5_2022_params): + def test_steer_delta_matches_the_eps_rate_limit_at_this_steer_step(self): + cx5_2022_params = _cx5_2022_params() # The EPS rate limit is per unit TIME (~1200 units/s), while STEER_DELTA_UP/DOWN are per # frame, so the two are only matched at STEER_STEP = 1. Changing one without the other # silently rescales the commanded slew rate. Both directions: the measured rail is # symmetric (p99 and p99.9 of the delivered step are 12 either way), and a winddown above # it only lets the command run ahead of the wheel. rate_hz = 1.0 / DT_CTRL / CarControllerParams.STEER_STEP - assert cx5_2022_params.STEER_DELTA_UP * rate_hz == pytest.approx(1200, rel=0.01) - assert cx5_2022_params.STEER_DELTA_DOWN * rate_hz == pytest.approx(1200, rel=0.01) - - @pytest.fixture - def pre_2022_params(self): - class FakeCP: - carFingerprint = CAR.MAZDA_CX5 - minSteerSpeed = 12.5 # no CX-5 EPS -> low-speed lockout, minSteerSpeed > 0 - return CarControllerParams(FakeCP()) + self.assertEqual(cx5_2022_params.STEER_DELTA_UP * rate_hz, _Approx(1200, abs(1200) * 0.01)) + self.assertEqual(cx5_2022_params.STEER_DELTA_DOWN * rate_hz, _Approx(1200, abs(1200) * 0.01)) - def test_cx5_2022_has_lookup(self, cx5_2022_params): - assert hasattr(cx5_2022_params, 'STEER_MAX_LOOKUP') - assert cx5_2022_params.STEER_MAX == 1200 + def test_cx5_2022_has_lookup(self): + cx5_2022_params = _cx5_2022_params() + self.assertTrue(hasattr(cx5_2022_params, 'STEER_MAX_LOOKUP')) + self.assertEqual(cx5_2022_params.STEER_MAX, 1200) - def test_cx5_2022_low_speed(self, cx5_2022_params): + def test_cx5_2022_low_speed(self): + cx5_2022_params = _cx5_2022_params() p = cx5_2022_params for v in [0.0, 5.0, 10.0, 14.2]: sm = round(float(np.interp(v, p.STEER_MAX_LOOKUP[0], p.STEER_MAX_LOOKUP[1]))) - assert sm == 1200 + self.assertEqual(sm, 1200) - def test_cx5_2022_high_speed(self, cx5_2022_params): + def test_cx5_2022_high_speed(self): + cx5_2022_params = _cx5_2022_params() p = cx5_2022_params for v in [14.5, 20.0, 30.0]: sm = round(float(np.interp(v, p.STEER_MAX_LOOKUP[0], p.STEER_MAX_LOOKUP[1]))) - assert sm == 800 + self.assertEqual(sm, 800) - def test_cx5_2022_rate_limits(self, cx5_2022_params): - assert cx5_2022_params.STEER_DELTA_UP == 12 - assert cx5_2022_params.STEER_DELTA_DOWN == 12 + def test_cx5_2022_rate_limits(self): + cx5_2022_params = _cx5_2022_params() + self.assertEqual(cx5_2022_params.STEER_DELTA_UP, 12) + self.assertEqual(cx5_2022_params.STEER_DELTA_DOWN, 12) - def test_cx5_2022_winddown_stays_within_the_panda_backstop(self, cx5_2022_params): + def test_cx5_2022_winddown_stays_within_the_panda_backstop(self): + cx5_2022_params = _cx5_2022_params() # opendbc/safety/modes/mazda.h declares max_rate_down = 25 for every Mazda; commanding a # tighter winddown is allowed, commanding a looser one would be blocked frame by frame. - assert cx5_2022_params.STEER_DELTA_DOWN <= 25 + self.assertLessEqual(cx5_2022_params.STEER_DELTA_DOWN, 25) - def test_cx5_eps_driver_multiplier(self, cx5_2022_params): + def test_cx5_eps_driver_multiplier(self): + cx5_2022_params = _cx5_2022_params() # 15 is the CX-5-EPS tune (upstream stock is 1) - assert cx5_2022_params.STEER_DRIVER_MULTIPLIER == 15 + self.assertEqual(cx5_2022_params.STEER_DRIVER_MULTIPLIER, 15) - def test_eps_swap_gets_cx5_tune(self, eps_swap_params): + def test_eps_swap_gets_cx5_tune(self): + eps_swap_params = _eps_swap_params() # EPS present (minSteerSpeed == 0) on a non-CX-5 model still gets the higher-authority tune - assert eps_swap_params.STEER_MAX == 1200 - assert eps_swap_params.STEER_DRIVER_MULTIPLIER == 15 - assert hasattr(eps_swap_params, 'STEER_MAX_LOOKUP') + self.assertEqual(eps_swap_params.STEER_MAX, 1200) + self.assertEqual(eps_swap_params.STEER_DRIVER_MULTIPLIER, 15) + self.assertTrue(hasattr(eps_swap_params, 'STEER_MAX_LOOKUP')) + + def test_no_eps_no_lookup(self): + pre_2022_params = _pre_2022_params() + self.assertFalse(hasattr(pre_2022_params, 'STEER_MAX_LOOKUP')) + self.assertEqual(pre_2022_params.STEER_MAX, 800) + self.assertEqual(pre_2022_params.STEER_DRIVER_MULTIPLIER, 1) - def test_no_eps_no_lookup(self, pre_2022_params): - assert not hasattr(pre_2022_params, 'STEER_MAX_LOOKUP') - assert pre_2022_params.STEER_MAX == 800 - assert pre_2022_params.STEER_DRIVER_MULTIPLIER == 1 + +class _Approx: + """`value == _Approx(expected, tol)` -- pytest.approx's tolerance check, without pytest.""" + + def __init__(self, expected, tol): + self.expected = expected + self.tol = tol + + def __eq__(self, other): + return abs(other - self.expected) <= self.tol + + def __repr__(self): + return f"{self.expected} +/- {self.tol}" def crz_info_reference_checksum(dat): @@ -124,15 +149,16 @@ def decode_accel_cmd_raw(dat): return (((dat[2] & 0x3) << 11) | (dat[3] << 3) | (dat[4] >> 5)) - 4096 -class TestMazdaLongitudinalMessages: +def _packer(): + return CANPacker("mazda_2017") + + +class TestMazdaLongitudinalMessages(unittest.TestCase): """The synthetic CRZ_INFO/CRZ_CTRL/radar frames must reproduce stock captures byte for byte; the hex values below come from real radar traffic.""" - @pytest.fixture - def packer(self): - return CANPacker("mazda_2017") - - def test_alert_command_relays_state_but_not_the_tja_churn(self, packer): + def test_alert_command_relays_state_but_not_the_tja_churn(self): + packer = _packer() # camera error and line state pass through to the dash; the camera's own TJA/CTS state # machine churns against steering it did not command (442 TJA_TRANSITION toggles in 22 # min, route 0000010b) and relaying it flapped the dash, so those two fields stay zeroed @@ -143,17 +169,19 @@ def test_alert_command_relays_state_but_not_the_tja_churn(self, packer): cp = CANParser("mazda_2017", [("CAM_LANEINFO", float("nan"))], 0) cp.update([(0, [(0x440, dat, 0)])]) out = cp.vl["CAM_LANEINFO"] - assert out["ERR_BIT"] == 1 and out["LINE_VISIBLE"] == 1 and out["LANE_LINES"] == 2 and out["S1"] == 1 - assert out["TJA"] == 0 and out["TJA_TRANSITION"] == 0 + self.assertTrue(out["ERR_BIT"] == 1 and out["LINE_VISIBLE"] == 1 and out["LANE_LINES"] == 2 and out["S1"] == 1) + self.assertTrue(out["TJA"] == 0 and out["TJA_TRANSITION"] == 0) - def test_crz_info_standby_matches_stock(self, packer): + def test_crz_info_standby_matches_stock(self): + packer = _packer() for counter in range(16): checksum = (0x5d - counter) & 0xff expected = f"01ffe3ffc000{counter:02x}{checksum:02x}" dat = mazdacan.create_acc_command(packer, 0, counter, 0.0, long_active=False, acc_available=False)[1] - assert dat.hex() == expected + self.assertEqual(dat.hex(), expected) - def test_crz_info_armed_idle_matches_stock(self, packer): + def test_crz_info_armed_idle_matches_stock(self): + packer = _packer() # armed-idle pegs the command like standby (47,752/47,752 stock armed-idle frames carry # raw 8190) and follows the brake on ACC_SET_ALLOWED; the zero-command armed-idle this # used to emit exists nowhere in the stock corpus @@ -163,9 +191,9 @@ def test_crz_info_armed_idle_matches_stock(self, packer): expected = f"01ffe3ff{byte4:02x}80{counter:02x}{checksum:02x}" dat = mazdacan.create_acc_command(packer, 0, counter, 0.0, long_active=False, acc_available=True, brake_pressed=brake_pressed)[1] - assert dat.hex() == expected + self.assertEqual(dat.hex(), expected) - @pytest.mark.parametrize(("accel", "stopping", "unlatching", "counter", "expected"), [ + CRZ_INFO_ENGAGED_GOLDEN_BYTES_CASES = [ (0.0, False, False, 0, "01ffe20006800097"), # engaged, zero command (2.0, False, False, 3, "01ffe2fa0680039a"), # ISO max accel, raw 2000 (-3.5, False, False, 7, "01ffe04a868007c8"), # ISO max brake, raw -3500 @@ -175,32 +203,37 @@ def test_crz_info_armed_idle_matches_stock(self, packer): # wire-attested twin: stock drive_0b's first pulse frame is 01ffe1ffe68040b9 -- the # checksum matches the counter-0 hold frame (b9) because the unlatch bit is not summed (-0.001, False, True, 0, "01ffe1ffe68040b9"), - ]) - def test_crz_info_engaged_golden_bytes(self, packer, accel, stopping, unlatching, counter, expected): - dat = mazdacan.create_acc_command(packer, 0, counter, accel, long_active=True, acc_available=False, - stopping=stopping, resume_unlatching=unlatching)[1] - assert dat.hex() == expected + ] - def test_crz_info_accel_encoding_and_checksum(self, packer): + def test_crz_info_engaged_golden_bytes(self): + packer = _packer() + for accel, stopping, unlatching, counter, expected in self.CRZ_INFO_ENGAGED_GOLDEN_BYTES_CASES: + with self.subTest(accel=accel, stopping=stopping, unlatching=unlatching, counter=counter, expected=expected): + dat = mazdacan.create_acc_command(packer, 0, counter, accel, long_active=True, acc_available=False, + stopping=stopping, resume_unlatching=unlatching)[1] + self.assertEqual(dat.hex(), expected) + + def test_crz_info_accel_encoding_and_checksum(self): + packer = _packer() # the packed command must round-trip at the 0.001 factor and carry a valid masked-bit # checksum over the whole command window, stop bits set or not for raw in range(-3500, 2001, 137): for stopping, unlatching in ((False, False), (True, False), (False, True)): dat = mazdacan.create_acc_command(packer, 0, raw % 16, raw / 1000.0, long_active=True, acc_available=False, stopping=stopping, resume_unlatching=unlatching)[1] - assert decode_accel_cmd_raw(dat) == raw - assert dat[7] == crz_info_reference_checksum(dat) - assert bool(dat[5] & 0x04) == stopping - assert bool(dat[6] & 0x10) == stopping - assert bool(dat[6] & 0x40) == unlatching + self.assertEqual(decode_accel_cmd_raw(dat), raw) + self.assertEqual(dat[7], crz_info_reference_checksum(dat)) + self.assertEqual(bool(dat[5] & 0x04), stopping) + self.assertEqual(bool(dat[6] & 0x10), stopping) + self.assertEqual(bool(dat[6] & 0x40), unlatching) # the excluded event bits must not move the checksum: stripping them yields the # same byte a bare frame carries (stock 0b: 8040b9 vs 8000b9, both chk b9) bare = bytearray(dat) bare[5] &= ~0x04 bare[6] &= ~0x40 - assert dat[7] == crz_info_reference_checksum(bytes(bare)) + self.assertEqual(dat[7], crz_info_reference_checksum(bytes(bare))) - @pytest.mark.parametrize(("long_active", "acc_available", "gap", "has_lead", "phase", "acc_active_2", "expected"), [ + CRZ_CTRL_GOLDEN_BYTES_CASES = [ (False, False, 0, False, 0, False, "0201010000000000"), # standby (False, True, 2, False, 0, False, "02010b0000000000"), # MRCC armed, SET allowed (True, True, 2, True, 1, True, "0a018b2000001000"), # engaged, cruise, no lead @@ -209,10 +242,15 @@ def test_crz_info_accel_encoding_and_checksum(self, packer): (True, True, 2, True, 4, True, "0a018b8000001000"), # stop-and-go hold (far phase) (True, True, 2, True, 3, False, "0a018b6000000000"), # relaxed hold, ACC_ACTIVE_2 drops (True, True, 1, True, 2, True, "0a01874000001000"), # driver gap 1 mirrored to the dash - ]) - def test_crz_ctrl_golden_bytes(self, packer, long_active, acc_available, gap, has_lead, phase, acc_active_2, expected): - dat = mazdacan.create_crz_ctrl(packer, 0, long_active, acc_available, gap, has_lead, phase, acc_active_2)[1] - assert dat.hex() == expected + ] + + def test_crz_ctrl_golden_bytes(self): + packer = _packer() + for long_active, acc_available, gap, has_lead, phase, acc_active_2, expected in self.CRZ_CTRL_GOLDEN_BYTES_CASES: + with self.subTest(long_active=long_active, acc_available=acc_available, gap=gap, has_lead=has_lead, + phase=phase, acc_active_2=acc_active_2, expected=expected): + dat = mazdacan.create_crz_ctrl(packer, 0, long_active, acc_available, gap, has_lead, phase, acc_active_2)[1] + self.assertEqual(dat.hex(), expected) def test_radar_frames_match_stock(self): expected = [ @@ -225,49 +263,51 @@ def test_radar_frames_match_stock(self): (0x366, "fff7fe7ffbff3fc0"), ] frames = mazdacan.create_radar_frames(0, 0, None) - assert [(f.address, f.dat.hex()) for f in frames] == expected + self.assertEqual([(f.address, f.dat.hex()) for f in frames], expected) def test_radar_frames_counter_and_lead_track(self): frames = mazdacan.create_radar_frames(2, 15, (10.25, 0.)) - assert all(f.src == 2 for f in frames) + self.assertTrue(all(f.src == 2 for f in frames)) # counter stamps the low nibble of the last byte on every track - assert [f.dat[7] & 0x0f for f in frames[1:]] == [15] * 6 + self.assertEqual([f.dat[7] & 0x0f for f in frames[1:]], [15] * 6) tracks = {f.address: f.dat.hex() for f in frames} - assert tracks[0x364] == "0a4e00001c00000f" + self.assertEqual(tracks[0x364], "0a4e00001c00000f") def test_lead_track_constant_bytes_match_the_stock_release_capture(self): # the template's measurement fields are zeroed, so a zero-range lead reproduces it exactly - assert mazdacan.create_lead_track(0., 0.) == mazdacan.LEAD_TRACK_TEMPLATE + self.assertEqual(mazdacan.create_lead_track(0., 0.), mazdacan.LEAD_TRACK_TEMPLATE) # the status pair the camera watches: drive_0b's occupied-slot 1c/00, never the # empty-slot c0 in byte 5 the old capture carried - assert mazdacan.LEAD_TRACK_TEMPLATE[4] & 0x1f == 0x1c - assert mazdacan.LEAD_TRACK_TEMPLATE[5] == 0x00 + self.assertEqual(mazdacan.LEAD_TRACK_TEMPLATE[4] & 0x1f, 0x1c) + self.assertEqual(mazdacan.LEAD_TRACK_TEMPLATE[5], 0x00) - @pytest.mark.parametrize("d_rel,v_rel", [ + LEAD_TRACK_ROUND_TRIPS_THROUGH_THE_DBC_CASES = [ (0., 0.), (6.5, 1.5), (10.25, -2.0), (29.4, 2.9375), (255.875, 63.9375), (400., 100.), (5., -80.), - ]) - def test_lead_track_round_trips_through_the_dbc(self, d_rel, v_rel): - dat = mazdacan.create_lead_track(d_rel, v_rel) - cp = CANParser("mazda_2017", [("RADAR_TRACK_364", float("nan"))], 0) - cp.update([(0, [(0x364, dat, 0)])]) - vl = cp.vl["RADAR_TRACK_364"] - assert vl["DIST_OBJ"] == pytest.approx(min(max(d_rel, 0.), 255.875), abs=0.0625) - assert vl["RELV_OBJ"] == pytest.approx(min(max(v_rel, -64.), 63.9375), abs=0.0625) - # the bits outside the two fields we drive stay exactly as captured - assert dat[1] & 0x0f == mazdacan.LEAD_TRACK_TEMPLATE[1] & 0x0f - assert dat[2] == mazdacan.LEAD_TRACK_TEMPLATE[2] - assert dat[4] & 0x1f == mazdacan.LEAD_TRACK_TEMPLATE[4] & 0x1f - assert dat[5:] == mazdacan.LEAD_TRACK_TEMPLATE[5:] - - -class TestStandstillHold: - - @pytest.fixture - def sm(self): - return StandstillHold() + ] + + def test_lead_track_round_trips_through_the_dbc(self): + for d_rel, v_rel in self.LEAD_TRACK_ROUND_TRIPS_THROUGH_THE_DBC_CASES: + with self.subTest(d_rel=d_rel, v_rel=v_rel): + dat = mazdacan.create_lead_track(d_rel, v_rel) + cp = CANParser("mazda_2017", [("RADAR_TRACK_364", float("nan"))], 0) + cp.update([(0, [(0x364, dat, 0)])]) + vl = cp.vl["RADAR_TRACK_364"] + self.assertEqual(vl["DIST_OBJ"], _Approx(min(max(d_rel, 0.), 255.875), 0.0625)) + self.assertEqual(vl["RELV_OBJ"], _Approx(min(max(v_rel, -64.), 63.9375), 0.0625)) + # the bits outside the two fields we drive stay exactly as captured + self.assertEqual(dat[1] & 0x0f, mazdacan.LEAD_TRACK_TEMPLATE[1] & 0x0f) + self.assertEqual(dat[2], mazdacan.LEAD_TRACK_TEMPLATE[2]) + self.assertEqual(dat[4] & 0x1f, mazdacan.LEAD_TRACK_TEMPLATE[4] & 0x1f) + self.assertEqual(dat[5:], mazdacan.LEAD_TRACK_TEMPLATE[5:]) + +def _sm(): + return StandstillHold() + + +class TestStandstillHold(unittest.TestCase): @staticmethod - def run(sm, frames, **kwargs): + def drive(sm, frames, **kwargs): defaults = dict(long_engaged=True, stopping=False, standstill=False, plan_accel=-1.024, brake_hold=False, gas_pressed=False) defaults.update(kwargs) @@ -275,241 +315,262 @@ def run(sm, frames, **kwargs): sm.update(**defaults) return sm - def test_holds_while_the_plan_is_stopping(self, sm): - self.run(sm, 1) - assert not sm.holding - self.run(sm, 1, stopping=True) - assert sm.holding and sm.stop_bits and sm.acc_active_2 + def test_holds_while_the_plan_is_stopping(self): + sm = _sm() + self.drive(sm, 1) + self.assertFalse(sm.holding) + self.drive(sm, 1, stopping=True) + self.assertTrue(sm.holding and sm.stop_bits and sm.acc_active_2) # arriving at a standstill changes nothing: the plan is still asking for the brakes - self.run(sm, 500, stopping=True, standstill=True) - assert sm.holding and sm.stop_bits + self.drive(sm, 500, stopping=True, standstill=True) + self.assertTrue(sm.holding and sm.stop_bits) - def test_hold_never_relaxes_on_its_own(self, sm): + def test_hold_never_relaxes_on_its_own(self): + sm = _sm() # the creep-into-the-lead regression: without the car taking the hold over, the command # must stay on the plan's brake no matter how long the stop lasts - self.run(sm, 1, stopping=True) - self.run(sm, int(30.0 / DT_CTRL), stopping=True, standstill=True) - assert sm.holding and sm.stop_bits and sm.acc_active_2 - assert not sm.car_has_hold - - def test_relax_follows_the_car_taking_the_hold(self, sm): - self.run(sm, 1, stopping=True) - self.run(sm, 10, stopping=True, standstill=True) - assert not sm.car_has_hold - self.run(sm, 1, stopping=True, standstill=True, brake_hold=True) + self.drive(sm, 1, stopping=True) + self.drive(sm, int(30.0 / DT_CTRL), stopping=True, standstill=True) + self.assertTrue(sm.holding and sm.stop_bits and sm.acc_active_2) + self.assertFalse(sm.car_has_hold) + + def test_relax_follows_the_car_taking_the_hold(self): + sm = _sm() + self.drive(sm, 1, stopping=True) + self.drive(sm, 10, stopping=True, standstill=True) + self.assertFalse(sm.car_has_hold) + self.drive(sm, 1, stopping=True, standstill=True, brake_hold=True) # stop bits and ACC_ACTIVE_2 drop with the command, together, exactly as stock does - assert sm.car_has_hold and not sm.stop_bits and not sm.acc_active_2 + self.assertTrue(sm.car_has_hold and not sm.stop_bits and not sm.acc_active_2) # and it is not a latch: if the car lets go, we brake again - self.run(sm, 1, stopping=True, standstill=True, brake_hold=False) - assert not sm.car_has_hold and sm.stop_bits and sm.acc_active_2 - - def test_released_when_the_plan_asks_to_move(self, sm): - self.run(sm, 1, stopping=True) - self.run(sm, 500, stopping=True, standstill=True, brake_hold=True) - assert sm.holding + self.drive(sm, 1, stopping=True, standstill=True, brake_hold=False) + self.assertTrue(not sm.car_has_hold and sm.stop_bits and sm.acc_active_2) + + def test_released_when_the_plan_asks_to_move(self): + sm = _sm() + self.drive(sm, 1, stopping=True) + self.drive(sm, 500, stopping=True, standstill=True, brake_hold=True) + self.assertTrue(sm.holding) # the release is debounced: a plan asking to move for less than the window changes nothing # (the body keeps its own latch until the pulse plays, so brake_hold stays up here) - self.run(sm, RELEASE_DEBOUNCE_FRAMES - 1, standstill=True, brake_hold=True, plan_accel=0.1) - assert sm.holding and not sm.resume_unlatching - self.run(sm, 1, standstill=True, brake_hold=True, plan_accel=0.1) - assert not sm.holding and not sm.car_has_hold + self.drive(sm, RELEASE_DEBOUNCE_FRAMES - 1, standstill=True, brake_hold=True, plan_accel=0.1) + self.assertTrue(sm.holding and not sm.resume_unlatching) + self.drive(sm, 1, standstill=True, brake_hold=True, plan_accel=0.1) + self.assertTrue(not sm.holding and not sm.car_has_hold) # the body owned the brakes, so this is the latched family: the pulse fires with the # release. The body answers nothing else -- deferring behind silence (route 0000011d) # and behind a positive nudge (route 0000012c) both just delayed the resume. - assert sm.latched_release and sm.resume_unlatching + self.assertTrue(sm.latched_release and sm.resume_unlatching) - def test_release_holds_for_as_long_as_the_plan_wants_to_move(self, sm): + def test_release_holds_for_as_long_as_the_plan_wants_to_move(self): + sm = _sm() # the failed-resume regression: no release window to run out from under the plan - self.run(sm, 1, stopping=True) - self.run(sm, 100, stopping=True, standstill=True) - self.run(sm, int(5.0 / DT_CTRL), standstill=True, plan_accel=0.4) - assert not sm.holding and not sm.stop_bits - - def test_hold_comes_back_if_the_plan_changes_its_mind(self, sm): - self.run(sm, 1, stopping=True) - self.run(sm, 100, stopping=True, standstill=True) - self.run(sm, RELEASE_DEBOUNCE_FRAMES, standstill=True, plan_accel=0.2) - assert not sm.holding + self.drive(sm, 1, stopping=True) + self.drive(sm, 100, stopping=True, standstill=True) + self.drive(sm, int(5.0 / DT_CTRL), standstill=True, plan_accel=0.4) + self.assertTrue(not sm.holding and not sm.stop_bits) + + def test_hold_comes_back_if_the_plan_changes_its_mind(self): + sm = _sm() + self.drive(sm, 1, stopping=True) + self.drive(sm, 100, stopping=True, standstill=True) + self.drive(sm, RELEASE_DEBOUNCE_FRAMES, standstill=True, plan_accel=0.2) + self.assertFalse(sm.holding) # nothing was latched, so this release emits no unlatch bit at all, deferred or otherwise - assert sm.unlatch_frames == 0 and not sm.resume_unlatching - self.run(sm, 1, stopping=True, standstill=True, plan_accel=-1.0) - assert sm.holding - assert not sm.resume_unlatching and sm.unlatch_frames == 0 - assert sm.stop_bits - - def test_never_latched_release_emits_no_pulse(self, sm): + self.assertTrue(sm.unlatch_frames == 0 and not sm.resume_unlatching) + self.drive(sm, 1, stopping=True, standstill=True, plan_accel=-1.0) + self.assertTrue(sm.holding) + self.assertTrue(not sm.resume_unlatching and sm.unlatch_frames == 0) + self.assertTrue(sm.stop_bits) + + def test_never_latched_release_emits_no_pulse(self): + sm = _sm() # a never-latched release has nothing latched to unlatch, so it puts no RESUME_UNLATCHING # on the wire at all. Stock blips here, but every pulse this port has emitted latched the # camera's SCBS fault (4/4), and mimicking a blip that unlatches nothing is not worth one - self.run(sm, 1, stopping=True) - self.run(sm, 100, stopping=True, standstill=True) - assert not sm.resume_unlatching - self.run(sm, RELEASE_DEBOUNCE_FRAMES, standstill=True, plan_accel=0.1) - assert not sm.holding and not sm.latched_release - self.run(sm, int(1.0 / DT_CTRL), standstill=True, plan_accel=0.1) - assert not sm.resume_unlatching and sm.unlatch_frames == 0 - - def test_latched_release_pulses_immediately_and_runs_its_length(self, sm): + self.drive(sm, 1, stopping=True) + self.drive(sm, 100, stopping=True, standstill=True) + self.assertFalse(sm.resume_unlatching) + self.drive(sm, RELEASE_DEBOUNCE_FRAMES, standstill=True, plan_accel=0.1) + self.assertTrue(not sm.holding and not sm.latched_release) + self.drive(sm, int(1.0 / DT_CTRL), standstill=True, plan_accel=0.1) + self.assertTrue(not sm.resume_unlatching and sm.unlatch_frames == 0) + + def test_latched_release_pulses_immediately_and_runs_its_length(self): + sm = _sm() # the pulse is the release protocol: the body ignores everything else (routes 0000011d # and 0000012c), so waiting only delays the resume. One pulse, stock's latched length. - self.run(sm, 1, stopping=True) - self.run(sm, 100, stopping=True, standstill=True, brake_hold=True) - self.run(sm, RELEASE_DEBOUNCE_FRAMES - 1, standstill=True, brake_hold=True, plan_accel=0.1) - assert not sm.resume_unlatching - self.run(sm, 1, standstill=True, brake_hold=True, plan_accel=0.1) - assert sm.resume_unlatching, "the pulse must fire with the release" - self.run(sm, RESUME_UNLATCH_LATCHED_FRAMES, standstill=True, brake_hold=True, plan_accel=0.1) - assert not sm.resume_unlatching, "pulse outran its length" - - def test_long_disengage_resets(self, sm): - self.run(sm, 1, stopping=True) - self.run(sm, 100, stopping=True, standstill=True, brake_hold=True) - self.run(sm, 1, long_engaged=False) - assert not sm.holding and not sm.car_has_hold and not sm.stop_bits - - def test_gas_override_drive_off_releases_the_hold(self, sm): + self.drive(sm, 1, stopping=True) + self.drive(sm, 100, stopping=True, standstill=True, brake_hold=True) + self.drive(sm, RELEASE_DEBOUNCE_FRAMES - 1, standstill=True, brake_hold=True, plan_accel=0.1) + self.assertFalse(sm.resume_unlatching) + self.drive(sm, 1, standstill=True, brake_hold=True, plan_accel=0.1) + self.assertTrue(sm.resume_unlatching, msg="the pulse must fire with the release") + self.drive(sm, RESUME_UNLATCH_LATCHED_FRAMES, standstill=True, brake_hold=True, plan_accel=0.1) + self.assertFalse(sm.resume_unlatching, msg="pulse outran its length") + + def test_long_disengage_resets(self): + sm = _sm() + self.drive(sm, 1, stopping=True) + self.drive(sm, 100, stopping=True, standstill=True, brake_hold=True) + self.drive(sm, 1, long_engaged=False) + self.assertTrue(not sm.holding and not sm.car_has_hold and not sm.stop_bits) + + def test_gas_override_drive_off_releases_the_hold(self): + sm = _sm() # a driver-gas drive-off under an override zeroes the plan's command, so the plan never # asks to move but the car does; the stop bits must not follow it up to speed. Stock keeps # STOPPING strictly to the final creep, below 0.55 m/s across all rolling frames. - self.run(sm, 1, stopping=True) - self.run(sm, 100, stopping=True, standstill=True) - assert sm.holding - self.run(sm, 1, plan_accel=0.0) - assert not sm.holding and not sm.stop_bits and not sm.resume_unlatching - - def test_stop_abort_releases(self, sm): - self.run(sm, 1, stopping=True) - assert sm.holding + self.drive(sm, 1, stopping=True) + self.drive(sm, 100, stopping=True, standstill=True) + self.assertTrue(sm.holding) + self.drive(sm, 1, plan_accel=0.0) + self.assertTrue(not sm.holding and not sm.stop_bits and not sm.resume_unlatching) + + def test_stop_abort_releases(self): + sm = _sm() + self.drive(sm, 1, stopping=True) + self.assertTrue(sm.holding) # lead speeds up again before the car reaches standstill - self.run(sm, 1, stopping=False, plan_accel=0.3) - assert not sm.holding + self.drive(sm, 1, stopping=False, plan_accel=0.3) + self.assertFalse(sm.holding) - def test_driver_gas_releases_the_hold_without_a_pulse(self, sm): + def test_driver_gas_releases_the_hold_without_a_pulse(self): + sm = _sm() # the driver's pedal outranks the hold, the way Toyota's PCM lets the pedal outrank its # standstill request -- but the pulse is the ACC's resume protocol, not the driver's: # stock's captured gas-ended hold drops the stop bits with no RESUME_UNLATCHING at all, # and pulsing there latched an SCBS fault (route 00000103 t+163.8) - self.run(sm, 1, stopping=True) - self.run(sm, 100, stopping=True, standstill=True) - assert sm.holding - self.run(sm, 1, stopping=True, standstill=True, gas_pressed=True) - assert not sm.holding and not sm.resume_unlatching, "gas release must not fire the ACC resume pulse" + self.drive(sm, 1, stopping=True) + self.drive(sm, 100, stopping=True, standstill=True) + self.assertTrue(sm.holding) + self.drive(sm, 1, stopping=True, standstill=True, gas_pressed=True) + self.assertTrue(not sm.holding and not sm.resume_unlatching, msg="gas release must not fire the ACC resume pulse") # no re-hold while the pedal is down, and a fresh hold once it lifts with the car stopped - self.run(sm, RESUME_UNLATCH_LATCHED_FRAMES + 5, stopping=True, standstill=True, gas_pressed=True) - assert not sm.holding and not sm.resume_unlatching - self.run(sm, 1, stopping=True, standstill=True) - assert sm.holding + self.drive(sm, RESUME_UNLATCH_LATCHED_FRAMES + 5, stopping=True, standstill=True, gas_pressed=True) + self.assertTrue(not sm.holding and not sm.resume_unlatching) + self.drive(sm, 1, stopping=True, standstill=True) + self.assertTrue(sm.holding) - def test_plan_flap_below_the_debounce_never_releases(self, sm): + def test_plan_flap_below_the_debounce_never_releases(self): + sm = _sm() # the SCBS-axis contamination shape: at a held standstill the lead inches forward and # stops, the plan flapping across zero. Sub-debounce flaps must not release at all, and # no frame may ever carry the stop bits and the release pulse together - self.run(sm, 1, stopping=True) - self.run(sm, 100, stopping=True, standstill=True) + self.drive(sm, 1, stopping=True) + self.drive(sm, 100, stopping=True, standstill=True) for i in range(600): accel = 0.3 if (i // 10) % 2 == 0 else -1.0 # 0.1 s swings, below the 0.2 s debounce sm.update(long_engaged=True, stopping=accel < 0, standstill=True, plan_accel=accel, brake_hold=False, gas_pressed=False) - assert not (sm.stop_bits and sm.resume_unlatching), "stop bits and pulse on one frame" - assert not sm.resume_unlatching, "a sub-debounce flap fired a release pulse" - assert sm.holding - - @pytest.mark.parametrize("brake_hold", [False, True]) - def test_slow_flap_never_mixes_stop_bits_with_the_pulse(self, sm, brake_hold): - # swings long enough to release each time. Nothing latched (brake_hold False) must never - # put an unlatch bit on the wire; a body that holds on through every swing falls back to - # at most one pulse per release, and a re-hold mid-pulse waits it out before re-asserting - # the stop bits - self.run(sm, 1, stopping=True) - self.run(sm, 100, stopping=True, standstill=True, brake_hold=brake_hold) - pulses = 0 - prev_unlatch = False - swing = RELEASE_DEBOUNCE_FRAMES + 30 # long enough for the release and its pulse to play - for i in range(1200): - accel = 0.3 if (i // swing) % 2 == 0 else -1.0 - sm.update(long_engaged=True, stopping=accel < 0, standstill=True, plan_accel=accel, - brake_hold=brake_hold, gas_pressed=False) - assert not (sm.stop_bits and sm.resume_unlatching), "stop bits and pulse on one frame" - pulses += int(sm.resume_unlatching and not prev_unlatch) - prev_unlatch = sm.resume_unlatching - if brake_hold: - assert pulses > 0 - assert pulses <= 1 + 1200 // (2 * swing), "more pulses than releases" - else: - assert pulses == 0, "a never-latched release put an unlatch bit on the wire" - - def test_latched_pulse_runs_to_completion_through_a_re_hold(self, sm): + self.assertFalse((sm.stop_bits and sm.resume_unlatching), msg="stop bits and pulse on one frame") + self.assertFalse(sm.resume_unlatching, msg="a sub-debounce flap fired a release pulse") + self.assertTrue(sm.holding) + + SLOW_FLAP_NEVER_MIXES_STOP_BITS_WITH_THE_PULSE_CASES = [False, True] + + def test_slow_flap_never_mixes_stop_bits_with_the_pulse(self): + sm = _sm() + for brake_hold in self.SLOW_FLAP_NEVER_MIXES_STOP_BITS_WITH_THE_PULSE_CASES: + with self.subTest(brake_hold=brake_hold): + # swings long enough to release each time. Nothing latched (brake_hold False) must never + # put an unlatch bit on the wire; a body that holds on through every swing falls back to + # at most one pulse per release, and a re-hold mid-pulse waits it out before re-asserting + # the stop bits + self.drive(sm, 1, stopping=True) + self.drive(sm, 100, stopping=True, standstill=True, brake_hold=brake_hold) + pulses = 0 + prev_unlatch = False + swing = RELEASE_DEBOUNCE_FRAMES + 30 # long enough for the release and its pulse to play + for i in range(1200): + accel = 0.3 if (i // swing) % 2 == 0 else -1.0 + sm.update(long_engaged=True, stopping=accel < 0, standstill=True, plan_accel=accel, + brake_hold=brake_hold, gas_pressed=False) + self.assertFalse((sm.stop_bits and sm.resume_unlatching), msg="stop bits and pulse on one frame") + pulses += int(sm.resume_unlatching and not prev_unlatch) + prev_unlatch = sm.resume_unlatching + if brake_hold: + self.assertGreater(pulses, 0) + self.assertLessEqual(pulses, 1 + 1200 // (2 * swing), msg="more pulses than releases") + else: + self.assertEqual(pulses, 0, msg="a never-latched release put an unlatch bit on the wire") + + def test_latched_pulse_runs_to_completion_through_a_re_hold(self): + sm = _sm() # a latched pulse spans the body's actual unlatch, so a re-hold mid-pulse waits it out # (stop bits blocked, stock never emits STOPPING with RESUME_UNLATCHING) instead of # canceling it; a second release cannot fire a fresh pulse before the first ends because # the release debounce is at least as long as any pulse window - assert RELEASE_DEBOUNCE_FRAMES >= RESUME_UNLATCH_LATCHED_FRAMES - self.run(sm, 1, stopping=True) - self.run(sm, 100, stopping=True, standstill=True, brake_hold=True) - self.run(sm, RELEASE_DEBOUNCE_FRAMES, standstill=True, brake_hold=True, plan_accel=0.3) - assert sm.latched_release and sm.resume_unlatching # the pulse fires with the release - self.run(sm, 3, standstill=True, plan_accel=0.3) - self.run(sm, 1, stopping=True, standstill=True) # re-hold mid-pulse, body already let go - assert sm.holding and not sm.stop_bits - assert sm.resume_unlatching, "a latched pulse mid-release must run to completion" - self.run(sm, RESUME_UNLATCH_LATCHED_FRAMES, stopping=True, standstill=True, plan_accel=-1.0) - assert sm.holding and sm.stop_bits and not sm.resume_unlatching - - -class TestAdvertisedLead: - """has_lead, the phase and the track slot are one decision, so they are asserted together.""" + self.assertGreaterEqual(RELEASE_DEBOUNCE_FRAMES, RESUME_UNLATCH_LATCHED_FRAMES) + self.drive(sm, 1, stopping=True) + self.drive(sm, 100, stopping=True, standstill=True, brake_hold=True) + self.drive(sm, RELEASE_DEBOUNCE_FRAMES, standstill=True, brake_hold=True, plan_accel=0.3) + self.assertTrue(sm.latched_release and sm.resume_unlatching) # the pulse fires with the release + self.drive(sm, 3, standstill=True, plan_accel=0.3) + self.drive(sm, 1, stopping=True, standstill=True) # re-hold mid-pulse, body already let go + self.assertTrue(sm.holding and not sm.stop_bits) + self.assertTrue(sm.resume_unlatching, msg="a latched pulse mid-release must run to completion") + self.drive(sm, RESUME_UNLATCH_LATCHED_FRAMES, stopping=True, standstill=True, plan_accel=-1.0) + self.assertTrue(sm.holding and sm.stop_bits and not sm.resume_unlatching) + + +def _al(): + return AdvertisedLead() - @pytest.fixture - def al(self): - return AdvertisedLead() + +class TestAdvertisedLead(unittest.TestCase): + """has_lead, the phase and the track slot are one decision, so they are asserted together.""" @staticmethod - def run(al, frames, **kwargs): + def drive(al, frames, **kwargs): defaults = dict(lead_visible=True, d_rel=40.0, v_rel=0.0, holding=False) defaults.update(kwargs) for _ in range(frames): al.update(**defaults) return al - def test_lead_follows_only_a_steady_state(self, al): + def test_lead_follows_only_a_steady_state(self): + al = _al() # a lead is adopted once leadVisible has held for the debounce window, not before - self.run(al, LEAD_DEBOUNCE_FRAMES - 1) - assert not al.has_lead and al.ctrl_phase == 0 - self.run(al, 1) - assert al.has_lead and al.lead == (40.0, 0.0) and al.ctrl_phase == 2 + self.drive(al, LEAD_DEBOUNCE_FRAMES - 1) + self.assertTrue(not al.has_lead and al.ctrl_phase == 0) + self.drive(al, 1) + self.assertTrue(al.has_lead and al.lead == (40.0, 0.0) and al.ctrl_phase == 2) # and dropped the same way - self.run(al, LEAD_DEBOUNCE_FRAMES - 1, lead_visible=False, d_rel=0.) - assert al.has_lead - self.run(al, 1, lead_visible=False, d_rel=0.) - assert not al.has_lead and al.ctrl_phase == 0 + self.drive(al, LEAD_DEBOUNCE_FRAMES - 1, lead_visible=False, d_rel=0.) + self.assertTrue(al.has_lead) + self.drive(al, 1, lead_visible=False, d_rel=0.) + self.assertTrue(not al.has_lead and al.ctrl_phase == 0) - def test_lead_flicker_never_reaches_the_bus(self, al): + def test_lead_flicker_never_reaches_the_bus(self): + al = _al() # the measured failure: a marginal 120 m vision lead toggled leadVisible 6 times in 1.4 s # (route 6bb2dc61c4 t+400); none of it may reach RADAR_HAS_LEAD or the track slot for frames, visible in ((15, True), (5, False), (7, True), (13, False), (10, True)): - self.run(al, frames, lead_visible=visible) - assert not al.has_lead, "a flickering lead leaked through the debounce" + self.drive(al, frames, lead_visible=visible) + self.assertFalse(al.has_lead, msg="a flickering lead leaked through the debounce") - def test_measurement_is_coasted_across_a_dropout(self, al): + def test_measurement_is_coasted_across_a_dropout(self): + al = _al() # leadOne goes to zero the instant vision drops the lead, well before the debounce expires. # Advertising a fabricated stand-in there put a stationary object 10.25 m dead ahead on the # bus at 22 m/s; the last real measurement carries the gap instead -- propagated by its own # range rate, never repeated frozen (a frozen range is the camera's proven SCBS trigger) - self.run(al, 2 * LEAD_DEBOUNCE_FRAMES, d_rel=120.0, v_rel=0.5) - assert al.lead == (120.0, 0.5) + self.drive(al, 2 * LEAD_DEBOUNCE_FRAMES, d_rel=120.0, v_rel=0.5) + self.assertEqual(al.lead, (120.0, 0.5)) coast_frames = LEAD_DEBOUNCE_FRAMES - 1 - self.run(al, coast_frames, lead_visible=False, d_rel=0., v_rel=0.) - assert al.lead is not None, "dropped the measurement inside the debounce window" + self.drive(al, coast_frames, lead_visible=False, d_rel=0., v_rel=0.) + self.assertIsNot(al.lead, None, msg="dropped the measurement inside the debounce window") d, v = al.lead - assert v == 0.5 - assert d == pytest.approx(120.0 + 0.5 * coast_frames * DT_CTRL, abs=1e-6), \ - "the coast must propagate the range, not freeze it" + self.assertEqual(v, 0.5) + self.assertEqual(d, _Approx(120.0 + 0.5 * coast_frames * DT_CTRL, 1e-6), msg="the coast must propagate the range, not freeze it") - def test_holding_reports_the_stop_phase_only_with_a_lead(self, al): - self.run(al, 2 * LEAD_DEBOUNCE_FRAMES, holding=True) - assert al.ctrl_phase == 3 - self.run(al, 2 * LEAD_DEBOUNCE_FRAMES, lead_visible=False, d_rel=0., holding=True) - assert not al.has_lead and al.ctrl_phase == 0 + def test_holding_reports_the_stop_phase_only_with_a_lead(self): + al = _al() + self.drive(al, 2 * LEAD_DEBOUNCE_FRAMES, holding=True) + self.assertEqual(al.ctrl_phase, 3) + self.drive(al, 2 * LEAD_DEBOUNCE_FRAMES, lead_visible=False, d_rel=0., holding=True) + self.assertTrue(not al.has_lead and al.ctrl_phase == 0) def _mock_cc(long_active=True, accel=0.5, long_state=None, standstill=False, gas=False, resume=False, cancel=False, lead_visible=True, gap=2, available=True, @@ -540,8 +601,7 @@ def _mock_cc(long_active=True, accel=0.5, long_state=None, standstill=False, gas return cc, cc_sp, cs -@pytest.fixture -def cc(): +def _cc(): CP = CarInterface.get_params(CAR.MAZDA_CX5_2022, {0: {}, 1: {}, 2: {}}, [], alpha_long=True, is_release=False, docs=False) CP_SP = CarInterface.get_params_sp(CP, CAR.MAZDA_CX5_2022, {0: {}, 1: {}, 2: {}}, [], True, False, False) @@ -549,8 +609,7 @@ def cc(): return CarController({Bus.pt: "mazda_2017"}, CP, CP_SP) -@pytest.fixture -def stock_cc(): +def _stock_cc(): CP = CarInterface.get_params(CAR.MAZDA_CX5_2022, {0: {}, 1: {}, 2: {}}, [], alpha_long=False, is_release=False, docs=False) CP_SP = CarInterface.get_params_sp(CP, CAR.MAZDA_CX5_2022, {0: {}, 1: {}, 2: {}}, [], False, False, False) @@ -614,11 +673,12 @@ def _step(cc, **kw): return sends -class TestLongitudinalIntegration: +class TestLongitudinalIntegration(unittest.TestCase): """Drives the real CarController.update_longitudinal through an engage -> cruise -> stop -> hold -> resume timeline and checks the emitted CAN, not just the state machine in isolation.""" - def test_engaged_frame_rates_and_counters(self, cc): + def test_engaged_frame_rates_and_counters(self): + cc = _cc() long = structs.CarControl.Actuators.LongControlState crz_info = crz_ctrl = radar_static = tester = 0 for _ in range(100): # 1 s at 100 Hz @@ -633,25 +693,27 @@ def test_engaged_frame_rates_and_counters(self, cc): tester += sum(1 for a, _, _ in sends if a == 0x764) # CRZ_INFO/CRZ_CTRL, when emitted, always go to both bus 0 and bus 2 if 0x21b in buses: - assert sorted(buses[0x21b]) == [0, 2] - assert sorted(buses[0x21c]) == [0, 2] + self.assertEqual(sorted(buses[0x21b]), [0, 2]) + self.assertEqual(sorted(buses[0x21c]), [0, 2]) # 100 Hz loop: long msgs at 50 Hz (x2 buses), radar at 10 Hz (x2), tester at 2 Hz - assert crz_info == crz_ctrl == 100 # 50 frames x 2 buses - assert radar_static == 20 # 10 frames x 2 buses - assert tester == 2 # 2 Hz, single bus - assert cc.long_counter == 50 and cc.radar_counter == 10 + self.assertTrue(crz_info == crz_ctrl == 100) # 50 frames x 2 buses + self.assertEqual(radar_static, 20) # 10 frames x 2 buses + self.assertEqual(tester, 2) # 2 Hz, single bus + self.assertTrue(cc.long_counter == 50 and cc.radar_counter == 10) - def test_gap_setting_mirrors_driver(self, cc): + def test_gap_setting_mirrors_driver(self): + cc = _cc() for gap in (1, 2, 3): cc.frame = 0 # force emission on the first step sends = _step(cc, gap=gap, long_state=structs.CarControl.Actuators.LongControlState.pid) ctrl = next(dat for a, dat, b in sends if a == 0x21c and b == 0) cp = CANParser("mazda_2017", [("CRZ_CTRL", float("nan"))], 0) cp.update([(0, [(0x21c, ctrl, 0)])]) - assert cp.vl["CRZ_CTRL"]["DISTANCE_SETTING"] == gap + self.assertEqual(cp.vl["CRZ_CTRL"]["DISTANCE_SETTING"], gap) - def test_stop_emits_hold_then_relaxes(self, cc): + def test_stop_emits_hold_then_relaxes(self): + cc = _cc() long = structs.CarControl.Actuators.LongControlState def accel_cmd(sends): @@ -669,7 +731,7 @@ def accel_cmd(sends): if cmd is not None: cmds.append(cmd) settled = cmds[len(cmds) // 2:] - assert settled and set(settled) == {-1024}, f"hold command drifted off the plan: {sorted(set(settled))}" + self.assertTrue(settled and set(settled) == {-1024}, msg=f"hold command drifted off the plan: {sorted(set(settled))}") # once the body ECU takes the hold over, stock stops asking for the brakes and so do we relaxed = [] @@ -678,9 +740,10 @@ def accel_cmd(sends): brake_hold=True)) if cmd is not None: relaxed.append(cmd) - assert relaxed and set(relaxed) == {round(CarControllerParams.ACCEL_HOLD_LATCHED * 1000)} + self.assertTrue(relaxed and set(relaxed) == {round(CarControllerParams.ACCEL_HOLD_LATCHED * 1000)}) - def test_gas_override_stays_engaged(self, cc): + def test_gas_override_stays_engaged(self): + cc = _cc() """A gas press is an override, not a disengagement. The command goes to zero as on every other port, but the engaged bits stay set the way Honda drives CONTROL_ON off CC.enabled. Clearing them mid-decel takes the PCM out of ACC mode (docs/mazda-gas-override.md).""" @@ -689,7 +752,7 @@ def test_gas_override_stays_engaged(self, cc): # braking hard, then the driver taps the gas for _ in range(200): _step(cc, long_state=long.pid, accel=-2.0, cruise_engaged=True) - assert cc.accel_last == pytest.approx(-2.0) + self.assertEqual(cc.accel_last, _Approx(-2.0, max(abs(-2.0) * 1e-6, 1e-12))) cmds = [] for _ in range(100): # 1 s of override @@ -700,23 +763,24 @@ def test_gas_override_stays_engaged(self, cc): cmds.append(frame) raw, acc_active, crz_active = zip(*cmds, strict=True) - assert all(acc_active), "ACC_ACTIVE dropped during a gas override" - assert all(crz_active), "CRZ_ACTIVE dropped during a gas override" - assert set(raw) == {0}, f"command should be zero through the override, got {sorted(set(raw))}" + self.assertTrue(all(acc_active), msg="ACC_ACTIVE dropped during a gas override") + self.assertTrue(all(crz_active), msg="CRZ_ACTIVE dropped during a gas override") + self.assertEqual(set(raw), {0}, msg=f"command should be zero through the override, got {sorted(set(raw))}") - def test_command_slew_is_rate_limited(self, cc): + def test_command_slew_is_rate_limited(self): + cc = _cc() """The plan can step; the wire should not. Windup is limited tightly because dumping the brake in one frame is what the driver feels, winddown loosely so braking is never delayed.""" long = structs.CarControl.Actuators.LongControlState for _ in range(200): _step(cc, long_state=long.pid, accel=-2.0, cruise_engaged=True) - assert cc.accel_last == pytest.approx(-2.0) + self.assertEqual(cc.accel_last, _Approx(-2.0, max(abs(-2.0) * 1e-6, 1e-12))) # plan jumps straight to +1.0: the command must ramp, not step prev = cc.accel_last for _ in range(5): _step(cc, long_state=long.pid, accel=1.0, cruise_engaged=True) - assert cc.accel_last - prev == pytest.approx(CarControllerParams.ACCEL_WINDUP_LIMIT, abs=1e-6) + self.assertEqual(cc.accel_last - prev, _Approx(CarControllerParams.ACCEL_WINDUP_LIMIT, 1e-6)) prev = cc.accel_last # and the other way, at the looser winddown limit @@ -725,10 +789,11 @@ def test_command_slew_is_rate_limited(self, cc): prev = cc.accel_last for _ in range(5): _step(cc, long_state=long.pid, accel=-3.0, cruise_engaged=True) - assert cc.accel_last - prev == pytest.approx(CarControllerParams.ACCEL_WINDDOWN_LIMIT, abs=1e-6) + self.assertEqual(cc.accel_last - prev, _Approx(CarControllerParams.ACCEL_WINDDOWN_LIMIT, 1e-6)) prev = cc.accel_last - def test_accel_last_tracks_the_wire_not_the_plan(self, cc): + def test_accel_last_tracks_the_wire_not_the_plan(self): + cc = _cc() # update() reports accel_last as actuatorsOutput.accel, the way Toyota, Ford and Honda # report the value they sent. It must be the wire value, clip and hold included. long = structs.CarControl.Actuators.LongControlState @@ -736,36 +801,38 @@ def test_accel_last_tracks_the_wire_not_the_plan(self, cc): # a plan beyond the envelope is reported clipped, not as asked for _ in range(400): sends = _step(cc, long_state=long.pid, accel=-9.0, cruise_engaged=True) - assert cc.accel_last == pytest.approx(CarControllerParams.ACCEL_MIN) + self.assertEqual(cc.accel_last, _Approx(CarControllerParams.ACCEL_MIN, max(abs(CarControllerParams.ACCEL_MIN) * 1e-6, 1e-12))) frame = _long_frames(sends) if frame is not None: - assert frame[0] == round(cc.accel_last * 1000) + self.assertEqual(frame[0], round(cc.accel_last * 1000)) # the standstill hold is the plan's own command, and that is what gets reported for _ in range(int(0.5 / 0.01)): _step(cc, long_state=long.stopping, accel=-1.5, standstill=True, cruise_engaged=True) - assert cc.accel_last == pytest.approx(-1.5) + self.assertEqual(cc.accel_last, _Approx(-1.5, max(abs(-1.5) * 1e-6, 1e-12))) # through a gas override we report the zero we actually send for _ in range(10): _step(cc, long_active=False, enabled=True, long_state=long.off, accel=0., gas=True, cruise_engaged=True) - assert cc.accel_last == 0. + self.assertEqual(cc.accel_last, 0.) - def test_gas_from_standstill_hold_releases_the_brake(self, cc): + def test_gas_from_standstill_hold_releases_the_brake(self): + cc = _cc() # gas out of a hold is a resume, not a slow release: the hold command must go straight to # zero rather than ramping off at the cruising override rate long = structs.CarControl.Actuators.LongControlState for _ in range(int(3.0 / 0.01)): _step(cc, long_state=long.stopping, accel=-1.5, standstill=True, cruise_engaged=True) - assert cc.accel_last < -0.5, "never reached the standstill hold" + self.assertLess(cc.accel_last, -0.5, msg="never reached the standstill hold") for _ in range(20): _step(cc, long_active=False, enabled=True, long_state=long.off, accel=0., gas=True, standstill=True, cruise_engaged=True) - assert cc.accel_last == 0., f"hold not released for the driver's gas: {cc.accel_last}" + self.assertEqual(cc.accel_last, 0., msg=f"hold not released for the driver's gas: {cc.accel_last}") - def test_release_command_holds_through_the_debounce_then_jumps(self, cc): + def test_release_command_holds_through_the_debounce_then_jumps(self): + cc = _cc() """Stock never lets ACCEL_CMD climb while STOPPING is asserted: through the release debounce the command stays at the hold value. Once the stop bits drop it relax-jumps into stock's release band and ramps. Pre-ramping toward the plan during the debounce put @@ -777,7 +844,7 @@ def test_release_command_holds_through_the_debounce_then_jumps(self, cc): _step(cc, long_state=long.stopping, accel=-1.5, standstill=False, **lead) for _ in range(int(3.0 / 0.01)): _step(cc, long_state=long.stopping, accel=-1.3, standstill=True, **lead) - assert cc.accel_last == pytest.approx(-1.3) + self.assertEqual(cc.accel_last, _Approx(-1.3, max(abs(-1.3) * 1e-6, 1e-12))) rows = [] for _ in range(int(1.5 / 0.01)): @@ -787,15 +854,16 @@ def test_release_command_holds_through_the_debounce_then_jumps(self, cc): rows.append((decode_accel_cmd_raw(dat), (dat[5] >> 2) & 1, (dat[6] >> 6) & 1)) debounce = [r for r in rows if r[1]] - assert debounce, "no stop-bit frames through the release debounce" - assert all(cmd == -1300 for cmd, _, _ in debounce), \ - f"command moved off the hold while STOPPING was asserted: {sorted({c for c, _, _ in debounce})}" + self.assertTrue(debounce, msg="no stop-bit frames through the release debounce") + self.assertTrue(all(cmd == -1300 for cmd, _, _ in debounce), + msg=f"command moved off the hold while STOPPING was asserted: {sorted({c for c, _, _ in debounce})}") # nothing was latched, so no unlatch bit goes out at all - assert not any(unl for _, _, unl in rows), "a never-latched release pulsed" - assert max(cmd for cmd, _, _ in rows) > 500, "command never ramped up after the release" + self.assertFalse(any(unl for _, _, unl in rows), msg="a never-latched release pulsed") + self.assertGreater(max(cmd for cmd, _, _ in rows), 500, msg="command never ramped up after the release") - def test_near_zero_hold_release_emits_no_pulse(self, cc): + def test_near_zero_hold_release_emits_no_pulse(self): + cc = _cc() # a no-lead hold relaxes the plan to ~0, so the release ramp would cross zero in the first # pulse frame -- the shape behind the routes 000000fe t+44.54 / 00000100 t+353.18 latches. # Nothing is latched here, so the release now carries no unlatch bit for it to land in. @@ -804,7 +872,7 @@ def test_near_zero_hold_release_emits_no_pulse(self, cc): _step(cc, long_state=long.stopping, accel=-0.5, standstill=False) for _ in range(int(2.0 / 0.01)): _step(cc, long_state=long.stopping, accel=-0.02, standstill=True) - assert cc.accel_last == pytest.approx(-0.02) + self.assertEqual(cc.accel_last, _Approx(-0.02, max(abs(-0.02) * 1e-6, 1e-12))) rows = [] for _ in range(int(1.5 / 0.01)): @@ -813,10 +881,11 @@ def test_near_zero_hold_release_emits_no_pulse(self, cc): if dat is not None: rows.append((decode_accel_cmd_raw(dat), (dat[6] >> 6) & 1)) - assert not any(unl for _, unl in rows), "a never-latched release pulsed" - assert max(cmd for cmd, _ in rows) > 500, "command never ramped up after the release" + self.assertFalse(any(unl for _, unl in rows), msg="a never-latched release pulsed") + self.assertGreater(max(cmd for cmd, _ in rows), 500, msg="command never ramped up after the release") - def test_release_keeps_climbing_until_the_car_actually_moves(self, cc): + def test_release_keeps_climbing_until_the_car_actually_moves(self): + cc = _cc() """Route 00000009--ad9e22f986 t+452.9 (EPS-swapped CX-9): the release ran correctly, the ramp caught the plan at +0.47 with a vision lead 2.5 m ahead, handed the command back -- and the car sat dead still for 1.5 s until the driver used the pedal. Mazda runs ki=0, so @@ -834,19 +903,20 @@ def test_release_keeps_climbing_until_the_car_actually_moves(self, cc): for _ in range(int(2.0 / 0.01)): _step(cc, long_state=long.pid, accel=0.47, standstill=True, **lead) peak = max(peak, cc.accel_last) - assert peak > 0.47 + 0.2, f"command plateaued at the plan and never asked harder: {peak:.2f}" - assert peak <= CarControllerParams.ACCEL_BREAKAWAY_MAX + 1e-6, f"climbed past the cap: {peak:.2f}" + self.assertGreater(peak, 0.47 + 0.2, msg=f"command plateaued at the plan and never asked harder: {peak:.2f}") + self.assertLessEqual(peak, CarControllerParams.ACCEL_BREAKAWAY_MAX + 1e-6, msg=f"climbed past the cap: {peak:.2f}") # the override sits on top of the plan, so it is bounded by what stock itself commands # pulling away from a stop: over all 31 stock stop->go episodes the breakaway command # spans +0.405..+1.425 (latched median +0.958), so this must not exceed stock's own worst - assert CarControllerParams.ACCEL_BREAKAWAY_MAX <= 1.45, "breakaway ceiling past stock's own max" + self.assertLessEqual(CarControllerParams.ACCEL_BREAKAWAY_MAX, 1.45, msg="breakaway ceiling past stock's own max") # once it moves, the plan owns the command again for _ in range(int(0.5 / 0.01)): _step(cc, long_state=long.pid, accel=0.47, standstill=False, **lead) - assert cc.accel_last == pytest.approx(0.47, abs=0.01) + self.assertEqual(cc.accel_last, _Approx(0.47, 0.01)) - def test_breakaway_gives_up_so_a_stuck_car_is_not_leaned_on(self, cc): + def test_breakaway_gives_up_so_a_stuck_car_is_not_leaned_on(self): + cc = _cc() # something we cannot see is holding the car (kerb, grade). Asking forever is worse than # settling back onto the plan, which the driver can then override with the pedal. long = structs.CarControl.Actuators.LongControlState @@ -854,10 +924,10 @@ def test_breakaway_gives_up_so_a_stuck_car_is_not_leaned_on(self, cc): _step(cc, long_state=long.stopping, accel=-1.024, standstill=True) for _ in range(BREAKAWAY_FRAMES + int(1.0 / 0.01)): _step(cc, long_state=long.pid, accel=0.3, standstill=True) - assert cc.accel_last == pytest.approx(0.3, abs=0.01), \ - f"still leaning on a car that never moved: {cc.accel_last:.2f}" + self.assertEqual(cc.accel_last, _Approx(0.3, 0.01), msg=f"still leaning on a car that never moved: {cc.accel_last:.2f}") - def test_breakaway_never_climbs_against_a_latched_body(self, cc): + def test_breakaway_never_climbs_against_a_latched_body(self): + cc = _cc() """The freeze that keeps the command pinned while GEAR.BRAKE_HOLD is still set (route 00000115 t+381.3, camera faulted 90 ms into the pulse) outranks the breakaway climb: a body-latched hold is pulsed, never leaned on.""" @@ -865,14 +935,15 @@ def test_breakaway_never_climbs_against_a_latched_body(self, cc): lead = dict(lead_visible=True, lead_d_rel=4.0, lead_v_rel=0.0) for _ in range(int(2.0 / 0.01)): _step(cc, long_state=long.stopping, accel=-1.3, standstill=True, brake_hold=True, **lead) - assert cc.stop_and_go.car_has_hold + self.assertTrue(cc.stop_and_go.car_has_hold) for _ in range(RESUME_UNLATCH_LATCHED_FRAMES + int(1.0 / 0.01)): _step(cc, long_state=long.pid, accel=0.5, standstill=True, brake_hold=True, **lead) - assert cc.accel_last <= CarControllerParams.ACCEL_RESUME_PULSE_MAX + 1e-6, \ - f"breakaway climbed against a still-latched body: {cc.accel_last:.2f}" + self.assertLessEqual(cc.accel_last, CarControllerParams.ACCEL_RESUME_PULSE_MAX + 1e-6, + msg=f"breakaway climbed against a still-latched body: {cc.accel_last:.2f}") - def test_never_latched_release_speaks_the_stock_wire_grammar(self, cc): + def test_never_latched_release_speaks_the_stock_wire_grammar(self): + cc = _cc() """Route 00000053 t+714.8 (second CX-5): slewing off the hold value under a 13-frame pulse put hold-grade braking beneath RESUME_UNLATCHING, a (stop, unlatch, cmd) tuple stock never emits, and the camera latched SCBS 90 ms in with a real departing lead advertised. Stock's @@ -886,7 +957,7 @@ def test_never_latched_release_speaks_the_stock_wire_grammar(self, cc): _step(cc, long_state=long.stopping, accel=-1.0, standstill=False, **lead) for _ in range(int(2.0 / 0.01)): _step(cc, long_state=long.stopping, accel=-1.024, standstill=True, **lead) - assert cc.accel_last == pytest.approx(-1.024) + self.assertEqual(cc.accel_last, _Approx(-1.024, max(abs(-1.024) * 1e-6, 1e-12))) rows = [] for _ in range(int(1.5 / 0.01)): @@ -895,63 +966,67 @@ def test_never_latched_release_speaks_the_stock_wire_grammar(self, cc): if dat is not None: rows.append((decode_accel_cmd_raw(dat), (dat[5] >> 2) & 1, (dat[6] >> 6) & 1)) - assert not any(stop and unl for _, stop, unl in rows), "stop bits and pulse on one frame" + self.assertFalse(any(stop and unl for _, stop, unl in rows), msg="stop bits and pulse on one frame") drop = next(i for i, (_, stop, _) in enumerate(rows) if not stop) post = rows[drop:] # the relax jump: no post-drop frame ever carries hold-grade braking again - assert all(cmd >= -280 for cmd, _, _ in post), f"command stayed at hold depth after the drop: {min(c for c, _, _ in post)}" - assert post[0][0] <= -180, f"release did not start inside the stock band: {post[0][0]}" - assert not any(unl for _, _, unl in rows), "a never-latched release pulsed" + self.assertTrue(all(cmd >= -280 for cmd, _, _ in post), msg=f"command stayed at hold depth after the drop: {min(c for c, _, _ in post)}") + self.assertLessEqual(post[0][0], -180, msg=f"release did not start inside the stock band: {post[0][0]}") + self.assertFalse(any(unl for _, _, unl in rows), msg="a never-latched release pulsed") # the ramp: stock's +25 raw per wire frame, straight through the drive-off ramping = [c for c, _, _ in post][:20] - assert all(20 <= b - a <= 30 for a, b in zip(ramping, ramping[1:], strict=False)), f"off the stock ramp: {ramping}" - - @pytest.mark.parametrize("drop_wire_frames", [1, 2, 3]) - def test_latched_release_speaks_the_stock_pulse_shape(self, cc, drop_wire_frames): - # a body-latched hold releases with a 9-wire-frame pulse: the command sits pinned at the - # relaxed -1 raw for as long as the body still reports GEAR.BRAKE_HOLD (as in every latched - # release of the census -- climbing before the drop faulted the camera 90 ms in, route - # 00000115 t+381.3), then climbs stock's ~+25 raw per frame ramp, peaking inside stock's - # family and never past the +0.25 ceiling (census: 6-11 frames, hold drop 1-3 frames in, - # cmd -1 climbing to +24..+342) - long = structs.CarControl.Actuators.LongControlState - lead = dict(lead_visible=True, lead_d_rel=4.0, lead_v_rel=0.0) - for _ in range(int(0.5 / 0.01)): - _step(cc, long_state=long.stopping, accel=-1.5, standstill=False, **lead) - for _ in range(int(2.0 / 0.01)): - _step(cc, long_state=long.stopping, accel=-1.3, standstill=True, brake_hold=True, **lead) - assert cc.accel_last == pytest.approx(CarControllerParams.ACCEL_HOLD_LATCHED) - - # the body reacts to the pulse: BRAKE_HOLD drops 1-3 wire frames after it starts (census) - rows = [] - pulse_started = None - window = RELEASE_DEBOUNCE_FRAMES + RESUME_UNLATCH_LATCHED_FRAMES + int(1.0 / 0.01) - for i in range(window): - body_holds = pulse_started is None or i < pulse_started + 2 * drop_wire_frames - sends = _step(cc, long_state=long.pid, accel=1.0, standstill=True, brake_hold=body_holds, **lead) - dat = next((d for a, d, b in sends if a == 0x21b and b == 0), None) - if dat is not None: - unl = (dat[6] >> 6) & 1 - rows.append((decode_accel_cmd_raw(dat), unl, body_holds)) - if pulse_started is None and unl: - pulse_started = i - - pulse = [(cmd, held) for cmd, unl, held in rows if unl] - cap = round(CarControllerParams.ACCEL_RESUME_PULSE_MAX * 1000) - assert len(pulse) == RESUME_UNLATCH_LATCHED_FRAMES // 2, f"pulse ran {len(pulse)} wire frames" - # the contract the route 115 fault turned on: no pulse frame moves off the relaxed hold - # while the body still reports its latch - pinned = [cmd for cmd, held in pulse if held] - assert len(pinned) == drop_wire_frames and all(cmd == -1 for cmd in pinned), \ - f"command moved under the latched hold: {pinned}" - # then the ramp: stock's +25 raw per wire frame from the relaxed hold - ramp = [cmd for cmd, held in pulse if not held] - assert -1 <= ramp[0] <= 15, f"ramp must start off the relaxed hold: {ramp[0]}" - assert all(20 <= b - a <= 30 for a, b in zip(ramp, ramp[1:], strict=False)), f"off the stock ramp: {ramp}" - assert -1 + (len(ramp) - 1) * 20 <= max(ramp) <= cap, f"in-pulse peak outside the ramp's own family: {max(ramp)}" - assert max(cmd for cmd, _, _ in rows) > cap, "command never ramped past the cap after the pulse" - - def test_latched_release_pulse_starts_at_the_release(self, cc): + self.assertTrue(all(20 <= b - a <= 30 for a, b in zip(ramping, ramping[1:], strict=False)), msg=f"off the stock ramp: {ramping}") + + LATCHED_RELEASE_SPEAKS_THE_STOCK_PULSE_SHAPE_CASES = [1, 2, 3] + + def test_latched_release_speaks_the_stock_pulse_shape(self): + cc = _cc() + for drop_wire_frames in self.LATCHED_RELEASE_SPEAKS_THE_STOCK_PULSE_SHAPE_CASES: + with self.subTest(drop_wire_frames=drop_wire_frames): + # a body-latched hold releases with a 9-wire-frame pulse: the command sits pinned at the + # relaxed -1 raw for as long as the body still reports GEAR.BRAKE_HOLD (as in every latched + # release of the census -- climbing before the drop faulted the camera 90 ms in, route + # 00000115 t+381.3), then climbs stock's ~+25 raw per frame ramp, peaking inside stock's + # family and never past the +0.25 ceiling (census: 6-11 frames, hold drop 1-3 frames in, + # cmd -1 climbing to +24..+342) + long = structs.CarControl.Actuators.LongControlState + lead = dict(lead_visible=True, lead_d_rel=4.0, lead_v_rel=0.0) + for _ in range(int(0.5 / 0.01)): + _step(cc, long_state=long.stopping, accel=-1.5, standstill=False, **lead) + for _ in range(int(2.0 / 0.01)): + _step(cc, long_state=long.stopping, accel=-1.3, standstill=True, brake_hold=True, **lead) + self.assertEqual(cc.accel_last, _Approx(CarControllerParams.ACCEL_HOLD_LATCHED, max(abs(CarControllerParams.ACCEL_HOLD_LATCHED) * 1e-6, 1e-12))) + + # the body reacts to the pulse: BRAKE_HOLD drops 1-3 wire frames after it starts (census) + rows = [] + pulse_started = None + window = RELEASE_DEBOUNCE_FRAMES + RESUME_UNLATCH_LATCHED_FRAMES + int(1.0 / 0.01) + for i in range(window): + body_holds = pulse_started is None or i < pulse_started + 2 * drop_wire_frames + sends = _step(cc, long_state=long.pid, accel=1.0, standstill=True, brake_hold=body_holds, **lead) + dat = next((d for a, d, b in sends if a == 0x21b and b == 0), None) + if dat is not None: + unl = (dat[6] >> 6) & 1 + rows.append((decode_accel_cmd_raw(dat), unl, body_holds)) + if pulse_started is None and unl: + pulse_started = i + + pulse = [(cmd, held) for cmd, unl, held in rows if unl] + cap = round(CarControllerParams.ACCEL_RESUME_PULSE_MAX * 1000) + self.assertEqual(len(pulse), RESUME_UNLATCH_LATCHED_FRAMES // 2, msg=f"pulse ran {len(pulse)} wire frames") + # the contract the route 115 fault turned on: no pulse frame moves off the relaxed hold + # while the body still reports its latch + pinned = [cmd for cmd, held in pulse if held] + self.assertTrue(len(pinned) == drop_wire_frames and all(cmd == -1 for cmd in pinned), msg=f"command moved under the latched hold: {pinned}") + # then the ramp: stock's +25 raw per wire frame from the relaxed hold + ramp = [cmd for cmd, held in pulse if not held] + self.assertTrue(-1 <= ramp[0] <= 15, msg=f"ramp must start off the relaxed hold: {ramp[0]}") + self.assertTrue(all(20 <= b - a <= 30 for a, b in zip(ramp, ramp[1:], strict=False)), msg=f"off the stock ramp: {ramp}") + self.assertTrue(-1 + (len(ramp) - 1) * 20 <= max(ramp) <= cap, msg=f"in-pulse peak outside the ramp's own family: {max(ramp)}") + self.assertGreater(max(cmd for cmd, _, _ in rows), cap, msg="command never ramped past the cap after the pulse") + + def test_latched_release_pulse_starts_at_the_release(self): + cc = _cc() """Routes 0000011d and 0000012c: the body ignores silence and a positive nudge alike, and answers the pulse within 2-3 wire frames. The pulse fires with the release, so the resume happens when the plan commands it.""" @@ -971,13 +1046,14 @@ def test_latched_release_pulse_starts_at_the_release(self, cc): # the stop bits are already down during a body-latched hold, so the pulse's deadline is # the debounce itself: it must start on the first wire frame after the plan's request lands - assert not any(stop for _, stop, _ in rows), "stop bits reappeared during a latched hold" + self.assertFalse(any(stop for _, stop, _ in rows), msg="stop bits reappeared during a latched hold") first_unl = next(i for i, (_, _, unl) in enumerate(rows) if unl) - assert first_unl <= RELEASE_DEBOUNCE_FRAMES // 2 + 1, \ - f"pulse lagged the release: {first_unl - RELEASE_DEBOUNCE_FRAMES // 2} wire frames late" - assert rows[first_unl][0] == -1, f"pulse did not start at the relaxed hold: {rows[first_unl][0]}" + self.assertLessEqual(first_unl, RELEASE_DEBOUNCE_FRAMES // 2 + 1, + msg=f"pulse lagged the release: {first_unl - RELEASE_DEBOUNCE_FRAMES // 2} wire frames late") + self.assertEqual(rows[first_unl][0], -1, msg=f"pulse did not start at the relaxed hold: {rows[first_unl][0]}") - def test_lead_track_follows_the_measured_lead(self, cc): + def test_lead_track_follows_the_measured_lead(self): + cc = _cc() # a frozen track is what latches the camera's SCBS fault, so the range we advertise has to # move with the lead we are actually following long = structs.CarControl.Actuators.LongControlState @@ -991,12 +1067,13 @@ def test_lead_track_follows_the_measured_lead(self, cc): track = next((d for a, d, b in sends if a == 0x364 and b == 0), None) if track is not None: seen.append(_lead_track(track)) - assert len(seen) > 1 + self.assertGreater(len(seen), 1) dists = [d for d, _ in seen] - assert all(a > b for a, b in zip(dists, dists[1:], strict=False)), f"range did not close with the lead: {dists}" - assert all(v == pytest.approx(-1.5, abs=0.0625) for _, v in seen) + self.assertTrue(all(a > b for a, b in zip(dists, dists[1:], strict=False)), msg=f"range did not close with the lead: {dists}") + self.assertTrue(all(abs(v - -1.5) <= 0.0625 for _, v in seen)) - def test_hold_with_nothing_ahead_advertises_nothing(self, cc): + def test_hold_with_nothing_ahead_advertises_nothing(self): + cc = _cc() # No fabricated object. The body does not decide the latch on the advertisement: across 32 # stock engaged standstills the radar said has_lead=1 in every one, yet 23 latched # GEAR.BRAKE_HOLD and 9 did not (one held 104 s), and 89 of 115 stock latches happened at @@ -1008,13 +1085,14 @@ def test_hold_with_nothing_ahead_advertises_nothing(self, cc): lead_visible=False, lead_d_rel=0.0, cruise_engaged=True) held += _frames(sends, 0x364) ctrls += _frames(sends, 0x21c) - assert held and ctrls - assert not any(map(_track_occupied, held)), "fabricated a lead for a hold with nothing ahead" - assert all(_crz_ctrl(d) == (0, 0) for d in ctrls), "advertised a lead with nothing in view" + self.assertTrue(held and ctrls) + self.assertFalse(any(map(_track_occupied, held)), msg="fabricated a lead for a hold with nothing ahead") + self.assertTrue(all(_crz_ctrl(d) == (0, 0) for d in ctrls), msg="advertised a lead with nothing in view") # and the hold itself is untouched: the plan's brake and the stop bits still go out - assert cc.stop_and_go.holding and cc.stop_and_go.stop_bits + self.assertTrue(cc.stop_and_go.holding and cc.stop_and_go.stop_bits) - def test_vision_lead_dropout_does_not_fabricate_a_lead_at_speed(self, cc): + def test_vision_lead_dropout_does_not_fabricate_a_lead_at_speed(self): + cc = _cc() # leadOne goes to zero the instant the vision lead drops while sm.lead_visible is still # latched. Falling through to the hold fallback there put a stationary object 10.25 m dead # ahead on the bus at 22 m/s, 20 times across the two 2026-08-25 drives. @@ -1027,12 +1105,13 @@ def test_vision_lead_dropout_does_not_fabricate_a_lead_at_speed(self, cc): for _ in range(int(LEAD_DEBOUNCE_FRAMES * 0.8)): # inside the debounce window dropped += _frames(_step(cc, long_state=long.pid, accel=0.5, lead_visible=False, lead_d_rel=0.0, lead_v_rel=0.0, cruise_engaged=True), 0x364) - assert dropped + self.assertTrue(dropped) for d in dropped: dist = _lead_track(d)[0] - assert dist == pytest.approx(120.0, abs=1.0), f"track teleported to {dist} m" + self.assertEqual(dist, _Approx(120.0, 1.0), msg=f"track teleported to {dist} m") - def test_has_lead_phase_and_track_never_disagree(self, cc): + def test_has_lead_phase_and_track_never_disagree(self): + cc = _cc() # stock pairs all three absolutely: has_lead=0 <=> phase=0, and RADAR_HAS_LEAD=1 with all six # slots empty appears 8 times in 1,095,826 stock samples. We shipped has_lead=0 with phase=1 # for 22-84% of every engaged drive before this was derived from one decision. @@ -1051,10 +1130,11 @@ def test_has_lead_phase_and_track_never_disagree(self, cc): if trk is None or ctl is None: continue has_lead, phase = _crz_ctrl(ctl) - assert bool(has_lead) == _track_occupied(trk), f"has_lead/track disagree for {kw}" - assert (phase == 0) == (has_lead == 0), f"has_lead/phase disagree for {kw}" + self.assertEqual(bool(has_lead), _track_occupied(trk), msg=f"has_lead/track disagree for {kw}") + self.assertEqual((phase == 0), (has_lead == 0), msg=f"has_lead/phase disagree for {kw}") - def test_lead_survives_disengagement(self, cc): + def test_lead_survives_disengagement(self): + cc = _cc() # perception is engagement-independent: stock advertises RADAR_HAS_LEAD=1 with cruise off in # 19.5% of all frames. Dropping the advertisement at disengage made a real car 4.5 m ahead # vanish from the bus in one frame while the driver braked toward it, and the camera ran its @@ -1069,72 +1149,77 @@ def test_lead_survives_disengagement(self, cc): if ctl is None: continue has_lead, phase = _crz_ctrl(ctl) - assert has_lead == 1 and phase != 0, "disengaging dropped a real lead off the bus" + self.assertTrue(has_lead == 1 and phase != 0, msg="disengaging dropped a real lead off the bus") if trk is not None: - assert _lead_track(trk)[0] == pytest.approx(4.8, abs=0.1) + self.assertEqual(_lead_track(trk)[0], _Approx(4.8, 0.1)) - def test_no_resume_button_while_openpilot_owns_longitudinal(self, cc): + def test_no_resume_button_while_openpilot_owns_longitudinal(self): + cc = _cc() # We are the ACC here, so the hold is released in-protocol. The car's own MRCC never presses # RES either: 0 of 23 stock body-latched-hold releases put one on the bus. A press would also # put a second writer on CRZ_BTNS, which ICBM owns. for accel in (0.3, -1.024): for standstill in (True, False): control, _, _ = _mock_cc(standstill=standstill, accel=accel, resume=True) - assert not cc.resume_requested(control) + self.assertFalse(cc.resume_requested(control)) - def test_resume_button_still_sent_with_stock_longitudinal(self, stock_cc): + def test_resume_button_still_sent_with_stock_longitudinal(self): + stock_cc = _stock_cc() # stock ACC owns the hold there, and the button is the only lever openpilot has on it control, _, _ = _mock_cc(standstill=True, accel=0.3, resume=True) - assert stock_cc.resume_requested(control) + self.assertTrue(stock_cc.resume_requested(control)) control, _, _ = _mock_cc(standstill=True, accel=0.3, resume=False) - assert not stock_cc.resume_requested(control) + self.assertFalse(stock_cc.resume_requested(control)) - def test_body_latched_hold_releases_in_protocol(self, cc): + def test_body_latched_hold_releases_in_protocol(self): + cc = _cc() # the release the button used to stand in for: stop bits already relaxed to the body, then # the plan asks to move and the unlatch pulse fires with the release long = structs.CarControl.Actuators.LongControlState for _ in range(200): _step(cc, long_state=long.stopping, accel=-1.024, standstill=True, cruise_engaged=True, brake_hold=True) - assert cc.stop_and_go.holding and cc.stop_and_go.car_has_hold - assert not cc.stop_and_go.stop_bits # body owns the brakes, stock relaxes here + self.assertTrue(cc.stop_and_go.holding and cc.stop_and_go.car_has_hold) + self.assertFalse(cc.stop_and_go.stop_bits) # body owns the brakes, stock relaxes here for _ in range(RELEASE_DEBOUNCE_FRAMES): sends = _step(cc, long_state=long.pid, accel=0.3, standstill=True, cruise_engaged=True, brake_hold=True) - assert not any(a == CRZ_BTNS for a, _, _ in sends), "CRZ_BTNS written at the release" - assert not cc.stop_and_go.holding - assert cc.stop_and_go.resume_unlatching, "the pulse must fire with the release" + self.assertFalse(any(a == CRZ_BTNS for a, _, _ in sends), msg="CRZ_BTNS written at the release") + self.assertFalse(cc.stop_and_go.holding) + self.assertTrue(cc.stop_and_go.resume_unlatching, msg="the pulse must fire with the release") - def test_gas_pedal_without_cruise_stays_disengaged(self, cc): + def test_gas_pedal_without_cruise_stays_disengaged(self): + cc = _cc() # gas pressed while openpilot is not enabled must not advertise an engaged ACC off = structs.CarControl.Actuators.LongControlState.off cc.frame = 0 sends = _step(cc, long_active=False, enabled=False, long_state=off, gas=True, available=True) info = next(dat for a, dat, b in sends if a == 0x21b and b == 0) - assert info.hex().startswith("01ffe3ffc480") # armed-but-idle pattern, command pegged + self.assertTrue(info.hex().startswith("01ffe3ffc480")) # armed-but-idle pattern, command pegged - def test_disengaged_emits_stock_patterns(self, cc): + def test_disengaged_emits_stock_patterns(self): + cc = _cc() off = structs.CarControl.Actuators.LongControlState.off # main off, not available: the exact standby pattern the panda allowlists byte-for-byte cc.frame = 0 sends = _step(cc, long_active=False, long_state=off, available=False) info = next(dat for a, dat, b in sends if a == 0x21b and b == 0) - assert info.hex().startswith("01ffe3ffc000") + self.assertTrue(info.hex().startswith("01ffe3ffc000")) # MRCC armed but not engaged: the command stays pegged and ACC_SET_ALLOWED follows the # brake, exactly the two patterns stock alternates between at an armed idle cc.frame = 0 sends = _step(cc, long_active=False, long_state=off, available=True) info = next(dat for a, dat, b in sends if a == 0x21b and b == 0) - assert info.hex().startswith("01ffe3ffc480") + self.assertTrue(info.hex().startswith("01ffe3ffc480")) cc.frame = 0 sends = _step(cc, long_active=False, long_state=off, available=True, brake_pressed=True) info = next(dat for a, dat, b in sends if a == 0x21b and b == 0) - assert info.hex().startswith("01ffe3ffc080") + self.assertTrue(info.hex().startswith("01ffe3ffc080")) -class TestCancelCarveOut: +class TestCancelCarveOut(unittest.TestCase): """controlsd raises cruiseControl.cancel whenever cruiseState.enabled has no matching CC.enabled (mazda reports pcmCruise). While the stock radar still owns the bus that engagement is the driver's own stock MRCC and a CANCEL turns its main off within ~100 ms, @@ -1151,20 +1236,23 @@ def _full_update(self, cc, cancel, radar_was_silenced, stock_radar_alive): _, sends = cc.update(control, control_sp, carstate, 0) return [a for a, _, _ in sends] - def test_no_cancel_while_the_radar_is_stock(self, cc): + def test_no_cancel_while_the_radar_is_stock(self): + cc = _cc() # pre-teardown settle window, and equally the silencing-failed drive: a driver SET is # their own stock MRCC and must be left alone addrs = self._full_update(cc, cancel=True, radar_was_silenced=False, stock_radar_alive=True) - assert CRZ_BTNS not in addrs, "CANCELed the driver's own stock MRCC" + self.assertTrue(CRZ_BTNS not in addrs, msg="CANCELed the driver's own stock MRCC") - def test_cancel_still_sent_after_the_teardown(self, cc): + def test_cancel_still_sent_after_the_teardown(self): + cc = _cc() # post-teardown a stock engagement is impossible: cancel keeps handling state desync addrs = self._full_update(cc, cancel=True, radar_was_silenced=True, stock_radar_alive=False) - assert CRZ_BTNS in addrs + self.assertTrue(CRZ_BTNS in addrs) - def test_stock_longitudinal_cancel_unaffected(self, stock_cc): + def test_stock_longitudinal_cancel_unaffected(self): + stock_cc = _stock_cc() addrs = self._full_update(stock_cc, cancel=True, radar_was_silenced=False, stock_radar_alive=True) - assert CRZ_BTNS in addrs + self.assertTrue(CRZ_BTNS in addrs) SESSION_PROG_DAT = bytes([0x02, 0x10, 0x02, 0, 0, 0, 0, 0]) @@ -1172,7 +1260,7 @@ def test_stock_longitudinal_cancel_unaffected(self, stock_cc): TESTER_PRESENT_DAT = bytes([0x02, 0x3e, 0x80, 0, 0, 0, 0, 0]) -class TestRadarSessionBounds: +class TestRadarSessionBounds(unittest.TestCase): """The fire-and-forget UDS session has no readable NRC, so every episode is bounded the way disable_ecu bounds its retries.""" @@ -1180,27 +1268,27 @@ def test_silencing_gives_up_bounded(self): m = RadarSessionManager() for _ in range(RADAR_SESSION_LIMIT_FRAMES + 2): state = m.update(True, True, False, standstill=True, session_refused=False) - assert state == RadarSessionState.STOCK and m.silencing_failed + self.assertTrue(state == RadarSessionState.STOCK and m.silencing_failed) # and stays given up for the drive: stock keeps the bus for _ in range(10): - assert m.update(True, True, False, standstill=True, session_refused=False) == RadarSessionState.STOCK + self.assertEqual(m.update(True, True, False, standstill=True, session_refused=False), RadarSessionState.STOCK) def test_negative_response_gives_up_immediately(self): # route 000000fe t+15.0 shows the radar answers a session request within 10 ms, so a # negative response is definitive: no reason to burn the silence budget m = RadarSessionManager() m.update(True, True, False, standstill=True, session_refused=False) - assert m.state == RadarSessionState.SILENCING - assert m.update(True, True, False, standstill=True, session_refused=True) == RadarSessionState.STOCK - assert m.silencing_failed + self.assertEqual(m.state, RadarSessionState.SILENCING) + self.assertEqual(m.update(True, True, False, standstill=True, session_refused=True), RadarSessionState.STOCK) + self.assertTrue(m.silencing_failed) def test_handback_stops_waiting_for_a_dead_radar(self): m = RadarSessionManager() m.update(True, False, False, standstill=True, session_refused=False) - assert m.state == RadarSessionState.SILENCED + self.assertEqual(m.state, RadarSessionState.SILENCED) for _ in range(RADAR_SESSION_LIMIT_FRAMES + 2): state = m.update(True, False, True, standstill=True, session_refused=False) - assert state == RadarSessionState.STOCK + self.assertEqual(state, RadarSessionState.STOCK) def test_completed_handback_never_resilences(self): # the parked toggle-off regression: the monitor's CC_SP assert used to drop after its done @@ -1209,14 +1297,14 @@ def test_completed_handback_never_resilences(self): # back, right before shutdown, leaving it to a degraded unattended S3 recovery m = RadarSessionManager() m.update(True, False, False, standstill=True, session_refused=False) - assert m.state == RadarSessionState.SILENCED + self.assertEqual(m.state, RadarSessionState.SILENCED) m.update(True, False, True, standstill=True, session_refused=False) - assert m.state == RadarSessionState.HANDBACK - assert m.update(True, True, True, standstill=True, session_refused=False) == RadarSessionState.STOCK + self.assertEqual(m.state, RadarSessionState.HANDBACK) + self.assertEqual(m.update(True, True, True, standstill=True, session_refused=False), RadarSessionState.STOCK) for handback in (True, False): for alive in (True, False): for _ in range(5): - assert m.update(True, alive, handback, standstill=True, session_refused=False) == RadarSessionState.STOCK + self.assertEqual(m.update(True, alive, handback, standstill=True, session_refused=False), RadarSessionState.STOCK) def test_withdrawn_handback_allows_retakeover(self): # only a hand-back that ran to completion latches: a genuine toggle-flip-back @@ -1224,34 +1312,35 @@ def test_withdrawn_handback_allows_retakeover(self): m = RadarSessionManager() m.update(True, False, False, standstill=True, session_refused=False) m.update(True, False, True, standstill=True, session_refused=False) - assert m.state == RadarSessionState.HANDBACK + self.assertEqual(m.state, RadarSessionState.HANDBACK) state = m.update(True, False, False, standstill=True, session_refused=False) - assert state == RadarSessionState.SILENCED and not m.handback_completed + self.assertTrue(state == RadarSessionState.SILENCED and not m.handback_completed) def test_silencing_waits_for_standstill_but_adoption_does_not(self): # actively silencing disables AEB, so it only starts pre-motion like disable_ecu; # adopting an already-quiet radar disables nothing and proceeds anywhere m = RadarSessionManager() for _ in range(10): - assert m.update(True, True, False, standstill=False, session_refused=False) == RadarSessionState.STOCK - assert m.update(True, True, False, standstill=True, session_refused=False) == RadarSessionState.SILENCING + self.assertEqual(m.update(True, True, False, standstill=False, session_refused=False), RadarSessionState.STOCK) + self.assertEqual(m.update(True, True, False, standstill=True, session_refused=False), RadarSessionState.SILENCING) m2 = RadarSessionManager() - assert m2.update(True, False, False, standstill=False, session_refused=False) == RadarSessionState.SILENCED + self.assertEqual(m2.update(True, False, False, standstill=False, session_refused=False), RadarSessionState.SILENCED) -def test_non_gen1_platform_refused_at_admission(): - # one init-time check instead of per-frame guards in the message builders, which every - # frame layout in mazdacan assumes; the fall-throughs used to emit an all-zero CAM_LKAS - # and return None from the button builder, straight into can_sends - CP = CarInterface.get_params(CAR.MAZDA_CX5_2022, {0: {}, 1: {}, 2: {}}, [], alpha_long=False, - is_release=False, docs=False) - CP_SP = CarInterface.get_params_sp(CP, CAR.MAZDA_CX5_2022, {0: {}, 1: {}, 2: {}}, [], False, False, False) - CP.flags = 0 - with pytest.raises(NotImplementedError): - CarController({Bus.pt: "mazda_2017"}, CP, CP_SP) +class TestPlatformAdmission(unittest.TestCase): + def test_non_gen1_platform_refused_at_admission(self): + # one init-time check instead of per-frame guards in the message builders, which every + # frame layout in mazdacan assumes; the fall-throughs used to emit an all-zero CAM_LKAS + # and return None from the button builder, straight into can_sends + CP = CarInterface.get_params(CAR.MAZDA_CX5_2022, {0: {}, 1: {}, 2: {}}, [], alpha_long=False, + is_release=False, docs=False) + CP_SP = CarInterface.get_params_sp(CP, CAR.MAZDA_CX5_2022, {0: {}, 1: {}, 2: {}}, [], False, False, False) + CP.flags = 0 + with self.assertRaises(NotImplementedError): + CarController({Bus.pt: "mazda_2017"}, CP, CP_SP) -class TestRadarSessionSequencing: +class TestRadarSessionSequencing(unittest.TestCase): """Boot teardown deferral and the ordered hand-back: what goes on the bus in each radar session state, driven through the real CarController.update_longitudinal.""" @@ -1270,34 +1359,37 @@ def _uds(sends): def _synthetic(sends): return [a for a, _, _ in sends if a in (0x21b, 0x21c, 0x499)] - def test_stock_state_is_silent(self, cc): + def test_stock_state_is_silent(self): + cc = _cc() # radar alive, gate not yet passed: nothing at all goes on the bus for _ in range(200): sends = self._step(cc, stock_radar_alive=True, fsc_settled=False) - assert sends == [] + self.assertEqual(sends, []) - def test_boot_teardown_sequence(self, cc): + def test_boot_teardown_sequence(self): + cc = _cc() # gate passes with the stock radar alive: programming-session requests at 2 Hz, # still no synthetic frames and no tester present for i in range(100): sends = self._step(cc, stock_radar_alive=True, fsc_settled=True) if i % CarControllerParams.RADAR_UDS_STEP == 0: - assert self._uds(sends) == [SESSION_PROG_DAT] + self.assertEqual(self._uds(sends), [SESSION_PROG_DAT]) else: - assert self._uds(sends) == [] - assert self._synthetic(sends) == [] + self.assertEqual(self._uds(sends), []) + self.assertEqual(self._synthetic(sends), []) # radar goes quiet: synthetic frames + tester present take over, session requests stop saw_tester = False for _ in range(100): frame = cc.frame sends = self._step(cc, stock_radar_alive=False, fsc_settled=True) - assert SESSION_PROG_DAT not in self._uds(sends) + self.assertTrue(SESSION_PROG_DAT not in self._uds(sends)) if frame % CarControllerParams.LONG_STEP == 0: - assert len(self._synthetic(sends)) > 0 + self.assertGreater(len(self._synthetic(sends)), 0) saw_tester |= TESTER_PRESENT_DAT in self._uds(sends) - assert saw_tester + self.assertTrue(saw_tester) - def test_handback_sequence(self, cc): + def test_handback_sequence(self): + cc = _cc() # reach SILENCED self._step(cc, stock_radar_alive=False, fsc_settled=True) # hand-back requested: default-session requests at 2 Hz, tester present stops, @@ -1306,35 +1398,38 @@ def test_handback_sequence(self, cc): for _ in range(100): frame = cc.frame sends = self._step(cc, stock_radar_alive=False, fsc_settled=True, handback=True) - assert TESTER_PRESENT_DAT not in self._uds(sends) + self.assertTrue(TESTER_PRESENT_DAT not in self._uds(sends)) saw_default |= SESSION_DFLT_DAT in self._uds(sends) if frame % CarControllerParams.LONG_STEP == 0: - assert len(self._synthetic(sends)) > 0 - assert saw_default + self.assertGreater(len(self._synthetic(sends)), 0) + self.assertTrue(saw_default) # stock radar returns: everything stops for _ in range(200): sends = self._step(cc, stock_radar_alive=True, fsc_settled=True, handback=True) - assert sends == [] + self.assertEqual(sends, []) - def test_handback_before_teardown_stops_everything(self, cc): + def test_handback_before_teardown_stops_everything(self): + cc = _cc() # toggle-off while still waiting on the gate: no session ever entered, so no # hand-back traffic either self._step(cc, stock_radar_alive=True, fsc_settled=False) for _ in range(120): sends = self._step(cc, stock_radar_alive=True, fsc_settled=False, handback=True) - assert sends == [] + self.assertEqual(sends, []) - def test_teardown_waits_for_stock_cruise_disengage(self, cc): + def test_teardown_waits_for_stock_cruise_disengage(self): + cc = _cc() # driver engaged stock MRCC before the gate passed (warm boot): hold the teardown for _ in range(120): sends = self._step(cc, stock_radar_alive=True, fsc_settled=True, cruise_engaged=True) - assert sends == [] + self.assertEqual(sends, []) # driver disengages: teardown proceeds cc.frame = 0 sends = self._step(cc, stock_radar_alive=True, fsc_settled=True, cruise_engaged=False) - assert SESSION_PROG_DAT in self._uds(sends) + self.assertTrue(SESSION_PROG_DAT in self._uds(sends)) - def test_completed_handback_stays_stock_after_the_assert_drops(self, cc): + def test_completed_handback_stays_stock_after_the_assert_drops(self): + cc = _cc() # CC_SP is rebuilt every frame, so once the toggle monitor's done latch stops asserting # the hand-back the manager sees handback=False; a completed hand-back must not turn # into a fresh takeover on the very next frame (parked => standstill, gate still passed) @@ -1343,14 +1438,15 @@ def test_completed_handback_stays_stock_after_the_assert_drops(self, cc): self._step(cc, stock_radar_alive=True, fsc_settled=True, handback=True) for _ in range(200): sends = self._step(cc, stock_radar_alive=True, fsc_settled=True, handback=False) - assert sends == [] + self.assertEqual(sends, []) - def test_s3_recovery_resilences(self, cc): + def test_s3_recovery_resilences(self): + cc = _cc() # radar reappears mid-drive (dropped tester present, S3 timeout): re-request the session self._step(cc, stock_radar_alive=False, fsc_settled=True) cc.frame = CarControllerParams.RADAR_UDS_STEP # align to a session-request frame sends = self._step(cc, stock_radar_alive=True, fsc_settled=True) - assert SESSION_PROG_DAT in self._uds(sends) + self.assertTrue(SESSION_PROG_DAT in self._uds(sends)) # and settles back to silenced once quiet again sends = self._step(cc, stock_radar_alive=False, fsc_settled=True) - assert SESSION_PROG_DAT not in self._uds(sends) + self.assertTrue(SESSION_PROG_DAT not in self._uds(sends)) diff --git a/opendbc/car/mazda/tests/test_mazda_interface.py b/opendbc/car/mazda/tests/test_mazda_interface.py index 7a0dd25c09a..950a27d28ff 100644 --- a/opendbc/car/mazda/tests/test_mazda_interface.py +++ b/opendbc/car/mazda/tests/test_mazda_interface.py @@ -1,4 +1,4 @@ -import pytest +import unittest from opendbc.car import structs from opendbc.car.common.conversions import Conversions as CV @@ -26,7 +26,7 @@ def _params(candidate, car_fw=None, alpha_long=False): alpha_long, is_release=False, docs=False) -class TestMazdaEpsSwap: +class TestMazdaEpsSwap(unittest.TestCase): """A 2022+ CX-5 EPS swapped into an older Mazda brings the EPS-derived behavior with it. Pre-2022 Mazdas are dashcam only because their EPS locks steering out after ~5 s hands-off @@ -36,47 +36,48 @@ class TestMazdaEpsSwap: def test_stock_older_mazda_is_dashcam_only(self): CP = _params(CAR.MAZDA_CX5, _eps_fw(STOCK_CX5_EPS_FW)) - assert CP.dashcamOnly - assert CP.minSteerSpeed == pytest.approx(LKAS_LIMITS.DISABLE_SPEED * CV.KPH_TO_MS) - assert CP.steerActuatorDelay == pytest.approx(0.1) + self.assertTrue(CP.dashcamOnly) + self.assertAlmostEqual(CP.minSteerSpeed, LKAS_LIMITS.DISABLE_SPEED * CV.KPH_TO_MS) + self.assertAlmostEqual(CP.steerActuatorDelay, 0.1) def test_swapped_eps_lifts_dashcam_and_the_speed_floor(self): CP = _params(CAR.MAZDA_CX5, _eps_fw(SWAPPED_EPS_FW)) - assert not CP.dashcamOnly - assert CP.minSteerSpeed == 0 - assert CP.steerActuatorDelay == pytest.approx(0.14) + self.assertFalse(CP.dashcamOnly) + self.assertEqual(CP.minSteerSpeed, 0) + self.assertAlmostEqual(CP.steerActuatorDelay, 0.14) def test_swapped_eps_does_not_unlock_longitudinal(self): # the radar and camera are not part of an EPS swap, and this car keeps its own pre-2022 pair CP = _params(CAR.MAZDA_CX5, _eps_fw(SWAPPED_EPS_FW), alpha_long=True) - assert not CP.alphaLongitudinalAvailable - assert not CP.openpilotLongitudinalControl + self.assertFalse(CP.alphaLongitudinalAvailable) + self.assertFalse(CP.openpilotLongitudinalControl) def test_swapped_eps_keeps_the_real_vehicle_specs(self): # the whole point of fixing detection is that the user no longer forces MAZDA_CX5_2022 and # inherits its mass, steer ratio and tire stiffness swapped = _params(CAR.MAZDA_CX5, _eps_fw(SWAPPED_EPS_FW)) cx5_2022 = _params(CAR.MAZDA_CX5_2022) - assert swapped.mass != cx5_2022.mass - assert swapped.steerRatio != cx5_2022.steerRatio - assert swapped.tireStiffnessFactor != cx5_2022.tireStiffnessFactor + self.assertNotEqual(swapped.mass, cx5_2022.mass) + self.assertNotEqual(swapped.steerRatio, cx5_2022.steerRatio) + self.assertNotEqual(swapped.tireStiffnessFactor, cx5_2022.tireStiffnessFactor) def test_supported_platforms_are_unchanged(self): cx5_2022 = _params(CAR.MAZDA_CX5_2022) - assert not cx5_2022.dashcamOnly - assert cx5_2022.minSteerSpeed == 0 - assert cx5_2022.steerActuatorDelay == pytest.approx(0.14) - assert cx5_2022.alphaLongitudinalAvailable + self.assertFalse(cx5_2022.dashcamOnly) + self.assertEqual(cx5_2022.minSteerSpeed, 0) + self.assertAlmostEqual(cx5_2022.steerActuatorDelay, 0.14) + self.assertTrue(cx5_2022.alphaLongitudinalAvailable) # the CX-9 2021 is supported without the CX-5 EPS, so it keeps the 45 kph floor cx9_2021 = _params(CAR.MAZDA_CX9_2021) - assert not cx9_2021.dashcamOnly - assert cx9_2021.minSteerSpeed == pytest.approx(LKAS_LIMITS.DISABLE_SPEED * CV.KPH_TO_MS) - assert cx9_2021.steerActuatorDelay == pytest.approx(0.1) + self.assertFalse(cx9_2021.dashcamOnly) + self.assertAlmostEqual(cx9_2021.minSteerSpeed, LKAS_LIMITS.DISABLE_SPEED * CV.KPH_TO_MS) + self.assertAlmostEqual(cx9_2021.steerActuatorDelay, 0.1) def test_docs_are_generated_without_firmware(self): # car_fw is empty when building CARS.md, so the docs must keep advertising dashcam mode for candidate in (CAR.MAZDA_CX5, CAR.MAZDA_CX9, CAR.MAZDA_3, CAR.MAZDA_6): - CP = CarInterface.get_params(candidate, {0: {}, 1: {}, 2: {}}, [], False, - is_release=False, docs=True) - assert CP.dashcamOnly, candidate + with self.subTest(candidate=candidate): + CP = CarInterface.get_params(candidate, {0: {}, 1: {}, 2: {}}, [], False, + is_release=False, docs=True) + self.assertTrue(CP.dashcamOnly) diff --git a/opendbc/car/mazda/tests/test_mazda_radar.py b/opendbc/car/mazda/tests/test_mazda_radar.py index 1e2baf6f3dc..9f7b43d7946 100644 --- a/opendbc/car/mazda/tests/test_mazda_radar.py +++ b/opendbc/car/mazda/tests/test_mazda_radar.py @@ -12,8 +12,7 @@ or not at all, except real tracks at exactly -1.0 m/s. """ import math - -import pytest +import unittest from opendbc.can import CANPacker from opendbc.car import gen_empty_fingerprint @@ -49,19 +48,20 @@ def burst(ri, t_ns, overrides=None): return rr -class TestRadarSentinels: +class TestRadarSentinels(unittest.TestCase): def test_empty_slots_produce_no_points(self): ri = _radar_interface() rr = burst(ri, 0) - assert len(rr.points) == 0 + self.assertEqual(len(rr.points), 0) def test_real_track_parses(self): ri = _radar_interface() rr = burst(ri, 0, {0x361: track_msg(0x361, dist=40.0, ang=2.0, relv=-5.0)}) - assert len(rr.points) == 1 + self.assertEqual(len(rr.points), 1) pt = rr.points[0] - assert pt.dRel == pytest.approx(math.cos(math.radians(2.0)) * 40.0, rel=1e-6) - assert pt.vRel == -5.0 + expected_drel = math.cos(math.radians(2.0)) * 40.0 + self.assertAlmostEqual(pt.dRel, expected_drel, delta=abs(expected_drel) * 1e-6) + self.assertEqual(pt.vRel, -5.0) def test_track_at_exactly_minus_one_mps_is_kept(self): """-1.0 m/s with a real distance is a live lead, not an empty slot. Dropping it @@ -73,34 +73,35 @@ def test_track_at_exactly_minus_one_mps_is_kept(self): # lead decelerates through exactly -1.0 m/s: the track must survive with its identity rr = burst(ri, int(0.1e9), {0x361: track_msg(0x361, dist=39.875, ang=0.0, relv=SENTINEL_RELV)}) - assert len(rr.points) == 1 - assert rr.points[0].vRel == SENTINEL_RELV - assert rr.points[0].trackId == track_id + self.assertEqual(len(rr.points), 1) + self.assertEqual(rr.points[0].vRel, SENTINEL_RELV) + self.assertEqual(rr.points[0].trackId, track_id) rr = burst(ri, int(0.2e9), {0x361: track_msg(0x361, dist=39.75, ang=0.0, relv=-0.9375)}) - assert rr.points[0].trackId == track_id + self.assertEqual(rr.points[0].trackId, track_id) def test_track_at_max_range_is_kept(self): ri = _radar_interface() rr = burst(ri, 0, {0x362: track_msg(0x362, dist=SENTINEL_DIST, ang=0.0, relv=-10.0)}) - assert len(rr.points) == 1 - assert rr.points[0].dRel == SENTINEL_DIST + self.assertEqual(len(rr.points), 1) + self.assertEqual(rr.points[0].dRel, SENTINEL_DIST) def test_track_at_sentinel_angle_is_kept(self): ri = _radar_interface() rr = burst(ri, 0, {0x363: track_msg(0x363, dist=40.0, ang=SENTINEL_ANG, relv=-5.0)}) - assert len(rr.points) == 1 - assert rr.points[0].yRel == pytest.approx(-math.sin(math.radians(SENTINEL_ANG)) * 40.0, rel=1e-6) + self.assertEqual(len(rr.points), 1) + expected_yrel = -math.sin(math.radians(SENTINEL_ANG)) * 40.0 + self.assertAlmostEqual(rr.points[0].yRel, expected_yrel, delta=abs(expected_yrel) * 1e-6) def test_all_sentinel_slot_deletes_a_prior_track(self): ri = _radar_interface() burst(ri, 0, {0x361: track_msg(0x361, dist=40.0, ang=0.0, relv=-5.0)}) rr = burst(ri, int(0.1e9)) # slot empties: all three sentinels - assert len(rr.points) == 0 + self.assertEqual(len(rr.points), 0) def test_undecoded_relv_addrs_never_produce_points(self): ri = _radar_interface() overrides = {addr: track_msg(addr, dist=40.0, ang=0.0, relv=-5.0) for addr in RADAR_TRACK_ADDRS if addr not in RADAR_USABLE_ADDRS} rr = burst(ri, 0, overrides) - assert len(rr.points) == 0 + self.assertEqual(len(rr.points), 0) diff --git a/opendbc/sunnypilot/car/tests/test_speed_dep_config.py b/opendbc/sunnypilot/car/tests/test_speed_dep_config.py index 536f125d8e5..a9d821df93d 100644 --- a/opendbc/sunnypilot/car/tests/test_speed_dep_config.py +++ b/opendbc/sunnypilot/car/tests/test_speed_dep_config.py @@ -4,10 +4,10 @@ This file is part of zoompilot and is licensed under the MIT License. See the LICENSE.md file in the root directory for more details. """ -import pytest +import unittest from opendbc.car.structs import CarParams -from opendbc.sunnypilot.car.interfaces import get_speed_dep_config_for_car, get_steer_max_schedule +from opendbc.sunnypilot.car.interfaces import get_speed_dep_config_for_car, get_steer_max_schedule, get_steer_rail_schedule def _cx5_cp(): @@ -18,57 +18,56 @@ def _cx5_cp(): return cp -class TestSteerMaxSchedule: +class TestSteerMaxSchedule(unittest.TestCase): def test_mazda_steer_to_zero_returns_lookup(self): schedule = get_steer_max_schedule(_cx5_cp()) - assert schedule == ([0.0, 14.2, 14.5], [1200.0, 1200.0, 800.0]) + self.assertEqual(schedule, ([0.0, 14.2, 14.5], [1200.0, 1200.0, 800.0])) def test_flat_steer_max_brand_returns_none(self): cp = CarParams() cp.carFingerprint = 'TOYOTA_RAV4_TSS2' cp.brand = 'toyota' - assert get_steer_max_schedule(cp) is None + self.assertIsNone(get_steer_max_schedule(cp)) def test_unknown_brand_returns_none(self): cp = CarParams() cp.brand = 'notabrand' - assert get_steer_max_schedule(cp) is None + self.assertIsNone(get_steer_max_schedule(cp)) def test_schedule_attached_to_active_entry(self): cfg = get_speed_dep_config_for_car(_cx5_cp()) - assert cfg['steer_max_schedule'] == ([0.0, 14.2, 14.5], [1200.0, 1200.0, 800.0]) + self.assertEqual(cfg['steer_max_schedule'], ([0.0, 14.2, 14.5], [1200.0, 1200.0, 800.0])) # the schedule's step must sit inside a bin span, or per-count interp cannot place it - assert cfg['speed_bp'][2] < 14.2 < 14.5 < cfg['speed_bp'][3] + self.assertLess(cfg['speed_bp'][2], 14.2) + self.assertLess(14.5, cfg['speed_bp'][3]) def test_inactive_entry_stays_empty(self): cp = CarParams() cp.carFingerprint = 'MAZDA_CX9_2021' cp.brand = 'mazda' cp.minSteerSpeed = 20.0 # stock EPS, requires_steer_to_zero suppresses the entry - assert get_speed_dep_config_for_car(cp) == {} + self.assertEqual(get_speed_dep_config_for_car(cp), {}) def test_config_copy_not_cached_dict(self): a = get_speed_dep_config_for_car(_cx5_cp()) a['steer_max_schedule'] = 'mutated' - assert get_speed_dep_config_for_car(_cx5_cp())['steer_max_schedule'] != 'mutated' + self.assertNotEqual(get_speed_dep_config_for_car(_cx5_cp())['steer_max_schedule'], 'mutated') -class TestSteerRailSchedule: +class TestSteerRailSchedule(unittest.TestCase): def test_mazda_steer_to_zero_rail(self): - from opendbc.sunnypilot.car.interfaces import get_steer_rail_schedule bp, rail = get_steer_rail_schedule(_cx5_cp()) - assert all(0.0 < r <= 1.0 for r in rail) + self.assertTrue(all(0.0 < r <= 1.0 for r in rail)) # the rail bottoms out just below the cliff (648/1200) and recovers above it (620/800) - assert rail[bp.index(14.2)] == min(rail) - assert rail[bp.index(14.2)] == pytest.approx(0.54, abs=0.01) - assert rail[bp.index(14.5)] == pytest.approx(620.0 / 800.0, abs=1e-6) + self.assertEqual(rail[bp.index(14.2)], min(rail)) + self.assertAlmostEqual(rail[bp.index(14.2)], 0.54, delta=0.01) + self.assertAlmostEqual(rail[bp.index(14.5)], 620.0 / 800.0, delta=1e-6) def test_no_ceiling_returns_none(self): - from opendbc.sunnypilot.car.interfaces import get_steer_rail_schedule cp = CarParams() cp.brand = 'mazda' cp.minSteerSpeed = 20.0 # stock EPS params have no ceiling lookup - assert get_steer_rail_schedule(cp) is None + self.assertIsNone(get_steer_rail_schedule(cp)) cp2 = CarParams() cp2.brand = 'toyota' - assert get_steer_rail_schedule(cp2) is None + self.assertIsNone(get_steer_rail_schedule(cp2))