|
|
|
@ -69,7 +69,7 @@ def run_DState_seq(driver_state_msgs, driver_car_interaction, openpilot_status, |
|
|
|
|
events_from_DM.append(event_per_state) # evaluate events at 10Hz for tests |
|
|
|
|
|
|
|
|
|
assert len(events_from_DM)==len(driver_state_msgs), 'somethings wrong' |
|
|
|
|
return events_from_DM |
|
|
|
|
return events_from_DM, DS |
|
|
|
|
|
|
|
|
|
class TestMonitoring(unittest.TestCase): |
|
|
|
|
# -1. rhd parser sanity check |
|
|
|
@ -87,12 +87,12 @@ class TestMonitoring(unittest.TestCase): |
|
|
|
|
|
|
|
|
|
# 0. op engaged, driver is doing fine all the time |
|
|
|
|
def test_fully_aware_driver(self): |
|
|
|
|
events_output = run_DState_seq(always_attentive, always_false, always_true, always_false) |
|
|
|
|
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) |
|
|
|
|
|
|
|
|
|
# 1. op engaged, driver is distracted and does nothing |
|
|
|
|
def test_fully_distracted_driver(self): |
|
|
|
|
events_output = run_DState_seq(always_distracted, always_false, always_true, always_false) |
|
|
|
|
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+\ |
|
|
|
|
((_DISTRACTED_PRE_TIME_TILL_TERMINAL-_DISTRACTED_PROMPT_TIME_TILL_TERMINAL)/2))/DT_DMON)][0].name, 'preDriverDistracted') |
|
|
|
@ -100,10 +100,11 @@ class TestMonitoring(unittest.TestCase): |
|
|
|
|
((_DISTRACTED_PROMPT_TIME_TILL_TERMINAL)/2))/DT_DMON)][0].name, 'promptDriverDistracted') |
|
|
|
|
self.assertEqual(events_output[int((_DISTRACTED_TIME+\ |
|
|
|
|
((_TEST_TIMESPAN-10-_DISTRACTED_TIME)/2))/DT_DMON)][0].name, 'driverDistracted') |
|
|
|
|
self.assertTrue(type(d_status.awareness)==float) |
|
|
|
|
|
|
|
|
|
# 2. op 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) |
|
|
|
|
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+\ |
|
|
|
|
((_AWARENESS_PRE_TIME_TILL_TERMINAL-_AWARENESS_PROMPT_TIME_TILL_TERMINAL)/2))/DT_DMON)][0].name, 'preDriverUnresponsive') |
|
|
|
@ -120,7 +121,7 @@ class TestMonitoring(unittest.TestCase): |
|
|
|
|
[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) |
|
|
|
|
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)][0].name, 'promptDriverDistracted') |
|
|
|
|
self.assertTrue(len(events_output[int(_DISTRACTED_SECONDS_TO_ORANGE*1.5/DT_DMON)])==0) |
|
|
|
@ -139,7 +140,7 @@ 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) |
|
|
|
|
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)][0].name, 'promptDriverDistracted') |
|
|
|
|
self.assertEqual(events_output[int((_DISTRACTED_SECONDS_TO_RED+1.5*_invisible_time)/DT_DMON)][0].name, 'driverDistracted') |
|
|
|
|
self.assertEqual(events_output[int((_DISTRACTED_SECONDS_TO_RED+2*_invisible_time+1.5)/DT_DMON)][0].name, 'driverDistracted') |
|
|
|
@ -154,7 +155,7 @@ class TestMonitoring(unittest.TestCase): |
|
|
|
|
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) |
|
|
|
|
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)][0].name, 'promptDriverUnresponsive') |
|
|
|
|
self.assertTrue(len(events_output[int((_INVISIBLE_SECONDS_TO_ORANGE+0.1)/DT_DMON)])==0) |
|
|
|
@ -177,7 +178,7 @@ 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) |
|
|
|
|
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)][0].name, 'promptDriverUnresponsive') |
|
|
|
|
self.assertEqual(events_output[int((_INVISIBLE_SECONDS_TO_RED-0.1)/DT_DMON)][0].name, 'driverUnresponsive') |
|
|
|
@ -188,7 +189,7 @@ class TestMonitoring(unittest.TestCase): |
|
|
|
|
# 7. op not engaged, 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) |
|
|
|
|
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) |
|
|
|
|
|
|
|
|
|
# 8. op engaged, car stops at traffic light, down to orange, no action, then car starts moving |
|
|
|
@ -197,7 +198,7 @@ class TestMonitoring(unittest.TestCase): |
|
|
|
|
_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) |
|
|
|
|
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)][0].name, 'preDriverDistracted') |
|
|
|
|
self.assertEqual(events_output[int((_redlight_time-0.1)/DT_DMON)][0].name, 'preDriverDistracted') |
|
|
|
|
self.assertEqual(events_output[int((_redlight_time+0.5)/DT_DMON)][0].name, 'promptDriverDistracted') |
|
|
|
@ -210,7 +211,7 @@ class TestMonitoring(unittest.TestCase): |
|
|
|
|
[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) |
|
|
|
|
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((_UNCERTAIN_SECONDS_TO_GREEN-0.1)/DT_DMON)][0].name, 'driverMonitorLowAcc') |
|
|
|
|
self.assertTrue(len(events_output[int((_UNCERTAIN_SECONDS_TO_GREEN+_DISTRACTED_SECONDS_TO_ORANGE-0.5)/DT_DMON)])==0) |
|
|
|
@ -221,7 +222,7 @@ class TestMonitoring(unittest.TestCase): |
|
|
|
|
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) |
|
|
|
|
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((_UNCERTAIN_SECONDS_TO_GREEN)/DT_DMON)][0].name, 'driverMonitorLowAcc') |
|
|
|
|
self.assertEqual(events_output[int((2.5*(_DISTRACTED_TIME-_DISTRACTED_PRE_TIME_TILL_TERMINAL))/DT_DMON)][1].name, 'preDriverDistracted') |
|
|
|
|