From 4060e5d9dd406c4d5775f7a6b326639b2207e3f7 Mon Sep 17 00:00:00 2001 From: LorenzzoQM Date: Thu, 14 Sep 2023 15:33:23 -0600 Subject: [PATCH 1/2] Issue #45: Added functions for Continuous Nadir Scanning --- .../scenario/data.py | 95 +++++++++++++++ .../scenario/environment_features.py | 14 +++ .../scenario/sat_actions.py | 3 + .../simulation/dynamics.py | 88 ++++++++++++++ .../simulation/fsw.py | 112 +++++++++++++++--- 5 files changed, 296 insertions(+), 16 deletions(-) diff --git a/bsk_rl/envs/general_satellite_tasking/scenario/data.py b/bsk_rl/envs/general_satellite_tasking/scenario/data.py index a9ba5001..19f5ba0a 100644 --- a/bsk_rl/envs/general_satellite_tasking/scenario/data.py +++ b/bsk_rl/envs/general_satellite_tasking/scenario/data.py @@ -371,3 +371,98 @@ def _calc_reward(self, new_data_dict: dict[str, UniqueImageData]) -> float: # reward += reward - self.data.rewards[target] # self.data += new_data # return reward + +################# +# Nadir Pointing# +################# + + +class NadirScanningTimeData(DataType): + def __init__(self, scanning_time: float = 0.0) -> None: + """DataType to log data generated scanning nadir + + Args: + scanning_time: Time spent scanning nadir + """ + self.scanning_time = scanning_time + + def __add__(self, other: "NadirScanningTimeData") -> "NadirScanningTimeData": + """Define the combination of two units of data""" + scanning_time = self.scanning_time + other.scanning_time + + return self.__class__(scanning_time) + + +class ScanningNadirTimeStore(DataStore): + DataType = NadirScanningTimeData + + def _get_log_state(self) -> LogStateType: + """Returns the amount of data stored in the storage unit.""" + + storage_unit = self.satellite.dynamics.storageUnit.storageUnitDataOutMsg.read() + stored_amount = storage_unit.storageLevel + + # return amount of data stored + return stored_amount + + def _compare_log_states( + self, old_state: float, new_state: float + ) -> "NadirScanningTimeData": + """Generate a unit of data based on previous step and current step stored + data amount. + + Args: + old_state: Previous amount of data in the storage unit + new_state: Current amount of data in the storage unit + + Returns: + DataType: Data generated + """ + instrument_baudrate = self.satellite.dynamics.instrument.nodeBaudRate + + if new_state > old_state: + data_generated = (new_state - old_state) / instrument_baudrate + else: + data_generated = 0.0 + + return NadirScanningTimeData(data_generated) + + +class NadirScanningManager(DataManager): + DataStore = ScanningNadirTimeStore # type of DataStore managed by the DataManager + + def __init__( + self, + env_features: Optional["EnvironmentFeatures"] = None, + reward_fn: Callable = None, + ) -> None: + """ + Args: + env_features: Information about the environment that can be collected as + data + reward_fn: Reward as function of time spend pointing nadir + """ + super().__init__(env_features) + if reward_fn is None: + + def reward_fn(p): + # Reward as a function of time send pointing nadir (p) and value + # per second + return p * self.env_features.value_per_second + + self.reward_fn = reward_fn + + def _calc_reward(self, new_data_dict: ["NadirScanningTimeData"]) -> float: + """Calculate step reward based on all satellite data from a step + + Args: + new_data_dict (dict): Satellite-DataType of new data from a step + + Returns: + Step reward + """ + reward = 0.0 + for scanning_time in new_data_dict.values(): + reward += self.reward_fn(scanning_time.scanning_time) + + return reward diff --git a/bsk_rl/envs/general_satellite_tasking/scenario/environment_features.py b/bsk_rl/envs/general_satellite_tasking/scenario/environment_features.py index 23187786..cc539dc4 100644 --- a/bsk_rl/envs/general_satellite_tasking/scenario/environment_features.py +++ b/bsk_rl/envs/general_satellite_tasking/scenario/environment_features.py @@ -150,3 +150,17 @@ def regenerate_targets(self) -> None: priority=self.priority_distribution(), ) ) + + +class UniformNadirFeature(EnvironmentFeatures): + """ + Defines a nadir target center at the center of the planet. + """ + + def __init__(self, value_per_second: float = 1.0) -> None: + """ " + Args: + value_per_second: Amount of reward per second imaging nadir. + """ + self.name = "NadirFeature" + self.value_per_second = value_per_second diff --git a/bsk_rl/envs/general_satellite_tasking/scenario/sat_actions.py b/bsk_rl/envs/general_satellite_tasking/scenario/sat_actions.py index abf77fc2..29d483ed 100644 --- a/bsk_rl/envs/general_satellite_tasking/scenario/sat_actions.py +++ b/bsk_rl/envs/general_satellite_tasking/scenario/sat_actions.py @@ -173,3 +173,6 @@ def set_action(self, action: Union[int, Target, str]): self.prev_action_key = self.image(action, self.prev_action_key) else: super().set_action(action) + + +NadirImagingAction = fsw_action_gen("action_nadir_scan") diff --git a/bsk_rl/envs/general_satellite_tasking/simulation/dynamics.py b/bsk_rl/envs/general_satellite_tasking/simulation/dynamics.py index 02a05d94..7df395d7 100644 --- a/bsk_rl/envs/general_satellite_tasking/simulation/dynamics.py +++ b/bsk_rl/envs/general_satellite_tasking/simulation/dynamics.py @@ -21,6 +21,7 @@ simpleNav, simplePowerSink, simpleSolarPanel, + simpleStorageUnit, spacecraft, spacecraftLocation, spaceToGroundTransmitter, @@ -786,6 +787,93 @@ def reset_for_action(self) -> None: self.instrumentPowerSink.powerStatus = 0 +class ContinuousImagingDynModel(ImagingDynModel): + """Equips the satellite with an instrument, storage unit, and transmitter + for continuous nadir imaging.""" + + @default_args(instrumentBaudRate=8e6) + def _set_instrument( + self, instrumentBaudRate: float, priority: int = 895, **kwargs + ) -> None: + """Create the continuous instrument model. + + Args: + instrumentBaudRate: Data generated in step by continuous imaging [bits] + priority: Model priority. + """ + self.instrument = simpleInstrument.SimpleInstrument() + self.instrument.ModelTag = "instrument" + self.satellite.id + self.instrument.nodeBaudRate = instrumentBaudRate # make imaging instantaneous + self.instrument.nodeDataName = "Instrument" + self.satellite.id + self.simulator.AddModelToTask( + self.task_name, self.instrument, ModelPriority=priority + ) + + @default_args(dataStorageCapacity=20 * 8e6) + def _set_storage_unit( + self, + dataStorageCapacity: int, + priority: int = 699, + **kwargs, + ) -> None: + """Configure the storage unit and its buffers. + + Args: + dataStorageCapacity: Maximum data to be stored [bits] + priority: Model priority. + """ + self.storageUnit = simpleStorageUnit.SimpleStorageUnit() + self.storageUnit.ModelTag = "storageUnit" + self.satellite.id + self.storageUnit.storageCapacity = dataStorageCapacity # bits + self.storageUnit.addDataNodeToModel(self.instrument.nodeDataOutMsg) + self.storageUnit.addDataNodeToModel(self.transmitter.nodeDataOutMsg) + + # Add the storage unit to the transmitter + self.transmitter.addStorageUnitToTransmitter( + self.storageUnit.storageUnitDataOutMsg + ) + + self.simulator.AddModelToTask( + self.task_name, self.storageUnit, ModelPriority=priority + ) + + @default_args( + imageTargetMaximumRange=-1, + ) + def _set_imaging_target( + self, + imageTargetMaximumRange: float = -1, + priority: int = 2000, + **kwargs, + ) -> None: + """Add a generic imaging target to dynamics. The target must be updated with a + particular location when used. + + Args: + imageTargetMaximumRange: Maximum range from target to satellite when + imaging. -1 to disable. [m] + priority: Model priority. + """ + self.imagingTarget = groundLocation.GroundLocation() + self.imagingTarget.ModelTag = "scanningTarget" + self.imagingTarget.planetRadius = 1e-6 + self.imagingTarget.specifyLocation(0, 0, 0) + self.imagingTarget.planetInMsg.subscribeTo( + self.environment.gravFactory.spiceObject.planetStateOutMsgs[ + self.environment.body_index + ] + ) + self.imagingTarget.minimumElevation = np.radians(-90) + self.imagingTarget.maximumRange = imageTargetMaximumRange + + self.simulator.AddModelToTask( + self.environment.env_task_name, + self.imagingTarget, + ModelPriority=priority, + ) + self.imagingTarget.addSpacecraftToModel(self.scObject.scStateOutMsg) + + class GroundStationDynModel(ImagingDynModel): """Model that connects satellite to environment ground stations""" diff --git a/bsk_rl/envs/general_satellite_tasking/simulation/fsw.py b/bsk_rl/envs/general_satellite_tasking/simulation/fsw.py index fcfdc33a..82973805 100644 --- a/bsk_rl/envs/general_satellite_tasking/simulation/fsw.py +++ b/bsk_rl/envs/general_satellite_tasking/simulation/fsw.py @@ -21,6 +21,7 @@ mrpSteering, rateServoFullNonlinear, rwMotorTorque, + scanningInstrumentController, simpleInstrumentController, thrForceMapping, thrMomentumDumping, @@ -575,7 +576,7 @@ def _make_task_list(self) -> list[Task]: def _set_gateway_msgs(self) -> None: super()._set_gateway_msgs() self.dynamics.instrument.nodeStatusInMsg.subscribeTo( - self.simpleInsControlConfig.deviceCmdOutMsg + self.insControlConfig.deviceCmdOutMsg ) class LocPointTask(Task): @@ -597,13 +598,13 @@ def create_module_data(self) -> None: self.locPointWrap.ModelTag = "locPoint" # SimpleInstrumentController configuration - self.simpleInsControlConfig = ( - self.fsw.simpleInsControlConfig + self.insControlConfig = ( + self.fsw.insControlConfig ) = simpleInstrumentController.simpleInstrumentControllerConfig() - self.simpleInsControlWrap = ( - self.fsw.simpleInsControlWrap - ) = self.fsw.simulator.setModelDataWrap(self.simpleInsControlConfig) - self.simpleInsControlWrap.ModelTag = "instrumentController" + self.insControlWrap = ( + self.fsw.insControlWrap + ) = self.fsw.simulator.setModelDataWrap(self.insControlConfig) + self.insControlWrap.ModelTag = "instrumentController" def init_objects(self, **kwargs) -> None: self._set_location_pointing(**kwargs) @@ -652,23 +653,23 @@ def _set_instrument_controller( imageRateErrorRequirement: Rate tolerance for imaging. Disable with None. [rad/s] """ - self.simpleInsControlConfig.attErrTolerance = imageAttErrorRequirement + self.insControlConfig.attErrTolerance = imageAttErrorRequirement if imageRateErrorRequirement is not None: - self.simpleInsControlConfig.useRateTolerance = 1 - self.simpleInsControlConfig.rateErrTolerance = imageRateErrorRequirement - self.simpleInsControlConfig.attGuidInMsg.subscribeTo(self.fsw.attGuidMsg) - self.simpleInsControlConfig.locationAccessInMsg.subscribeTo( + self.insControlConfig.useRateTolerance = 1 + self.insControlConfig.rateErrTolerance = imageRateErrorRequirement + self.insControlConfig.attGuidInMsg.subscribeTo(self.fsw.attGuidMsg) + self.insControlConfig.locationAccessInMsg.subscribeTo( self.fsw.dynamics.imagingTarget.accessOutMsgs[-1] ) self._add_model_to_task( - self.simpleInsControlWrap, self.simpleInsControlConfig, priority=987 + self.insControlWrap, self.insControlConfig, priority=987 ) def reset_for_action(self) -> None: self.fsw.dynamics.imagingTarget.Reset(self.fsw.simulator.sim_time_ns) self.locPointWrap.Reset(self.fsw.simulator.sim_time_ns) - self.simpleInsControlConfig.controllerStatus = 0 + self.insControlConfig.controllerStatus = 0 return super().reset_for_action() @action @@ -679,11 +680,11 @@ def action_image(self, location: Iterable[float], data_name: str) -> None: location: PCPF target location [m] data_name: Data buffer to store image data to """ - self.simpleInsControlConfig.controllerStatus = 1 + self.insControlConfig.controllerStatus = 1 self.dynamics.instrumentPowerSink.powerStatus = 1 self.dynamics.imagingTarget.r_LP_P_Init = location self.dynamics.instrument.nodeDataName = data_name - self.simpleInsControlConfig.imaged = 0 + self.insControlConfig.imaged = 0 self.simulator.enableTask(self.LocPointTask.name + self.satellite.id) @action @@ -699,6 +700,85 @@ def action_downlink(self) -> None: ) +class ContinuousImagingFSWModel(ImagingFSWModel): + class LocPointTask(ImagingFSWModel.LocPointTask): + """Task to point at targets and trigger the instrument""" + + def create_module_data(self) -> None: + # Location pointing configuration + self.locPointConfig = ( + self.fsw.locPointConfig + ) = locationPointing.locationPointingConfig() + self.locPointWrap = ( + self.fsw.locPointWrap + ) = self.fsw.simulator.setModelDataWrap(self.locPointConfig) + self.locPointWrap.ModelTag = "locPoint" + + # scanningInstrumentController configuration + self.insControlConfig = ( + self.fsw.insControlConfig + ) = scanningInstrumentController.scanningInstrumentControllerConfig() + self.insControlWrap = ( + self.fsw.simpleInsControlWrap + ) = self.fsw.simulator.setModelDataWrap(self.insControlConfig) + self.insControlWrap.ModelTag = "instrumentController" + + @default_args(imageAttErrorRequirement=0.01, imageRateErrorRequirement=None) + def _set_instrument_controller( + self, + imageAttErrorRequirement: float, + imageRateErrorRequirement: float, + **kwargs, + ) -> None: + """Defines the instrument controller parameters. + + Args: + imageAttErrorRequirement: Pointing attitude error tolerance for imaging + [MRP norm] + imageRateErrorRequirement: Rate tolerance for imaging. Disable with + None. [rad/s] + """ + self.insControlConfig.attErrTolerance = imageAttErrorRequirement + if imageRateErrorRequirement is not None: + self.insControlConfig.useRateTolerance = 1 + self.insControlConfig.rateErrTolerance = imageRateErrorRequirement + self.insControlConfig.attGuidInMsg.subscribeTo(self.fsw.attGuidMsg) + self.insControlConfig.accessInMsg.subscribeTo( + self.fsw.dynamics.imagingTarget.accessOutMsgs[-1] + ) + + self._add_model_to_task( + self.insControlWrap, self.insControlConfig, priority=987 + ) + + def reset_for_action(self) -> None: + self.instMsg = cMsgPy.DeviceCmdMsg_C() + self.instMsg.write(messaging.DeviceCmdMsgPayload()) + self.fsw.dynamics.instrument.nodeStatusInMsg.subscribeTo(self.instMsg) + return super().reset_for_action() + + @action + def action_nadir_scan(self) -> None: + """Action scan nadir. + + Args: + location: PCPF target location [m] + data_name: Data buffer to store image data to + """ + self.dynamics.instrument.nodeStatusInMsg.subscribeTo( + self.insControlConfig.deviceCmdOutMsg + ) + self.insControlConfig.controllerStatus = 1 + self.dynamics.instrumentPowerSink.powerStatus = 1 + self.dynamics.imagingTarget.r_LP_P_Init = np.array([0, 0, 0.1]) + self.dynamics.instrument.nodeDataName = "nadir" + self.simulator.enableTask(self.LocPointTask.name + self.satellite.id) + + @action + def action_image(self, *args, **kwargs) -> None: + raise NotImplementedError("Use action_nadir_scan instead") + + class SteeringFSWModel(BasicFSWModel): """FSW extending MRP control to use MRP steering instesd of MRP feedback.""" From 25d9c11a1bdd9e1334f23710a34a42833593553f Mon Sep 17 00:00:00 2001 From: LorenzzoQM Date: Thu, 14 Sep 2023 15:35:55 -0600 Subject: [PATCH 2/2] Issue #45: Tests for new features --- .../scenario/test_int_sat_actions.py | 40 +++++++++ .../scenario/test_data.py | 81 +++++++++++++++++++ .../scenario/test_environment_features.py | 7 ++ .../scenario/test_sat_actions.py | 9 +++ .../simulation/test_dynamics.py | 13 +++ 5 files changed, 150 insertions(+) diff --git a/tests/integration/envs/general_satellite_tasking/scenario/test_int_sat_actions.py b/tests/integration/envs/general_satellite_tasking/scenario/test_int_sat_actions.py index d31cbd80..76f68905 100644 --- a/tests/integration/envs/general_satellite_tasking/scenario/test_int_sat_actions.py +++ b/tests/integration/envs/general_satellite_tasking/scenario/test_int_sat_actions.py @@ -7,6 +7,7 @@ from bsk_rl.envs.general_satellite_tasking.scenario import sat_observations as so from bsk_rl.envs.general_satellite_tasking.scenario.environment_features import ( StaticTargets, + UniformNadirFeature, ) from bsk_rl.envs.general_satellite_tasking.simulation import dynamics, environment, fsw from bsk_rl.envs.general_satellite_tasking.utils.orbital import random_orbit @@ -148,3 +149,42 @@ def test_desat_action(self): assert np.linalg.norm( self.env.satellite.dynamics.wheel_speeds ) < np.linalg.norm(init_speeds) + + +class TestNadirImagingActions: + class ImageSat( + sa.NadirImagingAction, + so.TimeState, + ): + dyn_type = dynamics.ContinuousImagingDynModel + fsw_type = fsw.ContinuousImagingFSWModel + + env = gym.make( + "SingleSatelliteTasking-v1", + satellites=ImageSat( + "EO-1", + n_ahead_act=10, + sat_args=ImageSat.default_sat_args( + oe=random_orbit, + imageAttErrorRequirement=0.05, + imageRateErrorRequirement=0.05, + instrumentBaudRate=1.0, + dataStorageCapacity=3.0, + transmitterBaudRate=-1.0, + ), + ), + env_type=environment.BasicEnvironmentModel, + env_args=environment.BasicEnvironmentModel.default_env_args(), + env_features=UniformNadirFeature(), + data_manager=data.NoDataManager(), + sim_rate=1.0, + time_limit=10000.0, + max_step_duration=1e9, + disable_env_checker=True, + ) + + def test_image(self): + self.env.reset() + storage_init = self.env.satellite.dynamics.storage_level + self.env.step(0) + assert self.env.satellite.dynamics.storage_level > storage_init diff --git a/tests/unittest/envs/general_satellite_tasking/scenario/test_data.py b/tests/unittest/envs/general_satellite_tasking/scenario/test_data.py index f23c71b4..93f7ad14 100644 --- a/tests/unittest/envs/general_satellite_tasking/scenario/test_data.py +++ b/tests/unittest/envs/general_satellite_tasking/scenario/test_data.py @@ -183,3 +183,84 @@ def test_calc_reward_custom_fn(self): } ) assert reward == approx(1.5) + + +class TestNadirScanningTimeData: + def test_add_null(self): + dat1 = data.NadirScanningTimeData() + dat2 = data.NadirScanningTimeData() + dat = dat1 + dat2 + assert dat.scanning_time == 0.0 + + def test_add_to_null(self): + dat1 = data.NadirScanningTimeData(1.0) + dat2 = data.NadirScanningTimeData() + dat = dat1 + dat2 + assert dat.scanning_time == 1.0 + + def test_add(self): + dat1 = data.NadirScanningTimeData(1.0) + dat2 = data.NadirScanningTimeData(3.0) + dat = dat1 + dat2 + assert dat.scanning_time == 4.0 + + +class TestScanningNadirTimeStore: + def test_get_log_state(self): + sat = MagicMock() + sat.dynamics.storageUnit.storageUnitDataOutMsg.read().storageLevel = 6 + ds = data.ScanningNadirTimeStore(MagicMock(), sat) + assert ds._get_log_state() == 6.0 + + @pytest.mark.parametrize( + "before,after,new_time", + [ + (0, 3, 1), + (3, 6, 1), + (1, 1, 0), + (0, 6, 2), + ], + ) + def test_compare_log_states(self, before, after, new_time): + sat = MagicMock() + ds = data.ScanningNadirTimeStore(MagicMock(), sat) + sat.dynamics.instrument.nodeBaudRate = 3 + dat = ds._compare_log_states(before, after) + assert dat.scanning_time == new_time + + +class TestNadirScanningManager: + def test_calc_reward(self): + dm = data.NadirScanningManager(MagicMock()) + dm.data = data.NadirScanningTimeData([]) + dm.env_features.value_per_second = 1.0 + reward = dm._calc_reward( + { + "sat1": data.NadirScanningTimeData(1), + "sat2": data.NadirScanningTimeData(2), + } + ) + assert reward == approx(3) + + def test_calc_reward_existing(self): + dm = data.NadirScanningManager(MagicMock()) + dm.data = data.NadirScanningTimeData(1) + dm.env_features.value_per_second = 1.0 + reward = dm._calc_reward( + { + "sat1": data.NadirScanningTimeData(2), + "sat2": data.NadirScanningTimeData(3), + } + ) + assert reward == approx(5) + + def test_calc_reward_custom_fn(self): + dm = data.NadirScanningManager(MagicMock(), reward_fn=lambda x: 1 / x) + dm.data = data.NadirScanningTimeData([]) + reward = dm._calc_reward( + { + "sat1": data.NadirScanningTimeData(2), + "sat2": data.NadirScanningTimeData(2), + } + ) + assert reward == approx(1.0) diff --git a/tests/unittest/envs/general_satellite_tasking/scenario/test_environment_features.py b/tests/unittest/envs/general_satellite_tasking/scenario/test_environment_features.py index faead34a..95d95f2e 100644 --- a/tests/unittest/envs/general_satellite_tasking/scenario/test_environment_features.py +++ b/tests/unittest/envs/general_satellite_tasking/scenario/test_environment_features.py @@ -8,6 +8,7 @@ CityTargets, StaticTargets, Target, + UniformNadirFeature, lla2ecef, ) @@ -146,3 +147,9 @@ def test_regenerate_targets_offset(self, mock_read_csv, mock_lla2ecef): for target in ct.targets: assert np.linalg.norm(target.location - nominal) <= 0.03 assert np.linalg.norm(target.location) == approx(1.0) + + +class TestUniformNadirFeature: + def test_init(self): + st = UniformNadirFeature() + assert st.name == "NadirFeature" diff --git a/tests/unittest/envs/general_satellite_tasking/scenario/test_sat_actions.py b/tests/unittest/envs/general_satellite_tasking/scenario/test_sat_actions.py index cce8a1fb..088f3fc0 100644 --- a/tests/unittest/envs/general_satellite_tasking/scenario/test_sat_actions.py +++ b/tests/unittest/envs/general_satellite_tasking/scenario/test_sat_actions.py @@ -168,6 +168,15 @@ def test_set_action(self, sat_init, discrete_set, target): sat.image.assert_called_once() +@patch.multiple(sa.NadirImagingAction, __abstractmethods__=set()) +@patch("bsk_rl.envs.general_satellite_tasking.scenario.satellites.Satellite.__init__") +class TestNadirImagingActions: + def test_init(self, sat_init): + sat = sa.NadirImagingAction() + sat_init.assert_called_once() + assert sat.action_map == {"0": "action_nadir_scan"} + + @patch.multiple(sa.ChargingAction, __abstractmethods__=set()) @patch.multiple(sa.DriftAction, __abstractmethods__=set()) @patch.multiple(sa.DesatAction, __abstractmethods__=set()) diff --git a/tests/unittest/envs/general_satellite_tasking/simulation/test_dynamics.py b/tests/unittest/envs/general_satellite_tasking/simulation/test_dynamics.py index 8ab166ae..c57d4077 100644 --- a/tests/unittest/envs/general_satellite_tasking/simulation/test_dynamics.py +++ b/tests/unittest/envs/general_satellite_tasking/simulation/test_dynamics.py @@ -7,6 +7,7 @@ from bsk_rl.envs.general_satellite_tasking.simulation import environment from bsk_rl.envs.general_satellite_tasking.simulation.dynamics import ( BasicDynamicsModel, + ContinuousImagingDynModel, DynamicsModel, GroundStationDynModel, ImagingDynModel, @@ -256,3 +257,15 @@ def test_init_objects(self, *args): GroundStationDynModel(MagicMock(simulator=MagicMock()), 1.0) for setter in args: setter.assert_called_once() + + +@patch(imdyn + "requires_env", MagicMock(return_value=[])) +@patch(imdyn + "_init_dynamics_objects", MagicMock()) +class TestContinuousImagingDynModel: + def test_storage_properties(self): + dyn = ContinuousImagingDynModel(MagicMock(simulator=MagicMock()), 1.0) + dyn.storageUnit = MagicMock() + dyn.storageUnit.storageUnitDataOutMsg.read.return_value.storageLevel = 50.0 + dyn.storageUnit.storageCapacity = 100.0 + assert dyn.storage_level == 50.0 + assert dyn.storage_level_fraction == 0.5