diff --git a/starpilot/system/the_galaxy/assets/mobile/js/views/RoadScore.js b/starpilot/system/the_galaxy/assets/mobile/js/views/RoadScore.js index 4462393b41..de2c60cc91 100644 --- a/starpilot/system/the_galaxy/assets/mobile/js/views/RoadScore.js +++ b/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 = {

RoadScore

Music shaped by the road.

+
+
RoadScore
{{ status.live?.state || 'UNAVAILABLE' }}
+ +
+

{{status.live?.reason || 'Live driving readiness has not been confirmed.'}}

+ +
{{ status.state || 'UNAVAILABLE' }}
{{ status.composer ? status.composer.toUpperCase() : 'Composer not connected' }}{{ status.backend ? ' · ' + status.backend : '' }}{{ status.composer && status.profile ? ' · ' + status.profile.toUpperCase() : '' }}
diff --git a/starpilot/system/the_galaxy/roadscore.py b/starpilot/system/the_galaxy/roadscore.py index a4000e2a61..bde8667ab8 100644 --- a/starpilot/system/the_galaxy/roadscore.py +++ b/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: diff --git a/starpilot/system/the_galaxy/tests/test_roadscore_live.py b/starpilot/system/the_galaxy/tests/test_roadscore_live.py new file mode 100644 index 0000000000..d3310eeff0 --- /dev/null +++ b/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()