sadmen
/
openpilot_comma
mirror of https://github.com/commaai/openpilot.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

wait to leave elm mode before checking relay malfunction

Browse Source
pull/29759/head
Shane Smiskol 2 years ago
parent 9bef110ca7
commit f02b4622a7
1 changed files with 168 additions and 152 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 320
      selfdrive/car/tests/test_models.py

320
selfdrive/car/tests/test_models.py
Unescape View File

@ -90,6 +90,8 @@ class TestCarModelBase(unittest.TestCase):
test_segs = (2, 1, 0)
if cls.test_route.segment is not None:
test_segs = (cls.test_route.segment,)
cls.segment_num = None
no_elm = False
for seg in test_segs:
try:
@ -110,7 +112,20 @@ class TestCarModelBase(unittest.TestCase):
enabled_toggle = True
dashcam_only = False
for msg in lr:
if msg.which() == "can":
if msg.which() == 'pandaStateDEPREATED':
if msg.pandaState.safetyModel.raw != 3:
no_elm = True
elif msg.which() == 'pandaStates':
for ps in msg.pandaStates:
print(ps.safetyModel.raw)
if ps.safetyModel.raw != 3:
no_elm = True
break
elif msg.which() == "can":
if not no_elm:
continue
can_msgs.append(msg)
if len(can_msgs) <= FRAME_FINGERPRINT:
for m in msg.can:
@ -131,6 +146,7 @@ class TestCarModelBase(unittest.TestCase):
enabled_toggle = param.value.strip(b'\x00') == b'1'
if len(can_msgs) > int(50 / DT_CTRL):
cls.segment_num = seg
break
else:
raise Exception(f"Route: {repr(cls.test_route.route)} with segments: {test_segs} not found or no CAN msgs found. Is it uploaded?")
@ -161,42 +177,42 @@ class TestCarModelBase(unittest.TestCase):
self.assertEqual(0, set_status, f"failed to set safetyModel {cfg}")
self.safety.init_tests()
def test_car_params(self):
if self.CP.dashcamOnly:
self.skipTest("no need to check carParams for dashcamOnly")
# make sure car params are within a valid range
self.assertGreater(self.CP.mass, 1)
if self.CP.steerControlType != car.CarParams.SteerControlType.angle:
tuning = self.CP.lateralTuning.which()
if tuning == 'pid':
self.assertTrue(len(self.CP.lateralTuning.pid.kpV))
elif tuning == 'torque':
self.assertTrue(self.CP.lateralTuning.torque.kf > 0)
elif tuning == 'indi':
self.assertTrue(len(self.CP.lateralTuning.indi.outerLoopGainV))
else:
raise Exception("unknown tuning")
def test_car_interface(self):
# TODO: also check for checksum violations from can parser
can_invalid_cnt = 0
can_valid = False
CC = car.CarControl.new_message()
for i, msg in enumerate(self.can_msgs):
CS = self.CI.update(CC, (msg.as_builder().to_bytes(),))
self.CI.apply(CC, msg.logMonoTime)
if CS.canValid:
can_valid = True
# wait max of 2s for low frequency msgs to be seen
if i > 200 or can_valid:
can_invalid_cnt += not CS.canValid
self.assertEqual(can_invalid_cnt, 0)
# def test_car_params(self):
# if self.CP.dashcamOnly:
# self.skipTest("no need to check carParams for dashcamOnly")
#
# # make sure car params are within a valid range
# self.assertGreater(self.CP.mass, 1)
#
# if self.CP.steerControlType != car.CarParams.SteerControlType.angle:
# tuning = self.CP.lateralTuning.which()
# if tuning == 'pid':
# self.assertTrue(len(self.CP.lateralTuning.pid.kpV))
# elif tuning == 'torque':
# self.assertTrue(self.CP.lateralTuning.torque.kf > 0)
# elif tuning == 'indi':
# self.assertTrue(len(self.CP.lateralTuning.indi.outerLoopGainV))
# else:
# raise Exception("unknown tuning")
#
# def test_car_interface(self):
# # TODO: also check for checksum violations from can parser
# can_invalid_cnt = 0
# can_valid = False
# CC = car.CarControl.new_message()
#
# for i, msg in enumerate(self.can_msgs):
# CS = self.CI.update(CC, (msg.as_builder().to_bytes(),))
# self.CI.apply(CC, msg.logMonoTime)
#
# if CS.canValid:
# can_valid = True
#
# # wait max of 2s for low frequency msgs to be seen
# if i > 200 or can_valid:
# can_invalid_cnt += not CS.canValid
#
# self.assertEqual(can_invalid_cnt, 0)
def test_radar_interface(self):
os.environ['NO_RADAR_SLEEP'] = "1"
@ -207,7 +223,7 @@ class TestCarModelBase(unittest.TestCase):
error_cnt = 0
for i, msg in enumerate(self.can_msgs):
rr = RI.update((msg.as_builder().to_bytes(),))
if rr is not None and i > 50:
if rr is not None and (i > 50 and self.segment_num != 0 or i > 500 and self.segment_num == 0):
error_cnt += car.RadarData.Error.canError in rr.errors
self.assertEqual(error_cnt, 0)
@ -240,126 +256,126 @@ class TestCarModelBase(unittest.TestCase):
# No need to check relay malfunction on disabled routes (relay closed) or for reasonable fingerprinting time
# TODO: detect when relay has flipped to properly check relay malfunction
if self.openpilot_enabled and t > 5e6:
if self.openpilot_enabled:# and t > 5e6:
self.assertFalse(self.safety.get_relay_malfunction())
else:
self.safety.set_relay_malfunction(False)
self.assertFalse(len(failed_addrs), f"panda safety RX check failed: {failed_addrs}")
def test_panda_safety_tx_cases(self, data=None):
"""Asserts we can tx common messages"""
if self.CP.notCar:
self.skipTest("Skipping test for notCar")
def test_car_controller(car_control):
now_nanos = 0
msgs_sent = 0
CI = self.CarInterface(self.CP, self.CarController, self.CarState)
for _ in range(round(10.0 / DT_CTRL)): # make sure we hit the slowest messages
CI.update(car_control, [])
_, sendcan = CI.apply(car_control, now_nanos)
now_nanos += DT_CTRL * 1e9
msgs_sent += len(sendcan)
for addr, _, dat, bus in sendcan:
to_send = libpanda_py.make_CANPacket(addr, bus % 4, dat)
self.assertTrue(self.safety.safety_tx_hook(to_send), (addr, dat, bus))
# Make sure we attempted to send messages
self.assertGreater(msgs_sent, 50)
# Make sure we can send all messages while inactive
CC = car.CarControl.new_message()
test_car_controller(CC)
# Test cancel + general messages (controls_allowed=False & cruise_engaged=True)
self.safety.set_cruise_engaged_prev(True)
CC = car.CarControl.new_message(cruiseControl={'cancel': True})
test_car_controller(CC)
# Test resume + general messages (controls_allowed=True & cruise_engaged=True)
self.safety.set_controls_allowed(True)
CC = car.CarControl.new_message(cruiseControl={'resume': True})
test_car_controller(CC)
def test_panda_safety_carstate(self):
"""
Assert that panda safety matches openpilot's carState
"""
if self.CP.dashcamOnly:
self.skipTest("no need to check panda safety for dashcamOnly")
CC = car.CarControl.new_message()
# warm up pass, as initial states may be different
for can in self.can_msgs[:300]:
self.CI.update(CC, (can.as_builder().to_bytes(), ))
for msg in filter(lambda m: m.src in range(64), can.can):
to_send = libpanda_py.make_CANPacket(msg.address, msg.src % 4, msg.dat)
self.safety.safety_rx_hook(to_send)
controls_allowed_prev = False
CS_prev = car.CarState.new_message()
checks = defaultdict(lambda: 0)
for idx, can in enumerate(self.can_msgs):
CS = self.CI.update(CC, (can.as_builder().to_bytes(), ))
for msg in filter(lambda m: m.src in range(64), can.can):
to_send = libpanda_py.make_CANPacket(msg.address, msg.src % 4, msg.dat)
ret = self.safety.safety_rx_hook(to_send)
self.assertEqual(1, ret, f"safety rx failed ({ret=}): {to_send}")
# Skip first frame so CS_prev is properly initialized
if idx == 0:
CS_prev = CS
# Button may be left pressed in warm up period
if not self.CP.pcmCruise:
self.safety.set_controls_allowed(0)
continue
# TODO: check rest of panda's carstate (steering, ACC main on, etc.)
checks['gasPressed'] += CS.gasPressed != self.safety.get_gas_pressed_prev()
checks['standstill'] += CS.standstill == self.safety.get_vehicle_moving()
# TODO: remove this exception once this mismatch is resolved
brake_pressed = CS.brakePressed
if CS.brakePressed and not self.safety.get_brake_pressed_prev():
if self.CP.carFingerprint in (HONDA.PILOT, HONDA.RIDGELINE) and CS.brake > 0.05:
brake_pressed = False
checks['brakePressed'] += brake_pressed != self.safety.get_brake_pressed_prev()
checks['regenBraking'] += CS.regenBraking != self.safety.get_regen_braking_prev()
if self.CP.pcmCruise:
# On most pcmCruise cars, openpilot's state is always tied to the PCM's cruise state.
# On Honda Nidec, we always engage on the rising edge of the PCM cruise state, but
# openpilot brakes to zero even if the min ACC speed is non-zero (i.e. the PCM disengages).
if self.CP.carName == "honda" and self.CP.carFingerprint not in HONDA_BOSCH:
# only the rising edges are expected to match
if CS.cruiseState.enabled and not CS_prev.cruiseState.enabled:
checks['controlsAllowed'] += not self.safety.get_controls_allowed()
else:
checks['controlsAllowed'] += not CS.cruiseState.enabled and self.safety.get_controls_allowed()
# TODO: fix notCar mismatch
if not self.CP.notCar:
checks['cruiseState'] += CS.cruiseState.enabled != self.safety.get_cruise_engaged_prev()
else:
# Check for enable events on rising edge of controls allowed
button_enable = any(evt.enable for evt in CS.events)
mismatch = button_enable != (self.safety.get_controls_allowed() and not controls_allowed_prev)
checks['controlsAllowed'] += mismatch
controls_allowed_prev = self.safety.get_controls_allowed()
if button_enable and not mismatch:
self.safety.set_controls_allowed(False)
if self.CP.carName == "honda":
checks['mainOn'] += CS.cruiseState.available != self.safety.get_acc_main_on()
CS_prev = CS
failed_checks = {k: v for k, v in checks.items() if v > 0}
self.assertFalse(len(failed_checks), f"panda safety doesn't agree with openpilot: {failed_checks}")
# def test_panda_safety_tx_cases(self, data=None):
# """Asserts we can tx common messages"""
# if self.CP.notCar:
# self.skipTest("Skipping test for notCar")
#
# def test_car_controller(car_control):
# now_nanos = 0
# msgs_sent = 0
# CI = self.CarInterface(self.CP, self.CarController, self.CarState)
# for _ in range(round(10.0 / DT_CTRL)): # make sure we hit the slowest messages
# CI.update(car_control, [])
# _, sendcan = CI.apply(car_control, now_nanos)
#
# now_nanos += DT_CTRL * 1e9
# msgs_sent += len(sendcan)
# for addr, _, dat, bus in sendcan:
# to_send = libpanda_py.make_CANPacket(addr, bus % 4, dat)
# self.assertTrue(self.safety.safety_tx_hook(to_send), (addr, dat, bus))
#
# # Make sure we attempted to send messages
# self.assertGreater(msgs_sent, 50)
#
# # Make sure we can send all messages while inactive
# CC = car.CarControl.new_message()
# test_car_controller(CC)
#
# # Test cancel + general messages (controls_allowed=False & cruise_engaged=True)
# self.safety.set_cruise_engaged_prev(True)
# CC = car.CarControl.new_message(cruiseControl={'cancel': True})
# test_car_controller(CC)
#
# # Test resume + general messages (controls_allowed=True & cruise_engaged=True)
# self.safety.set_controls_allowed(True)
# CC = car.CarControl.new_message(cruiseControl={'resume': True})
# test_car_controller(CC)
#
# def test_panda_safety_carstate(self):
# """
# Assert that panda safety matches openpilot's carState
# """
# if self.CP.dashcamOnly:
# self.skipTest("no need to check panda safety for dashcamOnly")
#
# CC = car.CarControl.new_message()
#
# # warm up pass, as initial states may be different
# for can in self.can_msgs[:300]:
# self.CI.update(CC, (can.as_builder().to_bytes(), ))
# for msg in filter(lambda m: m.src in range(64), can.can):
# to_send = libpanda_py.make_CANPacket(msg.address, msg.src % 4, msg.dat)
# self.safety.safety_rx_hook(to_send)
#
# controls_allowed_prev = False
# CS_prev = car.CarState.new_message()
# checks = defaultdict(lambda: 0)
# for idx, can in enumerate(self.can_msgs):
# CS = self.CI.update(CC, (can.as_builder().to_bytes(), ))
# for msg in filter(lambda m: m.src in range(64), can.can):
# to_send = libpanda_py.make_CANPacket(msg.address, msg.src % 4, msg.dat)
# ret = self.safety.safety_rx_hook(to_send)
# self.assertEqual(1, ret, f"safety rx failed ({ret=}): {to_send}")
#
# # Skip first frame so CS_prev is properly initialized
# if idx == 0:
# CS_prev = CS
# # Button may be left pressed in warm up period
# if not self.CP.pcmCruise:
# self.safety.set_controls_allowed(0)
# continue
#
# # TODO: check rest of panda's carstate (steering, ACC main on, etc.)
#
# checks['gasPressed'] += CS.gasPressed != self.safety.get_gas_pressed_prev()
# checks['standstill'] += CS.standstill == self.safety.get_vehicle_moving()
#
# # TODO: remove this exception once this mismatch is resolved
# brake_pressed = CS.brakePressed
# if CS.brakePressed and not self.safety.get_brake_pressed_prev():
# if self.CP.carFingerprint in (HONDA.PILOT, HONDA.RIDGELINE) and CS.brake > 0.05:
# brake_pressed = False
# checks['brakePressed'] += brake_pressed != self.safety.get_brake_pressed_prev()
# checks['regenBraking'] += CS.regenBraking != self.safety.get_regen_braking_prev()
#
# if self.CP.pcmCruise:
# # On most pcmCruise cars, openpilot's state is always tied to the PCM's cruise state.
# # On Honda Nidec, we always engage on the rising edge of the PCM cruise state, but
# # openpilot brakes to zero even if the min ACC speed is non-zero (i.e. the PCM disengages).
# if self.CP.carName == "honda" and self.CP.carFingerprint not in HONDA_BOSCH:
# # only the rising edges are expected to match
# if CS.cruiseState.enabled and not CS_prev.cruiseState.enabled:
# checks['controlsAllowed'] += not self.safety.get_controls_allowed()
# else:
# checks['controlsAllowed'] += not CS.cruiseState.enabled and self.safety.get_controls_allowed()
#
# # TODO: fix notCar mismatch
# if not self.CP.notCar:
# checks['cruiseState'] += CS.cruiseState.enabled != self.safety.get_cruise_engaged_prev()
# else:
# # Check for enable events on rising edge of controls allowed
# button_enable = any(evt.enable for evt in CS.events)
# mismatch = button_enable != (self.safety.get_controls_allowed() and not controls_allowed_prev)
# checks['controlsAllowed'] += mismatch
# controls_allowed_prev = self.safety.get_controls_allowed()
# if button_enable and not mismatch:
# self.safety.set_controls_allowed(False)
#
# if self.CP.carName == "honda":
# checks['mainOn'] += CS.cruiseState.available != self.safety.get_acc_main_on()
#
# CS_prev = CS
#
# failed_checks = {k: v for k, v in checks.items() if v > 0}
# self.assertFalse(len(failed_checks), f"panda safety doesn't agree with openpilot: {failed_checks}")
@parameterized_class(('car_model', 'test_route'), get_test_cases())

Write Preview
Loading…
Cancel
Save