From ae87804add0707c0fc3d999e020eedaa989bc2ed Mon Sep 17 00:00:00 2001
From: firestar5683 <168790843+firestar5683@users.noreply.github.com>
Date: Sat, 19 Sep 2026 19:46:40 -0700
Subject: [PATCH] Extend timing alignment test to three measures
---
roadscore/prototype/operator_click_process.py | 10 +++++-----
roadscore/prototype/operator_output.py | 5 +++--
roadscore/prototype/test_operator_output.py | 6 ++++--
starpilot/system/the_galaxy/assets/mobile/index.html | 2 +-
starpilot/system/the_galaxy/assets/mobile/js/app.js | 2 +-
.../the_galaxy/assets/mobile/js/views/RoadScore.js | 4 ++--
6 files changed, 16 insertions(+), 13 deletions(-)
diff --git a/roadscore/prototype/operator_click_process.py b/roadscore/prototype/operator_click_process.py
index c1bf08c643..772ecb9024 100644
--- a/roadscore/prototype/operator_click_process.py
+++ b/roadscore/prototype/operator_click_process.py
@@ -1,7 +1,7 @@
"""Explicitly attended calibration process; no import-time audio output."""
import time
RATE=48000
-from operator_output import COUNT, COUNT_IN, INTERVAL, MARKER_OFFSETS
+from operator_output import COUNT, COUNT_IN, TEST_COUNT, INTERVAL, MARKER_OFFSETS
class ClickSequence:
"""Prebuilt low-level click PCM; callback only copies memory and timestamps."""
@@ -13,8 +13,8 @@ class ClickSequence:
self.index = 0
self.failed = False
start = 2.0
- if count not in (4,COUNT):raise ValueError('Unsupported calibration sequence')
- offsets=[start+i*INTERVAL for i in range(4 if count==4 else COUNT_IN)]
+ if count not in (TEST_COUNT,COUNT):raise ValueError('Unsupported calibration sequence')
+ offsets=[start+i*INTERVAL for i in range(TEST_COUNT if count==TEST_COUNT else COUNT_IN)]
if count==COUNT:offsets+=list(MARKER_OFFSETS)
self.beats = [round(at * RATE) for at in offsets]
self.pcm = np.zeros((self.beats[-1] + RATE, 2), dtype='float32')
@@ -28,7 +28,7 @@ class ClickSequence:
second = round(.08 * RATE)
marker[second:second + len(tone_t)] = .07 * np.sin(2 * np.pi * 1760 * tone_t) * envelope
for index,beat in enumerate(self.beats):
- click = marker if index >= COUNT_IN else downbeat if index % 4 == 0 else beat_click
+ click = marker if count == COUNT and index >= COUNT_IN else downbeat if index % 4 == 0 else beat_click
self.pcm[beat:beat + len(click)] = click[:, None]
self.stream = sd.OutputStream(device=device, samplerate=RATE, channels=2, dtype='float32', blocksize=480,
callback=self.callback)
@@ -60,7 +60,7 @@ def main():
from pathlib import Path
from bluetooth_output import prepare_output,select_device
from operator_output import real_offroad,selected_output
- p=argparse.ArgumentParser();p.add_argument('--address',required=True);p.add_argument('--count',type=int,choices=[4,COUNT],required=True);args=p.parse_args()
+ p=argparse.ArgumentParser();p.add_argument('--address',required=True);p.add_argument('--count',type=int,choices=[TEST_COUNT,COUNT],required=True);args=p.parse_args()
output=selected_output()
if not real_offroad() or not output or not output['connected'] or output['address']!=args.address:
raise RuntimeError('Park and reconnect the selected Bluetooth speaker before calibration')
diff --git a/roadscore/prototype/operator_output.py b/roadscore/prototype/operator_output.py
index cfcb8e3388..91100ee0fb 100644
--- a/roadscore/prototype/operator_output.py
+++ b/roadscore/prototype/operator_output.py
@@ -18,6 +18,7 @@ import time
BPM = 100
INTERVAL = 60 / BPM
+TEST_COUNT = 12
COUNT_IN = 8
COUNT = 20
MARKER_COUNT = COUNT - COUNT_IN
@@ -268,7 +269,7 @@ class OutputOwner:
token = secrets.token_hex(16)
clicks = {}
try:
- sink = self.sink_factory(lambda index, at: clicks.__setitem__(index, at), count=4 if action == 'test' else COUNT)
+ sink = self.sink_factory(lambda index, at: clicks.__setitem__(index, at), count=TEST_COUNT if action == 'test' else COUNT)
if self.output_provider() != state['output']:
sink.close()
raise ValueError('Output changed before calibration began')
@@ -283,7 +284,7 @@ class OutputOwner:
session_lease.__exit__(None, None, None); operator.__exit__(None, None, None)
raise
threading.Thread(target=self._watch, args=(token,), daemon=True).start()
- return dict(ok=True, session=token, interval_ms=round(INTERVAL * 1000), beats=4 if action=='test' else COUNT,
+ return dict(ok=True, session=token, interval_ms=round(INTERVAL * 1000), beats=TEST_COUNT if action=='test' else COUNT,
count_in=COUNT_IN,bpm=BPM,beats_per_bar=4,method=METHOD,marker_count=MARKER_COUNT,target_taps=MARKER_COUNT,
marker_offsets_ms=[round(at*1000) for at in MARKER_OFFSETS],
instructions='Listen to two bars without tapping. Then tap once at the START of each two-tone marker; wait through the silence. The estimate includes your reaction time.')
diff --git a/roadscore/prototype/test_operator_output.py b/roadscore/prototype/test_operator_output.py
index ddc4d12167..a16dbe3601 100644
--- a/roadscore/prototype/test_operator_output.py
+++ b/roadscore/prototype/test_operator_output.py
@@ -211,10 +211,12 @@ class OutputTests(unittest.TestCase):
stamps=[]
with patch.dict(sys.modules,{'sounddevice':fake}):
from operator_click_process import ClickSequence
- sink=ClickSequence(lambda i,at:stamps.append((i,at)),0,count=4)
- second=ClickSequence(lambda *_:None,0,count=4)
+ sink=ClickSequence(lambda i,at:stamps.append((i,at)),0,count=m.TEST_COUNT)
+ second=ClickSequence(lambda *_:None,0,count=m.TEST_COUNT)
markers=ClickSequence(lambda *_:None,0,count=m.COUNT)
markers_again=ClickSequence(lambda *_:None,0,count=m.COUNT)
+ self.assertEqual(len(sink.beats), 12)
+ self.assertTrue(all(b-a == 28800 for a,b in zip(sink.beats,sink.beats[1:])))
self.assertTrue(np.array_equal(sink.pcm,second.pcm))
self.assertTrue(np.array_equal(markers.pcm,markers_again.pcm))
self.assertEqual(len(markers.beats),20)
diff --git a/starpilot/system/the_galaxy/assets/mobile/index.html b/starpilot/system/the_galaxy/assets/mobile/index.html
index ecbc12ada3..21222bc5e5 100644
--- a/starpilot/system/the_galaxy/assets/mobile/index.html
+++ b/starpilot/system/the_galaxy/assets/mobile/index.html
@@ -36,7 +36,7 @@
-
+