mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-10-04 13:24:13 +08:00
Measure Bluetooth tap offsets with isolated irregular markers
This commit is contained in:
firestar5683
@@ -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')
|
||||
|
||||
@@ -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<session['clicks'][COUNT_IN]:
|
||||
raise ValueError('Listen for two full bars; start tapping on bar three')
|
||||
raise ValueError('Listen for two full bars; tap only the two-tone markers after the count-in')
|
||||
index=COUNT_IN+len(session['taps'])
|
||||
if index>=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-previous<INTERVAL*1000*.45:
|
||||
raise ValueError('Tap once per beat')
|
||||
raise ValueError('Tap once per marker')
|
||||
if not 0<=at-click<=1500:
|
||||
raise ValueError('Beat sequence was missed; restart with the two-bar count-in')
|
||||
session['pairing_error']='A marker was missed or tapped outside its 0–1500 ms window; restart the measurement'
|
||||
raise ValueError(session['pairing_error'])
|
||||
session['taps'][index] = (click, at)
|
||||
return dict(ok=True, accepted_taps=len(session['taps']))
|
||||
if action == 'calibration_result':
|
||||
try:
|
||||
if session.get('pairing_error'):raise ValueError(session['pairing_error'])
|
||||
if len(session['taps'])!=MARKER_COUNT:raise ValueError('Tap all twelve markers before finishing; restart if a marker was missed')
|
||||
result = robust_offset(list(session['taps'].values()))
|
||||
result.update(bpm=BPM,count_in_beats=COUNT_IN,beat_pairing='sequential after two-bar count-in',whole_beat_ambiguity_possible=True)
|
||||
result.update(method=METHOD,bpm=BPM,count_in_beats=COUNT_IN,marker_count=MARKER_COUNT,
|
||||
beat_pairing='sequential absolute marker timestamps; no modulo pairing',whole_beat_ambiguity_possible=False,
|
||||
measurement='Bluetooth/output delay plus human reaction time; not a physical latency measurement',
|
||||
output=session['output'],created_wall=time.time(),
|
||||
pairs=[dict(marker=index,server_ms=click,tap_ms=tap,offset_ms=tap-click)
|
||||
for index,(click,tap) in session['taps'].items()])
|
||||
path=self.root/'generated/calibration_latest_result.json'
|
||||
temp=path.with_suffix('.tmp');temp.write_text(json.dumps(result));temp.replace(path)
|
||||
finally:
|
||||
self._close()
|
||||
return dict(ok=True, **result)
|
||||
|
||||
@@ -52,21 +52,50 @@ class OutputTests(unittest.TestCase):
|
||||
token=self.owner.dispatch('calibration_start', attended=True)['session']
|
||||
for i in range(8):self.owner.session['sink'].callback(i,100000+i*600)
|
||||
with self.assertRaises(ValueError):self.owner.dispatch('calibration_tap',session=token,server_ms=100200,uncertainty_ms=3)
|
||||
offsets=[200,201,198,202,199,203,197,200,201,199,400,205,200,201,199,200]
|
||||
offsets=[200,201,198,202,199,203,197,200,201,199,400,205]
|
||||
for index, offset in enumerate(offsets):
|
||||
i=index+8;click=100000+i*600;self.clock.value=(click+offset)/1000
|
||||
i=index+8;click=100000+m.MARKER_OFFSETS[index]*1000;self.clock.value=(click+offset)/1000
|
||||
self.owner.session['sink'].callback(i,click)
|
||||
self.owner.dispatch('calibration_tap', session=token, server_ms=click+offset, uncertainty_ms=3)
|
||||
with self.assertRaises(ValueError):self.owner.dispatch('calibration_tap',session=token,server_ms=click+offset+5,uncertainty_ms=3)
|
||||
result=self.owner.dispatch('calibration_result',session=token)
|
||||
self.assertEqual(result['latency_ms'],200);self.assertEqual(result['rejected_taps'],1)
|
||||
self.assertEqual(result['count_in_beats'],8);self.assertTrue(result['whole_beat_ambiguity_possible'])
|
||||
self.assertEqual(result['count_in_beats'],8);self.assertFalse(result['whole_beat_ambiguity_possible'])
|
||||
saved=json.loads((self.root/'generated/calibration_latest_result.json').read_text())
|
||||
self.assertEqual(saved['output']['id'],'speaker-a');self.assertEqual(saved['method'],m.METHOD)
|
||||
self.assertTrue(saved['includes_human_tap_bias']);self.assertEqual(len(saved['pairs']),12)
|
||||
self.assertFalse((self.root/'generated/output_timing.json').exists())
|
||||
self.owner.dispatch('set_latency',latency_ms=result['latency_ms'])
|
||||
self.assertEqual(self.owner.status()['latency_ms'],200)
|
||||
self.output={**self.output,'id':'speaker-b'}
|
||||
self.assertEqual(self.owner.status()['latency_ms'],0)
|
||||
|
||||
def test_marker_pairing_distinguishes_whole_beat_offsets(self):
|
||||
for latency in (24,624):
|
||||
token=self.owner.dispatch('calibration_start',attended=True)['session']
|
||||
for index,offset in enumerate(m.MARKER_OFFSETS):
|
||||
click=100000+offset*1000;self.clock.value=(click+latency)/1000
|
||||
self.owner.session['sink'].callback(index+8,click)
|
||||
self.owner.dispatch('calibration_tap',session=token,server_ms=click+latency,uncertainty_ms=3)
|
||||
result=self.owner.dispatch('calibration_result',session=token)
|
||||
self.assertEqual(result['latency_ms'],latency)
|
||||
self.assertFalse((self.root/'generated/output_timing.json').exists())
|
||||
|
||||
def test_missed_marker_is_not_paired_to_next_or_partial_result(self):
|
||||
token=self.owner.dispatch('calibration_start',attended=True)['session']
|
||||
first=100000+m.MARKER_OFFSETS[0]*1000
|
||||
second=100000+m.MARKER_OFFSETS[1]*1000
|
||||
self.owner.session['sink'].callback(8,first);self.owner.session['sink'].callback(9,second)
|
||||
self.clock.value=(second+24)/1000
|
||||
with self.assertRaisesRegex(ValueError,'marker was missed'):
|
||||
self.owner.dispatch('calibration_tap',session=token,server_ms=second+24,uncertainty_ms=3)
|
||||
with self.assertRaisesRegex(ValueError,'marker was missed'):
|
||||
self.owner.dispatch('calibration_result',session=token)
|
||||
self.assertFalse((self.root/'generated/calibration_latest_result.json').exists())
|
||||
token=self.owner.dispatch('calibration_start',attended=True)['session']
|
||||
with self.assertRaisesRegex(ValueError,'all twelve'):
|
||||
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)
|
||||
@@ -184,7 +213,18 @@ class OutputTests(unittest.TestCase):
|
||||
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)
|
||||
markers=ClickSequence(lambda *_:None,0,count=m.COUNT)
|
||||
markers_again=ClickSequence(lambda *_:None,0,count=m.COUNT)
|
||||
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(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)
|
||||
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())
|
||||
|
||||
Reference in New Issue
Block a user