--- 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 ```bash 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: ```bash pip install pyvisa flask numpy ``` ## Step 2: Configure Instruments Edit `/home/adom/project/pyvisa-testscript/config.py`: ```python # 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: ```bash # Scan for instruments on SCPI port 5025 nmap -p 5025 10.0.3.0/24 --open ``` ## Step 3: Run the Server ```bash 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/_.py`. Subclass `ScopeDriver` and implement all abstract methods: ```python """ Driver for . """ 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`: ```python 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`: ```python 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`): | Endpoint | Method | Description | |----------|--------|-------------| | `/` | GET | Live monitoring UI | | `/api/waveform` | GET | Current waveform data (JSON) | | `/api/settings` | GET | Current scope settings | | `/api/settings` | POST | Apply new scope settings | | `/api/start` | POST | Start continuous capture | | `/api/stop` | POST | Stop capture | | `/api/single` | POST | Single-shot acquisition | | `/api/stream` | GET | SSE stream of waveform updates | | `/api/instruments` | GET | List connected instruments | | `/api/instruments//command` | POST | Send SCPI command to an instrument | | `/api/instruments//query` | POST | Query an instrument (returns response) | ## Troubleshooting | Symptom | Cause | Fix | |---------|-------|-----| | `VI_ERROR_RSRC_NFOUND` or connection timeout | Instrument not reachable on the network | Verify IP with `ping `. Check that port 5025 is open: `nc -zv 5025`. Ensure instrument and host are on the same subnet. | | `No driver found for ...` | `DRIVER_NAME` in `config.py` doesn't match any key in `REGISTRY` | Check `drivers/__init__.py` — the name must match exactly. | | Waveforms display but are flat/zero | Waveform format mismatch (byte order, data type) | Check `grab_waveform()` — ensure `is_big_endian` and `datatype` match the scope's binary format. Send `:WAVeform:FORMat?` to verify. | | Flask starts but UI shows "Disconnected" | SSE stream not connecting, or scope thread crashed | Check terminal for Python exceptions. Verify the scope IP is correct and the instrument is powered on. | | `pyvisa-py` can't find instruments | Missing transport backend | Install `psutil` and `zeroconf`: `pip install psutil zeroconf`. For USB instruments, install `pyusb`. | | `*IDN?` query hangs or times out | Instrument uses different termination characters | Check the instrument manual. Some need `\r\n` instead of `\n`. Adjust in the driver's `setup()` method. | | Multiple users see stale data | Flask runs single-threaded by default | `main.py` already sets `threaded=True`. For production use, run behind gunicorn: `gunicorn -w 1 --threads 4 -b 0.0.0.0:5000 server:app`. |