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

Revert "warm up kinda works"

Browse Source 
This reverts commit 7fc77b07d5.
pull/30443/head
Shane Smiskol 2 years ago
parent 7fc77b07d5
commit 237be49744
1 changed files with 208 additions and 212 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 420
      selfdrive/car/tests/test_models.py

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

@ -175,133 +175,131 @@ class TestCarModelBase(unittest.TestCase):
self.assertEqual(0, set_status, f"failed to set safetyModel {cfg}") self.assertEqual(0, set_status, f"failed to set safetyModel {cfg}")
self.safety.init_tests() self.safety.init_tests()
self.test_panda_safety_carstate() def test_car_params(self):
if self.CP.dashcamOnly:
# def test_car_params(self): self.skipTest("no need to check carParams for dashcamOnly")
# 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)
# # 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 self.CP.steerControlType != car.CarParams.SteerControlType.angle: if tuning == 'pid':
# tuning = self.CP.lateralTuning.which() self.assertTrue(len(self.CP.lateralTuning.pid.kpV))
# if tuning == 'pid': elif tuning == 'torque':
# self.assertTrue(len(self.CP.lateralTuning.pid.kpV)) self.assertTrue(self.CP.lateralTuning.torque.kf > 0)
# elif tuning == 'torque': else:
# self.assertTrue(self.CP.lateralTuning.torque.kf > 0) raise Exception("unknown tuning")
# else:
# raise Exception("unknown tuning") def test_car_interface(self):
# # TODO: also check for checksum violations from can parser
# def test_car_interface(self): can_invalid_cnt = 0
# # TODO: also check for checksum violations from can parser can_valid = False
# can_invalid_cnt = 0 CC = car.CarControl.new_message()
# 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(),))
# for i, msg in enumerate(self.can_msgs): self.CI.apply(CC, msg.logMonoTime)
# CS = self.CI.update(CC, (msg.as_builder().to_bytes(),))
# self.CI.apply(CC, msg.logMonoTime) if CS.canValid:
# can_valid = True
# if CS.canValid:
# can_valid = True # wait max of 2s for low frequency msgs to be seen
# if i > 200 or can_valid:
# # wait max of 2s for low frequency msgs to be seen can_invalid_cnt += not CS.canValid
# if i > 200 or can_valid:
# can_invalid_cnt += not CS.canValid self.assertEqual(can_invalid_cnt, 0)
#
# self.assertEqual(can_invalid_cnt, 0) def test_radar_interface(self):
# os.environ['NO_RADAR_SLEEP'] = "1"
# def test_radar_interface(self): RadarInterface = importlib.import_module(f'selfdrive.car.{self.CP.carName}.radar_interface').RadarInterface
# os.environ['NO_RADAR_SLEEP'] = "1" RI = RadarInterface(self.CP)
# RadarInterface = importlib.import_module(f'selfdrive.car.{self.CP.carName}.radar_interface').RadarInterface assert RI
# RI = RadarInterface(self.CP)
# assert RI # Since OBD port is multiplexed to bus 1 (commonly radar bus) while fingerprinting,
# # start parsing CAN messages after we've left ELM mode and can expect CAN traffic
# # Since OBD port is multiplexed to bus 1 (commonly radar bus) while fingerprinting, error_cnt = 0
# # start parsing CAN messages after we've left ELM mode and can expect CAN traffic for i, msg in enumerate(self.can_msgs[self.elm_frame:]):
# error_cnt = 0 rr = RI.update((msg.as_builder().to_bytes(),))
# for i, msg in enumerate(self.can_msgs[self.elm_frame:]): if rr is not None and i > 50:
# rr = RI.update((msg.as_builder().to_bytes(),)) error_cnt += car.RadarData.Error.canError in rr.errors
# if rr is not None and i > 50: self.assertEqual(error_cnt, 0)
# error_cnt += car.RadarData.Error.canError in rr.errors
# self.assertEqual(error_cnt, 0) def test_panda_safety_rx_checks(self):
# if self.CP.dashcamOnly:
# def test_panda_safety_rx_checks(self): self.skipTest("no need to check panda safety for dashcamOnly")
# if self.CP.dashcamOnly:
# self.skipTest("no need to check panda safety for dashcamOnly") start_ts = self.can_msgs[0].logMonoTime
#
# start_ts = self.can_msgs[0].logMonoTime failed_addrs = Counter()
# for can in self.can_msgs:
# failed_addrs = Counter() # update panda timer
# for can in self.can_msgs: t = (can.logMonoTime - start_ts) / 1e3
# # update panda timer self.safety.set_timer(int(t))
# t = (can.logMonoTime - start_ts) / 1e3
# self.safety.set_timer(int(t)) # run all msgs through the safety RX hook
# for msg in can.can:
# # run all msgs through the safety RX hook if msg.src >= 64:
# for msg in can.can: continue
# if msg.src >= 64:
# continue to_send = libpanda_py.make_CANPacket(msg.address, msg.src % 4, msg.dat)
# if self.safety.safety_rx_hook(to_send) != 1:
# to_send = libpanda_py.make_CANPacket(msg.address, msg.src % 4, msg.dat) failed_addrs[hex(msg.address)] += 1
# if self.safety.safety_rx_hook(to_send) != 1:
# failed_addrs[hex(msg.address)] += 1 # ensure all msgs defined in the addr checks are valid
# self.safety.safety_tick_current_safety_config()
# # ensure all msgs defined in the addr checks are valid if t > 1e6:
# self.safety.safety_tick_current_safety_config() self.assertTrue(self.safety.safety_config_valid())
# if t > 1e6:
# self.assertTrue(self.safety.safety_config_valid()) # Don't check relay malfunction on disabled routes (relay closed),
# # or before fingerprinting is done (elm327 and noOutput)
# # Don't check relay malfunction on disabled routes (relay closed), if self.openpilot_enabled and t / 1e4 > self.car_safety_mode_frame:
# # or before fingerprinting is done (elm327 and noOutput) self.assertFalse(self.safety.get_relay_malfunction())
# if self.openpilot_enabled and t / 1e4 > self.car_safety_mode_frame: else:
# self.assertFalse(self.safety.get_relay_malfunction()) self.safety.set_relay_malfunction(False)
# else:
# self.safety.set_relay_malfunction(False) self.assertFalse(len(failed_addrs), f"panda safety RX check failed: {failed_addrs}")
#
# self.assertFalse(len(failed_addrs), f"panda safety RX check failed: {failed_addrs}") # ensure RX checks go invalid after small time with no traffic
# self.safety.set_timer(int(t + (2*1e6)))
# # ensure RX checks go invalid after small time with no traffic self.safety.safety_tick_current_safety_config()
# self.safety.set_timer(int(t + (2*1e6))) self.assertFalse(self.safety.safety_config_valid())
# self.safety.safety_tick_current_safety_config()
# self.assertFalse(self.safety.safety_config_valid()) def test_panda_safety_tx_cases(self, data=None):
# """Asserts we can tx common messages"""
# def test_panda_safety_tx_cases(self, data=None): if self.CP.notCar:
# """Asserts we can tx common messages""" self.skipTest("Skipping test for notCar")
# if self.CP.notCar:
# self.skipTest("Skipping test for notCar") def test_car_controller(car_control):
# now_nanos = 0
# def test_car_controller(car_control): msgs_sent = 0
# now_nanos = 0 CI = self.CarInterface(self.CP, self.CarController, self.CarState)
# msgs_sent = 0 for _ in range(round(10.0 / DT_CTRL)): # make sure we hit the slowest messages
# CI = self.CarInterface(self.CP, self.CarController, self.CarState) CI.update(car_control, [])
# for _ in range(round(10.0 / DT_CTRL)): # make sure we hit the slowest messages _, sendcan = CI.apply(car_control, now_nanos)
# CI.update(car_control, [])
# _, sendcan = CI.apply(car_control, now_nanos) now_nanos += DT_CTRL * 1e9
# msgs_sent += len(sendcan)
# now_nanos += DT_CTRL * 1e9 for addr, _, dat, bus in sendcan:
# msgs_sent += len(sendcan) to_send = libpanda_py.make_CANPacket(addr, bus % 4, dat)
# for addr, _, dat, bus in sendcan: self.assertTrue(self.safety.safety_tx_hook(to_send), (addr, dat, bus))
# 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 attempted to send messages
# self.assertGreater(msgs_sent, 50) # Make sure we can send all messages while inactive
# CC = car.CarControl.new_message()
# # Make sure we can send all messages while inactive test_car_controller(CC)
# 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)
# # Test cancel + general messages (controls_allowed=False & cruise_engaged=True) CC = car.CarControl.new_message(cruiseControl={'cancel': True})
# self.safety.set_cruise_engaged_prev(True) test_car_controller(CC)
# 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)
# # Test resume + general messages (controls_allowed=True & cruise_engaged=True) CC = car.CarControl.new_message(cruiseControl={'resume': True})
# self.safety.set_controls_allowed(True) test_car_controller(CC)
# CC = car.CarControl.new_message(cruiseControl={'resume': True})
# test_car_controller(CC)
@pytest.mark.nocapture @pytest.mark.nocapture
@settings(max_examples=MAX_EXAMPLES, deadline=None, @settings(max_examples=MAX_EXAMPLES, deadline=None,
@ -345,31 +343,30 @@ class TestCarModelBase(unittest.TestCase):
CC = car.CarControl.new_message() CC = car.CarControl.new_message()
CS = self.CI.update(CC, (can.to_bytes(),)) CS = self.CI.update(CC, (can.to_bytes(),))
# if self.safety.get_gas_pressed_prev() != prev_panda_gas: if self.safety.get_gas_pressed_prev() != prev_panda_gas:
self.assertEqual(CS.gasPressed, self.safety.get_gas_pressed_prev()) self.assertEqual(CS.gasPressed, self.safety.get_gas_pressed_prev())
# if self.safety.get_brake_pressed_prev() != prev_panda_brake: if self.safety.get_brake_pressed_prev() != prev_panda_brake:
brake_pressed = CS.brakePressed brake_pressed = CS.brakePressed
if CS.brakePressed and not self.safety.get_brake_pressed_prev(): 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: if self.CP.carFingerprint in (HONDA.PILOT, HONDA.RIDGELINE) and CS.brake > 0.05:
brake_pressed = False brake_pressed = False
self.assertEqual(brake_pressed, self.safety.get_brake_pressed_prev()) self.assertEqual(brake_pressed, self.safety.get_brake_pressed_prev())
# if self.safety.get_regen_braking_prev() != prev_panda_regen_braking: if self.safety.get_regen_braking_prev() != prev_panda_regen_braking:
self.assertEqual(CS.regenBraking, self.safety.get_regen_braking_prev()) self.assertEqual(CS.regenBraking, self.safety.get_regen_braking_prev())
# if self.safety.get_vehicle_moving() != prev_panda_vehicle_moving: if self.safety.get_vehicle_moving() != prev_panda_vehicle_moving:
self.assertEqual(not CS.standstill, self.safety.get_vehicle_moving()) self.assertEqual(not CS.standstill, self.safety.get_vehicle_moving())
if self.CP.pcmCruise: if not (self.CP.carName == "honda" and self.CP.carFingerprint not in HONDA_BOSCH):
if not (self.CP.carName == "honda" and self.CP.carFingerprint not in HONDA_BOSCH):# and not self.CP.notCar: if self.safety.get_cruise_engaged_prev() != prev_panda_cruise_engaged:
# if self.safety.get_cruise_engaged_prev() != prev_panda_cruise_engaged:
self.assertEqual(CS.cruiseState.enabled, self.safety.get_cruise_engaged_prev()) self.assertEqual(CS.cruiseState.enabled, self.safety.get_cruise_engaged_prev())
if self.CP.carName == "honda": if self.CP.carName == "honda":
# if self.safety.get_acc_main_on() != prev_panda_acc_main_on: if self.safety.get_acc_main_on() != prev_panda_acc_main_on:
self.assertEqual(CS.cruiseState.available, self.safety.get_acc_main_on()) self.assertEqual(CS.cruiseState.available, self.safety.get_acc_main_on())
def test_panda_safety_carstate(self): def test_panda_safety_carstate(self):
""" """
@ -386,74 +383,73 @@ class TestCarModelBase(unittest.TestCase):
for msg in filter(lambda m: m.src in range(64), can.can): 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) to_send = libpanda_py.make_CANPacket(msg.address, msg.src % 4, msg.dat)
self.safety.safety_rx_hook(to_send) self.safety.safety_rx_hook(to_send)
return
controls_allowed_prev = False
# controls_allowed_prev = False CS_prev = car.CarState.new_message()
# CS_prev = car.CarState.new_message() checks = defaultdict(lambda: 0)
# checks = defaultdict(lambda: 0) controlsd = Controls(CI=self.CI)
# controlsd = Controls(CI=self.CI) controlsd.initialized = True
# controlsd.initialized = True for idx, can in enumerate(self.can_msgs):
# for idx, can in enumerate(self.can_msgs): CS = self.CI.update(CC, (can.as_builder().to_bytes(), ))
# CS = self.CI.update(CC, (can.as_builder().to_bytes(), )) for msg in filter(lambda m: m.src in range(64), can.can):
# 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)
# to_send = libpanda_py.make_CANPacket(msg.address, msg.src % 4, msg.dat) ret = self.safety.safety_rx_hook(to_send)
# ret = self.safety.safety_rx_hook(to_send) self.assertEqual(1, ret, f"safety rx failed ({ret=}): {to_send}")
# self.assertEqual(1, ret, f"safety rx failed ({ret=}): {to_send}")
# # Skip first frame so CS_prev is properly initialized
# # Skip first frame so CS_prev is properly initialized if idx == 0:
# if idx == 0: CS_prev = CS
# CS_prev = CS # Button may be left pressed in warm up period
# # Button may be left pressed in warm up period if not self.CP.pcmCruise:
# if not self.CP.pcmCruise: self.safety.set_controls_allowed(0)
# self.safety.set_controls_allowed(0) continue
# continue
# # TODO: check rest of panda's carstate (steering, ACC main on, etc.)
# # TODO: check rest of panda's carstate (steering, ACC main on, etc.)
# checks['gasPressed'] += CS.gasPressed != self.safety.get_gas_pressed_prev()
# checks['gasPressed'] += CS.gasPressed != self.safety.get_gas_pressed_prev() checks['standstill'] += CS.standstill == self.safety.get_vehicle_moving()
# checks['standstill'] += CS.standstill == self.safety.get_vehicle_moving()
# # TODO: remove this exception once this mismatch is resolved
# # TODO: remove this exception once this mismatch is resolved brake_pressed = CS.brakePressed
# brake_pressed = CS.brakePressed if CS.brakePressed and not self.safety.get_brake_pressed_prev():
# 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:
# if self.CP.carFingerprint in (HONDA.PILOT, HONDA.RIDGELINE) and CS.brake > 0.05: brake_pressed = False
# brake_pressed = False checks['brakePressed'] += brake_pressed != self.safety.get_brake_pressed_prev()
# checks['brakePressed'] += brake_pressed != self.safety.get_brake_pressed_prev() checks['regenBraking'] += CS.regenBraking != self.safety.get_regen_braking_prev()
# checks['regenBraking'] += CS.regenBraking != self.safety.get_regen_braking_prev()
# if self.CP.pcmCruise:
# if self.CP.pcmCruise: # On most pcmCruise cars, openpilot's state is always tied to the PCM's cruise state.
# # 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
# # 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).
# # 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:
# if self.CP.carName == "honda" and self.CP.carFingerprint not in HONDA_BOSCH: # only the rising edges are expected to match
# # only the rising edges are expected to match if CS.cruiseState.enabled and not CS_prev.cruiseState.enabled:
# if CS.cruiseState.enabled and not CS_prev.cruiseState.enabled: checks['controlsAllowed'] += not self.safety.get_controls_allowed()
# checks['controlsAllowed'] += not self.safety.get_controls_allowed() else:
# else: checks['controlsAllowed'] += not CS.cruiseState.enabled and self.safety.get_controls_allowed()
# checks['controlsAllowed'] += not CS.cruiseState.enabled and self.safety.get_controls_allowed()
# # TODO: fix notCar mismatch
# # TODO: fix notCar mismatch if not self.CP.notCar:
# if not self.CP.notCar: checks['cruiseState'] += CS.cruiseState.enabled != self.safety.get_cruise_engaged_prev()
# checks['cruiseState'] += CS.cruiseState.enabled != self.safety.get_cruise_engaged_prev() else:
# else: # Check for enable events on rising edge of controls allowed
# # Check for enable events on rising edge of controls allowed controlsd.update_events(CS)
# controlsd.update_events(CS) controlsd.CS_prev = CS
# controlsd.CS_prev = CS button_enable = (any(evt.enable for evt in CS.events) and
# button_enable = (any(evt.enable for evt in CS.events) and not any(evt == EventName.pedalPressed for evt in controlsd.events.names))
# not any(evt == EventName.pedalPressed for evt in controlsd.events.names)) mismatch = button_enable != (self.safety.get_controls_allowed() and not controls_allowed_prev)
# mismatch = button_enable != (self.safety.get_controls_allowed() and not controls_allowed_prev) checks['controlsAllowed'] += mismatch
# checks['controlsAllowed'] += mismatch controls_allowed_prev = self.safety.get_controls_allowed()
# controls_allowed_prev = self.safety.get_controls_allowed() if button_enable and not mismatch:
# if button_enable and not mismatch: self.safety.set_controls_allowed(False)
# self.safety.set_controls_allowed(False)
# if self.CP.carName == "honda":
# if self.CP.carName == "honda": checks['mainOn'] += CS.cruiseState.available != self.safety.get_acc_main_on()
# checks['mainOn'] += CS.cruiseState.available != self.safety.get_acc_main_on()
# CS_prev = CS
# CS_prev = CS
# failed_checks = {k: v for k, v in checks.items() if v > 0}
# 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}")
# self.assertFalse(len(failed_checks), f"panda safety doesn't agree with openpilot: {failed_checks}")
@parameterized_class(('car_model', 'test_route'), get_test_cases()) @parameterized_class(('car_model', 'test_route'), get_test_cases())

Write Preview
Loading…
Cancel
Save