Feature/allow reimaging with pending downlink

docs/source/release_notes.rst

@@ -5,13 +5,15 @@ Development - |version|
 -----------------------
 .. *Release Date: MMM. DD, YYYY*
 
+* Add ``block_pending_targets=False`` to :class:`~bsk_rl.scene.RSOTargets`
+  to allow reimaging before downlink, preserving cooldowns and image/reward
+  bookkeeping. Pending targets remain blocked by default. See issue #364.
 * Add :class:`~bsk_rl.scene.MixedScenario` and :class:`~bsk_rl.data.MixedReward`
   to combine independent scenarios and rewarders with different data types.
   Explicit scenario and reward-channel IDs bind named satellite groups, including
   overlapping assignments and passive target spacecraft. Communication shares
   channel data while preserving local data production and reward attribution.
   See :doc:`the mixed ground and space imaging example <examples/mixed_environment>`.
-
 * Add :doc:`the space-to-space imaging example <examples/space_to_space_rso_imaging>`
   for independently orbiting spacecraft:
   replayable RSO catalogs, explicitly named observers, moving-target access,
@@ -22,20 +24,17 @@ Development - |version|
   Basilisk controller update is tracked in `#358
   <https://github.com/AVSLab/bsk_rl/issues/358>`_.
   Existing RSO surface-point inspection is unchanged.
-
 * Validate satellite names at construction and renaming so simulator events use
   those names directly. Names must be valid Python identifiers, for example
   ``EO_1`` rather than ``EO-1``. Target catalog IDs may still contain punctuation.
   RSO image rewards default to full priority credit after complete downlink;
   acquisition shaping is available through explicit reward callbacks.
-
 * Fix incorrect eclipse-observation test assertions that assumed countdowns could
   not increase across a long step. The tests now use the same simulation start
   date, orbit, and random seed on every run, so the results are repeatable.
   They check shadow entry and countdown rollover to later orbits.
   Eclipse prediction is unchanged.
   See `#359 <https://github.com/AVSLab/bsk_rl/issues/359>`_.
-
 * Fix a bug where condensing ``NO_ACTION`` steps for sMDP discounting dropped the
   final observation and ``d_ts``, and could give the last super-action an empty
   reward. See issue #248.

src/bsk_rl/scene/rso_targets.py

@@ -254,9 +254,25 @@ class RSOTargets(Scenario):
     ``oe``, and attitude arguments and an ``EclipseDynModel`` for illumination.
     """
 
-    def __init__(self, catalog: RSOTargetCatalog, imager_names: list[str]) -> None:
-        """Select participants by explicit identity, independently of list order."""
+    def __init__(
+        self,
+        catalog: RSOTargetCatalog,
+        imager_names: list[str],
+        *,
+        block_pending_targets: bool = True,
+    ) -> None:
+        """Select participants by explicit identity, independently of list order.
+
+        Args:
+            catalog: Immutable target definitions for this scenario.
+            imager_names: Names of the satellites that can image catalog targets.
+            block_pending_targets: If True, exclude targets with undelivered images
+                from selection. Set False to allow reimaging before delivery while
+                retaining pending-product bookkeeping and delivery-based cooldowns.
+                Reward credit rules are unchanged. Defaults to True.
+        """
         super().__init__()
+        self.block_pending_targets = block_pending_targets
         self.catalog = catalog
         self.imager_names = tuple(imager_names)
         if not self.imager_names or len(set(self.imager_names)) != len(
@@ -373,8 +389,11 @@ def reset_during_sim_init(self) -> None:
                 imager.dynamics.bind_rso_target(target)
 
     def is_eligible(self, target_id: str, time_s: float) -> bool:
-        """Centralized pending visibility and quality-verified shared cooldown."""
-        return not self.pending.get(target_id) and time_s >= self.cooldown_until.get(
+        """Apply optional pending-product blocking and the shared cooldown."""
+        pending_blocks = self.block_pending_targets and bool(
+            self.pending.get(target_id)
+        )
+        return not pending_blocks and time_s >= self.cooldown_until.get(
             target_id, -np.inf
         )
 

tests/integration/test_int_rso_imaging_env.py

@@ -65,6 +65,7 @@ def make_env(
     composed=False,
     image_size=1000,
     sim_rate=1,
+    block_pending_targets=True,
 ):
     definitions = (
         scene.RSOTarget("83", "debris", (7e6 + 1000, 0, 0), (0, 7500, 0), priority=2),
@@ -110,14 +111,18 @@ def make_env(
     if composed:
         resource = data.ResourceReward(
             reward_weight=0,
-            resource_fn=lambda sat: sat.dynamics.battery_charge_fraction
-            if hasattr(sat.dynamics, "battery_charge_fraction")
-            else 0,
+            resource_fn=lambda sat: (
+                sat.dynamics.battery_charge_fraction
+                if hasattr(sat.dynamics, "battery_charge_fraction")
+                else 0
+            ),
         )
         rewarder = (resource, rewarder)
     return ConstellationTasking(
         satellites=satellites,
-        scenario=ReorderedScene(catalog, names),
+        scenario=ReorderedScene(
+            catalog, names, block_pending_targets=block_pending_targets
+        ),
         rewarder=rewarder,
         world_args=dict(utc_init=EPOCH, gsMinimumElevation=np.radians(-90)),
         time_limit=120,
@@ -133,9 +138,11 @@ def camera(env, name="camera_a"):
 
 def actions(env, imager_action=None, initial=False):
     return {
-        sat.name: imager_action
-        if sat.name in env.scenario.imager_names
-        else (0 if initial else None)
+        sat.name: (
+            imager_action
+            if sat.name in env.scenario.imager_names
+            else (0 if initial else None)
+        )
         for sat in env.satellites
     }
 
@@ -583,3 +590,43 @@ def test_evaluation_plots_and_step_budget(tmp_path, monkeypatch):
     assert result["captures"] == 1 and result["deliveries"] == 0
     for name in ("performance.png", "performance.pdf", "actions.png", "actions.pdf"):
         assert (output / name).stat().st_size > 1000
+
+
+@pytest.mark.parametrize("block_pending_targets", [False, True])
+def test_reimaging_before_downlink_preserves_physical_products(block_pending_targets):
+    env = make_env(block_pending_targets=block_pending_targets)
+    try:
+        env.reset(seed=0)
+        observer = camera(env)
+        env.step(actions(env, "83", initial=True))
+        for _ in range(20):
+            if observer.fsw.rso_capture.records:
+                break
+            env.step(actions(env))
+        first = observer.fsw.rso_capture.records[0]
+        assert env.scenario.pending["83"] == {first.record_id}
+        assert observer.dynamics.storage_level == 1000
+        assert env.rewarder.cum_reward["camera_a"] == pytest.approx(1.8)
+        if block_pending_targets:
+            with pytest.raises(ValueError, match="pending or in cooldown"):
+                env.step(actions(env, "83"))
+            assert len(observer.fsw.rso_capture.records) == 1
+        else:
+            env.step(actions(env, "83"))
+            for _ in range(20):
+                if len(observer.fsw.rso_capture.records) == 2:
+                    break
+                env.step(actions(env))
+            records = observer.fsw.rso_capture.records
+            assert [record.target_id for record in records] == ["83", "83"]
+            product_ids = {record.record_id for record in records}
+            assert len(product_ids) == 2
+            assert env.scenario.pending["83"] == product_ids
+            assert observer.dynamics.storage_level == 2000
+            store = rso_store(observer)
+            assert set(store.products) == product_ids
+            assert store.remaining_bits == {key: 1000 for key in product_ids}
+            assert not store.data.deliveries
+            assert env.rewarder.cum_reward["camera_a"] == pytest.approx(1.8)
+    finally:
+        env.close()

tests/unittest/act/test_rso_imaging_actions.py

@@ -6,6 +6,7 @@
 import pytest
 
 from bsk_rl.act import ImageRSO
+from bsk_rl.scene import RSOTarget, RSOTargetCatalog, RSOTargets
 from bsk_rl.utils.rso_imaging import CandidateSnapshot
 
 
@@ -69,3 +70,43 @@ def test_stale_ineligible_target_does_not_substitute_another_candidate():
         pytest.raises(ValueError, match="pending or in cooldown"),
     ):
         action.set_action(0)
+
+
+@pytest.mark.parametrize("block_pending_targets", [False, True])
+@pytest.mark.parametrize("selection", ["candidate", "target_id"])
+def test_pending_target_tasking_respects_scenario_setting(
+    block_pending_targets, selection
+):
+    definition = RSOTarget("observed", "debris", (7e6, 0, 0), (0, 7500, 0))
+    mission = RSOTargets(
+        RSOTargetCatalog((definition,)),
+        ["camera"],
+        block_pending_targets=block_pending_targets,
+    )
+    mission.targets_by_id[definition.id] = definition
+    mission.pending[definition.id] = {"camera:1"}
+    action = ImageRSO(1, variable_duration_imaging=False)
+    fsw = NS(action_image_rso=Mock())
+    action.satellite = NS(
+        rso_scenario=mission,
+        simulator=NS(sim_time=10, eventMap={}),
+        fsw=fsw,
+        update_timed_terminal_event=Mock(),
+        _rso_candidate_snapshot=CandidateSnapshot(10, 0, (definition,)),
+    )
+    action.simulator = action.satellite.simulator
+    select = (
+        (lambda: action.set_action(0))
+        if selection == "candidate"
+        else (lambda: action.set_action_override(definition.id))
+    )
+    if block_pending_targets:
+        with pytest.raises(ValueError, match="pending or in cooldown"):
+            select()
+        fsw.action_image_rso.assert_not_called()
+    else:
+        assert select() == definition.id
+        assert fsw.action_image_rso.call_args.args == (definition,)
+        action.satellite.update_timed_terminal_event.assert_called_once()
+        assert action.chosen_target_ids == [definition.id]
+    assert mission.pending[definition.id] == {"camera:1"}

tests/unittest/data/test_rso_imaging_data.py

@@ -13,6 +13,7 @@
     RSOImageStore,
 )
 from bsk_rl.data.composition import ComposedReward
+from bsk_rl.scene import RSOTarget, RSOTargetCatalog, RSOTargets
 
 
 def weighted_reward(alpha=0.1, **kwargs):
@@ -44,8 +45,9 @@ def test_custom_stage_rewards_receive_the_record_and_current_target():
     rewarder = RSOImageReward(
         multi_imager_credit="shared",
         acquisition_reward_fn=lambda record, target: record.quality * target.priority,
-        delivery_reward_fn=lambda record, target: target.priority
-        / (record.delivery_time - record.capture_time),
+        delivery_reward_fn=lambda record, target: (
+            target.priority / (record.delivery_time - record.capture_time)
+        ),
     )
     rewarder.link_scenario(mission)
     rewarder.reset_overwrite_previous()
@@ -276,3 +278,50 @@ def test_multi_imager_scientific_choice_is_required_and_per_owner_mode_works():
             ),
         }
     ) == {"cam-a": 0.2, "cam-b": 0.2}
+
+
+@pytest.mark.parametrize("block_pending_targets", [False, True])
+def test_selection_setting_preserves_pending_delivery_and_reward_accounting(
+    block_pending_targets,
+):
+    definition = RSOTarget("odd/83", "debris", (7e6, 0, 0), (0, 7500, 0), priority=2)
+    mission = RSOTargets(
+        RSOTargetCatalog((definition,)),
+        ["cam-a", "cam-b"],
+        block_pending_targets=block_pending_targets,
+    )
+    mission.targets_by_id = {definition.id: definition}
+    rewarder = weighted_reward(multi_imager_credit="shared", cooldown_s=100)
+    rewarder.link_scenario(mission)
+    rewarder.reset_overwrite_previous()
+    first, second = image(), image("a:2", time=15)
+    assert rewarder.calculate_reward(
+        {"cam-a": RSOImageData({first.record_id: first})}
+    ) == {"cam-a": 1.8}
+    assert mission.is_eligible(definition.id, 15) is (not block_pending_targets)
+    # Reimaging does not create additional acquisition credit for the same service.
+    assert rewarder.calculate_reward(
+        {"cam-a": RSOImageData({second.record_id: second})}
+    ) == {"cam-a": 0}
+    assert mission.pending[definition.id] == {first.record_id, second.record_id}
+    assert rewarder.calculate_reward(
+        {
+            "cam-a": RSOImageData(
+                deliveries={first.record_id: replace(first, delivery_time=30)}
+            )
+        }
+    ) == {"cam-a": 0.2}
+    assert mission.pending[definition.id] == {second.record_id}
+    assert mission.cooldown_until[definition.id] == 110
+    assert not mission.is_eligible(definition.id, 109)
+    assert mission.is_eligible(definition.id, 110) is (not block_pending_targets)
+    assert rewarder.calculate_reward(
+        {
+            "cam-a": RSOImageData(
+                deliveries={second.record_id: replace(second, delivery_time=35)}
+            )
+        }
+    ) == {"cam-a": 0}
+    assert mission.pending[definition.id] == set()
+    assert mission.cooldown_until[definition.id] == 115
+    assert mission.is_eligible(definition.id, 115)

tests/unittest/obs/test_rso_imaging_observations.py

@@ -10,6 +10,7 @@
 from bsk_rl.act import ImageRSO
 from bsk_rl.obs import RSOTargetProperties
 from bsk_rl.obs.rso_imaging import target_properties
+from bsk_rl.scene import RSOTarget, RSOTargetCatalog, RSOTargets
 from bsk_rl.utils.rso_imaging import candidate_snapshot, illumination_factor
 
 
@@ -157,3 +158,38 @@ def test_illumination_accessor_handles_equivalent_native_wrappers():
     payload = spacecraftLocation.EclipseMsgPayload()
     messaging.EclipseMsgPayload.illuminationFactor.fset(payload, 0.8)
     assert illumination_factor(payload) == 0.8
+
+
+@pytest.mark.parametrize("block_pending_targets", [False, True])
+def test_observations_filter_pending_targets_according_to_scenario_setting(
+    block_pending_targets,
+):
+    sat = observer()
+    targets = sat.rso_scenario.targets_by_id
+    catalog = RSOTargetCatalog(
+        tuple(
+            RSOTarget(
+                target.id,
+                f"target-{index}",
+                target.target_spacecraft.dynamics.r_BN_N,
+                target.target_spacecraft.dynamics.v_BN_N,
+            )
+            for index, target in enumerate(targets.values())
+        )
+    )
+    mission = RSOTargets(
+        catalog, ["camera"], block_pending_targets=block_pending_targets
+    )
+    mission.targets_by_id = targets
+    mission.pending["83"] = {"camera:1"}
+    sat.rso_scenario = mission
+    observation = RSOTargetProperties(dict(prop="valid"), n_ahead_observe=3)
+    observation.link_satellite(sat)
+    rows = observation.get_obs()
+    candidate_ids = observation.candidate_ids()
+    assert ("83" in candidate_ids) is (not block_pending_targets)
+    assert "far/blue" in candidate_ids
+    assert sum(row["valid"] for row in rows.values()) == (
+        1 if block_pending_targets else 2
+    )
+    assert mission.pending["83"] == {"camera:1"}

tests/unittest/scene/test_rso_targets.py

@@ -137,3 +137,46 @@ def test_ambiguous_original_names_are_rejected_before_environment_renaming():
             satellites=[NS(name="camera"), NS(name="camera"), NS(name="debris")],
             scenario=RSOTargets(catalog(), ["camera_0"]),
         )
+
+
+def test_pending_targets_are_blocked_until_all_products_are_delivered_by_default():
+    scene = RSOTargets(catalog(), ["camera"])
+    assert scene.is_eligible("id-83", 0)
+    scene.pending["id-83"] = {"camera:1", "camera:2"}
+    assert not scene.is_eligible("id-83", 1000)
+    scene.pending["id-83"].remove("camera:1")
+    assert not scene.is_eligible("id-83", 1000)
+    scene.pending["id-83"].clear()
+    assert scene.is_eligible("id-83", 1000)
+
+
+def test_pending_targets_can_be_selected_without_clearing_product_records():
+    scene = RSOTargets(catalog(), ["camera"], block_pending_targets=False)
+    scene.pending["id-83"] = {"camera:1"}
+    assert scene.is_eligible("id-83", 0)
+    assert scene.pending["id-83"] == {"camera:1"}
+
+
+def test_allowing_pending_targets_still_enforces_cooldown_boundary():
+    scene = RSOTargets(catalog(), ["camera"], block_pending_targets=False)
+    scene.pending["id-83"] = {"camera:1"}
+    scene.cooldown_until["id-83"] = 20
+    assert not scene.is_eligible("id-83", 19)
+    assert scene.is_eligible("id-83", 20)
+    assert scene.pending["id-83"] == {"camera:1"}
+
+
+@pytest.mark.parametrize("block_pending_targets", [False, True])
+def test_pending_selection_setting_survives_episode_reset(block_pending_targets):
+    scene = RSOTargets(
+        catalog(), ["camera"], block_pending_targets=block_pending_targets
+    )
+    scene.pending["id-83"] = {"camera:1"}
+    scene.cooldown_until["id-83"] = 20
+    scene.reset_overwrite_previous()
+    assert scene.block_pending_targets is block_pending_targets
+    assert scene.pending == {}
+    assert scene.cooldown_until == {}
+    assert scene.is_eligible("id-83", 0)
+    scene.pending["id-83"] = {"camera:2"}
+    assert scene.is_eligible("id-83", 0) is (not block_pending_targets)