From 2205038bcc0c7b4f42975e883d3b4b3d5aee8d3b Mon Sep 17 00:00:00 2001 From: isaac0804 Date: Wed, 3 Jun 2026 20:53:35 +0100 Subject: [PATCH 1/6] =?UTF-8?q?Filter=20vision=20detections=20to=20roster?= =?UTF-8?q?=20defined=20by=20vision=E2=86=92cmd=20mapping?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PositionRefiner now accepts allowed_yellow_ids and allowed_blue_ids. Any robot ID reported by SSL-Vision that is not in the allowlist is dropped before it enters the game state, preventing stray detections from permanently accumulating in friendly_robots/enemy_robots. StrategyRunner derives the allowlists from yellow/blue vision_to_cmd_mapping keys so the roster is implicitly defined by what is configured in gefr.py (or any other runner) — no extra config needed. --- .../data_processing/refiners/position.py | 30 +++++++++++++++---- utama_core/run/strategy_runner.py | 27 +++++++++++++++++ 2 files changed, 52 insertions(+), 5 deletions(-) diff --git a/utama_core/data_processing/refiners/position.py b/utama_core/data_processing/refiners/position.py index 9036efe4..0e8302a5 100644 --- a/utama_core/data_processing/refiners/position.py +++ b/utama_core/data_processing/refiners/position.py @@ -60,6 +60,8 @@ def __init__( full_field_dims: FieldDimensions, filtering: bool = True, exp_ball: bool = True, + allowed_yellow_ids: Optional[frozenset[int]] = None, + allowed_blue_ids: Optional[frozenset[int]] = None, ): # alpha=0 means no change in angle (inf smoothing), alpha=1 means no smoothing self.angle_smoother = AngleSmoother(alpha=1) @@ -80,6 +82,8 @@ def __init__( ) self.exp_ball = exp_ball + self.allowed_yellow_ids = allowed_yellow_ids + self.allowed_blue_ids = allowed_blue_ids if self.filtering: # Instantiate a dedicated Kalman filter for each robot so filtering can be kept independent. @@ -238,13 +242,22 @@ def _include_vanished_robots( game_frame.enemy_robots.keys(), ) - # Current vision IDs - yellow_present = {r.id for r in vision_data.yellow_robots} - blue_present = {r.id for r in vision_data.blue_robots} + # Current vision IDs (filtered to roster if provided) + yellow_robots = vision_data.yellow_robots + blue_robots = vision_data.blue_robots + if self.allowed_yellow_ids is not None: + yellow_robots = [r for r in yellow_robots if r.id in self.allowed_yellow_ids] + yellow_ids_last_frame = yellow_ids_last_frame & self.allowed_yellow_ids + if self.allowed_blue_ids is not None: + blue_robots = [r for r in blue_robots if r.id in self.allowed_blue_ids] + blue_ids_last_frame = blue_ids_last_frame & self.allowed_blue_ids + + yellow_present = {r.id for r in yellow_robots} + blue_present = {r.id for r in blue_robots} # Start with current measurements - yellow_vision_dict: dict[int, Optional[VisionRobotData]] = {r.id: r for r in vision_data.yellow_robots} - blue_vision_dict: dict[int, Optional[VisionRobotData]] = {r.id: r for r in vision_data.blue_robots} + yellow_vision_dict: dict[int, Optional[VisionRobotData]] = {r.id: r for r in yellow_robots} + blue_vision_dict: dict[int, Optional[VisionRobotData]] = {r.id: r for r in blue_robots} # Add None for vanished robots for robot_id in yellow_ids_last_frame - yellow_present: @@ -330,6 +343,13 @@ def _combine_both_teams_game_vision_positions( old_yellow_robots = game_frame.enemy_robots.copy() old_blue_robots = game_frame.friendly_robots.copy() + if self.allowed_yellow_ids is not None: + yellow_vision_robots = [r for r in yellow_vision_robots if r.id in self.allowed_yellow_ids] + old_yellow_robots = {k: v for k, v in old_yellow_robots.items() if k in self.allowed_yellow_ids} + if self.allowed_blue_ids is not None: + blue_vision_robots = [r for r in blue_vision_robots if r.id in self.allowed_blue_ids] + old_blue_robots = {k: v for k, v in old_blue_robots.items() if k in self.allowed_blue_ids} + new_yellow_robots = self._combine_single_team_positions( old_yellow_robots, yellow_vision_robots, diff --git a/utama_core/run/strategy_runner.py b/utama_core/run/strategy_runner.py index 35b4cf53..e8385643 100644 --- a/utama_core/run/strategy_runner.py +++ b/utama_core/run/strategy_runner.py @@ -208,6 +208,15 @@ def __init__( my_trusted_ir = yellow_trusted_ir_robots if my_team_is_yellow else blue_trusted_ir_robots opp_trusted_ir = blue_trusted_ir_robots if my_team_is_yellow else yellow_trusted_ir_robots + + # Derive per-color roster allowlists from the vision→cmd mappings (real mode only). + # Any robot ID seen by vision that is not in the allowlist is silently dropped so that + # stray detections from robots not in play never pollute the game state. + _allowed_yellow = ( + frozenset(yellow_vision_to_cmd_mapping.keys()) if yellow_vision_to_cmd_mapping is not None else None + ) + _allowed_blue = frozenset(blue_vision_to_cmd_mapping.keys()) if blue_vision_to_cmd_mapping is not None else None + self.my, self.opp = self._setup_sides_data( strategy, opp_strategy, @@ -216,6 +225,8 @@ def __init__( opp_control_scheme, my_trusted_ir_robots=my_trusted_ir, opp_trusted_ir_robots=opp_trusted_ir, + allowed_yellow_ids=_allowed_yellow, + allowed_blue_ids=_allowed_blue, ) ### functions below rely on self.my and self.opp ### @@ -481,6 +492,8 @@ def _setup_sides_data( opp_control_scheme: Optional[str], my_trusted_ir_robots: Optional[FrozenSet[int]] = None, opp_trusted_ir_robots: Optional[FrozenSet[int]] = None, + allowed_yellow_ids: Optional[FrozenSet[int]] = None, + allowed_blue_ids: Optional[FrozenSet[int]] = None, ) -> Tuple[SideRuntime, Optional[SideRuntime]]: """Setup the data structures for both sides (my team and opponent) Args: @@ -491,6 +504,8 @@ def _setup_sides_data( opp_control_scheme (Optional[str]): Name of the motion control scheme to use for the opponent team. If not set, uses same as friendly. my_trusted_ir_robots (FrozenSet[int], optional): Vision IDs of friendly robots whose IR sensor is trusted. opp_trusted_ir_robots (FrozenSet[int], optional): Vision IDs of opponent robots whose IR sensor is trusted. + allowed_yellow_ids (FrozenSet[int], optional): Vision IDs of yellow robots in play; others are ignored. + allowed_blue_ids (FrozenSet[int], optional): Vision IDs of blue robots in play; others are ignored. Side effect: Initializes the SideRuntime for both friendly and opponent sides, including their strategies, refiners, and motion controllers. @@ -503,6 +518,8 @@ def _setup_sides_data( filtering=filtering, exp_ball=self.exp_ball, trusted_ir_robots=my_trusted_ir_robots, + allowed_yellow_ids=allowed_yellow_ids, + allowed_blue_ids=allowed_blue_ids, ) my_motion_controller = get_control_scheme(control_scheme) my_strategy.setup_strategy_blackboard(is_opp_strat=False) @@ -520,6 +537,8 @@ def _setup_sides_data( filtering=filtering, exp_ball=self.exp_ball, trusted_ir_robots=opp_trusted_ir_robots, + allowed_yellow_ids=allowed_yellow_ids, + allowed_blue_ids=allowed_blue_ids, ) opp_motion_controller = ( get_control_scheme(opp_control_scheme) if opp_control_scheme is not None else my_motion_controller @@ -847,6 +866,8 @@ def _init_refiners( filtering: bool, exp_ball: bool = True, trusted_ir_robots: Optional[FrozenSet[int]] = None, + allowed_yellow_ids: Optional[FrozenSet[int]] = None, + allowed_blue_ids: Optional[FrozenSet[int]] = None, ) -> tuple[PositionRefiner, VelocityRefiner, RobotInfoRefiner]: """ Initialize the position, velocity, and robot info refiners. @@ -857,6 +878,10 @@ def _init_refiners( allowed to return None if no ball is detected in raw vision data. trusted_ir_robots (FrozenSet[int], optional): Vision IDs of robots whose IR sensor is trusted. See RobotInfoRefiner for details. + allowed_yellow_ids (FrozenSet[int], optional): Vision IDs of yellow robots that are in play. + Any robot ID seen by vision that is not in this set will be ignored. + allowed_blue_ids (FrozenSet[int], optional): Vision IDs of blue robots that are in play. + Any robot ID seen by vision that is not in this set will be ignored. Returns: tuple: The initialized PositionRefiner, VelocityRefiner, and RobotInfoRefiner. """ @@ -864,6 +889,8 @@ def _init_refiners( field_dims, filtering=filtering, exp_ball=exp_ball, + allowed_yellow_ids=allowed_yellow_ids, + allowed_blue_ids=allowed_blue_ids, ) velocity_refiner = VelocityRefiner() robot_info_refiner = RobotInfoRefiner(trusted_ir_robots=trusted_ir_robots) From d3afaef9ce3ca3ce83e202596e7f9250cf218625 Mon Sep 17 00:00:00 2001 From: isaac0804 Date: Wed, 3 Jun 2026 21:06:37 +0100 Subject: [PATCH 2/6] change cooldown kicker time --- utama_core/config/settings.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/utama_core/config/settings.py b/utama_core/config/settings.py index 17c16f5d..ddfdaaa0 100644 --- a/utama_core/config/settings.py +++ b/utama_core/config/settings.py @@ -31,7 +31,7 @@ BAUD_RATE = 115200 PORT = "/dev/ttyACM0" TIMEOUT = 0.1 -KICKER_COOLDOWN_TIME = 10 # in seconds to prevent kicker from being actuated too frequently +KICKER_COOLDOWN_TIME = 1 # in seconds to prevent kicker from being actuated too frequently KICKER_COOLDOWN_TIMESTEPS = int(KICKER_COOLDOWN_TIME * CONTROL_FREQUENCY) # in timesteps KICKER_PERSIST_TIMESTEPS = 10 # in timesteps to persist the kick command From 5743e9238960348456c7e36fae4794eaa623f646 Mon Sep 17 00:00:00 2001 From: isaac0804 Date: Wed, 3 Jun 2026 21:17:20 +0100 Subject: [PATCH 3/6] Enforce strict roster validation in real mode for single-team runs - Friendly vision_to_cmd_mapping is now required in real mode (single-team and PVP); previously only PVP enforced this. - Count check (mapping entries == exp_friendly/exp_enemy) now applies in single-team real mode, not just PVP. - Post-game-frame coverage check (_validate_mapping_covers_game_frame) now runs for single-team real mode, not just PVP. - Update tests: fix test that incorrectly assumed None mapping was valid for the friendly team; add 4 new tests for single-team validation paths. --- utama_core/run/strategy_runner.py | 41 ++++++++++------- .../strategy_runner/test_runner_misconfig.py | 45 +++++++++++++++++-- 2 files changed, 66 insertions(+), 20 deletions(-) diff --git a/utama_core/run/strategy_runner.py b/utama_core/run/strategy_runner.py index e8385643..4cc8dadc 100644 --- a/utama_core/run/strategy_runner.py +++ b/utama_core/run/strategy_runner.py @@ -308,12 +308,17 @@ def __init__( def _validate_vision_to_cmd_mapping(self, mapping: Optional[dict[int, int]], is_yellow: bool) -> dict[int, int]: if self.mode == Mode.REAL: - explicitly_provided = mapping is not None + is_my_team_color = is_yellow == self.my_team_is_yellow if mapping is None: if self.opp: raise ValueError( "explicit vision_to_cmd_mapping is required for both teams in real PVP/shared-transmitter mode." ) + if is_my_team_color: + raise ValueError( + "vision_to_cmd_mapping is required for the friendly team in real mode. " + "Provide a mapping from vision robot IDs to firmware command IDs." + ) return {} if not isinstance(mapping, dict): @@ -321,19 +326,22 @@ def _validate_vision_to_cmd_mapping(self, mapping: Optional[dict[int, int]], is_ f"vision_to_cmd_mapping must be a dictionary mapping vision robot IDs to command robot IDs; got {type(mapping).__name__}." ) - # if we are not running an opp strat, but mapping provided, warn that it will be ignored + # if we are not running an opp strat, but the opponent-color mapping provided, warn it will be ignored if self.opp is None and self.my_team_is_yellow ^ is_yellow: warnings.warn( "vision_to_cmd_mapping is provided but will be ignored since the opponent team is not being controlled." ) - if self.opp and explicitly_provided: - if is_yellow ^ self.my_team_is_yellow: - exp_count = self.exp_enemy - team_label = "opponent" - else: + # Count check: mapping must have exactly as many entries as expected robots for that team. + # Applied in both PVP and single-team real mode; the opponent-color mapping is skipped + # when there is no opp strategy (already warned above). + if is_my_team_color or self.opp: + if is_my_team_color: exp_count = self.exp_friendly team_label = "friendly" + else: + exp_count = self.exp_enemy + team_label = "opponent" # At init time we only know how many robots to expect, not their # actual vision IDs (those are non-contiguous in some deployments). @@ -342,8 +350,7 @@ def _validate_vision_to_cmd_mapping(self, mapping: Optional[dict[int, int]], is_ if len(mapping) != exp_count: raise ValueError( f"vision_to_cmd_mapping for {team_label} team has {len(mapping)} entries but " - f"{exp_count} robots are expected. Every robot must have a mapping entry " - "in shared-transmitter mode." + f"{exp_count} robots are expected. Mapping must include every robot in play." ) for vision_id, cmd_id in mapping.items(): @@ -932,19 +939,21 @@ def _load_game(self): # Validate mapping key coverage against the real observed vision IDs now # that we have a game frame (at init we only knew the expected count). - if self.opp: + # Applies in both PVP and single-team real mode. + if self.mode == Mode.REAL: my_mapping = ( self.yellow_vision_to_cmd_mapping if self.my_team_is_yellow else self.blue_vision_to_cmd_mapping ) - opp_mapping = ( - self.blue_vision_to_cmd_mapping if self.my_team_is_yellow else self.yellow_vision_to_cmd_mapping - ) self._validate_mapping_covers_game_frame( my_mapping, set(my_current_game_frame.friendly_robots.keys()), "friendly" ) - self._validate_mapping_covers_game_frame( - opp_mapping, set(opp_current_game_frame.friendly_robots.keys()), "opponent" - ) + if self.opp: + opp_mapping = ( + self.blue_vision_to_cmd_mapping if self.my_team_is_yellow else self.yellow_vision_to_cmd_mapping + ) + self._validate_mapping_covers_game_frame( + opp_mapping, set(opp_current_game_frame.friendly_robots.keys()), "opponent" + ) self.my.strategy.load_game(self.my.game) if self.opp: diff --git a/utama_core/tests/strategy_runner/test_runner_misconfig.py b/utama_core/tests/strategy_runner/test_runner_misconfig.py index 540f81ad..d6736e75 100644 --- a/utama_core/tests/strategy_runner/test_runner_misconfig.py +++ b/utama_core/tests/strategy_runner/test_runner_misconfig.py @@ -266,17 +266,53 @@ def test_strategy_runner_bounds_outside_non_standard_field_dims(): ) -def test_validate_vision_to_cmd_mapping_real_not_controlled(): +def test_validate_vision_to_cmd_mapping_real_opponent_not_controlled(): + # Opponent color mapping not required in single-team real mode runner = SimpleNamespace( mode=Mode.REAL, my_team_is_yellow=True, opp=None, - exp_friendly=3, + exp_friendly=1, ) - result = StrategyRunner._validate_vision_to_cmd_mapping(runner, None, True) + result = StrategyRunner._validate_vision_to_cmd_mapping(runner, None, False) assert result == {} +def test_validate_vision_to_cmd_mapping_real_friendly_mapping_required(): + # Friendly mapping is required in single-team real mode + runner = SimpleNamespace( + mode=Mode.REAL, + my_team_is_yellow=True, + opp=None, + exp_friendly=1, + ) + with pytest.raises(ValueError, match="vision_to_cmd_mapping is required for the friendly team"): + StrategyRunner._validate_vision_to_cmd_mapping(runner, None, True) + + +def test_validate_vision_to_cmd_mapping_single_team_count_check(): + # Single-team real mode: mapping count must match exp_friendly + runner = SimpleNamespace( + mode=Mode.REAL, + my_team_is_yellow=True, + opp=None, + exp_friendly=1, + ) + with pytest.raises(ValueError, match="has 2 entries but 1 robots are expected"): + StrategyRunner._validate_vision_to_cmd_mapping(runner, {0: 0, 1: 1}, True) + + +def test_validate_vision_to_cmd_mapping_single_team_correct_count_passes(): + runner = SimpleNamespace( + mode=Mode.REAL, + my_team_is_yellow=True, + opp=None, + exp_friendly=1, + ) + result = StrategyRunner._validate_vision_to_cmd_mapping(runner, {1: 0}, True) + assert result == {1: 0} + + def test_validate_vision_to_cmd_mapping_pvp_opp_mapping_missing(): runner = SimpleNamespace( mode=Mode.REAL, @@ -393,11 +429,12 @@ def test_validate_mapping_covers_game_frame_empty_mapping_passes(): def test_validate_vision_to_cmd_mapping_invalid_ids(): from utama_core.config.physical_constants import MAX_ROBOT_ID + # Use exp_friendly=1 so single-entry mappings pass the count check and hit the type/value checks runner = SimpleNamespace( mode=Mode.REAL, my_team_is_yellow=True, opp=None, - exp_friendly=3, + exp_friendly=1, ) with pytest.raises(TypeError, match="must map integers to integers"): StrategyRunner._validate_vision_to_cmd_mapping(runner, {0: "0"}, True) From 9f215d13b090a4b1da2b5ca29fb0c41a70b1f211 Mon Sep 17 00:00:00 2001 From: isaac0804 Date: Wed, 3 Jun 2026 21:34:55 +0100 Subject: [PATCH 4/6] Fix ordering: validate mappings before constructing PositionRefiner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously the allowlists passed to PositionRefiner were derived from raw (unvalidated) mapping params before _validate_vision_to_cmd_mapping ran, meaning a bad mapping could silently construct an incorrect refiner before raising. Also, rsim mode could temporarily get a non-None allowlist if a mapping was mistakenly provided. Fix: validate both mappings first, then derive frozenset allowlists from the validated self.*_vision_to_cmd_mapping dicts. Empty dict (rsim) → falsy frozenset → None (no filtering). Use opp_strategy as a sentinel for self.opp truthiness check during validation (overwritten immediately after by _setup_sides_data). --- utama_core/run/strategy_runner.py | 43 +++++++++++++++++-------------- 1 file changed, 23 insertions(+), 20 deletions(-) diff --git a/utama_core/run/strategy_runner.py b/utama_core/run/strategy_runner.py index 4cc8dadc..89bb4aa4 100644 --- a/utama_core/run/strategy_runner.py +++ b/utama_core/run/strategy_runner.py @@ -209,13 +209,31 @@ def __init__( my_trusted_ir = yellow_trusted_ir_robots if my_team_is_yellow else blue_trusted_ir_robots opp_trusted_ir = blue_trusted_ir_robots if my_team_is_yellow else yellow_trusted_ir_robots - # Derive per-color roster allowlists from the vision→cmd mappings (real mode only). + # Set self.opp to a sentinel before mapping validation so _validate_vision_to_cmd_mapping + # can check whether an opponent strategy is present (full SideRuntime is set later). + self.opp = opp_strategy # temporary sentinel; overwritten by _setup_sides_data below + + # Validate and store mappings before constructing refiners so that the + # allowlists passed to PositionRefiner are always derived from validated data. + self.yellow_vision_to_cmd_mapping = self._validate_vision_to_cmd_mapping( + yellow_vision_to_cmd_mapping, is_yellow=True + ) + self.blue_vision_to_cmd_mapping = self._validate_vision_to_cmd_mapping( + blue_vision_to_cmd_mapping, is_yellow=False + ) + self.yellow_cmd_to_vision_mapping = {v: k for k, v in self.yellow_vision_to_cmd_mapping.items()} + self.blue_cmd_to_vision_mapping = {v: k for k, v in self.blue_vision_to_cmd_mapping.items()} + + if self.opp and self.mode == Mode.REAL: + self._check_no_cmd_duplicate_if_transmission_sharing( + self.yellow_vision_to_cmd_mapping, self.blue_vision_to_cmd_mapping + ) + + # Derive per-color roster allowlists from the validated mappings (real mode only). # Any robot ID seen by vision that is not in the allowlist is silently dropped so that # stray detections from robots not in play never pollute the game state. - _allowed_yellow = ( - frozenset(yellow_vision_to_cmd_mapping.keys()) if yellow_vision_to_cmd_mapping is not None else None - ) - _allowed_blue = frozenset(blue_vision_to_cmd_mapping.keys()) if blue_vision_to_cmd_mapping is not None else None + _allowed_yellow = frozenset(self.yellow_vision_to_cmd_mapping) or None + _allowed_blue = frozenset(self.blue_vision_to_cmd_mapping) or None self.my, self.opp = self._setup_sides_data( strategy, @@ -234,21 +252,6 @@ def __init__( self.rsim_env, self.sim_controller = self._load_sim(rsim_noise, rsim_vanishing) self._assert_exp_robots_and_ball(exp_friendly, exp_enemy, exp_ball) - # mapping for mismatch between vision and cmd ids - self.yellow_vision_to_cmd_mapping = self._validate_vision_to_cmd_mapping( - yellow_vision_to_cmd_mapping, is_yellow=True - ) - self.blue_vision_to_cmd_mapping = self._validate_vision_to_cmd_mapping( - blue_vision_to_cmd_mapping, is_yellow=False - ) - self.yellow_cmd_to_vision_mapping = {v: k for k, v in self.yellow_vision_to_cmd_mapping.items()} - self.blue_cmd_to_vision_mapping = {v: k for k, v in self.blue_vision_to_cmd_mapping.items()} - - if self.opp and self.mode == Mode.REAL: - self._check_no_cmd_duplicate_if_transmission_sharing( - self.yellow_vision_to_cmd_mapping, self.blue_vision_to_cmd_mapping - ) - self._load_robot_controllers() # Remove Rsim ball. Rsim does not have the flexibilty to start without a ball. From 91888ec8f05280654c33fcae8dbea53a9d9c99b3 Mon Sep 17 00:00:00 2001 From: isaac0804 Date: Wed, 3 Jun 2026 21:56:44 +0100 Subject: [PATCH 5/6] Move mapping coverage check into GameGater _validate_mapping_covers_game_frame now lives in GameGater as a static method and is called from wait_until_game_valid, after the first valid game frame is available. StrategyRunner passes the real-mode mappings in; non-real modes pass None so no check runs. This keeps all "is the game state valid?" logic in the one place that owns waiting for a valid initial frame, rather than split between GameGater (robot/ball count) and StrategyRunner._load_game (ID coverage). --- utama_core/run/game_gater.py | 43 +++++++++++++++- utama_core/run/strategy_runner.py | 49 ++++--------------- .../strategy_runner/test_runner_misconfig.py | 15 +++--- 3 files changed, 57 insertions(+), 50 deletions(-) diff --git a/utama_core/run/game_gater.py b/utama_core/run/game_gater.py index 6e6a7318..39607c30 100644 --- a/utama_core/run/game_gater.py +++ b/utama_core/run/game_gater.py @@ -1,5 +1,5 @@ import time -from typing import Deque, List, Optional, Tuple +from typing import Deque, Dict, List, Optional, Tuple from utama_core.data_processing.refiners import PositionRefiner from utama_core.entities.data.raw_vision import RawVisionData @@ -8,6 +8,29 @@ class GameGater: + @staticmethod + def _validate_mapping_covers_game_frame(mapping: Dict[int, int], observed_ids: set, team_label: str) -> None: + """Verify that mapping keys exactly match observed vision IDs in the first game frame. + + Raises ValueError if any robot visible in the frame has no mapping entry, + or if the mapping names IDs that were not observed. + """ + if not mapping: + return + missing = observed_ids - mapping.keys() + extra = mapping.keys() - observed_ids + if missing or extra: + parts = [] + if missing: + parts.append(f"missing entries for observed IDs {sorted(missing)}") + if extra: + parts.append(f"extra entries for unseen IDs {sorted(extra)}") + raise ValueError( + f"vision_to_cmd_mapping for {team_label} team does not match observed vision IDs: " + + "; ".join(parts) + + f". Observed IDs: {sorted(observed_ids)}." + ) + @staticmethod def wait_until_game_valid( my_team_is_yellow: bool, @@ -20,6 +43,8 @@ def wait_until_game_valid( is_pvp: bool, rsim_env: SSLBaseEnv = None, wait_before_warn: float = 3.0, + my_vision_to_cmd_mapping: Optional[Dict[int, int]] = None, + opp_vision_to_cmd_mapping: Optional[Dict[int, int]] = None, ) -> Tuple[GameFrame, Optional[GameFrame]]: """ Waits until the game frame has the expected number of robots and a ball. @@ -88,4 +113,20 @@ def _add_frame(my_game_frame: GameFrame, opp_game_frame: GameFrame) -> Tuple[Gam if len(my_game_frame.enemy_robots) > exp_enemy: raise ValueError(f"Too many enemy robots: {len(my_game_frame.enemy_robots)} > {exp_enemy}") + # Validate that the vision→cmd mapping keys exactly match the observed robot IDs. + # Done here (after first valid frame) because actual vision IDs are non-contiguous + # in some deployments and are not known until a real frame arrives. + if my_vision_to_cmd_mapping is not None: + GameGater._validate_mapping_covers_game_frame( + my_vision_to_cmd_mapping, + set(my_game_frame.friendly_robots.keys()), + "friendly", + ) + if opp_vision_to_cmd_mapping is not None and opp_game_frame is not None: + GameGater._validate_mapping_covers_game_frame( + opp_vision_to_cmd_mapping, + set(opp_game_frame.friendly_robots.keys()), + "opponent", + ) + return my_game_frame, opp_game_frame diff --git a/utama_core/run/strategy_runner.py b/utama_core/run/strategy_runner.py index 89bb4aa4..a882a2c4 100644 --- a/utama_core/run/strategy_runner.py +++ b/utama_core/run/strategy_runner.py @@ -381,28 +381,6 @@ def _validate_vision_to_cmd_mapping(self, mapping: Optional[dict[int, int]], is_ ) return {} - def _validate_mapping_covers_game_frame(self, mapping: dict[int, int], observed_ids: set[int], team_label: str): - """After the first game frame loads, verify the mapping keys match the observed vision IDs. - - Called from _load_game() so we can validate against actual IDs rather than - assuming the contiguous 0..n-1 range (which fails non-contiguous deployments). - """ - if not mapping: - return - missing = observed_ids - mapping.keys() - extra = mapping.keys() - observed_ids - if missing or extra: - parts = [] - if missing: - parts.append(f"missing entries for observed IDs {sorted(missing)}") - if extra: - parts.append(f"extra entries for unseen IDs {sorted(extra)}") - raise ValueError( - f"vision_to_cmd_mapping for {team_label} team does not match observed vision IDs: " - + "; ".join(parts) - + f". Observed IDs: {sorted(observed_ids)}." - ) - def _check_no_cmd_duplicate_if_transmission_sharing( self, yellow_mapping: dict[int, int], blue_mapping: dict[int, int] ): @@ -913,6 +891,13 @@ def _load_game(self): Side effect: Populates game, game_history and current_game_frame on self.my (and self.opp if present). """ + my_mapping = self.yellow_vision_to_cmd_mapping if self.my_team_is_yellow else self.blue_vision_to_cmd_mapping + opp_mapping = ( + (self.blue_vision_to_cmd_mapping if self.my_team_is_yellow else self.yellow_vision_to_cmd_mapping) + if self.opp + else None + ) + my_current_game_frame, opp_current_game_frame = GameGater.wait_until_game_valid( self.my_team_is_yellow, self.my_team_is_right, @@ -923,6 +908,8 @@ def _load_game(self): self.my.position_refiner, is_pvp=self.opp is not None, rsim_env=self.rsim_env, + my_vision_to_cmd_mapping=my_mapping if self.mode == Mode.REAL else None, + opp_vision_to_cmd_mapping=opp_mapping if self.mode == Mode.REAL else None, ) self.my.position_refiner.start_filtering() @@ -940,24 +927,6 @@ def _load_game(self): self.opp.game = Game(self.opp.game_history, opp_current_game_frame, field=opp_field) self.opp.current_game_frame = opp_current_game_frame - # Validate mapping key coverage against the real observed vision IDs now - # that we have a game frame (at init we only knew the expected count). - # Applies in both PVP and single-team real mode. - if self.mode == Mode.REAL: - my_mapping = ( - self.yellow_vision_to_cmd_mapping if self.my_team_is_yellow else self.blue_vision_to_cmd_mapping - ) - self._validate_mapping_covers_game_frame( - my_mapping, set(my_current_game_frame.friendly_robots.keys()), "friendly" - ) - if self.opp: - opp_mapping = ( - self.blue_vision_to_cmd_mapping if self.my_team_is_yellow else self.yellow_vision_to_cmd_mapping - ) - self._validate_mapping_covers_game_frame( - opp_mapping, set(opp_current_game_frame.friendly_robots.keys()), "opponent" - ) - self.my.strategy.load_game(self.my.game) if self.opp: self.opp.strategy.load_game(self.opp.game) diff --git a/utama_core/tests/strategy_runner/test_runner_misconfig.py b/utama_core/tests/strategy_runner/test_runner_misconfig.py index d6736e75..9aa805b4 100644 --- a/utama_core/tests/strategy_runner/test_runner_misconfig.py +++ b/utama_core/tests/strategy_runner/test_runner_misconfig.py @@ -403,27 +403,24 @@ def test_validate_vision_to_cmd_mapping_correct_count_non_contiguous_ids_passes_ def test_validate_mapping_covers_game_frame_mismatch_raises(): - from utama_core.run.strategy_runner import StrategyRunner + from utama_core.run.game_gater import GameGater - runner = SimpleNamespace() with pytest.raises(ValueError, match="missing entries for observed IDs"): - StrategyRunner._validate_mapping_covers_game_frame(runner, {0: 0, 1: 1, 2: 2}, {5, 6, 7}, "friendly") + GameGater._validate_mapping_covers_game_frame({0: 0, 1: 1, 2: 2}, {5, 6, 7}, "friendly") def test_validate_mapping_covers_game_frame_match_passes(): - from utama_core.run.strategy_runner import StrategyRunner + from utama_core.run.game_gater import GameGater - runner = SimpleNamespace() # Should not raise - StrategyRunner._validate_mapping_covers_game_frame(runner, {5: 0, 6: 1, 7: 2}, {5, 6, 7}, "friendly") + GameGater._validate_mapping_covers_game_frame({5: 0, 6: 1, 7: 2}, {5, 6, 7}, "friendly") def test_validate_mapping_covers_game_frame_empty_mapping_passes(): - from utama_core.run.strategy_runner import StrategyRunner + from utama_core.run.game_gater import GameGater - runner = SimpleNamespace() # Empty mapping (non-PVP mode) always passes - StrategyRunner._validate_mapping_covers_game_frame(runner, {}, {0, 1, 2}, "friendly") + GameGater._validate_mapping_covers_game_frame({}, {0, 1, 2}, "friendly") def test_validate_vision_to_cmd_mapping_invalid_ids(): From f85d874bcbba82483a4831de03717db4073ad3d2 Mon Sep 17 00:00:00 2001 From: isaac0804 Date: Wed, 3 Jun 2026 22:03:37 +0100 Subject: [PATCH 6/6] Address Copilot review: print roster hint on gating hang, add allowlist tests When GameGater is waiting for a valid frame and an allowlist is active, the periodic status print now shows both the observed vision IDs and the expected IDs from the mapping so a misconfigured mapping is immediately visible in the terminal output rather than causing a silent hang. Also adds two PositionRefiner unit tests covering the allowlist path: one that verifies stray IDs are dropped when an allowlist is set, and one that verifies all IDs pass through when no allowlist is configured. --- utama_core/run/game_gater.py | 10 +++++-- .../tests/refiners/position_unit_test.py | 28 +++++++++++++++++++ 2 files changed, 36 insertions(+), 2 deletions(-) diff --git a/utama_core/run/game_gater.py b/utama_core/run/game_gater.py index 39607c30..0634805f 100644 --- a/utama_core/run/game_gater.py +++ b/utama_core/run/game_gater.py @@ -56,8 +56,14 @@ def wait_until_game_valid( def print_current_vision(game_frame: GameFrame): print("Waiting for valid game frame...") - print(f"Friendly robots: {len(game_frame.friendly_robots)}/{exp_friendly}") - print(f"Enemy robots: {len(game_frame.enemy_robots)}/{exp_enemy}") + friendly_ids = sorted(game_frame.friendly_robots.keys()) + enemy_ids = sorted(game_frame.enemy_robots.keys()) + print(f"Friendly robots: {len(game_frame.friendly_robots)}/{exp_friendly} observed IDs: {friendly_ids}") + if my_vision_to_cmd_mapping is not None: + print(f" expected IDs (from mapping): {sorted(my_vision_to_cmd_mapping.keys())}") + print(f"Enemy robots: {len(game_frame.enemy_robots)}/{exp_enemy} observed IDs: {enemy_ids}") + if opp_vision_to_cmd_mapping is not None: + print(f" expected IDs (from mapping): {sorted(opp_vision_to_cmd_mapping.keys())}") print(f"Ball present: {game_frame.ball is not None} (exp: {exp_ball})\n") def _add_frame(my_game_frame: GameFrame, opp_game_frame: GameFrame) -> Tuple[GameFrame, Optional[GameFrame]]: diff --git a/utama_core/tests/refiners/position_unit_test.py b/utama_core/tests/refiners/position_unit_test.py index e33d5b6f..a877b951 100644 --- a/utama_core/tests/refiners/position_unit_test.py +++ b/utama_core/tests/refiners/position_unit_test.py @@ -188,6 +188,34 @@ def test_out_of_bounds_friendly_not_added(): assert 1 not in result.friendly_robots +def test_allowlist_filters_stray_yellow_robot(): + # Vision sees yellow IDs 0 and 1; allowlist only permits ID 0. + raw_yellow = [RawRobotData(0, -1, -1, 0, 1), RawRobotData(1, -2, -2, 0, 1)] + raw_balls = [RawBallData(0, 0, 0, 0)] + frames = [RawVisionData(0, raw_yellow, [], raw_balls, 0)] + + p = PositionRefiner(full_field_dims, allowed_yellow_ids=frozenset({0})) + g = GameFrame(0, True, True, {}, {}, bfac(0, 0)) + result = p.refine(g, frames) + + assert 0 in result.friendly_robots + assert 1 not in result.friendly_robots + + +def test_no_allowlist_passes_all_robots(): + # Without an allowlist, both IDs should be present. + raw_yellow = [RawRobotData(0, -1, -1, 0, 1), RawRobotData(1, -2, -2, 0, 1)] + raw_balls = [RawBallData(0, 0, 0, 0)] + frames = [RawVisionData(0, raw_yellow, [], raw_balls, 0)] + + p = PositionRefiner(full_field_dims) + g = GameFrame(0, True, True, {}, {}, bfac(0, 0)) + result = p.refine(g, frames) + + assert 0 in result.friendly_robots + assert 1 in result.friendly_robots + + if __name__ == "__main__": test_combining_single_team_combines_single_robot() test_combining_with_robot_not_in_game_adds()