InstroAWG

Unstable APIInstroAWG ships in the instro-unstable package. Its API is not settled and may change without notice between releases. Install it with pip install "instro[unstable]". See Unstable modules for details.

InstroAWG is a hardware abstraction layer (HAL) that provides a unified interface for arbitrary waveform generators. The category class defines the vendor-independent API (set_waveform, set_amplitude, set_offset, set_modulation, …). A vendor-specific driver owns its connection details and translates those calls into vendor commands.

Supported Vendors

Rigol: DG1022Z (DG1000Z series) via SCPI/VISA (RigolDG1022Z)

If your vendor or model is not listed, see Custom Driver Development below.

Key Concepts

Driver Composition

An InstroAWG is built from a concrete driver:

InstroAWG("name", driver=RigolDG1022Z(visa_resource="USB0::..."), num_channels=2)

The RigolDG1022Z owns the connection setup and vendor-specific command mapping.
InstroAWG owns the category-level workflow: waveform programming, publishers, the background daemon.

Lifecycle

The typical InstroAWG workflow:

Construct: instantiate the vendor driver and pass it to InstroAWG, along with the channel count.
open(): establishes the connection to the instrument.
Configure and generate: define a waveform on a channel, set its amplitude and offset, and enable the output, etc.
start(): begins a periodic background daemon that polls output state. (Optional)
stop(): ends the background daemon (if started).
close(): disconnects from hardware.

Waveform Definitions

InstroAWG supports programming a channel with the following waveforms: Sine, Square, Sawtooth, Triangle, Pulse, Arbitrary, or StaticValue. These are all frozen dataclasses in instro.unstable.awg.

Driver support variesCertain features are only available on particular waveforms. Driver’s own their implementation of features in InstroAWG. Consult your specific driver for exact support.

Creating an InstroAWG Instance

from instro.unstable.awg.drivers import RigolDG1022Z
from instro.unstable.awg import InstroAWG

awg = InstroAWG(
    name="myAWG",
    driver=RigolDG1022Z(visa_resource="USB0::0x1AB1::0x0642::DG1ZA000000000::INSTR"),
    num_channels=2,
)

Parameters

name: A name for this AWG instance. Used as a prefix for channel names when publishing.
driver: A concrete AWGDriverBase instance (e.g. RigolDG1022Z) configured with the connection details for that model.
num_channels: Number of output channels on the waveform generator.
publishers: Optional list of publishers to attach.
**kwargs: Additional keyword arguments become default tags when using a publisher that supports tags (like NominalCorePublisher).

Choosing a Driver

Choose the concrete driver that matches the AWG model, then pass the instrument connection settings to that driver. For example, use RigolDG1022Z for the Rigol DG1022Z. To inspect a VISA instrument’s identity before choosing a driver:

from instro.lib.transports import VisaDriver

visa = VisaDriver("USB0::...")
try:
    visa.open()
    print(visa.query("*IDN?"))
finally:
    visa.close()

Examples

Basic Usage

from instro.unstable.awg.drivers import RigolDG1022Z
from instro.unstable.awg import InstroAWG
from instro.unstable.awg.types import AmplitudeMeasurementUnit, Sine
from instro.lib.publishers import NominalCorePublisher

VISA_RESOURCE = "USB0::0x1AB1::0x0642::DG1ZA000000000::INSTR"
DATASET_RID = "<your dataset here>"

awg = InstroAWG(
    name="myAWG",
    driver=RigolDG1022Z(visa_resource=VISA_RESOURCE),
    num_channels=2,
)
awg.add_publisher(NominalCorePublisher(dataset_rid=DATASET_RID))

awg.open()

# Configure channel 1
awg.set_waveform(1, Sine(frequency_hz=1000.0))
awg.set_amplitude(1, 2.0, AmplitudeMeasurementUnit.VPP)
awg.output_enable(1, True)

# Read channel 1
waveform = awg.get_waveform(1)
enabled = awg.get_output_state(1)
print(f"Waveform: {waveform}, Output enabled: {enabled.latest}")

awg.output_enable(1, False)
awg.close()

Important Note about PublishersData is published as a direct result of an instrument method being called.For example, when you call get_output_state(), this not only queries the instrument for the output state but also causes all attached Publishers to publish the measurement response automatically.

Published channels

Every measurement/command call produces a channel keyed under {name}.{descriptor}, where {name} is the constructor argument and {descriptor} is the row below. Substitute {N} with the actual channel number (1, 2, …).

Method	Descriptor	Type
`set_waveform(channel=N)`	`ch{N}.waveform.cmd`	command
`set_waveform(channel=N)`	`ch{N}.{param}.cmd` (one per numeric shape parameter, e.g. `frequency_hz`)	command
`set_amplitude(channel=N)`	`ch{N}.amplitude.cmd`	command
`set_offset(channel=N)`	`ch{N}.offset.cmd`	command
`output_enable(channel=N)`	`ch{N}.enabled.cmd`	command
`set_output_load(channel=N)`	`ch{N}.load.cmd`	command
`align_phase()`	`phase.align.cmd`	command
`set_modulation(channel=N)`	`ch{N}.modulation.cmd`	command
`modulation_enable(channel=N)`	`ch{N}.modulation_enabled.cmd`	command
`set_burst(channel=N)`	`ch{N}.burst.cmd`	command
`burst_enable(channel=N)`	`ch{N}.burst_enabled.cmd`	command
`set_burst_trigger(channel=N)`	`ch{N}.burst_trigger.cmd`	command
`set_burst_delay(channel=N)`	`ch{N}.burst_delay.cmd`	command
`set_gate_polarity(channel=N)`	`ch{N}.gate_polarity.cmd`	command
`set_ncycles(channel=N)`	`ch{N}.ncycles.cmd`	command
`set_burst_period(channel=N)`	`ch{N}.burst_period.cmd`	command
`get_offset(channel=N)`	`ch{N}.offset`	telemetry
`get_output_state(channel=N)`	`ch{N}.enabled`	telemetry
`get_output_load(channel=N)`	`ch{N}.load`	telemetry
`get_modulation_state(channel=N)`	`ch{N}.modulation_enabled`	telemetry
`get_burst_state(channel=N)`	`ch{N}.burst_enabled`	telemetry
`get_burst_delay(channel=N)`	`ch{N}.burst_delay`	telemetry
`get_burst_ncycles(channel=N)`	`ch{N}.ncycles`	telemetry
`get_burst_period(channel=N)`	`ch{N}.burst_period`	telemetry

get_waveform(), get_amplitude(), convert_amplitude(), get_modulation_type(), get_burst_type(), and get_gate_polarity() return a plain Python value (Waveform, tuple[float, AmplitudeMeasurementUnit], float, ModulationType, BurstType, GatePolarity) directly rather than a Measurement, and don’t publish. Every other readback listed above publishes a Measurement on the descriptor shown.

Method Reference

Method	Purpose
`InstroAWG(name, driver, num_channels, publishers=None, **kwargs)`	Construct an InstroAWG with a vendor driver
`open()`	Establish connection to the AWG
`close()`	Disconnect from the AWG and close all publishers
`set_waveform(channel, waveform)`	Program a channel with a `Waveform` definition
`get_waveform(channel)`	Read back the current waveform from a channel
`set_amplitude(channel, amplitude, unit)`	Set the output amplitude on a channel
`get_amplitude(channel)`	Read back the current amplitude and its unit on a channel
`convert_amplitude(channel, amplitude, from_unit, to_unit, impedance_ohms=None)`	Convert an amplitude value between units for a channel’s configured waveform
`set_offset(channel, offset_v)`	Set the DC offset in volts on a channel
`get_offset(channel)`	Read back the DC offset in volts on a channel
`output_enable(channel, enable)`	Enable (True) or disable (False) the output on a channel
`get_output_state(channel)`	Read back whether the output is enabled on a channel
`set_output_load(channel, load)`	Set the output load impedance (`None` means high-Z)
`get_output_load(channel)`	Read back the output load impedance on channel (high-Z publishes as `float('inf')`)
`align_phase()`	Sync the phase of all channels
`set_modulation(channel, mod_type, shape, magnitude)`	Configure a channel’s modulation (AM/FM/PM/PWM/ASK/FSK/PSK) without enabling it
`modulation_enable(channel, enable)`	Enable or disable modulation on channel
`get_modulation_type(channel)`	Read back the modulation type currently active on channel
`get_modulation_state(channel)`	Read back whether modulation is enabled on channel
`set_burst(channel, burst_type)`	Configure a channel’s burst type (`NCYCLE`/`GATED`/`INFINITE`) without enabling it
`burst_enable(channel, enable)`	Enable or disable burst mode on channel
`get_burst_type(channel)`	Read back the burst type currently active on channel
`get_burst_state(channel)`	Read back whether burst mode is enabled on channel
`set_burst_trigger(channel, source)`	Set the burst trigger source (`INTERNAL`/`EXTERNAL`/`MANUAL`) on channel
`set_burst_delay(channel, delay_s)`	Set the burst trigger delay in seconds on channel
`get_burst_delay(channel)`	Read back the burst trigger delay in seconds on channel
`set_gate_polarity(channel, gate_polarity)`	Set the gate polarity (`NORM`/`INV`) for GATED bursts on channel
`get_gate_polarity(channel)`	Read back the gate polarity for GATED bursts on channel
`set_ncycles(channel, n_cycles)`	Set the number of cycles per trigger for NCYCLE bursts on channel
`get_burst_ncycles(channel)`	Read back the number of cycles per trigger for NCYCLE bursts on channel
`set_burst_period(channel, period)`	Set the internal burst period in seconds on channel
`get_burst_period(channel)`	Read back the internal burst period in seconds on channel
`start()`	Begin background daemon
`stop()`	End background daemon

Custom Driver Development

This section is for developers implementing InstroAWG support for waveform generators that aren’t supported out of the box.

Overview

Driver developers subclass AWGDriverBase and own whatever transport their instrument needs. The caller chooses a concrete driver, and that concrete driver exposes connection parameters that make sense for its protocol:

awg = InstroAWG(
    name="labAWG",
    driver=MyVendorAWG(host="10.0.0.42"),
    num_channels=1,
)

The driver is responsible for translating InstroAWG’s vendor-independent API (set_waveform, set_amplitude, output_enable, …) into vendor-specific commands.

Driver Responsibilities

An AWG driver must:

Expose a protocol-native constructor: accept inputs like visa_resource, host, port, depending on the instrument.
Own transport setup: create and store the transport internally. Do not require users to pass a VisaDriver or other transport object.
Own lifecycle: implement open() and close() by opening and closing the underlying transport.
Map commands: translate each abstract method into vendor-specific commands.
Parse responses: convert instrument responses to the expected Python types (Waveform, float, bool, ModulationType, etc.).
Validate hardware constraints: if the instrument only supports a subset of waveform shapes, modulation types, or carrier/modulator combinations, raise ValueError for unsupported combinations rather than silently misprogramming the instrument.

AWGDriverBase Interface

All AWG drivers subclass AWGDriverBase. The required methods are declared @abc.abstractmethod:

def open(self) -> None:
    """Open the underlying transport."""

def close(self) -> None:
    """Close the underlying transport."""

def set_waveform(self, channel: int, waveform: Waveform) -> None:
    """Program channel with the waveform definition; raise ValueError if the definition is unsupported."""

def get_waveform(self, channel: int) -> Waveform:
    """Get the current waveform on channel; drivers may return the last-programmed definition if not readable."""

def set_amplitude(self, channel: int, amplitude: float, unit: AmplitudeMeasurementUnit) -> None:
    """Set the output amplitude on channel."""

def get_amplitude(self, channel: int) -> tuple[float, AmplitudeMeasurementUnit]:
    """Get the current output amplitude and voltage unit on channel."""

def set_offset(self, channel: int, offset: float) -> None:
    """Set the DC offset (volts) on channel."""

def get_offset(self, channel: int) -> float:
    """Get the DC offset (volts) on channel."""

def output_enable(self, channel: int, enable: bool) -> None:
    """Enable or disable the output on channel."""

def get_output_state(self, channel: int) -> bool:
    """Return True if the output on channel is enabled."""

Optional overrides (raise NotImplementedError if not supported):

set_output_load(channel, load) / get_output_load(channel): Set or read the output load impedance; None means high-Z.
align_phase(): Sync the phase of all channels.
set_modulation(channel, mod_type, shape, magnitude): Configure a channel’s modulation. Call modulation_enable() to activate it.
modulation_enable(channel, enable): Enable or disable modulation on channel.
get_modulation_type(channel): Read back the active modulation type from the instrument.
get_modulation_state(channel): Read back whether modulation is enabled from the instrument.
set_burst(channel, burst_type) / get_burst_type(channel): Configure or read back a channel’s burst type (NCYCLE/GATED/INFINITE). Call burst_enable() to activate it.
burst_enable(channel, enable) / get_burst_state(channel): Enable/disable burst mode, or read back whether it’s enabled.
set_burst_trigger(channel, source): Set the burst trigger source (INTERNAL/EXTERNAL/MANUAL).
set_burst_delay(channel, delay_s) / get_burst_delay(channel): Set or read back the burst trigger delay in seconds.
set_gate_polarity(channel, gate_polarity) / get_gate_polarity(channel): Set or read back the gate polarity (NORM/INV) for GATED bursts.
set_ncycles(channel, n_cycles) / get_burst_ncycles(channel): Set or read back the number of cycles per trigger for NCYCLE bursts.
set_burst_period(channel, period) / get_burst_period(channel): Set or read back the internal burst period in seconds.

Talking to the Instrument

Concrete drivers should hide transport details behind private attributes. For VISA-backed drivers, create a VisaDriver internally and use it for all I/O:

self._visa.write(command): Send a SCPI command (no response expected).
self._visa.query(command): Send a SCPI query and receive the response string.

VisaDriver owns the resource lock. Concurrent write / query calls against the same driver are serialized automatically. Use with self._visa.lock(): when a sequence of writes and their error check need to execute atomically. See the VisaDriver guide for the full transport reference, covering configuration, terminators, timeouts, serial settings, and the raw-byte I/O path. For non-VISA instruments, follow the same shape with the protocol client your driver needs. The important part is that the public constructor describes the instrument connection, while the transport object remains an implementation detail.

Implementation Example: RigolDG1022Z Driver

Here’s an abridged shape of the Rigol DG1022Z driver:

from instro.unstable.awg.awg import AWGDriverBase
from instro.unstable.awg.types import AmplitudeMeasurementUnit, Sine, Waveform
from instro.lib.transports import VisaConfig, VisaDriver


class RigolDG1022Z(AWGDriverBase):
    """SCPI driver for the Rigol DG1022Z two-channel arbitrary waveform generator."""

    def __init__(self, visa_resource: str | VisaConfig) -> None:
        self._visa = VisaDriver(visa_resource)

    def open(self) -> None:
        self._visa.open()

    def close(self) -> None:
        self._visa.close()

    def set_waveform(self, channel: int, waveform: Waveform) -> None:
        with self._visa.lock():
            if isinstance(waveform, Sine):
                self._visa.write(f":SOUR{channel}:FUNC SIN")
                self._visa.write(f":SOUR{channel}:FREQ {waveform.frequency_hz}")
                self._visa.write(f":SOUR{channel}:PHAS {waveform.phase_deg % 360.0}")
            # ... other Waveform shapes ...
            else:
                raise ValueError(f"unsupported waveform definition {type(waveform).__name__}")
            self._check_errors()  # one check per call, not one per write — see note above

    def set_amplitude(self, channel: int, amplitude: float, unit: AmplitudeMeasurementUnit) -> None:
        if unit is AmplitudeMeasurementUnit.VP:
            raise ValueError("the DG1022Z has no VP amplitude unit; convert to VPP, VRMS, or DBM")
        with self._visa.lock():
            self._visa.write(f":SOUR{channel}:VOLT:UNIT {unit.value}")
            self._visa.write(f":SOUR{channel}:VOLT {amplitude}")
            self._check_errors()

    # ... other required and optional methods ...

    def _check_errors(self) -> None:
        err = self._visa.query(":SYST:ERR?")
        code = err.strip().split(",", 1)[0].lstrip("+")
        if code != "0":
            raise RuntimeError(f"Rigol DG1022Z reported error: {err.strip()}")

Using a Custom Driver

For drivers that aren’t shipped in the library, construct InstroAWG with your own driver instance. The driver should accept connection settings directly and create its transport internally:

from instro.unstable.awg import InstroAWG
from instro.unstable.awg.awg import AWGDriverBase
from instro.unstable.awg.types import AmplitudeMeasurementUnit, Waveform, Sine
from instro.lib.transports import VisaDriver


class MyCustomAWGDriver(AWGDriverBase):
    """Custom driver for my lab's proprietary AWG."""

    def __init__(self, visa_resource: str) -> None:
        self._visa = VisaDriver(visa_resource)

    def open(self) -> None:
        self._visa.open()

    def close(self) -> None:
        self._visa.close()

    def set_waveform(self, channel: int, waveform: Waveform) -> None:
        if isinstance(waveform, Sine):
            self._visa.write(f"CH{channel}:WAVE SIN,{waveform.frequency_hz}")
            self._check_errors()
        else:
            raise ValueError(f"unsupported waveform definition {type(waveform).__name__}")

    # ... implement other required methods ...

    def get_output_state(self, channel: int) -> bool:
        return self._visa.query(f"CH{channel}:OUTP?") == "ON"

    def _check_errors(self) -> None:
        err = self._visa.query("ERR?")
        if err != "OK":
            raise RuntimeError(f"AWG error: {err}")


awg = InstroAWG(
    name="labAWG",
    driver=MyCustomAWGDriver(visa_resource="<VISA_ADDRESS>"),
    num_channels=1,
)

awg.open()
awg.set_waveform(1, Sine(frequency_hz=1000.0))
awg.close()

Summary

Driver development requires careful mapping of vendor-specific behavior to the unified InstroAWG interface. Focus on:

Subclassing AWGDriverBase
Designing a constructor around natural connection parameters for the instrument
Hiding transport construction inside the driver
Implementing all abstract methods on AWGDriverBase (and optional overrides where supported)
Using the correct vendor protocol or command syntax
Converting instrument responses to the expected Python types
Validating carrier/modulator/waveform compatibility your instrument actually supports, raising ValueError rather than misprogramming the instrument
Querying and reporting errors from the instrument’s error queue where one exists
Testing with actual hardware to ensure commands work as expected

Transports

Migration

Arbitrary Waveform Generator (AWG)

InstroAWG

Supported Vendors

Key Concepts

Driver Composition

Lifecycle

Waveform Definitions

Creating an InstroAWG Instance

Parameters

Choosing a Driver

Examples

Basic Usage

Published channels

Method Reference

Custom Driver Development

Overview

Driver Responsibilities

AWGDriverBase Interface

Talking to the Instrument

Implementation Example: RigolDG1022Z Driver

Using a Custom Driver

Summary

​InstroAWG

​Supported Vendors

​Key Concepts

​Driver Composition

​Lifecycle

​Waveform Definitions

​Creating an InstroAWG Instance

​Parameters

​Choosing a Driver

​Examples

​Basic Usage

​Published channels

​Method Reference

​Custom Driver Development

​Overview

​Driver Responsibilities

​AWGDriverBase Interface

​Talking to the Instrument

​Implementation Example: RigolDG1022Z Driver

​Using a Custom Driver

​Summary

InstroAWG

Supported Vendors

Key Concepts

Driver Composition

Lifecycle

Waveform Definitions

Creating an InstroAWG Instance

Parameters

Choosing a Driver

Examples

Basic Usage

Published channels

Method Reference

Custom Driver Development

Overview

Driver Responsibilities

AWGDriverBase Interface

Talking to the Instrument

Implementation Example: RigolDG1022Z Driver

Using a Custom Driver

Summary