254 lines
13 KiB
Python
254 lines
13 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
|
|
|
|
|
|
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:
|
|
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() not in ("DPO4034", "MDO4034C"):
|
|
raise ValueError(f"Ожидался Tektronix DPO4034 или 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
|