diff --git a/selfdrive/monitoring/dmonitoringd.py b/selfdrive/monitoring/dmonitoringd.py index d8fa4932ca..63df16c795 100755 --- a/selfdrive/monitoring/dmonitoringd.py +++ b/selfdrive/monitoring/dmonitoringd.py @@ -14,18 +14,13 @@ def dmonitoringd_thread(sm=None, pm=None): if sm is None: sm = messaging.SubMaster(['driverState', 'liveCalibration', 'carState', 'controlsState', 'model'], poll=['driverState']) - driver_status = DriverStatus() - driver_status.is_rhd_region = Params().get("IsRHD") == b"1" + driver_status = DriverStatus(rhd=Params().get("IsRHD") == b"1") offroad = Params().get("IsOffroad") == b"1" sm['liveCalibration'].calStatus = Calibration.INVALID sm['liveCalibration'].rpyCalib = [0, 0, 0] - sm['carState'].vEgo = 0. - sm['carState'].cruiseState.speed = 0. sm['carState'].buttonEvents = [] - sm['carState'].steeringPressed = False - sm['carState'].gasPressed = False sm['carState'].standstill = True v_cruise_last = 0 diff --git a/selfdrive/monitoring/driver_monitor.py b/selfdrive/monitoring/driver_monitor.py index 33005e19c6..56eac51109 100644 --- a/selfdrive/monitoring/driver_monitor.py +++ b/selfdrive/monitoring/driver_monitor.py @@ -1,11 +1,11 @@ -from common.numpy_fast import interp from math import atan2, sqrt + +from cereal import car +from common.numpy_fast import interp from common.realtime import DT_DMON from common.filter_simple import FirstOrderFilter from common.stat_live import RunningStatFilter -from cereal import car - EventName = car.CarEvent.EventName # ****************************************************************************************** @@ -54,7 +54,7 @@ MAX_TERMINAL_DURATION = 300 # 30s RESIZED_FOCAL = 320.0 H, W, FULL_W = 320, 160, 426 -class DistractedType(): +class DistractedType: NOT_DISTRACTED = 0 BAD_POSE = 1 BAD_BLINK = 2 @@ -63,22 +63,19 @@ def face_orientation_from_net(angles_desc, pos_desc, rpy_calib, is_rhd): # the output of these angles are in device frame # so from driver's perspective, pitch is up and yaw is right - pitch_net = angles_desc[0] - yaw_net = angles_desc[1] - roll_net = angles_desc[2] + pitch_net, yaw_net, roll_net = angles_desc face_pixel_position = ((pos_desc[0] + .5)*W - W + FULL_W, (pos_desc[1]+.5)*H) yaw_focal_angle = atan2(face_pixel_position[0] - FULL_W//2, RESIZED_FOCAL) pitch_focal_angle = atan2(face_pixel_position[1] - H//2, RESIZED_FOCAL) - roll = roll_net pitch = pitch_net + pitch_focal_angle yaw = -yaw_net + yaw_focal_angle # no calib for roll pitch -= rpy_calib[1] yaw -= rpy_calib[2] * (1 - 2 * int(is_rhd)) # lhd -> -=, rhd -> += - return roll, pitch, yaw + return roll_net, pitch, yaw class DriverPose(): def __init__(self): @@ -100,7 +97,8 @@ class DriverBlink(): self.cfactor = 1. class DriverStatus(): - def __init__(self): + def __init__(self, rhd=False): + self.is_rhd_region = rhd self.pose = DriverPose() self.pose_calibrated = self.pose.pitch_offseter.filtered_stat.n > _POSE_OFFSET_MIN_COUNT and \ self.pose.yaw_offseter.filtered_stat.n > _POSE_OFFSET_MIN_COUNT @@ -119,9 +117,6 @@ class DriverStatus(): self.hi_std_alert_enabled = True self.threshold_prompt = _DISTRACTED_PROMPT_TIME_TILL_TERMINAL / _DISTRACTED_TIME - self.is_rhd_region = False - self.is_rhd_region_checked = False - self._set_timers(active_monitoring=True) def _set_timers(self, active_monitoring): @@ -181,8 +176,8 @@ class DriverStatus(): self.blink.cfactor = interp(ep, [0, 0.5, 1], [_BLINK_THRESHOLD_STRICT, _BLINK_THRESHOLD, _BLINK_THRESHOLD_SLACK])/_BLINK_THRESHOLD def get_pose(self, driver_state, cal_rpy, car_speed, op_engaged): - # 10 Hz - if len(driver_state.faceOrientation) == 0 or len(driver_state.facePosition) == 0 or len(driver_state.faceOrientationStd) == 0 or len(driver_state.facePositionStd) == 0: + if not all(len(x) > 0 for x in [driver_state.faceOrientation, driver_state.facePosition, + driver_state.faceOrientationStd, driver_state.facePositionStd]): return self.pose.roll, self.pose.pitch, self.pose.yaw = face_orientation_from_net(driver_state.faceOrientation, driver_state.facePosition, cal_rpy, self.is_rhd_region) diff --git a/selfdrive/monitoring/test_monitoring.py b/selfdrive/monitoring/test_monitoring.py old mode 100644 new mode 100755 index c0292c81af..f914677587 --- a/selfdrive/monitoring/test_monitoring.py +++ b/selfdrive/monitoring/test_monitoring.py @@ -1,11 +1,11 @@ -# flake8: noqa - +#!/usr/bin/env python3 import unittest import numpy as np -from cereal import car + +from cereal import car, log from common.realtime import DT_DMON from selfdrive.controls.lib.events import Events -from selfdrive.monitoring.driver_monitor import DriverStatus, MAX_TERMINAL_ALERTS, \ +from selfdrive.monitoring.driver_monitor import DriverStatus, \ _AWARENESS_TIME, _AWARENESS_PRE_TIME_TILL_TERMINAL, \ _AWARENESS_PROMPT_TIME_TILL_TERMINAL, _DISTRACTED_TIME, \ _DISTRACTED_PRE_TIME_TILL_TERMINAL, _DISTRACTED_PROMPT_TIME_TILL_TERMINAL, \ @@ -20,40 +20,32 @@ _INVISIBLE_SECONDS_TO_ORANGE = _AWARENESS_TIME - _AWARENESS_PROMPT_TIME_TILL_TER _INVISIBLE_SECONDS_TO_RED = _AWARENESS_TIME + 1 _UNCERTAIN_SECONDS_TO_GREEN = _HI_STD_TIMEOUT + 0.5 -class fake_DM_msg(): - def __init__(self, is_face_detected, is_distracted=False, is_model_uncertain=False): - self.faceOrientation = [0., 0., 0.] - self.facePosition = [0., 0.] - self.faceProb = 1. * is_face_detected - self.leftEyeProb = 1. - self.rightEyeProb = 1. - self.leftBlinkProb = 1. * is_distracted - self.rightBlinkProb = 1. * is_distracted - self.faceOrientationStd = [1.*is_model_uncertain, 1.*is_model_uncertain, 1.*is_model_uncertain] - self.facePositionStd = [1.*is_model_uncertain, 1.*is_model_uncertain] - self.sgProb = 0. - +def make_msg(face_detected, distracted=False, model_uncertain=False): + ds = log.DriverState.new_message() + ds.faceOrientation = [0., 0., 0.] + ds.facePosition = [0., 0.] + ds.faceProb = 1. * face_detected + ds.leftEyeProb = 1. + ds.rightEyeProb = 1. + ds.leftBlinkProb = 1. * distracted + ds.rightBlinkProb = 1. * distracted + ds.faceOrientationStd = [1.*model_uncertain, 1.*model_uncertain, 1.*model_uncertain] + ds.facePositionStd = [1.*model_uncertain, 1.*model_uncertain] + ds.sgProb = 0. + return ds # driver state from neural net, 10Hz -msg_NO_FACE_DETECTED = fake_DM_msg(is_face_detected=False) -msg_ATTENTIVE = fake_DM_msg(is_face_detected=True) -msg_DISTRACTED = fake_DM_msg(is_face_detected=True, is_distracted=True) -msg_ATTENTIVE_UNCERTAIN = fake_DM_msg(is_face_detected=True, is_model_uncertain=True) -msg_DISTRACTED_UNCERTAIN = fake_DM_msg(is_face_detected=True, is_distracted=True, is_model_uncertain=True) -msg_DISTRACTED_BUT_SOMEHOW_UNCERTAIN = fake_DM_msg(is_face_detected=True, is_distracted=True, is_model_uncertain=_POSESTD_THRESHOLD*1.5) +msg_NO_FACE_DETECTED = make_msg(False) +msg_ATTENTIVE = make_msg(True) +msg_DISTRACTED = make_msg(True, distracted=True) +msg_ATTENTIVE_UNCERTAIN = make_msg(True, model_uncertain=True) +msg_DISTRACTED_UNCERTAIN = make_msg(True, distracted=True, model_uncertain=True) +msg_DISTRACTED_BUT_SOMEHOW_UNCERTAIN = make_msg(True, distracted=True, model_uncertain=_POSESTD_THRESHOLD*1.5) # driver interaction with car car_interaction_DETECTED = True car_interaction_NOT_DETECTED = False -# openpilot state -openpilot_ENGAGED = True -openpilot_NOT_ENGAGED = False - -# car standstill state -car_STANDSTILL = True -car_NOT_STANDSTILL = False - # some common state vectors always_no_face = [msg_NO_FACE_DETECTED] * int(_TEST_TIMESPAN/DT_DMON) always_attentive = [msg_ATTENTIVE] * int(_TEST_TIMESPAN/DT_DMON) @@ -61,66 +53,69 @@ always_distracted = [msg_DISTRACTED] * int(_TEST_TIMESPAN/DT_DMON) always_true = [True] * int(_TEST_TIMESPAN/DT_DMON) always_false = [False] * int(_TEST_TIMESPAN/DT_DMON) -def run_DState_seq(driver_state_msgs, driver_car_interaction, openpilot_status, car_standstill_status): - # inputs are all 10Hz - DS = DriverStatus() - events_from_DM = [] - for idx in range(len(driver_state_msgs)): - e = Events() - DS.get_pose(driver_state_msgs[idx], [0, 0, 0], 0, openpilot_status[idx]) - # cal_rpy and car_speed don't matter here - - # evaluate events at 10Hz for tests - DS.update(e, driver_car_interaction[idx], openpilot_status[idx], car_standstill_status[idx]) - events_from_DM.append(e) - assert len(events_from_DM) == len(driver_state_msgs), 'somethings wrong' - return events_from_DM, DS - +# TODO: this only tests DriverStatus class TestMonitoring(unittest.TestCase): - # 0. op engaged, driver is doing fine all the time + def _run_seq(self, msgs, interaction, engaged, standstill): + DS = DriverStatus() + events = [] + for idx in range(len(msgs)): + e = Events() + DS.get_pose(msgs[idx], [0, 0, 0], 0, engaged[idx]) + # cal_rpy and car_speed don't matter here + + # evaluate events at 10Hz for tests + DS.update(e, interaction[idx], engaged[idx], standstill[idx]) + events.append(e) + assert len(events) == len(msgs), f"got {len(events)} for {len(msgs)} driverState input msgs" + return events, DS + + def _assert_no_events(self, events): + self.assertTrue(all(not len(e) for e in events)) + + # engaged, driver is attentive all the time def test_fully_aware_driver(self): - events_output = run_DState_seq(always_attentive, always_false, always_true, always_false)[0] - self.assertTrue(np.sum([len(event) for event in events_output]) == 0) + events, _ = self._run_seq(always_attentive, always_false, always_true, always_false) + self._assert_no_events(events) - # 1. op engaged, driver is distracted and does nothing + # engaged, driver is distracted and does nothing def test_fully_distracted_driver(self): - events_output, d_status = run_DState_seq(always_distracted, always_false, always_true, always_false) - self.assertTrue(len(events_output[int((_DISTRACTED_TIME-_DISTRACTED_PRE_TIME_TILL_TERMINAL)/2/DT_DMON)]) == 0) - self.assertEqual(events_output[int((_DISTRACTED_TIME-_DISTRACTED_PRE_TIME_TILL_TERMINAL + + events, d_status = self._run_seq(always_distracted, always_false, always_true, always_false) + self.assertEqual(len(events[int((_DISTRACTED_TIME-_DISTRACTED_PRE_TIME_TILL_TERMINAL)/2/DT_DMON)]), 0) + self.assertEqual(events[int((_DISTRACTED_TIME-_DISTRACTED_PRE_TIME_TILL_TERMINAL + ((_DISTRACTED_PRE_TIME_TILL_TERMINAL-_DISTRACTED_PROMPT_TIME_TILL_TERMINAL)/2))/DT_DMON)].names[0], EventName.preDriverDistracted) - self.assertEqual(events_output[int((_DISTRACTED_TIME-_DISTRACTED_PROMPT_TIME_TILL_TERMINAL + + self.assertEqual(events[int((_DISTRACTED_TIME-_DISTRACTED_PROMPT_TIME_TILL_TERMINAL + ((_DISTRACTED_PROMPT_TIME_TILL_TERMINAL)/2))/DT_DMON)].names[0], EventName.promptDriverDistracted) - self.assertEqual(events_output[int((_DISTRACTED_TIME + + self.assertEqual(events[int((_DISTRACTED_TIME + ((_TEST_TIMESPAN-10-_DISTRACTED_TIME)/2))/DT_DMON)].names[0], EventName.driverDistracted) self.assertIs(type(d_status.awareness), float) - # 2. op engaged, no face detected the whole time, no action + # engaged, no face detected the whole time, no action def test_fully_invisible_driver(self): - events_output = run_DState_seq(always_no_face, always_false, always_true, always_false)[0] - self.assertTrue(len(events_output[int((_AWARENESS_TIME-_AWARENESS_PRE_TIME_TILL_TERMINAL)/2/DT_DMON)]) == 0) - self.assertEqual(events_output[int((_AWARENESS_TIME-_AWARENESS_PRE_TIME_TILL_TERMINAL + + events = self._run_seq(always_no_face, always_false, always_true, always_false)[0] + self.assertTrue(len(events[int((_AWARENESS_TIME-_AWARENESS_PRE_TIME_TILL_TERMINAL)/2/DT_DMON)]) == 0) + self.assertEqual(events[int((_AWARENESS_TIME-_AWARENESS_PRE_TIME_TILL_TERMINAL + ((_AWARENESS_PRE_TIME_TILL_TERMINAL-_AWARENESS_PROMPT_TIME_TILL_TERMINAL)/2))/DT_DMON)].names[0], EventName.preDriverUnresponsive) - self.assertEqual(events_output[int((_AWARENESS_TIME-_AWARENESS_PROMPT_TIME_TILL_TERMINAL + + self.assertEqual(events[int((_AWARENESS_TIME-_AWARENESS_PROMPT_TIME_TILL_TERMINAL + ((_AWARENESS_PROMPT_TIME_TILL_TERMINAL)/2))/DT_DMON)].names[0], EventName.promptDriverUnresponsive) - self.assertEqual(events_output[int((_AWARENESS_TIME + + self.assertEqual(events[int((_AWARENESS_TIME + ((_TEST_TIMESPAN-10-_AWARENESS_TIME)/2))/DT_DMON)].names[0], EventName.driverUnresponsive) - # 3. op engaged, down to orange, driver pays attention, back to normal; then down to orange, driver touches wheel + # engaged, down to orange, driver pays attention, back to normal; then down to orange, driver touches wheel # - should have short orange recovery time and no green afterwards; should recover rightaway on wheel touch def test_normal_driver(self): ds_vector = [msg_DISTRACTED] * int(_DISTRACTED_SECONDS_TO_ORANGE/DT_DMON) + \ [msg_ATTENTIVE] * int(_DISTRACTED_SECONDS_TO_ORANGE/DT_DMON) + \ [msg_DISTRACTED] * (int(_TEST_TIMESPAN/DT_DMON)-int(_DISTRACTED_SECONDS_TO_ORANGE*2/DT_DMON)) interaction_vector = [car_interaction_NOT_DETECTED] * int(_DISTRACTED_SECONDS_TO_ORANGE*3/DT_DMON) + \ - [car_interaction_DETECTED] * (int(_TEST_TIMESPAN/DT_DMON)-int(_DISTRACTED_SECONDS_TO_ORANGE*3/DT_DMON)) - events_output = run_DState_seq(ds_vector, interaction_vector, always_true, always_false)[0] - self.assertTrue(len(events_output[int(_DISTRACTED_SECONDS_TO_ORANGE*0.5/DT_DMON)]) == 0) - self.assertEqual(events_output[int((_DISTRACTED_SECONDS_TO_ORANGE-0.1)/DT_DMON)].names[0], EventName.promptDriverDistracted) - self.assertTrue(len(events_output[int(_DISTRACTED_SECONDS_TO_ORANGE*1.5/DT_DMON)]) == 0) - self.assertEqual(events_output[int((_DISTRACTED_SECONDS_TO_ORANGE*3-0.1)/DT_DMON)].names[0], EventName.promptDriverDistracted) - self.assertTrue(len(events_output[int((_DISTRACTED_SECONDS_TO_ORANGE*3+0.1)/DT_DMON)]) == 0) - - # 4. op engaged, down to orange, driver dodges camera, then comes back still distracted, down to red, \ + [car_interaction_DETECTED] * (int(_TEST_TIMESPAN/DT_DMON)-int(_DISTRACTED_SECONDS_TO_ORANGE*3/DT_DMON)) + events = self._run_seq(ds_vector, interaction_vector, always_true, always_false)[0] + self.assertEqual(len(events[int(_DISTRACTED_SECONDS_TO_ORANGE*0.5/DT_DMON)]), 0) + self.assertEqual(events[int((_DISTRACTED_SECONDS_TO_ORANGE-0.1)/DT_DMON)].names[0], EventName.promptDriverDistracted) + self.assertEqual(len(events[int(_DISTRACTED_SECONDS_TO_ORANGE*1.5/DT_DMON)]), 0) + self.assertEqual(events[int((_DISTRACTED_SECONDS_TO_ORANGE*3-0.1)/DT_DMON)].names[0], EventName.promptDriverDistracted) + self.assertEqual(len(events[int((_DISTRACTED_SECONDS_TO_ORANGE*3+0.1)/DT_DMON)]), 0) + + # engaged, down to orange, driver dodges camera, then comes back still distracted, down to red, \ # driver dodges, and then touches wheel to no avail, disengages and reengages # - orange/red alert should remain after disappearance, and only disengaging clears red def test_biggest_comma_fan(self): @@ -132,35 +127,32 @@ class TestMonitoring(unittest.TestCase): ds_vector[int((_DISTRACTED_SECONDS_TO_RED+_invisible_time)/DT_DMON):int((_DISTRACTED_SECONDS_TO_RED+2*_invisible_time)/DT_DMON)] = [msg_NO_FACE_DETECTED] * int(_invisible_time/DT_DMON) interaction_vector[int((_DISTRACTED_SECONDS_TO_RED+2*_invisible_time+0.5)/DT_DMON):int((_DISTRACTED_SECONDS_TO_RED+2*_invisible_time+1.5)/DT_DMON)] = [True] * int(1/DT_DMON) op_vector[int((_DISTRACTED_SECONDS_TO_RED+2*_invisible_time+2.5)/DT_DMON):int((_DISTRACTED_SECONDS_TO_RED+2*_invisible_time+3)/DT_DMON)] = [False] * int(0.5/DT_DMON) - events_output = run_DState_seq(ds_vector, interaction_vector, op_vector, always_false)[0] - self.assertEqual(events_output[int((_DISTRACTED_SECONDS_TO_ORANGE+0.5*_invisible_time)/DT_DMON)].names[0], EventName.promptDriverDistracted) - self.assertEqual(events_output[int((_DISTRACTED_SECONDS_TO_RED+1.5*_invisible_time)/DT_DMON)].names[0], EventName.driverDistracted) - self.assertEqual(events_output[int((_DISTRACTED_SECONDS_TO_RED+2*_invisible_time+1.5)/DT_DMON)].names[0], EventName.driverDistracted) - self.assertTrue(len(events_output[int((_DISTRACTED_SECONDS_TO_RED+2*_invisible_time+3.5)/DT_DMON)]) == 0) + events = self._run_seq(ds_vector, interaction_vector, op_vector, always_false)[0] + self.assertEqual(events[int((_DISTRACTED_SECONDS_TO_ORANGE+0.5*_invisible_time)/DT_DMON)].names[0], EventName.promptDriverDistracted) + self.assertEqual(events[int((_DISTRACTED_SECONDS_TO_RED+1.5*_invisible_time)/DT_DMON)].names[0], EventName.driverDistracted) + self.assertEqual(events[int((_DISTRACTED_SECONDS_TO_RED+2*_invisible_time+1.5)/DT_DMON)].names[0], EventName.driverDistracted) + self.assertTrue(len(events[int((_DISTRACTED_SECONDS_TO_RED+2*_invisible_time+3.5)/DT_DMON)]) == 0) - # 5. op engaged, invisible driver, down to orange, driver touches wheel; then down to orange again, driver appears + # engaged, invisible driver, down to orange, driver touches wheel; then down to orange again, driver appears # - both actions should clear the alert, but momentary appearence should not def test_sometimes_transparent_commuter(self): - _visible_time = np.random.choice([0.5, 10]) # seconds - # print _visible_time + _visible_time = np.random.choice([0.5, 10]) ds_vector = always_no_face[:]*2 interaction_vector = always_false[:]*2 ds_vector[int((2*_INVISIBLE_SECONDS_TO_ORANGE+1)/DT_DMON):int((2*_INVISIBLE_SECONDS_TO_ORANGE+1+_visible_time)/DT_DMON)] = [msg_ATTENTIVE] * int(_visible_time/DT_DMON) interaction_vector[int((_INVISIBLE_SECONDS_TO_ORANGE)/DT_DMON):int((_INVISIBLE_SECONDS_TO_ORANGE+1)/DT_DMON)] = [True] * int(1/DT_DMON) - events_output = run_DState_seq(ds_vector, interaction_vector, 2*always_true, 2*always_false)[0] - self.assertTrue(len(events_output[int(_INVISIBLE_SECONDS_TO_ORANGE*0.5/DT_DMON)]) == 0) - self.assertEqual(events_output[int((_INVISIBLE_SECONDS_TO_ORANGE-0.1)/DT_DMON)].names[0], EventName.promptDriverUnresponsive) - self.assertTrue(len(events_output[int((_INVISIBLE_SECONDS_TO_ORANGE+0.1)/DT_DMON)]) == 0) + events = self._run_seq(ds_vector, interaction_vector, 2*always_true, 2*always_false)[0] + self.assertTrue(len(events[int(_INVISIBLE_SECONDS_TO_ORANGE*0.5/DT_DMON)]) == 0) + self.assertEqual(events[int((_INVISIBLE_SECONDS_TO_ORANGE-0.1)/DT_DMON)].names[0], EventName.promptDriverUnresponsive) + self.assertTrue(len(events[int((_INVISIBLE_SECONDS_TO_ORANGE+0.1)/DT_DMON)]) == 0) if _visible_time == 0.5: - self.assertEqual(events_output[int((_INVISIBLE_SECONDS_TO_ORANGE*2+1-0.1)/DT_DMON)].names[0], EventName.promptDriverUnresponsive) - self.assertEqual(events_output[int((_INVISIBLE_SECONDS_TO_ORANGE*2+1+0.1+_visible_time)/DT_DMON)].names[0], EventName.preDriverUnresponsive) + self.assertEqual(events[int((_INVISIBLE_SECONDS_TO_ORANGE*2+1-0.1)/DT_DMON)].names[0], EventName.promptDriverUnresponsive) + self.assertEqual(events[int((_INVISIBLE_SECONDS_TO_ORANGE*2+1+0.1+_visible_time)/DT_DMON)].names[0], EventName.preDriverUnresponsive) elif _visible_time == 10: - self.assertEqual(events_output[int((_INVISIBLE_SECONDS_TO_ORANGE*2+1-0.1)/DT_DMON)].names[0], EventName.promptDriverUnresponsive) - self.assertTrue(len(events_output[int((_INVISIBLE_SECONDS_TO_ORANGE*2+1+0.1+_visible_time)/DT_DMON)]) == 0) - else: - pass + self.assertEqual(events[int((_INVISIBLE_SECONDS_TO_ORANGE*2+1-0.1)/DT_DMON)].names[0], EventName.promptDriverUnresponsive) + self.assertTrue(len(events[int((_INVISIBLE_SECONDS_TO_ORANGE*2+1+0.1+_visible_time)/DT_DMON)]) == 0) - # 6. op engaged, invisible driver, down to red, driver appears and then touches wheel, then disengages/reengages + # engaged, invisible driver, down to red, driver appears and then touches wheel, then disengages/reengages # - only disengage will clear the alert def test_last_second_responder(self): _visible_time = 2 # seconds @@ -170,57 +162,56 @@ class TestMonitoring(unittest.TestCase): ds_vector[int(_INVISIBLE_SECONDS_TO_RED/DT_DMON):int((_INVISIBLE_SECONDS_TO_RED+_visible_time)/DT_DMON)] = [msg_ATTENTIVE] * int(_visible_time/DT_DMON) interaction_vector[int((_INVISIBLE_SECONDS_TO_RED+_visible_time)/DT_DMON):int((_INVISIBLE_SECONDS_TO_RED+_visible_time+1)/DT_DMON)] = [True] * int(1/DT_DMON) op_vector[int((_INVISIBLE_SECONDS_TO_RED+_visible_time+1)/DT_DMON):int((_INVISIBLE_SECONDS_TO_RED+_visible_time+0.5)/DT_DMON)] = [False] * int(0.5/DT_DMON) - events_output = run_DState_seq(ds_vector, interaction_vector, op_vector, always_false)[0] - self.assertTrue(len(events_output[int(_INVISIBLE_SECONDS_TO_ORANGE*0.5/DT_DMON)]) == 0) - self.assertEqual(events_output[int((_INVISIBLE_SECONDS_TO_ORANGE-0.1)/DT_DMON)].names[0], EventName.promptDriverUnresponsive) - self.assertEqual(events_output[int((_INVISIBLE_SECONDS_TO_RED-0.1)/DT_DMON)].names[0], EventName.driverUnresponsive) - self.assertEqual(events_output[int((_INVISIBLE_SECONDS_TO_RED+0.5*_visible_time)/DT_DMON)].names[0], EventName.driverUnresponsive) - self.assertEqual(events_output[int((_INVISIBLE_SECONDS_TO_RED+_visible_time+0.5)/DT_DMON)].names[0], EventName.driverUnresponsive) - self.assertTrue(len(events_output[int((_INVISIBLE_SECONDS_TO_RED+_visible_time+1+0.1)/DT_DMON)]) == 0) - - # 7. op not engaged, always distracted driver + events = self._run_seq(ds_vector, interaction_vector, op_vector, always_false)[0] + self.assertTrue(len(events[int(_INVISIBLE_SECONDS_TO_ORANGE*0.5/DT_DMON)]) == 0) + self.assertEqual(events[int((_INVISIBLE_SECONDS_TO_ORANGE-0.1)/DT_DMON)].names[0], EventName.promptDriverUnresponsive) + self.assertEqual(events[int((_INVISIBLE_SECONDS_TO_RED-0.1)/DT_DMON)].names[0], EventName.driverUnresponsive) + self.assertEqual(events[int((_INVISIBLE_SECONDS_TO_RED+0.5*_visible_time)/DT_DMON)].names[0], EventName.driverUnresponsive) + self.assertEqual(events[int((_INVISIBLE_SECONDS_TO_RED+_visible_time+0.5)/DT_DMON)].names[0], EventName.driverUnresponsive) + self.assertTrue(len(events[int((_INVISIBLE_SECONDS_TO_RED+_visible_time+1+0.1)/DT_DMON)]) == 0) + + # disengaged, always distracted driver # - dm should stay quiet when not engaged def test_pure_dashcam_user(self): - events_output = run_DState_seq(always_distracted, always_false, always_false, always_false)[0] - self.assertTrue(np.sum([len(event) for event in events_output]) == 0) + events = self._run_seq(always_distracted, always_false, always_false, always_false)[0] + self.assertTrue(np.sum([len(event) for event in events]) == 0) - # 8. op engaged, car stops at traffic light, down to orange, no action, then car starts moving + # engaged, car stops at traffic light, down to orange, no action, then car starts moving # - should only reach green when stopped, but continues counting down on launch def test_long_traffic_light_victim(self): _redlight_time = 60 # seconds standstill_vector = always_true[:] standstill_vector[int(_redlight_time/DT_DMON):] = [False] * int((_TEST_TIMESPAN-_redlight_time)/DT_DMON) - events_output = run_DState_seq(always_distracted, always_false, always_true, standstill_vector)[0] - self.assertEqual(events_output[int((_DISTRACTED_TIME-_DISTRACTED_PRE_TIME_TILL_TERMINAL+1)/DT_DMON)].names[0], EventName.preDriverDistracted) - self.assertEqual(events_output[int((_redlight_time-0.1)/DT_DMON)].names[0], EventName.preDriverDistracted) - self.assertEqual(events_output[int((_redlight_time+0.5)/DT_DMON)].names[0], EventName.promptDriverDistracted) + events = self._run_seq(always_distracted, always_false, always_true, standstill_vector)[0] + self.assertEqual(events[int((_DISTRACTED_TIME-_DISTRACTED_PRE_TIME_TILL_TERMINAL+1)/DT_DMON)].names[0], EventName.preDriverDistracted) + self.assertEqual(events[int((_redlight_time-0.1)/DT_DMON)].names[0], EventName.preDriverDistracted) + self.assertEqual(events[int((_redlight_time+0.5)/DT_DMON)].names[0], EventName.promptDriverDistracted) - # 9. op engaged, model is extremely uncertain. driver first attentive, then distracted + # engaged, model is extremely uncertain. driver first attentive, then distracted # - should pop a uncertain message first, then slowly into active green/orange, finally back to wheel touch but timer locked by orange def test_one_indecisive_model(self): ds_vector = [msg_ATTENTIVE_UNCERTAIN] * int(_UNCERTAIN_SECONDS_TO_GREEN/DT_DMON) + \ [msg_ATTENTIVE] * int(_DISTRACTED_SECONDS_TO_ORANGE/DT_DMON) + \ [msg_DISTRACTED_UNCERTAIN] * (int(_TEST_TIMESPAN/DT_DMON)-int((_DISTRACTED_SECONDS_TO_ORANGE+_UNCERTAIN_SECONDS_TO_GREEN)/DT_DMON)) interaction_vector = always_false[:] - events_output = run_DState_seq(ds_vector, interaction_vector, always_true, always_false)[0] - self.assertTrue(len(events_output[int(_UNCERTAIN_SECONDS_TO_GREEN*0.5/DT_DMON)]) == 0) - self.assertEqual(events_output[int((_HI_STD_TIMEOUT)/DT_DMON)].names[0], EventName.driverMonitorLowAcc) - self.assertTrue(len(events_output[int((_UNCERTAIN_SECONDS_TO_GREEN+_DISTRACTED_SECONDS_TO_ORANGE-0.5)/DT_DMON)]) == 0) - self.assertTrue(EventName.promptDriverDistracted in events_output[int((_TEST_TIMESPAN-5.)/DT_DMON)].names) + events = self._run_seq(ds_vector, interaction_vector, always_true, always_false)[0] + self.assertTrue(len(events[int(_UNCERTAIN_SECONDS_TO_GREEN*0.5/DT_DMON)]) == 0) + self.assertEqual(events[int((_HI_STD_TIMEOUT)/DT_DMON)].names[0], EventName.driverMonitorLowAcc) + self.assertTrue(len(events[int((_UNCERTAIN_SECONDS_TO_GREEN+_DISTRACTED_SECONDS_TO_ORANGE-0.5)/DT_DMON)]) == 0) + self.assertTrue(EventName.promptDriverDistracted in events[int((_TEST_TIMESPAN-5.)/DT_DMON)].names) - # 10. op engaged, model is somehow uncertain and driver is distracted + # engaged, model is somehow uncertain and driver is distracted # - should slow down the alert countdown but it still gets there def test_somehow_indecisive_model(self): ds_vector = [msg_DISTRACTED_BUT_SOMEHOW_UNCERTAIN] * int(_TEST_TIMESPAN/DT_DMON) interaction_vector = always_false[:] - events_output = run_DState_seq(ds_vector, interaction_vector, always_true, always_false)[0] - self.assertTrue(len(events_output[int(_UNCERTAIN_SECONDS_TO_GREEN*0.5/DT_DMON)]) == 0) - self.assertEqual(events_output[int((_HI_STD_TIMEOUT)/DT_DMON)].names[0], EventName.driverMonitorLowAcc) - self.assertTrue(EventName.preDriverDistracted in events_output[int((2*(_DISTRACTED_TIME-_DISTRACTED_PRE_TIME_TILL_TERMINAL))/DT_DMON)].names) - self.assertTrue(EventName.promptDriverDistracted in events_output[int((2*(_DISTRACTED_TIME-_DISTRACTED_PROMPT_TIME_TILL_TERMINAL))/DT_DMON)].names) - self.assertEqual(events_output[int((_DISTRACTED_TIME+1)/DT_DMON)].names[0], EventName.promptDriverDistracted) - self.assertEqual(events_output[int((_DISTRACTED_TIME*2.5)/DT_DMON)].names[0], EventName.promptDriverDistracted) # set_timer blocked + events = self._run_seq(ds_vector, interaction_vector, always_true, always_false)[0] + self.assertEqual(len(events[int(_UNCERTAIN_SECONDS_TO_GREEN*0.5/DT_DMON)]), 0) + self.assertEqual(events[int((_HI_STD_TIMEOUT)/DT_DMON)].names[0], EventName.driverMonitorLowAcc) + self.assertTrue(EventName.preDriverDistracted in events[int((2*(_DISTRACTED_TIME-_DISTRACTED_PRE_TIME_TILL_TERMINAL))/DT_DMON)].names) + self.assertTrue(EventName.promptDriverDistracted in events[int((2*(_DISTRACTED_TIME-_DISTRACTED_PROMPT_TIME_TILL_TERMINAL))/DT_DMON)].names) + self.assertEqual(events[int((_DISTRACTED_TIME+1)/DT_DMON)].names[0], EventName.promptDriverDistracted) + self.assertEqual(events[int((_DISTRACTED_TIME*2.5)/DT_DMON)].names[0], EventName.promptDriverDistracted) # set_timer blocked if __name__ == "__main__": - print('MAX_TERMINAL_ALERTS', MAX_TERMINAL_ALERTS) unittest.main()