File size: 5,594 Bytes
b7e9b58 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | """Assemble a separate reproduction workspace; never run Unity or training.
The released originals remain unchanged. Only new job copies have output/report
paths rebased, to ReproducedResults and Reports in the new workspace.
"""
from pathlib import Path
import argparse, json, shutil
RELEASE = Path(__file__).resolve().parents[1]
TASKS = ('RevisionEvaluation_Local_20260910','RevisionAdditional_20260911')
def safe_relative(value):
p=Path(value.replace('\\','/'))
if p.is_absolute() or '..' in p.parts or ':' in value:
raise ValueError('Unsafe relative path: '+value)
return p
def prepare(destination, dry_run=False):
destination=Path(destination).resolve()
if destination==RELEASE or destination in RELEASE.parents or destination.is_relative_to(RELEASE):
raise ValueError('Use a separate directory outside Upload.')
if destination.exists():
raise FileExistsError('Choose a new destination; existing folders are never overwritten.')
copies={}
def add_tree(source,target):
if not source.is_dir():raise FileNotFoundError(source)
for p in sorted(source.rglob('*')):
if p.is_file():
dest=destination/target/p.relative_to(source)
if dest in copies:raise ValueError('Duplicate destination: '+str(dest))
copies[dest]=p
def add_file(source,target):
if not source.is_file():raise FileNotFoundError(source)
dest=destination/target
if dest in copies:raise ValueError('Duplicate destination: '+str(dest))
copies[dest]=source
add_tree(RELEASE/'01_Source_Code/Unity_Evaluation',Path('RevisionEval'))
add_tree(RELEASE/'01_Source_Code/Python_Training',Path('Document/ShootingGame_IDE_Handoff/reference_originals/Python'))
add_tree(RELEASE/'01_Source_Code/Python_Evaluation/ShootingGame_IDE_Handoff',Path('Document/ShootingGame_IDE_Handoff'))
add_file(RELEASE/'04_Model_Weights/PyTorch/DT/E_1_DT_C_5.pth',Path('Document/ShootingGame_IDE_Handoff/models/E_1_DT_C_5.pth'))
add_file(RELEASE/'05_ONNX_Models/DT/E_1_DT_C_5.onnx',Path('Document/ShootingGame_IDE_Handoff/models/E_1_DT_C_5.onnx'))
jobs=[]
for task in TASKS:
base=Path('Document')/task
add_tree(RELEASE/'01_Source_Code/Python_Evaluation'/task/'Tools',base/'Tools')
evidence=RELEASE/'06_Evaluation_Results'/task
add_tree(evidence/'Results',base/'Results')
add_tree(evidence/'Reports',base/'ArchivedReports')
input_base=RELEASE/'07_Reproduction'/task
for name in ['Verification','Verification_Local']:
if (input_base/name).exists():add_tree(input_base/name,base/name)
add_tree(input_base/'Jobs',base/'OriginalJobs')
for p in sorted((input_base/'Jobs').glob('*.json')):
job=json.loads(p.read_text(encoding='utf-8-sig'))
normal=job['output'].replace('\\','/')
marker=f'/Document/{task}/Results/'
if marker not in normal:raise ValueError('Unrecognized original job output: '+normal)
suffix=safe_relative(normal.split(marker,1)[1])
job['output']=(destination/base/'ReproducedResults'/suffix).as_posix()
job['report']=(destination/base/'Reports'/(p.stem+'.json')).as_posix()
for model in job['models']:
if not (RELEASE/'01_Source_Code/Unity_Evaluation'/safe_relative(model)).is_file():
raise FileNotFoundError('Job model not included: '+model)
safe_relative(job['scene'])
jobs.append((destination/base/'Jobs'/p.name,job))
weights=Path('Document')/TASKS[1]/'Verification/Weights'
for n in (1,2,3):
name=f'E_{n}_DT_BC_100.pth'
add_file(RELEASE/'04_Model_Weights/PyTorch/BC'/name,weights/name)
add_file(RELEASE/'04_Model_Weights/PyTorch/DT/E_3_DT_S_100.pth',weights/'E_3_DT_S_100.pth')
# This report is an input manifest, not a previous success result.
add_file(RELEASE/'06_Evaluation_Results'/TASKS[1]/'Reports/reference-weight-manifest.json',
Path('Document')/TASKS[1]/'Reports/reference-weight-manifest.json')
# Current parity verification imports these exact original module files.
for name in ('model_dt.py','dataset_dt.py'):
target=destination/'Document'/TASKS[1]/'Verification'/name
if target not in copies:add_file(RELEASE/'01_Source_Code/Python_Training'/name,target.relative_to(destination))
summary={'destination':str(destination),'copy_files':len(copies),'copy_bytes':sum(p.stat().st_size for p in copies.values()),'rebased_jobs':len(jobs),'dry_run':dry_run,'runs_unity_or_training':False}
if dry_run:
print(json.dumps(summary,indent=2));return summary
destination.mkdir(parents=True,exist_ok=False)
for target,source in copies.items():
target.parent.mkdir(parents=True,exist_ok=True);shutil.copy2(source,target)
for target,job in jobs:
target.parent.mkdir(parents=True,exist_ok=True)
target.write_text(json.dumps(job,indent=2),encoding='utf-8')
for task in TASKS:
for name in ('Reports','Logs','ReproducedResults'):(destination/'Document'/task/name).mkdir(exist_ok=True)
(destination/'PREPARATION.json').write_text(json.dumps(summary,indent=2),encoding='utf-8')
print(json.dumps(summary,indent=2));return summary
if __name__=='__main__':
p=argparse.ArgumentParser(description=__doc__)
p.add_argument('--destination',required=True,type=Path)
p.add_argument('--dry-run',action='store_true')
args=p.parse_args();prepare(args.destination,args.dry_run)
|