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