"""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 # Legacy name: local landing POSITION window only. 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)