File size: 4,976 Bytes
c28ffff | 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 98 99 100 101 102 103 104 105 106 107 108 109 110 | #!/usr/bin/env python3
"""Attach a downloaded ego-exo model bundle to code/core. Dry-run by default.
Creates links only where destinations are absent. Loader patches are applied
only to the exact known pre-sharding code version (or treated as already applied
if hashes match); other local edits are never overwritten. No inference/upload.
"""
import argparse
import hashlib
import json
from pathlib import Path
import tempfile
def digest(path):
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def safe_relative(name):
path = Path(name)
if path.is_absolute() or not path.parts or '..' in path.parts or '.' in path.parts:
raise ValueError('unsafe relative path in loader patch')
return path
def plan_install(bundle, code_root, patch_loaders=True):
bundle, code_root = Path(bundle).resolve(), Path(code_root).resolve()
if not (code_root / 'project_paths.py').is_file():
raise ValueError('--code-root must be the existing code/core directory')
if not (bundle / 'bundle_manifest.json').is_file():
raise ValueError('bundle is incomplete or has no verified bundle_manifest.json')
if json.loads((bundle / 'bundle_manifest.json').read_text()).get('status') != 'complete':
raise ValueError('bundle is not complete')
operations = []
for group in ('models', 'checkpoints'):
source_root = bundle / group
if not source_root.is_dir():
raise ValueError('missing bundle resource group: ' + group)
target_root = code_root / group
if target_root.exists() and not target_root.is_dir():
raise ValueError('resource parent is not a directory: ' + str(target_root))
for source in sorted(source_root.iterdir()):
if not source.is_dir():
continue
target = target_root / source.name
if target.is_symlink() and target.resolve() == source.resolve():
continue
if target.exists() or target.is_symlink():
raise ValueError('existing resource will not be overwritten: ' + str(target))
operations.append(dict(action='link', source=str(source), target=str(target)))
if patch_loaders:
patch_root = bundle / 'code_support'
manifest = json.loads((patch_root / 'loader_patch.json').read_text())
for row in manifest['files']:
relative = safe_relative(row['path'])
source, target = patch_root / relative, code_root / relative
if digest(source) != row['new_sha256']:
raise ValueError('bundle loader patch hash differs')
if target.is_symlink():
raise ValueError('refusing to patch a symlink: ' + str(target))
if target.exists() and not target.is_file():
raise ValueError('refusing to patch a non-file: ' + str(target))
current = digest(target) if target.is_file() else None
if current == row['new_sha256']:
continue
if current != row['previous_sha256']:
raise ValueError('unknown/local code changes will not be overwritten: ' + str(target))
operations.append(dict(action='patch', source=str(source), target=str(target)))
return operations
def install(bundle, code_root, apply=False, patch_loaders=True):
operations = plan_install(bundle, code_root, patch_loaders)
for row in operations:
print(row['action'].upper(), row['target'], '<-', row['source'])
if not apply:
print('DRY RUN: nothing changed. Add --apply after checking the paths.')
return operations
# Validate the entire plan before any writes. Never replace a resource directory.
for row in operations:
source, target = Path(row['source']), Path(row['target'])
target.parent.mkdir(parents=True, exist_ok=True)
if row['action'] == 'link':
target.symlink_to(source, target_is_directory=True)
else:
with tempfile.NamedTemporaryFile(dir=target.parent, prefix='.loader_patch-', delete=False) as handle:
temporary = Path(handle.name)
handle.write(source.read_bytes())
if target.exists():
temporary.chmod(target.stat().st_mode & 0o777)
else:
temporary.chmod(0o644)
temporary.replace(target)
print('ATTACHED. No training, inference, GPU use or upload occurred.')
return operations
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--bundle', type=Path, default=Path(__file__).resolve().parents[1])
parser.add_argument('--code-root', type=Path, required=True)
parser.add_argument('--apply', action='store_true')
parser.add_argument('--no-code-patch', action='store_true')
args = parser.parse_args()
install(args.bundle, args.code_root, args.apply, not args.no_code_patch)
if __name__ == '__main__':
main()
|