diff --git a/roadscore/prototype/operator_click_process.py b/roadscore/prototype/operator_click_process.py index 772ecb9024..52786110b9 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, TEST_COUNT, INTERVAL, MARKER_OFFSETS +from operator_output import COUNT, COUNT_IN, REFINE_COUNT, 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 (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 not in (TEST_COUNT,COUNT,REFINE_COUNT):raise ValueError('Unsupported calibration sequence') + offsets=[start+i*INTERVAL for i in range(COUNT_IN if count==COUNT else count)] 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') @@ -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=[TEST_COUNT,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,REFINE_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 91100ee0fb..8a1260ff21 100644 --- a/roadscore/prototype/operator_output.py +++ b/roadscore/prototype/operator_output.py @@ -21,6 +21,8 @@ INTERVAL = 60 / BPM TEST_COUNT = 12 COUNT_IN = 8 COUNT = 20 +REFINE_COUNT = 24 +REFINE_METHOD = 'coarse-anchored-rhythm-v1' MARKER_COUNT = COUNT - COUNT_IN METHOD = 'irregular-marker-reaction-v1' MARKER_INTERVALS = (1.8, 2.4, 2.1, 2.7, 1.9, 2.5, 2.2, 2.8, 2.0, 2.6, 2.3) @@ -124,6 +126,20 @@ def robust_offset(pairs): rejected_taps=len(pairs) - len(accepted), spread_ms=round(spread), includes_human_tap_bias=True) +def rhythm_pair(clicks, at, coarse): + interval_ms=INTERVAL*1000 + margin=max(60,3*coarse.get('spread_ms',0)) + if margin>=interval_ms/2:raise ValueError('Coarse estimate is too uncertain; repeat the chimes') + approximate=at-coarse['latency_ms'] + candidates=[(index,click) for index,click in clicks.items() if COUNT_IN<=index=interval_ms/2-margin: + raise ValueError('Whole-beat choice is ambiguous; repeat the chimes, then tap with the rhythm') + if not 0<=at-click<=1500:raise ValueError('Rhythmic offset is outside 0–1500 ms; repeat calibration') + return index,click + + class ClickSink: """Own a separate local process so each session binds ALSA before PortAudio loads.""" def __init__(self,on_click,count=COUNT): @@ -249,7 +265,7 @@ class OutputOwner: return dict(ok=True) raise ValueError('Calibration expired; start again') self.safe() - if action in ('calibration_start', 'test'): + if action in ('calibration_start', 'calibration_refine', 'test'): if data != {'attended': True}: raise ValueError('Confirm attended audio') if self.session: @@ -257,6 +273,14 @@ class OutputOwner: state = self.status() if state['judging_locked'] or state['playback_active'] or not state['calibration']: raise ValueError('Audio is busy, muted or unavailable') + coarse=None + if action=='calibration_refine': + coarse=read(self.root/'generated/calibration_coarse_result.json') or read(self.root/'generated/calibration_latest_result.json') + if (coarse.get('method')!=METHOD or coarse.get('output',{}).get('id')!=state['output']['id'] or + not 0<=time.time()-coarse.get('created_wall',0)<=600 or type(coarse.get('latency_ms')) is not int or + not 0<=coarse['latency_ms']<=1500 or coarse.get('accepted_taps',0)<8): + raise ValueError('First complete the chimes on this speaker; coarse estimates expire after ten minutes') + count=TEST_COUNT if action=='test' else REFINE_COUNT if coarse else COUNT operator = self.operator_lease() operator.__enter__() session_lease = lease(self.root / 'generated/session.lock') @@ -269,13 +293,13 @@ class OutputOwner: token = secrets.token_hex(16) clicks = {} try: - sink = self.sink_factory(lambda index, at: clicks.__setitem__(index, at), count=TEST_COUNT if action == 'test' else COUNT) + sink = self.sink_factory(lambda index, at: clicks.__setitem__(index, at), count=count) if self.output_provider() != state['output']: sink.close() raise ValueError('Output changed before calibration began') self.session = dict(token=token, sink=sink, clicks=clicks, taps={}, output=state['output'], operator_lease=operator, session_lease=session_lease, last_client=self.clock(), - deadline=self.clock() + (15 if action == 'test' else 60), test=action == 'test') + deadline=self.clock() + (15 if action == 'test' else 60), test=action == 'test', coarse=coarse) sink.start() except BaseException: if self.session: @@ -284,10 +308,13 @@ 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=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.') + return dict(ok=True, session=token, interval_ms=round(INTERVAL * 1000), beats=count, + count_in=COUNT_IN,bpm=BPM,beats_per_bar=4,method=REFINE_METHOD if coarse else METHOD, + marker_count=0 if coarse else MARKER_COUNT,target_taps=16 if coarse else MARKER_COUNT, + min_taps=8 if coarse else MARKER_COUNT,coarse_offset_ms=coarse['latency_ms'] if coarse else None, + marker_offsets_ms=[] if coarse else [round(at*1000) for at in MARKER_OFFSETS], + instructions=('Listen for two bars, then tap steadily with the heard beat. Missed beats are allowed.' if coarse else + '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.')) if action == 'set_latency': value = data.get('latency_ms') if type(value) is not int or not 0 <= value <= 1500: @@ -314,7 +341,7 @@ class OutputOwner: self._close(); return dict(ok=True) if action == 'calibration_poll': delay = correction(self.root, session['output']['id']) if session['test'] else 0 - return dict(ok=True, clicks=[{'beat': index, 'server_ms': at + delay, 'kind':'marker' if index>=COUNT_IN and not session['test'] else 'count_in'} for index, at in list(session['clicks'].items())], test=session['test']) + return dict(ok=True, clicks=[{'beat': index, 'server_ms': at + delay, 'kind':('rhythm' if session.get('coarse') else 'marker') if index>=COUNT_IN and not session['test'] else 'count_in'} for index, at in list(session['clicks'].items())], test=session['test']) if action == 'calibration_tap': at, uncertainty = data.get('server_ms'), data.get('uncertainty_ms') if type(at) not in (int, float) or not math.isfinite(at) or type(uncertainty) not in (int, float) or not 0 <= uncertainty <= 25: @@ -323,6 +350,11 @@ class OutputOwner: raise ValueError('Stale tap') if session['test']:raise ValueError('Test mode does not collect taps') if session.get('pairing_error'):raise ValueError(session['pairing_error']) + if session.get('coarse'): + index,click=rhythm_pair(session['clicks'],at,session['coarse']) + if index in session['taps']:raise ValueError('This beat already has a tap') + session['taps'][index]=(click,at) + return dict(ok=True,accepted_taps=len(session['taps']),paired_beat=index) if COUNT_IN not in session['clicks'] or at=interval_ms/2-margin or not 0<=branch<=1500: + raise ValueError('Whole-beat choice is ambiguous; repeat the chimes and rhythm') + result.update(latency_ms=branch,raw_offset_ms=branch,method=REFINE_METHOD,marker_count=0, + beat_pairing='nearest coarse-corrected rhythmic beat; no sequential tap shift', + whole_beat_ambiguity_possible=True, + phase_offset_ms=phase,coarse_offset_ms=coarse['latency_ms'], + coarse_result=coarse,branch_margin_ms=interval_ms/2-abs(branch-coarse['latency_ms']), + measurement='Coarse-anchored rhythmic tap estimate; human timing bias remains, not a physical latency measurement', + branch_assumption='Coarse reaction bias and rhythmic tap error must differ by less than half a beat; borderline estimates are rejected') + else: + path=self.root/'generated/calibration_coarse_result.json' + temp=path.with_suffix('.tmp');temp.write_text(json.dumps(result));temp.replace(path) path=self.root/'generated/calibration_latest_result.json' temp=path.with_suffix('.tmp');temp.write_text(json.dumps(result));temp.replace(path) finally: diff --git a/roadscore/prototype/test_operator_output.py b/roadscore/prototype/test_operator_output.py index a16dbe3601..6ffed86316 100644 --- a/roadscore/prototype/test_operator_output.py +++ b/roadscore/prototype/test_operator_output.py @@ -96,6 +96,56 @@ class OutputTests(unittest.TestCase): with self.assertRaisesRegex(ValueError,'all twelve'): self.owner.dispatch('calibration_result',session=token) + def coarse_fixture(self, offset=800, age=0, identity='speaker-a'): + import time + path=self.root/'generated/calibration_coarse_result.json';path.parent.mkdir(exist_ok=True) + path.write_text(json.dumps(dict(method=m.METHOD,output={'id':identity},created_wall=time.time()-age, + latency_ms=offset,spread_ms=4,accepted_taps=12))) + + def test_refine_requires_recent_same_speaker_coarse(self): + with self.assertRaisesRegex(ValueError,'First complete'): + self.owner.dispatch('calibration_refine',attended=True) + for age,identity in ((601,'speaker-a'),(0,'other')): + self.coarse_fixture(age=age,identity=identity) + with self.assertRaisesRegex(ValueError,'First complete'): + self.owner.dispatch('calibration_refine',attended=True) + self.coarse_fixture() + with self.assertRaisesRegex(ValueError,'attended'): + self.owner.dispatch('calibration_refine') + + def test_refine_recovers_branch_without_shift_on_missed_beats(self): + for coarse,actual in ((800,624),(200,24)): + self.coarse_fixture(coarse) + response=self.owner.dispatch('calibration_refine',attended=True);token=response['session'] + self.assertEqual(response['beats'],24);self.assertEqual(response['target_taps'],16) + self.assertEqual(self.owner.session['sink'].count,24) + for i in range(24):self.owner.session['sink'].callback(i,100000+i*600) + for index in (8,9,11,12,14,15,17,18,20,21,22,23): + at=100000+index*600+actual;self.clock.value=at/1000 + answer=self.owner.dispatch('calibration_tap',session=token,server_ms=at,uncertainty_ms=3) + self.assertEqual(answer['paired_beat'],index) + with self.assertRaisesRegex(ValueError,'already'): + self.owner.dispatch('calibration_tap',session=token,server_ms=at+2,uncertainty_ms=3) + result=self.owner.dispatch('calibration_result',session=token) + self.assertEqual(result['latency_ms'],actual);self.assertEqual(result['phase_offset_ms'],24) + self.assertEqual(result['method'],m.REFINE_METHOD);self.assertEqual(result['accepted_taps'],12) + self.assertEqual(result['coarse_result']['latency_ms'],coarse) + self.assertFalse((self.root/'generated/output_timing.json').exists()) + self.assertEqual(json.loads((self.root/'generated/calibration_latest_result.json').read_text())['latency_ms'],actual) + + def test_refine_rejects_half_beat_ambiguity_and_count_in(self): + self.coarse_fixture(324) + token=self.owner.dispatch('calibration_refine',attended=True)['session'] + for i in range(24):self.owner.session['sink'].callback(i,100000+i*600) + self.clock.value=105.124 + with self.assertRaisesRegex(ValueError,'ambiguous'): + self.owner.dispatch('calibration_tap',session=token,server_ms=104824,uncertainty_ms=3) + self.assertFalse(self.owner.session['taps']) + with self.assertRaisesRegex(ValueError,'ambiguous'): + self.owner.dispatch('calibration_tap',session=token,server_ms=104224,uncertainty_ms=3) + with self.assertRaisesRegex(ValueError,'eight'): + self.owner.dispatch('calibration_result',session=token) + def test_uncertain_clock_stale_token_and_judging_lock(self): token=self.owner.dispatch('calibration_start',attended=True)['session'] self.owner.session['sink'].callback(0,100000) @@ -215,11 +265,15 @@ class OutputTests(unittest.TestCase): 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) + refine=ClickSequence(lambda *_:None,0,count=m.REFINE_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) + self.assertEqual(len(refine.beats),24) + self.assertTrue(np.all(np.diff(refine.beats)==28800)) + self.assertFalse(np.any(refine.pcm[refine.beats[8]+2000:refine.beats[9]])) self.assertEqual(markers.beats[8],round(8.6*48000)) self.assertTrue(all(value>1.5 for value in np.diff(markers.beats[8:])/48000)) self.assertGreater(len(set(np.diff(markers.beats[8:]))),3) diff --git a/starpilot/system/the_galaxy/roadscore.py b/starpilot/system/the_galaxy/roadscore.py index f6b593cfe8..a4000e2a61 100644 --- a/starpilot/system/the_galaxy/roadscore.py +++ b/starpilot/system/the_galaxy/roadscore.py @@ -146,8 +146,8 @@ class Operator: with self.lock: if action == 'clock': return self.target('clock') - if action in {'calibration_start', 'calibration_tap', 'calibration_result', 'calibration_cancel', 'calibration_poll', 'test'}: - if action == 'calibration_start' and data != {'attended': True}: + if action in {'calibration_start', 'calibration_refine', 'calibration_tap', 'calibration_result', 'calibration_cancel', 'calibration_poll', 'test'}: + if action in {'calibration_start','calibration_refine'} and data != {'attended': True}: raise ValueError('Confirm that you are ready to hear the clicks') if action == 'calibration_tap': if set(data) != {'session', 'server_ms', 'uncertainty_ms'} or not isinstance(data['session'], str) or len(data['session']) > 100 or type(data['server_ms']) not in (int, float) or not 0 <= data['server_ms'] < 1e15: