qoherent/ria-toolkit-oss
19
6
Fork 0
Code Issues Pull Requests 1 Workflows Packages Projects Releases 6 Wiki Activity

Compare commits

base: qoherent:fc098a12ee513fb5568c21a2e4c1ca0238fd6ebf
Branches Tags
qoherent:main
qoherent:docs/sdr-guides-update
qoherent:agent-naming-fix
qoherent:zfp-oss
qoherent:screens-connection
qoherent:dataset-manager-v2
qoherent:v0.1.5
qoherent:0.1.4
qoherent:v0.1.3
qoherent:v0.1.2
qoherent:v0.1.1
qoherent:v0.1.0
..
compare: qoherent:4ac4e9c642f216707a34480fdfcdb6bcf95089a9
Branches Tags
qoherent:docs/sdr-guides-update
qoherent:agent-naming-fix
qoherent:main
qoherent:zfp-oss
qoherent:screens-connection
qoherent:dataset-manager-v2
qoherent:v0.1.5
qoherent:0.1.4
qoherent:v0.1.3
qoherent:v0.1.2
qoherent:v0.1.1
qoherent:v0.1.0

No commits in common. "fc098a12ee513fb5568c21a2e4c1ca0238fd6ebf" and "4ac4e9c642f216707a34480fdfcdb6bcf95089a9" have entirely different histories.
fc098a12ee ... 4ac4e9c642

9 changed files with 526 additions and 948 deletions
Show Stats Expand all files Collapse all files

244
src/ria_toolkit_oss/sdr/blade.py
View File

@ -1,4 +1,4 @@
import gc import time
import warnings import warnings
from typing import Optional from typing import Optional
@ -6,7 +6,7 @@ import numpy as np
from bladerf import _bladerf from bladerf import _bladerf
from ria_toolkit_oss.datatypes import Recording from ria_toolkit_oss.datatypes import Recording
from ria_toolkit_oss.sdr import SDR, SDRError, SDRParameterError from ria_toolkit_oss.sdr import SDR
class Blade(SDR): class Blade(SDR):
@ -22,7 +22,7 @@ class Blade(SDR):
""" """
if identifier != "": if identifier != "":
warnings.warn(f"Blade: Identifier '{identifier}' will be ignored", UserWarning) print(f"Warning, radio identifier {identifier} provided for Blade but will not be used.")
uut = self._probe_bladerf() uut = self._probe_bladerf()
@ -34,7 +34,6 @@ class Blade(SDR):
self.device = _bladerf.BladeRF(uut) self.device = _bladerf.BladeRF(uut)
self._print_versions(device=self.device) self._print_versions(device=self.device)
self.bytes_per_sample = 4
super().__init__() super().__init__()
@ -43,10 +42,8 @@ class Blade(SDR):
if board is not None: if board is not None:
board.close() board.close()
if error != 0: # TODO why does this create an error under any conditions?
raise OSError(f"BladeRF shutdown with error code: {error}") raise OSError("Shutdown initiated with error code: {}".format(error))
else:
print("BladeRF shutdown successfully")
def _probe_bladerf(self): def _probe_bladerf(self):
device = None device = None
@ -88,25 +85,24 @@ class Blade(SDR):
:type sample_rate: int or float :type sample_rate: int or float
:param center_frequency: The center frequency of the recording. :param center_frequency: The center frequency of the recording.
:type center_frequency: int or float :type center_frequency: int or float
:param gain: The gain set for receiving on the BladeRF. :param gain: The gain set for receiving on the BladeRF
:type gain: int :type gain: int
:param channel: The channel the BladeRF is set to. :param channel: The channel the BladeRF is set to.
:type channel: int :type channel: int
:param buffer_size: The buffer size during receive. Defaults to 8192. :param buffer_size: The buffer size during receive. Defaults to 8192.
:type buffer_size: int :type buffer_size: int
:param gain_mode: 'absolute' passes gain directly to the SDR; :param gain_mode: 'absolute' passes gain directly to the sdr,
'relative' means that gain should be a negative value, and it will be subtracted from the max gain (60). 'relative' means that gain should be a negative value, and it will be subtracted from the max gain (60).
:type gain_mode: str :type gain_mode: str
""" """
print("Initializing RX") print("Initializing RX")
# Configure BladeRF # Configure BladeRF
self.set_rx_channel(channel) self._set_rx_channel(channel)
self.set_rx_sample_rate(sample_rate) self._set_rx_sample_rate(sample_rate)
self.set_rx_center_frequency(center_frequency) self._set_rx_center_frequency(center_frequency)
self.set_rx_gain(channel, gain, gain_mode) self._set_rx_gain(channel, gain, gain_mode)
self.set_rx_buffer_size(buffer_size) self._set_rx_buffer_size(buffer_size)
bw = self.rx_sample_rate bw = self.rx_sample_rate
if bw < 200000: if bw < 200000:
@ -132,8 +128,10 @@ class Blade(SDR):
stream_timeout=3500000000, stream_timeout=3500000000,
) )
print("Blade Starting RX...")
self.rx_ch.enable = True self.rx_ch.enable = True
self.bytes_per_sample = 4
print("Blade Starting RX...")
self._enable_rx = True self._enable_rx = True
while self._enable_rx: while self._enable_rx:
@ -150,34 +148,18 @@ class Blade(SDR):
print("Blade RX Completed.") print("Blade RX Completed.")
self.rx_ch.enable = False self.rx_ch.enable = False
def record( def record(self, num_samples: Optional[int] = None, rx_time: Optional[int | float] = None):
self,
num_samples: Optional[int] = None,
rx_time: Optional[int | float] = None,
) -> Recording:
"""
Create a radio recording (iq samples and metadata) of a given length from the Blade.
Either num_samples or rx_time must be provided.
init_rx() must be called before record()
:param num_samples: The number of samples to record.
:type num_samples: int, optional
:param rx_time: The time to record.
:type rx_time: int or float, optional
returns: Recording object (iq samples and metadata)
"""
if not self._rx_initialized: if not self._rx_initialized:
raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()") raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()")
if num_samples is not None and rx_time is not None: if num_samples is not None and rx_time is not None:
raise SDRParameterError("Only input one of num_samples or rx_time") raise ValueError("Only input one of num_samples or rx_time")
elif num_samples is not None: elif num_samples is not None:
self._num_samples_to_record = num_samples self._num_samples_to_record = num_samples
elif rx_time is not None: elif rx_time is not None:
self._num_samples_to_record = int(rx_time * self.rx_sample_rate) self._num_samples_to_record = int(rx_time * self.rx_sample_rate)
else: else:
raise SDRParameterError("Must provide input of one of num_samples or rx_time") raise ValueError("Must provide input of one of num_samples or rx_time")
# Setup synchronous stream # Setup synchronous stream
self.device.sync_config( self.device.sync_config(
@ -189,10 +171,11 @@ class Blade(SDR):
stream_timeout=3500000000, stream_timeout=3500000000,
) )
print("Blade Starting RX...")
with self._param_lock:
self._enable_rx = True
self.rx_ch.enable = True self.rx_ch.enable = True
self.bytes_per_sample = 4
print("Blade Starting RX...")
self._enable_rx = True
store_array = np.zeros( store_array = np.zeros(
(1, (self._num_samples_to_record // self.rx_buffer_size + 1) * self.rx_buffer_size), dtype=np.complex64 (1, (self._num_samples_to_record // self.rx_buffer_size + 1) * self.rx_buffer_size), dtype=np.complex64
@ -208,7 +191,6 @@ class Blade(SDR):
# Disable module # Disable module
print("Blade RX Completed.") print("Blade RX Completed.")
with self._param_lock:
self.rx_ch.enable = False self.rx_ch.enable = False
metadata = { metadata = {
"source": self.__class__.__name__, "source": self.__class__.__name__,
@ -225,7 +207,7 @@ class Blade(SDR):
center_frequency: int | float, center_frequency: int | float,
gain: int, gain: int,
channel: int, channel: int,
buffer_size: Optional[int] = 32768, buffer_size: Optional[int] = 8192,
gain_mode: Optional[str] = "absolute", gain_mode: Optional[str] = "absolute",
): ):
""" """
@ -244,22 +226,14 @@ class Blade(SDR):
:param gain_mode: 'absolute' passes gain directly to the sdr, :param gain_mode: 'absolute' passes gain directly to the sdr,
'relative' means that gain should be a negative value, and it will be subtracted from the max gain (60). 'relative' means that gain should be a negative value, and it will be subtracted from the max gain (60).
:type gain_mode: str :type gain_mode: str
:return: 0 if successful, -1 if there's an error.
:rtype: int
""" """
# Configure BladeRF # Configure BladeRF
self.set_tx_channel(channel) self._set_tx_channel(channel)
self.set_tx_sample_rate(sample_rate) self._set_tx_sample_rate(sample_rate)
self.set_tx_center_frequency(center_frequency) self._set_tx_center_frequency(center_frequency)
self.set_tx_gain(channel=channel, gain=gain, gain_mode=gain_mode) self._set_tx_gain(channel=channel, gain=gain, gain_mode=gain_mode)
self.set_tx_buffer_size(buffer_size) self._set_tx_buffer_size(buffer_size)
if self.tx_sample_rate >= 7.5e6 and self.tx_buffer_size < 65536:
warnings.warn(
"Blade: For high sample rates, a buffer size of 65536, 131072, or 262144 is recommended", UserWarning
)
bw = self.tx_sample_rate bw = self.tx_sample_rate
if bw < 200000: if bw < 200000:
@ -328,13 +302,13 @@ class Blade(SDR):
""" """
if num_samples is not None and tx_time is not None: if num_samples is not None and tx_time is not None:
raise SDRParameterError("Only input one of num_samples or tx_time") raise ValueError("Only input one of num_samples or tx_time")
elif num_samples is not None: elif num_samples is not None:
pass tx_time = num_samples / self.tx_sample_rate
elif tx_time is not None: elif tx_time is not None:
num_samples = int(tx_time * self.tx_sample_rate) pass
else: else:
num_samples = len(recording) tx_time = len(recording) / self.tx_sample_rate
if isinstance(recording, np.ndarray): if isinstance(recording, np.ndarray):
samples = recording samples = recording
@ -343,15 +317,9 @@ class Blade(SDR):
warnings.warn("Recording object is multichannel, only channel 0 data was used for transmission") warnings.warn("Recording object is multichannel, only channel 0 data was used for transmission")
samples = recording.data[0] samples = recording.data[0]
else: else:
raise SDRParameterError("recording must be np.ndarray or Recording") raise TypeError("recording must be np.ndarray or Recording")
samples = samples.astype(np.complex64, copy=False) samples = samples.astype(np.complex64, copy=False)
tx_bytes = self._convert_tx_samples(samples)
# Transmit in chunks
samples_sent = 0
len_samples = len(samples)
chunk_size = self.tx_buffer_size
# Setup stream # Setup stream
self.device.sync_config( self.device.sync_config(
@ -367,21 +335,26 @@ class Blade(SDR):
self.tx_ch.enable = True self.tx_ch.enable = True
print("Blade Starting TX...") print("Blade Starting TX...")
# Transmit samples - repeat as needed for the duration
start_time = time.time()
sample_index = 0
try: try:
while samples_sent < num_samples: while time.time() - start_time < tx_time:
this_chunk_size = min(chunk_size, num_samples - samples_sent) # Get next chunk
chunk_size = min(self.tx_buffer_size, len(samples) - sample_index)
if chunk_size == 0:
# Reached end, loop back
sample_index = 0
chunk_size = min(self.tx_buffer_size, len(samples))
start_idx = (samples_sent % len_samples) * self.bytes_per_sample chunk = samples[sample_index : sample_index + chunk_size]
end_idx = start_idx + this_chunk_size * self.bytes_per_sample sample_index += chunk_size
end_idx %= len_samples * self.bytes_per_sample
if end_idx > start_idx: # Convert and transmit
chunk_bytes_arr = tx_bytes[start_idx:end_idx] byte_array = self._convert_tx_samples(chunk)
else: self.device.sync_tx(byte_array, len(chunk))
chunk_bytes_arr = tx_bytes[start_idx:] + tx_bytes[:end_idx]
self.device.sync_tx(chunk_bytes_arr, this_chunk_size)
samples_sent += this_chunk_size
except KeyboardInterrupt: except KeyboardInterrupt:
print("\nTransmission interrupted by user") print("\nTransmission interrupted by user")
@ -411,70 +384,29 @@ class Blade(SDR):
byte_array = tx_samples.tobytes() byte_array = tx_samples.tobytes()
return byte_array return byte_array
def set_rx_channel(self, channel): def _set_rx_channel(self, channel):
if channel != 0 and channel != 1:
raise SDRParameterError("Channel must be either 0 or 1.")
self.rx_channel = channel self.rx_channel = channel
self.rx_ch = self.device.Channel(_bladerf.CHANNEL_RX(channel)) self.rx_ch = self.device.Channel(_bladerf.CHANNEL_RX(channel))
print(f"\nBlade channel = {self.rx_ch}") print(f"\nBlade channel = {self.rx_ch}")
def set_rx_sample_rate(self, sample_rate): def _set_rx_sample_rate(self, sample_rate):
"""
Set the sample rate of the receiver.
Not callable during recording; Blade requires stream stop/restart to change sample rate.
"""
with self._param_lock:
if hasattr(self, "rx_channel"):
range_list = self.device.get_sample_rate_range(self.rx_channel)
min_rate, max_rate = range_list[0], range_list[1]
else:
raise SDRError("Must set channel before setting center frequency")
if sample_rate < min_rate or sample_rate > max_rate:
raise SDRParameterError(
f"{self.__class__.__name__}: Sample rate {sample_rate/1e6:.3f} Msps "
f"out of range: [{min_rate/1e6:.3f} - {max_rate/1e6:.3f} Msps]"
)
self.rx_sample_rate = sample_rate self.rx_sample_rate = sample_rate
self.rx_ch.sample_rate = self.rx_sample_rate self.rx_ch.sample_rate = self.rx_sample_rate
print(f"Blade sample rate = {self.rx_ch.sample_rate}") print(f"Blade sample rate = {self.rx_ch.sample_rate}")
def set_rx_center_frequency(self, center_frequency): def _set_rx_center_frequency(self, center_frequency):
"""
Set the center frequency of the receiver.
Not callable during recording; Blade requires stream stop/restart to change center frequency.
"""
with self._param_lock:
if hasattr(self, "rx_channel"):
range_list = self.device.get_frequency_range(self.rx_channel)
min_rate, max_rate = range_list[0], range_list[1]
else:
raise SDRError("Must set channel before setting center frequency")
if center_frequency < min_rate or center_frequency > max_rate:
raise SDRParameterError(
f"{self.__class__.__name__}: Center frequency {center_frequency/1e9:.3f} GHz "
f"out of range: [{min_rate/1e9:.3f} - {max_rate/1e9:.3f} GHz]"
)
self.rx_center_frequency = center_frequency self.rx_center_frequency = center_frequency
self.rx_ch.frequency = center_frequency self.rx_ch.frequency = center_frequency
print(f"Blade center frequency = {self.rx_ch.frequency}") print(f"Blade center frequency = {self.rx_ch.frequency}")
def set_rx_gain(self, channel, gain, gain_mode): def _set_rx_gain(self, channel, gain, gain_mode):
"""
Set the gain of the receiver.
Not callable during recording; Blade requires stream stop/restart to change gain.
"""
with self._param_lock:
rx_gain_min = self.device.get_gain_range(channel)[0] rx_gain_min = self.device.get_gain_range(channel)[0]
rx_gain_max = self.device.get_gain_range(channel)[1] rx_gain_max = self.device.get_gain_range(channel)[1]
if gain_mode == "relative": if gain_mode == "relative":
if gain > 0: if gain > 0:
raise SDRParameterError( raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This sets \ "When gain_mode = 'relative', gain must be < 0. This sets \
the gain relative to the maximum possible gain." the gain relative to the maximum possible gain."
) )
@ -493,64 +425,32 @@ class Blade(SDR):
print(f"Blade gain = {self.rx_ch.gain}") print(f"Blade gain = {self.rx_ch.gain}")
def set_rx_buffer_size(self, buffer_size): def _set_rx_buffer_size(self, buffer_size):
self.rx_buffer_size = buffer_size self.rx_buffer_size = buffer_size
def set_tx_channel(self, channel): def _set_tx_channel(self, channel):
if channel != 0 and channel != 1:
raise SDRParameterError("Channel must be either 0 or 1.")
self.tx_channel = channel self.tx_channel = channel
self.tx_ch = self.device.Channel(_bladerf.CHANNEL_TX(self.tx_channel)) self.tx_ch = self.device.Channel(_bladerf.CHANNEL_TX(self.tx_channel))
print(f"\nBlade channel = {self.tx_ch}") print(f"\nBlade channel = {self.tx_ch}")
def set_tx_sample_rate(self, sample_rate): def _set_tx_sample_rate(self, sample_rate):
if hasattr(self, "tx_channel"):
range_list = self.device.get_sample_rate_range(self.tx_channel)
min_rate, max_rate = range_list[0], range_list[1]
else:
raise SDRError("Must set channel before setting center frequency")
if sample_rate < min_rate or sample_rate > max_rate:
raise SDRParameterError(
f"{self.__class__.__name__}: Sample rate {sample_rate/1e6:.3f} Msps "
f"out of range: [{min_rate/1e6:.3f} - {max_rate/1e6:.3f} Msps]"
)
if sample_rate < min_rate or sample_rate > max_rate:
raise SDRParameterError(
f"{self.__class__.__name__}: Sample rate {sample_rate/1e6:.3f} Msps "
f"out of range: [{min_rate/1e6:.3f} - {max_rate/1e6:.3f} Msps]"
)
self.tx_sample_rate = sample_rate self.tx_sample_rate = sample_rate
self.tx_ch.sample_rate = self.tx_sample_rate self.tx_ch.sample_rate = self.tx_sample_rate
print(f"Blade sample rate = {self.tx_ch.sample_rate}") print(f"Blade sample rate = {self.tx_ch.sample_rate}")
def set_tx_center_frequency(self, center_frequency): def _set_tx_center_frequency(self, center_frequency):
if hasattr(self, "tx_channel"):
range_list = self.device.get_frequency_range(self.tx_channel)
min_rate, max_rate = range_list[0], range_list[1]
else:
raise SDRError("Must set channel before setting center frequency")
if center_frequency < min_rate or center_frequency > max_rate:
raise SDRParameterError(
f"{self.__class__.__name__}: Center frequency {center_frequency/1e9:.3f} GHz "
f"out of range: [{min_rate/1e9:.3f} - {max_rate/1e9:.3f} GHz]"
)
self.tx_center_frequency = center_frequency self.tx_center_frequency = center_frequency
self.tx_ch.frequency = center_frequency self.tx_ch.frequency = center_frequency
print(f"Blade center frequency = {self.tx_ch.frequency}") print(f"Blade center frequency = {self.tx_ch.frequency}")
def set_tx_gain(self, channel, gain, gain_mode): def _set_tx_gain(self, channel, gain, gain_mode):
tx_gain_min = self.device.get_gain_range(channel)[0] tx_gain_min = self.device.get_gain_range(channel)[0]
tx_gain_max = self.device.get_gain_range(channel)[1] tx_gain_max = self.device.get_gain_range(channel)[1]
if gain_mode == "relative": if gain_mode == "relative":
if gain > 0: if gain > 0:
raise SDRParameterError( raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This sets\ "When gain_mode = 'relative', gain must be < 0. This sets\
the gain relative to the maximum possible gain." the gain relative to the maximum possible gain."
) )
@ -569,7 +469,7 @@ class Blade(SDR):
print(f"Blade gain = {self.tx_ch.gain}") print(f"Blade gain = {self.tx_ch.gain}")
def set_tx_buffer_size(self, buffer_size): def _set_tx_buffer_size(self, buffer_size):
self.tx_buffer_size = buffer_size self.tx_buffer_size = buffer_size
def set_clock_source(self, source): def set_clock_source(self, source):
@ -599,20 +499,4 @@ class Blade(SDR):
print(f"BladeRF bias tee {state} on channel {channel}.") print(f"BladeRF bias tee {state} on channel {channel}.")
def close(self): def close(self):
if hasattr(self, "device") and self.device is not None:
try:
if hasattr(self, "tx_ch"):
self.tx_ch.enable = False
if hasattr(self, "rx_ch"):
self.rx_ch.enable = False
self.device.close() self.device.close()
except Exception as e:
print(f"Warning: error closing bladeRF: {e}")
finally:
del self.device
self.device = None
gc.collect()
def supports_dynamic_updates(self) -> dict:
return {"center_frequency": False, "sample_rate": False, "gain": False}

123
src/ria_toolkit_oss/sdr/hackrf.py
View File

@ -6,7 +6,7 @@ import numpy as np
from ria_toolkit_oss.datatypes.recording import Recording from ria_toolkit_oss.datatypes.recording import Recording
from ria_toolkit_oss.sdr._external.libhackrf import HackRF as hrf from ria_toolkit_oss.sdr._external.libhackrf import HackRF as hrf
from ria_toolkit_oss.sdr.sdr import SDR, SDRParameterError from ria_toolkit_oss.sdr.sdr import SDR
class HackRF(SDR): class HackRF(SDR):
@ -21,7 +21,7 @@ class HackRF(SDR):
""" """
if identifier != "": if identifier != "":
warnings.warn(f"HackRF: Identifier '{identifier}' will be ignored", UserWarning) print(f"Warning, radio identifier {identifier} provided for HackRF but will not be used.")
print("Initializing HackRF radio.") print("Initializing HackRF radio.")
try: try:
@ -33,6 +33,8 @@ class HackRF(SDR):
print("Failed to find HackRF radio.") print("Failed to find HackRF radio.")
raise e raise e
super().__init__()
def init_rx( def init_rx(
self, self,
sample_rate: int | float, sample_rate: int | float,
@ -62,8 +64,14 @@ class HackRF(SDR):
:type gain_mode: str :type gain_mode: str
""" """
print("Initializing RX") print("Initializing RX")
self.set_sample_rate(sample_rate=sample_rate)
self.set_center_frequency(center_frequency=center_frequency) self.rx_sample_rate = sample_rate
self.radio.sample_rate = int(sample_rate)
print(f"HackRF sample rate = {self.radio.sample_rate}")
self.rx_center_frequency = center_frequency
self.radio.center_freq = int(center_frequency)
print(f"HackRF center frequency = {self.radio.center_freq}")
# Distribute gain across amplifier stages # Distribute gain across amplifier stages
rx_gain_min = 0 rx_gain_min = 0
@ -71,7 +79,7 @@ class HackRF(SDR):
if gain_mode == "relative": if gain_mode == "relative":
if gain > 0: if gain > 0:
raise SDRParameterError( raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This " "When gain_mode = 'relative', gain must be < 0. This "
"sets the gain relative to the maximum possible gain." "sets the gain relative to the maximum possible gain."
) )
@ -91,9 +99,7 @@ class HackRF(SDR):
self.rx_gain = abs_gain self.rx_gain = abs_gain
print(f"HackRF gain distribution: Amp={self.amp_enabled}, LNA={self.rx_lna_gain}dB, VGA={self.rx_vga_gain}dB") print(f"HackRF gain distribution: Amp={self.amp_enabled}, LNA={self.rx_lna_gain}dB, VGA={self.rx_vga_gain}dB")
print( print("To individually modify the HackRF gains, use set_gain_amp(), set_rx_lna_gain(), and set_rx_vga_gain().")
"To individually modify the HackRF gains, use set_gain_amp(), set_rx_lna_gain(), and set_rx_vga_gain().\n"
)
self._tx_initialized = False self._tx_initialized = False
self._rx_initialized = True self._rx_initialized = True
@ -116,13 +122,13 @@ class HackRF(SDR):
raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()") raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()")
if num_samples is not None and rx_time is not None: if num_samples is not None and rx_time is not None:
raise SDRParameterError("Only input one of num_samples or rx_time") raise ValueError("Only input one of num_samples or rx_time")
elif num_samples is not None: elif num_samples is not None:
self._num_samples_to_record = num_samples self._num_samples_to_record = num_samples
elif rx_time is not None: elif rx_time is not None:
self._num_samples_to_record = int(rx_time * self.sample_rate) self._num_samples_to_record = int(rx_time * self.rx_sample_rate)
else: else:
raise SDRParameterError("Must provide input of one of num_samples or rx_time") raise ValueError("Must provide input of one of num_samples or rx_time")
print("HackRF Starting RX...") print("HackRF Starting RX...")
@ -131,15 +137,18 @@ class HackRF(SDR):
print("HackRF RX Completed.") print("HackRF RX Completed.")
rx_complex = self.convert_rx_samples(rx_samples=all_samples) # Create 1xN array for single-channel recording
store_array = np.zeros((1, self._num_samples_to_record), dtype=np.complex64)
store_array[0, :] = all_samples
metadata = { metadata = {
"source": self.__class__.__name__, "source": self.__class__.__name__,
"sample_rate": self.sample_rate, "sample_rate": self.rx_sample_rate,
"center_frequency": self.center_frequency, "center_frequency": self.rx_center_frequency,
"gain": self.rx_gain, "gain": self.rx_gain,
} }
return Recording(data=rx_complex, metadata=metadata) return Recording(data=store_array, metadata=metadata)
def init_tx( def init_tx(
self, self,
@ -165,14 +174,19 @@ class HackRF(SDR):
""" """
print("Initializing TX") print("Initializing TX")
self.set_sample_rate(sample_rate=sample_rate) self.tx_sample_rate = sample_rate
self.set_center_frequency(center_frequency=center_frequency) self.radio.sample_rate = int(sample_rate)
print(f"HackRF sample rate = {self.radio.sample_rate}")
self.tx_center_frequency = center_frequency
self.radio.center_freq = int(center_frequency)
print(f"HackRF center frequency = {self.radio.center_freq}")
tx_gain_min = 0 tx_gain_min = 0
tx_gain_max = 47 tx_gain_max = 47
if gain_mode == "relative": if gain_mode == "relative":
if gain > 0: if gain > 0:
raise SDRParameterError( raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This \ "When gain_mode = 'relative', gain must be < 0. This \
sets the gain relative to the maximum possible gain." sets the gain relative to the maximum possible gain."
) )
@ -183,14 +197,14 @@ class HackRF(SDR):
if abs_gain < tx_gain_min or abs_gain > tx_gain_max: if abs_gain < tx_gain_min or abs_gain > tx_gain_max:
abs_gain = min(max(gain, tx_gain_min), tx_gain_max) abs_gain = min(max(gain, tx_gain_min), tx_gain_max)
print(f"Gain {gain} out of range for HackRF.") print(f"Gain {gain} out of range for Pluto.")
print(f"Gain range: {tx_gain_min} to {tx_gain_max} dB") print(f"Gain range: {tx_gain_min} to {tx_gain_max} dB")
self.set_gain_amp(True) self.set_gain_amp(True)
self.set_tx_vga_gain(abs_gain) self.set_tx_vga_gain(abs_gain)
self.tx_gain = abs_gain self.tx_gain = abs_gain
print(f"HackRF gain distribution: Amp={self.amp_enabled}, VGA={self.tx_vga_gain}dB") print(f"HackRF gain distribution: Amp={self.amp_enabled}, VGA={self.tx_vga_gain}dB")
print("To individually modify the HackRF gains, use set_gain_amp() or set_tx_vga_gain().\n") print("To individually modify the HackRF gains, use set_gain_amp() or set_tx_vga_gain().")
self._tx_initialized = True self._tx_initialized = True
self._rx_initialized = False self._rx_initialized = False
@ -215,13 +229,13 @@ class HackRF(SDR):
:type tx_time: int or float, optional :type tx_time: int or float, optional
""" """
if num_samples is not None and tx_time is not None: if num_samples is not None and tx_time is not None:
raise SDRParameterError("Only input one of num_samples or tx_time") raise ValueError("Only input one of num_samples or tx_time")
elif num_samples is not None: elif num_samples is not None:
tx_time = num_samples / self.sample_rate tx_time = num_samples / self.tx_sample_rate
elif tx_time is not None: elif tx_time is not None:
pass pass
else: else:
tx_time = len(recording) / self.sample_rate tx_time = len(recording) / self.tx_sample_rate
if isinstance(recording, np.ndarray): if isinstance(recording, np.ndarray):
samples = recording samples = recording
@ -261,62 +275,6 @@ class HackRF(SDR):
self.radio.set_txvga_gain(vga_gain) self.radio.set_txvga_gain(vga_gain)
self.tx_vga_gain = vga_gain self.tx_vga_gain = vga_gain
def set_sample_rate(self, sample_rate):
if sample_rate < 2e6 or sample_rate > 20e6:
raise SDRParameterError(
f"{self.__class__.__name__}: Sample rate {sample_rate/1e6:.3f} Msps "
f"out of range: [{2:.3f} - {20:.3f} Msps]"
)
self.sample_rate = sample_rate
self.radio.sample_rate = int(sample_rate)
print(f"HackRF sample rate = {self.radio.sample_rate}")
def set_rx_sample_rate(self, sample_rate):
"""
Set the sample rate.
Not callable during recording; HackRF requires stream stop/restart to change sample rate.
"""
self.set_sample_rate(sample_rate=sample_rate)
def set_tx_sample_rate(self, sample_rate):
self.set_sample_rate(sample_rate=sample_rate)
def set_center_frequency(self, center_frequency):
with self._param_lock:
if center_frequency < 1e6 or center_frequency > 6e9:
raise SDRParameterError(
f"{self.__class__.__name__}: Center frequency {center_frequency/1e9:.3f} GHz "
f"out of range: [{1e6/1e9:.3f} - {6e9/1e9:.3f} GHz]"
)
self.center_frequency = center_frequency
self.radio.center_freq = int(center_frequency)
print(f"HackRF center frequency = {self.radio.center_freq}")
def set_rx_center_frequency(self, center_frequency):
"""
Set the center frequency. Callable during streaming.
"""
self.set_center_frequency(center_frequency=center_frequency)
def set_tx_center_frequency(self, center_frequency):
self.set_center_frequency(center_frequency=center_frequency)
def convert_rx_samples(self, rx_samples):
# Handle conversion depending on dtype
if np.issubdtype(rx_samples.dtype, np.complexfloating):
# Already complex: just normalize
rx_complex = rx_samples.astype(np.complex64) / 128.0
elif np.issubdtype(rx_samples.dtype, np.integer):
# Raw interleaved I/Q bytes: convert to complex
i_samples = rx_samples[0::2].astype(np.float32)
q_samples = rx_samples[1::2].astype(np.float32)
rx_complex = (i_samples + 1j * q_samples) / 128.0
else:
raise TypeError(f"Unexpected dtype from read_samples: {rx_samples.dtype}")
# Ensure 2D array: 1xN for single channel
return rx_complex.reshape((1, -1))
def set_clock_source(self, source): def set_clock_source(self, source):
self.radio.set_clock_source(source) self.radio.set_clock_source(source)
@ -330,11 +288,7 @@ class HackRF(SDR):
raise NotImplementedError("Underlying HackRF interface lacks bias-tee control") from exc raise NotImplementedError("Underlying HackRF interface lacks bias-tee control") from exc
def close(self): def close(self):
try:
self.radio.close() self.radio.close()
del self.radio
finally:
self._enable_rx = False
def _stream_rx(self, callback): def _stream_rx(self, callback):
""" """
@ -388,6 +342,3 @@ class HackRF(SDR):
def _stream_tx(self, callback): def _stream_tx(self, callback):
return super()._stream_tx(callback) return super()._stream_tx(callback)
def supports_dynamic_updates(self) -> dict:
return {"center_frequency": True, "sample_rate": False, "gain": False}

321
src/ria_toolkit_oss/sdr/pluto.py
View File

@ -8,7 +8,7 @@ import adi
import numpy as np import numpy as np
from ria_toolkit_oss.datatypes.recording import Recording from ria_toolkit_oss.datatypes.recording import Recording
from ria_toolkit_oss.sdr.sdr import SDR, SDRError, SDRParameterError from ria_toolkit_oss.sdr.sdr import SDR
class Pluto(SDR): class Pluto(SDR):
@ -28,7 +28,6 @@ class Pluto(SDR):
print(f"Initializing Pluto radio with identifier [{identifier}].") print(f"Initializing Pluto radio with identifier [{identifier}].")
try: try:
super().__init__() super().__init__()
self._tx_lock = threading.Lock()
if identifier is None: if identifier is None:
uri = "ip:pluto.local" uri = "ip:pluto.local"
@ -75,12 +74,10 @@ class Pluto(SDR):
:type center_frequency: int or float :type center_frequency: int or float
:param gain: The gain set for receiving on the Pluto :param gain: The gain set for receiving on the Pluto
:type gain: int :type gain: int
:param channel: The channel the Pluto is set to. Must be 0 or 1. 0 :param channel: The channel the Pluto is set to. Must be 0 or 1. 0 enables channel 1, 1 enables both channels.
enables channel 1, 1 enables both channels.
:type channel: int :type channel: int
:param gain_mode: 'absolute' passes gain directly to the sdr, :param gain_mode: 'absolute' passes gain directly to the sdr,
'relative' means that gain should be a negative value, and it will 'relative' means that gain should be a negative value, and it will be subtracted from the max gain (74).
be subtracted from the max gain (74).
:type gain_mode: str :type gain_mode: str
""" """
print("Initializing RX") print("Initializing RX")
@ -91,7 +88,20 @@ class Pluto(SDR):
self.set_rx_center_frequency(center_frequency=int(center_frequency)) self.set_rx_center_frequency(center_frequency=int(center_frequency))
print(f"Pluto center frequency = {self.radio.rx_lo}") print(f"Pluto center frequency = {self.radio.rx_lo}")
self.set_rx_channel(channel=channel) if channel == 0:
self.radio.rx_enabled_channels = [0]
print(f"Pluto channel(s) = {self.radio.rx_enabled_channels}")
elif channel == 1:
if not self._mimo_capable:
raise ValueError(
"Dual RX channel requested (channel=1) but hardware is not MIMO-capable. "
"Dual RX/TX requires Pluto Rev C/D. Detected hardware: Rev B (single channel only)."
)
self.radio.rx_enabled_channels = [0, 1]
print(f"Pluto channel(s) = {self.radio.rx_enabled_channels}")
else:
raise ValueError("Channel must be either 0 or 1.")
self.set_rx_gain(gain=gain, channel=channel, gain_mode=gain_mode) self.set_rx_gain(gain=gain, channel=channel, gain_mode=gain_mode)
if channel == 0: if channel == 0:
print(f"Pluto gain = {self.radio.rx_hardwaregain_chan0}") print(f"Pluto gain = {self.radio.rx_hardwaregain_chan0}")
@ -99,6 +109,8 @@ class Pluto(SDR):
self.set_rx_gain(gain=gain, channel=0, gain_mode=gain_mode) self.set_rx_gain(gain=gain, channel=0, gain_mode=gain_mode)
print(f"Pluto gain = {self.radio.rx_hardwaregain_chan0}, {self.radio.rx_hardwaregain_chan1}") print(f"Pluto gain = {self.radio.rx_hardwaregain_chan0}, {self.radio.rx_hardwaregain_chan1}")
self.set_rx_buffer_size(getattr(self, "rx_buffer_size", 1024))
self._rx_initialized = True self._rx_initialized = True
self._tx_initialized = False self._tx_initialized = False
@ -122,12 +134,10 @@ class Pluto(SDR):
:type center_frequency: int or float :type center_frequency: int or float
:param gain: The gain set for transmitting on the Pluto :param gain: The gain set for transmitting on the Pluto
:type gain: int :type gain: int
:param channel: The channel the Pluto is set to. Must be 0 or 1. 0 :param channel: The channel the Pluto is set to. Must be 0 or 1. 0 enables channel 1, 1 enables both channels.
enables channel 1, 1 enables both channels.
:type channel: int :type channel: int
:param gain_mode: 'absolute' passes gain directly to the sdr, :param gain_mode: 'absolute' passes gain directly to the sdr,
'relative' means that gain should be a negative value, and it will 'relative' means that gain should be a negative value, and it will be subtracted from the max gain (0).
be subtracted from the max gain (0).
:type gain_mode: str :type gain_mode: str
""" """
@ -139,7 +149,20 @@ class Pluto(SDR):
self.set_tx_center_frequency(center_frequency=int(center_frequency)) self.set_tx_center_frequency(center_frequency=int(center_frequency))
print(f"Pluto center frequency = {self.radio.tx_lo}") print(f"Pluto center frequency = {self.radio.tx_lo}")
self.set_tx_channel(channel=channel) if channel == 0:
self.radio.tx_enabled_channels = [0]
print(f"Pluto channel(s) = {self.radio.tx_enabled_channels}")
elif channel == 1:
if not self._mimo_capable:
raise ValueError(
"Dual TX channel requested (channel=1) but hardware is not MIMO-capable. "
"Dual RX/TX requires Pluto Rev C/D. Detected hardware: Rev B (single channel only)."
)
self.radio.tx_enabled_channels = [0, 1]
print(f"Pluto channel(s) = {self.radio.tx_enabled_channels}")
else:
raise ValueError("Channel must be either 0 or 1.")
self.set_tx_gain(gain=gain, channel=channel, gain_mode=gain_mode) self.set_tx_gain(gain=gain, channel=channel, gain_mode=gain_mode)
if channel == 0: if channel == 0:
print(f"Pluto gain = {self.radio.tx_hardwaregain_chan0}") print(f"Pluto gain = {self.radio.tx_hardwaregain_chan0}")
@ -156,74 +179,16 @@ class Pluto(SDR):
if not self._rx_initialized: if not self._rx_initialized:
raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()") raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()")
# print("Starting rx...")
self._enable_rx = True self._enable_rx = True
while self._enable_rx is True: while self._enable_rx is True:
# collect complex signa from radio
signal = self.radio.rx() signal = self.radio.rx()
signal = self._convert_rx_samples(signal)
# send callback complex signal # send callback complex signal
callback(buffer=signal, metadata=None) callback(buffer=signal, metadata=None)
def _record_fast(self, num_samples): def record(self, num_samples: Optional[int] = None, rx_time: Optional[int | float] = None):
"""Optimized single-buffer capture for ≤16M samples."""
self.set_rx_buffer_size(buffer_size=num_samples)
print("Pluto Starting RX...")
samples = self.radio.rx()
# Handle single/dual channel
if self.radio.rx_enabled_channels == [0]:
samples = [self._convert_rx_samples(samples)]
else:
samples = [self._convert_rx_samples(s) for s in samples]
print("Pluto RX Completed.")
metadata = {
"source": self.__class__.__name__,
"sample_rate": self.rx_sample_rate,
"center_frequency": self.rx_center_frequency,
"gain": self.rx_gain,
}
return Recording(data=samples, metadata=metadata)
def _record_chunked(self, num_samples):
"""Chunked streaming capture for >2M samples."""
# Use base class streaming with pre-allocation
chunk_size = 2_000_000 # 2M sample chunks (safe size)
self.set_rx_buffer_size(buffer_size=chunk_size)
self._max_num_buffers = (num_samples // chunk_size) + 1
self._num_buffers_processed = 0
self._accumulated_buffer = None
# Stream with accumulation callback
print("Pluto Starting RX...")
self._stream_rx(callback=self._accumulate_buffers_callback)
print("Pluto RX Completed.")
print(f"Corrupted buffer count: {self._corrupted_buffer_count}")
# Truncate to exact size
samples = self._accumulated_buffer[:, :num_samples]
samples_list = [self._convert_rx_samples(chan) for chan in samples]
metadata = {
"source": self.__class__.__name__,
"sample_rate": self.rx_sample_rate,
"center_frequency": self.rx_center_frequency,
"gain": self.rx_gain,
}
# Reset for next capture
self._accumulated_buffer = None
return Recording(data=samples_list, metadata=metadata)
def record(
self,
num_samples: Optional[int] = None,
rx_time: Optional[int | float] = None,
) -> Recording:
""" """
Create a radio recording (iq samples and metadata) of a given length from the SDR. Create a radio recording (iq samples and metadata) of a given length from the SDR.
Either num_samples or rx_time must be provided. Either num_samples or rx_time must be provided.
@ -240,19 +205,38 @@ class Pluto(SDR):
raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()") raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()")
if num_samples is not None and rx_time is not None: if num_samples is not None and rx_time is not None:
raise SDRParameterError("Only input one of num_samples or rx_time") raise ValueError("Only input one of num_samples or rx_time")
elif num_samples is not None: elif num_samples is not None:
self._num_samples_to_record = num_samples self._num_samples_to_record = num_samples
elif rx_time is not None: elif rx_time is not None:
self._num_samples_to_record = int(rx_time * self.rx_sample_rate) self._num_samples_to_record = int(rx_time * self.rx_sample_rate)
else: else:
raise SDRParameterError("Must provide input of one of num_samples or rx_time") raise ValueError("Must provide input of one of num_samples or rx_time")
# Record in one go if there are less than 2,000,000 samples to record, record in chunks otherwise if self._num_samples_to_record > 16000000:
if self._num_samples_to_record <= 2_000_000: raise NotImplementedError("Pluto record for num_samples>16M not implemented yet.")
return self._record_fast(self._num_samples_to_record) self.radio.rx_buffer_size = self._num_samples_to_record
print("Pluto Starting RX...")
samples = self.radio.rx()
if self.radio.rx_enabled_channels == [0]:
samples = self._convert_rx_samples(samples)
samples = [samples]
else: else:
return self._record_chunked(self._num_samples_to_record) channel1 = self._convert_rx_samples(samples[0])
channel2 = self._convert_rx_samples(samples[1])
samples = [channel1, channel2]
print("Pluto RX Completed.")
metadata = {
"source": self.__class__.__name__,
"sample_rate": self.rx_sample_rate,
"center_frequency": self.rx_center_frequency,
"gain": self.rx_gain,
}
recording = Recording(data=samples, metadata=metadata)
return recording
def _format_tx_data(self, recording: Recording | np.ndarray | list): def _format_tx_data(self, recording: Recording | np.ndarray | list):
if isinstance(recording, np.ndarray): if isinstance(recording, np.ndarray):
@ -305,7 +289,6 @@ class Pluto(SDR):
print("Pluto TX Completed.") print("Pluto TX Completed.")
def interrupt_transmit(self): def interrupt_transmit(self):
with self._tx_lock:
self.radio.tx_destroy_buffer() self.radio.tx_destroy_buffer()
self.radio.tx_cyclic_buffer = False self.radio.tx_cyclic_buffer = False
print("Pluto TX Completed.") print("Pluto TX Completed.")
@ -327,7 +310,7 @@ class Pluto(SDR):
:type mode: str, optional :type mode: str, optional
""" """
if num_samples is not None and tx_time is not None: if num_samples is not None and tx_time is not None:
raise SDRParameterError("Only input one of num_samples or tx_time") raise ValueError("Only input one of num_samples or tx_time")
elif num_samples is not None: elif num_samples is not None:
tx_time = num_samples / self.tx_sample_rate tx_time = num_samples / self.tx_sample_rate
elif tx_time is not None: elif tx_time is not None:
@ -337,7 +320,6 @@ class Pluto(SDR):
data = self._format_tx_data(recording=recording) data = self._format_tx_data(recording=recording)
with self._tx_lock:
try: try:
if self.radio.tx_cyclic_buffer: if self.radio.tx_cyclic_buffer:
print("Destroying existing TX buffer...") print("Destroying existing TX buffer...")
@ -358,76 +340,45 @@ class Pluto(SDR):
if self._tx_initialized is False: if self._tx_initialized is False:
raise RuntimeError("TX was not initialized, init_tx must be called before _stream_tx") raise RuntimeError("TX was not initialized, init_tx must be called before _stream_tx")
if not hasattr(self, "tx_buffer_size"): num_samples = 10000
self.tx_buffer_size = 10000 # TODO remove hardcode
self._enable_tx = True self._enable_tx = True
while self._enable_tx is True: while self._enable_tx is True:
buffer = self._convert_tx_samples(callback(self.tx_buffer_size)) buffer = self._convert_tx_samples(callback(num_samples))
self.radio.tx(buffer[0]) self.radio.tx(buffer[0])
def set_rx_center_frequency(self, center_frequency): def set_rx_center_frequency(self, center_frequency):
"""
Set the center frequency of the receiver. Callable during streaming.
"""
with self._param_lock:
if center_frequency < 70e6 or center_frequency > 6e9:
raise SDRParameterError(
f"{self.__class__.__name__}: Center frequency {center_frequency/1e9:.3f} GHz "
f"out of range:\nStandard:\t[{325e6/1e9:.3f} - {3.8e9/1e9:.3f} GHz]\nHacked:\t"
f"[{70e6/1e9:.3f} - {6e9/1e9:.3f} GHz]"
)
try: try:
self.radio.rx_lo = int(center_frequency) self.radio.rx_lo = int(center_frequency)
self.rx_center_frequency = center_frequency self.rx_center_frequency = center_frequency
except OSError as e: except OSError as e:
raise SDRError(e) _handle_OSError(e)
except ValueError: except ValueError as e:
raise SDRParameterError( _handle_OSError(e)
f"{self.__class__.__name__}: Center frequency {center_frequency/1e9:.3f} GHz "
f"out of range:\nStandard:\t[{325e6/1e9:.3f} - {3.8e9/1e9:.3f} GHz]\nHacked:\t"
f"[{70e6/1e9:.3f} - {6e9/1e9:.3f} GHz]"
)
def set_rx_sample_rate(self, sample_rate): def set_rx_sample_rate(self, sample_rate):
""" self.rx_sample_rate = sample_rate
Set the sample rate of the receiver. Callable during streaming.
""" # TODO add logic for limiting sample rate
with self._param_lock:
min_rate, max_rate = 65.1e3, 61.44e6
if sample_rate < min_rate or sample_rate > max_rate:
raise SDRParameterError(
f"{self.__class__.__name__}: Sample rate {sample_rate/1e6:.3f} Msps "
f"out of range: [{min_rate/1e6:.3f} - {max_rate/1e6:.3f} Msps]"
)
try: try:
# set the sample rate
self.radio.sample_rate = int(sample_rate) self.radio.sample_rate = int(sample_rate)
self.rx_sample_rate = sample_rate
# set the front end filter width # set the front end filter width
self.radio.rx_rf_bandwidth = int(sample_rate) self.radio.rx_rf_bandwidth = int(sample_rate)
except OSError as e: except OSError as e:
raise SDRError(e) _handle_OSError(e)
except ValueError: except ValueError as e:
raise SDRParameterError( _handle_OSError(e)
f"{self.__class__.__name__}: Sample rate {sample_rate/1e6:.3f} Msps "
f"out of range: [{min_rate/1e6:.3f} - {max_rate/1e6:.3f} Msps]"
)
def set_rx_gain(self, gain, channel=0, gain_mode="absolute"): def set_rx_gain(self, gain, channel=0, gain_mode="absolute"):
"""
Set the gain of the receiver. Callable during streaming.
"""
with self._param_lock:
rx_gain_min = 0 rx_gain_min = 0
rx_gain_max = 74 rx_gain_max = 74
if gain_mode == "relative": if gain_mode == "relative":
if gain > 0: if gain > 0:
raise SDRParameterError( raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This sets \ "When gain_mode = 'relative', gain must be < 0. This sets \
the gain relative to the maximum possible gain." the gain relative to the maximum possible gain."
) )
@ -442,7 +393,9 @@ class Pluto(SDR):
print(f"Gain range: {rx_gain_min} to {rx_gain_max} dB") print(f"Gain range: {rx_gain_min} to {rx_gain_max} dB")
self.rx_gain = abs_gain self.rx_gain = abs_gain
try:
if channel == 0: if channel == 0:
if abs_gain is None: if abs_gain is None:
self.radio.gain_control_mode_chan0 = "automatic" self.radio.gain_control_mode_chan0 = "automatic"
print("Using Pluto Automatic Gain Control.") print("Using Pluto Automatic Gain Control.")
@ -466,73 +419,59 @@ class Pluto(SDR):
raise e raise e
else: else:
raise SDRParameterError(f"Pluto channel must be 0 or 1 but was {channel}.") raise ValueError(f"Pluto channel must be 0 or 1 but was {channel}.")
except OSError as e:
_handle_OSError(e)
except ValueError as e:
_handle_OSError(e)
def set_rx_channel(self, channel): def set_rx_channel(self, channel):
if channel == 0: if channel == 0:
self.radio.rx_enabled_channels = [0] self.radio.rx_enabled_channels = [0]
elif channel == 1:
if not self._mimo_capable:
raise SDRParameterError(
"Dual RX channel requested (channel=1) but hardware is not MIMO-capable. "
"Dual RX/TX requires Pluto Rev C/D. Detected hardware: Rev B (single channel only)."
)
self.radio.rx_enabled_channels = [0, 1]
else:
raise SDRParameterError("Channel must be either 0 or 1.")
print(f"Pluto channel(s) = {self.radio.rx_enabled_channels}") print(f"Pluto channel(s) = {self.radio.rx_enabled_channels}")
elif channel == 1:
self.radio.rx_enabled_channels = [0, 1]
print(f"Pluto channel(s) = {self.radio.rx_enabled_channels}")
else:
raise ValueError("Channel must be either 0 or 1.")
def set_rx_buffer_size(self, buffer_size: int): def set_rx_buffer_size(self, buffer_size):
if buffer_size is None: if buffer_size is None:
raise SDRParameterError("Buffer_size must be provided.") raise ValueError("Buffer_size must be provided.")
buffer_size = int(buffer_size)
if buffer_size <= 0: if buffer_size <= 0:
raise SDRParameterError("Buffer_size must be a positive integer.") raise ValueError("Buffer_size must be a positive integer.")
self.rx_buffer_size = buffer_size
if hasattr(self, "radio"): if hasattr(self, "radio"):
try: try:
self.radio.rx_buffer_size = buffer_size self.radio.rx_buffer_size = buffer_size
except Exception as e: except OSError as e:
raise SDRError(e) _handle_OSError(e)
except ValueError as e:
_handle_OSError(e)
def set_tx_center_frequency(self, center_frequency): def set_tx_center_frequency(self, center_frequency):
if center_frequency < 70e6 or center_frequency > 6e9:
raise SDRParameterError(
f"{self.__class__.__name__}: Center frequency {center_frequency/1e9:.3f} GHz "
f"out of range:\nStandard:\t[{325e6/1e9:.3f} - {3.8e9/1e9:.3f} GHz]\nHacked:\t"
f"[{70e6/1e9:.3f} - {6e9/1e9:.3f} GHz]"
)
try: try:
self.radio.tx_lo = int(center_frequency) self.radio.tx_lo = int(center_frequency)
self.tx_center_frequency = center_frequency self.tx_center_frequency = center_frequency
except OSError as e: except OSError as e:
raise SDRError(e) _handle_OSError(e)
except ValueError: except ValueError as e:
raise SDRParameterError( _handle_OSError(e)
f"{self.__class__.__name__}: Center frequency {center_frequency/1e9:.3f} GHz "
f"out of range:\nStandard:\t[{325e6/1e9:.3f} - {3.8e9/1e9:.3f} GHz]\nHacked:\t"
f"[{70e6/1e9:.3f} - {6e9/1e9:.3f} GHz]"
)
def set_tx_sample_rate(self, sample_rate): def set_tx_sample_rate(self, sample_rate):
min_rate, max_rate = 65.1e3, 61.44e6
if sample_rate < min_rate or sample_rate > max_rate:
raise SDRParameterError(
f"{self.__class__.__name__}: Sample rate {sample_rate/1e6:.3f} Msps "
f"out of range: [{min_rate/1e6:.3f} - {max_rate/1e6:.3f} Msps]"
)
try: try:
self.radio.sample_rate = sample_rate self.radio.sample_rate = sample_rate
self.tx_sample_rate = sample_rate self.tx_sample_rate = sample_rate
except OSError as e: except OSError as e:
raise SDRError(e) _handle_OSError(e)
except ValueError: except ValueError as e:
raise SDRParameterError( _handle_OSError(e)
f"{self.__class__.__name__}: Sample rate {sample_rate/1e6:.3f} Msps "
f"out of range: [{min_rate/1e6:.3f} - {max_rate/1e6:.3f} Msps]"
)
def set_tx_gain(self, gain, channel=0, gain_mode="absolute"): def set_tx_gain(self, gain, channel=0, gain_mode="absolute"):
tx_gain_min = -89 tx_gain_min = -89
@ -540,7 +479,7 @@ class Pluto(SDR):
if gain_mode == "relative": if gain_mode == "relative":
if gain > 0: if gain > 0:
raise SDRParameterError( raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This sets\ "When gain_mode = 'relative', gain must be < 0. This sets\
the gain relative to the maximum possible gain." the gain relative to the maximum possible gain."
) )
@ -562,39 +501,34 @@ class Pluto(SDR):
elif channel == 1: elif channel == 1:
self.radio.tx_hardwaregain_chan1 = int(abs_gain) self.radio.tx_hardwaregain_chan1 = int(abs_gain)
else: else:
raise SDRParameterError(f"Pluto channel must be 0 or 1 but was {channel}.") raise ValueError(f"Pluto channel must be 0 or 1 but was {channel}.")
except Exception as e: except OSError as e:
raise SDRError(e) _handle_OSError(e)
except ValueError as e:
_handle_OSError(e)
def set_tx_channel(self, channel): def set_tx_channel(self, channel):
if channel == 0: if channel == 1:
self.radio.tx_enabled_channels = [0]
elif channel == 1:
if not self._mimo_capable:
raise SDRParameterError(
"Dual TX channel requested (channel=1) but hardware is not MIMO-capable. "
"Dual RX/TX requires Pluto Rev C/D. Detected hardware: Rev B (single channel only)."
)
self.radio.tx_enabled_channels = [0, 1] self.radio.tx_enabled_channels = [0, 1]
else:
raise SDRParameterError("Channel must be either 0 or 1.")
print(f"Pluto channel(s) = {self.radio.tx_enabled_channels}") print(f"Pluto channel(s) = {self.radio.tx_enabled_channels}")
elif channel == 0:
self.radio.tx_enabled_channels = [0]
print(f"Pluto channel(s) = {self.radio.tx_enabled_channels}")
else:
raise ValueError("Channel must be either 0 or 1.")
def set_tx_buffer_size(self, buffer_size: int): def set_tx_buffer_size(self, buffer_size):
if buffer_size is None: raise NotImplementedError
raise SDRParameterError("Buffer_size must be provided.")
if buffer_size <= 0:
raise SDRParameterError("Buffer_size must be a positive integer.")
self.tx_buffer_size = buffer_size
def close(self): def close(self):
if self.radio.tx_cyclic_buffer: if self.radio.tx_cyclic_buffer:
self.radio.tx_destroy_buffer() self.radio.tx_destroy_buffer()
del self.radio del self.radio
def shutdown(self):
del self.radio
def _convert_rx_samples(self, samples): def _convert_rx_samples(self, samples):
return samples / (2**11) return samples / (2**11)
@ -604,9 +538,6 @@ class Pluto(SDR):
def set_clock_source(self, source): def set_clock_source(self, source):
raise NotImplementedError raise NotImplementedError
def supports_dynamic_updates(self) -> dict:
return {"center_frequency": True, "sample_rate": True, "gain": True}
def _handle_OSError(e): def _handle_OSError(e):

67
src/ria_toolkit_oss/sdr/rtlsdr.py
View File

@ -12,7 +12,7 @@ except ImportError as exc: # pragma: no cover - dependency provided by end user
raise ImportError("pyrtlsdr is required to use the RTLSDR class") from exc raise ImportError("pyrtlsdr is required to use the RTLSDR class") from exc
from ria_toolkit_oss.datatypes.recording import Recording from ria_toolkit_oss.datatypes.recording import Recording
from ria_toolkit_oss.sdr.sdr import SDR, SDRParameterError from ria_toolkit_oss.sdr.sdr import SDR
class RTLSDR(SDR): class RTLSDR(SDR):
@ -45,7 +45,8 @@ class RTLSDR(SDR):
print(f"Initialized RTL-SDR with identifier [{identifier}].") print(f"Initialized RTL-SDR with identifier [{identifier}].")
except Exception as e: except Exception as e:
raise RuntimeError(f"RTL-SDR: Failed to find device with identifier '{identifier}'\nError: {e}") print(f"Failed to find RTL-SDR with identifier [{identifier}].")
raise e
def init_rx( def init_rx(
self, self,
@ -54,18 +55,18 @@ class RTLSDR(SDR):
gain: Optional[int], gain: Optional[int],
channel: int, channel: int,
gain_mode: Optional[str] = "absolute", gain_mode: Optional[str] = "absolute",
buffer_size: Optional[int] = 256_000,
bias_t: bool = False, bias_t: bool = False,
): ):
if channel not in (0, None): if channel not in (0, None):
raise SDRParameterError("RTL-SDR supports only channel 0 for RX.") raise ValueError("RTL-SDR supports only channel 0 for RX.")
self.set_rx_sample_rate(sample_rate=sample_rate) self.set_rx_sample_rate(sample_rate=sample_rate)
self.set_rx_center_frequency(center_frequency=center_frequency) self.set_rx_center_frequency(center_frequency=center_frequency)
self.set_rx_gain(gain=gain, gain_mode=gain_mode) self.set_rx_gain(gain=gain, gain_mode=gain_mode)
self.rx_buffer_size = int(buffer_size or self.rx_buffer_size)
self.rx_channel = 0 self.rx_channel = 0
self.rx_buffer_size = self._calculate_optimal_buffer_size(sample_rate)
print(f"RTL-SDR buffer: {self.rx_buffer_size} samples for {sample_rate/1e6:.1f} MS/s")
if bias_t: if bias_t:
self.set_bias_tee(True) self.set_bias_tee(True)
@ -77,42 +78,16 @@ class RTLSDR(SDR):
return {"sample_rate": self.rx_sample_rate, "center_frequency": self.rx_center_frequency, "gain": self.rx_gain} return {"sample_rate": self.rx_sample_rate, "center_frequency": self.rx_center_frequency, "gain": self.rx_gain}
def set_rx_sample_rate(self, sample_rate): def set_rx_sample_rate(self, sample_rate):
"""
Set the sample rate of the receiver.
Not callable during recording; RTL-SDR requires stream stop/restart to change sample rate.
"""
if not ((sample_rate > 230e3 and sample_rate < 300e3) or (sample_rate > 900 and sample_rate < 3.2e6)):
raise SDRParameterError(
f"{self.__class__.__name__}: Sample rate {sample_rate/1e6:.3f} Msps "
f"out of range: [{2:.3f} - {20:.3f} Msps]"
)
self.radio.sample_rate = float(sample_rate) self.radio.sample_rate = float(sample_rate)
self.rx_sample_rate = self.radio.sample_rate self.rx_sample_rate = self.radio.sample_rate
print(f"RTL RX Sample Rate = {self.radio.get_sample_rate()}") print(f"RTL RX Sample Rate = {self.radio.get_sample_rate()}")
def set_rx_center_frequency(self, center_frequency): def set_rx_center_frequency(self, center_frequency):
"""
Set the center frequency of the receiver.
Not callable during recording; RTL-SDR requires stream stop/restart to change center frequency.
"""
with self._param_lock:
min_rate, max_rate = 25e6, 1.75e9
if center_frequency < min_rate or center_frequency > max_rate:
raise SDRParameterError(
f"{self.__class__.__name__}: Center frequency {center_frequency/1e9:.3f} GHz "
f"out of range: [{min_rate/1e9:.3f} - {max_rate/1e9:.3f} GHz]"
)
self.radio.center_freq = float(center_frequency) self.radio.center_freq = float(center_frequency)
self.rx_center_frequency = self.radio.center_freq self.rx_center_frequency = self.radio.center_freq
print(f"RTL RX Center Frequency = {self.radio.get_center_freq()}") print(f"RTL RX Center Frequency = {self.radio.get_center_freq()}")
def set_rx_gain(self, gain, gain_mode="absolute"): def set_rx_gain(self, gain, gain_mode="absolute"):
"""
Set the gain of the receiver. Callable during streaming.
"""
with self._param_lock:
available_gains = self.radio.get_gains() available_gains = self.radio.get_gains()
if gain is None: if gain is None:
@ -131,7 +106,7 @@ class RTLSDR(SDR):
if gain_mode == "relative": if gain_mode == "relative":
if gain > 0: if gain > 0:
raise SDRParameterError( raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This sets\ "When gain_mode = 'relative', gain must be < 0. This sets\
the gain relative to the maximum possible gain." the gain relative to the maximum possible gain."
) )
@ -154,25 +129,7 @@ class RTLSDR(SDR):
print(f"RTL RX Gain = {self.radio.get_gain()}") print(f"RTL RX Gain = {self.radio.get_gain()}")
print(f"Available RTL RX Gains: {available_gains}") print(f"Available RTL RX Gains: {available_gains}")
def _calculate_optimal_buffer_size(self, sample_rate): def record(self, num_samples: Optional[int] = None, rx_time: Optional[int | float] = None):
"""USB packet alignment for stability."""
# RTL-SDR USB transfers in 16k chunks
min_size = 16384
max_size = 262144 # 256k
# Target: 50ms of data per buffer
target = int(sample_rate * 0.05)
# Round up to 16k boundary
size = ((target + 16383) // 16384) * 16384
return max(min_size, min(size, max_size))
def record(
self,
num_samples: Optional[int] = None,
rx_time: Optional[int | float] = None,
) -> Recording:
""" """
Create a radio recording (iq samples and metadata) of a given length from the RTL-SDR. Create a radio recording (iq samples and metadata) of a given length from the RTL-SDR.
Either num_samples or rx_time must be provided. Either num_samples or rx_time must be provided.
@ -190,13 +147,13 @@ class RTLSDR(SDR):
raise RuntimeError("RX was not initialized. init_rx() must be called before record().") raise RuntimeError("RX was not initialized. init_rx() must be called before record().")
if num_samples is not None and rx_time is not None: if num_samples is not None and rx_time is not None:
raise SDRParameterError("Only input one of num_samples or rx_time") raise ValueError("Only input one of num_samples or rx_time")
elif num_samples is not None: elif num_samples is not None:
pass pass
elif rx_time is not None: elif rx_time is not None:
num_samples = int(rx_time * self.rx_sample_rate) num_samples = int(rx_time * self.rx_sample_rate)
else: else:
raise SDRParameterError("Must provide input of one of num_samples or rx_time") raise ValueError("Must provide input of one of num_samples or rx_time")
# RTL-SDR has USB buffer limitations - use consistent 256k chunks # RTL-SDR has USB buffer limitations - use consistent 256k chunks
# Always read full chunks to avoid USB overflow issues with partial reads # Always read full chunks to avoid USB overflow issues with partial reads
@ -275,10 +232,6 @@ class RTLSDR(SDR):
def close(self): def close(self):
try: try:
self.radio.close() self.radio.close()
del self.radio
finally: finally:
self._enable_rx = False self._enable_rx = False
self._enable_tx = False self._enable_tx = False
def supports_dynamic_updates(self) -> dict:
return {"center_frequency": False, "sample_rate": False, "gain": True}

159
src/ria_toolkit_oss/sdr/sdr.py
View File

@ -1,6 +1,5 @@
import math import math
import pickle import pickle
import threading
import warnings import warnings
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from typing import Optional from typing import Optional
@ -28,21 +27,17 @@ class SDR(ABC):
self._tx_initialized = False self._tx_initialized = False
self._enable_rx = False self._enable_rx = False
self._enable_tx = False self._enable_tx = False
self._accumulated_buffer = None self._accumulated_buffer = None
self._max_num_buffers = None self._max_num_buffers = None
self._num_buffers_processed = 0 self._num_buffers_processed = 0
self._accumulated_buffer = None self._accumulated_buffer = None
self._last_buffer = None self._last_buffer = None
self._corrupted_buffer_count = 0
self.rx_sample_rate = None self.rx_sample_rate = None
self.rx_center_frequency = None self.rx_center_frequency = None
self.rx_gain = None self.rx_gain = None
self.tx_sample_rate = None self.tx_sample_rate = None
self.tx_center_frequency = None self.tx_center_frequency = None
self.tx_gain = None self.tx_gain = None
self._param_lock = threading.RLock() # Reentrant lock
def record(self, num_samples: Optional[int] = None, rx_time: Optional[int | float] = None) -> Recording: def record(self, num_samples: Optional[int] = None, rx_time: Optional[int | float] = None) -> Recording:
""" """
@ -76,6 +71,7 @@ class SDR(ABC):
self._max_num_buffers = num_buffers self._max_num_buffers = num_buffers
self._num_buffers_processed = 0 self._num_buffers_processed = 0
self._num_buffers_processed = 0
self._last_buffer = None self._last_buffer = None
self._accumulated_buffer = None self._accumulated_buffer = None
print("Starting stream") print("Starting stream")
@ -98,7 +94,6 @@ class SDR(ABC):
# reset to record again # reset to record again
self._accumulated_buffer = None self._accumulated_buffer = None
self._num_buffers_processed = 0
return recording return recording
def stream_to_zmq(self, zmq_address, n_samples: int, buffer_size: Optional[int] = 10000): def stream_to_zmq(self, zmq_address, n_samples: int, buffer_size: Optional[int] = 10000):
@ -115,7 +110,7 @@ class SDR(ABC):
:return: The trimmed Recording. :return: The trimmed Recording.
:rtype: Recording :rtype: Recording
""" """
try:
self._previous_buffer = None self._previous_buffer = None
self._max_num_buffers = np.inf if n_samples == np.inf else math.ceil(n_samples / buffer_size) self._max_num_buffers = np.inf if n_samples == np.inf else math.ceil(n_samples / buffer_size)
self._num_buffers_processed = 0 self._num_buffers_processed = 0
@ -127,11 +122,9 @@ class SDR(ABC):
self._stream_rx( self._stream_rx(
self._zmq_bytestream_callback, self._zmq_bytestream_callback,
) )
finally:
if hasattr(self, "socket"):
self.socket.close()
if hasattr(self, "context"):
self.context.destroy() self.context.destroy()
self.socket.close()
def _accumulate_buffers_callback(self, buffer, metadata=None): def _accumulate_buffers_callback(self, buffer, metadata=None):
""" """
@ -141,72 +134,62 @@ class SDR(ABC):
# save the buffer until max reached # save the buffer until max reached
# return a recording # return a recording
# Validate buffer
if not self._validate_buffer(buffer):
print("Warning: Corrupted buffer detected, skipping")
self._corrupted_buffer_count += 1
return # Skip this buffer
if isinstance(buffer, np.ndarray):
if buffer.ndim == 1:
buffer = buffer[np.newaxis, :] # make shape (1, N)
else:
buffer = np.array(buffer) # make it 1d buffer = np.array(buffer) # make it 1d
if len(buffer.shape) == 1: if len(buffer.shape) == 1:
buffer = np.array([buffer]) buffer = np.array([buffer])
# First call: pre-allocate if we know the final size # it runs these checks each time, is that an efficiency issue?
if self._accumulated_buffer is None:
# Check that _max_num_buffers is set
if self._max_num_buffers is None: if self._max_num_buffers is None:
# default then
# this should probably print, but that would happen every buffer...
raise ValueError("Number of buffers for block capture not set.") raise ValueError("Number of buffers for block capture not set.")
if self._num_samples_to_record is None:
raise ValueError("Number of samples not set before RX start.") # add the given buffer to the pre-allocated buffer
if metadata is not None: if metadata is not None:
self.received_metadata = metadata self.received_metadata = metadata
# Preallocate once (avoid np.zeros; use np.empty for speed) # TODO optimize, pre-allocate
num_channels = buffer.shape[0] if self._accumulated_buffer is not None:
self._accumulated_buffer = np.empty((num_channels, self._num_samples_to_record), dtype=buffer.dtype) self._accumulated_buffer = np.concatenate((self._accumulated_buffer, buffer), axis=1)
self._write_position = 0 else:
print(f"Pre-allocated buffer for {self._num_samples_to_record:,} samples.") # the first time
self._accumulated_buffer = buffer.copy()
# Write new buffer into pre-allocated array self._num_buffers_processed = self._num_buffers_processed + 1
n = buffer.shape[1]
start = self._write_position
end = min(start + n, self._num_samples_to_record)
samples_to_write = end - start
if samples_to_write > 0:
self._accumulated_buffer[:, start:end] = buffer[:, : end - start]
self._write_position = end
# Check if we're done
self._num_buffers_processed += 1
if self._num_buffers_processed >= self._max_num_buffers: if self._num_buffers_processed >= self._max_num_buffers:
self.stop() self.stop()
def _validate_buffer(self, buffer): if self._last_buffer is not None:
"""Check for obviously corrupt data.""" if (buffer == self._last_buffer).all():
# Check for all zeros print("\033[93mWarning: Buffer Overflow Detected\033[0m")
if np.all(buffer == 0): self._last_buffer = buffer.copy()
return False else:
# Check for all same value self._last_buffer = buffer.copy()
if np.all(buffer == buffer[0]):
return False # print("Number of buffers received: " + str(self._num_buffers_processed))
return True
def _zmq_bytestream_callback(self, buffer, metadata=None): def _zmq_bytestream_callback(self, buffer, metadata=None):
# push to ZMQ port # push to ZMQ port
data = np.array(buffer).tobytes() # convert to bytes for transport data = np.array(buffer).tobytes() # convert to bytes for transport
self.socket.send(data) self.socket.send(data)
# print(f"Sent {self._num_buffers_processed} ZMQ buffers to {self.zmq_address}")
self._num_buffers_processed = self._num_buffers_processed + 1 self._num_buffers_processed = self._num_buffers_processed + 1
if self._max_num_buffers is not None: if self._max_num_buffers is not None:
if self._num_buffers_processed >= self._max_num_buffers: if self._num_buffers_processed >= self._max_num_buffers:
self.pause_rx() self.pause_rx()
if self._previous_buffer is not None:
if (buffer == self._previous_buffer).all():
print("\033[93mWarning: Buffer Overflow Detected\033[0m")
# TODO: I suggest we think about moving this part to the top of this function
# and skip the rest of the function in case of overflow.
# like, it's not necessary to stream repeated IQ data anyways!
self._previous_buffer = buffer.copy()
def pickle_buffer_to_zmq(self, zmq_address, buffer_size, num_buffers): def pickle_buffer_to_zmq(self, zmq_address, buffer_size, num_buffers):
""" """
Stream samples to a zmq address, packaged in binary buffers using numpy.pickle. Stream samples to a zmq address, packaged in binary buffers using numpy.pickle.
@ -246,7 +229,7 @@ class SDR(ABC):
self.stop() self.stop()
if self._last_buffer is not None: if self._last_buffer is not None:
if np.array_equal(buffer, self._last_buffer): if (buffer == self._last_buffer).all():
print("\033[93mWarning: Buffer Overflow Detected\033[0m") print("\033[93mWarning: Buffer Overflow Detected\033[0m")
self._last_buffer = buffer.copy() self._last_buffer = buffer.copy()
else: else:
@ -390,58 +373,6 @@ class SDR(ABC):
""" """
return self.tx_gain return self.tx_gain
def set_rx_sample_rate(self):
"""
Set the sample rate of the receiver.
"""
raise NotImplementedError
def set_rx_center_frequency(self):
"""
Set the center frequency of the receiver.
"""
raise NotImplementedError
def set_rx_gain(self):
"""
Set the gain setting of the receiver.
"""
raise NotImplementedError
def set_tx_sample_rate(self):
"""
Set the sample rate of the transmitter.
"""
raise NotImplementedError
def set_tx_center_frequency(self):
"""
Set the center frequency of the transmitter.
"""
raise NotImplementedError
def set_tx_gain(self):
"""
Set the gain setting of the transmitter.
"""
raise NotImplementedError
def supports_dynamic_updates(self) -> dict:
"""
Report which parameters can be updated during streaming.
Returns:
dict: {'center_frequency': bool, 'sample_rate': bool, 'gain': bool}
"""
return {"center_frequency": False, "sample_rate": False, "gain": False}
def __del__(self):
"""Cleanup on garbage collection."""
try:
self.close()
except Exception:
pass
@abstractmethod @abstractmethod
def close(self): def close(self):
pass pass
@ -511,21 +442,3 @@ def _verify_sample_format(samples):
""" """
return np.max(np.abs(samples)) <= 1 return np.max(np.abs(samples)) <= 1
class SDRError(Exception):
"""Base exception for SDR errors."""
pass
class SDRParameterError(SDRError):
"""Invalid parameter (sample rate, freq, gain)."""
pass
class SDROverflowError(SDRError):
"""Buffer overflow detected."""
pass

62
src/ria_toolkit_oss/sdr/thinkrf.py
View File

@ -36,7 +36,7 @@ except SyntaxError as exc: # pragma: no cover - Python 2/3 compatibility issue
print("Manual fix: Run `python scripts/fix_pyrf_python3.py` from ria-toolkit-oss directory") print("Manual fix: Run `python scripts/fix_pyrf_python3.py` from ria-toolkit-oss directory")
raise exc raise exc
from ria_toolkit_oss.sdr.sdr import SDR, SDRParameterError from ria_toolkit_oss.sdr.sdr import SDR
class ThinkRF(SDR): class ThinkRF(SDR):
@ -51,7 +51,7 @@ class ThinkRF(SDR):
super().__init__() super().__init__()
if identifier is None: if identifier is None:
raise SDRParameterError("ThinkRF requires an IP address or hostname identifier") raise ValueError("ThinkRF requires an IP address or hostname identifier")
self.identifier = identifier self.identifier = identifier
try: try:
@ -90,7 +90,7 @@ class ThinkRF(SDR):
mode = capture_mode.lower() mode = capture_mode.lower()
if mode not in {"block", "stream"}: if mode not in {"block", "stream"}:
raise SDRParameterError("capture_mode must be either 'block' or 'stream'") raise ValueError("capture_mode must be either 'block' or 'stream'")
self._rfe_mode = rfe_mode self._rfe_mode = rfe_mode
self._attenuation = int(max(0, min(attenuation, 30))) self._attenuation = int(max(0, min(attenuation, 30)))
@ -113,12 +113,10 @@ class ThinkRF(SDR):
decimation: Optional[int] = None, decimation: Optional[int] = None,
): ):
if channel not in (0, None): if channel not in (0, None):
raise SDRParameterError("ThinkRF supports only channel 0 for RX.") raise ValueError("ThinkRF devices expose a single receive channel")
stream_mode = getattr(self, "_capture_mode", "block") == "stream" stream_mode = getattr(self, "_capture_mode", "block") == "stream"
actual_decimation, _ = self.set_rx_sample_rate( actual_decimation, actual_sample_rate = self.set_rx_sample_rate(sample_rate=sample_rate, decimation=decimation)
sample_rate=sample_rate, decimation=decimation, stream_mode=stream_mode
)
self.radio.reset() self.radio.reset()
self.radio.scpiset(":SYSTEM:FLUSH") self.radio.scpiset(":SYSTEM:FLUSH")
@ -129,7 +127,15 @@ class ThinkRF(SDR):
self.radio.rfe_mode(self._rfe_mode) self.radio.rfe_mode(self._rfe_mode)
self.set_rx_center_frequency(center_frequency=center_frequency) self.set_rx_center_frequency(center_frequency=center_frequency)
self.set_rx_gain(gain=gain, gain_mode=gain_mode, actual_decimation=actual_decimation)
attenuation = self._attenuation if gain is None else int(gain) # gain
attenuation = max(0, min(attenuation, 30))
self.radio.attenuator(attenuation)
gain_profile = self._gain_profile
if gain_mode and isinstance(gain_mode, str) and gain_mode.upper() in {"LOW", "MEDIUM", "HIGH", "VLOW"}:
gain_profile = gain_mode.upper()
self.radio.gain(gain_profile.lower()) # WSA.gain() expects lowercase
self.radio.decimation(actual_decimation) self.radio.decimation(actual_decimation)
if stream_mode: if stream_mode:
@ -147,6 +153,14 @@ class ThinkRF(SDR):
self.radio.scpiset(f":TRACE:BLOCK:PACKETS {self._packets_per_block}") self.radio.scpiset(f":TRACE:BLOCK:PACKETS {self._packets_per_block}")
self.radio.scpiset(":TRACE:BLOCK:DATA?") self.radio.scpiset(":TRACE:BLOCK:DATA?")
self.rx_gain = {
"attenuation_dB": attenuation,
"profile": gain_profile,
"decimation": actual_decimation,
"rfe_mode": self._rfe_mode,
"spp": self._samples_per_packet,
"ppb": self._packets_per_block,
}
self.rx_buffer_size = self._samples_per_packet self.rx_buffer_size = self._samples_per_packet
self.rx_channel = 0 self.rx_channel = 0
@ -154,10 +168,6 @@ class ThinkRF(SDR):
self._tx_initialized = False self._tx_initialized = False
def set_rx_sample_rate(self, sample_rate, decimation, stream_mode): def set_rx_sample_rate(self, sample_rate, decimation, stream_mode):
"""
Set the sample rate of the receiver.
Not callable during recording; ThinkRF requires stream stop/restart to change sample rate.
"""
# Enforce sample rate / decimation # Enforce sample rate / decimation
# Note: decimation parameter takes precedence if provided # Note: decimation parameter takes precedence if provided
actual_decimation, actual_sample_rate = self.enforce_sample_rate(sample_rate, decimation) actual_decimation, actual_sample_rate = self.enforce_sample_rate(sample_rate, decimation)
@ -178,33 +188,10 @@ class ThinkRF(SDR):
return actual_decimation, actual_sample_rate return actual_decimation, actual_sample_rate
def set_rx_center_frequency(self, center_frequency): def set_rx_center_frequency(self, center_frequency):
"""
Set the center frequency of the receiver. Callable during streaming.
"""
with self._param_lock:
self.radio.freq(int(center_frequency)) self.radio.freq(int(center_frequency))
self.rx_center_frequency = self.radio.freq self.rx_center_frequency = self.radio.freq
print(f"ThinkRF RX Center Frequency = {self.radio.freq}") print(f"ThinkRF RX Center Frequency = {self.radio.freq}")
def set_rx_gain(self, gain, gain_mode, actual_decimation):
attenuation = self._attenuation if gain is None else int(gain) # gain
attenuation = max(0, min(attenuation, 30))
self.radio.attenuator(attenuation)
gain_profile = self._gain_profile
if gain_mode and isinstance(gain_mode, str) and gain_mode.upper() in {"LOW", "MEDIUM", "HIGH", "VLOW"}:
gain_profile = gain_mode.upper()
self.radio.gain(gain_profile.lower()) # WSA.gain() expects lowercase
self.rx_gain = {
"attenuation_dB": attenuation,
"profile": gain_profile,
"decimation": actual_decimation,
"rfe_mode": self._rfe_mode,
"spp": self._samples_per_packet,
"ppb": self._packets_per_block,
}
def _stream_rx(self, callback): def _stream_rx(self, callback):
if not self._rx_initialized: if not self._rx_initialized:
raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record().") raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record().")
@ -444,7 +431,7 @@ class ThinkRF(SDR):
For decimation 1 or 2, block captures are limited by onboard RAM. For decimation 1 or 2, block captures are limited by onboard RAM.
""" """
if decimation <= 2 and num_samples > self.MAX_ONBOARD_SAMPLES: if decimation <= 2 and num_samples > self.MAX_ONBOARD_SAMPLES:
raise SDRParameterError( raise ValueError(
f"ThinkRF: Cannot capture {num_samples} samples at decimation {decimation}. " f"ThinkRF: Cannot capture {num_samples} samples at decimation {decimation}. "
f"Onboard RAM limit is ~{self.MAX_ONBOARD_SAMPLES} samples for dec 1/2. " f"Onboard RAM limit is ~{self.MAX_ONBOARD_SAMPLES} samples for dec 1/2. "
f"Either reduce num_samples or use stream mode (increase decimation to >=4)." f"Either reduce num_samples or use stream mode (increase decimation to >=4)."
@ -459,6 +446,3 @@ class ThinkRF(SDR):
"fstop": int(center_frequency) + half, "fstop": int(center_frequency) + half,
"amplitude": -100, "amplitude": -100,
} }
def supports_dynamic_updates(self) -> dict:
return {"center_frequency": True, "sample_rate": False, "gain": False}

150
src/ria_toolkit_oss/sdr/usrp.py
View File

@ -7,7 +7,7 @@ import numpy as np
import uhd import uhd
from ria_toolkit_oss.datatypes.recording import Recording from ria_toolkit_oss.datatypes.recording import Recording
from ria_toolkit_oss.sdr.sdr import SDR, SDRParameterError from ria_toolkit_oss.sdr.sdr import SDR
class USRP(SDR): class USRP(SDR):
@ -40,7 +40,7 @@ class USRP(SDR):
channel: int, channel: int,
gain: int, gain: int,
gain_mode: Optional[str] = "absolute", gain_mode: Optional[str] = "absolute",
rx_buffer_size: Optional[int] = None, rx_buffer_size: int = 960000,
): ):
""" """
Initializes the USRP for receiving. Initializes the USRP for receiving.
@ -63,6 +63,8 @@ class USRP(SDR):
:rtype: dict :rtype: dict
""" """
self.rx_buffer_size = rx_buffer_size
# build USRP object # build USRP object
usrp_args = _generate_usrp_config_string(sample_rate=sample_rate, device_dict=self.device_dict) usrp_args = _generate_usrp_config_string(sample_rate=sample_rate, device_dict=self.device_dict)
self.usrp = uhd.usrp.MultiUSRP(usrp_args) self.usrp = uhd.usrp.MultiUSRP(usrp_args)
@ -70,7 +72,7 @@ class USRP(SDR):
# check if channel arg is valid # check if channel arg is valid
max_num_channels = self.usrp.get_rx_num_channels() max_num_channels = self.usrp.get_rx_num_channels()
if channel + 1 > max_num_channels: if channel + 1 > max_num_channels:
raise SDRParameterError(f"Channel {channel} not valid for device with {max_num_channels} channels.") raise IOError(f"Channel {channel} not valid for device with {max_num_channels} channels.")
self.set_rx_sample_rate(sample_rate=sample_rate, channel=channel) self.set_rx_sample_rate(sample_rate=sample_rate, channel=channel)
self.set_rx_center_frequency(center_frequency=center_frequency, channel=channel) self.set_rx_center_frequency(center_frequency=center_frequency, channel=channel)
@ -79,20 +81,6 @@ class USRP(SDR):
self.rx_channel = channel self.rx_channel = channel
print(f"USRP RX Channel = {self.rx_channel}") print(f"USRP RX Channel = {self.rx_channel}")
stream_args = uhd.usrp.StreamArgs("fc32", "sc16")
stream_args.channels = [self.rx_channel]
self.metadata = uhd.types.RXMetadata()
self.rx_stream = self.usrp.get_rx_stream(stream_args)
if rx_buffer_size is None: # In case it's none
self.rx_buffer_size = self.rx_stream.get_max_num_samps()
else:
self.rx_buffer_size = rx_buffer_size
# set timeout based on buffer size and sample rate, with a safety factor of 5
self.timeout = (self.rx_buffer_size / self.rx_sample_rate) * 5
# flag to prevent user from calling certain functions before this one. # flag to prevent user from calling certain functions before this one.
self._rx_initialized = True self._rx_initialized = True
self._tx_initialized = False self._tx_initialized = False
@ -100,55 +88,41 @@ class USRP(SDR):
return {"sample_rate": self.rx_sample_rate, "center_frequency": self.rx_center_frequency, "gain": self.rx_gain} return {"sample_rate": self.rx_sample_rate, "center_frequency": self.rx_center_frequency, "gain": self.rx_gain}
def set_rx_sample_rate(self, sample_rate, channel=0): def set_rx_sample_rate(self, sample_rate, channel=0):
"""
Set the sample rate of the receiver. Callable during streaming.
"""
# check if sample rate arg is valid # check if sample rate arg is valid
# Note: B200/B210 devices auto-adjust master clock rate, so get_rx_rates() returns # Note: B200/B210 devices auto-adjust master clock rate, so get_rx_rates() returns
# the range for the CURRENT master clock, not the maximum possible range. # the range for the CURRENT master clock, not the maximum possible range.
# Skip validation for B-series devices and let UHD handle it. # Skip validation for B-series devices and let UHD handle it.
with self._param_lock:
device_type = self.device_dict.get("type", "").lower() device_type = self.device_dict.get("type", "").lower()
if device_type not in ["b200", "b210"]: if device_type not in ["b200", "b210"]:
sample_rate_range = self.usrp.get_rx_rates() sample_rate_range = self.usrp.get_rx_rates()
min_rate, max_rate = sample_rate_range.start(), sample_rate_range.stop() if sample_rate < sample_rate_range.start() or sample_rate > sample_rate_range.stop():
if sample_rate < min_rate or sample_rate > max_rate: raise IOError(
raise SDRParameterError( f"Sample rate {sample_rate} not valid for this USRP.\nValid\
f"{self.__class__.__name__}: Sample rate {sample_rate/1e6:.3f} Msps " range is {sample_rate_range.start()}\
f"out of range: [{min_rate/1e6:.3f} - {max_rate/1e6:.3f} Msps]" to {sample_rate_range.stop()}."
) )
self.usrp.set_rx_rate(sample_rate, channel) self.usrp.set_rx_rate(sample_rate, channel)
self.rx_sample_rate = self.usrp.get_rx_rate(channel) self.rx_sample_rate = self.usrp.get_rx_rate(channel)
print(f"USRP RX Sample Rate = {self.rx_sample_rate}") print(f"USRP RX Sample Rate = {self.rx_sample_rate}")
def set_rx_center_frequency(self, center_frequency, channel=0): def set_rx_center_frequency(self, center_frequency, channel=0):
"""
Set the center frequency of the receiver. Callable during streaming.
"""
with self._param_lock:
center_frequency_range = self.usrp.get_rx_freq_range() center_frequency_range = self.usrp.get_rx_freq_range()
min_rate, max_rate = center_frequency_range.start(), center_frequency_range.stop() if center_frequency < center_frequency_range.start() or center_frequency > center_frequency_range.stop():
if center_frequency < min_rate or center_frequency > max_rate: raise IOError(
raise SDRParameterError( f"Center frequency {center_frequency} out of range for USRP.\
f"{self.__class__.__name__}: Center frequency {center_frequency/1e9:.3f} GHz " \nValid range is {center_frequency_range.start()} \
f"out of range: [{min_rate/1e9:.3f} - {max_rate/1e9:.3f} GHz]" to {center_frequency_range.stop()}."
) )
self.usrp.set_rx_freq(uhd.libpyuhd.types.tune_request(center_frequency), channel) self.usrp.set_rx_freq(uhd.libpyuhd.types.tune_request(center_frequency), channel)
self.rx_center_frequency = self.usrp.get_rx_freq(channel) self.rx_center_frequency = self.usrp.get_rx_freq(channel)
print(f"USRP RX Center Frequency = {self.rx_center_frequency}") print(f"USRP RX Center Frequency = {self.rx_center_frequency}")
def set_rx_gain(self, gain, gain_mode="absolute", channel=0): def set_rx_gain(self, gain, gain_mode="absolute", channel=0):
"""
Set the gain of the receiver. Callable during streaming.
"""
with self._param_lock:
# check if gain arg is valid # check if gain arg is valid
gain_range = self.usrp.get_rx_gain_range() gain_range = self.usrp.get_rx_gain_range()
if gain_mode == "relative": if gain_mode == "relative":
if gain > 0: if gain > 0:
raise SDRParameterError( raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This sets\ "When gain_mode = 'relative', gain must be < 0. This sets\
the gain relative to the maximum possible gain." the gain relative to the maximum possible gain."
) )
@ -166,10 +140,16 @@ class USRP(SDR):
print(f"USRP RX Gain = {self.rx_gain}") print(f"USRP RX Gain = {self.rx_gain}")
def _stream_rx(self, callback): def _stream_rx(self, callback):
if not self._rx_initialized: if not self._rx_initialized:
raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()") raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()")
# send command to start the rx stream stream_args = uhd.usrp.StreamArgs("fc32", "sc16")
stream_args.channels = [self.rx_channel]
self.metadata = uhd.types.RXMetadata()
self.rx_stream = self.usrp.get_rx_stream(stream_args)
stream_command = uhd.types.StreamCMD(uhd.types.StreamMode.start_cont) stream_command = uhd.types.StreamCMD(uhd.types.StreamMode.start_cont)
stream_command.stream_now = True stream_command.stream_now = True
self.rx_stream.issue_stream_cmd(stream_command) self.rx_stream.issue_stream_cmd(stream_command)
@ -180,19 +160,19 @@ class USRP(SDR):
receive_buffer = np.zeros((1, self.rx_buffer_size), dtype=np.complex64) receive_buffer = np.zeros((1, self.rx_buffer_size), dtype=np.complex64)
while self._enable_rx: while self._enable_rx:
self.rx_stream.recv(receive_buffer, self.metadata, self.timeout)
# 1 is the timeout #TODO maybe set this intelligently based on the desired sample rate
self.rx_stream.recv(receive_buffer, self.metadata, 1)
# TODO set metadata correctly, sending real sample rate plus any error codes # TODO set metadata correctly, sending real sample rate plus any error codes
# sending complex signal # sending complex signal
callback(buffer=receive_buffer, metadata=self.metadata) callback(buffer=receive_buffer, metadata=self.metadata)
if self.metadata.error_code != uhd.types.RXMetadataErrorCode.none: if self.metadata.error_code != uhd.types.RXMetadataErrorCode.none:
if self.metadata.error_code == uhd.types.RXMetadataErrorCode.overflow: print(f"Error while receiving samples: {self.metadata.strerror()}")
print("\033[93mWarning: Buffer Overflow Detected.\033[0m")
if self.metadata.error_code == uhd.types.RXMetadataErrorCode.timeout: if self.metadata.error_code == uhd.types.RXMetadataErrorCode.timeout:
print("\033[91Stopping receive due to timeout error.\033[0m") print("Stopping receive due to timeout error.")
self.stop() self.stop()
# stop streaming
wait_time = 0.1 wait_time = 0.1
stop_time = self.usrp.get_time_now() + wait_time stop_time = self.usrp.get_time_now() + wait_time
stop_cmd = uhd.types.StreamCMD(uhd.types.StreamMode.stop_cont) stop_cmd = uhd.types.StreamCMD(uhd.types.StreamMode.stop_cont)
@ -200,14 +180,10 @@ class USRP(SDR):
stop_cmd.time_spec = stop_time stop_cmd.time_spec = stop_time
self.rx_stream.issue_stream_cmd(stop_cmd) self.rx_stream.issue_stream_cmd(stop_cmd)
time.sleep(wait_time) # TODO figure out what a realistic wait time is here. time.sleep(wait_time) # TODO figure out what a realistic wait time is here.
del self.rx_stream
print("USRP RX Completed.") print("USRP RX Completed.")
def record( def record(self, num_samples: Optional[int] = None, rx_time: Optional[int | float] = None):
self,
num_samples: Optional[int] = None,
rx_time: Optional[int | float] = None,
) -> Recording:
""" """
Create a radio recording (iq samples and metadata) of a given length from the USRP. Create a radio recording (iq samples and metadata) of a given length from the USRP.
Either num_samples or rx_time must be provided. Either num_samples or rx_time must be provided.
@ -224,31 +200,41 @@ class USRP(SDR):
raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()") raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()")
if num_samples is not None and rx_time is not None: if num_samples is not None and rx_time is not None:
raise SDRParameterError("Only input one of num_samples or rx_time") raise ValueError("Only input one of num_samples or rx_time")
elif num_samples is not None: elif num_samples is not None:
pass pass
elif rx_time is not None: elif rx_time is not None:
num_samples = int(rx_time * self.rx_sample_rate) num_samples = int(rx_time * self.rx_sample_rate)
else: else:
raise SDRParameterError("Must provide input of one of num_samples or rx_time") raise ValueError("Must provide input of one of num_samples or rx_time")
stream_args = uhd.usrp.StreamArgs("fc32", "sc16")
stream_args.channels = [self.rx_channel]
self.metadata = uhd.types.RXMetadata()
self.rx_stream = self.usrp.get_rx_stream(stream_args)
# send command to start the rx stream
stream_command = uhd.types.StreamCMD(uhd.types.StreamMode.start_cont) stream_command = uhd.types.StreamCMD(uhd.types.StreamMode.start_cont)
stream_command.stream_now = True stream_command.stream_now = True
self.rx_stream.issue_stream_cmd(stream_command) self.rx_stream.issue_stream_cmd(stream_command)
# receive loop # receive loop
self._enable_rx = True self._enable_rx = True
print("USRP Starting RX...")
store_array = np.zeros((1, (num_samples // self.rx_buffer_size + 1) * self.rx_buffer_size), dtype=np.complex64) store_array = np.zeros((1, (num_samples // self.rx_buffer_size + 1) * self.rx_buffer_size), dtype=np.complex64)
receive_buffer = np.zeros((1, self.rx_buffer_size), dtype=np.complex64) receive_buffer = np.zeros((1, self.rx_buffer_size), dtype=np.complex64)
print("USRP Starting RX...")
# write complex samples to receive buffer
for i in range(num_samples // self.rx_buffer_size + 1): for i in range(num_samples // self.rx_buffer_size + 1):
self.rx_stream.recv(receive_buffer, self.metadata, self.timeout)
# write samples to receive buffer
# they should already be complex
# 1 is the timeout #TODO maybe set this intelligently based on the desired sample rate
self.rx_stream.recv(receive_buffer, self.metadata, 1)
# TODO set metadata correctly, sending real sample rate plus any error codes
# sending complex signal
store_array[:, i * self.rx_buffer_size : (i + 1) * self.rx_buffer_size] = receive_buffer store_array[:, i * self.rx_buffer_size : (i + 1) * self.rx_buffer_size] = receive_buffer
# stop streaming
wait_time = 0.1 wait_time = 0.1
stop_time = self.usrp.get_time_now() + wait_time stop_time = self.usrp.get_time_now() + wait_time
stop_cmd = uhd.types.StreamCMD(uhd.types.StreamMode.stop_cont) stop_cmd = uhd.types.StreamCMD(uhd.types.StreamMode.stop_cont)
@ -256,7 +242,7 @@ class USRP(SDR):
stop_cmd.time_spec = stop_time stop_cmd.time_spec = stop_time
self.rx_stream.issue_stream_cmd(stop_cmd) self.rx_stream.issue_stream_cmd(stop_cmd)
time.sleep(wait_time) # TODO figure out what a realistic wait time is here. time.sleep(wait_time) # TODO figure out what a realistic wait time is here.
del self.rx_stream
print("USRP RX Completed.") print("USRP RX Completed.")
metadata = { metadata = {
"source": self.__class__.__name__, "source": self.__class__.__name__,
@ -301,7 +287,7 @@ class USRP(SDR):
# check if channel arg is valid # check if channel arg is valid
max_num_channels = self.usrp.get_rx_num_channels() max_num_channels = self.usrp.get_rx_num_channels()
if channel + 1 > max_num_channels: if channel + 1 > max_num_channels:
raise SDRParameterError(f"Channel {channel} not valid for device with {max_num_channels} channels.") raise IOError(f"Channel {channel} not valid for device with {max_num_channels} channels.")
self.set_tx_sample_rate(sample_rate=sample_rate, channel=channel) self.set_tx_sample_rate(sample_rate=sample_rate, channel=channel)
self.set_tx_center_frequency(center_frequency=center_frequency, channel=channel) self.set_tx_center_frequency(center_frequency=center_frequency, channel=channel)
@ -327,26 +313,23 @@ class USRP(SDR):
device_type = self.device_dict.get("type", "").lower() device_type = self.device_dict.get("type", "").lower()
if device_type not in ["b200", "b210"]: if device_type not in ["b200", "b210"]:
sample_rate_range = self.usrp.get_tx_rates() sample_rate_range = self.usrp.get_tx_rates()
min_rate, max_rate = sample_rate_range.start(), sample_rate_range.stop() if sample_rate < sample_rate_range.start() or sample_rate > sample_rate_range.stop():
if sample_rate < min_rate or sample_rate > max_rate: raise IOError(
raise SDRParameterError( f"Sample rate {sample_rate} not valid for this USRP.\nValid\
f"{self.__class__.__name__}: Sample rate {sample_rate/1e6:.3f} Msps " range is {sample_rate_range.start()} to {sample_rate_range.stop()}."
f"out of range: [{min_rate/1e6:.3f} - {max_rate/1e6:.3f} Msps]"
) )
self.usrp.set_tx_rate(sample_rate, channel) self.usrp.set_tx_rate(sample_rate, channel)
self.tx_sample_rate = self.usrp.get_tx_rate(channel) self.tx_sample_rate = self.usrp.get_tx_rate(channel)
print(f"USRP TX Sample Rate = {self.tx_sample_rate}") print(f"USRP TX Sample Rate = {self.tx_sample_rate}")
def set_tx_center_frequency(self, center_frequency, channel=0): def set_tx_center_frequency(self, center_frequency, channel=0):
center_frequency_range = self.usrp.get_tx_freq_range() center_frequency_range = self.usrp.get_tx_freq_range()
min_rate, max_rate = center_frequency_range.start(), center_frequency_range.stop() if center_frequency < center_frequency_range.start() or center_frequency > center_frequency_range.stop():
if center_frequency < min_rate or center_frequency > max_rate: raise IOError(
raise SDRParameterError( f"Center frequency {center_frequency} out of range for USRP.\
f"{self.__class__.__name__}: Center frequency {center_frequency/1e9:.3f} GHz " \nValid range is {center_frequency_range.start()}\
f"out of range: [{min_rate/1e9:.3f} - {max_rate/1e9:.3f} GHz]" to {center_frequency_range.stop()}."
) )
self.usrp.set_tx_freq(uhd.types.TuneRequest(center_frequency), channel) self.usrp.set_tx_freq(uhd.types.TuneRequest(center_frequency), channel)
self.tx_center_frequency = self.usrp.get_tx_freq(channel) self.tx_center_frequency = self.usrp.get_tx_freq(channel)
print(f"USRP TX Center Frequency = {self.tx_center_frequency}") print(f"USRP TX Center Frequency = {self.tx_center_frequency}")
@ -356,7 +339,7 @@ class USRP(SDR):
gain_range = self.usrp.get_tx_gain_range() gain_range = self.usrp.get_tx_gain_range()
if gain_mode == "relative": if gain_mode == "relative":
if gain > 0: if gain > 0:
raise SDRParameterError( raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This sets\ "When gain_mode = 'relative', gain must be < 0. This sets\
the gain relative to the maximum possible gain." the gain relative to the maximum possible gain."
) )
@ -375,13 +358,7 @@ class USRP(SDR):
print(f"USRP TX Gain = {self.tx_gain}") print(f"USRP TX Gain = {self.tx_gain}")
def close(self): def close(self):
self._tx_initialized = False pass
self._rx_initialized = False
if hasattr(self, "rx_stream"):
del self.rx_stream
if hasattr(self, "usrp"):
del self.usrp
self.usrp = None
def _stream_tx(self, callback): def _stream_tx(self, callback):
@ -462,9 +439,6 @@ class USRP(SDR):
print(f"USRP clock source set to {self.usrp.get_clock_source(0)}") print(f"USRP clock source set to {self.usrp.get_clock_source(0)}")
def supports_dynamic_updates(self) -> dict:
return {"center_frequency": True, "sample_rate": True, "gain": True}
def _create_device_dict(identifier_value=None): def _create_device_dict(identifier_value=None):
""" """

35
src/ria_toolkit_oss/view/view_signal.py
View File

@ -1,4 +1,3 @@
import gc
import os import os
import textwrap import textwrap
from typing import Optional from typing import Optional
@ -9,7 +8,6 @@ from matplotlib import gridspec
from PIL import Image from PIL import Image
from scipy.fft import fft, fftshift from scipy.fft import fft, fftshift
from scipy.signal import spectrogram from scipy.signal import spectrogram
from scipy.signal.windows import hann
from ria_toolkit_oss.datatypes.recording import Recording from ria_toolkit_oss.datatypes.recording import Recording
from ria_toolkit_oss.view.tools import ( from ria_toolkit_oss.view.tools import (
@ -124,7 +122,7 @@ def view_sig(
plot_y_indx = plot_y_indx + 2 plot_y_indx = plot_y_indx + 2
fft_size = get_fft_size(plot_length=plot_length) fft_size = get_fft_size(plot_length=plot_length)
_, t_spec, Sxx = spectrogram( f, t_spec, Sxx = spectrogram(
complex_signal[:plot_length], complex_signal[:plot_length],
fs=sample_rate, fs=sample_rate,
nperseg=fft_size, nperseg=fft_size,
@ -134,16 +132,14 @@ def view_sig(
) )
# shift frequencies so zero is centered # shift frequencies so zero is centered
f_bins = np.fft.fftfreq(fft_size, d=1.0 / sample_rate)
f_bins = np.fft.fftshift(f_bins)
f_bins = f_bins + center_frequency
Sxx = np.fft.fftshift(Sxx, axes=0) Sxx = np.fft.fftshift(Sxx, axes=0)
f = np.fft.fftshift(f) - sample_rate / 2 + center_frequency
spec_ax.imshow( spec_ax.imshow(
10 * np.log10(Sxx + 1e-12), 10 * np.log10(Sxx + 1e-12),
aspect="auto", aspect="auto",
origin="lower", origin="lower",
extent=[t_spec[0], t_spec[-1], f_bins[0], f_bins[-1]], extent=[t_spec[0], t_spec[-1], f[0], f[-1]],
cmap="twilight", cmap="twilight",
) )
@ -173,17 +169,18 @@ def view_sig(
freq_ax = plt.subplot(gs[plot_y_indx : plot_y_indx + 2, :]) freq_ax = plt.subplot(gs[plot_y_indx : plot_y_indx + 2, :])
plot_y_indx = plot_y_indx + 2 plot_y_indx = plot_y_indx + 2
# Apply window to reduce spectral leakage epsilon = 1e-10
window = hann(len(complex_signal[:plot_length])) spectrum = np.abs(fftshift(fft(complex_signal[0:plot_length])))
spectrum = np.abs(fftshift(fft(complex_signal[:plot_length] * window))) freqs = (
np.linspace(-1 * (sample_rate / 2), (sample_rate / 2), len(complex_signal[0:plot_length]))
+ center_frequency
)
# Convert to dB # Use semi-log for the y-axis
spectrum_db = 20 * np.log10(spectrum + 1e-12) # 20*log for magnitude freq_ax.semilogy(freqs, spectrum + epsilon, color=COLORS["accent"], linewidth=0.8)
freq_ax.set_xlabel("Frequency")
freqs = np.linspace(-sample_rate / 2, sample_rate / 2, len(complex_signal[:plot_length])) + center_frequency freq_ax.set_ylabel("Magnitude")
freq_ax.plot(freqs, spectrum_db, color=COLORS["accent"], linewidth=0.8) freq_ax.set_title("Frequency Spectrum", fontsize=subtitle_fontsize)
freq_ax.set_ylabel("Magnitude (dB)")
freq_ax.set_title("Frequency Spectrum (Windowed FFT)", fontsize=subtitle_fontsize)
set_spines(freq_ax, spines) set_spines(freq_ax, spines)
if constellation: if constellation:
@ -258,7 +255,3 @@ def view_sig(
output_path, _ = set_path(output_path=output_path) output_path, _ = set_path(output_path=output_path)
plt.savefig(output_path, dpi=dpi) plt.savefig(output_path, dpi=dpi)
print(f"Saved signal plot to {output_path}") print(f"Saved signal plot to {output_path}")
# Garbage collection and clean up to prevent memory overloading
plt.close("all")
gc.collect()

5
src/ria_toolkit_oss/view/view_signal_simple.py
View File

@ -2,7 +2,6 @@
from __future__ import annotations from __future__ import annotations
import gc
from typing import Optional from typing import Optional
import matplotlib import matplotlib
@ -319,10 +318,6 @@ def view_simple_sig(
return output_path return output_path
plt.show() plt.show()
# Garbage collection and clean up to prevent memory overloading
plt.close("all")
gc.collect()
return None return None