"""Simulated SDR device for testing without hardware.
Set ``recorder.device = "mock"`` (or ``"sim"``) in a campaign config to use
this driver. The inference loop can also use it by specifying ``device:
"mock"`` in the SDR start request.
The mock generates complex float32 AWGN samples normalised to [-1, 1].
It satisfies both interfaces used in this codebase:
- ``record(num_samples)`` / ``_stream_rx(callback)`` — used by
``CampaignExecutor`` (inherits from ``SDR`` base class).
- ``rx(num_samples)`` — PlutoSDR-style interface used by the controller
inference loop.
"""
from __future__ import annotations
import time
import numpy as np
from ria_toolkit_oss.sdr.sdr import SDR
_DEFAULT_BUFFER_SIZE = 4096
# Simulated sample rate throttle: sleep this long between buffers so the
# loop does not spin at 100% CPU. 10 ms ≈ 100 buffers/s which is fine for
# tests and campaign execution timing.
_SLEEP_PER_BUFFER_S = 0.01
[docs]
class MockSDR(SDR):
"""Software-simulated SDR that generates AWGN noise.
Args:
buffer_size: Number of complex samples per streaming buffer.
seed: Optional RNG seed for reproducible output.
"""
def __init__(self, buffer_size: int = _DEFAULT_BUFFER_SIZE, seed: int | None = None):
super().__init__()
self.rx_buffer_size: int = buffer_size
self._rng = np.random.default_rng(seed)
# Direct attribute aliases used by _apply_sdr_config in the controller.
self.center_freq: float = 2.45e9
self.sample_rate: float = 10e6
self.gain: float = 40.0
# ------------------------------------------------------------------
# Abstract method implementations
# ------------------------------------------------------------------
[docs]
def init_rx(
self,
sample_rate: float,
center_frequency: float,
gain,
channel: int = 0,
gain_mode: str = "manual",
) -> None:
self.rx_sample_rate = float(sample_rate)
self.rx_center_frequency = float(center_frequency)
self.rx_gain = 40.0 if gain is None else float(gain)
# Mirror to the attribute names used by _apply_sdr_config.
self.sample_rate = self.rx_sample_rate
self.center_freq = self.rx_center_frequency
self.gain = self.rx_gain
self._rx_initialized = True
[docs]
def init_tx(
self,
sample_rate: float,
center_frequency: float,
gain,
channel: int = 0,
gain_mode: str = "manual",
) -> None:
self.tx_sample_rate = float(sample_rate)
self.tx_center_frequency = float(center_frequency)
self.tx_gain = 40.0 if gain is None else float(gain)
self._tx_initialized = True
def _stream_rx(self, callback) -> None:
"""Generate 1-D AWGN buffers and pass each to *callback* until stopped.
Uses 1-D arrays so the base class ``_validate_buffer`` check does not
incorrectly flag them as corrupted (the (1, N) form triggers a false
positive in the all-same-value check).
"""
self._enable_rx = True
while self._enable_rx:
buf = self._awgn(self.rx_buffer_size)
callback(buf)
time.sleep(_SLEEP_PER_BUFFER_S)
def _stream_tx(self, callback) -> None:
self._enable_tx = True
while self._enable_tx:
callback(self.rx_buffer_size)
time.sleep(_SLEEP_PER_BUFFER_S)
[docs]
def set_clock_source(self, source: str) -> None:
pass # no-op
[docs]
def close(self) -> None:
self._enable_rx = False
self._enable_tx = False
self._rx_initialized = False
self._tx_initialized = False
# ------------------------------------------------------------------
# PlutoSDR-style interface used by the controller inference loop
# ------------------------------------------------------------------
[docs]
def rx(self, num_samples: int) -> np.ndarray:
"""Return *num_samples* complex64 AWGN samples (PlutoSDR-style)."""
return self._awgn(num_samples)
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _awgn(self, n: int) -> np.ndarray:
"""Return *n* normalised complex64 AWGN samples as a 1-D array."""
real = self._rng.standard_normal(n).astype(np.float32)
imag = self._rng.standard_normal(n).astype(np.float32)
buf = real + 1j * imag
peak = np.abs(buf).max()
if peak > 1e-9:
buf /= peak
return buf