Merge branch '25-signal-block' into 'master'

Optimize Signals With Null Sections

Closes #25

See merge request barkhauseninstitut/wicon/hermespy!179

_examples/hybrid_beamforming.py

@@ -57,7 +57,7 @@
     StreetCanyonLineOfSight(),
 )
 
-test_signal = Signal(np.outer(np.ones(1), np.exp(2j * pi * .25 * np.arange(100))), sampling_rate, carrier_frequency)
+test_signal = Signal.Create(np.outer(np.ones(1), np.exp(2j * pi * .25 * np.arange(100))), sampling_rate, carrier_frequency)
 beamformer = ConventionalBeamformer()
 base_station_transmitter = BeamformingTransmitter(test_signal, beamformer)
 base_station_transmitter.device = base_station_device

_examples/matlab/getting_started_example1.m

@@ -48,9 +48,9 @@
 % however this method is not guarateed to be sufficient always.
 signal = signal_tuple{1};
 
-% we are interested in the variable "signal.samples" which is a ndarray.
+% we are interested in the samples of a signal "signal[:, :]" which is a ndarray.
 % the function "python2matlab" converts ndarray to double
-signal_samples = python2matlab(signal.samples);
+signal_samples = python2matlab(signal[:, :]);
 
 % PLOTS
 figure

_examples/matlab/getting_started_example3.m

@@ -65,15 +65,15 @@
 input_signal_py.resize(py.int(1),py.int(num_samples)); 
 
 % create a Signal object
-signal = core.Signal(input_signal_py, py.int(transmitter.sampling_rate));
+signal = core.Signal.Create(input_signal_py, py.int(transmitter.sampling_rate));
 
 % propate input through the channel
 output_tupple = radar_channel.propagate(signal);
 
 % retrieve the out samples
 output_list = output_tupple{1};
 output_cell = output_list.cell{1};
-output = python2matlab(output_cell.samples);
+output = python2matlab(output_cell[:, :]);
 
 % plot received signal after the radar channel
 figure

docssource/notebooks/beamforming_usage.ipynb

@@ -186,7 +186,7 @@
    "source": [
     "# Configure the device to transmit a beamformed sinusoidal probing signal\n",
     "beamformer = ConventionalBeamformer()\n",
-    "sinusoid = Signal(np.outer(np.ones(1), np.exp(2j * pi * .25 * np.arange(100))), sampling_rate, carrier_frequency)\n",
+    "sinusoid = Signal.Create(np.outer(np.ones(1), np.exp(2j * pi * .25 * np.arange(100))), sampling_rate, carrier_frequency)\n",
     "base_station_transmitter = BeamformingTransmitter(sinusoid, beamformer)\n",
     "base_station_transmitter.device = base_station_device"
    ]

docssource/notebooks/channel.ipynb

@@ -70,8 +70,8 @@
                 "        interpolation: InterpolationMode,\n",
                 "     ) -> Signal:\n",
                 "\n",
-                "        shifted_samples = signal.samples * np.exp(1j * self.__phase_shift)\n",
-                "        return Signal(shifted_samples, signal.sampling_rate, signal.carrier_frequency)\n",
+                "        shifted_samples = signal[:, :] * np.exp(1j * self.__phase_shift)\n",
+                "        return signal.from_ndarray(shifted_samples)\n",
                 "\n",
                 "    def state(\n",
                 "        self,\n",

docssource/notebooks/waveform.ipynb

@@ -200,7 +200,7 @@
     "\n",
     "# Generate the base-band representation of a single communication frame\n",
     "baseband_transmission = ask.modulate(ask.place(mapped_symbols))\n",
-    "_ = Signal(baseband_transmission, ask.sampling_rate).plot()"
+    "_ = Signal.Create(baseband_transmission, ask.sampling_rate).plot()"
    ]
   },
   {
@@ -256,7 +256,7 @@
     "\n",
     "# Compute the baseband signal received over an AWGN channel\n",
     "baseband_reception = baseband_transmission + noise\n",
-    "_ = Signal(baseband_reception, ask.sampling_rate).plot()\n",
+    "_ = Signal.Create(baseband_reception, ask.sampling_rate).plot()\n",
     "\n",
     "# Demodulate the received signal\n",
     "received_symbols = ask.demodulate(baseband_reception)\n",

docssource/scripts/examples/simulation.py

@@ -36,7 +36,7 @@
 
 # Make both devices transmit 100 samples at 100 MHz
 ns, fs = 100, 1e8
-transmitted_signal = Signal(np.exp(2j * np.random.uniform(0, np.pi, (1, ns))), fs)
+transmitted_signal = Signal.Create(np.exp(2j * np.random.uniform(0, np.pi, (1, ns))), fs)
 alpha_transmitter = SignalTransmitter(transmitted_signal)
 beta_transmitter = SignalTransmitter(transmitted_signal)
 alpha_receiver = SignalReceiver(ns, fs, 1.0)

docssource/scripts/examples/simulation_SimulatedDevice.py

@@ -57,7 +57,7 @@
 device.velocity = np.array([1., 1., 0.], dtype=np.float_)
 
 # Transmit random white noise from the device
-transmitter = SignalTransmitter(Signal(
+transmitter = SignalTransmitter(Signal.Create(
     np.random.normal(size=(device.num_transmit_ports, 100)) +
     1j * np.random.normal(size=(device.num_transmit_ports, 100)),
     device.sampling_rate,
@@ -75,7 +75,7 @@
 transmission = device.transmit()
 
 # Receive a signal at the device
-impinging_signal = Signal(
+impinging_signal = Signal.Create(
     np.random.normal(size=(device.num_transmit_ports, 100)) +
     1j * np.random.normal(size=(device.num_transmit_ports, 100)),
     device.sampling_rate,

hermespy/beamforming/beamformer.py

@@ -512,15 +512,11 @@ def transmit(
             )
 
         carrier_frequency = signal.carrier_frequency
-        samples = signal.samples.copy()
+        samples = signal[:, :]
         focus_angles = np.array([focus.spherical_angles for focus in _focus], dtype=np.float_)
 
         steered_samples = self._encode(samples, carrier_frequency, focus_angles, _array)
-        return Signal(
-            steered_samples,
-            sampling_rate=signal.sampling_rate,
-            carrier_frequency=signal.carrier_frequency,
-        )
+        return signal.from_ndarray(steered_samples)
 
 
 class ReceiveBeamformer(BeamformerBase[Receiver], ReceiveStreamDecoder, ABC):
@@ -734,11 +730,11 @@ def receive(
             )
 
         carrier_frequency = signal.carrier_frequency
-        samples = signal.samples.copy()
+        samples = signal[:, :]
         focus_angles = np.array([[focus.spherical_angles for focus in _focus]], dtype=np.float_)
 
         beamformed_samples = self._decode(samples, carrier_frequency, focus_angles, _array)
-        return Signal(beamformed_samples[0, ::], signal.sampling_rate)
+        return signal.from_ndarray(beamformed_samples[0, ::])
 
     @property
     def probe_focus_points(self) -> np.ndarray:
@@ -807,6 +803,6 @@ def probe(self, signal: Signal, focus_points: np.ndarray | None = None) -> np.nd
             )
 
         carrier_frequency = signal.carrier_frequency
-        samples = signal.samples.copy()
+        samples = signal[:, :]
 
         return self._decode(samples, carrier_frequency, focus_points, self.operator.device.antennas)

hermespy/channel/channel.py

@@ -255,7 +255,9 @@ def propagate(
         propagated_signal = self._propagate(_signal, _transmitter, _receiver, _interpolation_mode)
 
         # Apply channel gain
-        propagated_signal.samples *= self.gain**0.5
+        gain_sqrt = self.gain**0.5
+        for block in propagated_signal:
+            block *= gain_sqrt
 
         # Return resulting channel propagation
         propagation = ChannelPropagation[CRT](

hermespy/channel/cluster_delay_lines.py

@@ -333,7 +333,7 @@ def _propagate(
         impulse = np.zeros(signal.num_samples, dtype=np.complex_)
         einsum_subscripts = "ij,jk,k->ik"
         einsum_path = np.einsum_path(
-            einsum_subscripts, channel, signal.samples, impulse, optimize="optimal"
+            einsum_subscripts, channel, signal[:, :], impulse, optimize="optimal"
         )[0]
 
         for channel, impulse, delay in self.__ray_impulse_generator(
@@ -348,16 +348,10 @@ def _propagate(
                 propagated_samples[
                     :, delay_in_samples : delay_in_samples + signal.num_samples
                 ] += np.einsum(
-                    einsum_subscripts, channel, signal.samples, impulse, optimize=einsum_path
+                    einsum_subscripts, channel, signal[:, :], impulse, optimize=einsum_path
                 )
 
-        return Signal(
-            propagated_samples,
-            signal.sampling_rate,
-            signal.carrier_frequency,
-            signal.delay,
-            signal.noise_power,
-        )
+        return signal.from_ndarray(propagated_samples)
 
     def state(
         self,

hermespy/channel/delay.py

@@ -140,13 +140,10 @@ def _propagate(
 
         propagated_samples = np.append(
             np.zeros((receiver.antennas.num_receive_antennas, delay_samples), np.complex_),
-            gain_factor * spatial_response @ signal.samples,
+            gain_factor * spatial_response @ signal[:, :],
         )
 
-        propagated_signal = Signal(
-            propagated_samples, signal.sampling_rate, signal.carrier_frequency
-        )
-        return propagated_signal
+        return signal.from_ndarray(propagated_samples)
 
     def state(
         self,

hermespy/channel/ideal.py

@@ -67,20 +67,20 @@ def _propagate(
         # Single antenna transmitter case
         if transmitter.num_transmit_antennas == 1:
             propagated_samples = np.repeat(
-                signal.samples[[0], :], receiver.num_receive_antennas, axis=0
+                signal[[0], :], receiver.num_receive_antennas, axis=0
             )
 
         # Single antenna receiver case
         elif receiver.num_receive_antennas == 1:
-            propagated_samples = np.sum(signal.samples, axis=0, keepdims=True)
+            propagated_samples = np.sum(signal[:, :], axis=0, keepdims=True)
 
         # No receiving antenna case
         elif receiver.num_receive_antennas == 0:
             propagated_samples = np.empty((0, signal.num_samples), dtype=np.complex_)
 
         # MIMO case
         else:
-            propagated_samples = signal.samples
+            propagated_samples = signal[:, :]
             if transmitter.num_transmit_antennas < receiver.num_receive_antennas:
                 propagated_samples = np.append(
                     propagated_samples,
@@ -94,13 +94,7 @@ def _propagate(
                     axis=0,
                 )
 
-        return Signal(
-            propagated_samples,
-            signal.sampling_rate,
-            signal.carrier_frequency,
-            signal.delay,
-            signal.noise_power,
-        )
+        return signal.from_ndarray(propagated_samples)
 
 
 class IdealChannel(Channel[IdealChannelRealization]):

hermespy/channel/multipath_fading_channel.py

@@ -435,7 +435,7 @@ def propagate(self, signal: Signal) -> np.ndarray:
         impulse_response = self._impulse_response(signal.timestamps)
 
         # Propagate the transmitted samples
-        propagated_samples = signal.samples * impulse_response[np.newaxis, :]
+        propagated_samples = signal[:, :] * impulse_response[np.newaxis, :]
         return propagated_samples
 
     def to_HDF(self, group: Group) -> None:
@@ -703,14 +703,7 @@ def _propagate(
         propagated_samples = spatial_response @ propagated_samples
 
         # Return the result
-        propagated_signal = Signal(
-            propagated_samples,
-            sampling_rate,
-            carrier_frequency=signal.carrier_frequency,
-            delay=signal.delay,
-            noise_power=signal.noise_power,
-        )
-        return propagated_signal
+        return signal.from_ndarray(propagated_samples)
 
     def plot_power_delay(self, axes: VAT | None = None) -> Tuple[plt.Figure, VAT]:
         if axes:

hermespy/channel/quadriga_channel.py

@@ -100,7 +100,7 @@ def _propagate(
             )
 
         max_delay_in_samples = ceil(np.max(self.path_delays) * signal.sampling_rate)
-        propagated_signal = np.zeros(
+        propagated_samples = np.zeros(
             (receiver.antennas.num_receive_antennas, signal.num_samples + max_delay_in_samples),
             dtype=np.complex_,
         )
@@ -117,13 +117,13 @@ def _propagate(
                 )
 
                 for delay_idx, delay_in_samples in enumerate(time_delay_in_samples_vec):
-                    propagated_signal[
+                    propagated_samples[
                         rx_antenna, delay_in_samples : delay_in_samples + signal.num_samples
                     ] += (
-                        cir_txa_rxa[delay_idx, : signal.num_samples] * signal.samples[tx_antenna, :]
+                        cir_txa_rxa[delay_idx, : signal.num_samples] * signal[tx_antenna, :].flatten()
                     )
 
-        return Signal(propagated_signal, signal.sampling_rate, signal.carrier_frequency)
+        return signal.from_ndarray(propagated_samples)
 
     def state(
         self,

hermespy/channel/radar_channel.py

@@ -389,7 +389,7 @@ def _propagate(
         # Apply the channel gain
         propagated_samples *= self.gain**0.5
 
-        return Signal(propagated_samples, signal.sampling_rate, signal.carrier_frequency)
+        return signal.from_ndarray(propagated_samples)
 
     def state(
         self,
@@ -717,7 +717,7 @@ def add_propagation(
 
         propagated_samples[
             :, delay_sample_offset : delay_sample_offset + signal.num_samples
-        ] += np.einsum("ij,jk,k->ik", propagation_response, signal.samples, echo_weights)
+        ] += np.einsum("ij,jk,k->ik", propagation_response, signal[:, :], echo_weights)
 
     def add_state(
         self,

hermespy/core/__init__.py

@@ -67,7 +67,7 @@
 from .random_node import RandomRealization, RandomNode
 from .drop import Drop, RecalledDrop
 from .scenario import Scenario, ScenarioMode, ScenarioType, ReplayScenario
-from .signal_model import Signal
+from .signal_model import Signal, DenseSignal, SparseSignal
 from .visualize import (
     ScatterVisualization,
     PlotVisualization,
@@ -173,6 +173,8 @@
     "ScenarioType",
     "ReplayScenario",
     "Signal",
+    "DenseSignal",
+    "SparseSignal",
     "ProcessedDeviceInput",
     "DeviceInput",
     "ScatterVisualization",

hermespy/core/channel.py

@@ -550,18 +550,13 @@ def propagate(self, signal: Signal) -> Signal:
             for tx_idx, rx_idx in product(range(state.shape[1]), range(state.shape[0])):
                 delayed_signal = (
                     state[rx_idx, tx_idx, : signal.num_samples, delay_index]
-                    * signal.samples[tx_idx, :]
+                    * signal[tx_idx, :].flatten()
                 )
                 propagated_samples[
                     rx_idx, delay_index : delay_index + signal.num_samples
                 ] += delayed_signal
 
-        return Signal(
-            propagated_samples,
-            sampling_rate=signal.sampling_rate,
-            carrier_frequency=signal.carrier_frequency,
-            delay=signal.delay,
-        )
+        return signal.from_ndarray(propagated_samples)
 
     def reciprocal(self) -> ChannelStateInformation:
         """Compute the reciprocal channel state.

hermespy/core/device.py

@@ -1797,7 +1797,7 @@ def generate_output(
         operator_streams = [o.selected_transmit_ports for o in self.transmitters]
 
         # Superimpose the operator transmissions to the device's RF configuration
-        superimposed_signal = Signal.empty(
+        superimposed_signal = Signal.Empty(
             self.sampling_rate, self.num_transmit_ports, carrier_frequency=self.carrier_frequency
         )
 
@@ -1865,7 +1865,7 @@ def process_input(
 
         # Superimpose the impinging signal models
         if len(impinging_signals) != 1:
-            superimposed_signal = Signal.empty(
+            superimposed_signal = Signal.Empty(
                 self.sampling_rate,
                 self.antennas.num_receive_antennas,
                 carrier_frequency=self.carrier_frequency,
@@ -1884,12 +1884,8 @@ def process_input(
                 slice(None) if selected_receive_antennas is None else selected_receive_antennas
             )
 
-            stream_samples = superimposed_signal.samples[stream_selector, :]  # type: ignore
-            operator_input = Signal(
-                stream_samples,
-                superimposed_signal.sampling_rate,
-                superimposed_signal.carrier_frequency,
-            )
+            stream_samples = superimposed_signal[stream_selector, :]
+            operator_input = superimposed_signal.from_ndarray(stream_samples)
             operator_inputs.append(operator_input)
 
         # Cache the operator inputs if the respective flag is enabled

hermespy/core/operators.py

@@ -99,7 +99,7 @@ def _transmit(self, duration: float = 0.0) -> Transmission:
         # Compute the number of samples to be transmitted
         num_samples = self.num_samples if duration <= 0.0 else int(duration * self.sampling_rate)
 
-        silence = Signal(
+        silence = Signal.Create(
             np.zeros((self.device.num_antennas, num_samples), dtype=complex),
             sampling_rate=self.sampling_rate,
             carrier_frequency=self.device.carrier_frequency,