| """Explicit, independently resolved H2 training settings.""" |
| from __future__ import annotations |
|
|
| from dataclasses import asdict, dataclass, fields |
| from pathlib import Path |
| import math |
|
|
| import yaml |
|
|
|
|
| @dataclass(frozen=True) |
| class RLConfig: |
| seed: int = 42 |
| learner_device: str = "cuda" |
| warmup_balls: int = 400 |
| online_balls: int = 7600 |
| checkpoint_interval_balls: int = 10 |
| action_scales_rad: tuple[float, ...] = (0.15, 0.12, 0.12, 0.18, 0.4, 0.3, 0.35) |
| warmup_sigma: float = 0.3 |
| exploration_sigma: float = 0.3 |
| exploration_sigma_final: float = 0.05 |
| exploration_hold_steps: int = 10 |
| batch_size: int = 256 |
| replay_capacity: int = 200000 |
| actor_learning_rate: float = 5e-6 |
| actor_residual_l2: float = 2.0 |
| critic_learning_rate: float = 1e-4 |
| critic_warmup_steps: int = 3000 |
| critic_updates_per_step: int = 1 |
| actor_update_delay: int = 4 |
| discount: float = 0.99 |
| n_step: int = 50 |
| target_noise: float = 0.10 |
| target_noise_clip: float = 0.3 |
| target_tau: float = 0.005 |
| hidden_sizes: tuple[int, ...] = (256, 256) |
| layer_norm: bool = True |
| optimizer_betas: tuple[float, float] = (0.9, 0.999) |
| optimizer_eps: float = 1e-8 |
| weight_decay: float = 0.01 |
| gradient_clip: float = 1.0 |
| normalizer_min_std: float = 0.1 |
| normalizer_clip: float = 5.0 |
| hit_reward: float = 0.0 |
| success_bonus: float = 0.0 |
| success_reward: float = 60.0 |
| landing_scale_m: float = 0.4 |
| failure_reward: float = -20.0 |
| miss_failure_reward: float = -80.0 |
| position_reward_coefficient: float = 2.0 |
| position_reward_scale_m: float = 0.15 |
| position_reward_window_before_s: float = 0.25 |
| position_reward_window_after_s: float = 0.30 |
| pre_hit_timeout_s: float = 0.40 |
| post_hit_timeout_s: float = 2.0 |
| velocity_window: int = 5 |
| incoming_vx_mps: float = -0.5 |
| outgoing_vx_mps: float = 0.3 |
| outgoing_confirmations: int = 2 |
| table_height_tolerance_m: float = 0.03 |
| net_x_tolerance_m: float = 0.04 |
| max_ball_gap_s: float = 0.05 |
| zmq_endpoint: str = "tcp://127.0.0.1:5556" |
| output_dir: str = "artifacts/real_rl_runs" |
| episode_queue_capacity: int = 16 |
| logging_queue_capacity: int = 512 |
| ball_queue_capacity: int = 4096 |
| max_episode_steps: int = 500 |
|
|
| def __post_init__(self): |
| for name in ("action_scales_rad", "hidden_sizes", "optimizer_betas"): |
| object.__setattr__(self, name, tuple(getattr(self, name))) |
| positive_ints = ("warmup_balls", "online_balls", "checkpoint_interval_balls", |
| "batch_size", "replay_capacity", "critic_updates_per_step", |
| "actor_update_delay", "n_step", "exploration_hold_steps", "velocity_window", |
| "outgoing_confirmations", "episode_queue_capacity", |
| "logging_queue_capacity", "ball_queue_capacity", "max_episode_steps") |
| for name in positive_ints: |
| value = getattr(self, name) |
| if isinstance(value, bool) or not isinstance(value, int) or value <= 0: |
| raise ValueError(f"rl.{name} must be a positive integer") |
| if not isinstance(self.seed, int) or self.seed < 0: |
| raise ValueError("rl.seed must be a nonnegative integer") |
| if not isinstance(self.critic_warmup_steps, int) or self.critic_warmup_steps < 0: |
| raise ValueError("rl.critic_warmup_steps must be a nonnegative integer") |
| for field in fields(self): |
| value = getattr(self, field.name) |
| if isinstance(value, float) and not math.isfinite(value): |
| raise ValueError(f"rl.{field.name} must be finite") |
| for name in ("actor_learning_rate", "critic_learning_rate", "optimizer_eps", |
| "gradient_clip", "normalizer_min_std", "normalizer_clip", |
| "landing_scale_m", "position_reward_scale_m", |
| "pre_hit_timeout_s", "post_hit_timeout_s", |
| "max_ball_gap_s", "table_height_tolerance_m", "net_x_tolerance_m"): |
| if getattr(self, name) <= 0: |
| raise ValueError(f"rl.{name} must be positive") |
| for name in ("warmup_sigma", "exploration_sigma", "exploration_sigma_final", "target_noise", "target_noise_clip", |
| "actor_residual_l2", "weight_decay", "position_reward_window_before_s", "position_reward_window_after_s", |
| "position_reward_coefficient", "hit_reward", "success_bonus", "success_reward"): |
| if getattr(self, name) < 0: |
| raise ValueError(f"rl.{name} must be nonnegative") |
| if len(self.action_scales_rad) != 7 or any(not math.isfinite(v) or v <= 0 for v in self.action_scales_rad): |
| raise ValueError("rl.action_scales_rad requires seven positive finite radians") |
| if not 0 < self.discount <= 1 or not 0 < self.target_tau <= 1: |
| raise ValueError("discount and target_tau must be in (0, 1]") |
| if self.hidden_sizes != (256, 256) or self.layer_norm is not True: |
| raise ValueError("H2 residual architecture is [256,256] with LayerNorm") |
| if len(self.optimizer_betas) != 2 or any(not 0 <= v < 1 for v in self.optimizer_betas): |
| raise ValueError("optimizer betas must be in [0,1)") |
| if self.failure_reward > 0 or self.miss_failure_reward > 0: |
| raise ValueError("rl failure rewards must be nonpositive") |
| if self.incoming_vx_mps >= 0 or self.outgoing_vx_mps <= 0 or self.velocity_window < 5: |
| raise ValueError("invalid real-ball velocity detector settings") |
| if self.replay_capacity < self.batch_size: |
| raise ValueError("replay_capacity must cover one batch") |
| if not str(self.learner_device).startswith("cuda"): |
| raise ValueError("production H2 learner_device must be CUDA") |
|
|
| def to_dict(self): |
| return asdict(self) |
|
|
|
|
| def load_rl_config(path: str | Path) -> RLConfig: |
| with Path(path).open() as stream: |
| data = yaml.safe_load(stream) or {} |
| section = data.get("rl", {}) |
| if not isinstance(section, dict): |
| raise ValueError("rl must be a mapping") |
| unknown = section.keys() - {f.name for f in fields(RLConfig)} |
| if unknown: |
| raise ValueError(f"unknown H2 RL settings: {sorted(unknown)}") |
| return RLConfig(**section) |
|
|
