"""Typed loading and validation for the standalone H2 deployment config.""" from __future__ import annotations from dataclasses import dataclass import math from pathlib import Path import yaml from .h2_contract import COMMAND_JOINT_MARGIN_RAD from .deployment_versions import deployment_bundle PROJECT_ROOT = Path(__file__).resolve().parents[1] def _resolve(value: str) -> Path: path = Path(str(value)).expanduser() return path.resolve() if path.is_absolute() else (PROJECT_ROOT / path).resolve() def _section(data: dict, key: str) -> dict: value = data.get(key, {}) if not isinstance(value, dict): raise ValueError(f"config section {key!r} must be a mapping") return value def _bounded_float(name: str, value: object, lower: float, upper: float) -> float: result = float(value) if not math.isfinite(result) or not lower <= result <= upper: raise ValueError(f"{name} must be finite and in [{lower},{upper}], got {value!r}") return result @dataclass(frozen=True) class DeploySettings: source: Path deployment_version: str policy_path: Path policy_device: str generator_device: str | None generator_sampling_steps: int | None generator_warmup_steps: int seed: int hardware_joint_map_revision: str landing_xy: tuple[float, float] active_latency_skip_s: float hit_plane_tolerance_m: float joint_limit_tolerance_rad: float command_joint_margin_rad: float stop_hold_torque_fraction: float max_joint_velocity_rad_s: float max_tilt_rad: float max_abs_incoming_velocity_mps: float robot_mocap_timeout_s: float ball_mocap_timeout_s: float max_control_gap_s: float ready_position_tolerance_rad: float ready_arm_position_tolerance_rad: float ready_velocity_tolerance_rad_s: float ready_settle_duration_s: float ready_post_interpolation_timeout_s: float max_initial_root_xy_error_m: float max_mocap_root_jump_m: float max_mocap_root_speed_mps: float max_mocap_root_xy_m: float max_mocap_heading_jump_rad: float max_mocap_heading_rate_rad_s: float def load_settings(path: str | Path, *, deployment_version: str | None = None) -> DeploySettings: source = Path(path).expanduser().resolve() with source.open("r", encoding="utf-8") as stream: data = yaml.safe_load(stream) or {} if not isinstance(data, dict): raise ValueError("top-level config must be a mapping") if str(data.get("robot", "")).lower() != "h2": raise ValueError("this deployment package only accepts robot: h2") policy = _section(data, "policy") version = deployment_version if deployment_version is not None else str(data.get("deployment_version", "original")) bundle = deployment_bundle(version) if "path" in policy and _resolve(str(policy["path"])) != bundle.policy_path: raise ValueError( f"policy.path must resolve exactly to the {version} bundle policy: {bundle.policy_path}; " "remove policy.path and select --deployment-version original|v1|v2" ) generator = _section(data, "generator") command = _section(data, "command") safety = _section(data, "safety") hardware = _section(data, "hardware") policy_device = str(policy.get("device", "cpu")).lower() if policy_device not in ("cpu", "cuda"): raise ValueError("policy.device must be cpu or cuda") generator_device = generator.get("device") if generator_device is not None: generator_device = str(generator_device).lower() if generator_device not in ("cpu", "cuda"): raise ValueError("generator.device must be cpu, cuda or null") sampling_steps = generator.get("sampling_steps") sampling_steps = None if sampling_steps is None else int(sampling_steps) landing = tuple(float(x) for x in command.get("target_landing_xy", [2.87, 0.0])) if len(landing) != 2 or not all(math.isfinite(value) for value in landing): raise ValueError("command.target_landing_xy must have two finite values") settings = DeploySettings( source=source, deployment_version=version, policy_path=bundle.policy_path, policy_device=policy_device, generator_device=generator_device, generator_sampling_steps=sampling_steps, generator_warmup_steps=max(0, int(generator.get("warmup_steps", 1))), seed=int(data.get("seed", 0)), hardware_joint_map_revision=str(hardware.get("joint_map_revision", "")), landing_xy=(landing[0], landing[1]), active_latency_skip_s=_bounded_float( "command.active_latency_skip_s", command.get("active_latency_skip_s", 0.0), 0.0, 1.0 ), hit_plane_tolerance_m=_bounded_float( "safety.hit_plane_tolerance_m", safety.get("hit_plane_tolerance_m", 0.06), 0.0, 0.20 ), joint_limit_tolerance_rad=_bounded_float( "safety.joint_limit_tolerance_rad", safety.get("joint_limit_tolerance_rad", 0.05), 0.0, 0.20, ), command_joint_margin_rad=_bounded_float( "safety.command_joint_margin_rad", safety.get("command_joint_margin_rad", COMMAND_JOINT_MARGIN_RAD), COMMAND_JOINT_MARGIN_RAD, 0.10, ), stop_hold_torque_fraction=_bounded_float( "safety.stop_hold_torque_fraction", safety.get("stop_hold_torque_fraction", 0.90), 0.25, 0.90, ), max_joint_velocity_rad_s=_bounded_float( "safety.max_joint_velocity_rad_s", safety.get("max_joint_velocity_rad_s", 40.0), 0.1, 60.0, ), max_tilt_rad=_bounded_float( "safety.max_tilt_rad", safety.get("max_tilt_rad", 0.80), 0.1, 1.2 ), max_abs_incoming_velocity_mps=_bounded_float( "safety.max_abs_incoming_velocity_mps", safety.get("max_abs_incoming_velocity_mps", 15.0), 0.1, 30.0, ), robot_mocap_timeout_s=_bounded_float( "safety.robot_mocap_timeout_s", safety.get("robot_mocap_timeout_s", 0.10), 0.02, 0.25, ), ball_mocap_timeout_s=_bounded_float( "safety.ball_mocap_timeout_s", safety.get("ball_mocap_timeout_s", 0.12), 0.02, 0.25, ), max_control_gap_s=_bounded_float( "safety.max_control_gap_s", safety.get("max_control_gap_s", 0.06), 0.02, 0.099 ), ready_position_tolerance_rad=_bounded_float( "safety.ready_position_tolerance_rad", safety.get("ready_position_tolerance_rad", 0.10), 0.001, 0.30, ), ready_arm_position_tolerance_rad=_bounded_float( "safety.ready_arm_position_tolerance_rad", safety.get("ready_arm_position_tolerance_rad", 0.55), 0.30, 0.60, ), ready_velocity_tolerance_rad_s=_bounded_float( "safety.ready_velocity_tolerance_rad_s", safety.get("ready_velocity_tolerance_rad_s", 1.0), 0.01, 3.0, ), ready_settle_duration_s=_bounded_float( "safety.ready_settle_duration_s", safety.get("ready_settle_duration_s", 0.50), 0.10, 3.0, ), ready_post_interpolation_timeout_s=_bounded_float( "safety.ready_post_interpolation_timeout_s", safety.get("ready_post_interpolation_timeout_s", 10.0), 2.0, 30.0, ), max_initial_root_xy_error_m=_bounded_float( "safety.max_initial_root_xy_error_m", safety.get("max_initial_root_xy_error_m", 0.30), 0.01, 0.50, ), max_mocap_root_jump_m=_bounded_float( "safety.max_mocap_root_jump_m", safety.get("max_mocap_root_jump_m", 0.10), 0.01, 0.50, ), max_mocap_root_speed_mps=_bounded_float( "safety.max_mocap_root_speed_mps", safety.get("max_mocap_root_speed_mps", 2.0), 0.1, 5.0, ), max_mocap_root_xy_m=_bounded_float( "safety.max_mocap_root_xy_m", safety.get("max_mocap_root_xy_m", 0.75), 0.10, 2.0, ), max_mocap_heading_jump_rad=_bounded_float( "safety.max_mocap_heading_jump_rad", safety.get("max_mocap_heading_jump_rad", 0.25), 0.01, 1.0, ), max_mocap_heading_rate_rad_s=_bounded_float( "safety.max_mocap_heading_rate_rad_s", safety.get("max_mocap_heading_rate_rad_s", 4.0), 0.1, 10.0, ), ) if not settings.hardware_joint_map_revision: raise ValueError("hardware.joint_map_revision is required for the H2 real-robot contract") if settings.max_control_gap_s >= 0.10: raise ValueError("safety.max_control_gap_s must stay below the 0.10s bridge watchdog") if settings.max_initial_root_xy_error_m > settings.max_mocap_root_xy_m: raise ValueError("initial root gate cannot exceed the runtime root-domain gate") if settings.ready_arm_position_tolerance_rad < settings.ready_position_tolerance_rad: raise ValueError("ready arm tolerance cannot be smaller than the common ready tolerance") if not settings.policy_path.is_file(): raise FileNotFoundError(f"H2 policy not found: {settings.policy_path}") return settings __all__ = [ "DeploySettings", "PROJECT_ROOT", "load_settings", ]