From b86a1b94753c27a192fbe4db95bc8786deaa8c40 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Sat, 19 Sep 2026 20:44:21 -0700 Subject: [PATCH] Give meaningful warnings bounded beat-locked fill priority --- roadscore/prototype/alert_accent.py | 55 +++++++++++++--------- roadscore/prototype/review_alert_accent.py | 54 +++++++++++++++++++++ roadscore/prototype/test_alert_accent.py | 20 ++++++-- 3 files changed, 105 insertions(+), 24 deletions(-) create mode 100644 roadscore/prototype/review_alert_accent.py diff --git a/roadscore/prototype/alert_accent.py b/roadscore/prototype/alert_accent.py index ffeff060c0..3379718b40 100644 --- a/roadscore/prototype/alert_accent.py +++ b/roadscore/prototype/alert_accent.py @@ -1,47 +1,60 @@ -"""Sparse native-alert accents using the signal shaker's percussion vocabulary.""" +"""Sparse beat-locked warning fills; bounded priority over routine musical cues.""" import math import numpy as np -from signal_shaker import SignalShaker class AlertAccent: def __init__(self, grid, rate=48000, enabled=False): self.grid=grid;self.rate=rate;self.enabled=enabled - self.grain=SignalShaker(grid,rate,peak=.016).grain + n=round(.11*rate);t=np.arange(n)/rate;rng=np.random.default_rng(1701) + frequency=np.fft.rfftfreq(n,1/rate);spectrum=np.fft.rfft(rng.standard_normal(n)) + spectrum*=np.clip((frequency-650)/700,0,1)*np.clip((6500-frequency)/2000,0,1) + grain=np.fft.irfft(spectrum,n);envelope=(1-np.exp(-t/.004))*np.exp(-t/.032) + grain*=envelope;grain/=max(float(np.max(np.abs(grain))),1e-9) + self.grain=np.column_stack((grain,grain)).astype(np.float32)*.045 + self.duck=(np.sin(np.pi*np.arange(n)/(n-1))**2).astype(np.float32) self.key=None;self.since=0;self.handled=False;self.pending=[] - self.last=-60*rate;self.events=[];self.rendered_active=False + self.last=-60*rate;self.events=[];self.rendered_active=False;self.priority_active=False def process(self, pcm, start, key, meaningful, fresh, competing=False): - self.rendered_active=False + self.rendered_active=False;self.priority_active=False if not self.enabled:return pcm current=key if meaningful and fresh else None if current!=self.key: - self.key=current;self.since=start;self.handled=False;self.pending=[] - if not fresh or competing: + self.key=current;self.since=start;self.handled=False + if not fresh: self.pending=[] - if current:self.handled=True return pcm - if self.grid.usable and current and not self.handled and start-self.since>=round(.2*self.rate): - self.handled=True - if start-self.last>=12*self.rate and sum(start-e['frame']<60*self.rate for e in self.events)<3: - beat=self.rate*60/self.grid.bpm;origin=self.grid.beat_phase*self.rate - frame=round(origin+math.ceil((start-origin)/beat-1e-10)*beat) - self.pending=[(frame,1.),(frame+round(beat/2),.55)] - self.last=start;self.events.append({'kind':'native_alert_accent','frame':start,'scheduled_frame':frame,'alert':current}) - end=start+len(pcm);overlay=None;remaining=[] + elapsed=(start-self.since)/self.rate + if self.grid.usable and current and not self.handled and elapsed>=.2: + # Ordinary cues can defer a warning briefly, but cannot permanently erase it. + if not competing or elapsed>=.7: + self.handled=True + if start-self.last>=12*self.rate and sum(start-e['frame']<60*self.rate for e in self.events)<3: + beat=self.rate*60/self.grid.bpm;origin=self.grid.beat_phase*self.rate + frame=round(origin+math.ceil((start-origin)/beat-1e-10)*beat) + self.pending=[(frame,1.),(frame+round(beat/2),.55),(frame+round(beat),.8)] + self.last=start;self.events.append({'kind':'native_alert_fill','frame':start,'scheduled_frame':frame,'alert':current, + 'competition_deferred':bool(competing),'priority':'warning over routine cues'}) + self.priority_active=bool(self.pending) + end=start+len(pcm);overlay=None;duck=None;remaining=[] for frame,gain in self.pending: left=max(start,frame);right=min(end,frame+len(self.grain)) if right>left: - if overlay is None:overlay=np.zeros_like(pcm) + if overlay is None:overlay=np.zeros_like(pcm);duck=np.zeros(len(pcm),np.float32) overlay[left-start:right-start]+=self.grain[left-frame:right-frame]*gain + duck[left-start:right-start]=np.maximum(duck[left-start:right-start],self.duck[left-frame:right-frame]*gain) if frame+len(self.grain)>end:remaining.append((frame,gain)) self.pending=remaining - # Never create an isolated notification sound in a musical rest. + # Do not fill an intentional musical rest with an isolated notification. if overlay is None or np.max(np.abs(pcm),initial=0)<.002:return pcm self.rendered_active=bool(np.any(overlay)) - return pcm+overlay + base=pcm*(1-(1-10**(-1.5/20))*duck[:,None]) + headroom=np.maximum(1-np.abs(base),0) + return base+np.clip(overlay,-headroom,headroom) def snapshot(self): return {'enabled':self.enabled,'rhythm_enabled':self.enabled and self.grid.usable, - 'rendered_active':self.rendered_active,'minimum_spacing_seconds':12, - 'maximum_per_minute':3,'source':'selfdriveState.alertStatus'} + 'rendered_active':self.rendered_active,'priority_active':self.priority_active, + 'minimum_spacing_seconds':12,'maximum_per_minute':3,'maximum_competition_deferral_seconds':.5, + 'peak_limit':.045,'duck_db':1.5,'fill':'beat, eighth, next beat','source':'selfdriveState.alertStatus'} diff --git a/roadscore/prototype/review_alert_accent.py b/roadscore/prototype/review_alert_accent.py new file mode 100644 index 0000000000..4eac5d9b9f --- /dev/null +++ b/roadscore/prototype/review_alert_accent.py @@ -0,0 +1,54 @@ +"""Offline warning A/B with recorded event timing and exactly the archived music.""" +import argparse,hashlib,json,time +from pathlib import Path +import numpy as np +import soundfile as sf +from alert_accent import AlertAccent +from signal_shaker import ShakerGrid +from engagement_presentation import EngagementPresentation,PresentationConfig + + +def sha(path):return hashlib.sha256(path.read_bytes()).hexdigest() + +def render(locked,out): + source=locked/'render/heard.wav';trace=locked/'render/trace.jsonl';timing=locked/'render/audio_blocks.jsonl';audit=locked/'source_alert_audit.private.json' + paths=(source,trace,timing,audit);hashes={str(p):sha(p) for p in paths} + rows=[json.loads(x) for x in trace.read_text().splitlines()];blocks=[json.loads(x) for x in timing.read_text().splitlines()] + events=[e for e in json.loads(audit.read_text())['intervals'] if e['eligible_status_and_size']] + if not events:raise ValueError('No recorded meaningful alert for comparison') + out.mkdir(parents=True,exist_ok=False);info=sf.info(source);rate=info.samplerate + accent=AlertAccent(ShakerGrid(128,0,0,0,False,'not yet observed'),rate,True) + delta_filter=EngagementPresentation(rate);config=PresentationConfig(enabled=True) + index=-1;observations=[];times=[] + with sf.SoundFile(source) as src,sf.SoundFile(out/'full_alert.wav','w',samplerate=rate,channels=2,subtype='FLOAT') as dst: + for i,block in enumerate(blocks): + start=round(block['audio_s']*rate);end=round(blocks[i+1]['audio_s']*rate) if i+1=0 else {};route_t=row.get('route_t',-1) + grid=row.get('signal_shaker',{}).get('grid') + if grid:accent.grid=ShakerGrid(**grid) + event=next((e for e in events if e['start_demo_s']<=route_t<=e['start_demo_s']+e['duration_observed_s']),None) + fresh=bool(block.get('engagement_fresh') and row and 0<=block['callback_wall']-row['command_wall']<=1.) + competing=bool(row.get('signal_shaker',{}).get('sequence_active') or row.get('core_apex',{}).get('rendered_active')) + before=time.perf_counter();wet=accent.process(pcm,start,str(event['key']) if event else None,event is not None,fresh,competing) + # Baseline already contains engagement processing. Filter only the new delta; + # this approximates placement before that stage without filtering music twice. + delta=delta_filter.process(wet-pcm,bool(block.get('engagement_active')),config) + y=pcm+delta;times.append(time.perf_counter()-before);dst.write(y) + observations.append(dict(audio_s=block['audio_s'],route_t=route_t,meaningful=event is not None,competing=competing,**accent.snapshot())) + first=next(x for x in observations if x['rendered_active']);start=max(0.,first['audio_s']-8);duration=min(24.,info.duration-start) + for name,path in [('A_baseline.wav',source),('B_alert.wav',out/'full_alert.wav')]: + with sf.SoundFile(path) as handle: + handle.seek(round(start*rate));wave=handle.read(round(duration*rate),dtype='float32',always_2d=True) + sf.write(out/name,wave,rate,subtype='FLOAT') + if hashes!={str(p):sha(p) for p in paths}:raise RuntimeError('Protected input changed') + report=dict(source_sha256=hashes,source_unchanged=True,excerpt_start_audio_s=start,duration_seconds=duration,events=accent.events, + approximation='New alert delta passed through reconstructed engagement filter; archived prior shaker/apex remains. Native priority ordering can differ. Same generated PCM, no normalization.', + runtime_cost='Mac offline only',median_ms=float(np.median(times)*1000),p95_ms=float(np.quantile(times,.95)*1000)) + (out/'report.json').write_text(json.dumps(report,indent=2));(out/'alert_states.json').write_text(json.dumps(observations)) + (out/'index.html').write_text('Warning fill A/B

Warning fill A/B

Same archived music and recorded warning. B adds one beat-locked percussion fill with a brief gentle duck. Existing archived cues remain; live priority will make them yield. This is an offline approximation, not a new hardware recording.

A · Baseline

B · Warning fill

') + return report + +if __name__=='__main__': + p=argparse.ArgumentParser();p.add_argument('locked',type=Path);p.add_argument('output',type=Path);a=p.parse_args();print(json.dumps(render(a.locked,a.output),indent=2)) diff --git a/roadscore/prototype/test_alert_accent.py b/roadscore/prototype/test_alert_accent.py index 028ebac0f7..0d149078cb 100644 --- a/roadscore/prototype/test_alert_accent.py +++ b/roadscore/prototype/test_alert_accent.py @@ -12,15 +12,18 @@ class AccentTests(unittest.TestCase): for i,key in enumerate(alerts): wet=accent.process(source,i*800,key,bool(key),fresh,competing) heard |= bool(np.any(wet!=source)) - self.assertLessEqual(float(np.max(np.abs(wet-source))),.016001) + self.assertLessEqual(float(np.max(np.abs(wet-source))),.061) return accent,heard def test_persistent_alert_once_and_grid_aligned(self): accent,heard=self.render(['takeover']*200) self.assertTrue(heard);self.assertEqual(len(accent.events),1) self.assertEqual(accent.events[0]['scheduled_frame'],5000) - def test_tiny_changes_and_competing_cues_are_omitted(self): + def test_tiny_changes_omitted_but_competing_cues_do_not_erase_warning(self): self.assertFalse(self.render(['a','b']*100)[1]) - self.assertFalse(self.render(['a']*100,competing=True)[1]) + accent,heard=self.render(['a']*100,competing=True) + self.assertTrue(heard);self.assertEqual(len(accent.events),1) + self.assertGreaterEqual(accent.events[0]['frame'],round(.7*8000)) + self.assertLess(accent.events[0]['scheduled_frame'],round(1.3*8000)) self.assertFalse(self.render(['a']*100,fresh=False)[1]) def test_density_limit(self): accent,_=self.render([str(i//30) for i in range(600)]) @@ -33,6 +36,17 @@ class AccentTests(unittest.TestCase): for i in range(3):a.process(x,i*800,'a',True,True) self.assertTrue(a.pending) self.assertIs(a.process(x,2400,'a',True,False),x);self.assertFalse(a.pending) + def test_headroom_and_no_notification_in_rest(self): + a=AlertAccent(GRID,rate=8000,enabled=True);x=np.full((800,2),.999,np.float32) + for i in range(20): + y=a.process(x,i*800,'warning',True,True,True) + self.assertLessEqual(np.max(np.abs(y)),1.) + b=AlertAccent(GRID,rate=8000,enabled=True);silence=np.zeros_like(x) + for i in range(20):self.assertIs(b.process(silence,i*800,'warning',True,True,True),silence) + def test_disabling_and_stale_input_remain_bypass(self): + a=AlertAccent(GRID,rate=8000,enabled=False);x=np.full((800,2),.1,np.float32) + self.assertIs(a.process(x,8000,'warning',True,True),x) + self.assertFalse(a.priority_active) def test_short_opening_assessed_without_assuming_confidence(self): # Real estimator must accept a 23s opening; silence must still veto cues. grid,_=assess_grid(np.zeros((23*8000,2),np.float32),8000,128)