[#337] Allow continuous aliveness checkers

docs/source/release_notes.rst

@@ -63,6 +63,10 @@ Development - |version|
 * Update files for compatibility with Ruff 0.16.0's new default lint rules, excluding C408.
 * Update ``np2EigenMatrix3d`` and ``np2EigenVectorXd`` import path to prevent deprecation warning while preserving compatibility with older versions of Basilisk.
 * Add support for configurable RSO attitude dynamics and new relative attitude observations, enabling inspection of spinning and tumbling targets. Update the `RSO Inspection <examples/rso_inspection.ipynb>`_ example to demonstrate non-nadir-pointing targets. See issue #355.
+* Allow :func:`~bsk_rl.utils.functional.aliveness_checker` to run as a continuous Basilisk
+  event that interrupts the environment step on failure. Battery charge and reaction
+  wheel speed checks use this behavior by default, at the simulator rate. See issues
+  #337 and #59.
 
 Version 1.3.0
 -------------

src/bsk_rl/sats/satellite.py

@@ -19,9 +19,12 @@
 from bsk_rl.utils.functional import (
     AbstractClassProperty,
     Resetable,
+    _log_aliveness_failure,
     collect_default_args,
     compose_types,
+    is_property,
     safe_dict_merge,
+    valid_func_name,
 )
 from bsk_rl.utils.orbital import TrajectorySimulator
 
@@ -275,6 +278,46 @@ def set_fsw(self, fsw_rate: float) -> "fsw.FSWModelABC":
         self.fsw = proxy(fsw)
         return fsw
 
+    def setup_aliveness_events(self, log_failure: bool = True) -> None:
+        """Create simulator events for continuous aliveness checkers.
+
+        :meta private:
+        """
+        for model in (self.dynamics, self.fsw):
+            for name in dir(model):
+                if (
+                    not name.startswith("__")
+                    and not is_property(model, name)
+                    and callable(getattr(model, name))
+                    and hasattr(getattr(model, name), "is_aliveness_checker")
+                    and getattr(getattr(model, name), "continuous", False)
+                ):
+                    checker = getattr(model, name)
+                    check_rate = checker.check_rate
+                    if check_rate is None:
+                        check_rate = self.simulator.sim_rate
+
+                    def condition(sim, checker=checker):
+                        return self._is_alive and not checker()
+
+                    def side_effect(sim, checker_name=checker.__name__):
+                        if log_failure:
+                            _log_aliveness_failure(self, checker_name)
+                        self._is_alive = False
+                        self.record_death(sim.sim_time)
+
+                    self.simulator.createNewEvent(
+                        valid_func_name(
+                            f"aliveness_{self.name}_{model.__class__.__name__}_{name}"
+                        ),
+                        macros.sec2nano(check_rate),
+                        True,
+                        conditionFunction=condition,
+                        actionFunction=side_effect,
+                        terminal=True,
+                        exactRateMatch=False,
+                    )
+
     def reset_during_sim_init(self) -> None:
         """Called during environment reset, during Basilisk simulation initialization."""
         self.observation_builder.reset_during_sim_init()
@@ -285,6 +328,7 @@ def reset_post_sim_init(self) -> None:
         """Called during environment reset, after Basilisk simulation initialization."""
         self.observation_builder.reset_post_sim_init()
         self.action_builder.reset_post_sim_init()
+        self.setup_aliveness_events()
 
     @property
     def observation_space(self) -> spaces.Space:

src/bsk_rl/sim/dyn/__init__.py

@@ -31,8 +31,13 @@
 ------------------
 
 Certain functions in the dynamics model are decorated with the :func:`~bsk_rl.utils.functional.aliveness_checker`
-decorator. These functions are called at each step to check if the satellite is still
-operational, returning true if the satellite is still alive.
+decorator. These functions are called at each environment step to check if the satellite
+is still operational, returning true if the satellite is still alive. Passing
+``continuous=True`` to the decorator also evaluates the checker during integration and
+stops the environment step if it fails.
+
+Battery charge and reaction wheel speed checks use continuous checking by default,
+at the simulator rate.
 
 """
 

src/bsk_rl/sim/dyn/base.py

@@ -623,9 +623,9 @@ def setup_reaction_wheel_dyn_effector(
             self.task_name, self.rwStateEffector, ModelPriority=priority
         )
 
-    @aliveness_checker
+    @aliveness_checker(continuous=True)
     def rw_speeds_valid(self) -> bool:
-        """Check if any wheel speed exceeds the ``maxWheelSpeed``."""
+        """Check wheel speeds against ``maxWheelSpeed`` during integration."""
         valid = all(
             abs(speed) < self.maxWheelSpeed * macros.rpm2radsec
             for speed in self.wheel_speeds
@@ -742,13 +742,12 @@ def setup_battery(
             self.task_name, self.powerMonitor, ModelPriority=priority
         )
 
-    @aliveness_checker
+    @aliveness_checker(continuous=True)
     def battery_valid(self) -> bool:
         """Check if the battery has charge remaining.
 
-        Note that this check is instantaneous. If a satellite runs out of power during a
-        environment step but then recharges to have positive power at the end of the step,
-        the satellite will still be considered alive.
+        Evaluated during integration so battery depletion interrupts the environment
+        step, even if the satellite would recharge later in the step.
         """
         return self.battery_charge > 0
 

src/bsk_rl/sim/fsw/__init__.py

@@ -25,8 +25,10 @@
 ------------------
 
 Certain functions in the FSW models are decorated with the :func:`~bsk_rl.utils.functional.aliveness_checker`
-decorator. These functions are called at each step to check if the satellite is still
-operational, returning true if the satellite is still alive.
+decorator. These functions are called at each environment step to check if the satellite
+is still operational, returning true if the satellite is still alive. Passing
+``continuous=True`` to the decorator also evaluates the checker during integration and
+stops the environment step if it fails.
 """
 
 from bsk_rl.sim.fsw.base import (

src/bsk_rl/utils/functional.py

@@ -5,7 +5,7 @@
 from copy import deepcopy
 from functools import wraps
 from types import new_class
-from typing import Any, Callable, ParamSpec, TypeVar
+from typing import Any, Callable, Optional, ParamSpec, TypeVar
 from unittest.mock import MagicMock
 
 import numpy as np
@@ -115,23 +115,47 @@ def vectorize_nested_dict(dictionary: dict) -> tuple[list[str], np.ndarray]:
     return list(np.concatenate(keys)), np.concatenate(values)
 
 
-def aliveness_checker(func: Callable[..., bool]) -> Callable[..., bool]:
-    """Decorate function to evaluate when checking for satellite aliveness."""
+def _log_aliveness_failure(satellite: Any, checker_name: str) -> None:
+    """Log a failed aliveness check without evaluating it again."""
+    satellite.logger.warning(f"failed {checker_name} check")
 
-    @wraps(func)
-    def inner(*args, log_failure=False, **kwargs) -> bool:
-        self = args[0]
-        alive = func(*args, **kwargs)
-        if not alive and log_failure:
-            self.satellite.logger.warning(f"failed {func.__name__} check")
-        return alive
 
-    inner.__doc__ = (
-        "*Decorated with* :class:`~bsk_rl.utils.functional.aliveness_checker`\n\n"
-        + str(func.__doc__)
-    )
-    inner.is_aliveness_checker = True
-    return inner
+def aliveness_checker(
+    func: Optional[Callable[..., bool]] = None,
+    *,
+    continuous: bool = False,
+    check_rate: Optional[float] = None,
+) -> Callable:
+    """Decorate function to evaluate when checking for satellite aliveness.
+
+    Checkers are evaluated at environment steps. Set ``continuous=True`` to also
+    evaluate the checker during integration via a Basilisk event that terminates
+    the environment step on failure. ``check_rate`` (seconds) sets the minimum
+    interval between evaluations and defaults to the simulator rate. Checks run
+    at simulation timesteps, so the actual interval may be longer.
+    """
+
+    def decorator(func: Callable[..., bool]) -> Callable[..., bool]:
+        @wraps(func)
+        def inner(*args, log_failure=False, **kwargs) -> bool:
+            self = args[0]
+            alive = func(*args, **kwargs)
+            if not alive and log_failure:
+                _log_aliveness_failure(self.satellite, func.__name__)
+            return alive
+
+        inner.__doc__ = (
+            "*Decorated with* :class:`~bsk_rl.utils.functional.aliveness_checker`\n\n"
+            + str(func.__doc__)
+        )
+        inner.is_aliveness_checker = True
+        inner.continuous = continuous
+        inner.check_rate = check_rate
+        return inner
+
+    if func is None:
+        return decorator
+    return decorator(func)
 
 
 def check_aliveness_checkers(model: Any, log_failure=False) -> bool:
@@ -158,7 +182,7 @@ def check_aliveness_checkers(model: Any, log_failure=False) -> bool:
 
 def is_property(obj: Any, attr_name: str) -> bool:
     """Check if obj has an ``@property`` called ``attr_name`` without calling it."""
-    cls = type(obj)
+    cls = obj.__class__
     attribute = getattr(cls, attr_name, None)
     return attribute is not None and isinstance(attribute, property)
 

tests/integration/sim/test_int_dynamics.py

@@ -2,9 +2,11 @@
 
 import gymnasium as gym
 import pytest
+from Basilisk.utilities import macros
 
 from bsk_rl import act, data, obs, sats, scene
 from bsk_rl.sim import dyn, fsw
+from bsk_rl.utils.functional import aliveness_checker
 from bsk_rl.utils.orbital import random_orbit
 
 
@@ -240,3 +242,109 @@ class DeputySat(sats.Satellite):
             assert sat2.dynamics.out_of_ranges == []
             assert not terminated["Chief"]
             assert not terminated["Deputy"]
+
+
+class TestContinuousAlivenessChecker:
+    @pytest.mark.parametrize(
+        "continuous,check_rate,expected_time",
+        [
+            (False, None, 100.0),
+            (True, None, 5.0),
+            (True, 0.5, 5.0),
+            (True, 2.0, 6.0),
+            (True, 2.5, 6.0),
+            (True, 3.0, 6.0),
+        ],
+    )
+    def test_step_vs_continuous(
+        self, continuous, check_rate, expected_time, caplog
+    ):
+        class BoomDyn(dyn.DynamicsModel):
+            @aliveness_checker(continuous=continuous, check_rate=check_rate)
+            def time_valid(self):
+                return self.simulator.sim_time < 5.0
+
+        class BoomSat(sats.Satellite):
+            fsw_type = fsw.FSWModel
+            dyn_type = BoomDyn
+            observation_spec: ClassVar[list[obs.Observation]] = [obs.Time()]
+            action_spec: ClassVar[list[act.Action]] = [act.Drift()]
+
+        env = gym.make(
+            "SatelliteTasking-v1",
+            satellite=BoomSat("Boom", sat_args=dict(oe=random_orbit)),
+            sim_rate=1.0,
+            time_limit=100.0,
+            max_step_duration=100.0,
+            disable_env_checker=True,
+        )
+
+        env.reset(seed=0)
+        _, _, terminated, _, _ = env.step(0)
+
+        assert env.unwrapped.simulator.sim_time == pytest.approx(expected_time)
+        assert not env.unwrapped.satellite.is_alive()
+        assert env.unwrapped.satellite.time_of_death == pytest.approx(expected_time)
+        assert terminated
+        assert caplog.messages.count("failed time_valid check") == 1
+        env.close()
+
+    @pytest.mark.parametrize("checker_name", ["battery_valid", "rw_speeds_valid"])
+    @pytest.mark.parametrize("continuous", [False, True])
+    def test_transient_failure(self, checker_name, continuous, caplog):
+        class TransientFailureDyn(dyn.BasicDynamicsModel):
+            @property
+            def battery_charge(self):
+                if (
+                    checker_name == "battery_valid"
+                    and 5.0 <= self.simulator.sim_time < 10.0
+                ):
+                    return 0.0
+                return super().battery_charge
+
+            @property
+            def wheel_speeds(self):
+                if (
+                    checker_name == "rw_speeds_valid"
+                    and 5.0 <= self.simulator.sim_time < 10.0
+                ):
+                    return [1.1 * self.maxWheelSpeed * macros.rpm2radsec] * 3
+                return super().wheel_speeds
+
+        if not continuous:
+            checker = getattr(TransientFailureDyn, checker_name)
+            setattr(
+                TransientFailureDyn,
+                checker_name,
+                aliveness_checker(checker.__wrapped__),
+            )
+
+        class FailureSat(sats.Satellite):
+            dyn_type = TransientFailureDyn
+            fsw_type = fsw.BasicFSWModel
+            observation_spec: ClassVar[list[obs.Observation]] = [obs.Time()]
+            action_spec: ClassVar[list[act.Action]] = [act.Drift()]
+
+        env = gym.make(
+            "SatelliteTasking-v1",
+            satellite=FailureSat("Failure", sat_args=dict(oe=random_orbit)),
+            sim_rate=1.0,
+            time_limit=100.0,
+            max_step_duration=100.0,
+            disable_env_checker=True,
+        )
+
+        env.reset(seed=0)
+        _, _, terminated, _, _ = env.step(0)
+
+        expected_time = 5.0 if continuous else 100.0
+        assert env.unwrapped.simulator.sim_time == pytest.approx(expected_time)
+        assert terminated is continuous
+        assert env.unwrapped.satellite.is_alive() is not continuous
+        if continuous:
+            assert env.unwrapped.satellite.time_of_death == pytest.approx(5.0)
+            assert caplog.messages.count(f"failed {checker_name} check") == 1
+        else:
+            assert env.unwrapped.satellite.time_of_death is None
+            assert f"failed {checker_name} check" not in caplog.messages
+        env.close()

tests/integration/sim/test_int_fsw.py

@@ -1 +1,55 @@
-# For fsw models not tested in other tests
+from typing import ClassVar
+
+import gymnasium as gym
+import pytest
+
+from bsk_rl import act, obs, sats
+from bsk_rl.sim import dyn, fsw
+from bsk_rl.utils.functional import aliveness_checker
+from bsk_rl.utils.orbital import random_orbit
+
+
+class TestContinuousAlivenessChecker:
+    @pytest.mark.parametrize(
+        "continuous,check_rate,expected_time",
+        [
+            (False, None, 100.0),
+            (True, None, 5.0),
+            (True, 0.5, 5.0),
+            (True, 2.0, 6.0),
+            (True, 2.5, 6.0),
+            (True, 3.0, 6.0),
+        ],
+    )
+    def test_step_vs_continuous(
+        self, continuous, check_rate, expected_time, caplog
+    ):
+        class BoomFSW(fsw.FSWModel):
+            @aliveness_checker(continuous=continuous, check_rate=check_rate)
+            def time_valid(self):
+                return self.simulator.sim_time < 5.0
+
+        class BoomSat(sats.Satellite):
+            fsw_type = BoomFSW
+            dyn_type = dyn.DynamicsModel
+            observation_spec: ClassVar[list[obs.Observation]] = [obs.Time()]
+            action_spec: ClassVar[list[act.Action]] = [act.Drift()]
+
+        env = gym.make(
+            "SatelliteTasking-v1",
+            satellite=BoomSat("Boom", sat_args=dict(oe=random_orbit)),
+            sim_rate=1.0,
+            time_limit=100.0,
+            max_step_duration=100.0,
+            disable_env_checker=True,
+        )
+
+        env.reset(seed=0)
+        _, _, terminated, _, _ = env.step(0)
+
+        assert env.unwrapped.simulator.sim_time == pytest.approx(expected_time)
+        assert not env.unwrapped.satellite.is_alive()
+        assert env.unwrapped.satellite.time_of_death == pytest.approx(expected_time)
+        assert terminated
+        assert caplog.messages.count("failed time_valid check") == 1
+        env.close()