[#265] Add maximum range dynamics model

Author: Mark2000

docs/source/release_notes.rst

@@ -29,6 +29,8 @@ Development - |version|
 * Allow for the ``time_limit`` to be randomized.
 * Added observation for arbitrary relative states between two satellites in :class:`~bsk_rl.obs.RelativeProperties`.
 * Allow for the ``transmitterPacketSize`` to be specified. The default sets it to the instrument's baud rate.
+* Add a maximum range checking dynamics model in :class:`~bsk_rl.dynamics.MaxRangeDynModel`. Useful for keeping an agent
+  in the vicinity of a target early in training.
 
 
 Version 1.1.0

src/bsk_rl/sats/satellite.py

@@ -282,10 +282,7 @@ def record_death(self, time: float) -> None:
     @property
     def _satellite_command(self) -> str:
         """Generate string that refers to self in simBase."""
-        return (
-            "[satellite for satellite in self.satellites "
-            + f"if satellite.name=='{self.name}'][0]"
-        )
+        return f"self.get_satellite('{self.name}')"
 
     def _info_command(self, info: str) -> str:
         """Generate command to log to info from an event.

src/bsk_rl/sim/dyn.py

@@ -1256,6 +1256,65 @@ def setup_conjunctions(self, conjunction_radius: float, **kwargs) -> None:
                 )
 
 
+class MaxRangeDynModel(BasicDynamicsModel):
+    """For evaluating a maximum range limitation between satellites."""
+
+    def __init__(self, *args, **kwargs) -> None:
+        """Model that checks for maximum range violations between satellites.
+
+        The simulation is terminated at the time of separation and a range_valid failure is reported.
+        """
+        super().__init__(*args, **kwargs)
+        self.out_of_ranges = []
+
+    def _setup_dynamics_objects(self, **kwargs) -> None:
+        super()._setup_dynamics_objects(**kwargs)
+        self.setup_range(**kwargs)
+
+    @aliveness_checker
+    def range_valid(self) -> bool:
+        """Check if conjunction has not occurred."""
+        return len(self.out_of_ranges) == 0
+
+    @default_args(max_range_radius=5000, chief_name=None)
+    def setup_range(self, max_range_radius: float, chief_name: str, **kwargs) -> None:
+        """Set up maximum distance checking relative to a chief satellite.
+
+        Args:
+            max_range_radius: [m] Maximum allowed range from the chief satellite.
+            chief_name: Chief satellite to check range against.
+            kwargs: Passed to other setup functions.
+        """
+        self.max_range_radius = max_range_radius
+        self.chief_name = chief_name
+
+        if self.chief_name is None:
+            self.logger.warning(
+                "No chief satellite specified for maximum range checking. "
+                "Range checking is disabled."
+            )
+            return
+
+        self.simulator.createNewEvent(
+            valid_func_name(f"range_{self.satellite.name}_{self.chief_name}"),
+            macros.sec2nano(self.simulator.sim_rate),
+            True,
+            [
+                f"np.linalg.norm(np.array({self.satellite._satellite_command}.dynamics.r_BN_N)"
+                + f"- np.array(self.get_satellite('{self.chief_name}').dynamics.r_BN_N))"
+                + " >= "
+                + f"{self.satellite._satellite_command}.dynamics.max_range_radius"
+            ],
+            [
+                self.satellite._info_command(
+                    f"Exceeded maximum range of {max_range_radius} m from {self.chief_name}"
+                ),
+                f"{self.satellite._satellite_command}.dynamics.out_of_ranges.append(self.get_satellite('{self.chief_name}'))",
+            ],
+            terminal=True,
+        )
+
+
 __doc_title__ = "Dynamics Sims"
 __all__ = [
     "DynamicsModel",
@@ -1265,5 +1324,6 @@ def setup_conjunctions(self, conjunction_radius: float, **kwargs) -> None:
     "ContinuousImagingDynModel",
     "GroundStationDynModel",
     "ConjunctionDynModel",
+    "MaxRangeDynModel",
     "FullFeaturedDynModel",
 ]

src/bsk_rl/sim/simulator.py

@@ -162,6 +162,20 @@ def delete_event(self, event_name) -> None:
         self.eventList.remove(event)
         del self.eventMap[event_name]
 
+    def get_satellite(self, name: str) -> "Satellite":
+        """Get a satellite by name.
+
+        Args:
+            name: Name of the satellite to retrieve.
+
+        Returns:
+            The satellite object with the specified name.
+        """
+        for sat in self.satellites:
+            if sat.name == name:
+                return sat
+        raise ValueError(f"Satellite with name '{name}' not found.")
+
     def __del__(self):
         """Log when simulator is deleted."""
         logger.debug("Basilisk simulator deleted")

tests/integration/sim/test_int_dynamics.py

@@ -158,3 +158,66 @@ class CollisionSat(sats.Satellite):
             assert sat1.dynamics.conjunctions == []
             assert sat2.dynamics.conjunctions == []
             assert sat3.dynamics.conjunctions == []
+
+
+class TestMaxRangeDynModel:
+    @pytest.mark.parametrize(
+        "rN1,rN2,max_range_violation",
+        [
+            ([1e8, 0, 0], [1e8 + 10, 0, 0], False),
+            ([1e8, 0, 0], [1e8 + 200, 0, 0], True),
+            ([0, 1e8, 0], [1e8, 0, 0], True),
+        ],
+    )
+    def test_max_range(self, rN1, rN2, max_range_violation):
+        class ChiefSat(sats.Satellite):
+            fsw_type = fsw.BasicFSWModel
+            dyn_type = dyn.BasicDynamicsModel
+            observation_spec = [obs.Time()]
+            action_spec = [act.Drift()]
+
+        class DeputySat(sats.Satellite):
+            fsw_type = fsw.BasicFSWModel
+            dyn_type = dyn.MaxRangeDynModel
+            observation_spec = [obs.Time()]
+            action_spec = [act.Drift()]
+
+        env = gym.make(
+            "ConstellationTasking-v1",
+            satellites=[
+                ChiefSat(
+                    "Chief",
+                    sat_args=dict(
+                        rN=rN1,
+                        vN=[0, 0, 0],
+                        oe=None,
+                    ),
+                ),
+                DeputySat(
+                    "Deputy",
+                    sat_args=dict(
+                        rN=rN2,
+                        vN=[0, 0, 0],
+                        oe=None,
+                        chief_name="Chief",
+                        max_range_radius=100,
+                    ),
+                ),
+            ],
+            sim_rate=1.0,
+            time_limit=100.0,
+            max_step_duration=1e9,
+            disable_env_checker=True,
+        )
+
+        env.reset()
+
+        env.step(dict(Chief=0, Deputy=0))
+
+        sat1 = env.unwrapped.satellites[0]
+        sat2 = env.unwrapped.satellites[1]
+
+        if max_range_violation:
+            assert sat2.dynamics.out_of_ranges == [sat1]
+        else:
+            assert sat2.dynamics.out_of_ranges == []

tests/unittest/sats/test_satellite.py

@@ -1,9 +1,11 @@
+from functools import partial
 from unittest.mock import MagicMock, patch
 
 import numpy as np
 import pytest
 
 from bsk_rl import sats
+from bsk_rl.sim import Simulator
 from bsk_rl.sim.fsw import Task
 
 
@@ -110,6 +112,7 @@ def test_satellite_command(self):
         sat1 = sats.Satellite(name="TestSat_1", sat_args={})
         sat2 = sats.Satellite(name="TestSat_2", sat_args={})
         self.satellites = [sat1, sat2]
+        self.get_satellite = partial(Simulator.get_satellite, self)
         assert sat1 == eval(sat1._satellite_command)
         assert sat1 != eval(sat2._satellite_command)
         assert sat2 == eval(sat2._satellite_command)