name: pyvisa-testscript
description: Use when the user wants to set up, run, configure, or extend the pyvisa-testscript instrument control system. Covers starting the live oscilloscope monitor, configuring instrument IP addresses and VISA connections, adding new oscilloscope or instrument drivers, writing PyVISA SCPI commands, and troubleshooting instrument connectivity. Trigger phrases include "pyvisa", "testscript", "oscilloscope monitor", "scope driver", "VISA instrument", "live waveform", "add instrument driver".

PyVISA Test Script — Instrument Control Skill

This skill covers the pyvisa-testscript project: a PyVISA-based instrument control system with a Flask web UI for live oscilloscope monitoring.

Repository: noah-adom-industries/pyvisa-testscript

Project Overview

The system connects to SCPI-capable instruments (oscilloscopes, DAQs) over TCP/IP using PyVISA, captures waveform data in a background thread, and serves a real-time browser UI via Flask + Plotly.

Architecture

main.py                     ← entry point: starts scope thread + Flask server
config.py                   ← IP addresses, VISA address, driver selection
drivers/
    base.py                 ← ScopeDriver abstract base class
    keysight_dsox1204g.py   ← Keysight DSOX1204G oscilloscope driver
    keysight_daq970a.py     ← Keysight DAQ970A data acquisition driver
    __init__.py             ← REGISTRY dict mapping driver names → classes
instruments/
    manager.py              ← InstrumentManager: auto-connect instruments via *IDN?
scope/
    controller.py           ← background capture thread (start/stop/single)
    state.py                ← ScopeState: thread-safe shared state
server/
    __init__.py             ← Flask app factory
    routes.py               ← REST API endpoints + SSE streaming
    templates/index.html    ← live monitoring UI (Plotly waveforms)

Key Concepts

• ScopeDriver (drivers/base.py): Abstract base class. Every scope driver implements setup(), read_settings(), apply_settings(), grab_waveform(), and single_shot().
• REGISTRY (drivers/__init__.py): Maps string names to driver classes. config.DRIVER_NAME selects the active driver.
• ScopeState (scope/state.py): Thread-safe container holding current waveform data, settings, and status. Shared between the capture thread and Flask routes.
• InstrumentManager (instruments/manager.py): Connects to instruments at startup via *IDN? query. Provides send_command() and query() for generic SCPI control.

Step 1: Clone and Install

cd /home/adom/project
git clone https://github.com/noah-adom-industries/pyvisa-testscript.git
cd pyvisa-testscript
pip install pyvisa pyvisa-py flask numpy

If using NI-VISA backend instead of pyvisa-py, install the NI-VISA runtime and use:

pip install pyvisa flask numpy

Step 2: Configure Instruments

Edit /home/adom/project/pyvisa-testscript/config.py:

# Primary oscilloscope
IP_ADDRESS   = "10.0.3.127"                          # scope IP on the network
VISA_ADDRESS = f"TCPIP0::{IP_ADDRESS}::5025::SOCKET" # SCPI socket address

# Driver selection — must match a key in drivers/__init__.py REGISTRY
DRIVER_NAME = "keysight_dsox1204g"

# Additional instruments auto-connected at startup
STARTUP_INSTRUMENTS: list[dict] = [
    {"ip": "10.0.3.83", "port": 5025},   # e.g. Keysight DAQ970A
]

To find instrument IPs, scan the network:

# Scan for instruments on SCPI port 5025
nmap -p 5025 10.0.3.0/24 --open

Step 3: Run the Server

cd /home/adom/project/pyvisa-testscript
python main.py

The server starts on http://0.0.0.0:5000. Open it in a browser to see live waveforms.

Step 4: Adding a New Oscilloscope Driver

To support a new oscilloscope model:

4a. Create the driver file

Create drivers/<manufacturer>_<model>.py. Subclass ScopeDriver and implement all abstract methods:

"""
Driver for <Manufacturer> <Model>.
"""
import struct
import numpy as np
from drivers.base import ScopeDriver


class MyNewScopeDriver(ScopeDriver):

    NAME = "My New Scope"
    N_CHANNELS = 4

    CHANNEL_COLORS = {
        "1": "#FFD700",
        "2": "#00E5FF",
        "3": "#FF00FF",
        "4": "#00FF88",
    }

    TIMEBASE_OPTIONS = [
        (5e-9, "5 ns"), (10e-9, "10 ns"), (20e-9, "20 ns"),
        (50e-9, "50 ns"), (100e-9, "100 ns"), (200e-9, "200 ns"),
        # ... add all supported timebases
    ]

    VSCALE_OPTIONS = [
        (0.001, "1 mV"), (0.002, "2 mV"), (0.005, "5 mV"),
        (0.01, "10 mV"), (0.02, "20 mV"), (0.05, "50 mV"),
        # ... add all supported voltage scales
    ]

    def setup(self, resource) -> None:
        """Configure VISA resource after connection."""
        resource.read_termination = "\n"
        resource.write_termination = "\n"
        resource.timeout = 5000
        resource.write("*RST")
        resource.write("*CLS")
        # Set waveform transfer format (binary, byte order, etc.)

    def read_settings(self, resource) -> dict:
        """Query scope and return settings dict."""
        timebase = float(resource.query(":TIMebase:SCALe?"))
        channels = {}
        for ch in range(1, self.N_CHANNELS + 1):
            scale = float(resource.query(f":CHANnel{ch}:SCALe?"))
            enabled = resource.query(f":CHANnel{ch}:DISPlay?").strip() == "1"
            channels[str(ch)] = {"scale": scale, "enabled": enabled}
        trigger_source = resource.query(":TRIGger:SOURce?").strip()
        trigger_slope = resource.query(":TRIGger:SLOPe?").strip()
        trigger_level = float(resource.query(":TRIGger:LEVel?"))
        return {
            "timebase": timebase,
            "channels": channels,
            "trigger": {
                "source": trigger_source,
                "slope": trigger_slope,
                "level": trigger_level,
            },
        }

    def apply_settings(self, resource, settings: dict) -> None:
        """Push settings dict to scope."""
        if "timebase" in settings:
            resource.write(f":TIMebase:SCALe {settings['timebase']}")
        if "channels" in settings:
            for ch, cfg in settings["channels"].items():
                if "scale" in cfg:
                    resource.write(f":CHANnel{ch}:SCALe {cfg['scale']}")
                if "enabled" in cfg:
                    resource.write(f":CHANnel{ch}:DISPlay {'1' if cfg['enabled'] else '0'}")

    def grab_waveform(self, resource, channels: list[int]) -> dict[str, np.ndarray]:
        """Capture waveform data from the specified channels. Return {ch: array}."""
        result = {}
        for ch in channels:
            resource.write(f":WAVeform:SOURce CHANnel{ch}")
            raw = resource.query_binary_values(":WAVeform:DATA?", datatype="h", is_big_endian=True)
            result[str(ch)] = np.array(raw, dtype=np.float64)
        return result

    def single_shot(self, resource) -> None:
        """Trigger a single acquisition."""
        resource.write(":SINGle")

4b. Register the driver

Edit drivers/__init__.py and add to REGISTRY:

from drivers.my_new_scope import MyNewScopeDriver

REGISTRY = {
    "keysight_dsox1204g": KeysightDSOX1204GDriver,
    "my_new_scope": MyNewScopeDriver,
}

4c. Select the driver

Set DRIVER_NAME = "my_new_scope" in config.py.

Step 5: Adding a New Instrument (non-scope)

For instruments managed by InstrumentManager (DAQs, multimeters, power supplies):

1. Add the instrument to STARTUP_INSTRUMENTS in config.py:

   STARTUP_INSTRUMENTS = [
       {"ip": "10.0.3.83", "port": 5025},
       {"ip": "10.0.3.100", "port": 5025},   # new instrument
   ]
   

2. Optionally create a dedicated driver in drivers/ with helper methods, then query it through the instrument manager's send_command() / query() API.

REST API Endpoints

The Flask server exposes these endpoints (defined in server/routes.py):

Troubleshooting