Select complete validated derived playback caches automatically

roadscore/prototype/playback_cache.py

@@ -0,0 +1,52 @@
+"""Select a complete derived video cache without changing original route assets."""
+import json
+from pathlib import Path, PurePosixPath
+
+SCHEMA='roadscore-playback-cache-v1'
+
+
+def relative(value):
+  if not isinstance(value,str) or '\\' in value:
+    raise ValueError('Invalid cache path')
+  path=PurePosixPath(value)
+  if path.is_absolute() or '..' in path.parts or len(path.parts)!=2:
+    raise ValueError('Invalid cache path')
+  return path
+
+
+def validated_playback(parent,route):
+  parent=Path(parent);derived=parent/'playback'
+  try:
+    if (parent/'.acquiring').exists() or (derived/'.acquiring').exists():return None
+    original=json.loads((parent/'cache_manifest.json').read_text())
+    manifest=json.loads((derived/'playback_manifest.json').read_text())
+    canonical=route.replace('|','/')
+    if (manifest.get('schema')!=SCHEMA or manifest.get('route')!=canonical
+        or original.get('route','').replace('|','/')!=canonical or manifest.get('complete') is not True):return None
+    originals={entry['path']:entry for entry in original['files']}
+    files=manifest['files']
+    if not originals or len(files)!=len(originals) or len(originals)!=len(original['files']):return None
+    if {entry['source'] for entry in files}!=set(originals) or len({entry['path'] for entry in files})!=len(files):return None
+    name=canonical.split('/')[1]
+    for entry in files:
+      source=relative(entry['source']);target=relative(entry['path'])
+      if source.parent!=target.parent or not source.parts[0].startswith(name+'--'):return None
+      if not source.parts[0][len(name)+2:].isdigit():return None
+      before=(parent/source).stat();after=(derived/target).stat()
+      if not (parent/source).is_file() or not (derived/target).is_file():return None
+      if before.st_size!=originals[str(source)]['bytes'] or before.st_size!=entry['source_bytes'] or before.st_mtime_ns!=entry['source_mtime_ns']:return None
+      if after.st_size!=entry['bytes'] or after.st_size<=0:return None
+      if source==target:
+        if before.st_size!=after.st_size:return None
+      else:
+        if source.suffix!='.hevc' or target.suffix!='.h264' or source.stem!=target.stem:return None
+        if (derived/source).exists():return None
+        video=entry['video_validation']
+        if (type(video['source_frames']) is not int or video['source_frames']<=0
+            or video['source_frames']!=video['output_frames']
+            or type(video['width']) is not int or video['width']<=0
+            or type(video['height']) is not int or video['height']<=0
+            or video['frame_order_verified'] is not True or video['timing_verified'] is not True):return None
+    return derived
+  except (OSError,ValueError,KeyError,TypeError,IndexError):
+    return None

roadscore/prototype/route_library.py

@@ -1,6 +1,7 @@
 """Private native-layout route cache. Does not feed pre-analysis to RoadScore."""
 import argparse,json,os,re,shutil,hashlib
 from pathlib import Path
+from playback_cache import validated_playback
 ROOT=Path(__file__).resolve().parents[1]
 def identity(route):
  parts=route.replace('|','/').split('/')
@@ -21,9 +22,10 @@ def local_source(route,root=ROOT):
  candidates += [Path(x) for x in os.environ.get('ROADSCORE_ROUTE_PATHS','/data/media/0/realdata').split(':') if x]
  for parent in candidates:
   if (parent/'.acquiring').exists() or cache_complete(parent) is False:continue
-  if any(parent.glob(name+'--*/rlog*')) or any(parent.glob(name+'--*/qlog*')):return parent
+  if any(parent.glob(name+'--*/rlog*')) or any(parent.glob(name+'--*/qlog*')):return validated_playback(parent,route) or parent
  from cache_discovery import prepared_source
- return prepared_source(route,root)
+ parent=prepared_source(route,root)
+ return (validated_playback(parent,route) or parent) if parent is not None else None
 
 def fetch(route):
  # Same authenticated endpoints and host fallback as native replay. Never prints signed URLs.
@@ -73,7 +75,7 @@ def inventory():
    if source is None:continue
    segments=sorted(source.glob(name+'--*'),key=lambda p:int(p.name.rsplit('--',1)[1]))
    logs=all(any(x.glob('rlog*')) or any(x.glob('qlog*')) for x in segments)
-   video=all((x/'fcamera.hevc').exists() or (x/'qcamera.ts').exists() for x in segments)
+   video=all((x/'fcamera.h264').exists() or (x/'fcamera.hevc').exists() or (x/'qcamera.ts').exists() for x in segments)
    evidence=ROOT/'routes'/dongle.name/name/'acceptance.json'
    tested=json.loads(evidence.read_text()) if evidence.exists() else None
    audit_path=ROOT/'routes'/dongle.name/name/'inventory.json';audit=json.loads(audit_path.read_text()) if audit_path.exists() else {}

roadscore/prototype/test_playback_cache.py

@@ -0,0 +1,43 @@
+import json
+from pathlib import Path
+import tempfile
+import unittest
+from playback_cache import SCHEMA,validated_playback
+from route_library import local_source
+
+ROUTE='0123456789abcdef/2026-09-19--00-00-00'
+NAME=ROUTE.split('/')[1]
+class PlaybackCacheTests(unittest.TestCase):
+ def setUp(self):
+  self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
+  self.root=Path(self.tmp.name);self.parent=self.root/'routes'/ROUTE
+  entries=[];derived=[]
+  for segment in range(2):
+   for name in ['rlog.zst','fcamera.hevc']:
+    source=f'{NAME}--{segment}/{name}';p=self.parent/source;p.parent.mkdir(parents=True,exist_ok=True);p.write_bytes(b'original')
+    target=source.replace('.hevc','.h264');q=self.parent/'playback'/target;q.parent.mkdir(parents=True,exist_ok=True);q.write_bytes(b'original')
+    entries.append(dict(path=source,bytes=8));row=dict(source=source,path=target,source_bytes=8,source_mtime_ns=p.stat().st_mtime_ns,bytes=8)
+    if name.endswith('hevc'):row['video_validation']=dict(source_frames=1200,output_frames=1200,width=1928,height=1208,frame_order_verified=True,timing_verified=True)
+    derived.append(row)
+  (self.parent/'cache_manifest.json').write_text(json.dumps(dict(route=ROUTE,files=entries)))
+  self.manifest=dict(schema=SCHEMA,route=ROUTE,complete=True,files=derived);self.save()
+ def save(self):
+  (self.parent/'playback/playback_manifest.json').write_text(json.dumps(self.manifest))
+ def test_normal_path_selects_complete_derived_cache(self):
+  self.assertEqual(local_source(ROUTE,self.root),self.parent/'playback')
+ def test_stale_source_falls_back(self):
+  (self.parent/self.manifest['files'][0]['source']).write_bytes(b'changed!')
+  self.assertEqual(local_source(ROUTE,self.root),self.parent)
+ def test_incomplete_wrong_route_and_acquiring_fall_back(self):
+  for field,value in [('route','other'),('complete',False),('files',self.manifest['files'][:-1])]:
+   before=self.manifest[field];self.manifest[field]=value;self.save();self.assertIsNone(validated_playback(self.parent,ROUTE));self.manifest[field]=before
+  self.save();(self.parent/'playback/.acquiring').touch();self.assertEqual(local_source(ROUTE,self.root),self.parent)
+ def test_video_count_order_and_missing_output_rejected(self):
+  video=self.manifest['files'][1]['video_validation']
+  for field,value in [('output_frames',1199),('frame_order_verified',False),('timing_verified',False)]:
+   before=video[field];video[field]=value;self.save();self.assertIsNone(validated_playback(self.parent,ROUTE));video[field]=before
+  self.save();(self.parent/'playback'/self.manifest['files'][1]['path']).unlink();self.assertIsNone(validated_playback(self.parent,ROUTE))
+ def test_path_escape_and_duplicate_rejected(self):
+  self.manifest['files'][0]['path']='../outside';self.save();self.assertIsNone(validated_playback(self.parent,ROUTE))
+
+if __name__=='__main__':unittest.main()