InstroSDR

class InstroSDR(

name: str,

driver: SDRDriverBase,

**kwargs: Any,

)

Bases: Instrument

Software defined radio instrument that publishes IQ blocks and spectrum features.

Each IQ block publishes as one Measurement with paired .i and .q channels per signal path, timestamped from the sample period the device reports.

Initialize an InstroSDR.

Parameters:

• name (str)

  Channel-name prefix for published data.

• driver (SDRDriverBase)

  Concrete vendor driver.

• **kwargs

  Passed to Instrument: publishers and background_config, dataset_rid to attach a NominalCorePublisher, and any other keyword as a default tag.

Example

from instro.unstable.sdr import InstroSDR

sdr = InstroSDR(name="sdr", driver=MyVendorSDR(...))  # any concrete SDRDriverBase
sdr.open()
try:
    sdr.set_center_freq(100e6)
    sdr.set_sample_rate(2.4e6)
    iq = sdr.measure_iq(n_samples=4096)
    spectrum = sdr.measure_spectrum(n_samples=4096)
finally:
    sdr.close()

Attributes

background_interval	Background daemon loop period.
channel_names	Return an insertion-ordered snapshot of all published channel names.
driver	The underlying hardware driver.

Methods

add_background_daemon_function	Append method to the daemon's call list.
add_publisher	Register a publisher to receive this instrument's Measurement/Command data.
close	Stop the background daemon and close the driver.
compute_psd	Acquire one block and return one channel's power spectral density.
define_background_daemon	Replace all daemon functions with a single method (called with the given args).
fetch_iq	Return the next contiguous block from the running stream and publish it.
get_antenna	Query the selected antenna port and publish it as antenna.
get_backlog	Return how many samples are waiting on the receive stream.
get_bandwidth	Query the IF or filter bandwidth and publish it as bandwidth.
get_center_freq	Query the RF center frequency and publish it as center_freq.
get_channel	Return the most recent length samples for channel_name from the in-memory buffer.
get_freq_correction	Query the reference oscillator correction and publish it as freq_correction.
get_gain	Query the gain and publish it as gain.
get_gain_mode	Query the gain mode and publish it as gain_mode.
get_sample_rate	Query the IQ sample rate and publish it as sample_rate.
get_single_channel_value	Return the most recent sample for channel_name from the in-memory buffer.
is_streaming	Report whether a stream is running.
measure_iq	Acquire one time-aligned IQ block and publish it as paired .i and .q channels.
measure_spectrum	Acquire one block and publish four scalar spectrum features per channel.
open	Open the underlying driver.
publish	Fan data out to every configured publisher; kwargs pass through.
set_antenna	Set the antenna port and publish it as antenna.cmd.
set_bandwidth	Set the IF or filter bandwidth and publish it as bandwidth.cmd.
set_center_freq	Set the RF center frequency and publish it as center_freq.cmd.
set_freq_correction	Set the reference oscillator correction and publish it as freq_correction.cmd.
set_gain	Set the gain and publish it as gain.cmd.
set_gain_mode	Hand gain to the AGC or take manual control, and publish it as gain_mode.cmd.
set_sample_rate	Set the IQ sample rate and publish it as sample_rate.cmd.
start	Begin continuous acquisition.
stop	Stop the background daemon and every running hardware stream.

open() → None

Open the underlying driver.

close() → None

Stop the background daemon and close the driver.

The driver is closed even if the daemon does not shut down cleanly; that failure is logged.

property driver: SDRDriverBase

The underlying hardware driver.

Returns:

SDRDriverBase – The driver passed at construction.

measure_iq(

n_samples: int = 1024,

*,

channels: Sequence[str] | None = None,

**kwargs: Any,

) → Measurement | None

Acquire one time-aligned IQ block and publish it as paired .i and .q channels.

While a receive stream runs, the block is taken from that stream rather than the device, so it stays contiguous with the surrounding fetches and also publishes backlog and overflow.

Parameters:

• n_samples (int)

  Samples per channel.

• channels (Sequence[str] | None)

  Receive channels to read. None means every channel the running stream covers, or ("0",) with no stream running.

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Measurement | None – The published block, or None if the read returned no samples.

Raises:

ValueError – If n_samples is not positive.

start(

*,

direction: Direction = Direction.RX,

channels: Sequence[str] = ('0',),

background: bool = False,

n_samples: int = 1024,

publish_spectrum: bool = False,

) → None

Begin continuous acquisition.

With background=True the daemon runs fetch_iq in a loop, which the blocking fetch paces, so blocks publish without a loop of your own.

Parameters:

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channels (Sequence[str])

  Channels the stream covers.

• background (bool)

  Hand the fetch loop to the background daemon.

• n_samples (int)

  Samples per channel per block, when background is set.

• publish_spectrum (bool)

  Also publish spectrum features for each block, when background is set.

property background_interval: float

Background daemon loop period.

start(background=True) sets it to 0: fetch_iq blocks until the radio has the samples, so it paces the loop itself.

Returns:

float – Loop period in seconds.

stop(**kwargs: Any) → None

Stop the background daemon and every running hardware stream.

Parameters:

**kwargs

Ignored.

fetch_iq(

n_samples: int = 1024,

*,

publish_spectrum: bool = False,

**kwargs: Any,

) → Measurement | None

Return the next contiguous block from the running stream and publish it.

Publishes the block as paired .i and .q channels for every channel the stream covers, plus backlog and overflow stream-health channels.

Parameters:

• n_samples (int)

  Samples per channel.

• publish_spectrum (bool)

  Also publish spectrum features computed from this block, without consuming another.

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Measurement | None – The published block, or None if the fetch returned no samples.

Raises:

ValueError – If n_samples is not positive.

get_backlog() → int

Return how many samples are waiting on the receive stream.

Returns:

int – Samples per channel.

Raises:

NotImplementedError – If the driver does not support streaming.

is_streaming(

*,

direction: Direction = Direction.RX,

) → bool

Report whether a stream is running.

Parameters:

direction (Direction)

Direction to check.

Returns:

bool – True while a stream started through this instrument is running.

compute_psd(

n_samples: int = 1024,

*,

channel: str = '0',

) → tuple[ndarray, ndarray] | None

Acquire one block and return one channel’s power spectral density. Publishes nothing.

Uses a Hann-windowed FFT. Frequencies are absolute, offset by the tuned center frequency.

Parameters:

• n_samples (int)

  Samples to acquire, which sets the frequency resolution.

• channel (str)

  Receive channel to analyze.

Returns:

tuple[numpy.ndarray, numpy.ndarray] | None – (frequencies_hz, power_db), frequencies in Hz and power in dB relative to one squared sample unit per Hz. None if the read returned no samples.

Raises:

ValueError – If n_samples is not positive.

measure_spectrum(

n_samples: int = 1024,

*,

channels: Sequence[str] | None = None,

**kwargs: Any,

) → Measurement | None

Acquire one block and publish four scalar spectrum features per channel.

Publishes spectrum.peak_power_db, spectrum.peak_freq_hz (Hz), spectrum.mean_power_db, and spectrum.occupied_bw_hz (Hz, the band holding 99% of total power). The spectrum array itself is not published; use compute_psd for it.

Parameters:

• n_samples (int)

  Samples per channel.

• channels (Sequence[str] | None)

  Receive channels to analyze. None means every channel the running stream covers, or ("0",) with no stream running.

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Measurement | None – The published features, or None if the read returned no samples.

Raises:

ValueError – If n_samples is not positive.

set_center_freq(

frequency_hz: float,

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Command

Set the RF center frequency and publish it as center_freq.cmd.

Parameters:

• frequency_hz (float)

  Center frequency in Hz.

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Command – The published command.

get_center_freq(

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Measurement

Query the RF center frequency and publish it as center_freq.

Parameters:

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Measurement – The published reading, in Hz.

set_sample_rate(

sample_rate_hz: float,

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Command

Set the IQ sample rate and publish it as sample_rate.cmd.

Parameters:

• sample_rate_hz (float)

  Sample rate in samples per second.

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Command – The published command.

get_sample_rate(

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Measurement

Query the IQ sample rate and publish it as sample_rate.

Parameters:

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Measurement – The published reading, in samples per second.

set_gain(

gain_db: float,

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Command

Set the gain and publish it as gain.cmd.

Parameters:

• gain_db (float)

  Gain in dB.

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Command – The published command.

get_gain(

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Measurement

Query the gain and publish it as gain.

Parameters:

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Measurement – The published reading, in dB.

set_gain_mode(

automatic: bool,

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Command

Hand gain to the AGC or take manual control, and publish it as gain_mode.cmd.

Parameters:

• automatic (bool)

  True for AGC, False for manual gain.

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Command – The published command.

get_gain_mode(

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Measurement

Query the gain mode and publish it as gain_mode.

Parameters:

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Measurement – The published reading, True when AGC is in control.

set_bandwidth(

bandwidth_hz: float,

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Command

Set the IF or filter bandwidth and publish it as bandwidth.cmd.

Parameters:

• bandwidth_hz (float)

  Bandwidth in Hz.

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Command – The published command.

get_bandwidth(

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Measurement

Query the IF or filter bandwidth and publish it as bandwidth.

Parameters:

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Measurement – The published reading, in Hz.

set_freq_correction(

ppm: float,

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Command

Set the reference oscillator correction and publish it as freq_correction.cmd.

Parameters:

• ppm (float)

  Correction in parts per million.

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Command – The published command.

get_freq_correction(

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Measurement

Query the reference oscillator correction and publish it as freq_correction.

Parameters:

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Measurement – The published reading, in parts per million.

set_antenna(

antenna: str,

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Command

Set the antenna port and publish it as antenna.cmd.

Parameters:

• antenna (str)

  Port name, one of the driver’s list_antennas.

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Command – The published command.

add_background_daemon_function(

method: Callable,

*args,

**kwargs,

)

Append method to the daemon’s call list. Use define_background_daemon to replace instead.

add_publisher(

publisher: Publisher,

)

Register a publisher to receive this instrument’s Measurement/Command data.

property channel_names: tuple[str, ...]

Return an insertion-ordered snapshot of all published channel names.

Names are complete after one full daemon iteration: {name}.loop_time publishes every iteration, so call get_channel(“loop_time”, wait_for_new_samples=True) then read channel_names to ensure the complete set has arrived.

Raises:

RuntimeError – No background buffer; start() was not called.

define_background_daemon(

method: Callable,

*args,

**kwargs,

)

Replace all daemon functions with a single method (called with the given args).

get_antenna(

*,

direction: Direction = Direction.RX,

channel: str = '0',

**kwargs: Any,

) → Measurement

Query the selected antenna port and publish it as antenna.

Parameters:

• direction (Direction)

  Signal path, Direction.RX or Direction.TX.

• channel (str)

  Channel on that path, e.g. "0".

• **kwargs

  Tags added to the published value, on top of the instrument’s default tags.

Returns:

Measurement – The published reading, the port name.

get_channel(

channel_name: str,

length: int = 1,

wait_for_new_samples: bool = False,

timeout: float = 10.0,

) → Measurement

Return the most recent length samples for channel_name from the in-memory buffer.

If the channel does not exist yet, or no sample is available, the code will always block until timeout expires.

Parameters:

• channel_name

  Name of the channel to retrieve.

• length

  Number of trailing samples to return.

• wait_for_new_samples

  Block until at least length new values arrive.

• timeout

  Seconds to wait when insufficient data exists or wait_for_new_samples=True.

Raises:

• RuntimeError – No background buffer; start() was not called.

• ChannelNotFoundError – channel had no values and no data appeared before timeout. wait_for_new_samples=True and channel did not appear within timeout.

• ChannelValueTimeoutError – wait_for_new_samples=True and values did not arrive within timeout.

get_single_channel_value(

channel_name: str,

) → float | str | None

Return the most recent sample for channel_name from the in-memory buffer.

This will not wait, if data is not available then None is returned.

Parameters:

channel_name

Name of the channel to retrieve.

Raises:

RuntimeError – No background buffer; start() was not called.

publish(

data: Measurement | Command,

**kwargs,

)

Fan data out to every configured publisher; kwargs pass through.