diff --git a/docs/tektronix-dpo4034.md b/docs/tektronix-dpo4034.md new file mode 100644 index 0000000..c4558a4 --- /dev/null +++ b/docs/tektronix-dpo4034.md @@ -0,0 +1,67 @@ +# Tektronix DPO4034 + +Независимый от Qt драйвер: `python/set_devices/tektronix.py`. +Установка: `pip install "setcorp-templates[tektronix]"` из пакета templates. +Нужна системная библиотека NI-VISA или TekVISA соответствующей разрядности. +Для Python 3.8 / Win7 выбирайте совместимую версию PyVISA и VISA runtime. + +## Подключение + +- Ethernet: включить LAN на приборе, проверить IP и доступность из сети ПК. + Ресурс VISA `TCPIP0::192.168.1.100::inst0::INSTR` (VXI-11). + В SETGUI можно ввести только IP/hostname или полное имя TCPIP INSTR. +- GPIB: контроллер GPIB на ПК с установленным драйвером, кабель GPIB, + адаптер TEK-USB-488, USB device port осциллографа. У адаптера должно быть + питание. Адрес задаётся на приборе в Utility → I/O → GPIB. + Пример ресурса `GPIB0::1::INSTR`; адреса 0…30, номер контроллера отдельно. + +## Реализованный обмен + +Команды SCPI с LF. После открытия `*IDN?` проверяет производителя и точную +модель DPO4034; `HEADER OFF` отключает заголовки ответов. + +Для снимка запрашивается `ACQUIRE:STATE?`, затем отправляется +`ACQUIRE:STATE STOP` и ожидается `*OPC?`. Это позволяет читать несколько +каналов одной остановленной записи. Для каждого CH1…CH4: + +1. `SELECT:CHn?`: канал должен быть включён пользователем на приборе. +2. `DATA:SOURCE CHn`, `DATA:ENCDG RIBINARY`, `DATA:WIDTH 2`, + `DATA:START 1`, `DATA:STOP N`. +3. Чтение `WFMOUTPRE:PT_FMT?`: поддерживается Y; ENV отклоняется явно. +4. Чтение `WFMOUTPRE:XINCR?`, `XZERO?`, `PT_OFF?`, `YMULT?`, `YOFF?`, + `YZERO?`, `YUNIT?`, `NR_PT?` (каждое имя с префиксом `WFMOUTPRE:`). +5. `CURVE?`: IEEE 488.2 definite-length binary block, знаковые 16-битные + отсчёты, старший байт первым. Разбор блока выполняет PyVISA, в том числе + при разделении ответа между транспортными пакетами. + +Для индекса i, начиная с нуля: + +`t[i] = XZERO + (i - PT_OFF) * XINCR` (секунды), +`y[i] = YZERO + (raw[i] - YOFF) * YMULT` (единицы YUNIT). + +Проверяются конечность коэффициентов, положительный шаг времени и число +полученных отсчётов. В `finally` восстанавливается RUN, если до снимка сбор +шёл; исходное состояние STOP сохраняется. При ошибке UI закрывает сеанс. +Тайм-аут операции VISA — 5 с. Драйвер не сбрасывает настройки прибора. + +Снимок читает текущую запись, не ожидает нового аппаратного триггера. +DATA:STOP ограничивает начало записи, а не прореживает всю запись. +В UI предел 250 000 точек на канал (1 млн суммарно). + +## Пример + +```python +from set_devices.tektronix import DPO4034, resource_name + +scope = DPO4034() +try: + print(scope.connect(resource_name("Ethernet", "192.168.1.100"))) + waveforms = scope.capture(["CH1", "CH2"], points=10000) +finally: + scope.close() +``` + +Источник команд и аппаратного подключения: +[MSO4000 / DPO4000 Programmer Manual, 077024801](https://download.tek.com/manual/077024801web.pdf). +Проверка с реальным прибором требуется для обоих интерфейсов; офлайн-тесты +проверяют последовательность команд, масштабирование, ошибки и восстановление RUN. diff --git a/python/pyproject.toml b/python/pyproject.toml index 64fb8f6..bd1277b 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -9,6 +9,7 @@ description = "Reusable SET device protocols, transports and logic analyzers" requires-python = ">=3.8" [project.optional-dependencies] +tektronix = ["PyVISA>=1.13,<2"] qt = ["PySide6>=6.7,<7; python_version >= '3.9'", "PySide2>=5.15,<6; python_version < '3.9'"] saleae = ["logic2-automation>=1.0.7,<2"] diff --git a/python/set_devices/tektronix.py b/python/set_devices/tektronix.py new file mode 100644 index 0000000..408d977 --- /dev/null +++ b/python/set_devices/tektronix.py @@ -0,0 +1,123 @@ +"""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