Keep owned native preparation screen awake until playback or failure

roadscore/prototype/normal_ui_audit.py

@@ -1,7 +1,9 @@
 """Observe the unmodified normal UI's own update/draw calls. No custom drawing."""
 import os,json,time,runpy
 from pathlib import Path
-from openpilot.selfdrive.ui.ui_state import UIState
+from openpilot.selfdrive.ui.ui_state import UIState, device
+from preparing_awake import PreparationWake
+preparation_wake=PreparationWake(os.environ.get("ROADSCORE_STATUS_FILE"),Path("/TICI").exists())
 from openpilot.selfdrive.ui.onroad.cameraview import CameraView
 from openpilot.selfdrive.ui.mici.onroad.model_renderer import ModelRenderer
 counts={'accepted_camera_frames':0,'path_draw_calls':0,'lane_draw_calls':0,'nonempty_path_draws':0,'nonempty_lane_draws':0,'camera_texture_draws':0};last=0.;original_update=UIState.update;original_accept=CameraView._accept_frame;original_path=ModelRenderer._draw_path;original_lanes=ModelRenderer._draw_lane_lines;original_textures=CameraView._render_textures
@@ -23,6 +25,7 @@ def textures(self,*args,**kw):
  return original_textures(self,*args,**kw)
 def update(self,*args,**kw):
  global last
+ preparation_wake.update(self,device)
  result=original_update(self,*args,**kw);now=time.monotonic()
  if now-last>=1:
   out.write(json.dumps({'wall':now,'started':bool(self.started),'speed':float(self.sm['carState'].vEgo),'model_mono_ns':self.sm.logMonoTime['modelV2'],'model_points':len(self.sm['modelV2'].position.x),**counts})+'\n');last=now

roadscore/prototype/preparing_awake.py

@@ -0,0 +1,29 @@
+"""Keep the owned native preparation screen interactive until playback or failure."""
+import json
+from pathlib import Path
+import time
+
+
+class PreparationWake:
+    def __init__(self, status_path, native):
+        self.path = Path(status_path) if status_path else None
+        self.native = native
+        self.next_check = 0.
+
+    def update(self, ui_state, device, now=None):
+        if not self.native or self.path is None or ui_state.started:
+            return False
+        now = time.monotonic() if now is None else now
+        if now < self.next_check:
+            return False
+        self.next_check = now + 1.
+        try:
+            status = json.loads(self.path.read_text())
+        except (OSError, ValueError):
+            return False
+        if not isinstance(status, dict) or str(status.get('readiness', '')).upper() != 'PREPARING':
+            return False
+        if status.get('worker_failed') or status.get('failure_kind'):
+            return False
+        device.reset_interactive_timeout()
+        return True

roadscore/prototype/test_preparing_awake.py

@@ -0,0 +1,40 @@
+import json
+from pathlib import Path
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import Mock
+from preparing_awake import PreparationWake
+
+class WakeTests(unittest.TestCase):
+ def setUp(self):
+  self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
+  self.path=Path(self.tmp.name)/'status.json';self.device=Mock();self.ui=SimpleNamespace(started=False)
+  self.wake=PreparationWake(self.path,True)
+ def state(self,**values):self.path.write_text(json.dumps(values))
+ def test_long_cold_preparation_remains_visible_then_stops_on_ready(self):
+  self.state(readiness='Preparing')
+  for second in range(0,901):self.assertTrue(self.wake.update(self.ui,self.device,second))
+  self.assertEqual(self.device.reset_interactive_timeout.call_count,901)
+  self.state(readiness='READY');self.assertFalse(self.wake.update(self.ui,self.device,901))
+  self.assertEqual(self.device.reset_interactive_timeout.call_count,901)
+ def test_no_other_ui_or_playback_is_kept_awake(self):
+  self.state(readiness='PREPARING')
+  self.ui.started=True;self.assertFalse(self.wake.update(self.ui,self.device,0))
+  self.ui.started=False
+  self.assertFalse(PreparationWake(self.path,False).update(self.ui,self.device,0))
+  self.assertFalse(PreparationWake(None,True).update(self.ui,self.device,0))
+  self.device.reset_interactive_timeout.assert_not_called()
+ def test_failure_missing_and_invalid_status_stop_refresh(self):
+  for second,value in enumerate([{'readiness':'DEGRADED'},{'readiness':'Preparing','worker_failed':True},{'readiness':'Preparing','failure_kind':'connection_lost'},[]]):
+   self.path.write_text(json.dumps(value));self.assertFalse(self.wake.update(self.ui,self.device,second))
+  self.path.unlink();self.assertFalse(self.wake.update(self.ui,self.device,4))
+  self.path.write_text('{');self.assertFalse(self.wake.update(self.ui,self.device,5))
+  self.device.reset_interactive_timeout.assert_not_called()
+ def test_bounded_polling(self):
+  self.state(readiness='PREPARING')
+  self.assertTrue(self.wake.update(self.ui,self.device,0))
+  self.assertFalse(self.wake.update(self.ui,self.device,.1))
+  self.assertTrue(self.wake.update(self.ui,self.device,1))
+
+if __name__=='__main__':unittest.main()