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

st_edits #6

Merged
madrigal merged 14 commits from st_edits into main 2025-10-24 10:21:12 -04:00
Conversation 0 Commits 14 Files Changed 15 +958 -396
6 changed files with 958 additions and 396 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit a9f8ad4bee - Show all commits

15
pyproject.toml
View File

@ -47,6 +47,21 @@ dependencies = [
"pyzmq (>=27.1.0,<28.0.0)", "pyzmq (>=27.1.0,<28.0.0)",
] ]
[project.optional-dependencies]
# SDR hardware-specific dependencies (optional installs)
rtlsdr = ["pyrtlsdr>=0.2.9"]
pluto = ["pyadi-iio>=0.0.14"]
usrp = [] # Requires system UHD installation
hackrf = ["pyhackrf>=0.2.0"]
bladerf = [] # Requires system libbladerf installation
# All SDR hardware support
all-sdr = [
"pyrtlsdr>=0.2.9",
"pyadi-iio>=0.0.14",
"pyhackrf>=0.2.0",
]
[tool.poetry] [tool.poetry]
packages = [ packages = [
{ include = "ria_toolkit_oss", from = "src" } { include = "ria_toolkit_oss", from = "src" }

83
src/ria_toolkit_oss/sdr/_external/libhackrf.py
View File

@ -325,8 +325,8 @@ f.argtypes = [p_hackrf_device, POINTER(read_partid_serialno_t)]
# libhackrf.hackrf_set_txvga_gain.argtypes = [POINTER(hackrf_device), c_uint32] # libhackrf.hackrf_set_txvga_gain.argtypes = [POINTER(hackrf_device), c_uint32]
## extern ADDAPI int ADDCALL hackrf_set_antenna_enable(hackrf_device* ## extern ADDAPI int ADDCALL hackrf_set_antenna_enable(hackrf_device*
## device, const uint8_t value); ## device, const uint8_t value);
# libhackrf.hackrf_set_antenna_enable.restype = c_int libhackrf.hackrf_set_antenna_enable.restype = c_int
# libhackrf.hackrf_set_antenna_enable.argtypes = [POINTER(hackrf_device), c_uint8] libhackrf.hackrf_set_antenna_enable.argtypes = [p_hackrf_device, c_uint8]
# #
## extern ADDAPI const char* ADDCALL hackrf_error_name(enum hackrf_error errcode); ## extern ADDAPI const char* ADDCALL hackrf_error_name(enum hackrf_error errcode);
## libhackrf.hackrf_error_name.restype = POINTER(c_char) ## libhackrf.hackrf_error_name.restype = POINTER(c_char)
@ -537,6 +537,16 @@ class HackRF(object):
raise IOError("error disabling amp") raise IOError("error disabling amp")
return 0 return 0
def set_antenna_enable(self, enable):
value = 1 if enable else 0
result = libhackrf.hackrf_set_antenna_enable(self.dev_p, value)
if result != 0:
error_name = get_error_name(result)
raise IOError(f"Error setting antenna bias tee: {error_name} (Code {result})")
state = "enabled" if enable else "disabled"
print(f"HackRF antenna bias tee {state}.")
return 0
# rounds down to multiple of 8 (15 -> 8, 39 -> 32), etc. # rounds down to multiple of 8 (15 -> 8, 39 -> 32), etc.
# internally, hackrf_set_lna_gain does the same thing # internally, hackrf_set_lna_gain does the same thing
# But we take care of it so we can keep track of the correct gain # But we take care of it so we can keep track of the correct gain
@ -582,6 +592,75 @@ class HackRF(object):
if result != 0: if result != 0:
raise IOError("stop_rx failure") raise IOError("stop_rx failure")
def _rx_capture_callback(self, hackrf_transfer):
"""Instance method callback for RX capture - prevents garbage collection"""
try:
c = hackrf_transfer.contents
# Append bytes to buffer using string_at
from ctypes import string_at
byte_chunk = string_at(c.buffer, c.valid_length)
self._capture_buffer.extend(byte_chunk)
# Check if we have enough
if len(self._capture_buffer) >= self._capture_target:
self._capture_done = True
return 1 # Stop streaming
return 0
except Exception as e:
print(f"Error in RX capture callback: {e}")
import traceback
traceback.print_exc()
self._capture_done = True
return 1
def read_samples(self, num_samples):
"""
Block capture mode for HackRF - captures exactly num_samples.
This is safer than streaming for USB2 and avoids buffer overflow issues.
:param num_samples: Number of complex samples to capture
:return: numpy array of complex64 samples
"""
# Initialize capture state as instance variables
self._capture_buffer = bytearray()
self._capture_target = num_samples * 2 # 2 bytes per complex sample (I+Q as int8)
self._capture_done = False
# Store callback as instance variable to prevent garbage collection (like TX does)
self._rx_cb = _callback(self._rx_capture_callback)
# Start RX with the callback
result = libhackrf.hackrf_start_rx(self.dev_p, self._rx_cb, None)
if result != 0:
raise IOError("start_rx failure during read_samples")
# Wait for capture to complete
import time
timeout = num_samples / self.sample_rate + 5.0 # Add 5 second buffer
start_time = time.time()
while not self._capture_done:
if time.time() - start_time > timeout:
print("HackRF capture timeout!")
break
time.sleep(0.01)
# Stop RX
self.stop_rx()
# Convert bytes to complex samples
byte_data = bytes(self._capture_buffer[:self._capture_target])
all_samples = np.frombuffer(byte_data, dtype=np.int8).astype(np.float32).view(np.complex64)
# Clean up instance variables
del self._capture_buffer
del self._capture_target
del self._capture_done
del self._rx_cb
return all_samples[:num_samples]
# Add transmit gain property # Add transmit gain property
def set_txvga_gain(self, gain): def set_txvga_gain(self, gain):
if gain < 0 or gain > 47: if gain < 0 or gain > 47:

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

@ -1,383 +1,484 @@
from typing import Optional from typing import Optional
import numpy as np 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 from ria_toolkit_oss.sdr import SDR
class Blade(SDR): class Blade(SDR):
def __init__(self, identifier=""): def __init__(self, identifier=""):
""" """
Initialize a BladeRF device object and connect to the SDR hardware. Initialize a BladeRF device object and connect to the SDR hardware.
:param identifier: Not used for BladeRF. :param identifier: Not used for BladeRF.
BladeRF devices cannot currently be selected with and identifier value. BladeRF devices cannot currently be selected with and identifier value.
If there are multiple connected devices, the device in use may be selected randomly. If there are multiple connected devices, the device in use may be selected randomly.
""" """
if identifier != "": if identifier != "":
print(f"Warning, radio identifier {identifier} provided for Blade but will not be used.") print(f"Warning, radio identifier {identifier} provided for Blade but will not be used.")
uut = self._probe_bladerf() uut = self._probe_bladerf()
if uut is None: if uut is None:
print("No bladeRFs detected. Exiting.") print("No bladeRFs detected. Exiting.")
self._shutdown(error=-1, board=None) self._shutdown(error=-1, board=None)
print(uut) print(uut)
self.device = _bladerf.BladeRF(uut) self.device = _bladerf.BladeRF(uut)
self._print_versions(device=self.device) self._print_versions(device=self.device)
super().__init__() super().__init__()
def _shutdown(self, error=0, board=None): def supports_bias_tee(self) -> bool:
print("Shutting down with error code: " + str(error)) return True
if board is not None:
board.close() def set_bias_tee(self, enable: bool, channel: Optional[int] = None):
if channel is None:
# TODO why does this create an error under any conditions? channel = getattr(self, "rx_channel", getattr(self, "tx_channel", 0))
raise OSError("Shutdown initiated with error code: {}".format(error))
try:
def _probe_bladerf(self): bladerf_channel = _bladerf.CHANNEL_RX(channel)
device = None self.device.set_bias_tee(bladerf_channel, bool(enable))
print("Searching for bladeRF devices...") except AttributeError as exc: # pragma: no cover - depends on libbladeRF version
try: raise NotImplementedError("bladeRF binding lacks bias-tee control") from exc
devinfos = _bladerf.get_device_list()
if len(devinfos) == 1: state = "enabled" if enable else "disabled"
device = "{backend}:device={usb_bus}:{usb_addr}".format(**devinfos[0]._asdict()) print(f"BladeRF bias tee {state} on channel {channel}.")
print("Found bladeRF device: " + str(device))
if len(devinfos) > 1: def _shutdown(self, error=0, board=None):
print("Unsupported feature: more than one bladeRFs detected.") print("Shutting down with error code: " + str(error))
print("\n".join([str(devinfo) for devinfo in devinfos])) if board is not None:
self._shutdown(error=-1, board=None) board.close()
except _bladerf.BladeRFError:
print("No bladeRF devices found.") # TODO why does this create an error under any conditions?
pass raise OSError("Shutdown initiated with error code: {}".format(error))
return device
def _probe_bladerf(self):
def _print_versions(self, device=None): device = None
print("libbladeRF version:\t" + str(_bladerf.version())) print("Searching for bladeRF devices...")
if device is not None: try:
print("Firmware version:\t" + str(device.get_fw_version())) devinfos = _bladerf.get_device_list()
print("FPGA version:\t\t" + str(device.get_fpga_version())) if len(devinfos) == 1:
return 0 device = "{backend}:device={usb_bus}:{usb_addr}".format(**devinfos[0]._asdict())
print("Found bladeRF device: " + str(device))
def close(self): if len(devinfos) > 1:
self.device.close() print("Unsupported feature: more than one bladeRFs detected.")
print("\n".join([str(devinfo) for devinfo in devinfos]))
def init_rx( self._shutdown(error=-1, board=None)
self, except _bladerf.BladeRFError:
sample_rate: int | float, print("No bladeRF devices found.")
center_frequency: int | float, pass
gain: int, return device
channel: int,
buffer_size: Optional[int] = 8192, def _print_versions(self, device=None):
gain_mode: Optional[str] = "absolute", print("libbladeRF version:\t" + str(_bladerf.version()))
): if device is not None:
""" print("Firmware version:\t" + str(device.get_fw_version()))
Initializes the BladeRF for receiving. print("FPGA version:\t\t" + str(device.get_fpga_version()))
return 0
:param sample_rate: The sample rate for receiving.
:type sample_rate: int or float def close(self):
:param center_frequency: The center frequency of the recording. self.device.close()
:type center_frequency: int or float
:param gain: The gain set for receiving on the BladeRF def init_rx(
:type gain: int self,
:param channel: The channel the BladeRF is set to. sample_rate: int | float,
:type channel: int center_frequency: int | float,
:param buffer_size: The buffer size during receive. Defaults to 8192. gain: int,
:type buffer_size: int channel: int,
""" buffer_size: Optional[int] = 8192,
print("Initializing RX") gain_mode: Optional[str] = "absolute",
):
# Configure BladeRF """
self._set_rx_channel(channel) Initializes the BladeRF for receiving.
self._set_rx_sample_rate(sample_rate)
self._set_rx_center_frequency(center_frequency) :param sample_rate: The sample rate for receiving.
self._set_rx_gain(channel, gain, gain_mode) :type sample_rate: int or float
self._set_rx_buffer_size(buffer_size) :param center_frequency: The center frequency of the recording.
:type center_frequency: int or float
bw = self.rx_sample_rate :param gain: The gain set for receiving on the BladeRF
if bw < 200000: :type gain: int
bw = 200000 :param channel: The channel the BladeRF is set to.
elif bw > 56000000: :type channel: int
bw = 56000000 :param buffer_size: The buffer size during receive. Defaults to 8192.
self.rx_ch.bandwidth = bw :type buffer_size: int
"""
self._rx_initialized = True print("Initializing RX")
self._tx_initialized = False
# Configure BladeRF
def init_tx( self._set_rx_channel(channel)
self, self._set_rx_sample_rate(sample_rate)
sample_rate: int | float, self._set_rx_center_frequency(center_frequency)
center_frequency: int | float, self._set_rx_gain(channel, gain, gain_mode)
gain: int, self._set_rx_buffer_size(buffer_size)
channel: int,
buffer_size: Optional[int] = 8192, bw = self.rx_sample_rate
gain_mode: Optional[str] = "absolute", if bw < 200000:
): bw = 200000
""" elif bw > 56000000:
Initializes the BladeRF for transmitting. bw = 56000000
self.rx_ch.bandwidth = bw
:param sample_rate: The sample rate for transmitting.
:type sample_rate: int or float self._rx_initialized = True
:param center_frequency: The center frequency of the recording. self._tx_initialized = False
:type center_frequency: int or float
:param gain: The gain set for transmitting on the BladeRF def init_tx(
:type gain: int self,
:param channel: The channel the BladeRF is set to. sample_rate: int | float,
:type channel: int center_frequency: int | float,
:param buffer_size: The buffer size during transmission. Defaults to 8192. gain: int,
:type buffer_size: int channel: int,
""" buffer_size: Optional[int] = 8192,
gain_mode: Optional[str] = "absolute",
# Configure BladeRF ):
self._set_tx_channel(channel) """
self._set_tx_sample_rate(sample_rate) Initializes the BladeRF for transmitting.
self._set_tx_center_frequency(center_frequency)
self._set_tx_gain(channel=channel, gain=gain, gain_mode=gain_mode) :param sample_rate: The sample rate for transmitting.
self._set_tx_buffer_size(buffer_size) :type sample_rate: int or float
:param center_frequency: The center frequency of the recording.
bw = self.tx_sample_rate :type center_frequency: int or float
if bw < 200000: :param gain: The gain set for transmitting on the BladeRF
bw = 200000 :type gain: int
elif bw > 56000000: :param channel: The channel the BladeRF is set to.
bw = 56000000 :type channel: int
self.tx_ch.bandwidth = bw :param buffer_size: The buffer size during transmission. Defaults to 8192.
:type buffer_size: int
if self.device is None: """
print("TX: Invalid device handle.")
return -1 # Configure BladeRF
self._set_tx_channel(channel)
if self.tx_channel is None: self._set_tx_sample_rate(sample_rate)
print("TX: Invalid channel.") self._set_tx_center_frequency(center_frequency)
return -1 self._set_tx_gain(channel=channel, gain=gain, gain_mode=gain_mode)
self._set_tx_buffer_size(buffer_size)
self._tx_initialized = True
self._rx_initialized = False bw = self.tx_sample_rate
return 0 if bw < 200000:
bw = 200000
def _stream_rx(self, callback): elif bw > 56000000:
if not self._rx_initialized: bw = 56000000
raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()") self.tx_ch.bandwidth = bw
# Setup synchronous stream if self.device is None:
self.device.sync_config( print("TX: Invalid device handle.")
layout=_bladerf.ChannelLayout.RX_X1, return -1
fmt=_bladerf.Format.SC16_Q11,
num_buffers=16, if self.tx_channel is None:
buffer_size=self.rx_buffer_size, print("TX: Invalid channel.")
num_transfers=8, return -1
stream_timeout=3500000000,
) self._tx_initialized = True
self._rx_initialized = False
self.rx_ch.enable = True return 0
self.bytes_per_sample = 4
def _stream_rx(self, callback):
print("Blade Starting RX...") if not self._rx_initialized:
self._enable_rx = True raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()")
while self._enable_rx: # Setup synchronous stream
# Create receive buffer and read in samples to buffer self.device.sync_config(
# Add them to a list to convert and save after stream is finished layout=_bladerf.ChannelLayout.RX_X1,
buffer = bytearray(self.rx_buffer_size * self.bytes_per_sample) fmt=_bladerf.Format.SC16_Q11,
self.device.sync_rx(buffer, self.rx_buffer_size) num_buffers=16,
signal = self._convert_rx_samples(buffer) buffer_size=self.rx_buffer_size,
# samples = convert_to_2xn(signal) num_transfers=8,
self.buffer = buffer stream_timeout=3500000000,
# send callback complex signal )
callback(buffer=signal, metadata=None)
self.rx_ch.enable = True
# Disable module self.bytes_per_sample = 4
print("Blade RX Completed.")
self.rx_ch.enable = False print("Blade Starting RX...")
self._enable_rx = True
def record(self, num_samples):
if not self._rx_initialized: while self._enable_rx:
raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()") # Create receive buffer and read in samples to buffer
# Add them to a list to convert and save after stream is finished
# Setup synchronous stream buffer = bytearray(self.rx_buffer_size * self.bytes_per_sample)
self.device.sync_config( self.device.sync_rx(buffer, self.rx_buffer_size)
layout=_bladerf.ChannelLayout.RX_X1, signal = self._convert_rx_samples(buffer)
fmt=_bladerf.Format.SC16_Q11, # samples = convert_to_2xn(signal)
num_buffers=16, self.buffer = buffer
buffer_size=self.rx_buffer_size, # send callback complex signal
num_transfers=8, callback(buffer=signal, metadata=None)
stream_timeout=3500000000,
) # Disable module
print("Blade RX Completed.")
self.rx_ch.enable = True self.rx_ch.enable = False
self.bytes_per_sample = 4
def record(self, num_samples):
print("Blade Starting RX...") if not self._rx_initialized:
self._enable_rx = True raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record()")
store_array = np.zeros((1, (num_samples // self.rx_buffer_size + 1) * self.rx_buffer_size), dtype=np.complex64) # Setup synchronous stream
self.device.sync_config(
for i in range(num_samples // self.rx_buffer_size + 1): layout=_bladerf.ChannelLayout.RX_X1,
# Create receive buffer and read in samples to buffer fmt=_bladerf.Format.SC16_Q11,
# Add them to a list to convert and save after stream is finished num_buffers=16,
buffer = bytearray(self.rx_buffer_size * self.bytes_per_sample) buffer_size=self.rx_buffer_size,
self.device.sync_rx(buffer, self.rx_buffer_size) num_transfers=8,
signal = self._convert_rx_samples(buffer) stream_timeout=3500000000,
# samples = convert_to_2xn(signal) )
store_array[:, i * self.rx_buffer_size : (i + 1) * self.rx_buffer_size] = signal
self.rx_ch.enable = True
# Disable module self.bytes_per_sample = 4
print("Blade RX Completed.")
self.rx_ch.enable = False print("Blade Starting RX...")
metadata = { self._enable_rx = True
"source": self.__class__.__name__,
"sample_rate": self.rx_sample_rate, store_array = np.zeros((1, (num_samples // self.rx_buffer_size + 1) * self.rx_buffer_size), dtype=np.complex64)
"center_frequency": self.rx_center_frequency,
"gain": self.rx_gain, for i in range(num_samples // self.rx_buffer_size + 1):
} # Create receive buffer and read in samples to buffer
# Add them to a list to convert and save after stream is finished
return Recording(data=store_array[:, :num_samples], metadata=metadata) buffer = bytearray(self.rx_buffer_size * self.bytes_per_sample)
self.device.sync_rx(buffer, self.rx_buffer_size)
def _stream_tx(self, callback): signal = self._convert_rx_samples(buffer)
# samples = convert_to_2xn(signal)
# Setup stream store_array[:, i * self.rx_buffer_size : (i + 1) * self.rx_buffer_size] = signal
self.device.sync_config(
layout=_bladerf.ChannelLayout.TX_X1, # Disable module
fmt=_bladerf.Format.SC16_Q11, print("Blade RX Completed.")
num_buffers=16, self.rx_ch.enable = False
buffer_size=8192, metadata = {
num_transfers=8, "source": self.__class__.__name__,
stream_timeout=3500, "sample_rate": self.rx_sample_rate,
) "center_frequency": self.rx_center_frequency,
"gain": self.rx_gain,
# Enable module }
self.tx_ch.enable = True
self._enable_tx = True return Recording(data=store_array[:, :num_samples], metadata=metadata)
print("Blade Starting TX...") def tx_recording(
self,
while self._enable_tx: recording: "Recording | np.ndarray",
buffer = callback(self.tx_buffer_size) # [0] num_samples: Optional[int] = None,
byte_array = self._convert_tx_samples(buffer) tx_time: Optional[int | float] = None,
self.device.sync_tx(byte_array, len(buffer)) ):
"""
# Disable module Transmit the given IQ samples from the provided recording.
print("Blade TX Completed.") init_tx() must be called before this function.
self.tx_ch.enable = False
:param recording: The recording to transmit.
def _convert_rx_samples(self, samples): :type recording: Recording or np.ndarray
samples = np.frombuffer(samples, dtype=np.int16).astype(np.float32) :param num_samples: The number of samples to transmit, will repeat or
samples /= 2048 truncate the recording to this length. Defaults to None.
samples = samples[::2] + 1j * samples[1::2] :type num_samples: int, optional
return samples :param tx_time: The time to transmit, will repeat or truncate the
recording to this length. Defaults to None.
def _convert_tx_samples(self, samples): :type tx_time: int or float, optional
tx_samples = np.empty(samples.size * 2, dtype=np.float32) """
tx_samples[::2] = np.real(samples) # Real part import warnings
tx_samples[1::2] = np.imag(samples) # Imaginary part import time
from ria_toolkit_oss.datatypes.recording import Recording
tx_samples *= 2048
tx_samples = tx_samples.astype(np.int16) if num_samples is not None and tx_time is not None:
byte_array = tx_samples.tobytes() raise ValueError("Only input one of num_samples or tx_time")
elif num_samples is not None:
return byte_array tx_time = num_samples / self.tx_sample_rate
elif tx_time is not None:
def _set_rx_channel(self, channel): pass
self.rx_channel = channel else:
self.rx_ch = self.device.Channel(_bladerf.CHANNEL_RX(channel)) tx_time = len(recording) / self.tx_sample_rate
print(f"\nBlade channel = {self.rx_ch}")
if isinstance(recording, np.ndarray):
def _set_rx_sample_rate(self, sample_rate): samples = recording
self.rx_sample_rate = sample_rate elif isinstance(recording, Recording):
self.rx_ch.sample_rate = self.rx_sample_rate if len(recording.data) > 1:
print(f"Blade sample rate = {self.rx_ch.sample_rate}") warnings.warn("Recording object is multichannel, only channel 0 data was used for transmission")
samples = recording.data[0]
def _set_rx_center_frequency(self, center_frequency): else:
self.rx_center_frequency = center_frequency raise TypeError("recording must be np.ndarray or Recording")
self.rx_ch.frequency = center_frequency
print(f"Blade center frequency = {self.rx_ch.frequency}") samples = samples.astype(np.complex64, copy=False)
def _set_rx_gain(self, channel, gain, gain_mode): # Setup stream
self.device.sync_config(
rx_gain_min = self.device.get_gain_range(channel)[0] layout=_bladerf.ChannelLayout.TX_X1,
rx_gain_max = self.device.get_gain_range(channel)[1] fmt=_bladerf.Format.SC16_Q11,
num_buffers=16,
if gain_mode == "relative": buffer_size=self.tx_buffer_size,
if gain > 0: num_transfers=8,
raise ValueError( stream_timeout=3500,
"When gain_mode = 'relative', gain must be < 0. This sets \ )
the gain relative to the maximum possible gain."
) # Enable module
else: self.tx_ch.enable = True
abs_gain = rx_gain_max + gain
else: print("Blade Starting TX...")
abs_gain = gain
# Transmit samples - repeat as needed for the duration
if abs_gain < rx_gain_min or abs_gain > rx_gain_max: start_time = time.time()
abs_gain = min(max(gain, rx_gain_min), rx_gain_max) sample_index = 0
print(f"Gain {abs_gain} out of range for Blade.")
print(f"Gain range: {rx_gain_min} to {rx_gain_max} dB") try:
while time.time() - start_time < tx_time:
self.rx_gain = abs_gain # Get next chunk
self.rx_ch.gain = abs_gain chunk_size = min(self.tx_buffer_size, len(samples) - sample_index)
if chunk_size == 0:
print(f"Blade gain = {self.rx_ch.gain}") # Reached end, loop back
sample_index = 0
def _set_rx_buffer_size(self, buffer_size): chunk_size = min(self.tx_buffer_size, len(samples))
self.rx_buffer_size = buffer_size
chunk = samples[sample_index:sample_index + chunk_size]
def _set_tx_channel(self, channel): sample_index += chunk_size
self.tx_channel = channel
self.tx_ch = self.device.Channel(_bladerf.CHANNEL_TX(self.tx_channel)) # Convert and transmit
print(f"\nBlade channel = {self.tx_ch}") byte_array = self._convert_tx_samples(chunk)
self.device.sync_tx(byte_array, len(chunk))
def _set_tx_sample_rate(self, sample_rate):
self.tx_sample_rate = sample_rate except KeyboardInterrupt:
self.tx_ch.sample_rate = self.tx_sample_rate print("\nTransmission interrupted by user")
print(f"Blade sample rate = {self.tx_ch.sample_rate}")
# Disable module
def _set_tx_center_frequency(self, center_frequency): print("Blade TX Completed.")
self.tx_center_frequency = center_frequency self.tx_ch.enable = False
self.tx_ch.frequency = center_frequency
print(f"Blade center frequency = {self.tx_ch.frequency}") def _stream_tx(self, callback):
def _set_tx_gain(self, channel, gain, gain_mode): # Setup stream
self.device.sync_config(
tx_gain_min = self.device.get_gain_range(channel)[0] layout=_bladerf.ChannelLayout.TX_X1,
tx_gain_max = self.device.get_gain_range(channel)[1] fmt=_bladerf.Format.SC16_Q11,
num_buffers=16,
if gain_mode == "relative": buffer_size=8192,
if gain > 0: num_transfers=8,
raise ValueError( stream_timeout=3500,
"When gain_mode = 'relative', gain must be < 0. This sets\ )
the gain relative to the maximum possible gain."
) # Enable module
else: self.tx_ch.enable = True
abs_gain = tx_gain_max + gain self._enable_tx = True
else:
abs_gain = gain print("Blade Starting TX...")
if abs_gain < tx_gain_min or abs_gain > tx_gain_max: while self._enable_tx:
abs_gain = min(max(gain, tx_gain_min), tx_gain_max) buffer = callback(self.tx_buffer_size) # [0]
print(f"Gain {abs_gain} out of range for Blade.") byte_array = self._convert_tx_samples(buffer)
print(f"Gain range: {tx_gain_min} to {tx_gain_max} dB") self.device.sync_tx(byte_array, len(buffer))
self.tx_gain = abs_gain # Disable module
self.tx_ch.gain = abs_gain print("Blade TX Completed.")
self.tx_ch.enable = False
print(f"Blade gain = {self.tx_ch.gain}")
def _convert_rx_samples(self, samples):
def _set_tx_buffer_size(self, buffer_size): samples = np.frombuffer(samples, dtype=np.int16).astype(np.float32)
self.tx_buffer_size = buffer_size samples /= 2048
samples = samples[::2] + 1j * samples[1::2]
def set_clock_source(self, source): return samples
if source.lower() == "external":
self.device.set_pll_enable(True) def _convert_tx_samples(self, samples):
elif source.lower() == "internal": tx_samples = np.empty(samples.size * 2, dtype=np.float32)
print("Disabling PLL") tx_samples[::2] = np.real(samples) # Real part
self.device.set_pll_enable(False) tx_samples[1::2] = np.imag(samples) # Imaginary part
print(f"Clock source set to {self.device.get_clock_select()}") tx_samples *= 2048
print(f"PLL Reference set to {self.device.get_pll_refclk()}") tx_samples = tx_samples.astype(np.int16)
byte_array = tx_samples.tobytes()
return byte_array
def _set_rx_channel(self, channel):
self.rx_channel = channel
self.rx_ch = self.device.Channel(_bladerf.CHANNEL_RX(channel))
print(f"\nBlade channel = {self.rx_ch}")
def _set_rx_sample_rate(self, sample_rate):
self.rx_sample_rate = sample_rate
self.rx_ch.sample_rate = self.rx_sample_rate
print(f"Blade sample rate = {self.rx_ch.sample_rate}")
def _set_rx_center_frequency(self, center_frequency):
self.rx_center_frequency = center_frequency
self.rx_ch.frequency = center_frequency
print(f"Blade center frequency = {self.rx_ch.frequency}")
def _set_rx_gain(self, channel, gain, gain_mode):
rx_gain_min = self.device.get_gain_range(channel)[0]
rx_gain_max = self.device.get_gain_range(channel)[1]
if gain_mode == "relative":
if gain > 0:
raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This sets \
the gain relative to the maximum possible gain."
)
else:
abs_gain = rx_gain_max + gain
else:
abs_gain = gain
if abs_gain < rx_gain_min or abs_gain > rx_gain_max:
abs_gain = min(max(gain, rx_gain_min), rx_gain_max)
print(f"Gain {abs_gain} out of range for Blade.")
print(f"Gain range: {rx_gain_min} to {rx_gain_max} dB")
self.rx_gain = abs_gain
self.rx_ch.gain = abs_gain
print(f"Blade gain = {self.rx_ch.gain}")
def _set_rx_buffer_size(self, buffer_size):
self.rx_buffer_size = buffer_size
def _set_tx_channel(self, channel):
self.tx_channel = channel
self.tx_ch = self.device.Channel(_bladerf.CHANNEL_TX(self.tx_channel))
print(f"\nBlade channel = {self.tx_ch}")
def _set_tx_sample_rate(self, sample_rate):
self.tx_sample_rate = sample_rate
self.tx_ch.sample_rate = self.tx_sample_rate
print(f"Blade sample rate = {self.tx_ch.sample_rate}")
def _set_tx_center_frequency(self, center_frequency):
self.tx_center_frequency = center_frequency
self.tx_ch.frequency = center_frequency
print(f"Blade center frequency = {self.tx_ch.frequency}")
def _set_tx_gain(self, channel, gain, gain_mode):
tx_gain_min = self.device.get_gain_range(channel)[0]
tx_gain_max = self.device.get_gain_range(channel)[1]
if gain_mode == "relative":
if gain > 0:
raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This sets\
the gain relative to the maximum possible gain."
)
else:
abs_gain = tx_gain_max + gain
else:
abs_gain = gain
if abs_gain < tx_gain_min or abs_gain > tx_gain_max:
abs_gain = min(max(gain, tx_gain_min), tx_gain_max)
print(f"Gain {abs_gain} out of range for Blade.")
print(f"Gain range: {tx_gain_min} to {tx_gain_max} dB")
self.tx_gain = abs_gain
self.tx_ch.gain = abs_gain
print(f"Blade gain = {self.tx_ch.gain}")
def _set_tx_buffer_size(self, buffer_size):
self.tx_buffer_size = buffer_size
def set_clock_source(self, source):
if source.lower() == "external":
self.device.set_pll_enable(True)
elif source.lower() == "internal":
print("Disabling PLL")
self.device.set_pll_enable(False)
print(f"Clock source set to {self.device.get_clock_select()}")
print(f"PLL Reference set to {self.device.get_pll_refclk()}")

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

@ -1,5 +1,6 @@
import time import time
import warnings import warnings
import math
from typing import Optional from typing import Optional
import numpy as np import numpy as np
@ -35,10 +36,101 @@ class HackRF(SDR):
super().__init__() super().__init__()
def supports_bias_tee(self) -> bool:
return True
def set_bias_tee(self, enable: bool):
try:
self.radio.set_antenna_enable(bool(enable))
except AttributeError as exc: # pragma: no cover - defensive
raise NotImplementedError("Underlying HackRF interface lacks bias-tee control") from exc
def init_rx(self, sample_rate, center_frequency, gain, channel, gain_mode): def init_rx(self, sample_rate, center_frequency, gain, channel, gain_mode):
self._tx_initialized = False """
Initializes the HackRF for receiving.
HackRF has 3 gain stages:
- 14 dB front-end amplifier (on/off)
- LNA gain: 0-40 dB in 8 dB steps
- VGA gain: 0-62 dB in 2 dB steps
:param sample_rate: The sample rate for receiving.
:type sample_rate: int or float
:param center_frequency: The center frequency of the recording.
:type center_frequency: int or float
:param gain: The total gain set for receiving on the HackRF (distributed across stages)
:type gain: int
:param channel: The channel the HackRF is set to. (Not actually used)
:type channel: int
:param gain_mode: Gain mode setting. Currently only "absolute" is supported.
:type gain_mode: str
"""
print("Initializing RX")
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
rx_gain_min = 0
rx_gain_max = 116 # 14 (amp) + 40 (LNA) + 62 (VGA)
if gain_mode == "relative":
if gain > 0:
raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This "
"sets the gain relative to the maximum possible gain."
)
else:
abs_gain = rx_gain_max + gain
else:
abs_gain = gain
if abs_gain < rx_gain_min or abs_gain > rx_gain_max:
abs_gain = min(max(abs_gain, rx_gain_min), rx_gain_max)
print(f"Gain {gain} out of range for HackRF.")
print(f"Gain range: {rx_gain_min} to {rx_gain_max} dB")
# Distribute gain using the signal-testbed algorithm
enable_amp = False
remaining_gain = abs_gain
# Enable 14 dB pre-amp if gain is high enough
if remaining_gain > 30:
remaining_gain = remaining_gain - 14
enable_amp = True
print("HackRF: 14dB front-end amplifier enabled.")
# Distribute remaining gain between LNA and VGA
# LNA gets 60% of remaining gain, rounded down to 8 dB steps
lna_gain = math.floor(remaining_gain * 0.6)
lna_gain = lna_gain - (lna_gain % 8) # Round to 8 dB steps
if lna_gain > 40:
lna_gain = 40
# VGA gets the rest
vga_gain = remaining_gain - lna_gain
if vga_gain > 62:
vga_gain = 62
# Apply gain settings
if enable_amp:
self.radio.enable_amp()
else:
self.radio.disable_amp()
self.radio.set_lna_gain(lna_gain)
self.radio.set_vga_gain(vga_gain)
self.rx_gain = abs_gain
print(f"HackRF gain distribution: Amp={enable_amp}, LNA={lna_gain}dB, VGA={vga_gain}dB")
self._rx_initialized = True self._rx_initialized = True
return NotImplementedError("RX not yet implemented for HackRF") self._tx_initialized = False
def init_tx( def init_tx(
self, self,
@ -151,10 +243,87 @@ class HackRF(SDR):
def close(self): def close(self):
self.radio.close() self.radio.close()
def _stream_rx(self, callback): def record(self, num_samples):
"""
Record a specified number of samples from the HackRF using block capture mode.
This is more reliable than streaming for USB2 connections.
:param num_samples: Number of samples to capture
:type num_samples: int
:return: Recording object containing the captured data
:rtype: Recording
"""
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 record()")
return NotImplementedError("RX not yet implemented for HackRF")
print("HackRF Starting RX...")
# Use libhackrf's block capture method
all_samples = self.radio.read_samples(num_samples)
print("HackRF RX Completed.")
# Create 1xN array for single-channel recording
store_array = np.zeros((1, num_samples), dtype=np.complex64)
store_array[0, :] = all_samples
metadata = {
"source": self.__class__.__name__,
"sample_rate": self.rx_sample_rate,
"center_frequency": self.rx_center_frequency,
"gain": self.rx_gain,
}
return Recording(data=store_array, metadata=metadata)
def _stream_rx(self, callback):
"""
Stream samples from the HackRF using a callback function.
:param callback: Function to call for each buffer of samples
:type callback: callable
"""
if not self._rx_initialized:
raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx()")
print("HackRF Starting RX stream...")
self._enable_rx = True
def rx_callback(hackrf_transfer):
"""Internal callback that wraps the user's callback"""
try:
if not self._enable_rx:
return 1 # Stop
c = hackrf_transfer.contents
# Use ctypes string_at to safely copy the buffer
from ctypes import string_at
byte_data = string_at(c.buffer, c.valid_length)
# Convert bytes to int8, then to float32, then view as complex64
samples = np.frombuffer(byte_data, dtype=np.int8).astype(np.float32).view(np.complex64)
# Call user's callback
callback(buffer=samples, metadata=None)
return 0 if self._enable_rx else 1
except Exception as e:
print(f"Error in rx_callback: {e}")
return 1 # Stop on error
# Start RX
self.radio.start_rx(rx_callback)
# Wait while streaming
while self._enable_rx:
time.sleep(0.1)
# Stop RX
self.radio.stop_rx()
print("HackRF RX stream completed.")
def _stream_tx(self, callback): def _stream_tx(self, callback):
return super()._stream_tx(callback) return super()._stream_tx(callback)

190
src/ria_toolkit_oss/sdr/rtlsdr.py Normal file
View File

@ -0,0 +1,190 @@
"""RTL-SDR device integration for the RIA Toolkit."""
import time
import warnings
from typing import Optional
import numpy as np
try:
from rtlsdr import RtlSdr
except ImportError as exc: # pragma: no cover - dependency provided by end user
raise ImportError("pyrtlsdr is required to use the RTLSDR class") from exc
from ria_toolkit_oss.sdr.sdr import SDR
class RTLSDR(SDR):
"""SDR interface for RTL-SDR dongles using pyrtlsdr."""
def __init__(self, identifier: Optional[int | str] = None):
super().__init__()
try:
if identifier is None:
self.radio = RtlSdr()
else:
self.radio = RtlSdr(identifier)
print(f"Initialized RTL-SDR with identifier [{identifier}].")
except Exception as exc:
print(f"Failed to initialize RTL-SDR with identifier [{identifier}].")
raise exc
self.rx_buffer_size = 256_000
self.rx_channel = 0
def supports_bias_tee(self) -> bool:
return True
def set_bias_tee(self, enable: bool):
self.radio.set_bias_tee(bool(enable))
state = "enabled" if enable else "disabled"
print(f"RTL-SDR bias tee {state}.")
def init_rx(
self,
sample_rate: int | float,
center_frequency: int | float,
gain: Optional[int],
channel: int,
gain_mode: Optional[str] = "absolute",
buffer_size: Optional[int] = 256_000,
bias_t: bool = False,
):
if channel not in (0, None):
raise ValueError("RTL-SDR supports only channel 0 for RX.")
self.radio.sample_rate = float(sample_rate)
self.rx_sample_rate = self.radio.sample_rate
self.radio.center_freq = float(center_frequency)
self.rx_center_frequency = self.radio.center_freq
available_gains = getattr(self.radio, "gains", [])
if gain is None:
self.radio.gain = "auto"
self.rx_gain = "auto"
else:
if not available_gains:
warnings.warn(
"No gain table reported by RTL-SDR; applying requested gain directly.",
RuntimeWarning,
)
target_gain = gain
else:
if gain_mode == "relative":
if gain > 0:
raise ValueError(
"When gain_mode = 'relative', gain must be < 0. This sets the gain relative to the maximum."
)
target_gain = max(available_gains) + gain
else:
target_gain = gain
min_gain = min(available_gains)
max_gain = max(available_gains)
if target_gain < min_gain or target_gain > max_gain:
print(f"Requested gain {target_gain} dB out of range; clamping to valid span {min_gain}-{max_gain} dB.")
target_gain = min(max(target_gain, min_gain), max_gain)
target_gain = min(available_gains, key=lambda g: abs(g - target_gain))
self.radio.gain = target_gain
self.rx_gain = self.radio.gain
self.rx_buffer_size = int(buffer_size or self.rx_buffer_size)
self.rx_channel = 0
if bias_t:
self.set_bias_tee(True)
time.sleep(1)
self._rx_initialized = True
self._tx_initialized = False
def init_tx(self, *args, **kwargs): # pragma: no cover - RTL-SDR is RX only
raise NotImplementedError("RTL-SDR does not support transmit operations")
def record(self, num_samples):
"""
Record a fixed number of samples from RTL-SDR.
Args:
num_samples: Number of samples to capture
Returns:
Recording object with captured samples
"""
from ria_toolkit_oss.datatypes.recording import Recording
if not self._rx_initialized:
raise RuntimeError("RX was not initialized. init_rx() must be called before record().")
print("RTL-SDR Starting RX...")
# RTL-SDR has USB buffer limitations - use consistent 256k chunks
# Always read full chunks to avoid USB overflow issues with partial reads
max_samples_per_read = 262144 # 256k samples = stable chunk size
num_full_reads = num_samples // max_samples_per_read
remainder = num_samples % max_samples_per_read
signal = np.array([], dtype=np.complex64)
# Read full chunks
for i in range(num_full_reads):
try:
chunk = self.radio.read_samples(max_samples_per_read)
signal = np.append(signal, chunk)
except Exception as e:
print(f"Error while reading samples: {e}")
break
# Read remainder if needed (round up to power of 2 for USB compatibility)
if remainder > 0 and len(signal) == num_full_reads * max_samples_per_read:
# Round up to next 16k boundary for USB stability
padded_remainder = ((remainder + 16383) // 16384) * 16384
try:
chunk = self.radio.read_samples(padded_remainder)
signal = np.append(signal, chunk[:remainder]) # Only keep what we need
except Exception as e:
print(f"Error while reading final chunk: {e}")
print("RTL-SDR RX Completed.")
# Create 1xN array for single-channel recording
store_array = np.zeros((1, len(signal)), dtype=np.complex64)
store_array[0, :] = signal
metadata = {
"source": self.__class__.__name__,
"sample_rate": self.rx_sample_rate,
"center_frequency": self.rx_center_frequency,
"gain": self.rx_gain,
}
return Recording(data=store_array, metadata=metadata)
def _stream_rx(self, callback):
if not self._rx_initialized:
raise RuntimeError("RX was not initialized. init_rx() must be called before _stream_rx() or record().")
print("RTL-SDR Starting RX...")
self._enable_rx = True
try:
while self._enable_rx:
samples = self.radio.read_samples(self.rx_buffer_size)
callback(buffer=np.asarray(samples, dtype=np.complex64), metadata=None)
finally:
print("RTL-SDR RX Completed.")
def _stream_tx(self, callback): # pragma: no cover - RTL-SDR is RX only
raise NotImplementedError("RTL-SDR does not support transmit operations")
def close(self):
try:
self.radio.close()
finally:
self._enable_rx = False
self._enable_tx = False
def set_clock_source(self, source): # pragma: no cover - not applicable to RTL-SDR
raise NotImplementedError("RTL-SDR does not support external clock configuration")

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

@ -301,12 +301,20 @@ class SDR(ABC):
def pause_tx(self): def pause_tx(self):
self._enable_tx = False self._enable_tx = False
def stop(self): def stop(self):
self.pause_rx() self.pause_rx()
@abstractmethod def supports_bias_tee(self) -> bool:
def close(self): """Return True when the radio supports bias-tee control."""
pass return False
def set_bias_tee(self, enable: bool):
"""Enable or disable bias-tee power when supported by the radio."""
raise NotImplementedError(f"{self.__class__.__name__} does not support bias-tee control")
@abstractmethod
def close(self):
pass
@abstractmethod @abstractmethod
def init_rx(self, sample_rate, center_frequency, gain, channel, gain_mode): def init_rx(self, sample_rate, center_frequency, gain, channel, gain_mode):