Preserve archived replay origins, audio tails and capture boundaries

roadscore/prototype/capture_alignment.py

@@ -19,7 +19,7 @@ def audio_alignment(run, video_origin):
         if origin is None:
             raise ValueError('Stored audio has no DAC timeline')
         score = Path(summary['score']) / 'score.flac'
-        start = summary['source_frame_start'] / 48000 + video_origin - origin
+        start = summary['source_frame_start'] / summary.get('sample_rate',48000) + video_origin - origin
         kind = 'stored score DAC origin'
     if not score.is_file():
         raise FileNotFoundError(score)
@@ -29,11 +29,14 @@ def audio_alignment(run, video_origin):
 def video_frames(run, frames):
     """Trim preparation using the recorded host capture interval, never a manual offset."""
     fresh = Path(run) / 'host_audio_summary.json'
-    if not fresh.exists():
-        return list(range(len(frames)))
-    summary = json.loads(fresh.read_text())
-    start = summary['first_host_dac_wall']
-    end = start + summary['host_frames'] / 48000
+    if fresh.exists():
+        summary = json.loads(fresh.read_text())
+        start = summary['first_host_dac_wall']
+        end = start + summary['host_frames'] / 48000
+    else:
+        summary = json.loads((Path(run) / 'stored_summary.json').read_text())
+        start = summary['first_dac_wall']
+        end = start + (summary['source_frame_end'] - summary['source_frame_start']) / summary.get('sample_rate',48000)
     selected = [i for i, frame in enumerate(frames) if start <= frame['wall'] <= end]
     if len(selected) < 2:
         raise ValueError('Insufficient UI/audio timeline overlap')

roadscore/prototype/normal_onroad.py

@@ -15,7 +15,8 @@ R=Path(__file__).resolve().parents[1]
 native=Path('/TICI').exists()
 def interrupt(*_):raise KeyboardInterrupt
 signal.signal(signal.SIGTERM,interrupt)
-p=argparse.ArgumentParser();p.add_argument('--roadscore-seed',type=seed_argument,help='Reproduce an ACE session; normal launches choose a fresh seed');p.add_argument('--render-mode',choices=['current','gold-core'],default=None,help='ACE rendering mode');p.add_argument('--roadscore-presentation',choices=['conservative-v1','off','frozen']);p.add_argument('route',nargs='?');p.add_argument('--routeid');p.add_argument('--roadscore',action='store_true',required=True);p.add_argument('--replay',action='store_true',help='Play recorded final score without Chestnut');p.add_argument('--start',type=int,default=None);p.add_argument('--duration',type=float,default=float('inf'),help='Optional duration limit; normally replay to route EOF');p.add_argument('--audible',action='store_true',help='Compatibility flag; output is audible by default outside automated sessions');p.add_argument('--muted',action='store_true');p.add_argument('--no-overlay',action='store_true');p.add_argument('--capture-ui',action='store_true',help='Record the normal UI internally without speaker output');p.add_argument('--audio-device',default=None,help='Development host output device; default is the system output');p.add_argument('--transport-only',action='store_true');p.add_argument('--headless',action='store_true');p.add_argument('--runtime',type=Path,default=Path('/data/openpilot') if native else Path(os.environ.get('ROADSCORE_RUNTIME','/Users/dominickthompson/starpilot/.host_runtime/darwin/worktree')));p.add_argument('--bench',default=device_target());p.add_argument('--composer',choices=['sa3','ace'],default=choice(),help='ACE Prism is the event default; SA3 is an explicit fallback');p.add_argument('--profile',choices=['prism','aurora'],default='prism');a=p.parse_args()
+p=argparse.ArgumentParser();p.add_argument('--roadscore-seed',type=seed_argument,help='Reproduce an ACE session; normal launches choose a fresh seed');p.add_argument('--render-mode',choices=['current','gold-core'],default=None,help='ACE rendering mode');p.add_argument('--roadscore-presentation',choices=['conservative-v1','off','frozen']);p.add_argument('route',nargs='?');p.add_argument('--routeid');p.add_argument('--roadscore',action='store_true',required=True);p.add_argument('--replay',action='store_true',help='Play recorded final score without Chestnut');p.add_argument('--score-archive',type=Path,help='Imported local archive for --replay');p.add_argument('--start',type=int,default=None);p.add_argument('--duration',type=float,default=float('inf'),help='Optional duration limit; normally replay to route EOF');p.add_argument('--audible',action='store_true',help='Compatibility flag; output is audible by default outside automated sessions');p.add_argument('--muted',action='store_true');p.add_argument('--no-overlay',action='store_true');p.add_argument('--capture-ui',action='store_true',help='Record the normal UI internally without speaker output');p.add_argument('--audio-device',default=None,help='Development host output device; default is the system output');p.add_argument('--transport-only',action='store_true');p.add_argument('--headless',action='store_true');p.add_argument('--runtime',type=Path,default=Path('/data/openpilot') if native else Path(os.environ.get('ROADSCORE_RUNTIME','/Users/dominickthompson/starpilot/.host_runtime/darwin/worktree')));p.add_argument('--bench',default=device_target());p.add_argument('--composer',choices=['sa3','ace'],default=choice(),help='ACE Prism is the event default; SA3 is an explicit fallback');p.add_argument('--profile',choices=['prism','aurora'],default='prism');a=p.parse_args()
+explicit_start=a.start
 if a.render_mode=='gold-core' and a.composer!='ace':raise SystemExit('Gold core requires ACE')
 if a.roadscore_seed is not None and (a.replay or a.composer!='ace'):raise SystemExit('--roadscore-seed applies only to fresh ACE generation')
 session=None
@@ -71,14 +72,18 @@ env['PYTHONPATH']+=':'+('/data/roadscore-feasibility/venv/lib/python3.12/site-pa
 from route_library import local_source
 local=local_source(a.routeid)
 services='roadEncodeIdx,wideRoadEncodeIdx,driverEncodeIdx,modelV2,controlsState,onroadEvents,liveCalibration,radarState,deviceState,pandaStates,carParams,driverMonitoringState,carState,driverStateV2,roadCameraState,wideRoadCameraState,managerState,selfdriveState,longitudinalPlan,gpsLocationExternal,mapdOut,carOutput,carControl,liveParameters,starpilotCarState,starpilotPlan,starpilotRadarState,starpilotSelfdriveState,liveTracks,liveDelay,liveTorqueParameters,navInstruction,navRoute,livePose'
+score=None;stored_limit=None
+if a.score_archive and not a.replay:p.error('--score-archive requires --replay')
+if a.replay:
+ if not native and not local:raise SystemExit('Archived Mac replay requires a local route cache; no network fetch is started')
+ from score_archive import latest
+ from stored_replay_policy import replay_archive
+ try:score=a.score_archive.resolve() if a.score_archive else latest(a.routeid)
+ except FileNotFoundError:raise SystemExit('No stored RoadScore exists for this route. Generate a score first.')
+ a.start,stored_limit=replay_archive(score,a.routeid,explicit_start)
 args=[a.routeid,'--allow',services,'--start',str(a.start),'--no-loop','--headless','--cache','2']
 if local:args+=['--data_dir',str(local)]
 if native:args+=['--no-hw-decoder']
-score=None
-if a.replay:
- from score_archive import latest
- try:score=latest(a.routeid)
- except FileNotFoundError:raise SystemExit('No stored RoadScore exists for this route. Generate a score first.')
 display=None;children=[];named_children={};failure=None;logs=[];launch_started=time.monotonic()
 print(('Preparing stored score replay; no generation. ' if a.replay else ('Preparing ACE replay; first preparation may take 10–15 minutes. ' if a.composer=='ace' else 'Preparing RoadScore replay; cold preparation can take 2–3 minutes. '))+('Host speaker enabled.' if a.audible else 'Muted host capture.'),flush=True)
 def launch(cmd,name,**kw):
@@ -141,11 +146,14 @@ try:
    except OSError:pass
   try:
    native_state=json.loads(state_path.read_text())
-   if end_watch.observe(native_state,(out/'replay.log').read_text(),time.monotonic()):end_reason='native final segment exhausted';break
+   if end_watch.observe(native_state,(out/'replay.log').read_text(),time.monotonic()):
+    end_reason='native final segment exhausted'
+    if not a.replay:break
   except (FileNotFoundError,json.JSONDecodeError):pass
   if player.poll() is not None and time.monotonic()-started>2:
    if player.returncode:raise RuntimeError('Native replay failed; see replay.log')
-   break
+   if not a.replay:break
+  if a.replay and time.monotonic()-started>stored_limit:raise TimeoutError('Stored score did not finish within its recorded extent')
   if audio_host is not None and audio_host.poll() is not None:raise RuntimeError('Host PCM stream stopped; see host_audio.log')
   if not a.headless and ui.poll() is not None:raise RuntimeError('Existing normal UI exited; see ui.log')
   if time.monotonic()-started>a.duration+90:raise TimeoutError('No progressing replay messages')

roadscore/prototype/score_archive.py

@@ -5,9 +5,13 @@ from route_library import ROOT,identity
 
 def prefer_new_score(new,old):
  def coverage(m):return (bool(m.get('full_route')) and m.get('replay_start_seconds',0)==0,m.get('replay_start_seconds',0)==0,float(m.get('audio_seconds',0)))
+ if old.get('complete') is True and new.get('complete') is False:return False
  if coverage(new)[0] and coverage(old)[0]:return True
  return coverage(new)>=coverage(old)
 
+def recording_complete(blocks, launch, bridge):
+ return bool(blocks) and launch.get('end_reason')=='native final segment exhausted' and not bridge.get('failure')
+
 def archive(route,run,start=0):
  dongle,name=identity(route);parent=ROOT/'routes'/dongle/name/'roadscore';parent.mkdir(parents=True,exist_ok=True)
  dest=parent/run.name;dest.mkdir(exist_ok=False)
@@ -28,7 +32,10 @@ def archive(route,run,start=0):
  if (run/'runtime_manifest.json').exists():
   manifest=json.loads((run/'runtime_manifest.json').read_text());meta['source_identity']=manifest.get('source_identity',{'note':'See runtime manifest; source hash unavailable in this older run'});meta['backend']=manifest.get('backend',meta['backend']);meta['composer']=manifest.get('composer','sa3');meta['profile']=manifest.get('ace_initial_provenance',{}).get('prepared_profile')
  launch=json.loads((run/'launch.json').read_text()) if (run/'launch.json').exists() else {}
- meta['full_route']=launch.get('end_reason')=='native final segment exhausted' and start==0
+ bridge=json.loads((run/'bridge.json').read_text()) if (run/'bridge.json').exists() else {}
+ meta['complete']=recording_complete(blocks,launch,bridge)
+ meta['completion_reason']='native EOF with captured audio' if meta['complete'] else 'partial or unverified end of recording'
+ meta['full_route']=meta['complete'] and start==0
  meta['audio_seconds']=blocks[-1]['audio_s']+.1 if blocks else 0
  meta['output_underruns']=audio_stats.get('portaudio_flags',0)
  meta['capture_timing_clean']=not any(audio_stats.get(k,0) for k in ['portaudio_flags','starved_callbacks','late_frames'])

roadscore/prototype/stored_replay_policy.py

@@ -0,0 +1,25 @@
+"""Validate imported PCM extent without loading a composer or contacting a device."""
+import json
+import math
+from pathlib import Path
+
+
+def replay_archive(score, route, explicit_start=None):
+    import soundfile as sf
+    score = Path(score)
+    meta = json.loads((score / 'metadata.json').read_text())
+    if meta.get('route') != route:
+        raise ValueError('Stored score belongs to a different route')
+    if meta.get('first_model_ns') is None:
+        raise ValueError('Stored score lacks its measured route clock')
+    offset = meta.get('audio_file_start_relative_first_model')
+    if not isinstance(offset, (int, float)) or not math.isfinite(offset):
+        raise ValueError('Stored score lacks its measured audio clock')
+    info = sf.info(score / 'score.flac')
+    if info.frames <= 0:
+        raise ValueError('Stored score is empty')
+    start = meta.get('replay_start_seconds', 0) if explicit_start is None else explicit_start
+    if not isinstance(start, int) or start < 0:
+        raise ValueError('Stored replay start must be a nonnegative integer')
+    # Allow replay startup plus the entire captured tail, including positive DAC delay.
+    return start, info.duration + abs(offset) + max(0, meta.get('replay_start_seconds',0) - start) + 90

roadscore/prototype/stored_score.py

@@ -53,6 +53,6 @@ def main():
      tmp=a.out/'roadscore_status.tmp';tmp.write_text(json.dumps(state));tmp.replace(a.out/'roadscore_status.json');last_status=time.monotonic()
     if started and time.monotonic()-started>=a.duration:break
  finally:
-  (a.out/'stored_summary.json').write_text(json.dumps({'score':str(a.score),'first_dac_wall':first_dac_wall,'frames_presented':count,'portaudio_flags':flags,'muted':not a.audible,'first_model_ns':initial[0] if initial else None,'generation_invoked':False,'source_frame_start':first_frame,'source_frame_end':last_frame,'contiguous_samples_verified':first_frame is not None and digest.hexdigest()==hashlib.sha256(audio[first_frame:last_frame].tobytes()).hexdigest(),'max_clock_alignment_error_seconds':max_error,'synchronization':'original logMonoTime plus recorded host DAC/sample origin'},indent=2))
+  (a.out/'stored_summary.json').write_text(json.dumps({'score':str(a.score),'first_dac_wall':first_dac_wall,'sample_rate':rate,'frames_presented':count,'portaudio_flags':flags,'muted':not a.audible,'first_model_ns':initial[0] if initial else None,'generation_invoked':False,'source_frame_start':first_frame,'source_frame_end':last_frame,'contiguous_samples_verified':first_frame is not None and digest.hexdigest()==hashlib.sha256(audio[first_frame:last_frame].tobytes()).hexdigest(),'max_clock_alignment_error_seconds':max_error,'synchronization':'original logMonoTime plus recorded host DAC/sample origin'},indent=2))
  if max_error>.05:raise RuntimeError('Stored-score audio clock drift exceeded 50ms')
 if __name__=='__main__':main()

roadscore/prototype/test_stored_replay_policy.py

@@ -0,0 +1,40 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+from stored_replay_policy import replay_archive
+from score_archive import recording_complete, prefer_new_score
+from capture_alignment import video_frames
+
+class StoredReplayTests(unittest.TestCase):
+ def test_archive_offset_and_explicit_seek(self):
+  with tempfile.TemporaryDirectory() as d:
+   p=Path(d)
+   meta={'route':'route','first_model_ns':123,'audio_file_start_relative_first_model':.25,'replay_start_seconds':149}
+   (p/'metadata.json').write_text(json.dumps(meta))
+   fake=SimpleNamespace(info=lambda _:SimpleNamespace(frames=14400000,duration=300))
+   with patch.dict('sys.modules',soundfile=fake):
+    self.assertEqual(replay_archive(p,'route'),(149,390.25))
+    self.assertEqual(replay_archive(p,'route',160),(160,390.25))
+    self.assertEqual(replay_archive(p,'route',0),(0,539.25))
+    with self.assertRaisesRegex(ValueError,'different route'):replay_archive(p,'other')
+    meta['audio_file_start_relative_first_model']=None
+    (p/'metadata.json').write_text(json.dumps(meta))
+    with self.assertRaisesRegex(ValueError,'audio clock'):replay_archive(p,'route')
+ def test_partial_does_not_replace_complete_excerpt(self):
+  self.assertFalse(prefer_new_score({'complete':False,'audio_seconds':400},{'complete':True,'replay_start_seconds':149,'audio_seconds':300}))
+ def test_partial_audio_is_not_complete(self):
+  eof={'end_reason':'native final segment exhausted'}
+  self.assertTrue(recording_complete([{}],eof,{}))
+  self.assertFalse(recording_complete([],eof,{}))
+  self.assertFalse(recording_complete([{}],{'end_reason':'requested duration'},{}))
+  self.assertFalse(recording_complete([{}],eof,{'failure':'clock lost'}))
+ def test_stored_video_trims_startup_and_preserves_tail(self):
+  with tempfile.TemporaryDirectory() as d:
+   p=Path(d)
+   (p/'stored_summary.json').write_text(json.dumps({'first_dac_wall':200,'source_frame_start':44100,'source_frame_end':132300,'sample_rate':44100}))
+   self.assertEqual(video_frames(p,[{'wall':w} for w in [10,199,200,201,202,203]]),[2,3,4])
+   with self.assertRaises(ValueError):video_frames(p,[{'wall':10}])
+if __name__=='__main__':unittest.main()