> ## Documentation Index
> Fetch the complete documentation index at: https://nominal-instro-524-sw-timed-background-daemon.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Arbitrary Waveform Generator (AWG)

> Using InstroAWG for SCPI-based arbitrary waveform generators

# InstroAWG

<Warning>
  **Unstable API**

  `InstroAWG` 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](/instrumentation/installation#unstable-modules) for details.
</Warning>

`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](#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:

1. **Construct**: instantiate the vendor driver and pass it to `InstroAWG`, along with the channel count.
2. **`open()`**: establishes the VISA connection.
3. **Configure and generate**: define a waveform on a channel, set its amplitude and offset, and enable the output.
4. **`start()`**: begins a periodic background daemon that polls output state. (Optional)
5. **`stop()`**: ends the background daemon (if started).
6. **`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.

<Note>
  **Driver support varies**

  Not 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.
</Note>

### 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:

```python theme={null}
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

```python theme={null}
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:

```python theme={null}
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`](/instrumentation/library#measurement) objects. This is common amongst all `Instrument` objects.

### Basic Usage

```python theme={null}
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

```python theme={null}
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.

<Note>
  **Default AWG Background Daemon**

  For each <Tooltip tip="A named signal for a series of measurements or computed values (example: voltage, pressure, system state).">channel</Tooltip>:

  * 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.
</Note>

**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](/instrumentation/overview#two-ways-to-get-data) for more information regarding background fetching of measurements.

<Note>
  **Important Note about Publishers**

  Data 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!
</Note>

## 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:

```python theme={null}
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:

1. **Expose a protocol-native constructor**: accept inputs like `visa_resource`, `host`, `port`, `unit_id`, `interface`, or `node_id`, depending on the instrument.
2. **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.
3. **Own lifecycle**: implement `open()` and `close()` by opening and closing the underlying transport.
4. **Map commands**: translate each abstract method into vendor-specific commands, raising `ValueError` for a `Waveform` definition the instrument can't produce.
5. **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:

```python theme={null}
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:

```python theme={null}
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](/instrumentation/transports/visa) 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:

```python theme={null}
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 ...
```

<Tip>
  **Vendor SCPI Variations**

  Different 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.
</Tip>

<Warning>
  **USB arbitrary downloads on some Rigol DG1000Z units**

  Some 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.
</Warning>

## 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:

```python theme={null}
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