# InstroSDR

## class InstroSDR( name: str, driver: SDRDriverBase, **kwargs: Any, )[source]

Bases: [`Instrument`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument "instro.lib.instrument.Instrument")

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

Each IQ block publishes as one [`Measurement`](https://dev.nominal.io/instro/ref/library/instro.lib.types.Measurement.md#instro.lib.types.Measurement "instro.lib.types.Measurement") with paired `.i` and `.q`
channels per signal path, timestamped from the sample period the device reports.

Initialize an InstroSDR.

Parameters:
:   - **name** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      Channel-name prefix for published data.
    - **driver** ([*SDRDriverBase*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.SDRDriverBase.md#instro.unstable.sdr.SDRDriverBase "instro.unstable.sdr.SDRDriverBase"))

      Concrete vendor driver.
    - ****kwargs**

      Passed to [`Instrument`](https://dev.nominal.io/instro/ref/library/instro.lib.instrument.Instrument.md#instro.lib.instrument.Instrument "instro.lib.instrument.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`](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.InstroSDR.md#instro.unstable.sdr.InstroSDR.background_interval "instro.unstable.sdr.InstroSDR.background_interval") | Background daemon loop period. |
| --- | --- |
| [`channel_names`](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.InstroSDR.md#instro.unstable.sdr.InstroSDR.channel_names "instro.unstable.sdr.InstroSDR.channel_names") | Return an insertion-ordered snapshot of all published channel names. |
| [`driver`](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.InstroSDR.md#instro.unstable.sdr.InstroSDR.driver "instro.unstable.sdr.InstroSDR.driver") | The underlying hardware driver. |

Methods

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

### open() → None[source]

Open the underlying driver.

### close() → None[source]

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[source]

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*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      Samples per channel.
    - **channels** (*Sequence**[*[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*]* *|* *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**](https://docs.python.org/3/builtins/exceptions.html#ValueError "(in Python v3.15)") – 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[source]

Begin continuous acquisition.

With `background=True` the daemon runs [`fetch_iq`](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.InstroSDR.md#instro.unstable.sdr.InstroSDR.fetch_iq "instro.unstable.sdr.InstroSDR.fetch_iq") in a loop, which the blocking fetch
paces, so blocks publish without a loop of your own.

Parameters:
:   - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channels** (*Sequence**[*[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*]*)

      Channels the stream covers.
    - **background** ([*bool*](https://docs.python.org/3/builtins/functions.html#bool "(in Python v3.15)"))

      Hand the fetch loop to the background daemon.
    - **n_samples** ([*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      Samples per channel per block, when `background` is set.
    - **publish_spectrum** ([*bool*](https://docs.python.org/3/builtins/functions.html#bool "(in Python v3.15)"))

      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`](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.InstroSDR.md#instro.unstable.sdr.InstroSDR.fetch_iq "instro.unstable.sdr.InstroSDR.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[source]

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[source]

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*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      Samples per channel.
    - **publish_spectrum** ([*bool*](https://docs.python.org/3/builtins/functions.html#bool "(in Python v3.15)"))

      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**](https://docs.python.org/3/builtins/exceptions.html#ValueError "(in Python v3.15)") – If `n_samples` is not positive.

### get_backlog() → int[source]

Return how many samples are waiting on the receive stream.

Returns:
:   *int* – Samples per channel.

Raises:
:   [**NotImplementedError**](https://docs.python.org/3/builtins/exceptions.html#NotImplementedError "(in Python v3.15)") – If the driver does not support streaming.

### is_streaming( *, direction: Direction = Direction.RX, ) → bool[source]

Report whether a stream is running.

Parameters:
:   **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.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[source]

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*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      Samples to acquire, which sets the frequency resolution.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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**](https://docs.python.org/3/builtins/exceptions.html#ValueError "(in Python v3.15)") – If `n_samples` is not positive.

### measure_spectrum( n_samples: int = 1024, *, channels: Sequence[str] | None = None, **kwargs: Any, ) → Measurement | None[source]

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`](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.InstroSDR.md#instro.unstable.sdr.InstroSDR.compute_psd "instro.unstable.sdr.InstroSDR.compute_psd") for it.

Parameters:
:   - **n_samples** ([*int*](https://docs.python.org/3/builtins/functions.html#int "(in Python v3.15)"))

      Samples per channel.
    - **channels** (*Sequence**[*[*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)")*]* *|* *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**](https://docs.python.org/3/builtins/exceptions.html#ValueError "(in Python v3.15)") – If `n_samples` is not positive.

### set_center_freq( frequency_hz: float, *, direction: Direction = Direction.RX, channel: str = '0', **kwargs: Any, ) → Command[source]

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

Parameters:
:   - **frequency_hz** ([*float*](https://docs.python.org/3/builtins/functions.html#float "(in Python v3.15)"))

      Center frequency in Hz.
    - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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[source]

Query the RF center frequency and publish it as `center_freq`.

Parameters:
:   - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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[source]

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

Parameters:
:   - **sample_rate_hz** ([*float*](https://docs.python.org/3/builtins/functions.html#float "(in Python v3.15)"))

      Sample rate in samples per second.
    - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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[source]

Query the IQ sample rate and publish it as `sample_rate`.

Parameters:
:   - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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[source]

Set the gain and publish it as `gain.cmd`.

Parameters:
:   - **gain_db** ([*float*](https://docs.python.org/3/builtins/functions.html#float "(in Python v3.15)"))

      Gain in dB.
    - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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[source]

Query the gain and publish it as `gain`.

Parameters:
:   - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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[source]

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

Parameters:
:   - **automatic** ([*bool*](https://docs.python.org/3/builtins/functions.html#bool "(in Python v3.15)"))

      `True` for AGC, `False` for manual gain.
    - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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[source]

Query the gain mode and publish it as `gain_mode`.

Parameters:
:   - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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[source]

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

Parameters:
:   - **bandwidth_hz** ([*float*](https://docs.python.org/3/builtins/functions.html#float "(in Python v3.15)"))

      Bandwidth in Hz.
    - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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[source]

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

Parameters:
:   - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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[source]

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

Parameters:
:   - **ppm** ([*float*](https://docs.python.org/3/builtins/functions.html#float "(in Python v3.15)"))

      Correction in parts per million.
    - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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[source]

Query the reference oscillator correction and publish it as `freq_correction`.

Parameters:
:   - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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[source]

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

Parameters:
:   - **antenna** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      Port name, one of the driver’s [`list_antennas`](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.SDRDriverBase.md#instro.unstable.sdr.SDRDriverBase.list_antennas "instro.unstable.sdr.SDRDriverBase.list_antennas").
    - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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**](https://docs.python.org/3/builtins/exceptions.html#RuntimeError "(in Python v3.15)") – 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[source]

Query the selected antenna port and publish it as `antenna`.

Parameters:
:   - **direction** ([*Direction*](https://dev.nominal.io/instro/ref/instruments/generated/instro.unstable.sdr.types.Direction.md#instro.unstable.sdr.types.Direction "instro.unstable.sdr.Direction"))

      Signal path, `Direction.RX` or `Direction.TX`.
    - **channel** ([*str*](https://docs.python.org/3/builtins/stdtypes.html#str "(in Python v3.15)"))

      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**](https://docs.python.org/3/builtins/exceptions.html#RuntimeError "(in Python v3.15)") – No background buffer; `start()` was not called.
    - [**ChannelNotFoundError**](https://dev.nominal.io/instro/ref/library/instro.lib.publishers.channel_buffer.ChannelNotFoundError.md#instro.lib.publishers.channel_buffer.ChannelNotFoundError "instro.lib.publishers.channel_buffer.ChannelNotFoundError") – channel had no values and no data appeared before `timeout`.
      `wait_for_new_samples=True` and channel did not appear within `timeout`.
    - [**ChannelValueTimeoutError**](https://dev.nominal.io/instro/ref/library/instro.lib.publishers.channel_buffer.ChannelValueTimeoutError.md#instro.lib.publishers.channel_buffer.ChannelValueTimeoutError "instro.lib.publishers.channel_buffer.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**](https://docs.python.org/3/builtins/exceptions.html#RuntimeError "(in Python v3.15)") – No background buffer; `start()` was not called.

### publish( data: Measurement | Command, **kwargs, )

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