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, get_offset, …). A vendor-specific driver (e.g. RigolDG1022Z) 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)

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 VISA connection.
Configure and generate: define a waveform on a channel, set its amplitude and offset, and enable the output.
start(): begins a periodic background daemon that polls output state. (Optional)
stop(): ends the background daemon (if started).
close(): disconnects from hardware.

Waveform Definitions

Each channel is programmed with a Waveform, one of the frozen dataclasses in instro.unstable.awg:

Waveform	Parameters	Notes
`Sine`	`frequency_hz`, `phase_deg`
`Square`	`frequency_hz`, `duty_cycle_pct`, `phase_deg`
`Sawtooth`	`frequency_hz`, `phase_deg`
`Triangle`	`frequency_hz`, `phase_deg`
`Pulse`	`frequency_hz`, `width_s`, `delay_s`	`width_s + delay_s` must fit within one period
`Arbitrary`	`samples`, `sample_rate_hz`	`samples` are normalized to `[-1.0, 1.0]`; at least 2 samples required
`StaticValue`	`value`	Constant (DC) output

Each definition validates its own parameters at construction time (raising ValueError for out-of-range values), so a Waveform is always well-formed before it reaches a driver.

Driver support variesNot every driver supports every waveform or every parameter combination. For example, RigolDG1022Z raises ValueError for a nonzero Pulse.delay_s, since the DG1000Z command set has no pulse-delay parameter. Consult your instrument’s manual and driver implementation for exact support.

Amplitude Units and Conversion

Amplitude is set and read together with an AmplitudeMeasurementUnit: VPP, VP, VRMS, or DBM. Use convert_amplitude() to convert a value between units for a channel’s currently configured waveform:

vrms = awg.convert_amplitude(channel=1, amplitude=5.0, from_unit=AmplitudeMeasurementUnit.VPP, to_unit=AmplitudeMeasurementUnit.VRMS)

VPP/VP/VRMS conversions depend on the waveform’s crest factor (shared math, not vendor-specific). Converting to or from DBM additionally requires the load impedance driving the output: pass impedance_ohms explicitly, or let it fall back to the channel’s get_output_load() value if the driver supports it. set_waveform must be called for the channel before converting, since the crest factor depends on the waveform shape.

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 instrument. Must be at least 1.
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 Rigol DG1000Z series generators, use RigolDG1022Z with the VISA resource string for the instrument. 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

All measurement methods return Measurement objects. This is common amongst all Instrument objects.

Basic Usage

import time

from instro.unstable.awg.drivers import RigolDG1022Z
from instro.unstable.awg import AmplitudeMeasurementUnit, InstroAWG, 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()

# Program a 1 kHz sine wave on channel 1
awg.set_waveform(channel=1, waveform=Sine(frequency_hz=1000.0))
awg.set_amplitude(channel=1, amplitude=2.0, unit=AmplitudeMeasurementUnit.VPP)
awg.set_offset(channel=1, offset_v=0.0)

# Enable the output
awg.output_enable(channel=1, enable=True)

time.sleep(0.5)

# Read back output state
enabled = awg.get_output_state(channel=1)
print(f"channel 1 enabled: {enabled.latest}")

# Disable the output
awg.output_enable(channel=1, enable=False)
awg.close()

Background Daemon for Continuous Monitoring

import time

from instro.unstable.awg.drivers import RigolDG1022Z
from instro.unstable.awg import AmplitudeMeasurementUnit, InstroAWG, Square
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))

# Configure telemetry rate (optional, default is 1 second)
awg.background_interval = 0.5  # Poll every 500ms

awg.open()

# set_waveform must be called for at least one channel before start()
awg.set_waveform(channel=1, waveform=Square(frequency_hz=500.0, duty_cycle_pct=25.0))
awg.set_amplitude(channel=1, amplitude=3.3, unit=AmplitudeMeasurementUnit.VPP)

# Start background daemon
awg.start()

awg.output_enable(channel=1, enable=True)

# Data flows automatically to publishers
while True:
    try:
        # Optionally grab data being produced by the background daemon.
        ch1_enabled = awg.get_channel(
            channel_name = "myAWG.ch1.enabled",
            length = 1,
            wait_for_new_samples = True)  # This will block for the next sample from the background daemon

    except KeyboardInterrupt:
        print("Stopping...")
        break

awg.output_enable(channel=1, enable=False)

# Stop background daemon
awg.stop()
awg.close()

In this mode:

start() begins a background daemon, executing a function or list of functions periodically.
stop() ends the background daemon.

Default AWG Background DaemonFor each :

Output enabled state (via get_output_state())

The default polling interval is 1 second, configurable via the background_interval property. Other readbacks (get_offset(), get_output_load(), waveform parameters) are opt-in: call add_background_daemon_function() to add them to the daemon’s call list.start() raises ValueError unless set_waveform() has been called for at least one channel first.

Custom Background Daemon

To define your own background daemon, call define_background_daemon(method, *args, **kwargs), which replaces the registered daemon functions.
To add a method to the background daemon stack, call add_background_daemon_function().

See Two ways to get data for more information regarding background fetching of measurements.

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.Therefore the background daemon, when calling these instrument methods, is publishing data in the background as well!

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
`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

Method Reference

Method	Purpose
`InstroAWG(name, driver, num_channels, publishers=None, **kwargs)`	Construct an InstroAWG with a vendor driver
`open()`	Establish VISA connection to the AWG
`close()`	Disconnect from the AWG and close all publishers
`set_waveform(channel, waveform)`	Program channel with a `Waveform` definition
`get_waveform(channel)`	Read back the current waveform definition on channel
`set_amplitude(channel, amplitude, unit)`	Set the output amplitude on channel
`get_amplitude(channel)`	Read back the current amplitude and its unit on channel
`convert_amplitude(channel, amplitude, from_unit, to_unit, impedance_ohms=None)`	Convert an amplitude value between units for channel’s configured waveform
`set_offset(channel, offset_v)`	Set the DC offset (volts) on channel
`get_offset(channel)`	Read back the DC offset (volts) on channel
`output_enable(channel, enable)`	Enable (True) or disable (False) the output on channel
`get_output_state(channel)`	Read back whether the output is enabled on 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
`start()`	Begin background telemetry daemon
`stop()`	End background telemetry 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", unit_id=1),
    num_channels=2,
)

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

Driver Responsibilities

An AWG driver must:

Expose a protocol-native constructor: accept inputs like visa_resource, host, port, unit_id, interface, or node_id, depending on the instrument.
Own transport setup: create and store the transport internally. Do not require users to pass a VisaDriver, socket client, Modbus client, 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, raising ValueError for a Waveform definition the instrument can’t produce.
Parse responses: convert instrument responses to the expected Python types (float, bool, Waveform subclasses).

AWGDriverBase Interface

All AWG drivers subclass AWGDriverBase and implement these abstract methods:

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

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

def check_errors(self) -> None:
    """Drain the instrument error queue; raise if any error is pending."""

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."""

Two methods are optional and raise NotImplementedError by default, for drivers whose instrument doesn’t support them:

def set_output_load(self, channel: int, load: float | None) -> None:
    """Set the output load impedance; None means high-Z."""

def get_output_load(self, channel: int) -> float | None:
    """Get the output load impedance; None means high-Z."""

def align_phase(self) -> None:
    """Sync the phase of all channels."""

Error checking is not part of the base contract beyond check_errors(): if your vendor exposes an error queue, implement check_errors() to drain and raise on it (see the representative driver below).

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 self._visa.lock() to hold the lock across a multi-command sequence (for example, programming an Arbitrary waveform point-by-point). See the VisaDriver guide for the full transport reference, covering configuration, terminators, timeouts, serial settings, and the raw-byte I/O path.

Implementation Example: Rigol DG1022Z Driver

Here’s the driver implementation for the Rigol DG1022Z (DG1000Z series) two-channel arbitrary waveform generator:

from instro.lib.transports.visa import VisaConfig, VisaDriver
from instro.unstable.awg.awg import AWGDriverBase
from instro.unstable.awg.types import (
    AmplitudeMeasurementUnit,
    Arbitrary,
    Pulse,
    Sawtooth,
    Sine,
    Square,
    StaticValue,
    Triangle,
    Waveform,
)

_HIGH_Z_SENTINEL = 9.9e37
_ARB_MIN_POINTS = 9
_ARB_MAX_POINTS = 16384


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)
        self._arb_waveforms: dict[int, Arbitrary] = {}

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

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

    def check_errors(self) -> None:
        """Drain :SYSTem:ERRor? and raise on the first non-zero code."""
        while True:
            resp = self._visa.query(":SYST:ERR?")
            code_str, _, msg = resp.partition(",")
            code = int(code_str)
            if code == 0:
                return
            raise RuntimeError(f"Rigol DG1022Z reported error {code}: {msg.strip().strip(chr(34))}")

    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}")
            elif isinstance(waveform, Arbitrary):
                # Per-point downloads keep both USB and Ethernet compatible.
                num_points = len(waveform.samples)
                if not _ARB_MIN_POINTS <= num_points <= _ARB_MAX_POINTS:
                    raise ValueError(
                        f"the DG1022Z accepts {_ARB_MIN_POINTS} to {_ARB_MAX_POINTS} arbitrary points"
                        f" per download, got {num_points}"
                    )
                self._visa.write(f":SOUR{channel}:APPL:ARB {waveform.sample_rate_hz}")
                self.check_errors()
                self._visa.write(f":SOUR{channel}:TRAC:DATA:POIN VOLATILE,{num_points}")
                self.check_errors()
                for point, sample in enumerate(waveform.samples, start=1):
                    decimal_value = round((sample + 1) / 2 * 16383)
                    self._visa.write(f":SOUR{channel}:TRAC:DATA:VAL VOLATILE,{point},{decimal_value}")
                    self.check_errors()
                self._arb_waveforms[channel] = waveform
            elif isinstance(waveform, StaticValue):
                self._visa.write(f":SOUR{channel}:FUNC DC")
                self._visa.write(f":SOUR{channel}:VOLT:OFFS {waveform.value}")
            # ... Square, Sawtooth, Triangle, Pulse map similarly ...
            else:
                raise ValueError(f"unsupported waveform definition {type(waveform).__name__}")

    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}")

    def set_offset(self, channel: int, offset: float) -> None:
        self._visa.write(f":SOUR{channel}:VOLT:OFFS {offset}")

    def get_offset(self, channel: int) -> float:
        return float(self._visa.query(f":SOUR{channel}:VOLT:OFFS?"))

    def output_enable(self, channel: int, enable: bool) -> None:
        self._visa.write(f":OUTP{channel} ON" if enable else f":OUTP{channel} OFF")

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

    def get_output_load(self, channel: int) -> float | None:
        load = float(self._visa.query(f":OUTP{channel}:LOAD?"))
        return None if load >= _HIGH_Z_SENTINEL else load

    def align_phase(self) -> None:
        self._visa.write(":SOUR1:PHAS:SYNC")

    # ... get_waveform, set_output_load, get_amplitude follow the same query/parse shape ...

Vendor SCPI VariationsDifferent AWG vendors use different SCPI command sets and support a different subset of waveform shapes. Always consult your instrument’s programming manual for the correct SCPI syntax and validate unsupported combinations with ValueError rather than sending malformed commands.

USB arbitrary downloads on some Rigol DG1000Z unitsSome DG1000Z-series units (observed on a DG1062Z, firmware 03.01.12) hang when downloading volatile arbitrary waveform data over USB, regardless of client (including Rigol’s own reference examples). If set_waveform() with an Arbitrary waveform hangs on your unit, try a LAN (TCPIP::...::INSTR) connection instead of USB, or update the instrument firmware. This is an instrument firmware issue, not a driver bug.

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 AWGDriverBase, InstroAWG, Sine
from instro.unstable.awg.types import AmplitudeMeasurementUnit, Waveform
from instro.lib.transports import VisaDriver


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

    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 check_errors(self) -> None:
        err = self._visa.query("ERR?")
        if err != "OK":
            raise RuntimeError(f"AWG error: {err}")

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

    # ... implement other required methods ...


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

awg.open()
awg.set_waveform(channel=1, waveform=Sine(frequency_hz=1000.0))
awg.output_enable(channel=1, enable=True)
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 the optional load/phase methods your instrument supports
Raising ValueError for Waveform definitions and parameter combinations the instrument can’t produce
Using the correct vendor protocol or command syntax
Converting instrument responses to the expected Python types
Implementing check_errors() against your vendor’s error queue or status register
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

Amplitude Units and Conversion

Creating an InstroAWG Instance

Parameters

Choosing a Driver

Examples

Basic Usage

Background Daemon for Continuous Monitoring

Published channels

Method Reference

Custom Driver Development

Overview

Driver Responsibilities

AWGDriverBase Interface

Talking to the Instrument

Implementation Example: Rigol DG1022Z Driver

Using a Custom Driver

Summary

​InstroAWG

​Supported Vendors

​Key Concepts

​Driver Composition

​Lifecycle

​Waveform Definitions

​Amplitude Units and Conversion

​Creating an InstroAWG Instance

​Parameters

​Choosing a Driver

​Examples

​Basic Usage

​Background Daemon for Continuous Monitoring

​Published channels

​Method Reference

​Custom Driver Development

​Overview

​Driver Responsibilities

​AWGDriverBase Interface

​Talking to the Instrument

​Implementation Example: Rigol DG1022Z Driver

​Using a Custom Driver

​Summary

InstroAWG

Supported Vendors

Key Concepts

Driver Composition

Lifecycle

Waveform Definitions

Amplitude Units and Conversion

Creating an InstroAWG Instance

Parameters

Choosing a Driver

Examples

Basic Usage

Background Daemon for Continuous Monitoring

Published channels

Method Reference

Custom Driver Development

Overview

Driver Responsibilities

AWGDriverBase Interface

Talking to the Instrument

Implementation Example: Rigol DG1022Z Driver

Using a Custom Driver

Summary