diff --git a/roadscore/prototype/playback_cache.py b/roadscore/prototype/playback_cache.py new file mode 100644 index 0000000000..131d199772 --- /dev/null +++ b/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 diff --git a/roadscore/prototype/route_library.py b/roadscore/prototype/route_library.py index f924ab9300..ca9cec643d 100644 --- a/roadscore/prototype/route_library.py +++ b/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 {} diff --git a/roadscore/prototype/test_playback_cache.py b/roadscore/prototype/test_playback_cache.py new file mode 100644 index 0000000000..8f947af4d5 --- /dev/null +++ b/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()