Add interactive local Mac showcase from prepared ACE core

roadscore/MAC_SHOWCASE.md

@@ -0,0 +1,19 @@
+The prepared Mac showcase runs native onroad replay with a preserved ACE core recording and the current real-time presentation layer. It does not load ACE, contact Chestnut, or regenerate music.
+
+```sh
+./onroad --roadscore route1 --prepared-showcase
+```
+
+The launcher opens the normal UI and a local control page. Engage/disengage and turn-signal buttons use the same Galaxy command writer and presentation controls as the comma replay. The displayed vehicle state is explicitly a replay simulation. Ctrl+C stops the Mac session and its children.
+
+Local setup lives in ignored `roadscore/assets/prepared_showcase.json`:
+
+```json
+{"route":"DONGLE/ROUTE","archive":"/absolute/path/to/completed/core/archive","curve_plan":"/absolute/path/to/optional/showcase_curve_plan.json"}
+```
+
+The archive must contain `dry.wav`, `launch.json`, `audio_blocks.jsonl`, `replay_origin.json`, and `rhythm_timeline.json`. The route must already be cached locally. The dry source is used at unity gain; the recorded final mix is never processed a second time. Original model timestamps and measured DAC/sample offsets keep the music aligned to the replay. An optional staged curve plan must match the exact route and replay start.
+
+`--score-archive PATH` selects another compatible complete recording explicitly. `--check` validates the local prerequisites without starting playback. `--muted`, `--duration SECONDS`, and `--no-browser` support verification. Live generation and the existing stored final-score replay retain their separate launch modes.
+
+The Mac uses its selected system audio output. Bluetooth delay estimates from the comma do not transfer automatically to a different Mac output. Current controls do not synchronize a second device's playback.

roadscore/prototype/mac_showcase.html

@@ -0,0 +1,7 @@
+<!doctype html><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>RoadScore replay controls</title>
+<style>body{background:#101215;color:#eee;font:20px -apple-system,system-ui;margin:35px auto;max-width:760px;padding:15px}h1{font-size:34px;margin-bottom:8px}p{color:#aaa;line-height:1.5}section{margin-top:30px}button{font:inherit;background:#282d34;color:white;border:1px solid #454d59;border-radius:12px;padding:18px;margin:6px;cursor:pointer}button.on{background:#245e95;border-color:#78c7ff}button:disabled{opacity:.4;cursor:default}small{color:#aaa}#error{color:#ffba94}</style>
+<h1>RoadScore replay</h1><p>Prepared Prism music · live presentation controls<br>These controls simulate the replay display and music only.</p>
+<section><button data-field="mode" data-value="engaged">Simulate engage</button><button data-field="mode" data-value="disengaged">Simulate disengage</button><button data-field="mode" data-value="recorded">Recorded engagement</button></section>
+<section><button data-field="signal_mode" data-value="left">Left signal</button><button data-field="signal_mode" data-value="right">Right signal</button><button data-field="signal_mode" data-value="off">Signals off</button><button data-field="signal_mode" data-value="recorded">Recorded signals</button></section>
+<p id="state">Waiting for replay…</p><p id="error"></p><small>No model generation, vehicle control, or connection to the comma.</small>
+<script>let session;const buttons=[...document.querySelectorAll('button')];buttons.forEach(b=>b.onclick=async()=>{try{const r=await fetch('/control',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({session_id:session,[b.dataset.field]:b.dataset.value})});const j=await r.json();if(!r.ok)throw Error(j.error);document.querySelector('#error').textContent=''}catch(e){document.querySelector('#error').textContent=e.message}});async function refresh(){try{const r=await fetch('/status');const j=await r.json();session=j.demo.session_id;buttons.forEach(b=>{b.disabled=!j.demo.available;b.classList.toggle('on',j.demo[b.dataset.field]===b.dataset.value)});document.querySelector('#state').textContent=j.state+' · '+Math.max(0,j.audio_s||0).toFixed(1)+' s · '+(j.muted?'MUTED':'system audio output')}catch(e){buttons.forEach(b=>b.disabled=true);document.querySelector('#state').textContent='Replay stopped'}setTimeout(refresh,150)}refresh();</script>

roadscore/prototype/mac_showcase.py

@@ -0,0 +1,251 @@
+"""Local native replay + prepared ACE core + interactive presentation, without inference."""
+import argparse
+import importlib.util
+import hashlib
+import json
+import os
+from pathlib import Path
+import signal
+import subprocess
+import sys
+import threading
+import time
+import uuid
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+HERE = Path(__file__).resolve().parent
+
+
+def write_json(path, value):
+  temporary = Path(str(path)+'.tmp')
+  temporary.write_text(json.dumps(value))
+  temporary.replace(path)
+
+
+def control_server(project, out, shared, port):
+  spec = importlib.util.spec_from_file_location('showcase_galaxy', project/'starpilot/system/the_galaxy/roadscore.py')
+  module = importlib.util.module_from_spec(spec)
+  spec.loader.exec_module(module)
+  operator_root = out/'operator'
+  (operator_root/'results').mkdir(parents=True)
+  (operator_root/'results/current').symlink_to(out)
+  operator = module.Operator(operator_root, device=False, offroad=lambda: True)
+  class Handler(BaseHTTPRequestHandler):
+    def log_message(self, *args):pass
+    def send_json(self, value, status=200):
+      data = json.dumps(value).encode()
+      self.send_response(status)
+      self.send_header('Content-Type', 'application/json')
+      self.send_header('Content-Length', str(len(data)))
+      self.send_header('Cache-Control', 'no-store')
+      self.end_headers();self.wfile.write(data)
+    def do_GET(self):
+      if self.path == '/status':
+        self.send_json({'demo': operator.demo_status(), **shared})
+      elif self.path == '/':
+        data = (HERE/'mac_showcase.html').read_bytes()
+        self.send_response(200);self.send_header('Content-Type', 'text/html; charset=utf-8');self.send_header('Content-Length', str(len(data)));self.end_headers();self.wfile.write(data)
+      else:self.send_error(404)
+    def do_POST(self):
+      if self.path != '/control':self.send_error(404);return
+      if not module.control_origin_allowed(self.headers.get('Origin'), self.headers.get('Host'), 'http', self.headers.get('Sec-Fetch-Site')):
+        self.send_json({'error':'Cross-origin controls are not allowed'},403);return
+      try:
+        length = int(self.headers.get('Content-Length', '0'))
+        if not 0 < length <= 1024:raise ValueError('Invalid control size')
+        data = json.loads(self.rfile.read(length))
+        field = 'signal_mode' if 'signal_mode' in data else 'mode'
+        self.send_json(operator.demo_engagement(data, field))
+      except (ValueError, TypeError) as error:self.send_json({'error':str(error)},400)
+  server = ThreadingHTTPServer(('127.0.0.1',port),Handler)
+  threading.Thread(target=server.serve_forever,daemon=True).start()
+  return server
+
+
+def audio_worker(a):
+  import numpy as np
+  import sounddevice as sd
+  from cereal import messaging
+  from core import Conductor
+  from demo_engagement import DemoEngagement
+  from operator_output import PresentationDelay
+  from prepared_core import load_archive, initial_frame, PreparedPresentation
+  audio, rate, meta = load_archive(a.score_archive, a.route)
+  processor = PreparedPresentation(a.score_archive, rate)
+  session = os.environ['ROADSCORE_SHOWCASE_SESSION']
+  controls = DemoEngagement(a.out, session, 'replay')
+  delay = PresentationDelay(a.out,output_provider=lambda: None)
+  sm = messaging.SubMaster(['modelV2','carState','selfdriveState'],poll='modelV2')
+  conductor = Conductor(handoff=True)
+  staged = None
+  plan_path = a.curve_plan
+  if plan_path is not None and plan_path.exists():
+    from demo_curve_plan import ReplayCurvePlan
+    value = json.loads(plan_path.read_text())
+    if value['route'] != a.route or value['replay_start'] != int(meta['native_replay_args'][meta['native_replay_args'].index('--start')+1]):
+      raise ValueError('Staged curve plan does not match prepared route/start')
+    staged = ReplayCurvePlan(value)
+    write_json(a.out/'demo_curve_plan.json',value)
+  state = dict(active=False,signal_on=False,fresh=False,car_fresh=False,model_fresh=False,speed=0.,alert_key='',alert_meaningful=False,curve={'kind':'curve','phase':'neutral','amount':0.,'activation':None})
+  received = {name:float('-inf') for name in sm.services}
+  anchor = None
+  position = None
+  rendered = None
+  done = False
+  errors = []
+  flags = 0
+  max_drift = 0.
+  first_frame = None
+  shared = dict(state='PREPARING',audio_s=0.,muted=a.muted)
+  server = control_server(a.project_root,a.out,shared,a.port)
+  write_json(a.out/'controls.json',{'url':f'http://127.0.0.1:{server.server_port}/'})
+  def callback(out, n, ti, status):
+    nonlocal position,rendered,done,flags,max_drift,first_frame
+    out.fill(0)
+    if anchor is None or done:return
+    now=time.monotonic();dac=now+float(ti.outputBufferDacTime-ti.currentTime)
+    expected=initial_frame(meta,anchor[0],dac-anchor[1],rate)
+    if position is None:position=expected;first_frame=position
+    start=position;position+=n
+    max_drift=max(max_drift,abs(start-expected)/rate)
+    if status:flags+=1
+    chunk=np.zeros((n,2),np.float32)
+    lo=max(0,-start);hi=min(n,len(audio)-start)
+    if hi>lo:chunk[lo:hi]=audio[start+lo:start+hi]
+    try:
+      wet,cues=processor.process(chunk,start,state,controls.selection)
+      rendered=dict(sequence=start,callback_wall=now,dac_wall=dac,cues=cues)
+      if not a.muted:out[:]=wet
+      if start+n>=len(audio):done=True
+    except Exception as error:
+      errors.append(repr(error));done=True
+  started=None;last_status=0.
+  try:
+    with sd.OutputStream(device=a.audio_device,samplerate=rate,channels=2,blocksize=960,dtype='float32',callback=callback):
+      (a.out/'prepared_ready').write_text('ready')
+      while not done:
+        sm.update(50);now=time.monotonic();controls.poll()
+        for name in received:
+          if sm.updated[name]:received[name]=now
+        latest=max(sm.logMonoTime.values())
+        def fresh(name):return bool(sm.valid[name] and 0<=now-received[name]<=.6 and 0<=(latest-sm.logMonoTime[name])/1e9<=1.)
+        c=sm['carState'];s=sm['selfdriveState']
+        update=dict(active=bool(s.active) and fresh('selfdriveState'),signal_on=bool(c.leftBlinker or c.rightBlinker),fresh=fresh('selfdriveState'),car_fresh=fresh('carState'),model_fresh=fresh('modelV2'),speed=float(c.vEgo),alert_key=str(s.alertType),alert_meaningful=int(s.alertStatus.raw)>0 and int(s.alertSize.raw)>0,curve=state['curve'])
+        if sm.updated['modelV2']:
+          mono=sm.logMonoTime['modelV2']
+          if anchor is None:anchor=(mono,now);started=now
+          if abs((mono-anchor[0])/1e9-(now-anchor[1]))>.75:raise RuntimeError('Prepared showcase left its 1x replay clock')
+          route_t=(mono-meta['first_model_ns'])/1e9
+          m=sm['modelV2']
+          curve=conductor.update(route_t,{'mono':mono,'eof':m.timestampEof,'t':list(m.orientationRate.t),'yaw':list(m.orientationRate.z),'v':list(m.velocity.x)},float(c.vEgo)) if fresh('modelV2') and len(m.position.t)==33 else conductor.state(route_t)
+          update['curve']=staged.state(route_t,a.route,curve) if staged else curve
+        state=update
+        if now-last_status>=.08:
+          shared.update(state='READY' if anchor else 'PREPARING',audio_s=max(0,(position or 0)/rate))
+          snapshot=dict(command_wall=now,input_mode='replay',compute='prepared-core',composer='ace',profile='prism',style='Prism',readiness=shared['state'],section='PREPARED PRISM',route=a.route,buffered=max(0,(len(audio)-(position or 0))/rate),presentation_session_id=session,demo_engagement_mode=controls.mode,demo_signal_mode=controls.signal_mode,engagement_presentation={'enabled':True},generation_invoked=False)
+          snapshot=delay.apply(snapshot,rendered)
+          # The command API must remain available during initial DAC lead-in.
+          snapshot.setdefault('engagement_presentation',{'enabled':True})
+          write_json(a.out/'status.json',snapshot);last_status=now
+        if started and now-started>=a.duration:break
+        if started and now-received['modelV2']>2 and (position or 0)/rate < len(audio)/rate-2:raise RuntimeError('Replay model stream stopped before prepared audio ended')
+  finally:
+    server.shutdown()
+    write_json(a.out/'prepared_summary.json',dict(generation_invoked=False,source=str(a.score_archive),first_source_frame=first_frame,last_source_frame=position,sample_rate=rate,portaudio_flags=flags,max_clock_error_seconds=max_drift,callback_errors=errors,muted=a.muted,session_id=session,manual_scope='isolated replay display and presentation only'))
+  if errors:raise RuntimeError(errors[0])
+  if max_drift>.05:raise RuntimeError('Prepared audio clock drift exceeded 50 ms')
+
+
+def parser():
+  p=argparse.ArgumentParser()
+  p.add_argument('--roadscore',action='store_true');p.add_argument('--prepared-showcase',action='store_true')
+  p.add_argument('route',nargs='?',default='route1')
+  p.add_argument('--score-archive',type=Path)
+  p.add_argument('--curve-plan',type=Path)
+  p.add_argument('--project-root',type=Path,default=HERE.parents[1])
+  p.add_argument('--runtime',type=Path)
+  p.add_argument('--out',type=Path)
+  p.add_argument('--duration',type=float,default=float('inf'))
+  p.add_argument('--port',type=int,default=0)
+  p.add_argument('--muted',action='store_true');p.add_argument('--headless',action='store_true')
+  p.add_argument('--no-browser',action='store_true');p.add_argument('--audio-device');p.add_argument('--check',action='store_true');p.add_argument('--audio-worker',action='store_true',help=argparse.SUPPRESS)
+  return p
+
+
+def main():
+  a=parser().parse_args()
+  if a.audio_worker:return audio_worker(a)
+  if sys.platform!='darwin' or Path('/TICI').exists():raise SystemExit('Prepared Mac showcase runs only on the Mac')
+  project=a.project_root.resolve();rt=a.runtime or project/'.host_runtime/darwin/worktree'
+  py=rt.parent/'venv/bin/python'
+  from route_favorites import resolve_favorite
+  a.route,_=resolve_favorite(a.route,project/'roadscore/routes/favorites.json')
+  if a.score_archive is None:
+    config=json.loads((project/'roadscore/assets/prepared_showcase.json').read_text())
+    if config['route']!=a.route:raise SystemExit('No prepared showcase configured for this route')
+    a.score_archive=Path(config['archive'])
+    if a.curve_plan is None and config.get('curve_plan'):a.curve_plan=Path(config['curve_plan'])
+  launch=json.loads((a.score_archive/'launch.json').read_text())
+  if launch['route']!=a.route:raise SystemExit('Prepared score belongs to a different route')
+  from route_library import local_source
+  local=local_source(a.route,project/'roadscore')
+  if local is None:raise SystemExit('Local route missing; no network fetch will be started')
+  # This Mac binary predates native H.264 cache support. Prefer the complete
+  # local originals, which VideoToolbox decodes, instead of its qcamera fallback.
+  if local.name=='playback':
+    original=local.parent
+    segments=list(original.glob(a.route.split('/')[1]+'--*'))
+    if segments and all((segment/'fcamera.hevc').is_file() for segment in segments):local=original
+  for path in (py,rt/'tools/replay/replay',rt/'selfdrive/ui/ui.py'):
+    if not path.exists():raise SystemExit('Missing existing host runtime: '+str(path))
+  session=uuid.uuid4().hex
+  out=a.out or project/'roadscore/results'/('mac_showcase_'+str(int(time.time())))
+  out.mkdir(parents=True,exist_ok=False)
+  env=os.environ.copy()
+  env.update(PYTHONDONTWRITEBYTECODE='1',ZMQ='1',OPENPILOT_ZMQ_NAMESPACE='roadscore-showcase-'+session,ROADSCORE_SHOWCASE_SESSION=session,ROADSCORE_PREPARED_SHOWCASE='1',ROADSCORE_REPLAY_UI_CONTROLS='1',PARAMS_ROOT=str(out/'params'),BASEDIR=str(rt),NOBOARD='1',SIMULATION='1',SKIP_FW_QUERY='1',BIG='0',SP_ALLOW_DESKTOP_FAKE_WIFI='0',SP_ALLOW_DESKTOP_FAKE_BLUETOOTH='0',SP_ONROAD_NAV_DEMO='0',SP_ONROAD_CEM_DEMO='0',ROADSCORE_CLEAN_DEMO_UI='1',ROADSCORE_OVERLAY='1',ROADSCORE_STATUS_FILE=str(out/'status.json'),ROADSCORE_UI_AUDIT=str(out/'ui_audit.jsonl'),ROADSCORE_OVERLAY_CAPTURE=str(out/'overlay.png'),ROADSCORE_PRESENTATION_POLICY='conservative-v4')
+  env.pop('OPENPILOT_PREFIX',None)
+  env['PWD']=str(rt)
+  env['PYTHONPATH']=':'.join(map(str,[HERE,rt,rt/'starpilot/third_party',*rt.glob('*_repo'),project/'roadscore/.analysis-venv/lib/python3.12/site-packages']))
+  args=launch['native_replay_args'][:]
+  args[args.index('--data_dir')+1]=str(local)
+  if '--no-hw-decoder' in args:args.remove('--no-hw-decoder')
+  write_json(out/'status.json',dict(readiness='PREPARING',style='Prism',compute='prepared-core'))
+  write_json(out/'launch.json',dict(core_sha256=hashlib.sha256((a.score_archive/'dry.wav').read_bytes()).hexdigest(),curve_plan_sha256=hashlib.sha256(a.curve_plan.read_bytes()).hexdigest() if a.curve_plan else None,mode='prepared-interactive-showcase',route=a.route,source=str(a.score_archive),runtime=str(rt),native_replay_args=args,generation_invoked=False,network_required=False,session_id=session,muted=a.muted or a.headless))
+  check=subprocess.run([str(py),'-c','from cereal import messaging; import sounddevice,soundfile; from prepared_core import load_archive; import sys; a,r,m=load_archive(sys.argv[1],sys.argv[2]); print("Prepared core:",len(a)/r,"seconds; local replay ready")',str(a.score_archive),a.route],cwd=rt,env=env)
+  if check.returncode:raise SystemExit(check.returncode)
+  if a.check:return
+  children=[];logs=[]
+  def start(cmd,name):
+    log=(out/(name+'.log')).open('w');logs.append(log)
+    proc=subprocess.Popen(cmd,env=env,cwd=rt,stdout=log,stderr=subprocess.STDOUT,start_new_session=True)
+    children.append(proc);return proc
+  def stop(*_):raise KeyboardInterrupt
+  signal.signal(signal.SIGTERM,stop)
+  try:
+    with (out/'seed.log').open('w') as log:subprocess.run([str(py),str(rt/'tools/replay/onroad_config.py'),'seed',*args],env=env,cwd=rt,stdout=log,stderr=subprocess.STDOUT,check=True)
+    if not a.headless:ui=start([str(py),str(HERE/'normal_ui_audit.py')],'ui')
+    audio=start([str(py),str(__file__),'--audio-worker',a.route,'--score-archive',str(a.score_archive),'--project-root',str(project),'--out',str(out),'--duration',str(a.duration),'--port',str(a.port)]+(['--curve-plan',str(a.curve_plan)] if a.curve_plan else [])+(['--muted'] if a.muted or a.headless else [])+(['--audio-device',a.audio_device] if a.audio_device else []),'audio')
+    deadline=time.monotonic()+30
+    while not (out/'prepared_ready').exists():
+      if audio.poll() is not None or time.monotonic()>deadline:raise RuntimeError('Prepared audio did not become ready; see '+str(out/'audio.log'))
+      time.sleep(.1)
+    controls_url=json.loads((out/'controls.json').read_text())['url']
+    print('Replay controls:',controls_url,flush=True)
+    if not a.no_browser and not a.headless:subprocess.Popen(['/usr/bin/open',controls_url],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
+    replay=start([str(rt/'tools/replay/replay'),*args],'replay')
+    print('Prepared RoadScore showcase running locally. Ctrl+C stops it. Output:',out,flush=True)
+    while audio.poll() is None:
+      if replay.poll() not in (None,0):raise RuntimeError('Local replay failed; see '+str(out/'replay.log'))
+      if not a.headless and ui.poll() is not None:raise RuntimeError('Native UI exited; see '+str(out/'ui.log'))
+      time.sleep(.2)
+    if audio.returncode:raise RuntimeError('Prepared audio failed; see '+str(out/'audio.log'))
+  finally:
+    for child in reversed(children):
+      if child.poll() is None:
+        os.killpg(child.pid,signal.SIGTERM)
+        try:child.wait(timeout=5)
+        except subprocess.TimeoutExpired:os.killpg(child.pid,signal.SIGKILL);child.wait()
+    for log in logs:log.close()
+    print('Local prepared showcase stopped.',flush=True)
+
+if __name__=='__main__':main()

roadscore/prototype/prepared_core.py

@@ -0,0 +1,87 @@
+"""Prepared ACE PCM with the normal real-time presentation layers; no model imports."""
+import json
+import math
+from pathlib import Path
+
+import numpy as np
+import soundfile as sf
+
+from alert_accent import AlertAccent
+from curve_reaction import CurveReaction
+from demo_engagement import presentation_active, presentation_signal, controls_snapshot, annotate
+from engagement_presentation import EngagementPresentation, PresentationConfig
+from motion_presentation import MotionPresentation
+from presentation_policy import effective_config
+from rhythm_timeline import RhythmTimeline
+from signal_shaker import SignalShaker, ShakerGrid
+
+
+def load_archive(path, route):
+  path = Path(path)
+  launch = json.loads((path/'launch.json').read_text())
+  origin = json.loads((path/'replay_origin.json').read_text())
+  if launch.get('route') != route or launch.get('render_mode') != 'gold-core':
+    raise ValueError('Prepared showcase requires matching route and original ACE core render')
+  if launch.get('end_reason') != 'native final segment exhausted':
+    raise ValueError('Prepared showcase requires a complete route capture')
+  with (path/'audio_blocks.jsonl').open() as timing:
+    first = json.loads(next(timing))
+  values = [first.get('audio_s'), first.get('callback_wall'), first.get('dac_delay'), origin.get('host_received_wall')]
+  if not all(type(x) in (int, float) and math.isfinite(x) for x in values) or first['audio_s'] != 0:
+    raise ValueError('Prepared core lacks its original sample/DAC clock')
+  if type(origin.get('first_model_ns')) is not int or origin['first_model_ns'] <= 0:
+    raise ValueError('Prepared core lacks its original model clock')
+  audio, rate = sf.read(path/'dry.wav', dtype='float32', always_2d=True)
+  if rate != 48000 or audio.shape[1] != 2 or not len(audio) or not np.isfinite(audio).all():
+    raise ValueError('Prepared core must be finite stereo 48 kHz PCM')
+  offset = first['callback_wall'] + first['dac_delay'] - origin['host_received_wall']
+  if abs(offset) > 2:
+    raise ValueError('Unexpected original audio clock offset')
+  return audio, rate, {**launch, **origin, 'audio_offset': offset, 'duration': len(audio)/rate}
+
+
+def initial_frame(meta, mono_ns, dac_since_receipt, rate):
+  return round(((mono_ns-meta['first_model_ns'])/1e9+dac_since_receipt-meta['audio_offset'])*rate)
+
+
+class PreparedPresentation:
+  """Identical presentation primitives/order to ACE gold-core playback."""
+  def __init__(self, archive, rate=48000, policy='conservative-v4'):
+    config = effective_config({}, {'ROADSCORE_PRESENTATION_POLICY': policy})
+    self.config = PresentationConfig.read(config['engagement_presentation'])
+    self.engagement = EngagementPresentation(rate, cutoff_hz=self.config.cutoff_hz, width=self.config.width, gain=self.config.gain)
+    self.motion = MotionPresentation(rate, enabled=config.get('stopped_motion', {}).get('enabled', False))
+    self.rhythm = RhythmTimeline(rate, None)
+    rows = json.loads((Path(archive)/'rhythm_timeline.json').read_text())
+    self.rhythm.entries = tuple((int(row['start_frame']), int(row['end_frame']), ShakerGrid(**row['grid'])) for row in rows)
+    grid = self.rhythm.at(0)
+    self.shaker = SignalShaker(grid, rate, enabled=True, peak=config['signal_shaker'].get('peak', .018))
+    self.curve = CurveReaction(grid, rate, enabled=True, bass_build=config.get('curve_reaction', {}).get('bass_build', False))
+    self.alert = AlertAccent(grid, rate, enabled=True)
+    self.contained_gain = config['signal_shaker'].get('contained_gain', 1.)
+
+  def process(self, pcm, frame, state, selection):
+    mode, signal_mode = selection
+    active = presentation_active(mode, state['active'])
+    signal_on = presentation_signal(signal_mode, state['signal_on'])
+    grid = self.rhythm.at(frame)
+    self.shaker.set_grid(grid, frame)
+    self.curve.grid = self.alert.grid = grid
+    competing = self.shaker.active or abs(self.engagement.mix-float(active)) > .01
+    result = self.alert.process(pcm, frame, state['alert_key'], state['alert_meaningful'], state['fresh'], competing)
+    result = self.curve.process(result, frame, state['curve'], source_fresh=state['model_fresh'], blocked=self.alert.priority_active or (self.motion.stopped and self.motion.fresh))
+    result = self.motion.process(result, speed=state['speed'], source_fresh=state['car_fresh'])
+    result = self.engagement.process(result, active, self.config)
+    gain = self.contained_gain+(1-self.contained_gain)*min(self.engagement.mix, self.motion.dsp.mix)
+    shaken = self.shaker.process(result, frame, signal_on, state['car_fresh'], sequence_key=signal_mode, presentation_gain=gain)
+    if not self.alert.priority_active:
+      result = shaken
+    else:
+      self.shaker.rendered_active = False
+      self.shaker.rendered_peak = 0.
+      self.shaker.suppression_reason = 'meaningful alert priority'
+    cues = {**state['curve'], 'section': 'PREPARED PRISM', 'replay_demo': controls_snapshot(mode, signal_mode),
+            'signal_shaker': self.shaker.snapshot(), 'alert_accent': self.alert.snapshot(),
+            **self.motion.snapshot(), **self.curve.snapshot(),
+            **annotate(self.engagement.snapshot(self.config, active, state['fresh']), mode, state['active'])}
+    return result, cues

roadscore/prototype/replay_ui_controls.py

@@ -1,6 +1,7 @@
 """Native replay UI view of acknowledged demo controls; never publishes messages."""
 import json
 import math
+import re
 import time
 from pathlib import Path
 
@@ -9,10 +10,15 @@ from demo_engagement import MODES, SIGNAL_MODES
 
 
 def isolated_replay(environ):
-  return (environ.get('ROADSCORE_REPLAY_UI_CONTROLS') == '1'
-          and environ.get('SIMULATION') == '1'
-          and not environ.get('ZMQ')
-          and environ.get('OPENPILOT_PREFIX') == 'roadscore_replay')
+  if environ.get('ROADSCORE_REPLAY_UI_CONTROLS') != '1' or environ.get('SIMULATION') != '1':
+    return False
+  if not environ.get('ZMQ'):
+    return environ.get('OPENPILOT_PREFIX') == 'roadscore_replay'
+  session = environ.get('ROADSCORE_SHOWCASE_SESSION', '')
+  return (environ.get('ROADSCORE_PREPARED_SHOWCASE') == '1'
+          and environ.get('ZMQ') == '1'
+          and bool(re.fullmatch(r'[a-zA-Z0-9-]{8,80}', session))
+          and environ.get('OPENPILOT_ZMQ_NAMESPACE') == 'roadscore-showcase-' + session)
 
 
 def apply_turn_intent(widget, signal_mode):

roadscore/prototype/test_prepared_core.py

@@ -0,0 +1,50 @@
+import json
+from pathlib import Path
+import tempfile
+import unittest
+import numpy as np
+import soundfile as sf
+from prepared_core import load_archive, initial_frame, PreparedPresentation
+from replay_ui_controls import isolated_replay
+
+
+class PreparedTests(unittest.TestCase):
+  def setUp(self):
+    self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
+    self.path=Path(self.tmp.name)
+    (self.path/'launch.json').write_text(json.dumps(dict(route='fixture',render_mode='gold-core',end_reason='native final segment exhausted')))
+    (self.path/'replay_origin.json').write_text(json.dumps(dict(first_model_ns=1000000000,host_received_wall=10.)))
+    (self.path/'audio_blocks.jsonl').write_text(json.dumps(dict(audio_s=0.,callback_wall=10.05,dac_delay=.1))+'\n')
+    self.pcm=np.ones((48000,2),np.float32)*.1
+    sf.write(self.path/'dry.wav',self.pcm,48000,subtype='FLOAT')
+    (self.path/'rhythm_timeline.json').write_text(json.dumps([dict(start_frame=0,end_frame=480000,grid=dict(bpm=120.,beat_phase=0.,tempo_confidence=1.,phase_confidence=1.,coherent=True,reason='fixture'))]))
+  def test_original_clock_and_pcm_preserved(self):
+    pcm,rate,meta=load_archive(self.path,'fixture')
+    np.testing.assert_array_equal(pcm,self.pcm)
+    self.assertEqual(initial_frame(meta,1000000000,.15,rate),0)
+    self.assertEqual(initial_frame(meta,1500000000,.15,rate),24000)
+  def test_wrong_route_and_final_pcm_rejected(self):
+    with self.assertRaises(ValueError):load_archive(self.path,'other')
+    (self.path/'launch.json').write_text(json.dumps(dict(route='fixture',render_mode='current',end_reason='native final segment exhausted')))
+    with self.assertRaises(ValueError):load_archive(self.path,'fixture')
+  def test_reaction_changes_without_changing_source(self):
+    engine=PreparedPresentation(self.path)
+    state=dict(active=True,signal_on=False,fresh=True,car_fresh=True,model_fresh=True,speed=10.,alert_key='',alert_meaningful=False,curve=dict(kind='curve',phase='neutral',amount=0.,activation=None))
+    before=self.pcm.copy()
+    chunks=[];cues={}
+    for frame in range(0,48000,960):
+      wet,cues=engine.process(self.pcm[frame:frame+960],frame,state,('disengaged','left'));chunks.append(wet)
+    result=np.concatenate(chunks)
+    self.assertLess(engine.engagement.mix,.01)
+    self.assertGreater(len(engine.shaker.pulse_frames),0)
+    self.assertTrue(cues['replay_demo']['simulated'])
+    np.testing.assert_array_equal(self.pcm,before)
+    self.assertEqual(result.shape,before.shape)
+  def test_mac_isolation_requires_explicit_matching_namespace(self):
+    env=dict(ROADSCORE_REPLAY_UI_CONTROLS='1',SIMULATION='1',ZMQ='1',ROADSCORE_PREPARED_SHOWCASE='1',ROADSCORE_SHOWCASE_SESSION='abcdefgh',OPENPILOT_ZMQ_NAMESPACE='roadscore-showcase-abcdefgh')
+    self.assertTrue(isolated_replay(env))
+    for key in ('ROADSCORE_PREPARED_SHOWCASE','ROADSCORE_SHOWCASE_SESSION','SIMULATION'):
+      self.assertFalse(isolated_replay({**env,key:''}))
+    self.assertFalse(isolated_replay({**env,'OPENPILOT_ZMQ_NAMESPACE':'roadscore-native-abc'}))
+
+if __name__=='__main__':unittest.main()