File size: 2,593 Bytes
91c1e9e | 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 | """Shared configuration parsing for PRIMA commands."""
import argparse
from pathlib import Path
def parse_with_config(parser, argv=None):
import yaml
parser.add_argument('--config', type=Path, help='YAML configuration; CLI options override it.')
probe = argparse.ArgumentParser(add_help=False)
probe.add_argument('--config', type=Path)
known, _ = probe.parse_known_args(argv)
if known.config:
with known.config.open() as handle:
values = yaml.safe_load(handle) or {}
if not isinstance(values, dict):
parser.error('Configuration must be a mapping of option names to values.')
actions = {action.dest: action for action in parser._actions}
unknown = set(values) - set(actions)
if unknown:
parser.error('Unknown configuration options: ' + ', '.join(sorted(unknown)))
for key, value in values.items():
action = actions[key]
if isinstance(action, (argparse._StoreTrueAction, argparse._StoreFalseAction,
argparse.BooleanOptionalAction)) and not isinstance(value, bool):
parser.error(f'{key} must be a YAML boolean.')
if action.choices and value not in action.choices:
parser.error(f'{key} must be one of {action.choices}.')
parser.set_defaults(**values)
return parser.parse_args(argv)
def validate_paths(args, *, checkpoint=True, dataset=False):
merged = getattr(args, 'merged_ckpt_path', '')
fields = ['merged_ckpt_path'] if merged else ['version', 'q_former_model', 'qformer_bert_path', 'eva_weights']
overlay = getattr(args, 'ckpt_path', '')
if checkpoint and (not merged or (overlay and not overlay.startswith('/path/to/'))):
fields.append('ckpt_path')
if dataset:
fields.append('dataset_dir')
errors = []
for field in fields:
value = getattr(args, field, None)
if not value or str(value).startswith('/path/to/'):
errors.append(f'{field}: replace the placeholder with a local path')
elif field == 'merged_ckpt_path' and not Path(value).expanduser().exists():
import re
if not re.fullmatch(r'[\w.-]+/[\w.-]+', str(value)):
errors.append(f'{field}: provide an existing model directory or organization/model Hub ID')
elif not Path(value).expanduser().exists():
errors.append(f'{field}: path does not exist: {value}')
if errors:
raise ValueError('Configure the required paths before running:\n ' + '\n '.join(errors))
|