124 lines
5.6 KiB
Python
124 lines
5.6 KiB
Python
"""DPO4034 SCPI acquisition over VISA (GPIB or Ethernet), without Qt.
|
|
|
|
See docs/tektronix-dpo4034.md for wiring, commands and scaling.
|
|
"""
|
|
from dataclasses import dataclass
|
|
import math
|
|
|
|
|
|
def resource_name(interface, address, board=0):
|
|
address = str(address).strip()
|
|
if interface == "GPIB":
|
|
if not address.isdigit() or not 0 <= int(address) <= 30:
|
|
raise ValueError("Адрес GPIB должен быть в диапазоне 0…30")
|
|
if not 0 <= int(board) <= 99:
|
|
raise ValueError("Неверный номер контроллера GPIB")
|
|
return f"GPIB{int(board)}::{int(address)}::INSTR"
|
|
if interface != "Ethernet":
|
|
raise ValueError("Неизвестный интерфейс")
|
|
if not address or any(c.isspace() for c in address) or any(c in address for c in ";\r\n"):
|
|
raise ValueError("Укажите IP/имя прибора или TCPIP VISA resource")
|
|
if "::" in address:
|
|
if not address.upper().startswith("TCPIP") or not address.upper().endswith("::INSTR"):
|
|
raise ValueError("Ожидается TCPIP…::INSTR (VXI-11)")
|
|
return address
|
|
return f"TCPIP0::{address}::inst0::INSTR"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Waveform:
|
|
channel: str
|
|
seconds: tuple
|
|
values: tuple
|
|
unit: str
|
|
|
|
|
|
class DPO4034:
|
|
"""Single-thread-owned session. Inject a ResourceManager for offline tests."""
|
|
|
|
def __init__(self, manager_factory=None):
|
|
self._factory = manager_factory
|
|
self.manager = None
|
|
self.instrument = None
|
|
|
|
def connect(self, resource, timeout_ms=5000):
|
|
self.close()
|
|
factory = self._factory
|
|
if factory is None:
|
|
try:
|
|
import pyvisa
|
|
except ImportError as exc:
|
|
raise RuntimeError("Установите PyVISA и NI-VISA или TekVISA") from exc
|
|
factory = pyvisa.ResourceManager
|
|
try:
|
|
self.manager = factory()
|
|
self.instrument = self.manager.open_resource(resource, open_timeout=timeout_ms)
|
|
self.instrument.timeout = timeout_ms
|
|
self.instrument.write_termination = "\n"
|
|
self.instrument.read_termination = "\n"
|
|
identity = self.instrument.query("*IDN?").strip()
|
|
fields = identity.upper().split(",")
|
|
if len(fields) < 2 or "TEKTRONIX" not in fields[0] or fields[1].strip() != "DPO4034":
|
|
raise ValueError(f"Ожидался Tektronix DPO4034, получено: {identity}")
|
|
self.instrument.write("HEADER OFF")
|
|
return identity
|
|
except Exception:
|
|
self.close()
|
|
raise
|
|
|
|
def close(self):
|
|
instrument, manager = self.instrument, self.manager
|
|
self.instrument = self.manager = None
|
|
try:
|
|
if instrument is not None:
|
|
instrument.close()
|
|
finally:
|
|
if manager is not None:
|
|
manager.close()
|
|
|
|
def capture(self, channels, points=100000):
|
|
channels = tuple(channels)
|
|
if not channels or len(set(channels)) != len(channels) or any(c not in ("CH1", "CH2", "CH3", "CH4") for c in channels):
|
|
raise ValueError("Выберите каналы CH1…CH4")
|
|
if not 1 <= points <= 1000000:
|
|
raise ValueError("Число точек должно быть 1…1000000")
|
|
scope = self.instrument
|
|
if scope is None:
|
|
raise RuntimeError("Прибор не подключён")
|
|
# Freeze the record so all selected channels belong to the same capture.
|
|
running = int(scope.query("ACQUIRE:STATE?").strip())
|
|
result = []
|
|
try:
|
|
scope.write("ACQUIRE:STATE STOP")
|
|
scope.query("*OPC?")
|
|
for channel in channels:
|
|
if not int(scope.query(f"SELECT:{channel}?").strip()):
|
|
raise ValueError(f"Включите {channel} на осциллографе")
|
|
scope.write(f"DATA:SOURCE {channel}")
|
|
scope.write("DATA:ENCDG RIBINARY")
|
|
scope.write("DATA:WIDTH 2")
|
|
scope.write("DATA:START 1")
|
|
scope.write(f"DATA:STOP {points}")
|
|
if scope.query("WFMOUTPRE:PT_FMT?").strip().upper() != "Y":
|
|
raise ValueError("ENV/Peak Detect не поддерживается; выберите Sample или Average")
|
|
params = [float(scope.query(f"WFMOUTPRE:{key}?").strip()) for key in
|
|
("XINCR", "XZERO", "PT_OFF", "YMULT", "YOFF", "YZERO")]
|
|
dx, x0, offset, mult, yoff, y0 = params
|
|
if not all(math.isfinite(v) for v in params) or dx <= 0:
|
|
raise ValueError("Некорректная преамбула осциллограммы")
|
|
unit = scope.query("WFMOUTPRE:YUNIT?").strip().strip('"')
|
|
count = int(scope.query("WFMOUTPRE:NR_PT?").strip())
|
|
if not 1 <= count <= points:
|
|
raise ValueError("Некорректная длина осциллограммы")
|
|
raw = scope.query_binary_values("CURVE?", datatype="h", is_big_endian=True,
|
|
container=list)
|
|
if len(raw) != count:
|
|
raise ValueError(f"Неполная осциллограмма: {len(raw)} из {count}")
|
|
result.append(Waveform(channel,
|
|
tuple(x0 + (i - offset) * dx for i in range(count)),
|
|
tuple(y0 + (v - yoff) * mult for v in raw), unit))
|
|
finally:
|
|
if running:
|
|
scope.write("ACQUIRE:STATE RUN")
|
|
return result
|