269 lines
14 KiB
Python
269 lines
14 KiB
Python
# Получение осциллограмм Tektronix DPO4034 через SCPI/VISA. Параметры масштаба из прибора
|
||
# переводят сырые отсчёты в физические значения; выбор VISA-ресурса отделён от графического
|
||
# интерфейса.
|
||
|
||
"""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
|
||
import struct
|
||
import sys
|
||
|
||
|
||
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=15000):
|
||
self.close()
|
||
factory = self._factory
|
||
if factory is None:
|
||
try:
|
||
import pyvisa
|
||
except ImportError as exc:
|
||
if getattr(sys, "frozen", False):
|
||
message = "В сборке SETGUI недоступна PyVISA. Используйте исправленную сборку EXE."
|
||
else:
|
||
message = "В Python-окружении недоступна PyVISA. Установите PyVISA (для Python 3.8: версия 1.14.1)."
|
||
raise RuntimeError(message + " Причина: " + str(exc)) from exc
|
||
factory = pyvisa.ResourceManager
|
||
try:
|
||
try:
|
||
self.manager = factory()
|
||
except (OSError, ValueError) as exc:
|
||
if self._factory is not None:
|
||
raise
|
||
bits = struct.calcsize("P") * 8
|
||
raise RuntimeError(
|
||
f"Не удалось загрузить VISA-драйвер. Установите NI-VISA или TekVISA {bits}-bit "
|
||
"с поддержкой вашей Windows (для Win7 нужна совместимая версия). Причина: " + str(exc)
|
||
) from exc
|
||
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() not in ("DPO4034", "DPO4034B", "MDO4034C"):
|
||
raise ValueError(f"Ожидался Tektronix DPO4034, DPO4034B или MDO4034C, получено: {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()
|
||
|
||
@staticmethod
|
||
def _channel(channel):
|
||
if channel not in ("CH1", "CH2", "CH3", "CH4"):
|
||
raise ValueError("Выберите канал CH1…CH4")
|
||
return channel
|
||
|
||
def is_running(self):
|
||
"""Read hardware Run/Stop without changing acquisition or transferring data."""
|
||
if self.instrument is None:
|
||
raise RuntimeError("Прибор не подключён")
|
||
state = self.instrument.query("ACQUIRE:STATE?").strip().upper()
|
||
if state in ("0", "OFF", "STOP"):
|
||
return False
|
||
if state in ("1", "ON", "RUN"):
|
||
return True
|
||
raise ValueError("Некорректное состояние сбора: " + state)
|
||
|
||
def acquisition_status(self):
|
||
if self.instrument is None:
|
||
raise RuntimeError("Прибор не подключён")
|
||
fields = self.instrument.query("ACQUIRE:STATE?;:ACQUIRE:STOPAFTER?;:ACQUIRE:NUMACQ?").strip().split(";")
|
||
if len(fields) != 3:
|
||
raise ValueError("Некорректное состояние сбора")
|
||
state, mode, count = (value.strip().upper() for value in fields)
|
||
if state not in ("0", "1", "OFF", "ON", "STOP", "RUN"):
|
||
raise ValueError("Некорректное состояние сбора")
|
||
if mode not in ("RUNSTOP", "RUNST", "SEQUENCE", "SEQ"):
|
||
raise ValueError("Некорректный режим сбора")
|
||
return state in ("1", "ON", "RUN"), mode in ("SEQUENCE", "SEQ"), int(count)
|
||
|
||
def settings(self, channel="CH1"):
|
||
self._channel(channel)
|
||
if self.instrument is None:
|
||
raise RuntimeError("Прибор не подключён")
|
||
commands = ["TRIGGER:A:TYPE?", "TRIGGER:A:MODE?", "TRIGGER:A:EDGE:SOURCE?",
|
||
"TRIGGER:A:EDGE:SLOPE?", f"TRIGGER:A:LEVEL:{channel}?",
|
||
"HORIZONTAL:SCALE?", f"SELECT:{channel}?", f"{channel}:SCALE?",
|
||
f"{channel}:POSITION?", f"{channel}:COUPLING?", "ACQUIRE:MODE?",
|
||
"ACQUIRE:NUMAVG?", "ACQUIRE:STATE?", "TRIGGER:STATE?", "HORIZONTAL:RECORDLENGTH?"]
|
||
values = self.instrument.query(";:".join(commands)).strip().split(";")
|
||
if len(values) != len(commands):
|
||
raise ValueError("Некорректный ответ настроек прибора")
|
||
keys = ("trigger_type", "trigger_mode", "trigger_source", "trigger_slope", "trigger_level",
|
||
"time_scale", "channel_enabled", "channel_scale", "channel_position", "coupling",
|
||
"acquire_mode", "averages", "running", "trigger_state", "record_length")
|
||
result = dict(zip(keys, (v.strip().strip('"') for v in values)))
|
||
result["channel"] = channel
|
||
for key in ("trigger_level", "time_scale", "channel_scale", "channel_position"):
|
||
result[key] = float(result[key])
|
||
if not math.isfinite(result[key]):
|
||
raise ValueError("Некорректное числовое значение настроек")
|
||
for key in ("channel_enabled", "averages", "running", "record_length"):
|
||
result[key] = int(result[key])
|
||
source = result["trigger_source"]
|
||
if source in ("CH1", "CH2", "CH3", "CH4") and source != channel:
|
||
result["trigger_level"] = float(self.instrument.query(f"TRIGGER:A:LEVEL:{source}?").strip())
|
||
return result
|
||
|
||
def configure(self, section, values, channel="CH1"):
|
||
self._channel(channel)
|
||
if self.instrument is None:
|
||
raise RuntimeError("Прибор не подключён")
|
||
def number(key, positive=False):
|
||
value = float(values[key])
|
||
if not math.isfinite(value) or (positive and value <= 0):
|
||
raise ValueError("Некорректное значение: " + key)
|
||
return format(value, ".12g")
|
||
def choice(key, allowed):
|
||
value = values[key]
|
||
if value not in allowed:
|
||
raise ValueError("Некорректное значение: " + key)
|
||
return value
|
||
if section == "preset":
|
||
from .tektronix_presets import PRESETS
|
||
if values.get("preset") not in PRESETS:
|
||
raise ValueError("Неизвестный пресет")
|
||
preset = PRESETS[values["preset"]]
|
||
commands = ["ACQUIRE:MODE " + preset["mode"],
|
||
"HORIZONTAL:RECORDLENGTH " + str(preset["record_length"])]
|
||
if preset["live"]:
|
||
commands += ["SELECT:CH1 ON", "ACQUIRE:STOPAFTER RUNSTOP", "ACQUIRE:STATE RUN"]
|
||
elif section == "trigger":
|
||
source = self._channel(values["source"])
|
||
commands = ["TRIGGER:A:TYPE EDGE", "TRIGGER:A:MODE " + choice("mode", ("AUTO", "NORMAL")),
|
||
"TRIGGER:A:EDGE:SOURCE " + source,
|
||
"TRIGGER:A:EDGE:SLOPE " + choice("slope", ("RISE", "FALL", "EITHER")),
|
||
f"TRIGGER:A:LEVEL:{source} " + number("level")]
|
||
channel = source
|
||
elif section == "horizontal":
|
||
commands = ["HORIZONTAL:SCALE " + number("scale", True)]
|
||
elif section == "record":
|
||
length = int(values["length"])
|
||
if not 1000 <= length <= 20000000:
|
||
raise ValueError("Длина записи должна быть 1000…20000000 точек")
|
||
commands = [f"HORIZONTAL:RECORDLENGTH {length}"]
|
||
elif section == "channel":
|
||
commands = [f"SELECT:{channel} " + choice("enabled", ("ON", "OFF")),
|
||
f"{channel}:SCALE " + number("scale", True),
|
||
f"{channel}:POSITION " + number("position"),
|
||
f"{channel}:COUPLING " + choice("coupling", ("AC", "DC", "GND"))]
|
||
elif section == "acquisition":
|
||
averages = int(values["averages"])
|
||
if averages not in (2, 4, 8, 16, 32, 64, 128, 256, 512):
|
||
raise ValueError("Число усреднений: 2…512, степень двойки")
|
||
commands = ["ACQUIRE:MODE " + choice("mode", ("SAMPLE", "AVERAGE", "HIRES")),
|
||
f"ACQUIRE:NUMAVG {averages}"]
|
||
else:
|
||
raise ValueError("Неизвестный раздел настроек")
|
||
self.instrument.write(";:".join(commands))
|
||
return self.settings(channel)
|
||
|
||
def acquisition(self, action, channel="CH1"):
|
||
self._channel(channel)
|
||
if self.instrument is None:
|
||
raise RuntimeError("Прибор не подключён")
|
||
commands = {"run": "ACQUIRE:STOPAFTER RUNSTOP;:ACQUIRE:STATE RUN",
|
||
"stop": "ACQUIRE:STATE STOP",
|
||
"single": "ACQUIRE:STATE STOP;:ACQUIRE:STOPAFTER SEQUENCE;:ACQUIRE:STATE RUN"}
|
||
if action not in commands:
|
||
raise ValueError("Неизвестная команда сбора")
|
||
self.instrument.write(commands[action])
|
||
# Do not wait for *OPC?: a single acquisition may wait indefinitely for a trigger.
|
||
return self.settings(channel)
|
||
|
||
def capture(self, channels, points=100000, live=False):
|
||
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 live and len(channels) != 1:
|
||
raise ValueError("Быстрое чтение поддерживает один канал")
|
||
if not 1 <= points <= 2000000:
|
||
raise ValueError("Число точек должно быть 1…2000000")
|
||
scope = self.instrument
|
||
if scope is None:
|
||
raise RuntimeError("Прибор не подключён")
|
||
# Freeze the record so all selected channels belong to the same capture.
|
||
running = 0 if live else int(scope.query("ACQUIRE:STATE?").strip())
|
||
result = []
|
||
try:
|
||
if not live:
|
||
scope.query("ACQUIRE:STATE STOP;*OPC?")
|
||
for channel in channels:
|
||
scope.write(f"DATA:SOURCE {channel};:DATA:ENCDG RIBINARY;:DATA:WIDTH 2;:DATA:START 1;:DATA:STOP {points}")
|
||
keys = ("PT_FMT", "XINCR", "XZERO", "PT_OFF", "YMULT", "YOFF", "YZERO", "YUNIT", "NR_PT")
|
||
query = f"SELECT:{channel}?;" + ";".join(f":WFMOUTPRE:{key}?" for key in keys)
|
||
fields = scope.query(query).strip().split(";")
|
||
if len(fields) != 10:
|
||
raise ValueError("Некорректный ответ преамбулы осциллограммы")
|
||
if not int(fields[0]):
|
||
raise ValueError(f"Включите {channel} на осциллографе")
|
||
if fields[1].strip().upper() != "Y":
|
||
raise ValueError("ENV/Peak Detect не поддерживается; выберите Sample или Average")
|
||
params = [float(value) for value in fields[2:8]]
|
||
dx, x0, offset, mult, yoff, y0 = params
|
||
if not all(math.isfinite(v) for v in params) or dx <= 0:
|
||
raise ValueError("Некорректная преамбула осциллограммы")
|
||
unit = fields[8].strip().strip('"')
|
||
count = int(fields[9].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
|