From ca17a91ef250ce16f76e99a1ebd65317baa0da1a Mon Sep 17 00:00:00 2001 From: Daniel Huterer Prats Date: Tue, 3 Nov 2026 17:46:15 -0700 Subject: [PATCH 1/5] Issue #370: Stop downlinks when storage empties --- docs/source/release_notes.rst | 4 ++ src/bsk_rl/act/discrete_actions.py | 95 +++++++++++++++++++++++++++++- 2 files changed, 98 insertions(+), 1 deletion(-) diff --git a/docs/source/release_notes.rst b/docs/source/release_notes.rst index af8aa696..c1b8e7e4 100644 --- a/docs/source/release_notes.rst +++ b/docs/source/release_notes.rst @@ -5,6 +5,10 @@ Development - |version| ----------------------- .. *Release Date: MMM. DD, YYYY* +* Add opt-in buffer-conditioned termination to :class:`~bsk_rl.act.Downlink` with + ``variable_duration_downlink=True``. Downlink actions stop once storage is empty, + subject to a configurable minimum duration that defaults to 10 seconds. Existing + fixed-duration behavior remains the default. See issue #370. * Add optional hybrid action spaces with ``hybrid=True`` on supported discrete actions. Mode, broadcast, and ground-imaging actions accept a requested duration. Ground imaging instead uses a desired capture-time offset from the current diff --git a/src/bsk_rl/act/discrete_actions.py b/src/bsk_rl/act/discrete_actions.py index 66a87c94..44cc6adc 100644 --- a/src/bsk_rl/act/discrete_actions.py +++ b/src/bsk_rl/act/discrete_actions.py @@ -9,6 +9,7 @@ from gymnasium import spaces from bsk_rl.act.actions import Action, ActionBuilder +from bsk_rl.utils.functional import valid_func_name if TYPE_CHECKING: # pragma: no cover from bsk_rl.sats import Satellite @@ -67,6 +68,9 @@ def __init__(self, satellite: "Satellite") -> None: def reset_post_sim_init(self) -> None: """Clear the previous action key after simulator initialization.""" + active_downlink_action = getattr(self.satellite, "_active_downlink_action", None) + if active_downlink_action is not None: + active_downlink_action._disable_downlink_empty_event() super().reset_post_sim_init() self.prev_action_key = None @@ -100,6 +104,9 @@ def set_action(self, action: Union[int, tuple[int, float]]) -> None: Hybrid actions receive ``(local_index, parameter)``. Other actions receive only their local index, even when the overall action space is hybrid. """ + active_downlink_action = getattr(self.satellite, "_active_downlink_action", None) + if active_downlink_action is not None: + active_downlink_action._disable_downlink_empty_event() action_index, parameter = clean_action(action, hybrid=self.hybrid) self.satellite.disable_timed_terminal_event() if not np.issubdtype(type(action_index), np.integer): @@ -326,6 +333,9 @@ def __init__( name: Optional[str] = None, duration: Optional[float] = None, hybrid: bool = False, + variable_duration_downlink: bool = False, + empty_storage_threshold_bits: float = 1.0, + minimum_duration: float = 10.0, ): """Action to transmit data from the data buffer (:class:`~bsk_rl.sim.fsw.ImagingFSWModel.action_downlink`). @@ -335,12 +345,95 @@ def __init__( Args: name: Action name. - duration: Time to task action, in seconds. + duration: Maximum time to task action, in seconds. hybrid: Whether to use a hybrid action space. + variable_duration_downlink: If true, stop early once storage is empty. + empty_storage_threshold_bits: Storage level considered empty, in bits. + minimum_duration: Minimum downlink time in seconds, including when + the buffer is empty at action start. """ super().__init__( fsw_action="action_downlink", name=name, duration=duration, hybrid=hybrid ) + if not np.isfinite(empty_storage_threshold_bits) or empty_storage_threshold_bits < 0: + raise ValueError("empty_storage_threshold_bits must be finite and non-negative.") + if not np.isfinite(minimum_duration) or minimum_duration < 0: + raise ValueError("minimum_duration must be finite and non-negative.") + self.variable_duration_downlink = bool(variable_duration_downlink) + self.empty_storage_threshold_bits = float(empty_storage_threshold_bits) + self.minimum_duration = float(minimum_duration) + self._downlink_empty_event_name = None + self._downlink_earliest_stop_time = None + + def _storage_is_empty(self) -> bool: + """Return whether the satellite data buffer is effectively empty.""" + return float(self.satellite.dynamics.storage_level) <= self.empty_storage_threshold_bits + + def _disable_downlink_empty_event(self) -> None: + """Delete the active empty-buffer event, if one exists.""" + if ( + self._downlink_empty_event_name is not None + and self._downlink_empty_event_name in self.simulator.eventMap + ): + self.simulator.delete_event(self._downlink_empty_event_name) + self._downlink_empty_event_name = None + self._downlink_earliest_stop_time = None + if getattr(self.satellite, "_active_downlink_action", None) is self: + self.satellite._active_downlink_action = None + + def _downlink_empty_event_ready(self) -> bool: + """Wait for the configured minimum duration and an empty buffer.""" + earliest_stop = self._downlink_earliest_stop_time + return ( + earliest_stop is not None + and float(self.simulator.sim_time) >= earliest_stop + and self._storage_is_empty() + ) + + def _finish_downlink_early(self) -> None: + """Request a new action and keep event state available for cleanup.""" + self.satellite.requires_retasking = True + self._downlink_earliest_stop_time = None + + def _enable_downlink_empty_event(self) -> None: + """Stop the action when its storage buffer is empty.""" + self._disable_downlink_empty_event() + self.satellite._active_downlink_action = self + # Only require the guard when there is no data to transmit at action start. + # If a nonempty buffer drains, end as soon as the next FSW event observes it. + minimum_step = ( + max(self.minimum_duration, float(self.simulator.sim_rate)) + if self._storage_is_empty() + else 0.0 + ) + self._downlink_earliest_stop_time = float(self.simulator.sim_time) + minimum_step + self._downlink_empty_event_name = valid_func_name( + f"downlink_empty_{self.satellite.name}" + ) + satellite_command = self.satellite._satellite_command + self.simulator.createNewEvent( + self._downlink_empty_event_name, + mc.sec2nano(self.satellite.fsw.fsw_rate), + True, + [ + f"{satellite_command}._active_downlink_action is not None and " + f"{satellite_command}._active_downlink_action._downlink_empty_event_ready()" + ], + [ + self.satellite._info_command("downlink storage empty"), + f"{satellite_command}._active_downlink_action._finish_downlink_early()", + ], + terminal=self.satellite.variable_interval, + ) + + def set_action( + self, action: Union[int, tuple[int, float]], prev_action_key=None + ) -> str: + """Activate downlink and optionally stop when storage is empty.""" + action_key = super().set_action(action, prev_action_key=prev_action_key) + if self.variable_duration_downlink: + self._enable_downlink_empty_event() + return action_key class Scan(DiscreteFSWAction): From 38f4cf23e1797c3b79e8add8f64b7926e7734bc1 Mon Sep 17 00:00:00 2001 From: Daniel Huterer Prats Date: Tue, 3 Nov 2026 18:07:19 -0700 Subject: [PATCH 2/5] Issue #370: Fix downlink event reset and cover behavior --- src/bsk_rl/act/discrete_actions.py | 20 +++++++--- tests/unittest/act/test_actions.py | 62 ++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+), 5 deletions(-) diff --git a/src/bsk_rl/act/discrete_actions.py b/src/bsk_rl/act/discrete_actions.py index 44cc6adc..69e646ae 100644 --- a/src/bsk_rl/act/discrete_actions.py +++ b/src/bsk_rl/act/discrete_actions.py @@ -68,10 +68,10 @@ def __init__(self, satellite: "Satellite") -> None: def reset_post_sim_init(self) -> None: """Clear the previous action key after simulator initialization.""" + super().reset_post_sim_init() active_downlink_action = getattr(self.satellite, "_active_downlink_action", None) if active_downlink_action is not None: active_downlink_action._disable_downlink_empty_event() - super().reset_post_sim_init() self.prev_action_key = None @property @@ -355,8 +355,13 @@ def __init__( super().__init__( fsw_action="action_downlink", name=name, duration=duration, hybrid=hybrid ) - if not np.isfinite(empty_storage_threshold_bits) or empty_storage_threshold_bits < 0: - raise ValueError("empty_storage_threshold_bits must be finite and non-negative.") + if ( + not np.isfinite(empty_storage_threshold_bits) + or empty_storage_threshold_bits < 0 + ): + raise ValueError( + "empty_storage_threshold_bits must be finite and non-negative." + ) if not np.isfinite(minimum_duration) or minimum_duration < 0: raise ValueError("minimum_duration must be finite and non-negative.") self.variable_duration_downlink = bool(variable_duration_downlink) @@ -367,7 +372,10 @@ def __init__( def _storage_is_empty(self) -> bool: """Return whether the satellite data buffer is effectively empty.""" - return float(self.satellite.dynamics.storage_level) <= self.empty_storage_threshold_bits + return ( + float(self.satellite.dynamics.storage_level) + <= self.empty_storage_threshold_bits + ) def _disable_downlink_empty_event(self) -> None: """Delete the active empty-buffer event, if one exists.""" @@ -406,7 +414,9 @@ def _enable_downlink_empty_event(self) -> None: if self._storage_is_empty() else 0.0 ) - self._downlink_earliest_stop_time = float(self.simulator.sim_time) + minimum_step + self._downlink_earliest_stop_time = ( + float(self.simulator.sim_time) + minimum_step + ) self._downlink_empty_event_name = valid_func_name( f"downlink_empty_{self.satellite.name}" ) diff --git a/tests/unittest/act/test_actions.py b/tests/unittest/act/test_actions.py index a37fbb67..1906915c 100644 --- a/tests/unittest/act/test_actions.py +++ b/tests/unittest/act/test_actions.py @@ -133,6 +133,68 @@ def test_set_action_reset(self): fswact.satellite.fsw.action_fsw.assert_called_once() +class TestVariableDurationDownlink: + @staticmethod + def make_downlink(storage_level, *, minimum_duration=10.0): + downlink = act.Downlink( + variable_duration_downlink=True, + minimum_duration=minimum_duration, + empty_storage_threshold_bits=1.0, + ) + downlink.satellite = MagicMock() + downlink.satellite.name = "sat" + downlink.satellite.dynamics.storage_level = storage_level + downlink.satellite.fsw.fsw_rate = 2.0 + downlink.satellite.variable_interval = True + downlink.satellite._satellite_command = "satellite" + downlink.satellite._info_command.return_value = "log_empty_buffer" + downlink.simulator = SimpleNamespace( + sim_rate=1.0, + sim_time=20.0, + eventMap={}, + createNewEvent=MagicMock(), + delete_event=MagicMock(), + ) + return downlink + + def test_empty_at_start_waits_for_minimum_duration(self): + downlink = self.make_downlink(storage_level=0.0) + + downlink._enable_downlink_empty_event() + + assert downlink._downlink_earliest_stop_time == 30.0 + assert not downlink._downlink_empty_event_ready() + downlink.simulator.sim_time = 30.0 + assert downlink._downlink_empty_event_ready() + + def test_nonempty_buffer_stops_as_soon_as_it_drains(self): + downlink = self.make_downlink(storage_level=5.0) + + downlink._enable_downlink_empty_event() + + assert downlink._downlink_earliest_stop_time == 20.0 + assert not downlink._downlink_empty_event_ready() + downlink.satellite.dynamics.storage_level = 0.5 + assert downlink._downlink_empty_event_ready() + + def test_finishing_and_disabling_event_cleans_up_state(self): + downlink = self.make_downlink(storage_level=0.0) + downlink._enable_downlink_empty_event() + event_name = downlink._downlink_empty_event_name + downlink.simulator.eventMap[event_name] = object() + + downlink._finish_downlink_early() + assert downlink.satellite.requires_retasking + assert not downlink._downlink_empty_event_ready() + downlink._disable_downlink_empty_event() + + downlink.simulator.delete_event.assert_called_once_with(event_name) + assert downlink.satellite._active_downlink_action is None + + def test_behavior_remains_opt_in(self): + assert act.Downlink().variable_duration_downlink is False + + @pytest.mark.parametrize("kind", ["fsw", "broadcast"]) @pytest.mark.parametrize( "requested,cap,sim_rate,expected_duration", From 9abde9f8036842a404a3082ab2ec41d65e76048f Mon Sep 17 00:00:00 2001 From: Daniel Huterer Prats Date: Tue, 3 Nov 2026 18:08:53 -0700 Subject: [PATCH 3/5] Issue #370: Clarify downlink duration precedence --- docs/source/release_notes.rst | 5 +++-- src/bsk_rl/act/discrete_actions.py | 3 ++- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/docs/source/release_notes.rst b/docs/source/release_notes.rst index c1b8e7e4..94d8ff57 100644 --- a/docs/source/release_notes.rst +++ b/docs/source/release_notes.rst @@ -7,8 +7,9 @@ Development - |version| * Add opt-in buffer-conditioned termination to :class:`~bsk_rl.act.Downlink` with ``variable_duration_downlink=True``. Downlink actions stop once storage is empty, - subject to a configurable minimum duration that defaults to 10 seconds. Existing - fixed-duration behavior remains the default. See issue #370. + subject to a configurable 10-second minimum when the buffer is empty at action + start. A shorter configured action duration remains the cap. Existing fixed-duration + behavior remains the default. See issue #370. * Add optional hybrid action spaces with ``hybrid=True`` on supported discrete actions. Mode, broadcast, and ground-imaging actions accept a requested duration. Ground imaging instead uses a desired capture-time offset from the current diff --git a/src/bsk_rl/act/discrete_actions.py b/src/bsk_rl/act/discrete_actions.py index 69e646ae..227f1218 100644 --- a/src/bsk_rl/act/discrete_actions.py +++ b/src/bsk_rl/act/discrete_actions.py @@ -350,7 +350,8 @@ def __init__( variable_duration_downlink: If true, stop early once storage is empty. empty_storage_threshold_bits: Storage level considered empty, in bits. minimum_duration: Minimum downlink time in seconds, including when - the buffer is empty at action start. + the buffer is empty at action start. The configured action + duration remains a hard maximum if it is shorter. """ super().__init__( fsw_action="action_downlink", name=name, duration=duration, hybrid=hybrid From 990a8a156fb3d0b932185899a3b1ac146eebade7 Mon Sep 17 00:00:00 2001 From: Daniel Huterer Prats Date: Tue, 3 Nov 2026 18:55:21 -0700 Subject: [PATCH 4/5] Issue #370: Exercise empty-buffer downlink event --- .../act/test_int_actions_discrete.py | 39 +++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/tests/integration/act/test_int_actions_discrete.py b/tests/integration/act/test_int_actions_discrete.py index 11479cac..34d860d7 100644 --- a/tests/integration/act/test_int_actions_discrete.py +++ b/tests/integration/act/test_int_actions_discrete.py @@ -146,6 +146,45 @@ def test_image_by_name(self): self.env.step(target.id) +class TestVariableDurationDownlink: + class ImageSat(sats.ImagingSatellite): + dyn_type = dyn.GroundStationDynModel + fsw_type = fsw.ImagingFSWModel + observation_spec: ClassVar[list[obs.Observation]] = [obs.Time()] + action_spec: ClassVar[list[act.Action]] = [ + act.Downlink(variable_duration_downlink=True) + ] + + env = gym.make( + "SatelliteTasking-v1", + satellite=ImageSat( + "EO_empty_buffer", + sat_args=ImageSat.default_sat_args( + oe=random_orbit, + instrumentBaudRate=1.0, + dataStorageCapacity=3.0, + transmitterBaudRate=-1.0, + ), + ), + scenario=UniformTargets(n_targets=0), + rewarder=data.UniqueImageReward(), + sim_rate=1.0, + time_limit=100.0, + max_step_duration=100.0, + disable_env_checker=True, + ) + + def test_empty_buffer_action_waits_for_minimum_then_retasks(self): + self.env.reset(seed=0) + start_time = self.env.unwrapped.simulator.sim_time + + self.env.step(0) + + elapsed = self.env.unwrapped.simulator.sim_time - start_time + assert elapsed >= 10.0 + assert elapsed < 100.0 + + ########################### # Individual Action Tests # ########################### From 6679e1bdb4bcb32ef5607e1f94b20192bb8b80f3 Mon Sep 17 00:00:00 2001 From: Daniel Huterer Prats Date: Wed, 4 Nov 2026 11:57:02 -0700 Subject: [PATCH 5/5] Issue #370: Use action-owned cleanup and callable events --- src/bsk_rl/act/actions.py | 4 ++ src/bsk_rl/act/discrete_actions.py | 37 +++++++------------ .../act/test_int_actions_discrete.py | 22 ++++++++++- tests/unittest/act/test_actions.py | 6 +-- 4 files changed, 40 insertions(+), 29 deletions(-) diff --git a/src/bsk_rl/act/actions.py b/src/bsk_rl/act/actions.py index 29814b41..eeeaa64c 100644 --- a/src/bsk_rl/act/actions.py +++ b/src/bsk_rl/act/actions.py @@ -116,6 +116,10 @@ def link_simulator(self, simulator: "Simulator") -> None: """ self.simulator = simulator # already a proxy + def _clear_action_events(self) -> None: + """Remove simulator events that should expire on the next action selection.""" + pass + @abstractmethod def set_action(self, action: Any) -> None: # pragma: no cover """Execute code to perform an action.""" diff --git a/src/bsk_rl/act/discrete_actions.py b/src/bsk_rl/act/discrete_actions.py index 227f1218..0ca4b406 100644 --- a/src/bsk_rl/act/discrete_actions.py +++ b/src/bsk_rl/act/discrete_actions.py @@ -9,7 +9,6 @@ from gymnasium import spaces from bsk_rl.act.actions import Action, ActionBuilder -from bsk_rl.utils.functional import valid_func_name if TYPE_CHECKING: # pragma: no cover from bsk_rl.sats import Satellite @@ -69,9 +68,6 @@ def __init__(self, satellite: "Satellite") -> None: def reset_post_sim_init(self) -> None: """Clear the previous action key after simulator initialization.""" super().reset_post_sim_init() - active_downlink_action = getattr(self.satellite, "_active_downlink_action", None) - if active_downlink_action is not None: - active_downlink_action._disable_downlink_empty_event() self.prev_action_key = None @property @@ -104,10 +100,9 @@ def set_action(self, action: Union[int, tuple[int, float]]) -> None: Hybrid actions receive ``(local_index, parameter)``. Other actions receive only their local index, even when the overall action space is hybrid. """ - active_downlink_action = getattr(self.satellite, "_active_downlink_action", None) - if active_downlink_action is not None: - active_downlink_action._disable_downlink_empty_event() action_index, parameter = clean_action(action, hybrid=self.hybrid) + for act in self.action_spec: + act._clear_action_events() self.satellite.disable_timed_terminal_event() if not np.issubdtype(type(action_index), np.integer): logger.warning( @@ -378,7 +373,12 @@ def _storage_is_empty(self) -> bool: <= self.empty_storage_threshold_bits ) - def _disable_downlink_empty_event(self) -> None: + def reset_overwrite_previous(self) -> None: + """Forget event state from the previous episode.""" + self._downlink_empty_event_name = None + self._downlink_earliest_stop_time = None + + def _clear_action_events(self) -> None: """Delete the active empty-buffer event, if one exists.""" if ( self._downlink_empty_event_name is not None @@ -387,8 +387,6 @@ def _disable_downlink_empty_event(self) -> None: self.simulator.delete_event(self._downlink_empty_event_name) self._downlink_empty_event_name = None self._downlink_earliest_stop_time = None - if getattr(self.satellite, "_active_downlink_action", None) is self: - self.satellite._active_downlink_action = None def _downlink_empty_event_ready(self) -> bool: """Wait for the configured minimum duration and an empty buffer.""" @@ -401,13 +399,13 @@ def _downlink_empty_event_ready(self) -> bool: def _finish_downlink_early(self) -> None: """Request a new action and keep event state available for cleanup.""" + self.satellite.logger.info("Downlink storage empty") self.satellite.requires_retasking = True self._downlink_earliest_stop_time = None def _enable_downlink_empty_event(self) -> None: """Stop the action when its storage buffer is empty.""" - self._disable_downlink_empty_event() - self.satellite._active_downlink_action = self + self._clear_action_events() # Only require the guard when there is no data to transmit at action start. # If a nonempty buffer drains, end as soon as the next FSW event observes it. minimum_step = ( @@ -418,22 +416,13 @@ def _enable_downlink_empty_event(self) -> None: self._downlink_earliest_stop_time = ( float(self.simulator.sim_time) + minimum_step ) - self._downlink_empty_event_name = valid_func_name( - f"downlink_empty_{self.satellite.name}" - ) - satellite_command = self.satellite._satellite_command + self._downlink_empty_event_name = f"downlink_empty_{self.satellite.name}" self.simulator.createNewEvent( self._downlink_empty_event_name, mc.sec2nano(self.satellite.fsw.fsw_rate), True, - [ - f"{satellite_command}._active_downlink_action is not None and " - f"{satellite_command}._active_downlink_action._downlink_empty_event_ready()" - ], - [ - self.satellite._info_command("downlink storage empty"), - f"{satellite_command}._active_downlink_action._finish_downlink_early()", - ], + conditionFunction=lambda sim: self._downlink_empty_event_ready(), + actionFunction=lambda sim: self._finish_downlink_early(), terminal=self.satellite.variable_interval, ) diff --git a/tests/integration/act/test_int_actions_discrete.py b/tests/integration/act/test_int_actions_discrete.py index 34d860d7..f07d6350 100644 --- a/tests/integration/act/test_int_actions_discrete.py +++ b/tests/integration/act/test_int_actions_discrete.py @@ -152,7 +152,8 @@ class ImageSat(sats.ImagingSatellite): fsw_type = fsw.ImagingFSWModel observation_spec: ClassVar[list[obs.Observation]] = [obs.Time()] action_spec: ClassVar[list[act.Action]] = [ - act.Downlink(variable_duration_downlink=True) + act.Downlink(variable_duration_downlink=True), + act.Drift(duration=20.0), ] env = gym.make( @@ -185,6 +186,25 @@ def test_empty_buffer_action_waits_for_minimum_then_retasks(self): assert elapsed < 100.0 + def test_downlink_event_is_removed_on_retask_and_reset(self): + self.env.reset(seed=0) + simulator = self.env.unwrapped.simulator + event_name = "downlink_empty_EO_empty_buffer" + + self.env.step(0) + assert event_name in simulator.eventMap + self.env.step(1) + assert event_name not in simulator.eventMap + + self.env.step(0) + assert event_name in simulator.eventMap + del simulator + self.env.reset(seed=0) + assert event_name not in self.env.unwrapped.simulator.eventMap + self.env.step(0) + assert self.env.unwrapped.simulator.sim_time == approx(10.0) + + ########################### # Individual Action Tests # ########################### diff --git a/tests/unittest/act/test_actions.py b/tests/unittest/act/test_actions.py index 1906915c..091ff68f 100644 --- a/tests/unittest/act/test_actions.py +++ b/tests/unittest/act/test_actions.py @@ -146,8 +146,6 @@ def make_downlink(storage_level, *, minimum_duration=10.0): downlink.satellite.dynamics.storage_level = storage_level downlink.satellite.fsw.fsw_rate = 2.0 downlink.satellite.variable_interval = True - downlink.satellite._satellite_command = "satellite" - downlink.satellite._info_command.return_value = "log_empty_buffer" downlink.simulator = SimpleNamespace( sim_rate=1.0, sim_time=20.0, @@ -186,10 +184,10 @@ def test_finishing_and_disabling_event_cleans_up_state(self): downlink._finish_downlink_early() assert downlink.satellite.requires_retasking assert not downlink._downlink_empty_event_ready() - downlink._disable_downlink_empty_event() + downlink._clear_action_events() downlink.simulator.delete_event.assert_called_once_with(event_name) - assert downlink.satellite._active_downlink_action is None + assert downlink._downlink_empty_event_name is None def test_behavior_remains_opt_in(self): assert act.Downlink().variable_duration_downlink is False