Merge pull request #115 from ARRC-Rocket/fix/a-non-finite-action-is-not-a-command

thc1006 · web-flow · commit 915b09af88cd · 2026-07-29T22:48:22.000+08:00
An action the environment cannot use is not a run the agent should lose
diff --git a/BalloonPoppingGymEnv/envs/balloon_world.py b/BalloonPoppingGymEnv/envs/balloon_world.py
@@ -30,6 +30,81 @@
 
 logger = logging.getLogger(__name__)
 
+# How many numbers each action field holds. The environment indexes `tvc` and
+# `launch_inclination_heading` by position, so a wrong count is not usable
+# either, and an empty array passes a finiteness test on its own.
+ACTION_FIELD_SIZES = {
+    "launch_inclination_heading": 2,
+    "tvc": 2,
+    "roll": 1,
+    "throttle": 1,
+}
+
+
+# Which repetition of an unusable action gets a log line. A broken policy stays
+# broken, and one line per step for the rest of the horizon is its own problem.
+_UNUSABLE_ACTION_LOG_STEPS = frozenset({1, 10, 100, 1000})
+
+
+def _is_launch_command(action):
+    """Whether `action["launch"]` asks for a launch, without guessing.
+
+    `bool(np.nan)` is True and a two element array raises, so the truthiness of
+    whatever arrives is not a safe question to ask directly.
+    """
+    try:
+        values = np.asarray(action["launch"]).reshape(-1)
+    except (KeyError, TypeError, ValueError):
+        return False
+    if values.size != 1:
+        return False
+    try:
+        value = float(values[0])
+    except (TypeError, ValueError, OverflowError):
+        return False
+    # Finite first: `nan != 0` is True, so a NaN would otherwise launch.
+    return bool(np.isfinite(value)) and value != 0.0
+
+
+def check_action(action):
+    """The action fields the environment cannot use, by name.
+
+    A policy that diverges emits NaN, and it is the most ordinary failure there
+    is while training one. An empty list means every field is usable.
+
+    Public so an agent can ask the same question before returning an action:
+
+        if check_action(action):
+            action = last_good_action
+    """
+    return sorted(_usable_action_fields(action)[1])
+
+
+def _usable_action_fields(action):
+    """The fields as arrays of the right size, and the names of the rest.
+
+    Validating a converted copy and then handing the original to the actuators
+    would check something other than what is used: a scalar `tvc` converts to a
+    finite array and then fails at `action["tvc"][0]`. So the converted values
+    are what the caller gets.
+    """
+    usable = {}
+    unusable = set()
+    for field, size in ACTION_FIELD_SIZES.items():
+        if field not in action:
+            unusable.add(field)
+            continue
+        try:
+            values = np.atleast_1d(np.asarray(action[field], dtype=float))
+        except (TypeError, ValueError, OverflowError):
+            unusable.add(field)
+            continue
+        if values.size != size or not np.all(np.isfinite(values)):
+            unusable.add(field)
+            continue
+        usable[field] = values
+    return usable, unusable
+
 
 def _remove_monte_carlo_workspace(directory):
     """Delete a Monte Carlo scratch directory, saying so if it will not go.
@@ -148,6 +223,8 @@ def __init__(self, render_mode, parameters):
         self._balloon_states = np.array(np.zeros((self.balloon_parameters["num"], 6)))
         # (gyroX, gyroY, gyroZ, accX, accY, accZ, posX, posY, posZ, velX, velY, velZ)
         self._rocket_sensors = np.full(12, np.nan)
+        # Steps whose action the environment could not use, for rate limiting.
+        self._unusable_action_steps = 0
         # (posX, posY, posZ, velX, velY, velZ, e0, e1, e2, e3, w1, w2, w3)
         self._rocket_states = np.full(13, np.nan)
         # Where the next pop sweep starts. Tracked separately from
@@ -294,6 +371,8 @@ def reset(self, seed=None, options=None):
 
         self._balloon_states = self._balloon_flights[:, :, 0]
         self._rocket_sensors = np.full(12, np.nan)
+        # Steps whose action the environment could not use, for rate limiting.
+        self._unusable_action_steps = 0
         self._rocket_states = np.full(13, np.nan)
         self._sweep_origin = None
         self.trajectories = None
@@ -325,28 +404,53 @@ def step(self, action):
         ground_mask = self._balloon_status[:, 0] == 0
         self._balloon_status[released_mask & ground_mask, 0] = 1
 
+        # A diverging policy emits NaN, and it is the most ordinary failure there
+        # is while training one. A step it cannot command is not a run it should
+        # lose, so the command is dropped and the episode carries on.
+        commands, unusable = _usable_action_fields(action)
+        if unusable:
+            # Rate limited, since a policy that breaks stays broken and one line
+            # per step for the rest of the horizon is its own denial of service.
+            self._unusable_action_steps += 1
+            if self._unusable_action_steps in _UNUSABLE_ACTION_LOG_STEPS:
+                logger.warning(
+                    "Step %d: ignoring %s, which the environment cannot use "
+                    "(%d such steps so far)",
+                    self.current_step,
+                    ", ".join(sorted(unusable)),
+                    self._unusable_action_steps,
+                )
+
         if not self.rocket_launched:
             _rocket_finished = False
-            if action["launch"]:  # Init rocket flight with first launch action
-                self.rocket_launched = True
-                self.__get_init_rocket_states(
-                    action["launch_inclination_heading"][0],
-                    action["launch_inclination_heading"][1],
-                )
+            # Refused rather than dropped, because there is no previous attitude
+            # to fall back on. The agent can launch on any later step.
+            if _is_launch_command(action) and "launch_inclination_heading" in commands:
+                attitude = commands["launch_inclination_heading"]
+                self.__get_init_rocket_states(attitude[0], attitude[1])
                 self.initial_solution[0] = (
                     self.current_step * self.simulation_parameters["time_step"]
                 )
                 self.__init_rocket_simulation()
                 self._sweep_origin = np.asarray(self.initial_solution[1:4], dtype=float)
+                # Last, so a failure above leaves the environment unlaunched
+                # rather than launched with no flight for the next step to use.
+                self.rocket_launched = True
         else:  # Apply action to step the rocket simulation and get sensor measurements
-            self._rocket_flight.rocket.roll_control.roll_torque = action["roll"]
-            self._rocket_flight.rocket.thrust_vector_control.gimbal_angle_x = action[
-                "tvc"
-            ][0]
-            self._rocket_flight.rocket.thrust_vector_control.gimbal_angle_y = action[
-                "tvc"
-            ][1]
-            self._rocket_flight.rocket.throttle_control.throttle = action["throttle"]
+            # Each actuator keeps the value it already holds when its field is
+            # dropped, which is what a real one does between commands.
+            if "roll" in commands:
+                self._rocket_flight.rocket.roll_control.roll_torque = float(
+                    commands["roll"][0]
+                )
+            if "tvc" in commands:
+                thrust_vector = self._rocket_flight.rocket.thrust_vector_control
+                thrust_vector.gimbal_angle_x = float(commands["tvc"][0])
+                thrust_vector.gimbal_angle_y = float(commands["tvc"][1])
+            if "throttle" in commands:
+                self._rocket_flight.rocket.throttle_control.throttle = float(
+                    commands["throttle"][0]
+                )
             self._rocket_flight.step_simulation()
             _sensor = self._rocket_flight.sensors
             self._rocket_sensors[:3] = _sensor[0].measurement  # gyro
diff --git a/tests/test_non_finite_actions.py b/tests/test_non_finite_actions.py
@@ -0,0 +1,211 @@
+"""What the environment does with an action it cannot use.
+
+A policy that diverges emits NaN, and it is the most ordinary failure there is
+while training one. Measured on the pin these tests were written against: a
+single NaN in `tvc` was harmless, the run finished with the same score as a
+clean one, while a NaN in `launch_inclination_heading` raised
+`ValueError: All components of the initial state y0 must be finite` and ended
+the evaluation. ActiveRocketPy #19 makes the actuator setters refuse a
+non-finite command too, so without this the first kind becomes the second.
+
+The rule here is that a field the environment cannot use is dropped, the
+actuator keeps what it holds, and the episode carries on. A run is not lost to
+one bad step.
+"""
+
+import logging
+import unittest
+from importlib.util import find_spec
+
+import numpy as np
+
+_STACK_AVAILABLE = find_spec("rocketpy") is not None
+
+if _STACK_AVAILABLE:
+    from BalloonPoppingGymEnv.envs.balloon_world import (
+        BalloonPoppingEnv,
+        check_action,
+    )
+    from BalloonPoppingGymEnv.evaluation.evaluate import load_scenario_parameters
+
+SCENARIO = 0
+LAUNCH_STEP = 1
+STEPS_AFTER_LAUNCH = 8
+
+
+_CLEAN = {
+    "launch_inclination_heading": np.array([80.0, 0.0]),
+    "tvc": np.array([0.0, 0.0]),
+    "roll": np.array([0.0]),
+    "throttle": np.array([1.0]),
+}
+
+
+class TestWhichFieldsAreUnusable(unittest.TestCase):
+    """The reader, tested on its own because everything below trusts it."""
+
+    def test_a_clean_action_has_none(self):
+        self.assertEqual(check_action(dict(_CLEAN)), [])
+
+    def test_each_field_is_found_by_name(self):
+        for field, shape in (
+            ("launch_inclination_heading", 2),
+            ("tvc", 2),
+            ("roll", 1),
+            ("throttle", 1),
+        ):
+            with self.subTest(field=field):
+                action = dict(_CLEAN, **{field: np.full(shape, np.nan)})
+
+                self.assertEqual(check_action(action), [field])
+
+    def test_infinity_counts_too(self):
+        """`np.clip` would carry an infinity straight through, and the actuator
+        would store it, so it is not only NaN that has to be caught."""
+        action = dict(_CLEAN, tvc=np.array([np.inf, 0.0]))
+
+        self.assertEqual(check_action(action), ["tvc"])
+
+    def test_a_field_that_is_not_a_number_at_all(self):
+        """`np.asarray(..., dtype=float)` raises on these rather than giving a
+        NaN, so they need naming separately or they reach the actuator."""
+        for value in ("nope", None, {"x": 1}):
+            with self.subTest(value=repr(value)):
+                self.assertIn("roll", check_action(dict(_CLEAN, roll=value)))
+
+    def test_a_missing_field_is_not_usable(self):
+        """Every shipped agent returns all four, and `step()` indexes them, so a
+        partial action passing the helper and failing in the environment is the
+        split this exists to remove."""
+        self.assertEqual(
+            check_action({"roll": np.array([0.0])}),
+            ["launch_inclination_heading", "throttle", "tvc"],
+        )
+
+    def test_the_wrong_number_of_values_is_not_usable(self):
+        """An empty array passes a finiteness test on its own, and `step()`
+        indexes `tvc[1]`."""
+        for field, value in (
+            ("tvc", np.array([])),
+            ("tvc", np.zeros(3)),
+            ("tvc", 0.0),
+            ("launch_inclination_heading", np.array([80.0])),
+        ):
+            with self.subTest(field=field, size=np.size(value)):
+                self.assertIn(field, check_action(dict(_CLEAN, **{field: value})))
+
+
+@unittest.skipUnless(_STACK_AVAILABLE, "simulation stack not installed")
+class TestAnActionTheEnvironmentCannotUse(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.parameters, _given = load_scenario_parameters(SCENARIO)
+
+    def _fresh(self):
+        env = BalloonPoppingEnv(render_mode=None, parameters=self.parameters)
+        env.reset(seed=self.parameters["scenario"]["random_seed"])
+        return env
+
+    @staticmethod
+    def _action(env, launch=False):
+        action = env.action_space.sample()
+        action["launch"] = np.array(int(launch), dtype=action["launch"].dtype)
+        action["launch_inclination_heading"] = np.array([80.0, 0.0], dtype=np.float64)
+        action["throttle"] = np.ones_like(action["throttle"])
+        action["tvc"] = np.zeros_like(action["tvc"])
+        action["roll"] = np.zeros_like(action["roll"])
+        return action
+
+    def _flown(self, env, steps, corrupt=None):
+        """Launch, then step, optionally corrupting one field throughout."""
+        for index in range(LAUNCH_STEP + steps):
+            action = self._action(env, launch=index == LAUNCH_STEP - 1)
+            if corrupt and index >= LAUNCH_STEP:
+                action[corrupt] = np.full_like(
+                    np.asarray(action[corrupt], dtype=float), np.nan
+                )
+            env.step(action)
+        return env
+
+    def test_a_control_field_of_nans_does_not_end_the_run(self):
+        """The case ActiveRocketPy #19 turns into a ValueError without this."""
+        for field in ("tvc", "roll", "throttle"):
+            with self.subTest(field=field):
+                env = self._flown(self._fresh(), STEPS_AFTER_LAUNCH, corrupt=field)
+
+                self.assertTrue(env.rocket_launched)
+                self.assertEqual(
+                    len(env.trajectories), LAUNCH_STEP + STEPS_AFTER_LAUNCH
+                )
+
+    def test_the_actuator_keeps_the_value_it_had(self):
+        """Dropping the command has to mean keeping the last one, not zeroing
+        it, or a single NaN would silently cut the throttle."""
+        env = self._fresh()
+        for index in range(4):
+            env.step(self._action(env, launch=index == LAUNCH_STEP - 1))
+        held = float(env._rocket_flight.rocket.throttle_control.throttle)
+
+        action = self._action(env)
+        action["throttle"] = np.full_like(
+            np.asarray(action["throttle"], dtype=float), np.nan
+        )
+        env.step(action)
+
+        self.assertEqual(
+            float(env._rocket_flight.rocket.throttle_control.throttle), held
+        )
+
+    def test_a_launch_attitude_of_nans_does_not_launch(self):
+        """There is no previous attitude to fall back on, so the launch is
+        refused rather than dropped. Measured without this: the flight is built
+        from a NaN quaternion and raises out of the integrator."""
+        env = self._fresh()
+        action = self._action(env, launch=True)
+        action["launch_inclination_heading"] = np.array([np.nan, 0.0])
+
+        env.step(action)
+
+        self.assertFalse(env.rocket_launched)
+
+    def test_the_agent_can_launch_on_a_later_step(self):
+        """The half that stops the refusal above being a way to lose the run."""
+        env = self._fresh()
+        broken = self._action(env, launch=True)
+        broken["launch_inclination_heading"] = np.array([np.nan, 0.0])
+        env.step(broken)
+
+        env.step(self._action(env, launch=True))
+
+        self.assertTrue(env.rocket_launched)
+
+    def test_the_dropped_field_is_named_in_the_log(self):
+        """A competitor reading the output has to be able to tell that a command
+        went nowhere, and which one."""
+        env = self._fresh()
+        for index in range(3):
+            env.step(self._action(env, launch=index == LAUNCH_STEP - 1))
+        action = self._action(env)
+        action["tvc"] = np.full_like(np.asarray(action["tvc"], dtype=float), np.nan)
+
+        with self.assertLogs(
+            "BalloonPoppingGymEnv.envs.balloon_world", logging.WARNING
+        ) as caught:
+            env.step(action)
+
+        self.assertIn("tvc", "".join(caught.output))
+
+    def test_a_clean_run_logs_nothing_and_scores_the_same(self):
+        """The half that stops all of this being satisfied by dropping
+        everything."""
+        env = self._fresh()
+        logger = logging.getLogger("BalloonPoppingGymEnv.envs.balloon_world")
+        with self.assertNoLogs(logger, logging.WARNING):
+            for index in range(LAUNCH_STEP + STEPS_AFTER_LAUNCH):
+                env.step(self._action(env, launch=index == LAUNCH_STEP - 1))
+
+        self.assertTrue(env.rocket_launched)
+
+
+if __name__ == "__main__":
+    unittest.main()
