From 5f8c535bc15912256d50b6749beb04c4ac3c2d71 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Sat, 19 Sep 2026 19:38:20 -0700 Subject: [PATCH] Measure Bluetooth tap offsets with isolated irregular markers --- roadscore/prototype/operator_click_process.py | 19 ++++++-- roadscore/prototype/operator_output.py | 42 ++++++++++++----- roadscore/prototype/test_operator_output.py | 46 +++++++++++++++++-- 3 files changed, 88 insertions(+), 19 deletions(-) diff --git a/roadscore/prototype/operator_click_process.py b/roadscore/prototype/operator_click_process.py index 77c797fe61..c1bf08c643 100644 --- a/roadscore/prototype/operator_click_process.py +++ b/roadscore/prototype/operator_click_process.py @@ -1,11 +1,11 @@ """Explicitly attended calibration process; no import-time audio output.""" import time RATE=48000 -INTERVAL=.6 +from operator_output import COUNT, COUNT_IN, INTERVAL, MARKER_OFFSETS class ClickSequence: """Prebuilt low-level click PCM; callback only copies memory and timestamps.""" - def __init__(self, on_click, device, count=24): + def __init__(self, on_click, device, count=COUNT): import numpy as np import sounddevice as sd self.sd = sd @@ -13,13 +13,22 @@ class ClickSequence: self.index = 0 self.failed = False start = 2.0 - self.beats = [round((start + i * INTERVAL) * RATE) for i in range(count)] + 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==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') t = np.arange(round(.018 * RATE)) / RATE downbeat = (.07 * np.sin(2 * np.pi * 1760 * t) * np.exp(-t * 180)).astype('float32') beat_click = (.0455 * np.sin(2 * np.pi * 880 * t) * np.exp(-t * 180)).astype('float32') + tone_t = np.arange(round(.045 * RATE)) / RATE + envelope = np.sin(np.pi * np.arange(len(tone_t)) / len(tone_t)) ** 2 + marker = np.zeros(round(.13 * RATE), dtype='float32') + marker[:len(tone_t)] = .07 * np.sin(2 * np.pi * 880 * tone_t) * envelope + 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 = downbeat if index % 4 == 0 else beat_click + click = marker if 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) @@ -51,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,24],required=True);args=p.parse_args() + p=argparse.ArgumentParser();p.add_argument('--address',required=True);p.add_argument('--count',type=int,choices=[4,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 136e803342..cfcb8e3388 100644 --- a/roadscore/prototype/operator_output.py +++ b/roadscore/prototype/operator_output.py @@ -19,7 +19,13 @@ import time BPM = 100 INTERVAL = 60 / BPM COUNT_IN = 8 -COUNT = 24 +COUNT = 20 +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) +MARKER_OFFSETS = (8.6,) +for _interval in MARKER_INTERVALS: + MARKER_OFFSETS += (round(MARKER_OFFSETS[-1] + _interval, 3),) RATE = 48000 @@ -102,13 +108,13 @@ def correction(root, identity): def robust_offset(pairs): if len(pairs) < 8: - raise ValueError('At least eight different beats are required') + raise ValueError('At least eight different markers are required') differences = [tap - click for click, tap in pairs] center = median(differences) mad = median(abs(value - center) for value in differences) accepted = [value for value in differences if abs(value - center) <= max(35, 4.4478 * mad)] if len(accepted) < 8 or len(accepted) < len(pairs) * .65: - raise ValueError('Tap rhythm was inconsistent; retry') + raise ValueError('Tap timing was inconsistent; retry') spread = median(abs(value - median(accepted)) for value in accepted) if spread > 80: raise ValueError('Tap timing varied too much; retry') @@ -268,7 +274,7 @@ class OutputOwner: 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 40), test=action == 'test') + deadline=self.clock() + (15 if action == 'test' else 60), test=action == 'test') sink.start() except BaseException: if self.session: @@ -277,7 +283,10 @@ 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=COUNT,count_in=COUNT_IN,bpm=BPM,beats_per_bar=4) + return dict(ok=True, session=token, interval_ms=round(INTERVAL * 1000), beats=4 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.') if action == 'set_latency': value = data.get('latency_ms') if type(value) is not int or not 0 <= value <= 1500: @@ -304,7 +313,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} for index, at in list(session['clicks'].items())], test=session['test']) + 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']) 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: @@ -312,23 +321,34 @@ class OutputOwner: if abs(at - self.clock() * 1000) > 2000: 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 COUNT_IN not in session['clicks'] or at=COUNT or index not in session['clicks']: - raise ValueError('Wait for the next beat') + raise ValueError('Wait for the next two-tone marker') click=session['clicks'][index] previous=max((tap for _,tap in session['taps'].values()),default=float('-inf')) if at-previous1.5 for value in np.diff(markers.beats[8:])/48000)) + self.assertGreater(len(set(np.diff(markers.beats[8:]))),3) + for beat,next_beat in zip(markers.beats[8:],markers.beats[9:]): + self.assertTrue(np.any(markers.pcm[beat:beat+2160])) + self.assertTrue(np.any(markers.pcm[beat+3840:beat+6000])) + self.assertFalse(np.any(markers.pcm[beat+6240:next_beat])) self.assertLessEqual(abs(sink.pcm).max(),.071) self.assertEqual(sink.beats[1]-sink.beats[0],28800) self.assertGreater(abs(sink.pcm[sink.beats[0]:sink.beats[0]+1000]).max(),abs(sink.pcm[sink.beats[1]:sink.beats[1]+1000]).max())