| """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/<task_name>.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" |
|
|
| |
| |
| 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 |
| return out |
|
|
