"""PyVISA driver for the Rigol DHO4804 oscilloscope.
SCPI commands and waveform preamble scaling follow the Rigol DHO1000/DHO4000
Programming Guide, publication PGA34101-1110.
"""
from __future__ import annotations
import math
from dataclasses import dataclass
from enum import StrEnum
from typing import TYPE_CHECKING, ClassVar, Literal
import numpy as np
import pyvisa
if TYPE_CHECKING:
from collections.abc import Mapping
from numpy.typing import NDArray
from pyvisa.resources import MessageBasedResource
type ChannelNumber = Literal[1, 2, 3, 4]
type Coupling = Literal["AC", "DC", "GND"]
type EdgeSlope = Literal["positive", "negative", "either"]
type WordByteOrder = Literal["little", "big"]
[docs]
class DHO4804Measurement(StrEnum):
"""Built-in single-channel measurement items exposed by this driver."""
VPP = "VPP"
VAVG = "VAVG"
VRMS = "VRMS"
VMAX = "VMAX"
VMIN = "VMIN"
FREQUENCY = "FREQuency"
PERIOD = "PERiod"
RISE_TIME = "RTIMe"
FALL_TIME = "FTIMe"
POSITIVE_PULSE_WIDTH = "PWIDth"
NEGATIVE_PULSE_WIDTH = "NWIDth"
POSITIVE_DUTY_CYCLE = "PDUTy"
NEGATIVE_DUTY_CYCLE = "NDUTy"
[docs]
@dataclass(frozen=True)
class EdgeTriggerSettings:
"""Current edge-trigger source, slope, and voltage level."""
source: str
slope: EdgeSlope
level_volts: float
[docs]
class RigolDHO4804Error(RuntimeError):
"""Base exception for DHO4804 driver errors."""
[docs]
class RigolConnectionError(RigolDHO4804Error):
"""Raised when the VISA session cannot be established with a DHO4804."""
[docs]
class RigolNotConnectedError(RigolDHO4804Error):
"""Raised when an instrument operation is attempted while disconnected."""
[docs]
class RigolDHO4804:
"""Control a Rigol DHO4804 through a caller-supplied VISA resource.
Construction does not open hardware. Call :meth:`connect` before issuing
commands and :meth:`disconnect` when finished. VISA addresses for USB, LAN,
or other supported interfaces are passed directly to PyVISA.
"""
_WAVEFORM_PREAMBLE_FIELD_COUNT: ClassVar[int] = 10
_NORMAL_WAVEFORM_POINTS: ClassVar[int] = 1000
_SLOPE_TO_SCPI: ClassVar[Mapping[EdgeSlope, str]] = {
"positive": "POSitive",
"negative": "NEGative",
"either": "RFALl",
}
_SCPI_TO_SLOPE: ClassVar[Mapping[str, EdgeSlope]] = {value.upper(): key for key, value in _SLOPE_TO_SCPI.items()}
def __init__(
self,
resource_name: str,
timeout_ms: int = 10_000,
waveform_word_byte_order: WordByteOrder = "little",
) -> None:
if not resource_name.strip():
raise ValueError("resource_name must be a non-empty VISA resource string.")
if isinstance(timeout_ms, bool) or not isinstance(timeout_ms, int) or timeout_ms <= 0:
raise ValueError("timeout_ms must be a positive integer.")
if waveform_word_byte_order not in {"little", "big"}:
raise ValueError("waveform_word_byte_order must be 'little' or 'big'.")
self.resource_name = resource_name
self.timeout_ms = timeout_ms
self.waveform_word_byte_order = waveform_word_byte_order
self._resource_manager: pyvisa.highlevel.ResourceManager | None = None
self._resource: MessageBasedResource | None = None
self._idn: str | None = None
@property
def connected(self) -> bool:
"""Whether a VISA resource is currently open."""
return self._resource is not None
@property
def idn(self) -> str | None:
"""The identification string returned during the last connection."""
return self._idn
[docs]
def connect(self) -> str:
"""Open the VISA resource, identify the scope, and return its ID string."""
if self._resource is not None:
return self._idn or self._query("*IDN?").strip()
manager: pyvisa.highlevel.ResourceManager | None = None
resource: MessageBasedResource | None = None
try:
manager = pyvisa.ResourceManager()
resource = manager.open_resource(self.resource_name)
resource.timeout = self.timeout_ms
resource.read_termination = "\n"
resource.write_termination = "\n"
idn = resource.query("*IDN?").strip()
if "RIGOL" not in idn.upper() or "DHO4804" not in idn.upper():
raise RigolConnectionError(f"Expected a Rigol DHO4804 at {self.resource_name!r}; instrument replied {idn!r}.")
except Exception as exc:
if resource is not None:
try:
resource.close()
except Exception:
pass
if manager is not None:
try:
manager.close()
except Exception:
pass
if isinstance(exc, RigolDHO4804Error):
raise
raise RigolConnectionError(f"Could not connect to Rigol DHO4804 at {self.resource_name!r}: {exc}") from exc
self._resource_manager = manager
self._resource = resource
self._idn = idn
return idn
[docs]
def disconnect(self) -> None:
"""Close the VISA resource and its resource manager, if open."""
resource = self._resource
manager = self._resource_manager
self._resource = None
self._resource_manager = None
self._idn = None
try:
if resource is not None:
resource.close()
finally:
if manager is not None:
manager.close()
[docs]
def run(self) -> None:
"""Start continuous acquisition."""
self._write(":RUN")
[docs]
def stop(self) -> None:
"""Stop acquisition."""
self._write(":STOP")
[docs]
def single(self) -> None:
"""Arm the scope for a single acquisition."""
self._write(":SINGle")
[docs]
def set_channel_enabled(self, channel: int, enabled: bool) -> None:
"""Enable or disable an analog channel's display and acquisition."""
channel = self._validate_channel(channel)
if not isinstance(enabled, bool):
raise ValueError("enabled must be a bool.")
self._write(f":CHANnel{channel}:DISPlay {'ON' if enabled else 'OFF'}")
[docs]
def get_channel_enabled(self, channel: int) -> bool:
"""Return whether an analog channel is enabled."""
channel = self._validate_channel(channel)
value = self._query(f":CHANnel{channel}:DISPlay?").strip().upper()
if value in {"1", "ON"}:
return True
if value in {"0", "OFF"}:
return False
raise RigolDHO4804Error(f"Unexpected channel display response: {value!r}.")
[docs]
def set_channel_coupling(self, channel: int, coupling: Coupling) -> None:
"""Set a channel's input coupling to AC, DC, or GND."""
channel = self._validate_channel(channel)
if not isinstance(coupling, str):
raise ValueError("coupling must be 'AC', 'DC', or 'GND'.")
normalized = coupling.upper()
if normalized not in {"AC", "DC", "GND"}:
raise ValueError("coupling must be 'AC', 'DC', or 'GND'.")
self._write(f":CHANnel{channel}:COUPling {normalized}")
[docs]
def get_channel_coupling(self, channel: int) -> Coupling:
"""Return a channel's input coupling."""
channel = self._validate_channel(channel)
value = self._query(f":CHANnel{channel}:COUPling?").strip().upper()
if value not in {"AC", "DC", "GND"}:
raise RigolDHO4804Error(f"Unexpected coupling response: {value!r}.")
return value # type: ignore[return-value]
[docs]
def set_vertical_scale(self, channel: int, volts_per_division: float) -> None:
"""Set a channel's vertical scale in volts per division."""
channel = self._validate_channel(channel)
scale = self._validate_finite(volts_per_division, "volts_per_division")
if scale <= 0:
raise ValueError("volts_per_division must be greater than zero.")
self._write(f":CHANnel{channel}:SCALe {scale:.12g}")
[docs]
def get_vertical_scale(self, channel: int) -> float:
"""Return a channel's vertical scale in volts per division."""
channel = self._validate_channel(channel)
return self._query_float(f":CHANnel{channel}:SCALe?")
[docs]
def set_vertical_offset(self, channel: int, offset_volts: float) -> None:
"""Set a channel's vertical offset in volts."""
channel = self._validate_channel(channel)
offset = self._validate_finite(offset_volts, "offset_volts")
self._write(f":CHANnel{channel}:OFFSet {offset:.12g}")
[docs]
def get_vertical_offset(self, channel: int) -> float:
"""Return a channel's vertical offset in volts."""
channel = self._validate_channel(channel)
return self._query_float(f":CHANnel{channel}:OFFSet?")
[docs]
def set_timebase_scale(self, seconds_per_division: float) -> None:
"""Set the main horizontal timebase in seconds per division."""
scale = self._validate_finite(seconds_per_division, "seconds_per_division")
if scale <= 0:
raise ValueError("seconds_per_division must be greater than zero.")
self._write(f":TIMebase:MAIN:SCALe {scale:.12g}")
[docs]
def get_timebase_scale(self) -> float:
"""Return the main horizontal timebase in seconds per division."""
return self._query_float(":TIMebase:MAIN:SCALe?")
[docs]
def get_edge_trigger(self) -> EdgeTriggerSettings:
"""Return the active edge-trigger source, slope, and level."""
source = self._query(":TRIGger:EDGE:SOURce?").strip().upper()
slope_response = self._query(":TRIGger:EDGE:SLOPe?").strip().upper()
try:
slope = self._SCPI_TO_SLOPE[slope_response]
except KeyError as exc:
raise RigolDHO4804Error(f"Unexpected edge-trigger slope response: {slope_response!r}.") from exc
return EdgeTriggerSettings(
source=source,
slope=slope,
level_volts=self._query_float(":TRIGger:EDGE:LEVel?"),
)
[docs]
def get_trigger_status(self) -> str:
"""Return the scope trigger state reported by ``:TRIGger:STATus?``."""
status = self._query(":TRIGger:STATus?").strip().upper()
if status not in {"TD", "WAIT", "RUN", "AUTO", "STOP"}:
raise RigolDHO4804Error(f"Unexpected trigger status response: {status!r}.")
return status
[docs]
def measure(
self,
channel: int,
measurement: DHO4804Measurement,
) -> float:
"""Query a built-in measurement for one channel.
Voltage measurements return volts, frequency returns hertz, period and
timing measurements return seconds, and duty-cycle measurements use the
value returned by the instrument.
"""
channel = self._validate_channel(channel)
try:
item = DHO4804Measurement(measurement)
except ValueError as exc:
allowed = ", ".join(item.name for item in DHO4804Measurement)
raise ValueError(f"Unsupported measurement. Choose one of: {allowed}.") from exc
return self._query_float(f":MEASure:ITEM? {item.value},CHANnel{channel}")
@staticmethod
def _validate_channel(channel: int) -> ChannelNumber:
if isinstance(channel, bool) or not isinstance(channel, int) or channel not in {1, 2, 3, 4}:
raise ValueError("channel must be one of 1, 2, 3, or 4.")
return channel # type: ignore[return-value]
@staticmethod
def _validate_finite(value: float, name: str) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)):
raise ValueError(f"{name} must be a finite number.")
numeric_value = float(value)
if not math.isfinite(numeric_value):
raise ValueError(f"{name} must be a finite number.")
return numeric_value
@staticmethod
def _parse_waveform_preamble(response: str) -> tuple[int | float, ...]:
fields = response.strip().split(",")
if len(fields) != RigolDHO4804._WAVEFORM_PREAMBLE_FIELD_COUNT:
raise RigolDHO4804Error(f"Expected {RigolDHO4804._WAVEFORM_PREAMBLE_FIELD_COUNT} waveform preamble fields, received {len(fields)}.")
try:
return (
int(fields[0]),
int(fields[1]),
int(fields[2]),
int(fields[3]),
*(float(field) for field in fields[4:]),
)
except ValueError as exc:
raise RigolDHO4804Error(f"Could not parse waveform preamble: {response!r}.") from exc
def _query_float(self, command: str) -> float:
response = self._query(command).strip()
try:
return float(response)
except ValueError as exc:
raise RigolDHO4804Error(f"Expected a numeric response to {command!r}, got {response!r}.") from exc
def _session(self) -> MessageBasedResource:
if self._resource is None:
raise RigolNotConnectedError("Not connected. Call connect() before using the oscilloscope.")
return self._resource
def _write(self, command: str) -> None:
self._session().write(command)
def _query(self, command: str) -> str:
return self._session().query(command)