Merge branch '211-constellation-evm' into 'master'

Constellation EVM Evaluation

Closes #211

See merge request barkhauseninstitut/wicon/hermespy!172

docssource/api/modem.evaluators._table.rst

@@ -1,8 +1,9 @@
-================================== ========================================================
+================================== =================================================================
 Performance Indicator              Performance Indicator
-================================== ========================================================
+================================== =================================================================
 :doc:`modem.evaluators.ber`        Errors comparing two bit streams
 :doc:`modem.evaluators.bler`       Errors comparing two bit streams divided into blocks
 :doc:`modem.evaluators.fer`        Errors comparing two bit streams divided into frames
 :doc:`modem.evaluators.throughput` Rate of correct frames multiplied by the frame bit rate
-================================== ========================================================
+:doc:`modem.evaluators.evm`        Root mean square error between transmitted and received symbols
+================================== =================================================================

docssource/api/modem.evaluators.evm.rst

@@ -0,0 +1,45 @@
+======================
+Error Vector Magnitude
+======================
+
+.. inheritance-diagram:: hermespy.modem.evaluators.ConstellationEVM hermespy.modem.evaluators.EVMArtifact hermespy.modem.evaluators.EVMEvaluation
+   :parts: 1
+   :top-classes: hermespy.core.monte_carlo.Evaluator, hermespy.core.monte_carlo.Evaluation, hermespy.core.monte_carlo.Artifact
+
+Considering two linked modems denoted by :math:`(\alpha)` and :math:`(\beta)`,
+with modem :math:`(\alpha)` transmitting a symbol sequence
+
+.. math::
+
+   \mathbf{s}_{\mathrm{Tx}}^{(\alpha)} = \left[ s_{\mathrm{Tx}}^{(\alpha,1)}, s_{\mathrm{Tx}}^{(\alpha,2)}, \ldots, s_{\mathrm{Tx}}^{(\alpha,B)} \right]^{\mathsf{T}} \in \mathbb{C}^{S}
+   
+and modem :math:`(\beta)` receiving a decoded symbol sequence
+
+.. math::
+
+   \mathbf{s}_{\mathrm{Rx}}^{(\beta)} = \left[ s_{\mathrm{Rx}}^{(\beta,1)}, s_{\mathrm{rx}}^{(\beta,2)}, \ldots, s_{\mathrm{Rx}}^{(\beta,B)} \right]^{\mathsf{T}} \in \mathbb{C}^{S}
+
+Hermes defines the symbol Error Vector Magnitude (EVM) as the root mean square (RMS) of the difference between the transmitted and received symbols
+
+.. math::
+
+   \mathrm{EVM}^{(\alpha,\beta)} = \sqrt{\frac{ \lVert \mathbf{s}_{\mathrm{Tx}}^{(\alpha)} - \mathbf{s}_{\mathrm{Rx}}^{(\beta)} \rVert_2^2 }{S}} \ \text{.}
+
+In practice, the number of symbols :math:`S` may differ between transmitter and receiver.
+In this case, the longer sequence is truncated to the length of the shorter sequence.
+
+The following minimal examples outlines how to configure this evaluator
+within the context of a simulation campaign:
+
+.. literalinclude:: ../scripts/examples/modem_evaluators_evm.py
+   :language: python
+   :linenos:
+   :lines: 10-31
+
+.. autoclass:: hermespy.modem.evaluators.ConstellationEVM
+
+.. autoclass:: hermespy.modem.evaluators.EVMArtifact
+
+.. autoclass:: hermespy.modem.evaluators.EVMEvaluation
+
+.. footbibliography::

docssource/api/modem.evaluators.rst

@@ -93,10 +93,30 @@ communication links between modems.
       +str title
    }
 
+   class EVMArtifact {
+
+      +float artifact
+      +to_scalar() float
+   }
+
+   class EVMEvaluation {
+
+      +str title
+      +artifact() EVMArtifact
+   }
+
+   class ConstellationEVM {
+
+      +evaluate() EVMEvaluation
+      +str abbreviation
+      +str title
+   }
+
    BitErrorEvaluator --|> CommunicationEvaluator
    BlockErrorEvaluator --|> CommunicationEvaluator
    FrameErrorEvaluator --|> CommunicationEvaluator
    ThroughputEvaluator --|> CommunicationEvaluator
+   ConstellationEVM --|> CommunicationEvaluator
 
    BitErrorEvaluator --> BitErrorArtifact : create
    BitErrorEvaluator --> BitErrorEvaluation : create
@@ -106,6 +126,8 @@ communication links between modems.
    FrameErrorEvaluator --> FrameErrorEvaluation : create
    ThroughputEvaluator --> ThroughputArtifact : create
    ThroughputEvaluator --> ThroughputEvaluation : create
+   ConstellationEVM --> EVMEvaluation : create
+   EVMEvaluation --> EVMArtifact : create
 
    link CommunicationEvaluator "modem.evaluators.CommunicationEvaluator.html"
    link BitErrorArtifact "modem.evaluators.BitErrorArtifact.html"
@@ -120,7 +142,9 @@ communication links between modems.
    link ThroughputArtifact "modem.evaluators.ThroughputArtifact.html"
    link ThroughputEvaluation "modem.evaluators.ThroughputEvaluation.html"
    link ThroughputEvaluator "modem.evaluators.ThroughputEvaluator.html"
-
+   link EVMArtifact "modem.evaluators.evm.html#hermespy.modem.evaluators.evm.EVMArtifact"
+   link EVMEvaluation "modem.evaluators.evm.html#hermespy.modem.evaluators.evm.EVMEvaluation"
+   link ConstellationEVM "modem.evaluators.evm.html#hermespy.modem.evaluators.evm.ConstellationEVM"
 
 The implemented :doc:`Communication Evaluators<modem.evaluators.CommunicationEvaluator>` all inherit from the identically named common
 base which gets initialized by selecting the two :doc:`Modem<modem.modem.BaseModem>` instances whose communication
@@ -154,6 +178,7 @@ Configuring :doc:`Communication Evaluators<modem.evaluators.CommunicationEvaluat
    modem.evaluators.bler
    modem.evaluators.fer
    modem.evaluators.throughput
+   modem.evaluators.evm
    modem.evaluators.CommunicationEvaluator
 
 .. footbibliography::

docssource/scripts/examples/modem_evaluators_evm.py

@@ -0,0 +1,34 @@
+# -*- coding: utf-8 -*-
+
+import matplotlib.pyplot as plt
+
+from hermespy.core import dB
+from hermespy.modem import ConstellationEVM, TransmittingModem, ReceivingModem, RootRaisedCosineWaveform
+from hermespy.simulation import Simulation
+
+
+# Create a new simulation featuring two devices
+simulation = Simulation()
+device_alpha = simulation.new_device()
+device_beta = simulation.new_device()
+
+# Create a transmitting and receiving modem for each device, respectively
+modem_alpha = TransmittingModem(device=device_alpha)
+modem_beta = ReceivingModem(device=device_beta)
+
+# Configure the modem's waveform
+waveform_configuration = {
+    'symbol_rate': 1e8,
+    'num_preamble_symbols': 10,
+    'num_data_symbols': 100,
+}
+modem_alpha.waveform = RootRaisedCosineWaveform(**waveform_configuration)
+modem_beta.waveform = RootRaisedCosineWaveform(**waveform_configuration)
+
+simulation.add_evaluator(ConstellationEVM(modem_alpha, modem_beta))
+simulation.new_dimension('snr', dB(0, 2, 4, 8, 10, 12, 14, 16, 18, 20), device_beta)
+simulation.num_samples = 1000
+result = simulation.run()
+
+result.plot()
+plt.show()

hermespy/channel/delay.py

@@ -315,8 +315,7 @@ def model_propagation_loss(self, value: bool) -> None:
         self.__model_propagation_loss = value
 
     @abstractmethod
-    def _init_realization(self, *args, **kwargs) -> DCRT:
-        ...  # pragma: no cover
+    def _init_realization(self, *args, **kwargs) -> DCRT: ...  # pragma: no cover
 
     def _realize(self) -> DCRT:
         delay = self._realize_delay()

hermespy/channel/radar_channel.py

@@ -912,9 +912,7 @@ def propagation_response(
 
             wavelength = speed_of_light / carrier_frequency
             amplitude_factor = (
-                wavelength
-                * self.cross_section**0.5
-                / ((4 * pi) ** 1.5 * tx_distance * rx_distance)
+                wavelength * self.cross_section**0.5 / ((4 * pi) ** 1.5 * tx_distance * rx_distance)
             )
 
         else:

hermespy/core/device.py

@@ -1245,12 +1245,10 @@ def min_num_samples_per_frame(self) -> int:
         return ceil(frame_duration * sampling_rate)
 
     @overload
-    def __getitem__(self, item: int) -> OperatorType:
-        ...  # pragma:  no cover
+    def __getitem__(self, item: int) -> OperatorType: ...  # pragma:  no cover
 
     @overload
-    def __getitem__(self, item: slice) -> Sequence[OperatorType]:
-        ...  # pragma: no cover
+    def __getitem__(self, item: slice) -> Sequence[OperatorType]: ...  # pragma: no cover
 
     def __getitem__(self, item: int | slice) -> OperatorType | Sequence[OperatorType]:
         return self.__operators[item]

hermespy/core/duplex.py

@@ -83,31 +83,24 @@ def slot(self, value: OperatorSlot[Transmitter]) -> None:
         Transmitter.slot.fset(self, value)
 
     @abstractmethod
-    def _transmit(self, duration: float = 0.0) -> TransmissionType:
-        ...  # pragma: no cover
+    def _transmit(self, duration: float = 0.0) -> TransmissionType: ...  # pragma: no cover
 
     @abstractmethod
-    def _receive(self, signal: Signal) -> ReceptionType:
-        ...  # pragma: no cover
+    def _receive(self, signal: Signal) -> ReceptionType: ...  # pragma: no cover
 
     @property
     @abstractmethod
-    def sampling_rate(self) -> float:
-        ...  # pragma: no cover
+    def sampling_rate(self) -> float: ...  # pragma: no cover
 
     @property
     @abstractmethod
-    def frame_duration(self) -> float:
-        ...  # pragma: no cover
+    def frame_duration(self) -> float: ...  # pragma: no cover
 
     @abstractmethod
-    def _noise_power(self, strength: float, snr_type: SNRType) -> float:
-        ...  # pragma: no cover
+    def _noise_power(self, strength: float, snr_type: SNRType) -> float: ...  # pragma: no cover
 
     @abstractmethod
-    def _recall_transmission(self, group: Group) -> TransmissionType:
-        ...  # pragma: no cover
+    def _recall_transmission(self, group: Group) -> TransmissionType: ...  # pragma: no cover
 
     @abstractmethod
-    def _recall_reception(self, group: Group) -> ReceptionType:
-        ...  # pragma: no cover
+    def _recall_reception(self, group: Group) -> ReceptionType: ...  # pragma: no cover

hermespy/core/logarithmic.py

@@ -393,15 +393,13 @@ def __reduce__(
 @overload
 def dB(
     *values: Sequence[Union[int, float]], conversion: DbConversionType = DbConversionType.POWER
-) -> LogarithmicSequence:
-    ...  # pragma no cover
+) -> LogarithmicSequence: ...  # pragma no cover
 
 
 @overload
 def dB(
     *values: Union[int, float], conversion: DbConversionType = DbConversionType.POWER
-) -> Union[Logarithmic, LogarithmicSequence]:
-    ...  # pragma no cover
+) -> Union[Logarithmic, LogarithmicSequence]: ...  # pragma no cover
 
 
 def dB(

hermespy/core/scenario.py

@@ -687,20 +687,17 @@ def transmit_devices(self) -> Sequence[DeviceTransmission]:
     @overload
     def process_inputs(
         self, impinging_signals: Sequence[DeviceInput], cache: bool = True
-    ) -> Sequence[ProcessedDeviceInput]:
-        ...  # pragma: no cover
+    ) -> Sequence[ProcessedDeviceInput]: ...  # pragma: no cover
 
     @overload
     def process_inputs(
         self, impinging_signals: Sequence[Signal], cache: bool = True
-    ) -> Sequence[ProcessedDeviceInput]:
-        ...  # pragma: no cover
+    ) -> Sequence[ProcessedDeviceInput]: ...  # pragma: no cover
 
     @overload
     def process_inputs(
         self, impinging_signals: Sequence[Sequence[Signal]], cache: bool = True
-    ) -> Sequence[ProcessedDeviceInput]:
-        ...  # pragma: no cover
+    ) -> Sequence[ProcessedDeviceInput]: ...  # pragma: no cover
 
     def process_inputs(
         self,
@@ -738,18 +735,15 @@ def process_inputs(
     @overload
     def receive_operators(
         self, operator_inputs: Sequence[ProcessedDeviceInput], cache: bool = True
-    ) -> Sequence[Sequence[Reception]]:
-        ...  # pragma: no cover
+    ) -> Sequence[Sequence[Reception]]: ...  # pragma: no cover
 
     @overload
     def receive_operators(
         self, operator_inputs: Sequence[Sequence[Signal]], cache: bool = True
-    ) -> Sequence[Sequence[Reception]]:
-        ...  # pragma: no cover
+    ) -> Sequence[Sequence[Reception]]: ...  # pragma: no cover
 
     @overload
-    def receive_operators(self) -> Sequence[Sequence[Reception]]:
-        ...  # pragma: no cover
+    def receive_operators(self) -> Sequence[Sequence[Reception]]: ...  # pragma: no cover
 
     def receive_operators(
         self,
@@ -792,20 +786,17 @@ def receive_operators(
     @overload
     def receive_devices(
         self, impinging_signals: Sequence[DeviceInput], cache: bool = True
-    ) -> Sequence[DeviceReception]:
-        ...  # pragma: no cover
+    ) -> Sequence[DeviceReception]: ...  # pragma: no cover
 
     @overload
     def receive_devices(
         self, impinging_signals: Sequence[Signal], cache: bool = True
-    ) -> Sequence[DeviceReception]:
-        ...  # pragma: no cover
+    ) -> Sequence[DeviceReception]: ...  # pragma: no cover
 
     @overload
     def receive_devices(
         self, impinging_signals: Sequence[Sequence[Signal]], cache: bool = True
-    ) -> Sequence[DeviceReception]:
-        ...  # pragma: no cover
+    ) -> Sequence[DeviceReception]: ...  # pragma: no cover
 
     def receive_devices(
         self,

hermespy/core/signal_model.py

@@ -336,9 +336,9 @@ def _update_visualization(
             values = np.empty(num_visualized_symbols * values_per_symbol, dtype=np.complex_)
             for n in range(num_symbols - 2 * num_cutoff_symbols):
                 sample_offset = (n + num_cutoff_symbols) * symbol_num_samples
-                values[
-                    n * values_per_symbol : (n + 1) * values_per_symbol - 1
-                ] = self.signal.samples[0, sample_offset : sample_offset + values_per_symbol - 1]
+                values[n * values_per_symbol : (n + 1) * values_per_symbol - 1] = (
+                    self.signal.samples[0, sample_offset : sample_offset + values_per_symbol - 1]
+                )
             values[values_per_symbol - 1 :: values_per_symbol] = float("nan") + 1j * float("nan")
 
             # Normalize values

hermespy/core/transformation.py

@@ -571,18 +571,19 @@ def backwards_transformation(self) -> Transformation:
         return np.linalg.inv(self.forwards_transformation).view(Transformation)
 
     @overload
-    def to_local_coordinates(self, global_object: Transformable) -> Transformation:
-        ...  # pragma no cover
+    def to_local_coordinates(
+        self, global_object: Transformable
+    ) -> Transformation: ...  # pragma no cover
 
     @overload
-    def to_local_coordinates(self, global_object: Transformation) -> Transformation:
-        ...  # pragma no cover
+    def to_local_coordinates(
+        self, global_object: Transformation
+    ) -> Transformation: ...  # pragma no cover
 
     @overload
     def to_local_coordinates(
         self, position: np.ndarray, orientation: np.ndarray | None = None
-    ) -> Transformation:
-        ...  # pragma no cover
+    ) -> Transformation: ...  # pragma no cover
 
     def to_local_coordinates(self, arg_0: Transformable | Transformation | np.ndarray, arg_1: np.ndarray | None = None) -> Transformation:  # type: ignore
         if isinstance(arg_0, Transformable):

hermespy/fec/coding.py

@@ -449,9 +449,9 @@ def encode(self, data_bits: np.ndarray, num_code_bits: Optional[int] = None) ->
                 encoded_block = encoder.encode(
                     data_state[block_idx * data_block_size : (1 + block_idx) * data_block_size]
                 )
-                code_state[
-                    block_idx * code_block_size : (1 + block_idx) * code_block_size
-                ] = encoded_block
+                code_state[block_idx * code_block_size : (1 + block_idx) * code_block_size] = (
+                    encoded_block
+                )
 
         if num_code_bits and len(code_state) > num_code_bits:
             raise RuntimeError(
@@ -543,10 +543,10 @@ def decode(self, encoded_bits: np.ndarray, num_data_bits: Optional[int] = None)
 
             # Decode all blocks sequentially
             for block_idx in range(num_blocks):
-                data_state[
-                    block_idx * data_block_size : (1 + block_idx) * data_block_size
-                ] = encoder.decode(
-                    code_state[block_idx * code_block_size : (1 + block_idx) * code_block_size]
+                data_state[block_idx * data_block_size : (1 + block_idx) * data_block_size] = (
+                    encoder.decode(
+                        code_state[block_idx * code_block_size : (1 + block_idx) * code_block_size]
+                    )
                 )
 
         # Return resulting data

hermespy/hardware_loop/physical_device.py

@@ -572,8 +572,9 @@ def __init__(self, device: PhysicalDevice | None = None) -> None:
 
     @classmethod
     @abstractmethod
-    def _configure_slot(cls: Type[CT], device: PhysicalDevice, value: CT | None) -> None:
-        ...  # pragma: no cover
+    def _configure_slot(
+        cls: Type[CT], device: PhysicalDevice, value: CT | None
+    ) -> None: ...  # pragma: no cover
 
     @property
     def device(self) -> PhysicalDevice | None: