Keep calibration taps responsive with direct real Params safety reads

roadscore/prototype/operator_output.py

@@ -50,13 +50,19 @@ def busy(path):
     return True
 
 
-def real_offroad():
-  env=os.environ.copy()
-  for key in ('OPENPILOT_PREFIX','PARAMS_ROOT','ZMQ'):env.pop(key,None)
+def real_offroad(params=Path('/data/params/d')):
+  # Native Params uses /data/params + /d and getBool compares exact bytes to "1".
+  # Explicit paths deliberately ignore the replay process's Params namespace.
   try:
-    code='from openpilot.common.params import Params; p=Params(); print(int(p.get_bool("IsOffroad") and not p.get_bool("IsOnroad")))'
-    return subprocess.check_output(['/usr/local/venv/bin/python','-c',code],cwd='/data/openpilot',env=env,text=True,timeout=3).strip()=='1'
-  except (OSError,subprocess.SubprocessError):return False
+    return (params/'IsOffroad').read_bytes()==b'1' and (params/'IsOnroad').read_bytes()==b'0'
+  except OSError:return False
+
+
+def real_bluetooth_selection(params=Path('/data/params/d')):
+  try:
+    return {'enabled':(params/'BluetoothEnabled').read_bytes()==b'1',
+            'address':(params/'BluetoothAudioAddress').read_text().strip()}
+  except (OSError,UnicodeError):return {'enabled':False,'address':''}
 
 
 def describe_output(selected,status):
@@ -70,10 +76,9 @@ def describe_output(selected,status):
 
 
 def selected_output():
-  from bluetooth_output import real_selection
   try:
     from openpilot.starpilot.system.bluetooth.protocol import BluetoothClient
-    selected=real_selection()
+    selected=real_bluetooth_selection()
     status=BluetoothClient(timeout=2).status()
     return describe_output(selected,BluetoothClient.serialize_status(status))
   except (OSError,ValueError,ImportError,subprocess.SubprocessError):return None
@@ -173,8 +178,8 @@ class OutputOwner:
     if not self.offroad():
       raise ValueError('Park before changing RoadScore')
 
-  def status(self):
-    output = self.output_provider()
+  def status(self, output=None, supplied=False):
+    output = output if supplied else self.output_provider()
     muted = (self.root / '.session-muted').exists() or os.environ.get('ROADSCORE_FORCE_MUTE') == '1'
     active = playback_process_active() or busy(self.root / 'generated/session.lock')
     worker = read(self.root / 'generated/ace_worker_state.json')
@@ -203,14 +208,15 @@ class OutputOwner:
         session['operator_lease'].__exit__(None, None, None)
 
   def _check_session(self, token):
+    # Potentially slow BlueZ IPC must not serialize the tap/poll handlers.
+    output=self.output_provider();parked=self.offroad();playing=playback_process_active()
     with self.lock:
       session = self.session
       if session is None or session['token'] != token:
         return False
-      output = self.output_provider()
-      if (not self.offroad() or self.clock() > session['deadline'] or
+      if (not parked or self.clock() > session['deadline'] or
           self.clock() - session['last_client'] > 7 or not output or output['id'] != session['output']['id'] or
-          output['muted'] or not output['connected'] or playback_process_active() or session['sink'].failed or
+          output['muted'] or not output['connected'] or playing or session['sink'].failed or
           (self.root / '.session-muted').exists() or os.environ.get('ROADSCORE_FORCE_MUTE') == '1'):
         self._close()
         return False
@@ -224,9 +230,10 @@ class OutputOwner:
         return
 
   def dispatch(self, action, **data):
+    if action=='status':
+      output=self.output_provider()
+      with self.lock:return self.status(output=output,supplied=True)
     with self.lock:
-      if action == 'status':
-        return self.status()
       if action == 'clock':
         return dict(ok=True, server_ms=self.clock() * 1000)
       if action == 'calibration_cancel':

roadscore/prototype/test_operator_output.py

@@ -131,6 +131,33 @@ class OutputTests(unittest.TestCase):
     failure=json.loads((self.root/'generated/calibration_failure.json').read_text())
     self.assertIn('PortAudioError',failure['error'])
 
+  def test_real_params_reads_ignore_replay_namespace_and_fail_closed(self):
+    import os
+    params=self.root/'real';params.mkdir()
+    with patch.dict(os.environ,{'OPENPILOT_PREFIX':'replay','PARAMS_ROOT':'/not-real'}),patch.object(m.subprocess,'check_output',side_effect=AssertionError('No subprocess')):
+      self.assertFalse(m.real_offroad(params))
+      (params/'IsOffroad').write_bytes(b'1');(params/'IsOnroad').write_bytes(b'0')
+      self.assertTrue(m.real_offroad(params))
+      (params/'IsOnroad').write_bytes(b'1');self.assertFalse(m.real_offroad(params))
+      (params/'BluetoothEnabled').write_bytes(b'1');(params/'BluetoothAudioAddress').write_text('AA:BB:CC:DD:EE:FF')
+      self.assertEqual(m.real_bluetooth_selection(params)['address'],'AA:BB:CC:DD:EE:FF')
+
+  def test_slow_watchdog_output_lookup_does_not_block_taps(self):
+    import threading,time
+    token=self.owner.dispatch('calibration_start',attended=True)['session']
+    for i in range(9):self.owner.session['sink'].callback(i,100000+i*600)
+    self.clock.value=105.
+    entered=threading.Event();release=threading.Event()
+    def slow():entered.set();release.wait(2);return self.output
+    self.owner.output_provider=slow
+    thread=threading.Thread(target=self.owner._check_session,args=(token,));thread.start();self.assertTrue(entered.wait(1))
+    start=time.perf_counter()
+    self.owner.dispatch('calibration_poll',session=token)
+    self.owner.dispatch('calibration_tap',session=token,server_ms=105000,uncertainty_ms=3)
+    elapsed=time.perf_counter()-start
+    release.set();thread.join(2)
+    self.assertLess(elapsed,.1)
+
   def test_selected_speaker_identity_never_falls_back(self):
     selected=dict(enabled=True,address='AA:BB:CC:DD:EE:FF')
     status=dict(enabled=True,powered=True,devices=[dict(address='11:22:33:44:55:66',name='Other',connected=True,audio=True)])