Merge branch '223-isolation-access' into 'master'

Isolation Access

Closes #233

See merge request barkhauseninstitut/wicon/hermespy!210

hermespy/core/__init__.py

@@ -65,7 +65,7 @@
     register,
 )
 from .random_node import RandomRealization, RandomNode
-from .drop import Drop, RecalledDrop
+from .drop import Drop
 from .scenario import Scenario, ScenarioMode, ScenarioType, ReplayScenario
 from .signal_model import Signal, SignalBlock, DenseSignal, SparseSignal
 from .visualize import (
@@ -168,7 +168,6 @@
     "RandomRealization",
     "RandomNode",
     "Drop",
-    "RecalledDrop",
     "Scenario",
     "ScenarioMode",
     "ScenarioType",

hermespy/core/device.py

@@ -406,6 +406,10 @@ def to_HDF(self, group: Group) -> None:
         self.mixed_signal.to_HDF(self._create_group(group, "mixed_signal"))
 
 
+DTT = TypeVar("DTT", bound="DeviceTransmission")
+"""Type of device transmission."""
+
+
 class DeviceTransmission(DeviceOutput):
     """Information generated by transmitting over a device."""
 
@@ -465,17 +469,38 @@ def num_operator_transmissions(self) -> int:
         return len(self.__operator_transmissions)
 
     @classmethod
-    def from_HDF(cls: Type[DeviceTransmission], group: Group) -> DeviceTransmission:
+    def from_HDF(
+        cls: Type[DeviceTransmission], group: Group, operators: Sequence[Transmitter] | None = None
+    ) -> DeviceTransmission:
+        """Recall a device transmission from a serialization.
+
+        Args:
+
+            group (Group):
+                HDF5 group containing the serialized device transmission.
+
+            operators (Sequence[Transmitter], optional):
+                List of device transmitters to recall the specific transmissions.
+                If not specified, the transmissions are recalled as their base class.
+        """
+
         # Recall base class
         device_output = DeviceOutput.from_HDF(group)
 
         # Recall attributes
         num_transmissions = group.attrs.get("num_transmissions", 1)
 
         # Recall transmissions
-        transmissions = [
-            Transmission.from_HDF(group[f"transmission_{t:02d}"]) for t in range(num_transmissions)
-        ]
+        if operators is None:
+            transmissions = [
+                Transmission.from_HDF(group[f"transmission_{t:02d}"])
+                for t in range(num_transmissions)
+            ]
+        else:
+            transmissions = [
+                operator.recall_transmission(group[f"transmission_{t:02d}"])
+                for t, operator in zip(range(num_transmissions), operators)
+            ]
 
         # Initialize object
         return cls.From_Output(device_output, transmissions)
@@ -706,20 +731,26 @@ def num_operator_receptions(self) -> int:
         return len(self.__operator_receptions)
 
     @classmethod
-    def from_HDF(cls: Type[DRT], group: Group) -> DRT:
+    def from_HDF(cls: Type[DRT], group: Group, operators: Sequence[Receiver] | None = None) -> DRT:
         # Recall base class
         device_input = ProcessedDeviceInput.from_HDF(group)
 
         # Recall individual operator receptions
         num_receptions = group.attrs.get("num_operator_receptions", 0)
 
-        # Recall operator receptions
-        operator_receptions = [
-            Reception.from_HDF(group[f"reception_{f:02d}"]) for f in range(num_receptions)
-        ]
+        # Recall receptions
+        if operators is None:
+            receptions = [
+                Reception.from_HDF(group[f"reception_{r:02d}"]) for r in range(num_receptions)
+            ]
+        else:
+            receptions = [
+                operator.recall_reception(group[f"reception_{r:02d}"])
+                for r, operator in zip(range(num_receptions), operators)
+            ]
 
         # Initialize object
-        return cls.From_ProcessedDeviceInput(device_input, operator_receptions)
+        return cls.From_ProcessedDeviceInput(device_input, receptions)
 
     @classmethod
     def Recall(cls: Type[DRT], group: Group, device: Device) -> DRT:
@@ -1809,6 +1840,21 @@ def transmit(self, clear_cache: bool = True) -> DeviceTransmission:
 
         return DeviceTransmission.From_Output(device_output, operator_transmissions)
 
+    def recall_transmission(self, group: Group) -> DeviceTransmission:
+        """Recall a specific transmission from a HDF5 serialization.
+
+        Args:
+
+            group (Group):
+                HDF group containing the transmission.
+
+        Returns: The recalled transmission.
+        """
+
+        # Recall the specific operator transmissions
+
+        return DeviceTransmission.from_HDF(group, list(self.transmitters))
+
     def cache_transmission(self, transmission: DeviceTransmission) -> None:
         for transmitter, operator_transmission in zip(
             self.transmitters, transmission.operator_transmissions
@@ -1953,3 +1999,16 @@ def receive(
 
         # Generate device reception
         return DeviceReception.From_ProcessedDeviceInput(processed_input, receptions)
+
+    def recall_reception(self, group: Group) -> DeviceReception:
+        """Recall a specific reception from a HDF5 serialization.
+
+        Args:
+
+            group (Group):
+                HDF group containing the reception.
+
+        Returns: The recalled reception.
+        """
+
+        return DeviceReception.Recall(group, self)

hermespy/core/drop.py

@@ -7,18 +7,15 @@
 
 from __future__ import annotations
 from collections.abc import Sequence
-from typing import Type, TYPE_CHECKING
+from typing import Generic, Type, TypeVar
 
 from h5py import Group
 
-from .device import DeviceReception, DeviceTransmission
+from .device import Device, DeviceReception, DeviceTransmission, DRT, DTT
 from .factory import HDFSerializable
 from .signal_model import Signal
 from .monte_carlo import Artifact
 
-if TYPE_CHECKING:
-    from .scenario import Scenario  # pragma: no cover
-
 __author__ = "Jan Adler"
 __copyright__ = "Copyright 2024, Barkhausen Institut gGmbH"
 __credits__ = ["Jan Adler"]
@@ -29,30 +26,30 @@
 __status__ = "Prototype"
 
 
-class Drop(HDFSerializable):
+class Drop(Generic[DTT, DRT], HDFSerializable):
     """Drop containing the information transmitted and received by all devices
     within a scenario."""
 
     __timestamp: float  # Time at which the drop was generated
-    __device_transmissions: Sequence[DeviceTransmission]  # Transmitted device information
-    __device_receptions: Sequence[DeviceReception]  # Received device information
+    __device_transmissions: Sequence[DTT]  # Transmitted device information
+    __device_receptions: Sequence[DRT]  # Received device information
 
     def __init__(
         self,
         timestamp: float,
-        device_transmissions: Sequence[DeviceTransmission],
-        device_receptions: Sequence[DeviceReception],
+        device_transmissions: Sequence[DTT],
+        device_receptions: Sequence[DRT],
     ) -> None:
         """
         Args:
 
             timestamp (float):
                 Time at which the drop was generated.
 
-            device_transmissions (Sequence[DeviceTransmission]):
+            device_transmissions (Sequence[DTT]):
                 Transmitted device information.
 
-            device_receptions (Sequence[DeviceReception]):
+            device_receptions (Sequence[DRT]):
                 Received device information.
         """
 
@@ -67,13 +64,13 @@ def timestamp(self) -> float:
         return self.__timestamp
 
     @property
-    def device_transmissions(self) -> Sequence[DeviceTransmission]:
+    def device_transmissions(self) -> Sequence[DTT]:
         """Transmitted device information within this drop."""
 
         return self.__device_transmissions
 
     @property
-    def device_receptions(self) -> Sequence[DeviceReception]:
+    def device_receptions(self) -> Sequence[DRT]:
         """Received device information within this drop."""
 
         return self.__device_receptions
@@ -100,19 +97,29 @@ def operator_inputs(self) -> Sequence[Sequence[Signal]]:
         return [reception.operator_inputs for reception in self.device_receptions]
 
     @classmethod
-    def from_HDF(cls: Type[Drop], group: Group) -> Drop:
+    def from_HDF(cls: Type[Drop], group: Group, devices: Sequence[Device] | None = None) -> Drop:
         # Recall attributes
         timestamp = group.attrs.get("timestamp", 0.0)
-        num_transmissions = group.attrs.get("num_transmissions", 0)
-        num_receptions = group.attrs.get("num_receptions", 0)
-
-        transmissions = [
-            DeviceTransmission.from_HDF(group[f"transmission_{t:02d}"])
-            for t in range(num_transmissions)
-        ]
-        receptions = [
-            DeviceReception.from_HDF(group[f"reception_{r:02d}"]) for r in range(num_receptions)
-        ]
+        num_transmissions: int = group.attrs.get("num_transmissions", 0)
+        num_receptions: int = group.attrs.get("num_receptions", 0)
+
+        if devices is None:
+            transmissions = [
+                DeviceTransmission.from_HDF(group[f"transmission_{t:02d}"])
+                for t in range(num_transmissions)
+            ]
+            receptions = [
+                DeviceReception.from_HDF(group[f"reception_{r:02d}"]) for r in range(num_receptions)
+            ]
+        else:
+            transmissions = [
+                device.recall_transmission(group[f"transmission_{t:02d}"])
+                for t, device in zip(range(num_transmissions), devices)
+            ]
+            receptions = [
+                device.recall_reception(group[f"reception_{r:02d}"])
+                for r, device in zip(range(num_receptions), devices)
+            ]
 
         drop = cls(
             timestamp=timestamp, device_transmissions=transmissions, device_receptions=receptions
@@ -133,46 +140,8 @@ def to_HDF(self, group: Group) -> None:
         group.attrs["num_receptions"] = self.num_device_receptions
 
 
-class RecalledDrop(Drop):
-    """Drop recalled from serialization containing the information transmitted and received by all devices
-    within a scenario."""
-
-    __group: Group
-
-    def __init__(self, group: Group, scenario: Scenario) -> None:
-        # Recall attributes
-        timestamp = group.attrs.get("timestamp", 0.0)
-        num_transmissions = group.attrs.get("num_transmissions", 0)
-        num_receptions = group.attrs.get("num_receptions", 0)
-
-        device_transmissions = [
-            DeviceTransmission.Recall(group[f"transmission_{t:02d}"], device)
-            for t, device in zip(range(num_transmissions), scenario.devices)
-        ]
-        device_receptions = [
-            DeviceReception.Recall(group[f"reception_{r:02d}"], device)
-            for r, device in zip(range(num_receptions), scenario.devices)
-        ]
-
-        # Initialize base class
-        Drop.__init__(
-            self,
-            timestamp=timestamp,
-            device_transmissions=device_transmissions,
-            device_receptions=device_receptions,
-        )
-
-        # Initialize class attributes
-        self.__group = group
-
-    @property
-    def group(self) -> Group:
-        """HDF group this drop was recalled from.
-
-        Returns: Handle to an HDF group.
-        """
-
-        return self.__group
+DropType = TypeVar("DropType", bound=Drop)
+"""Type of a drop."""
 
 
 class EvaluatedDrop(Drop):

hermespy/core/scenario.py

@@ -16,6 +16,7 @@
 from h5py import File, Group
 
 from .device import (
+    Device,
     DeviceInput,
     DeviceOutput,
     DeviceReception,
@@ -28,7 +29,7 @@
     Receiver,
     Operator,
 )
-from .drop import Drop, RecalledDrop
+from .drop import Drop, DropType
 from .factory import Factory
 from .random_node import RandomNode
 from .signal_model import Signal
@@ -66,7 +67,7 @@ class ScenarioMode(IntEnum):
     """
 
 
-class Scenario(ABC, RandomNode, TransformableBase, Generic[DeviceType]):
+class Scenario(ABC, RandomNode, TransformableBase, Generic[DeviceType, DropType]):
     """A wireless scenario.
 
     Scenarios consist of several devices transmitting and receiving electromagnetic signals.
@@ -857,7 +858,7 @@ def num_drops(self) -> int | None:
         return None
 
     @abstractmethod
-    def _drop(self) -> Drop:
+    def _drop(self) -> DropType:
         """Generate a single scenario drop.
 
         Wrapped by the scenario base class :meth:`.drop` method.
@@ -867,22 +868,33 @@ def _drop(self) -> Drop:
         """
         ...  # pragma no cover
 
-    def drop(self) -> Drop:
+    @abstractmethod
+    def _recall_drop(self, group: Group) -> DropType:
+        """Recall a recorded drop from a HDF5 group.
+
+        Args:
+
+            group (Group):
+                HDF5 group containing the drop information.
+
+        Returns: The recalled drop.
+        """
+        ...  # pragma no cover
+
+    def drop(self) -> DropType:
         """Generate a single data drop from all scenario devices.
 
         Return: The generated drop information.
         """
 
-        drop: Drop
-
         if self.mode == ScenarioMode.REPLAY:
             # Recall the drop from the savefile
             for _ in range(self.__file.attrs["num_drops"]):
                 drop_path = f"/campaigns/{self.__campaign}/drop_{self.__drop_counter:02d}"
                 self.__drop_counter = (self.__drop_counter + 1) % self.__file.attrs["num_drops"]
 
                 if drop_path in self.__file:
-                    drop = RecalledDrop(self.__file[drop_path], self)
+                    drop = self._recall_drop(self.__file[drop_path])
                     break
 
             # Replay device operator transmissions
@@ -912,8 +924,11 @@ def drop(self) -> Drop:
 """Type of scenario."""
 
 
-class ReplayScenario(Scenario):
+class ReplayScenario(Scenario[Device, Drop]):
     """Scenario which is unable to generate drops."""
 
     def _drop(self) -> Drop:
         raise RuntimeError("Replay scenario may not generate data drops.")
+
+    def _recall_drop(self, group: Group) -> Drop:
+        return Drop.from_HDF(group)