| """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", |
| ] |
|
|
