Actor Status Monitoring & Saving to File

paseos/actors/base_actor.py

@@ -43,6 +43,11 @@ class BaseActor(ABC):
     # Tracks the current activity
     _current_activity = None
 
+    # The following variables are used to track last evaluated state vectors to avoid recomputation.
+    _last_position = None
+    _last_velocity = None
+    _last_eclipse_status = None
+
     def __init__(self, name: str, epoch: pk.epoch) -> None:
         """Constructor for a base actor
 
@@ -57,6 +62,15 @@ def __init__(self, name: str, epoch: pk.epoch) -> None:
 
         self._communication_devices = DotMap(_dynamic=False)
 
+    @property
+    def current_activity(self) -> str:
+        """Returns the name of the activity the actor is currently performing.
+
+        Returns:
+            str: Activity name. None if no activity being performed.
+        """
+        return self._current_activity
+
     @property
     def local_time(self) -> pk.epoch:
         """Returns local time of the actor as pykep epoch. Use e.g. epoch.mjd2000 to get time in days.
@@ -145,6 +159,7 @@ def get_position(self, epoch: pk.epoch):
         # If the actor has no orbit, return position
         if self._orbital_parameters is None:
             if self._position is not None:
+                self._last_position = self._position
                 return self._position
         else:
             return self._orbital_parameters.eph(epoch)[0]
@@ -174,7 +189,10 @@ def get_position_velocity(self, epoch: pk.epoch):
             + str(epoch.mjd2000)
             + " (mjd2000)."
         )
-        return self._orbital_parameters.eph(epoch)
+        pos, vel = self._orbital_parameters.eph(epoch)
+        self._last_position = pos
+        self._last_velocity = vel
+        return pos, vel
 
     def is_in_line_of_sight(
         self,
@@ -208,7 +226,8 @@ def is_in_eclipse(self, t: pk.epoch = None):
         """
         if t is None:
             t = self._local_time
-        return is_in_eclipse(self, self._central_body, t)
+        self._last_eclipse_status = is_in_eclipse(self, self._central_body, t)
+        return self._last_eclipse_status
 
     def get_communication_window(
         self,

paseos/actors/spacecraft_actor.py

@@ -49,7 +49,7 @@ def battery_level_in_Ws(self):
         return self._battery_level_in_Ws
 
     @property
-    def battery_level_ratio(self):
+    def state_of_charge(self):
         """Get the current battery level as ratio of maximum.
 
         Returns:

paseos/paseos.py

@@ -1,12 +1,14 @@
 import types
 import asyncio
+import sys
 
 from dotmap import DotMap
 from loguru import logger
 import pykep as pk
 
 from paseos.actors.base_actor import BaseActor
 from paseos.activities.activity_manager import ActivityManager
+from paseos.utils.operations_monitor import OperationsMonitor
 
 
 class PASEOS:
@@ -32,6 +34,10 @@ class PASEOS:
     # Semaphore to track if an activity is currently running
     _is_running_activity = False
 
+    # Used to monitor the local actor over execution and write performance stats
+    _operations_monitor = None
+    _time_since_last_log = sys.float_info.max
+
     # Use automatic clock (default on for now)
     use_automatic_clock = True
 
@@ -60,6 +66,20 @@ def __init__(self, local_actor: BaseActor, cfg=None):
         # Update local actor time to simulation start time.
         self._local_actor.set_time(pk.epoch(self._cfg.sim.start_time * pk.SEC2DAY))
         self._activity_manager = ActivityManager(self, self._cfg.sim.activity_timestep)
+        self._operations_monitor = OperationsMonitor(self._local_actor.name)
+
+    def save_status_log_csv(self, filename) -> None:
+        """Saves the status log incl. all kinds of information such as battery charge,
+        running activtiy, etc.
+
+        Args:
+            filename (str): File to save the log in.
+        """
+        self._operations_monitor.save_to_csv(filename)
+
+    def log_status(self):
+        """Updates the status log."""
+        self._operations_monitor.log(self._local_actor, self.known_actor_names)
 
     def advance_time(self, time_to_advance: float):
         """Advances the simulation by a specified amount of time
@@ -90,6 +110,13 @@ def advance_time(self, time_to_advance: float):
             self._state.time += dt
             self._local_actor.set_time(pk.epoch(self._state.time * pk.SEC2DAY))
 
+            # Check if we should update the status log
+            if self._time_since_last_log > self._cfg.io.logging_interval:
+                self.log_status()
+                self._time_since_last_log = 0
+            else:
+                self._time_since_last_log += dt
+
         logger.debug("New time is: " + str(self._state.time) + " s.")
 
     def add_known_actor(self, actor: BaseActor):

paseos/resources/default_cfg.toml

@@ -3,6 +3,9 @@ start_time = 0 # [s] Start time of the simulation in seconds after MJD2000
 dt = 10 # [s] Maximal Internal timestep used for computing charging, etc.
 activity_timestep = 1 # [s] Internal timestep at which activities update, try to charge and discharge etc. 
 
+[io]
+logging_interval = 10 # [s] after how many seconds should paseos log the actor status
+
 [power]
 
 [comms]

paseos/tests/actor_builder_test.py

@@ -40,7 +40,7 @@ def test_add_power_devices():
     _, sat1, _ = get_default_instance()
     ActorBuilder.set_power_devices(sat1, 42, 42, 42)
     assert sat1.battery_level_in_Ws == 42
-    assert sat1.battery_level_ratio == 1
+    assert sat1.state_of_charge == 1
     assert sat1.charging_rate_in_W == 42
 
 

paseos/tests/operations_monitor_test.py

@@ -0,0 +1,59 @@
+"""Simple test for the operations monitor"""
+
+import asyncio
+import pytest
+import pykep as pk
+
+import paseos
+from paseos import ActorBuilder, SpacecraftActor, load_default_cfg
+
+
+async def wait_for_activity(sim):
+    while sim._is_running_activity is True:
+        await asyncio.sleep(0.1)
+
+
+# tell pytest to create an event loop and execute the tests using the event loop
+@pytest.mark.asyncio
+async def test_monitor():
+    """Test to see if we can log while performing an activity and write to file then.
+    To fully judge the outcome, have a look at the generated test file."""
+    # Define central body
+    earth = pk.planet.jpl_lp("earth")
+
+    # Define local actor
+    sat1 = ActorBuilder.get_actor_scaffold("sat1", SpacecraftActor, pk.epoch(0))
+    ActorBuilder.set_orbit(sat1, [10000000, 0, 0], [0, 8000.0, 0], pk.epoch(0), earth)
+    ActorBuilder.set_power_devices(sat1, 500, 10000, 1)
+
+    # init simulation
+    cfg = load_default_cfg()  # loading cfg to modify defaults
+    cfg.sim.dt = 0.1  # setting lower timestep to run things quickly
+    cfg.sim.activity_timestep = 0.1
+    cfg.io.logging_interval = 0.25  # log every 0.25 seconds
+    cfg.sim.time_multiplier = 10  # speed up execution for convenience
+    sim = paseos.init_sim(sat1, cfg)
+
+    async def func1(args):
+        await asyncio.sleep(0.5)
+
+    async def func2(args):
+        await asyncio.sleep(1.0)
+
+    # Register an activity that draws 10 watt per second
+    sim.register_activity(
+        "Activity_1", activity_function=func1, power_consumption_in_watt=2
+    )
+
+    sim.register_activity(
+        "Activity_2", activity_function=func2, power_consumption_in_watt=10
+    )
+
+    # Run the activity
+    sim.perform_activity("Activity_1")
+    await wait_for_activity(sim)
+
+    sim.perform_activity("Activity_2")
+    await wait_for_activity(sim)
+
+    sim.save_status_log_csv("test.csv")

paseos/utils/operations_monitor.py

@@ -0,0 +1,81 @@
+import csv
+
+from loguru import logger
+from dotmap import DotMap
+import pykep as pk
+
+from paseos.actors.base_actor import BaseActor
+from paseos.actors.spacecraft_actor import SpacecraftActor
+
+
+class OperationsMonitor:
+    """This class is used to track actor status and activities over time."""
+
+    def __init__(self, actor_name):
+        """Initializes the OperationsMonitor
+
+        Args:
+            actor_name (str): Name of the local actor.
+        """
+        logger.trace("Initializing OperationsMonitor for " + actor_name)
+        self._actor_name = actor_name
+        self._log = DotMap(_dynamic=False)
+        self._log.timesteps = []
+        self._log.current_activity = []
+        self._log.state_of_charge = []
+        self._log.is_in_eclipse = []
+        self._log.known_actors = []
+        self._log.position = []
+        self._log.velocity = []
+
+    def log(
+        self,
+        local_actor: BaseActor,
+        known_actors: list,
+    ):
+        """Log the current time step.
+
+        Args:
+            local_actor (BaseActor): The local actors whose status we are monitoring.
+            known_actors (list): List of names of the known actors.
+        """
+        logger.trace("Logging iteration")
+        assert local_actor.name == self._actor_name, (
+            "Expected actor's name was" + self._actor_name
+        )
+        self._log.timesteps.append(local_actor.local_time.mjd2000 * pk.DAY2SEC)
+        self._log.current_activity.append(local_actor.current_activity)
+        self._log.position.append(local_actor._last_position)
+        self._log.velocity.append(local_actor._last_velocity)
+        self._log.known_actors.append(known_actors)
+        if isinstance(local_actor, SpacecraftActor):
+            self._log.state_of_charge.append(local_actor.state_of_charge)
+        else:
+            self._log.state_of_charge.append(1.0)
+
+        if local_actor._last_eclipse_status is None:
+            self._log.is_in_eclipse.append(False)
+        else:
+            self._log.is_in_eclipse.append(local_actor._last_eclipse_status)
+
+    def save_to_csv(self, filename):
+        """Write the created log file to a csv file.
+
+        Args:
+            filename (str): File to store the log in.
+        """
+        logger.trace("Writing status log file to " + filename)
+        with open(filename, "w", newline="") as f:
+            w = csv.DictWriter(f, self._log.keys())
+            w.writeheader()
+            for i in range(len(self._log.timesteps)):
+                row = {
+                    "timesteps": self._log.timesteps[i],
+                    "current_activity": self._log.current_activity[i],
+                    "position": self._log.position[i],
+                    "velocity": self._log.velocity[i],
+                    "known_actors": self._log.known_actors[i],
+                    "state_of_charge": self._log.state_of_charge[i],
+                    "is_in_eclipse": self._log.is_in_eclipse[i],
+                }
+                w.writerow(row)

paseos/visualization/space_animation.py

@@ -147,7 +147,7 @@ def _populate_textbox(self, actor: BaseActor) -> str:
         info_str = f"{actor.name}"
         if isinstance(actor, SpacecraftActor):
             if actor.battery_level_in_Ws is not None:
-                battery_level = actor.battery_level_ratio * 100
+                battery_level = actor.state_of_charge * 100
                 info_str += f"\nBattery: {battery_level:.0f}%"
 
             for name in actor.communication_devices.keys():