"""YAML task configuration: scene parameters live in data, not only in code. Every task file declares its parameters as class attributes (ManiSkill's style, and what the solver reads). This module lets the same parameters be *overridden from a YAML file* without touching Python, so a scene can be re-tuned -- object scales, spawn points, jitter ranges, clamp force -- by editing data. yam/tasks/configs/.yaml Precedence, weakest first: class attribute -> YAML file -> --set CLI override. A YAML key that does not correspond to a class attribute is refused rather than silently attached: a typo'd key that quietly does nothing is worse than a missing file, because the run looks configured and is not. """ from __future__ import annotations from pathlib import Path CONFIG_DIR = Path(__file__).resolve().parents[1]/"tasks"/"configs" # these are structural, not scene parameters -- overriding them from data would change what the # task IS, not how it is set up PROTECTED = {"task_name", "title", "tags", "instruction", "solve", "evaluate", "_load_scene"} def config_path(task_name: str) -> Path: return CONFIG_DIR/f"{task_name}.yaml" def load(task_name: str) -> dict: """Read the task's YAML config, or {} if it has none.""" p = config_path(task_name) if not p.exists(): return {} try: import yaml except ImportError: print(f"[cfg] PyYAML not installed -- ignoring {p.name}", flush=True) return {} with open(p) as fh: data = yaml.safe_load(fh) or {} if not isinstance(data, dict): raise SystemExit(f"[cfg] {p} must be a mapping of parameter -> value, got {type(data).__name__}") return data def _coerce(value, template): """Restore the container types the code expects, using the class value as the template. YAML has no tuple, so `(90, 0, 0)` round-trips as a list and `{"a": (0.1, 0.2)}` comes back with a list inside. Most call sites index and would not care, but some unpack (`x, y = ...`), so match the template recursively rather than only at the top level. """ if isinstance(template, tuple) and isinstance(value, (list, tuple)): return tuple(_coerce(v, template[i] if i < len(template) else None) for i, v in enumerate(value)) if isinstance(template, dict) and isinstance(value, dict): return {k: _coerce(v, template.get(k)) for k, v in value.items()} if isinstance(template, list) and isinstance(value, (list, tuple)): return [_coerce(v, template[i] if i < len(template) else None) for i, v in enumerate(value)] return value def apply(task, overrides: dict, source: str = "yaml"): """Set validated overrides on a task INSTANCE (leaving the class untouched).""" for key, value in (overrides or {}).items(): if key in PROTECTED: raise SystemExit(f"[cfg] {source}: {key!r} is structural and cannot be overridden") if not hasattr(type(task), key): known = ", ".join(sorted(k for k in vars(type(task)) if not k.startswith("__"))) raise SystemExit(f"[cfg] {source}: {type(task).__name__} has no parameter {key!r}." f"\n known parameters: {known}") setattr(task, key, _coerce(value, getattr(type(task), key))) if overrides: print(f"[cfg] {source}: applied {sorted(overrides)}", flush=True) def parse_cli(pairs) -> dict: """`--set gripper_effort=70 --set grape_spawn=[-0.03,0.10]` -> dict.""" import ast out = {} for pair in pairs or []: if "=" not in pair: raise SystemExit(f"[cfg] --set expects key=value, got {pair!r}") k, v = pair.split("=", 1) try: out[k.strip()] = ast.literal_eval(v) except (ValueError, SyntaxError): out[k.strip()] = v # bare strings need no quoting on the command line return out