Upload/07_Reproduction/prepare_workspace.py

"""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)