Connect Galaxy live toggle to gated owned service with unconditional stop

starpilot/system/the_galaxy/assets/mobile/js/views/RoadScore.js

@@ -2,10 +2,19 @@ import { GalaxySection } from '../components/GalaxySection.js'
 
 export const RoadScore = {
   name: 'RoadScore', components: { GalaxySection },
-  data() { return { status: {}, error: '', busy: false, profile: 'prism', latency: 0, session: null, taps: 0, result: null, timer: null, calibrationTimer: null, clockOffset: 0, uncertainty: null, testMode: false, beat: '', sessionStarted: 0, latencyInitialized: false, outputIdentity: null, testClicks: [], animation: null, countIn: 8, bpm: 100, beatIndex: -1, cue: 'Getting ready…', lastTapInput: -1000, targetTaps: 12, finishing: false, calibrationStage: 'markers', coarseResult: null } },
+  data() { return { status: {}, error: '', busy: false, liveStarting: false, profile: 'prism', latency: 0, session: null, taps: 0, result: null, timer: null, calibrationTimer: null, clockOffset: 0, uncertainty: null, testMode: false, beat: '', sessionStarted: 0, latencyInitialized: false, outputIdentity: null, testClicks: [], animation: null, countIn: 8, bpm: 100, beatIndex: -1, cue: 'Getting ready…', lastTapInput: -1000, targetTaps: 12, finishing: false, calibrationStage: 'markers', coarseResult: null } },
   mounted() { window.addEventListener('keydown', this.keyTap); this.refresh(); this.timer = setInterval(() => this.refresh(), 2000) },
   beforeUnmount() { window.removeEventListener('keydown', this.keyTap); clearInterval(this.timer); clearInterval(this.calibrationTimer); cancelAnimationFrame(this.animation); if (this.session) this.action('calibration_cancel', {session: this.session}) },
   methods: {
+    async setLive(enabled, event) {
+      if (event) event.target.checked = this.status.live?.enabled === true
+      if (enabled) this.liveStarting = true
+      this.error = ''
+      try {
+        await this.quick('live', {enabled})
+      } catch (error) { this.error = error.message }
+      finally { this.liveStarting = false; await this.refresh() }
+    },
     async refresh() {
       try {
         const response = await fetch('/api/roadscore/status', {cache: 'no-store'})
@@ -119,6 +128,13 @@ export const RoadScore = {
     <div class="gx-view">
       <h2 style="margin-top:0"><i class="bi bi-music-note-beamed"></i> RoadScore</h2>
       <p style="color:var(--text-muted)">Music shaped by the road.</p>
+      <GalaxySection title="Live music" icon="bi-music-note-beamed" :collapsible="false"><div style="padding:16px">
+        <div class="gx-row"><div><strong>RoadScore</strong><div class="gx-row__desc">{{ status.live?.state || 'UNAVAILABLE' }}</div></div>
+          <label class="gx-switch"><input type="checkbox" aria-label="Enable live RoadScore" :checked="status.live?.enabled === true" :disabled="!status.live?.available || (!status.live?.enabled && (!status.live?.can_enable || liveStarting))" @change="setLive($event.target.checked, $event)"><span class="gx-switch__track"></span><span class="gx-switch__thumb"></span></label>
+        </div>
+        <p class="gx-row__desc">{{status.live?.reason || 'Live driving readiness has not been confirmed.'}}</p>
+        <button class="gx-btn" @click="setLive(false)">Stop RoadScore</button>
+      </div></GalaxySection>
       <GalaxySection title="Composer" icon="bi-music-note-beamed" :collapsible="false"><div style="padding:16px">
         <div class="gx-row"><div><strong>{{ status.state || 'UNAVAILABLE' }}</strong><div class="gx-row__desc">{{ status.composer ? status.composer.toUpperCase() : 'Composer not connected' }}{{ status.backend ? ' · ' + status.backend : '' }}{{ status.composer && status.profile ? ' · ' + status.profile.toUpperCase() : '' }}</div></div></div>
         <div class="gx-row"><label for="roadscore-style">Next style</label><select class="gx-field" id="roadscore-style" v-model="profile" :disabled="busy || !status.can_edit" @change="action('settings', {profile})"><option v-for="p in status.profiles || []" :value="p.id">{{p.name}}</option></select></div>

starpilot/system/the_galaxy/roadscore.py

@@ -99,8 +99,35 @@ class Operator:
     self.preparing = False
     self.error = None
     self.output_owner = None
+    self.live_owner = None
+    self.live_init_lock = threading.Lock()
     self.output_init_lock = threading.Lock()
 
+  def live_controller(self):
+    if not self.device:
+      raise ValueError('Live RoadScore is available only on the comma')
+    with self.live_init_lock:
+      if self.live_owner is None:
+        path = self.root / 'prototype/live_controller.py'
+        if not path.is_file():
+          raise ValueError('Live RoadScore is not available on this installation')
+        if str(path.parent) not in sys.path:sys.path.insert(0, str(path.parent))
+        spec = importlib.util.spec_from_file_location('roadscore_live_controller', path)
+        module = importlib.util.module_from_spec(spec)
+        sys.modules[spec.name] = module
+        spec.loader.exec_module(module)
+        self.live_owner = module.LiveController(root=self.root)
+    return self.live_owner
+
+  def live_status(self):
+    try:
+      value = self.live_controller().status()
+      if not isinstance(value, dict):raise ValueError('Invalid live status')
+      return value
+    except (OSError, ValueError, ImportError):
+      return {'available': False, 'enabled': None, 'state': 'UNAVAILABLE', 'can_enable': False,
+              'reason': 'Live status cannot be verified. Enable is unavailable.'}
+
   def target(self, action, **payload):
     if not self.device:
       raise ValueError('Target output is available only on the comma')
@@ -129,7 +156,7 @@ class Operator:
     if live and (self.root/'generated/busy').exists():state='GENERATING'
     if output.get('state') in STATES:
       state = output['state']
-    return dict(available=self.device and self.root.exists(), state=state, profiles=PROFILES,
+    return dict(available=self.device and self.root.exists(), state=state, profiles=PROFILES, live=self.live_status(),
                 profile=worker.get('profile') if live else settings.get('profile', 'prism'),
                 selected_profile=settings.get('profile', 'prism'),
                 composer='ace' if live else None, backend='Chestnut' if live else None,
@@ -141,6 +168,15 @@ class Operator:
                 calibrating=output.get('calibrating', False), session_muted=output.get('session_muted', True), output=output.get('output'), latency_ms=output.get('latency_ms'), error=self.error or output.get('error'))
 
   def operate(self, action, data, offroad):
+    if action == 'live':
+      if set(data) != {'enabled'} or type(data['enabled']) is not bool:
+        raise ValueError('Live RoadScore requires an enabled boolean')
+      controller = self.live_controller()
+      if data['enabled']:
+        status = controller.status()
+        if status.get('available') is not True or status.get('can_enable') is not True:
+          raise ValueError(status.get('reason') or 'Live driving readiness has not been confirmed')
+      return controller.set_enabled(data['enabled'])
     if not offroad and action != 'calibration_cancel':
       raise ValueError('RoadScore controls are available while parked')
     with self.lock:

starpilot/system/the_galaxy/tests/test_roadscore_live.py

@@ -0,0 +1,59 @@
+import importlib.util
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import Mock
+
+spec = importlib.util.spec_from_file_location('galaxy_live_operator', Path(__file__).resolve().parents[1] / 'roadscore.py')
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+
+class LiveTests(unittest.TestCase):
+  def setUp(self):
+    self.folder = tempfile.TemporaryDirectory()
+    self.addCleanup(self.folder.cleanup)
+    self.operator = module.Operator(root=Path(self.folder.name), device=True)
+    self.controller = Mock()
+    self.controller.status.return_value = {'available': True, 'enabled': False, 'can_enable': False, 'reason': 'Car baseline not authorized'}
+    self.operator.live_owner = self.controller
+    self.operator.target = Mock(side_effect=AssertionError('Bench/output owner must not be controlled'))
+
+  def test_off_is_allowed_onroad_without_authorization_or_status(self):
+    self.controller.status.side_effect = RuntimeError('Health snapshot unavailable')
+    self.controller.set_enabled.return_value = {'enabled': False, 'state': 'STOPPING'}
+    self.assertEqual(self.operator.operate('live', {'enabled': False}, False)['state'], 'STOPPING')
+    self.controller.status.assert_not_called()
+    self.controller.set_enabled.assert_called_once_with(False)
+    self.operator.target.assert_not_called()
+
+  def test_on_rejects_missing_authorization(self):
+    with self.assertRaisesRegex(ValueError, 'Car baseline'):
+      self.operator.operate('live', {'enabled': True}, False)
+    self.controller.set_enabled.assert_not_called()
+
+  def test_on_uses_controller_readiness_not_offroad_parameter(self):
+    self.controller.status.return_value.update(can_enable=True)
+    self.controller.set_enabled.return_value = {'enabled': True, 'state': 'PREPARING'}
+    result = self.operator.operate('live', {'enabled': True}, False)
+    self.assertEqual(result['state'], 'PREPARING')
+    self.controller.set_enabled.assert_called_once_with(True)
+
+  def test_invalid_payload_never_reaches_controller(self):
+    for data in ({}, {'enabled': 1}, {'enabled': 'true'}, {'enabled': False, 'pid': 123}):
+      with self.assertRaises(ValueError):self.operator.operate('live', data, True)
+    self.controller.set_enabled.assert_not_called()
+
+  def test_missing_backend_has_unknown_enabled_state(self):
+    self.operator.live_owner = None
+    state = self.operator.live_status()
+    self.assertFalse(state['available']);self.assertFalse(state['can_enable'])
+    self.assertIsNone(state['enabled'])
+    with self.assertRaisesRegex(ValueError, 'not available'):
+      self.operator.operate('live', {'enabled': True}, True)
+
+  def test_other_controls_still_require_parked(self):
+    with self.assertRaisesRegex(ValueError, 'parked'):
+      self.operator.operate('prepare', {}, False)
+    self.controller.set_enabled.assert_not_called()
+
+if __name__ == '__main__':unittest.main()