File size: 6,178 Bytes
06b25e4 | 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 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 | """Identify evaluation examples and protect saved prediction runs."""
import hashlib
from importlib import metadata
import json
import os
from pathlib import Path
def sample_identifier(record):
value = record.get('example_id', record.get('pair_id'))
if value is None:
raise ValueError('Each annotation must have an example_id or pair_id.')
value = str(value)
if not value or value in ('.', '..') or Path(value).name != value or '\\' in value:
raise ValueError('Sample IDs must be nonempty and contain no path separators.')
return value
def file_digest(path):
digest = hashlib.sha256()
with Path(path).open('rb') as handle:
for block in iter(lambda: handle.read(1024 * 1024), b''):
digest.update(block)
return digest.hexdigest()
def checkpoint_digest(path):
path = Path(path)
if path.is_file():
return file_digest(path)
suffixes = {'.safetensors', '.bin', '.pth', '.pt', '.json', '.model',
'.txt', '.py', '.vocab', '.merges'}
files = sorted(file for file in path.rglob('*')
if file.is_file() and file.suffix in suffixes
and not any(part.startswith('.') or part == 'code'
for part in file.relative_to(path).parts))
if not files:
raise ValueError(f'No model or tokenizer assets found in {path}.')
digest = hashlib.sha256()
for file in files:
digest.update(file.relative_to(path).as_posix().encode() + b'\0')
digest.update(bytes.fromhex(file_digest(file)))
return digest.hexdigest()
def index_prediction_files(directory):
files = {}
for path in sorted(Path(directory).rglob('*.pth')):
if path.stem in files:
raise ValueError(
f'Duplicate predictions for {path.stem}: {files[path.stem]} and {path}. '
'Keep one prediction per example before scoring.'
)
files[path.stem] = path
return files
def generation_identity(args, checkpoint):
asset_digests = {}
def digest_asset(value):
path = Path(value).resolve()
if path not in asset_digests:
asset_digests[path] = checkpoint_digest(path)
return asset_digests[path]
settings = ('precision', 'image_size', 'vit_image_size', 'model_max_length',
'max_num_images', 'num_query_tokens', 'lora_r', 'lora_alpha',
'lora_dropout', 'lora_target_modules', 'conv_type',
'use_mm_start_end', 'max_new_tokens', 'm4seg_data',
'n_chunks', 'chunk_idx')
identity = {
'schema_version': 2,
'annotation_sha256': file_digest(args.m4seg_annpath),
'checkpoint_sha256': digest_asset(checkpoint),
'initialization': 'merged' if args.merged_ckpt_path else 'backbone',
'dataset_root': str(Path(args.dataset_dir).resolve()),
'settings': {name: getattr(args, name) for name in settings},
'preprocessing': {
'truncate_from_left': os.getenv('PRIMA_TRUNCATE_FROM_LEFT', '0').strip().lower()
in ('1', 'true', 'yes', 'y', 'on'),
},
}
if args.merged_ckpt_path:
if args.ckpt_path and not args.ckpt_path.startswith('/path/to/'):
identity['overlay_sha256'] = digest_asset(args.ckpt_path)
else:
assets = ('version', 'q_former_model', 'pretrained_q_former_path',
'qformer_bert_path', 'eva_weights', 'pretrain_mm_mlp_adapter')
identity['initialization_assets'] = {
name: digest_asset(getattr(args, name)) for name in assets
if getattr(args, name, '') and not str(getattr(args, name)).startswith('/path/to/')
}
root = Path(__file__).resolve().parents[1]
digest = hashlib.sha256()
for directory in ('model', 'dataset', 'tools', 'eval'):
for path in sorted((root / directory).rglob('*.py')):
digest.update(path.relative_to(root).as_posix().encode() + b'\0')
digest.update(bytes.fromhex(file_digest(path)))
identity['source_sha256'] = digest.hexdigest()
dependencies = ('torch', 'torchvision', 'transformers', 'tokenizers', 'peft',
'timm', 'numpy', 'Pillow', 'opencv-python', 'opencv-python-headless',
'sentencepiece', 'einops')
identity['dependencies'] = {}
for name in dependencies:
try:
identity['dependencies'][name] = metadata.version(name)
except metadata.PackageNotFoundError:
identity['dependencies'][name] = None
return identity
def prepare_prediction_directory(output, identity, *, resume):
output = Path(output)
output.mkdir(parents=True, exist_ok=True)
provenance = output / 'generation.json'
existing = list(output.rglob('*.pth'))
if any(path.parent != output for path in existing):
raise ValueError('Use a separate output directory for each generation run or shard.')
if existing and not resume:
raise ValueError('Use an empty output directory or explicitly enable --resume.')
if provenance.exists():
if not resume:
raise ValueError('This output directory already has generation provenance. '
'Use --resume or a new directory.')
try:
previous = json.loads(provenance.read_text())
except (OSError, ValueError) as error:
raise ValueError('Cannot verify existing generation provenance. Use a new output directory.') from error
if previous != identity:
raise ValueError('Resume configuration does not match existing predictions. '
'Use a new output directory.')
return
if existing:
raise ValueError('Existing predictions have no generation provenance. '
'Use a new output directory; their configuration cannot be verified.')
try:
with provenance.open('x') as handle:
json.dump(identity, handle, indent=2)
handle.write('\n')
except FileExistsError as error:
raise ValueError('Another run has already claimed this output directory.') from error
|