InstroScope

InstroScope is a hardware abstraction layer (HAL) that provides a unified interface for SCPI-based oscilloscopes. The category class defines the vendor-independent API (set_vertical_scale, fetch_waveform, measure, set_trigger_level, …). A vendor-specific driver translates those calls into vendor commands.

Supported Vendors

Keysight: InfiniiVision 1200 X-Series via SCPI/VISA (Keysight1200X)
Siglent: SDS1000X-E series via SCPI/VISA (SiglentSDS1000XE)
Tektronix: 2 Series MSO via SCPI/VISA (Tektronix2SeriesMSO)

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

Key Concepts

Driver Composition

An InstroScope is built from a concrete driver:

InstroScope("name", driver=Keysight1200X("<visa_resource>"), num_channels=4)

Keysight1200X owns the connection setup and vendor-specific command mapping.
InstroScope owns the category-level workflow: configuration tracking, acquisition control, measurements, waveform fetches, publishers, the background daemon.

Lifecycle

Construct: instantiate the vendor driver and pass it to InstroScope.
open(): establish the VISA connection.
Configure: set vertical (per channel), horizontal, acquisition, and trigger settings.
Acquire: arm a single-shot with single(), or free-run with run().
Read: fetch a waveform with fetch_waveform() or take a built-in measurement with measure().
close(): disconnect from hardware.

Acquisition Control vs. the Background Daemon

InstroScope has two independent start/stop concepts:

Acquisition control drives the instrument: run(), single(), and stop_acquisition(). These map to the front-panel Run, Single, and Stop keys. After single(), both fetch_waveform() and measure() block up to their timeout for the trigger to fire, then read. They raise TimeoutError if it does not.
The background daemon drives polling on the host: start() and stop() (inherited from Instrument) periodically call the methods you register and stream their results to publishers.

Configuration Tracking

InstroScope tracks vertical, horizontal, acquisition, and trigger state locally as you issue commands. Call sync_configuration() after open() (or after load_settings()) to bulk-query the instrument and overwrite the tracked state.

sync_configuration() does not bulk-query trigger source, type, level, slope, or mode. Not all scopes expose those through a simple query, so set them explicitly to populate the tracked state.

Measurements

measure(measurement_type, channel) reads a built-in measurement: VPP, VMAX, VMIN, VAVG, VRMS, FREQUENCY, PERIOD, or DUTY_CYCLE. It returns NaN when the scope has no valid result (no acquisition yet, channel off). fetch_waveform(channel) returns a Measurement whose timestamps are relative to the trigger point (negative values are pre-trigger), with voltages already scaled through the configured probe attenuation.

Creating an InstroScope Instance

from instro.scope.drivers import Keysight1200X
from instro.scope import InstroScope

scope = InstroScope(
    name="myScope",
    driver=Keysight1200X("<visa_resource>"),
    num_channels=4,
)

Parameters

name: A name for this scope instance. Used as a prefix for channel names when publishing.
driver: A concrete ScopeDriverBase instance configured with the connection details for that model.
num_channels: Number of analog-input channels on the oscilloscope.
publishers: Optional list of publishers to attach.
**kwargs: Additional keyword arguments become default tags when using a publisher that supports tags (like NominalCorePublisher). Pass legacy_naming=True here to publish pre-v1.0 channel names.

Choosing a Driver

Choose the driver that matches the model, then pass it the connection settings. Use Keysight1200X, SiglentSDS1000XE, or Tektronix2SeriesMSO. Each accepts a VISA resource string or a VisaConfig. To inspect a VISA instrument’s identity before choosing a driver:

from instro.lib.transports import VisaDriver

visa = VisaDriver("<visa_resource>")
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

from instro.lib.publishers import NominalCorePublisher
from instro.scope import (
    AcquisitionMode,
    Coupling,
    InstroScope,
    ScopeMeasurementType,
    TriggerMode,
    TriggerSlope,
    TriggerType,
)
from instro.scope.drivers import Keysight1200X

VISA_RESOURCE = "<visa_resource>"
DATASET_RID = "<your dataset here>"

scope = InstroScope(
    name="myScope",
    driver=Keysight1200X(VISA_RESOURCE),
    num_channels=4,
)
scope.add_publisher(NominalCorePublisher(dataset_rid=DATASET_RID))

scope.open()

# Configure channel 1
scope.set_coupling(Coupling.DC, channel=1)
scope.set_probe_attenuation(1.0, channel=1)
scope.set_vertical_scale(1.0, channel=1)  # 1 V/div
scope.set_vertical_offset(0.0, channel=1)

# Configure timebase and acquisition
scope.set_horizontal_scale(0.01)  # 10 ms/div
scope.set_acquisition_mode(AcquisitionMode.NORMAL)

# Configure the trigger
scope.set_trigger_source(channel=1)
scope.set_trigger_type(TriggerType.EDGE)
scope.set_trigger_slope(TriggerSlope.RISING)
scope.set_trigger_level(0.0)
scope.set_trigger_mode(TriggerMode.NORMAL)

# Arm a single-shot, then fetch and measure (both block for the trigger)
scope.single()
waveform = scope.fetch_waveform(channel=1)
vpp = scope.measure(ScopeMeasurementType.VPP, channel=1)
freq = scope.measure(ScopeMeasurementType.FREQUENCY, channel=1)
print(f"Vpp: {vpp.latest}, Frequency: {freq.latest}")

scope.save_screenshot("capture.png")  # Saves to the host
scope.close()

Important Note about PublishersData is published as a direct result of an instrument method being called. Calling measure() reads the value from the instrument and causes all attached publishers to publish it automatically.

Background Daemon for Continuous Monitoring

from instro.lib.publishers import NominalCorePublisher
from instro.scope import AcquisitionMode, Coupling, InstroScope, ScopeMeasurementType
from instro.scope import TriggerMode, TriggerSlope, TriggerType
from instro.scope.drivers import Keysight1200X

scope = InstroScope(
    name="myScope",
    driver=Keysight1200X("<visa_resource>"),
    num_channels=2,
)
scope.add_publisher(NominalCorePublisher(dataset_rid="<your dataset here>"))
scope.background_interval = 0.5  # Poll every 500ms (default is 1 second)

scope.open()

# Configure channel 1, timebase, acquisition, and trigger
scope.set_coupling(Coupling.DC, channel=1)
scope.set_vertical_scale(1.0, channel=1)  # 1 V/div
scope.set_horizontal_scale(0.01)  # 10 ms/div
scope.set_acquisition_mode(AcquisitionMode.NORMAL)
scope.set_trigger_source(channel=1)
scope.set_trigger_type(TriggerType.EDGE)
scope.set_trigger_slope(TriggerSlope.RISING)
scope.set_trigger_level(0.0)
scope.set_trigger_mode(TriggerMode.NORMAL)

# Register the measurements the daemon should poll
scope.add_background_daemon_function(scope.measure, ScopeMeasurementType.VRMS, channel=1)
scope.add_background_daemon_function(scope.measure, ScopeMeasurementType.VPP, channel=1)

# Free-run the instrument, then start the daemon
scope.run()
scope.start()

while True:
    try:
        vrms = scope.get_channel("myScope.ch1.vrms", length=1, wait_for_new_samples=True)
        vpp = scope.get_channel("myScope.ch1.vpp", length=1)
        print(f"Vrms: {vrms.latest}, Vpp: {vpp.latest}")
    except KeyboardInterrupt:
        break

scope.stop()  # Stop the daemon
scope.stop_acquisition()  # Stop the instrument
scope.close()

Scope Background DaemonInstroScope registers no default daemon functions. Register the measurements with add_background_daemon_function() before start(), or use define_background_daemon() to replace the registered set. The default polling interval is 1 second, configurable via background_interval.

See Two ways to get data for more on background fetching.

Published channels

Every call produces a channel keyed under {name}.{descriptor}, where {name} is the constructor argument and {descriptor} is the row below. Substitute {N} with the channel number and {type} with the lowercased measurement type (e.g. vpp). Pass legacy_naming=True to the constructor for the pre-v1.0 names.

Method	Descriptor	Type
`set_vertical_scale(channel=N)`	`ch{N}.vscale.cmd`	command
`get_vertical_scale(channel=N)`	`ch{N}.vscale`	telemetry
`set_vertical_offset(channel=N)`	`ch{N}.voffset.cmd`	command
`get_vertical_offset(channel=N)`	`ch{N}.voffset`	telemetry
`set_coupling(channel=N)`	`ch{N}.coupling.cmd`	command
`get_coupling(channel=N)`	`ch{N}.coupling.cmd`	command
`set_probe_attenuation(channel=N)`	`ch{N}.probe_attenuation.cmd`	command
`get_probe_attenuation(channel=N)`	`ch{N}.probe_attenuation`	telemetry
`set_horizontal_scale()`	`hscale.cmd`	command
`get_horizontal_scale()`	`hscale`	telemetry
`get_sample_rate()`	`sample_rate`	telemetry
`set_acquisition_mode()`	`acquisition_mode.cmd`	command
`get_acquisition_mode()`	`acquisition_mode.cmd`	command
`set_average_count()`	`average_count.cmd`	command
`get_average_count()`	`average_count`	telemetry
`run()` / `stop_acquisition()` / `single()`	`acquisition_control.cmd`	command
`get_acquisition_state()`	`acquisition_state.cmd`	command
`fetch_waveform(channel=N)`	`ch{N}.waveform`	telemetry
`measure(type, channel=N)`	`ch{N}.{type}` (e.g. `ch{N}.vpp`)	telemetry
`set_trigger_source()`	`trigger_source.cmd`	command
`set_trigger_type()`	`trigger_type.cmd`	command
`set_trigger_level()`	`trigger_level.cmd`	command
`set_trigger_slope()`	`trigger_slope.cmd`	command
`set_trigger_mode()`	`trigger_mode.cmd`	command
`force_trigger()`	`trigger_control.cmd`	command
`get_trigger_status()`	`trigger_status.cmd`	command
`save_screenshot()`	`screenshot.cmd`	command
`save_settings()`	`save_settings.cmd`	command
`load_settings()`	`load_settings.cmd`	command

Method Reference

Method	Purpose
`InstroScope(name, driver, num_channels, publishers=None, **kwargs)`	Construct an InstroScope with a vendor driver
`open()`	Establish the VISA connection to the scope
`close()`	Disconnect from scope and close all publishers
`sync_configuration()`	Bulk-query the instrument and overwrite the tracked configuration
`set_vertical_scale(volts_per_div, channel=N)`	Set the vertical scale in volts/div
`get_vertical_scale(channel=N)`	Read the vertical scale in volts/div
`set_vertical_offset(offset, channel=N)`	Set the vertical offset in volts
`get_vertical_offset(channel=N)`	Read the vertical offset in volts
`set_coupling(coupling, channel=N)`	Set AC/DC input coupling
`get_coupling(channel=N)`	Read the input coupling
`set_probe_attenuation(factor, channel=N)`	Set the probe attenuation ratio (e.g. 1, 10, 100)
`get_probe_attenuation(channel=N)`	Read the probe attenuation ratio
`set_horizontal_scale(seconds_per_div)`	Set the timebase in seconds/div (applies to all channels)
`get_horizontal_scale()`	Read the timebase in seconds/div
`get_sample_rate()`	Read the effective sample rate in samples/second
`set_acquisition_mode(mode)`	Set the acquisition mode (NORMAL, AVERAGE, HIGH_RESOLUTION, PEAK_DETECT, ENVELOPE)
`get_acquisition_mode()`	Read the acquisition mode
`set_average_count(count)`	Set the waveforms-to-average count (used in AVERAGE mode)
`get_average_count()`	Read the waveforms-to-average count
`run()`	Start continuous (free-running) acquisition
`stop_acquisition()`	Stop acquisition, leaving captured data intact
`single()`	Arm a single-shot acquisition
`get_acquisition_state()`	Read the acquisition run state (RUNNING / STOPPED)
`fetch_waveform(channel=N, timeout=5.0)`	Fetch the most recently acquired waveform
`measure(measurement_type, channel=1, timeout=5.0)`	Take a built-in measurement
`set_trigger_source(channel)`	Set the trigger source channel
`set_trigger_type(trigger_type)`	Set the trigger type (EDGE, PULSE)
`set_trigger_level(level)`	Set the trigger threshold in volts
`set_trigger_slope(slope)`	Set the trigger edge slope (RISING, FALLING, EITHER)
`set_trigger_mode(mode)`	Set the trigger sweep mode (AUTO, NORMAL)
`force_trigger()`	Force a trigger event immediately
`get_trigger_status()`	Read the trigger status (ARMED, READY, TRIGGERED, …)
`save_screenshot(filepath, to_instrument=False)`	Capture a screenshot to the host or the scope’s filesystem
`save_settings(name, to_instrument=False)`	Save the current scope setup
`load_settings(name, from_instrument=False)`	Recall a scope setup (invalidates tracked configuration)
`start()`	Begin the background telemetry daemon
`stop()`	End the background telemetry daemon

Custom Driver Development

This section is for developers implementing InstroScope support for oscilloscopes that aren’t supported out of the box. A scope driver subclasses ScopeDriverBase and owns its transport. Its responsibilities:

Expose a protocol-native constructor: accept a visa_resource string or VisaConfig.
Own transport setup: create and store the transport internally. Do not require callers to pass a VisaDriver.
Own lifecycle: implement open() and close().
Map commands: translate each abstract method into vendor commands.
Parse responses: convert responses to the expected types (float, enums, WaveformData).
Drain the error queue: implement check_errors(). InstroScope calls it between setup commands and blocking queries, so a pending syntax error raises instead of hanging a data query.

Channels are 1-indexed analog input numbers throughout.

ScopeDriverBase Interface

All scope drivers subclass ScopeDriverBase and implement these abstract methods:

def open(self) -> None: ...
def close(self) -> None: ...
def check_errors(self) -> None: ...

# Channel vertical settings
def set_vertical_scale(self, volts_per_div: float, channel: int) -> None: ...
def get_vertical_scale(self, channel: int) -> float: ...
def set_vertical_offset(self, offset: float, channel: int) -> None: ...
def get_vertical_offset(self, channel: int) -> float: ...
def set_coupling(self, coupling: Coupling, channel: int) -> None: ...
def get_coupling(self, channel: int) -> Coupling: ...
def set_probe_attenuation(self, factor: float, channel: int) -> None: ...
def get_probe_attenuation(self, channel: int) -> float: ...

# Horizontal (timebase) and sample rate
def set_horizontal_scale(self, seconds_per_div: float) -> None: ...
def get_horizontal_scale(self) -> float: ...
def get_sample_rate(self) -> float: ...

# Acquisition
def set_acquisition_mode(self, mode: AcquisitionMode) -> None: ...
def get_acquisition_mode(self) -> AcquisitionMode: ...
def set_average_count(self, count: int) -> None: ...
def get_average_count(self) -> int: ...
def run(self) -> None: ...
def stop(self) -> None: ...
def single(self) -> None: ...
def digitize(self, timeout: float) -> None: ...
def get_acquisition_state(self) -> AcquisitionState: ...

# Waveform data and measurements
def fetch_waveform(self, channel: int) -> WaveformData: ...
def measure(self, measurement_type: ScopeMeasurementType, channel: int) -> float: ...

# Trigger
def set_trigger_source(self, channel: int) -> None: ...
def set_trigger_type(self, trigger_type: TriggerType) -> None: ...
def set_trigger_level(self, level: float) -> None: ...
def set_trigger_slope(self, slope: TriggerSlope) -> None: ...
def set_trigger_mode(self, mode: TriggerMode) -> None: ...
def force_trigger(self) -> None: ...
def get_trigger_status(self) -> TriggerStatus: ...

# File operations
def save_screenshot(self, filepath: str, to_instrument: bool = False) -> bytes: ...
def save_settings(self, name: str, to_instrument: bool = False) -> bytes: ...
def load_settings(self, name: str, from_instrument: bool = False) -> None: ...

setup_measurement(measurement_type, channel) is an optional override (default no-op). Implement it for instruments that compute measurements during acquisition (e.g. Tektronix), where the measurement slot must exist at trigger time. Raise NotImplementedError for AcquisitionMode values the scope does not support (e.g. the Keysight 1200X has no ENVELOPE mode). See Exceptions for unsupported-feature handling. For VISA-backed drivers, create a VisaDriver internally and use self._visa.write(command) and self._visa.query(command) for all I/O. VisaDriver owns the resource lock and serializes concurrent calls. See the VisaDriver guide for the full transport reference.

Implementation Example: Keysight 1200 X-Series Driver

from __future__ import annotations

import math

from instro.lib.transports.visa import VisaConfig, VisaDriver
from instro.scope import ScopeDriverBase

# IEEE-488.2 sentinel returned when a measurement has no valid result.
_VENDOR_INVALID_MEASUREMENT = 9.91e37


class Keysight1200X(ScopeDriverBase):
    """SCPI driver for Keysight InfiniiVision 1200 X-Series oscilloscopes."""

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

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

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

    def check_errors(self) -> None:
        while True:
            resp = self._visa.query(":SYSTem:ERRor?")
            parts = resp.split(",", 1)
            code = int(parts[0])
            if code == 0:
                return
            msg = parts[1].strip().strip('"') if len(parts) > 1 else "Unknown error"
            raise RuntimeError(f"Keysight SCPI error {code}: {msg}")

    def set_vertical_scale(self, volts_per_div: float, channel: int) -> None:
        self._visa.write(f":CHANnel{channel}:SCALe {volts_per_div}")

    def get_vertical_scale(self, channel: int) -> float:
        return float(self._visa.query(f":CHANnel{channel}:SCALe?"))

    # ...remaining channel, timebase, acquisition, trigger, and file methods...

Vendor SCPI VariationsScope vendors differ in SCPI command sets, error-query forms, and waveform-preamble formats. The Keysight 1200X computes measurements on demand, while the Tektronix 2 Series requires measurement slots installed before the trigger fires (setup_measurement). Always consult your oscilloscope’s programming manual.

Using a Custom Driver

Construct InstroScope with your own driver instance:

from instro.scope import InstroScope, ScopeDriverBase
from instro.lib.transports import VisaDriver


class MyCustomScopeDriver(ScopeDriverBase):
    """Custom driver for my lab's proprietary oscilloscope."""

    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("SYST:ERR?")
        if not err.startswith("0"):
            raise RuntimeError(f"Scope error: {err}")

    def set_vertical_scale(self, volts_per_div: float, channel: int) -> None:
        self._visa.write(f"CH{channel}:SCALE {volts_per_div}")

    # ... implement the other abstract methods ...


scope = InstroScope(
    name="labScope",
    driver=MyCustomScopeDriver(visa_resource="<visa_resource>"),
    num_channels=4,
)
scope.open()
scope.set_vertical_scale(1.0, channel=1)
scope.close()

​InstroScope

​Supported Vendors

​Key Concepts

​Driver Composition

​Lifecycle

​Acquisition Control vs. the Background Daemon

​Configuration Tracking

​Measurements

​Creating an InstroScope Instance

​Parameters

​Choosing a Driver

​Examples

​Basic Usage

​Background Daemon for Continuous Monitoring

​Published channels

​Method Reference

​Custom Driver Development

​ScopeDriverBase Interface

​Implementation Example: Keysight 1200 X-Series Driver

​Using a Custom Driver

InstroScope

Supported Vendors

Key Concepts

Driver Composition

Lifecycle

Acquisition Control vs. the Background Daemon

Configuration Tracking

Measurements

Creating an InstroScope Instance

Parameters

Choosing a Driver

Examples

Basic Usage

Background Daemon for Continuous Monitoring

Published channels

Method Reference

Custom Driver Development

ScopeDriverBase Interface

Implementation Example: Keysight 1200 X-Series Driver

Using a Custom Driver