> ## 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 **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. `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](#custom-driver-development) below. ## Key Concepts ### Driver Composition An `InstroAWG` is built from a concrete driver: ```python theme={null} 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: 1. **Construct**: instantiate the vendor driver and pass it to `InstroAWG`, along with the channel count. 2. **`open()`**: establishes the connection to the instrument. 3. **Configure and generate**: define a waveform on a channel, set its amplitude and offset, and enable the output, etc. 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 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 varies** Certain 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 ```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 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: ```python theme={null} from instro.lib.transports import VisaDriver visa = VisaDriver("USB0::...") try: visa.open() print(visa.query("*IDN?")) finally: visa.close() ``` ## Examples ### Basic Usage ```python theme={null} 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 = "" 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 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. ## 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 | | `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_waveform()`, `get_amplitude()`, `convert_amplitude()`, and `get_modulation_type()` return a plain Python value (`Waveform`, `tuple[float, AmplitudeMeasurementUnit]`, `float`, `ModulationType`) 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 | | `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: ```python theme={null} 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: 1. **Expose a protocol-native constructor**: accept inputs like `visa_resource`, `host`, `port`, depending on the instrument. 2. **Own transport setup**: create and store the transport internally. Do not require users to pass a `VisaDriver` 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. 5. **Parse responses**: convert instrument responses to the expected Python types (`Waveform`, `float`, `bool`, `ModulationType`, etc.). 6. **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`: ```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.""" ``` 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. ### 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](/instrumentation/transports/visa) 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: ```python theme={null} 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 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()}") 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__}") 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}") # ... other required and optional methods ... ``` ## 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 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 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}:WAVE SIN,{waveform.frequency_hz}") 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" awg = InstroAWG( name="labAWG", driver=MyCustomAWGDriver(visa_resource=""), 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