From a9a79b6dd7dd48876c29642526efa5c0ff1ab487 Mon Sep 17 00:00:00 2001 From: S Groesz Date: Mon, 24 Aug 2026 18:43:08 -0500 Subject: [PATCH] Replace progpib.py stub with a proper installable package Builds a modern, testable API for the Prologix GPIB-ETHERNET controller (GPIB-USB structurally supported but untested, no hardware available): Transport abstraction (Ethernet now, serial later), GpibController for the full "++" command set, a clean-room netfinder discovery client that fixes a real chr()/struct-format bug present in vendortools/nfutil.py, and a CLI (discover/config/info/terminal). vendortools/, pty-gpib-emulator/, sampledata.txt and the manual PDF are kept untouched as reference material. Depends on the sibling bits package (editable, not yet pinned -- see its own commit) for MAC formatting and IEEE-488.2 status-byte bit access. Co-Authored-By: Claude Sonnet 5 --- .gitignore | 2 + LICENSE | 21 ++ README.md | 53 ++++- progpib.py | 53 ----- progpib/__init__.py | 28 +++ progpib/cli.py | 210 +++++++++++++++++ progpib/controller.py | 283 ++++++++++++++++++++++ progpib/discovery.py | 389 +++++++++++++++++++++++++++++++ progpib/escaping.py | 21 ++ progpib/exceptions.py | 22 ++ progpib/transport/__init__.py | 5 + progpib/transport/base.py | 109 +++++++++ progpib/transport/ethernet.py | 68 ++++++ progpib/transport/serial.py | 73 ++++++ pyproject.toml | 24 ++ tests/conftest.py | 134 +++++++++++ tests/test_controller.py | 84 +++++++ tests/test_discovery_codec.py | 97 ++++++++ tests/test_transport_ethernet.py | 101 ++++++++ 19 files changed, 1723 insertions(+), 54 deletions(-) create mode 100644 LICENSE delete mode 100644 progpib.py create mode 100644 progpib/__init__.py create mode 100644 progpib/cli.py create mode 100644 progpib/controller.py create mode 100644 progpib/discovery.py create mode 100644 progpib/escaping.py create mode 100644 progpib/exceptions.py create mode 100644 progpib/transport/__init__.py create mode 100644 progpib/transport/base.py create mode 100644 progpib/transport/ethernet.py create mode 100644 progpib/transport/serial.py create mode 100644 pyproject.toml create mode 100644 tests/conftest.py create mode 100644 tests/test_controller.py create mode 100644 tests/test_discovery_codec.py create mode 100644 tests/test_transport_ethernet.py diff --git a/.gitignore b/.gitignore index ad6b940..3dd0971 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,5 @@ __pycache__ *.swp *.egg-info *.bak2 +.venv +.pytest_cache diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..5e50557 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 SGroesz + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md index 6be01e2..1872049 100644 --- a/README.md +++ b/README.md @@ -1,3 +1,54 @@ # progpib -Prologix GPIB python tools. \ No newline at end of file +A modern Python 3 API for the [Prologix](https://prologix.biz/) GPIB-ETHERNET +controller. GPIB-USB is supported behind the same interface but is +**untested** -- no GPIB-USB hardware was available while building this +package (see `progpib/transport/serial.py`). + +## Install (development) + +This package depends on [`bits`](https://git.groesz.org/Groesz.org/bits) +(distribution name `binary-bits`), currently used as a local editable +checkout rather than a pinned PyPI release -- install it first: + +```sh +python -m venv .venv +.venv/bin/pip install -e ../bits +.venv/bin/pip install -e .[dev] +``` + +## Usage + +```python +from progpib import discover, GpibController + +# Find controllers on the LAN (broadcasts on every local interface): +for device in discover(): + print(device.mac_str, device.ip_address, device.name) + +# Talk to one: +with GpibController.ethernet("192.168.1.50") as gpib: + print(gpib.ver()) + gpib.addr = 5 # address the instrument at GPIB primary address 5 + gpib.eos = 3 # no GPIB terminator appended + gpib.eoi = 1 + response = gpib.query("*IDN?") + print(response.decode()) +``` + +## CLI + +```sh +progpib discover # find controllers on the LAN +progpib info # controller version/mode/config +progpib config --mac AA:BB:CC:DD:EE:FF --dhcp +progpib config --mac AA:BB:CC:DD:EE:FF --static 192.168.1.50 255.255.255.0 192.168.1.1 +progpib terminal --addr 5 # interactive pass-through terminal +``` + +## Reference material + +`vendortools/`, `pty-gpib-emulator/`, `sampledata.txt` and +`PrologixGpibEthernetManual.pdf` are kept as historical/reference material +(original vendor tooling, a separate PTY-based emulator side project, and a +captured protocol session) -- nothing in `progpib/` imports from them. diff --git a/progpib.py b/progpib.py deleted file mode 100644 index 119a982..0000000 --- a/progpib.py +++ /dev/null @@ -1,53 +0,0 @@ -#!/usr/env python3 - -import time -import socket - - -class ProGPIB: - """ - ===== - ProGPIB - ===== - - Provides an interface to Prologix GPIB adapters - """ - - #from bits import Bytes - - def __init__(self, IP="0.0.0.0", MAC="02:01:01:01:01:01", Virtual=False): - """ - MAC: Media Access Control (MAC) Address of device - Virtual: (bool) Set True to emulate a Prologix GPIB device - IP: The IPv4 address of the device - """ - self.__start = time.time() - self.__socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP) - self.__socket.settimeout(0.1) - self.__ipaddr = IP - self.__netaddr = MAC - self.__virtual = Virtual - # Prologix configurables - self.__addr = None - self.__auto = None - self.__eoi = None - self.__eos = None - self.__eot_enable = None - self.__eot_char = None - self.__mode = None - self.__read_tmo_ms = None - self.__savecfg = None - self.__ver = None - # Prologix commands - # clr - # ifc - # llo - # loc - # lon - # read - # rst - # spoll - # srq - # status - # trg - # help \ No newline at end of file diff --git a/progpib/__init__.py b/progpib/__init__.py new file mode 100644 index 0000000..f4e0220 --- /dev/null +++ b/progpib/__init__.py @@ -0,0 +1,28 @@ +from .controller import GpibController +from .discovery import DeviceInfo, IpType, discover, identify, set_network_config, test +from .exceptions import ( + CommandError, + DiscoveryError, + ProgpibError, + TransportClosed, + TransportError, + TransportTimeout, +) + +__version__ = "0.1.0" + +__all__ = [ + "GpibController", + "DeviceInfo", + "IpType", + "discover", + "identify", + "set_network_config", + "test", + "ProgpibError", + "TransportError", + "TransportTimeout", + "TransportClosed", + "CommandError", + "DiscoveryError", +] diff --git a/progpib/cli.py b/progpib/cli.py new file mode 100644 index 0000000..102c605 --- /dev/null +++ b/progpib/cli.py @@ -0,0 +1,210 @@ +"""progpib CLI: discovery, network configuration, controller info, and an +interactive terminal -- a modernized, corrected replacement for +`vendortools/nfcli.py`'s discovery/config ability, plus commands it never +had (talking to the command channel at all). +""" + +import argparse +import json +import socket +import sys + +from bits import Bytes + +from . import __version__ +from .controller import GpibController +from .discovery import IpType, discover as discover_devices, identify, set_network_config +from .exceptions import ProgpibError + + +def _parse_mac(value: str) -> bytes: + cleaned = value.strip().replace(":", "").replace("-", "").replace(" ", "") + try: + mac = bytes.fromhex(cleaned) + except ValueError as exc: + raise argparse.ArgumentTypeError(f"invalid MAC address: {value}") from exc + if len(mac) != 6: + raise argparse.ArgumentTypeError(f"invalid MAC address: {value}") + return mac + + +def _is_valid_ipv4(value: str) -> bool: + try: + socket.inet_aton(value) + return True + except OSError: + return False + + +def _cmd_discover(args: argparse.Namespace) -> int: + on_progress = None if args.json else (lambda msg: print(msg, file=sys.stderr)) + devices = discover_devices(timeout=args.timeout, on_progress=on_progress) + + if args.json: + payload = [ + { + "mac": d.mac_str, + "ip_address": d.ip_address, + "netmask": d.netmask, + "gateway": d.gateway, + "ip_type": d.ip_type.name, + "name": d.name, + "app_version": d.app_version, + "boot_version": d.boot_version, + "hw_version": d.hw_version, + "host_ip": d.host_ip, + } + for d in devices + ] + print(json.dumps(payload, indent=2)) + return 0 + + if not devices: + print("No Prologix GPIB-ETHERNET controllers found.") + return 0 + + print(f"Found {len(devices)} controller(s):") + for d in devices: + print( + f" {d.mac_str} {d.ip_address:<15} {d.ip_type.name:<7} {d.name} " + f"(app {d.app_version}, boot {d.boot_version}, hw {d.hw_version})" + ) + return 0 + + +def _cmd_config(args: argparse.Namespace) -> int: + mac_str = Bytes(args.mac).hex(sep=":", bytes_per_sep=1) + + if args.dhcp: + ip_type = IpType.DYNAMIC + ip_address = netmask = gateway = "0.0.0.0" + else: + ip_address, netmask, gateway = args.static + for label, value in (("IP", ip_address), ("netmask", netmask), ("gateway", gateway)): + if not _is_valid_ipv4(value): + print(f"error: invalid {label} address: {value}", file=sys.stderr) + return 1 + ip_type = IpType.STATIC + + device = identify(args.mac, timeout=args.timeout) + if device is None: + print(f"No controller with MAC {mac_str} found.", file=sys.stderr) + return 1 + + print(f"Current config: {device.ip_type.name} {device.ip_address} {device.netmask} {device.gateway}") + print("New config: DHCP" if ip_type == IpType.DYNAMIC else + f"New config: STATIC {ip_address} {netmask} {gateway}") + + if not args.yes: + answer = input("Apply this configuration? [y/N] ").strip().lower() + if answer != "y": + print("Aborted.") + return 1 + + ok = set_network_config( + args.mac, + ip_type=ip_type, + ip_address=ip_address, + netmask=netmask, + gateway=gateway, + host_ip=device.host_ip, + ) + print("Configuration updated." if ok else "Configuration update failed.") + return 0 if ok else 1 + + +def _cmd_info(args: argparse.Namespace) -> int: + with GpibController.ethernet(args.host, port=args.port) as ctrl: + mode = ctrl.mode + print(f"Version: {ctrl.ver()}") + print(f"Mode: {mode} ({'CONTROLLER' if mode == 1 else 'DEVICE'})") + print(f"Address: {ctrl.addr}") + print(f"Auto: {ctrl.auto}") + print(f"EOI: {ctrl.eoi}") + print(f"EOS: {ctrl.eos}") + print(f"EOT enable: {ctrl.eot_enable}") + print(f"EOT char: {ctrl.eot_char}") + print(f"Read tmo ms: {ctrl.read_tmo_ms}") + return 0 + + +def _cmd_terminal(args: argparse.Namespace) -> int: + with GpibController.ethernet(args.host, port=args.port) as ctrl: + ctrl.addr = (args.addr, args.sad) if args.sad is not None else args.addr + if args.eos is not None: + ctrl.eos = args.eos + if args.eoi is not None: + ctrl.eoi = args.eoi + + print(f"Connected to {args.host}:{args.port}, addressed to {ctrl.addr}.") + print("Type a command/query to send, 'exit' or Ctrl-D to quit.") + while True: + try: + line = input("gpib> ") + except EOFError: + print() + break + if line.strip().lower() == "exit": + break + if not line: + continue + try: + response = ctrl.query(line) + except ProgpibError as exc: + print(f"error: {exc}", file=sys.stderr) + continue + if response: + print(response.decode("ascii", errors="replace")) + return 0 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="progpib", description="Prologix GPIB controller toolkit") + parser.add_argument("--version", action="version", version=f"progpib {__version__}") + subparsers = parser.add_subparsers(dest="command", required=True) + + p_discover = subparsers.add_parser("discover", help="Find Prologix GPIB-ETHERNET controllers on the LAN") + p_discover.add_argument("--timeout", type=float, default=0.5) + p_discover.add_argument("--json", action="store_true") + p_discover.set_defaults(func=_cmd_discover) + + p_config = subparsers.add_parser("config", help="View or change a controller's network configuration") + p_config.add_argument("--mac", type=_parse_mac, required=True) + p_config.add_argument("--timeout", type=float, default=2.0) + network_group = p_config.add_mutually_exclusive_group(required=True) + network_group.add_argument("--dhcp", action="store_true") + network_group.add_argument("--static", nargs=3, metavar=("IP", "NETMASK", "GATEWAY")) + p_config.add_argument("--yes", action="store_true", help="Apply without interactive confirmation") + p_config.set_defaults(func=_cmd_config) + + p_info = subparsers.add_parser("info", help="Query a controller's ++ command state") + p_info.add_argument("host") + p_info.add_argument("--port", type=int, default=1234) + p_info.set_defaults(func=_cmd_info) + + p_terminal = subparsers.add_parser( + "terminal", help="Interactive pass-through terminal to an addressed instrument" + ) + p_terminal.add_argument("host") + p_terminal.add_argument("--addr", type=int, required=True) + p_terminal.add_argument("--sad", type=int, default=None) + p_terminal.add_argument("--port", type=int, default=1234) + p_terminal.add_argument("--eos", type=int, default=None, choices=[0, 1, 2, 3]) + p_terminal.add_argument("--eoi", type=int, default=None, choices=[0, 1]) + p_terminal.set_defaults(func=_cmd_terminal) + + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + return args.func(args) + except ProgpibError as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/progpib/controller.py b/progpib/controller.py new file mode 100644 index 0000000..470b624 --- /dev/null +++ b/progpib/controller.py @@ -0,0 +1,283 @@ +"""GpibController: a Prologix "++" command-set client over any Transport.""" + +from bits import Bits + +from . import escaping +from .exceptions import CommandError +from .transport import EthernetTransport, SerialTransport, Transport + +_DEFAULT_COMMAND_TIMEOUT = 1.0 # seconds, for ++cmd query replies +_DEFAULT_READ_TIMEOUT = 2.0 # seconds, for ++read / query() instrument data + + +class GpibController: + """A Prologix GPIB controller reached over some `Transport`. + + Use `GpibController.ethernet(host)` or `GpibController.serial(port)` to + build one, or pass an existing `Transport` directly. Supports use as a + context manager, which opens/closes the underlying transport. + """ + + def __init__(self, transport: Transport): + self._transport = transport + + @classmethod + def ethernet(cls, host: str, port: int = 1234, **kwargs) -> "GpibController": + return cls(EthernetTransport(host, port, **kwargs)) + + @classmethod + def serial(cls, port: str, **kwargs) -> "GpibController": + return cls(SerialTransport(port, **kwargs)) + + def __enter__(self) -> "GpibController": + self._transport.open() + return self + + def __exit__(self, *exc_info) -> None: + self._transport.close() + + # -- low-level command helpers ------------------------------------- + + def _send_command(self, command: str) -> None: + self._transport.write(f"++{command}\n".encode("ascii")) + + def _query_command(self, command: str, *, timeout: float | None = None) -> str: + self._send_command(command) + line = self._transport.read_line(timeout=timeout or _DEFAULT_COMMAND_TIMEOUT) + return line.decode("ascii", errors="replace").strip() + + def _get_int(self, name: str) -> int: + return int(self._query_command(name)) + + def _set_int(self, name: str, value: int, lo: int, hi: int) -> None: + value = int(value) + if not (lo <= value <= hi): + raise ValueError(f"{name} must be between {lo} and {hi}, got {value}") + self._send_command(f"{name} {value}") + + def _require_device_mode(self, command_name: str) -> None: + if self.mode != 0: + raise CommandError(f"++{command_name} is only valid in DEVICE mode") + + # -- config properties ------------------------------------------------ + + @property + def mode(self) -> int: + """0 = DEVICE, 1 = CONTROLLER.""" + return self._get_int("mode") + + @mode.setter + def mode(self, value: int) -> None: + self._set_int("mode", value, 0, 1) + + @property + def auto(self) -> int: + return self._get_int("auto") + + @auto.setter + def auto(self, value: int) -> None: + self._set_int("auto", value, 0, 1) + + @property + def eoi(self) -> int: + return self._get_int("eoi") + + @eoi.setter + def eoi(self, value: int) -> None: + self._set_int("eoi", value, 0, 1) + + @property + def eos(self) -> int: + """0 = CR+LF, 1 = CR, 2 = LF, 3 = None.""" + return self._get_int("eos") + + @eos.setter + def eos(self, value: int) -> None: + self._set_int("eos", value, 0, 3) + + @property + def eot_enable(self) -> int: + return self._get_int("eot_enable") + + @eot_enable.setter + def eot_enable(self, value: int) -> None: + self._set_int("eot_enable", value, 0, 1) + + @property + def eot_char(self) -> int: + return self._get_int("eot_char") + + @eot_char.setter + def eot_char(self, value: int) -> None: + self._set_int("eot_char", value, 0, 255) + + @property + def read_tmo_ms(self) -> int: + return self._get_int("read_tmo_ms") + + @read_tmo_ms.setter + def read_tmo_ms(self, value: int) -> None: + self._set_int("read_tmo_ms", value, 1, 3000) + + @property + def savecfg(self) -> int: + return self._get_int("savecfg") + + @savecfg.setter + def savecfg(self, value: int) -> None: + self._set_int("savecfg", value, 0, 1) + + @property + def lon(self) -> int: + """DEVICE mode only: "listen-only" mode.""" + self._require_device_mode("lon") + return self._get_int("lon") + + @lon.setter + def lon(self, value: int) -> None: + self._require_device_mode("lon") + self._set_int("lon", value, 0, 1) + + @property + def status(self) -> Bits: + """DEVICE mode only: the serial-poll status byte. + + Returned as a `bits.Bits` (constructed with `msb_last=True`, i.e. + LSB-first indexing) so the RQS flag (bit #6 per IEEE-488.2) can be + read directly as `controller.status[6]`; `int(status)` gives the + raw byte. + """ + self._require_device_mode("status") + return Bits(self._get_int("status"), msb_last=True) + + @status.setter + def status(self, value: int) -> None: + self._require_device_mode("status") + value = int(value) + if not (0 <= value <= 255): + raise ValueError(f"status must be between 0 and 255, got {value}") + self._send_command(f"status {value}") + + @property + def addr(self) -> int | tuple[int, int]: + parts = self._query_command("addr").split() + if len(parts) == 1: + return int(parts[0]) + pad, sad = parts + return (int(pad), int(sad)) + + @addr.setter + def addr(self, value: int | tuple[int, int]) -> None: + pad, sad = value if isinstance(value, tuple) else (value, None) + pad = int(pad) + if not (0 <= pad <= 30): + raise ValueError(f"primary address must be between 0 and 30, got {pad}") + if sad is None: + self._send_command(f"addr {pad}") + return + sad = int(sad) + if not (96 <= sad <= 126): + raise ValueError(f"secondary address must be between 96 and 126, got {sad}") + self._send_command(f"addr {pad} {sad}") + + # -- action commands ---------------------------------------------------- + + def clr(self) -> None: + """Send Selected Device Clear (SDC) to the currently addressed instrument.""" + self._send_command("clr") + + def ifc(self) -> None: + """Assert GPIB IFC for 150us, becoming Controller-In-Charge.""" + self._send_command("ifc") + + def llo(self) -> None: + """Disable front panel operation of the currently addressed instrument.""" + self._send_command("llo") + + def loc(self) -> None: + """Enable front panel operation of the currently addressed instrument.""" + self._send_command("loc") + + def rst(self) -> None: + """Power-on reset (~5s); input received during that time is ignored.""" + self._send_command("rst") + + def srq(self) -> bool: + """Current state of the GPIB SRQ signal.""" + return bool(self._get_int("srq")) + + def spoll(self, pad: int | None = None, sad: int | None = None) -> Bits: + """Serial-poll an instrument (default: the currently addressed one). + + Returns the status byte as a `bits.Bits` (LSB-first indexing, see + `status` above) for bit-level RQS inspection (`result[6]`); + `int(result)` gives the raw byte. + """ + if pad is None: + command = "spoll" + elif sad is None: + command = f"spoll {pad}" + else: + command = f"spoll {pad} {sad}" + return Bits(int(self._query_command(command)), msb_last=True) + + def trg(self, *addrs: int | tuple[int, int]) -> None: + """Group Execute Trigger, up to 15 addresses (default: current).""" + if len(addrs) > 15: + raise ValueError("trg accepts at most 15 addresses") + parts: list[str] = [] + for addr in addrs: + if isinstance(addr, tuple): + pad, sad = addr + parts.extend((str(int(pad)), str(int(sad)))) + else: + parts.append(str(int(addr))) + self._send_command("trg" if not parts else "trg " + " ".join(parts)) + + def ver(self) -> str: + """The controller's version string.""" + return self._query_command("ver") + + def help(self) -> str: + """A summary of available commands, as printed by the controller.""" + self._send_command("help") + raw = self._transport.read_until_quiet(timeout=1.0, idle=0.3) + return raw.decode("ascii", errors="replace").strip() + + def read(self, mode: str | int | None = None, timeout: float | None = None) -> bytes: + """Read data from the addressed instrument. + + `mode` is `"eoi"` (read until EOI), an int 0-255 (read until that + terminator byte), or `None` (read until `read_tmo_ms` timeout). + """ + if mode is None: + command = "read" + elif mode == "eoi": + command = "read eoi" + else: + char = int(mode) + if not (0 <= char <= 255): + raise ValueError("terminator char must be between 0 and 255") + command = f"read {char}" + self._send_command(command) + return self._transport.read_until_quiet(timeout=timeout or _DEFAULT_READ_TIMEOUT) + + # -- raw instrument I/O --------------------------------------------- + + def write(self, data: bytes | str) -> None: + """Send `data` to the addressed instrument, escaped per the protocol.""" + if isinstance(data, str): + data = data.encode("ascii") + self._transport.write(escaping.escape(data) + b"\n") + + def query(self, data: bytes | str, *, read_timeout: float | None = None) -> bytes: + """`write(data)` then explicitly `++read eoi` the response. + + Deliberately does not rely on `++auto 1`: Prologix documents that + some instruments hang when addressed to talk after a command that + generates no response. Matches the safer pattern used in + `vendortools/arb_eth.py` (`++auto 0` + explicit `++read`). + """ + self.write(data) + self._send_command("read eoi") + return self._transport.read_until_quiet(timeout=read_timeout or _DEFAULT_READ_TIMEOUT) diff --git a/progpib/discovery.py b/progpib/discovery.py new file mode 100644 index 0000000..ae46242 --- /dev/null +++ b/progpib/discovery.py @@ -0,0 +1,389 @@ +"""Netfinder: the Prologix GPIB-ETHERNET controller's UDP discovery and +network-configuration protocol (broadcast, port 3040). + +Struct layouts below are ported from `vendortools/nfutil.py`, which are +correct, but that file has a real Python 3 bug: `MkIdentifyReply` and +`MkAssignment` build single-byte struct fields with `chr(x)`, which produces +a `str`, when `struct.pack`'s `c` format needs a length-1 `bytes` object +(`bytes([x])`). `sampledata.txt` captures this exact call raising +`struct.error: char format requires a bytes object of length 1` against +real hardware, and the working fix used there. Only the pack/unpack pairs a +discovery *client* needs (Identify, Assignment, Test) are implemented here; +the rest (block write/verify/reboot/etc.) are firmware/bootloader tooling, +out of scope. +""" + +import contextlib +import dataclasses +import enum +import random +import socket +import struct +import sys +import time +from typing import Callable, Iterator + +from bits import Bytes + +from .exceptions import DiscoveryError + +NETFINDER_PORT = 3040 +NF_MAGIC = 0x5A + + +class NfCommand(enum.IntEnum): + IDENTIFY = 0 + IDENTIFY_REPLY = 1 + ASSIGNMENT = 2 + ASSIGNMENT_REPLY = 3 + FLASH_ERASE = 4 + FLASH_ERASE_REPLY = 5 + BLOCK_SIZE = 6 + BLOCK_SIZE_REPLY = 7 + BLOCK_WRITE = 8 + BLOCK_WRITE_REPLY = 9 + VERIFY = 10 + VERIFY_REPLY = 11 + REBOOT = 12 + SET_ETHERNET_ADDRESS = 13 + SET_ETHERNET_ADDRESS_REPLY = 14 + TEST = 15 + TEST_REPLY = 16 + + +class NfResult(enum.IntEnum): + SUCCESS = 0 + CRC_MISMATCH = 1 + INVALID_MEMORY_TYPE = 2 + INVALID_SIZE = 3 + INVALID_IP_TYPE = 4 + + +class IpType(enum.IntEnum): + DYNAMIC = 0 + STATIC = 1 + + +_HEADER_FMT = "!2cH6s2x" +_IDENTIFY_REPLY_FMT = "!H6c4s4s4s4s4s4s32s" +_ASSIGNMENT_FMT = "!3xc4s4s4s32x" +_ASSIGNMENT_REPLY_FMT = "!c3x" +_TEST_REPLY_FMT = "!32s" + +_HEADER_LEN = struct.calcsize(_HEADER_FMT) + + +@dataclasses.dataclass(frozen=True) +class DeviceInfo: + """A Prologix GPIB-ETHERNET controller found (or targeted) via netfinder.""" + + mac: bytes + mode: int + alert: int + ip_type: IpType + ip_address: str + netmask: str + gateway: str + app_version: str + boot_version: str + hw_version: str + name: str + uptime_days: int + uptime_hours: int + uptime_minutes: int + uptime_seconds: int + host_ip: str | None = None # which local interface discovered it + + @property + def mac_str(self) -> str: + return Bytes(self.mac).hex(sep=":", bytes_per_sep=1) + + +def _pack_header(command: NfCommand, seq: int, mac: bytes) -> bytes: + return struct.pack(_HEADER_FMT, bytes([NF_MAGIC]), bytes([int(command)]), seq, mac) + + +def _unpack_header(msg: bytes) -> dict: + magic, command, seq, mac = struct.unpack(_HEADER_FMT, msg) + return {"magic": ord(magic), "command": ord(command), "seq": seq, "mac": mac} + + +def _pack_identify(seq: int) -> bytes: + return _pack_header(NfCommand.IDENTIFY, seq, b"\xff" * 6) + + +def _unpack_identify_reply(msg: bytes) -> DeviceInfo: + header = _unpack_header(msg[:_HEADER_LEN]) + ( + uptime_days, + uptime_hrs, + uptime_mins, + uptime_secs, + mode, + alert, + ip_type, + ip_addr, + netmask, + gateway, + app_ver, + boot_ver, + hw_ver, + name, + ) = struct.unpack(_IDENTIFY_REPLY_FMT, msg[_HEADER_LEN:]) + return DeviceInfo( + mac=header["mac"], + mode=ord(mode), + alert=ord(alert), + ip_type=IpType(ord(ip_type)), + ip_address=socket.inet_ntoa(ip_addr), + netmask=socket.inet_ntoa(netmask), + gateway=socket.inet_ntoa(gateway), + app_version=socket.inet_ntoa(app_ver), + boot_version=socket.inet_ntoa(boot_ver), + hw_version=socket.inet_ntoa(hw_ver), + name=name.split(b"\x00", 1)[0].decode("ascii", errors="replace"), + uptime_days=uptime_days, + uptime_hours=ord(uptime_hrs), + uptime_minutes=ord(uptime_mins), + uptime_seconds=ord(uptime_secs), + ) + + +def _pack_assignment( + seq: int, + mac: bytes, + ip_type: IpType, + ip_address: str, + netmask: str, + gateway: str, +) -> bytes: + return _pack_header(NfCommand.ASSIGNMENT, seq, mac) + struct.pack( + _ASSIGNMENT_FMT, + bytes([int(ip_type)]), # fixed: was chr(ip_type) in vendortools/nfutil.py + socket.inet_aton(ip_address), + socket.inet_aton(netmask), + socket.inet_aton(gateway), + ) + + +def _unpack_assignment_reply(msg: bytes) -> NfResult: + (result,) = struct.unpack(_ASSIGNMENT_REPLY_FMT, msg[_HEADER_LEN:]) + return NfResult(ord(result)) + + +def _pack_test(seq: int, mac: bytes) -> bytes: + return _pack_header(NfCommand.TEST, seq, mac) + + +def _unpack_test_reply(msg: bytes) -> str: + (raw,) = struct.unpack(_TEST_REPLY_FMT, msg[_HEADER_LEN:]) + return raw.split(b"\x00", 1)[0].decode("ascii", errors="replace") + + +def _validate_reply_header( + reply: bytes, expected_command: NfCommand, seq: int, expected_len: int +) -> bool: + if len(reply) != expected_len: + return False + header = _unpack_header(reply[:_HEADER_LEN]) + return ( + header["magic"] == NF_MAGIC + and header["command"] == int(expected_command) + and header["seq"] == seq + ) + + +def _local_ipv4_addresses() -> list[str]: + """Local, non-loopback IPv4 addresses to broadcast netfinder requests from.""" + import platform + + if platform.system() in ("Windows", "Microsoft"): + addresses = socket.gethostbyname_ex(socket.gethostname())[2] + else: + addresses = _local_ipv4_addresses_unix() + return [ip for ip in addresses if ip != "127.0.0.1"] + + +def _local_ipv4_addresses_unix() -> list[str]: + import array + import fcntl + + siocgifconf = 0x8912 + struct_size = 40 if sys.maxsize > 2**32 else 32 + max_interfaces = 128 + inbytes = max_interfaces * struct_size + ifreq = array.array("B", b"\0" * inbytes) + ifconf = struct.pack("iL", inbytes, ifreq.buffer_info()[0]) + + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + ifconf = fcntl.ioctl(sock.fileno(), siocgifconf, ifconf) + finally: + sock.close() + + outbytes = struct.unpack("iL", ifconf)[0] + return [ + socket.inet_ntoa(ifreq[i + 20 : i + 24]) + for i in range(0, outbytes, struct_size) + ] + + +@contextlib.contextmanager +def _netfinder_sockets(interface: str): + """A bound (send, recv) UDP socket pair for broadcasting on `interface`.""" + send_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + recv_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + send_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + send_sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + send_sock.bind((interface, 0)) + port = send_sock.getsockname()[1] + + recv_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + recv_sock.settimeout(0.1) + recv_sock.bind(("", port)) + + yield send_sock, recv_sock + finally: + send_sock.close() + recv_sock.close() + + +def _broadcast_request( + send_sock: socket.socket, recv_sock: socket.socket, packet: bytes, timeout: float +) -> Iterator[bytes]: + """Broadcast `packet` and yield every reply datagram received within `timeout`.""" + send_sock.sendto(packet, ("", NETFINDER_PORT)) + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + yield recv_sock.recv(256) + except socket.timeout: + continue + + +def discover( + *, + timeout: float = 0.5, + attempts: int = 2, + interfaces: list[str] | None = None, + on_progress: Callable[[str], None] | None = None, +) -> list[DeviceInfo]: + """Broadcast-discover Prologix GPIB-ETHERNET controllers on the LAN. + + Sends an Identify request on each local IPv4 interface (or `interfaces`, + if given) and collects IdentifyReply packets for `timeout` seconds, + `attempts` times per interface. Devices found on more than one interface + are deduplicated by MAC address (last interface wins for `host_ip`). + """ + if interfaces is None: + interfaces = _local_ipv4_addresses() + if not interfaces: + raise DiscoveryError("no local IPv4 interfaces found to broadcast from") + + reply_len = _HEADER_LEN + struct.calcsize(_IDENTIFY_REPLY_FMT) + devices: dict[bytes, DeviceInfo] = {} + + for interface in interfaces: + if on_progress: + on_progress(f"scanning via {interface}") + with _netfinder_sockets(interface) as (send_sock, recv_sock): + for _ in range(attempts): + seq = random.randint(1, 65535) + packet = _pack_identify(seq) + for reply in _broadcast_request(send_sock, recv_sock, packet, timeout): + if not _validate_reply_header( + reply, NfCommand.IDENTIFY_REPLY, seq, reply_len + ): + continue + try: + info = _unpack_identify_reply(reply) + except struct.error: + continue + info = dataclasses.replace(info, host_ip=interface) + devices[info.mac] = info + if on_progress: + on_progress(f"found {info.mac_str} at {info.ip_address}") + + return list(devices.values()) + + +def identify( + mac: bytes, + *, + timeout: float = 2.0, + attempts: int = 2, + interfaces: list[str] | None = None, +) -> DeviceInfo | None: + """Discover and return the device with the given MAC, or `None`.""" + for info in discover(timeout=timeout, attempts=attempts, interfaces=interfaces): + if info.mac == mac: + return info + return None + + +def set_network_config( + mac: bytes, + *, + ip_type: IpType, + ip_address: str = "0.0.0.0", + netmask: str = "0.0.0.0", + gateway: str = "0.0.0.0", + host_ip: str | None = None, + timeout: float = 0.5, + attempts: int = 10, +) -> bool: + """Set the network configuration of the device with the given MAC. + + `host_ip` should be the local interface that can reach the device (e.g. + `DeviceInfo.host_ip` from a prior `discover()`/`identify()` call); if + omitted, every local interface is tried in turn. Returns whether the + device acknowledged success. Raises `DiscoveryError` if no reply is + received at all after exhausting `attempts` (the vendor tool instead + returns a falsy empty dict, which is easy to silently mishandle). + """ + interfaces = [host_ip] if host_ip else _local_ipv4_addresses() + if not interfaces: + raise DiscoveryError("no local IPv4 interfaces found to broadcast from") + + reply_len = _HEADER_LEN + struct.calcsize(_ASSIGNMENT_REPLY_FMT) + + for interface in interfaces: + with _netfinder_sockets(interface) as (send_sock, recv_sock): + for _ in range(attempts): + seq = random.randint(1, 65535) + packet = _pack_assignment(seq, mac, ip_type, ip_address, netmask, gateway) + for reply in _broadcast_request(send_sock, recv_sock, packet, timeout): + if not _validate_reply_header( + reply, NfCommand.ASSIGNMENT_REPLY, seq, reply_len + ): + continue + header = _unpack_header(reply[:_HEADER_LEN]) + if header["mac"] != mac: + continue + return _unpack_assignment_reply(reply) == NfResult.SUCCESS + + raise DiscoveryError(f"no response configuring {mac.hex()}") + + +def test(mac: bytes, *, host_ip: str | None = None, timeout: float = 0.5) -> str | None: + """Send a Test request to the device with the given MAC; a simple liveness + check, e.g. after `set_network_config`. Returns the reply string, or + `None` if there was no reply. + """ + interfaces = [host_ip] if host_ip else _local_ipv4_addresses() + reply_len = _HEADER_LEN + struct.calcsize(_TEST_REPLY_FMT) + + for interface in interfaces: + with _netfinder_sockets(interface) as (send_sock, recv_sock): + seq = random.randint(1, 65535) + packet = _pack_test(seq, mac) + for reply in _broadcast_request(send_sock, recv_sock, packet, timeout): + if not _validate_reply_header(reply, NfCommand.TEST_REPLY, seq, reply_len): + continue + header = _unpack_header(reply[:_HEADER_LEN]) + if header["mac"] != mac: + continue + return _unpack_test_reply(reply) + + return None diff --git a/progpib/escaping.py b/progpib/escaping.py new file mode 100644 index 0000000..68d599f --- /dev/null +++ b/progpib/escaping.py @@ -0,0 +1,21 @@ +"""Byte escaping for the Prologix host-to-controller data channel. + +Per the Prologix GPIB-ETHERNET manual, section 8 (Data Transmission): CR, +LF, ESC and '+' occurring in data sent *to* the controller must be preceded +by an ESC byte, since an unescaped CR/LF terminates the line and an +unescaped leading "++" is interpreted as a command. Data the controller +sends back (instrument responses) is never escaped and must not be +re-escaped here. +""" + +_ESCAPED = frozenset(b"\r\n\x1b+") # CR, LF, ESC, '+' + + +def escape(data: bytes) -> bytes: + """Return `data` with each CR/LF/ESC/'+' byte preceded by an ESC byte.""" + out = bytearray() + for b in data: + if b in _ESCAPED: + out.append(0x1B) + out.append(b) + return bytes(out) diff --git a/progpib/exceptions.py b/progpib/exceptions.py new file mode 100644 index 0000000..5c5c0bc --- /dev/null +++ b/progpib/exceptions.py @@ -0,0 +1,22 @@ +class ProgpibError(Exception): + """Base class for all progpib errors.""" + + +class TransportError(ProgpibError): + """A transport-level connect or I/O failure.""" + + +class TransportTimeout(TransportError): + """A read timed out with no data available.""" + + +class TransportClosed(TransportError): + """The peer closed the connection (EOF).""" + + +class CommandError(ProgpibError): + """The controller rejected a command, or it isn't valid in the current mode.""" + + +class DiscoveryError(ProgpibError): + """A netfinder discovery/configuration request failed or got no reply.""" diff --git a/progpib/transport/__init__.py b/progpib/transport/__init__.py new file mode 100644 index 0000000..d04f6a7 --- /dev/null +++ b/progpib/transport/__init__.py @@ -0,0 +1,5 @@ +from .base import Transport +from .ethernet import EthernetTransport +from .serial import SerialTransport + +__all__ = ["Transport", "EthernetTransport", "SerialTransport"] diff --git a/progpib/transport/base.py b/progpib/transport/base.py new file mode 100644 index 0000000..cfd72f9 --- /dev/null +++ b/progpib/transport/base.py @@ -0,0 +1,109 @@ +"""Abstract transport interface shared by the Ethernet and serial backends.""" + +import abc +import time + +from ..exceptions import TransportTimeout + + +class Transport(abc.ABC): + """A single conversation with a Prologix controller. + + Implementations own exactly one physical connection (a TCP socket or a + serial port) and speak in raw bytes. Escaping/formatting of the "++" + command protocol is `controller.GpibController`'s job, not the + transport's. + + Subclasses implement `open`, `close`, `is_open`, `write` and the + protected `_recv_chunk`; `read_line`/`read_until_quiet` are provided + here, built on top of `_recv_chunk`, so the line-buffering and + idle-timeout logic exists exactly once for both backends. + """ + + def __init__(self) -> None: + self._buffer = bytearray() + + @abc.abstractmethod + def open(self) -> None: + """Open the underlying connection. Safe to call once per instance.""" + + @abc.abstractmethod + def close(self) -> None: + """Close the underlying connection. Safe to call even if not open.""" + + @property + @abc.abstractmethod + def is_open(self) -> bool: + """Whether the underlying connection is currently open.""" + + @abc.abstractmethod + def write(self, data: bytes) -> None: + """Send `data` exactly as given.""" + + @abc.abstractmethod + def _recv_chunk(self, timeout: float | None) -> bytes: + """Receive up to an implementation-defined amount of data. + + `timeout` is the maximum time to wait, or `None` to block + indefinitely. Raises `TransportTimeout` if nothing arrives within + `timeout`, and `TransportClosed` if the peer is gone. Always + returns a non-empty `bytes` otherwise. + """ + + def read_line(self, timeout: float | None = None) -> bytes: + """Read one `\\n`- or `\\r\\n`-terminated line (terminator stripped). + + Used for "++cmd" query replies. Raises `TransportTimeout` if no + complete line arrives within `timeout` seconds (or `None` for no + timeout), and `TransportClosed` if the peer closes first. + """ + deadline = None if timeout is None else time.monotonic() + timeout + while True: + newline = self._buffer.find(b"\n") + if newline != -1: + line = bytes(self._buffer[:newline]) + del self._buffer[: newline + 1] + if line.endswith(b"\r"): + line = line[:-1] + return line + remaining = None if deadline is None else deadline - time.monotonic() + if remaining is not None and remaining <= 0: + raise TransportTimeout("read_line timed out") + self._buffer.extend(self._recv_chunk(remaining)) + + def read_until_quiet(self, timeout: float, idle: float | None = None) -> bytes: + """Accumulate bytes until no more arrive for `idle` seconds. + + Used for "++read" instrument data, which is not reliably + newline-terminated and can be binary. `timeout` bounds the total + time spent; `idle` (default: `timeout`) bounds the gap between + successive reads. Returns whatever was accumulated -- an empty + result (no data at all) is a valid outcome, e.g. no instrument on + the bus, and does not raise. + """ + if idle is None: + idle = timeout + start = time.monotonic() + hard_deadline = start + timeout + idle_deadline = start + idle + result = bytearray(self._buffer) + self._buffer.clear() + while True: + deadline = min(hard_deadline, idle_deadline) + remaining = deadline - time.monotonic() + if remaining <= 0: + break + try: + chunk = self._recv_chunk(remaining) + except TransportTimeout: + break + result.extend(chunk) + idle_deadline = time.monotonic() + idle + return bytes(result) + + def __enter__(self) -> "Transport": + self.open() + return self + + def __exit__(self, *exc_info) -> None: + self.close() diff --git a/progpib/transport/ethernet.py b/progpib/transport/ethernet.py new file mode 100644 index 0000000..aecf627 --- /dev/null +++ b/progpib/transport/ethernet.py @@ -0,0 +1,68 @@ +"""TCP transport for the Prologix GPIB-ETHERNET controller (command port 1234).""" + +import socket + +from .base import Transport +from ..exceptions import TransportClosed, TransportError, TransportTimeout + + +class EthernetTransport(Transport): + """Talk to a Prologix GPIB-ETHERNET controller over its TCP command port.""" + + def __init__(self, host: str, port: int = 1234, connect_timeout: float = 3.0): + super().__init__() + self._host = host + self._port = port + self._connect_timeout = connect_timeout + self._sock: socket.socket | None = None + + def open(self) -> None: + if self._sock is not None: + return + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + sock.settimeout(self._connect_timeout) + sock.connect((self._host, self._port)) + except OSError as exc: + sock.close() + raise TransportError( + f"failed to connect to {self._host}:{self._port}: {exc}" + ) from exc + self._sock = sock + + def close(self) -> None: + if self._sock is not None: + self._sock.close() + self._sock = None + self._buffer.clear() + + @property + def is_open(self) -> bool: + return self._sock is not None + + def _require_open(self) -> socket.socket: + if self._sock is None: + raise TransportError("transport is not open") + return self._sock + + def write(self, data: bytes) -> None: + sock = self._require_open() + try: + sock.sendall(data) + except socket.timeout as exc: + raise TransportTimeout("write timed out") from exc + except OSError as exc: + raise TransportError(f"write failed: {exc}") from exc + + def _recv_chunk(self, timeout: float | None) -> bytes: + sock = self._require_open() + sock.settimeout(timeout) + try: + chunk = sock.recv(4096) + except socket.timeout as exc: + raise TransportTimeout("read timed out") from exc + except OSError as exc: + raise TransportError(f"read failed: {exc}") from exc + if chunk == b"": + raise TransportClosed("connection closed by peer") + return chunk diff --git a/progpib/transport/serial.py b/progpib/transport/serial.py new file mode 100644 index 0000000..bf79d64 --- /dev/null +++ b/progpib/transport/serial.py @@ -0,0 +1,73 @@ +"""Serial transport for the Prologix GPIB-USB controller. + +UNTESTED: no GPIB-USB hardware was available while building this package. +This is implemented from the Prologix manual's "++" command protocol and +the vendor tools' `arb.py`/`pxread.py` usage of a Prologix GPIB-USB serial +connection, but has never been exercised against real hardware. If you hit +a problem here, it's likely. +""" + +import serial + +from .base import Transport +from ..exceptions import TransportError, TransportTimeout + + +class SerialTransport(Transport): + """Talk to a Prologix GPIB-USB controller over its virtual COM port. + + The Prologix GPIB-USB controller presents as a USB-CDC virtual serial + port; per Prologix's documentation the configured baud rate is not + meaningful to the device itself. `baudrate` is kept only so this class + matches the shape `pyserial` expects. + """ + + def __init__(self, port: str, baudrate: int = 9600, connect_timeout: float = 3.0): + super().__init__() + self._port = port + self._baudrate = baudrate + self._connect_timeout = connect_timeout + self._ser: serial.Serial | None = None + + def open(self) -> None: + if self._ser is not None: + return + try: + self._ser = serial.Serial( + self._port, self._baudrate, timeout=self._connect_timeout + ) + except serial.SerialException as exc: + raise TransportError(f"failed to open serial port {self._port}: {exc}") from exc + + def close(self) -> None: + if self._ser is not None: + self._ser.close() + self._ser = None + self._buffer.clear() + + @property + def is_open(self) -> bool: + return self._ser is not None and self._ser.is_open + + def _require_open(self) -> serial.Serial: + if self._ser is None: + raise TransportError("transport is not open") + return self._ser + + def write(self, data: bytes) -> None: + ser = self._require_open() + try: + ser.write(data) + except serial.SerialException as exc: + raise TransportError(f"write failed: {exc}") from exc + + def _recv_chunk(self, timeout: float | None) -> bytes: + ser = self._require_open() + ser.timeout = timeout + try: + chunk = ser.read(4096) + except serial.SerialException as exc: + raise TransportError(f"read failed: {exc}") from exc + if not chunk: + raise TransportTimeout("read timed out") + return chunk diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..a386ac3 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,24 @@ +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[project] +name = "progpib" +version = "0.1.0" +description = "Modern Python 3 API for Prologix GPIB-ETHERNET (and, structurally but unvalidated, GPIB-USB) controllers" +readme = "README.md" +requires-python = ">=3.10" +license = { text = "MIT" } +dependencies = [ + "pyserial>=3.5", + "binary-bits", +] + +[project.optional-dependencies] +dev = ["pytest>=7"] + +[project.scripts] +progpib = "progpib.cli:main" + +[tool.pytest.ini_options] +testpaths = ["tests"] diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..1b031f6 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,134 @@ +import socket +import threading + +import pytest + + +class MockTcpServer: + """A single-connection TCP server on 127.0.0.1, driven by `handler(conn)` + in a background thread. Used to test transports/controller without + hardware.""" + + def __init__(self, handler): + self._handler = handler + self._listen_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._listen_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + self._listen_sock.bind(("127.0.0.1", 0)) + self._listen_sock.listen(1) + self.host, self.port = self._listen_sock.getsockname() + self._thread = threading.Thread(target=self._serve, daemon=True) + self._thread.start() + + def _serve(self) -> None: + try: + conn, _ = self._listen_sock.accept() + except OSError: + return + try: + self._handler(conn) + finally: + conn.close() + + def close(self) -> None: + self._listen_sock.close() + self._thread.join(timeout=2.0) + + +@pytest.fixture +def mock_server(): + servers: list[MockTcpServer] = [] + + def make(handler) -> MockTcpServer: + server = MockTcpServer(handler) + servers.append(server) + return server + + yield make + for server in servers: + server.close() + + +class FakeGpibServer: + """Emulates enough of the Prologix "++" command protocol to unit-test + `GpibController` without hardware. + + Tracks config command state in `self.state` and answers query/set + semantics per the Prologix manual. Also records every raw byte + received in `self.raw_received`, independent of line parsing, so tests + can check exact wire bytes (e.g. escaping) without the naive + line-splitter tripping over an escaped LF/CR inside a payload. + """ + + def __init__(self, *, version: str = "Fake Prologix GPIB-ETHERNET Version 1.0"): + self.state = { + "mode": "1", + "addr": "0", + "auto": "0", + "eoi": "1", + "eos": "0", + "eot_enable": "0", + "eot_char": "0", + "read_tmo_ms": "500", + "savecfg": "1", + "status": "0", + "lon": "0", + } + self.version = version + self.spoll_response = 0 + self.read_response = b"" + self.raw_received = bytearray() + + def handle(self, conn: socket.socket) -> None: + buf = bytearray() + conn.settimeout(5.0) + while True: + try: + chunk = conn.recv(4096) + except (socket.timeout, OSError): + break + if not chunk: + break + self.raw_received.extend(chunk) + buf.extend(chunk) + while b"\n" in buf: + idx = buf.index(b"\n") + line = bytes(buf[:idx]) + del buf[: idx + 1] + self._handle_line(conn, line) + + def _handle_line(self, conn: socket.socket, line: bytes) -> None: + if not line.startswith(b"++"): + return # raw instrument data; use raw_received to inspect it + self._handle_command(conn, line[2:].decode("ascii")) + + def _handle_command(self, conn: socket.socket, command: str) -> None: + parts = command.split() + if not parts: + return + name, args = parts[0], parts[1:] + + if name == "ver": + conn.sendall((self.version + "\n").encode("ascii")) + elif name == "read": + conn.sendall(self.read_response) + elif name == "spoll": + conn.sendall((str(self.spoll_response) + "\n").encode("ascii")) + elif name == "addr": + if args: + self.state["addr"] = " ".join(args) + else: + conn.sendall((self.state["addr"] + "\n").encode("ascii")) + elif name in self.state: + if args: + self.state[name] = args[0] + else: + conn.sendall((str(self.state[name]) + "\n").encode("ascii")) + # else: silently ignore action commands with no return value we + # don't need for these tests (clr, ifc, llo, loc, rst, srq, trg, help) + + +@pytest.fixture +def fake_gpib_server(mock_server): + server_obj = FakeGpibServer() + tcp_server = mock_server(server_obj.handle) + return server_obj, tcp_server diff --git a/tests/test_controller.py b/tests/test_controller.py new file mode 100644 index 0000000..ab5b888 --- /dev/null +++ b/tests/test_controller.py @@ -0,0 +1,84 @@ +import time + +import pytest +from bits import Bit + +from progpib.controller import GpibController +from progpib.exceptions import CommandError + + +def _wait_until(condition, timeout=2.0, interval=0.02): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if condition(): + return True + time.sleep(interval) + return condition() + + +def test_mode_property_roundtrip(fake_gpib_server): + server, tcp_server = fake_gpib_server + with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl: + ctrl.mode = 1 + assert ctrl.mode == 1 + ctrl.mode = 0 + assert ctrl.mode == 0 + + +def test_addr_property_roundtrip_as_tuple(fake_gpib_server): + server, tcp_server = fake_gpib_server + with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl: + ctrl.addr = (5, 96) + assert ctrl.addr == (5, 96) + ctrl.addr = 12 + assert ctrl.addr == 12 + + +def test_ver_returns_server_version(fake_gpib_server): + server, tcp_server = fake_gpib_server + with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl: + assert ctrl.ver() == server.version + + +def test_write_escapes_special_bytes_on_the_wire(fake_gpib_server): + server, tcp_server = fake_gpib_server + with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl: + ctrl.write("*IDN?\r\n+test") + expected = b"*IDN?\x1b\r\x1b\n\x1b+test\n" + assert _wait_until(lambda: bytes(server.raw_received) == expected) + + +def test_read_tmo_ms_range_validation_without_network(fake_gpib_server): + server, tcp_server = fake_gpib_server + with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl: + with pytest.raises(ValueError): + ctrl.read_tmo_ms = 5000 + # unchanged: the invalid value was never sent + assert ctrl.read_tmo_ms == 500 + + +def test_status_and_lon_raise_in_controller_mode(fake_gpib_server): + server, tcp_server = fake_gpib_server + server.state["mode"] = "1" # CONTROLLER + with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl: + with pytest.raises(CommandError): + _ = ctrl.status + with pytest.raises(CommandError): + _ = ctrl.lon + + +def test_status_and_spoll_return_bits_with_rqs_at_index_6(fake_gpib_server): + server, tcp_server = fake_gpib_server + server.state["mode"] = "0" # DEVICE + server.state["status"] = "72" # 0x48: RQS (bit 6) and bit 3 set + server.spoll_response = 72 + + with GpibController.ethernet(tcp_server.host, tcp_server.port) as ctrl: + status = ctrl.status + assert int(status) == 72 + assert status[6] == Bit(1) + assert status[5] == Bit(0) + + result = ctrl.spoll() + assert int(result) == 72 + assert result[6] == Bit(1) diff --git a/tests/test_discovery_codec.py b/tests/test_discovery_codec.py new file mode 100644 index 0000000..1525364 --- /dev/null +++ b/tests/test_discovery_codec.py @@ -0,0 +1,97 @@ +"""Pure, offline tests for the netfinder packet codec in progpib.discovery. + +Fixture bytes are taken from sampledata.txt, a captured Python 3 REPL +session against real Prologix GPIB-ETHERNET hardware (all using the same +sequence number, 30327 = 0x7677, reused across the whole session). +""" + +from progpib.discovery import ( + _HEADER_LEN, + DeviceInfo, + IpType, + NfCommand, + NfResult, + _pack_assignment, + _pack_header, + _pack_identify, + _unpack_assignment_reply, + _unpack_identify_reply, + _unpack_test_reply, +) + +SEQ = 30327 +MAC = b"\x00!i\x01-\xaf" + + +def test_pack_identify_matches_captured_bytes(): + assert _pack_identify(SEQ) == b"Z\x00vw\xff\xff\xff\xff\xff\xff\x00\x00" + + +def test_unpack_identify_reply_matches_captured_session(): + reply = ( + b"Z\x01vw\x00!i\x01-\xaf\x00\x00\x00%\x0572\x01\x00\x00\xc0\xa8\x00\xe5" + b"\xff\xff\xff\x00\xc0\xa8\x00\x01\x01\x06\x06\x00\x01\x03\x00\x00" + b"\x01\x03\x00\x00600" + b"\x00" * 29 + ) + assert len(reply) == 76 + info = _unpack_identify_reply(reply) + + assert info == DeviceInfo( + mac=MAC, + mode=1, + alert=0, + ip_type=IpType.DYNAMIC, + ip_address="192.168.0.229", + netmask="255.255.255.0", + gateway="192.168.0.1", + app_version="1.6.6.0", + boot_version="1.3.0.0", + hw_version="1.3.0.0", + name="600", + uptime_days=37, + uptime_hours=5, + uptime_minutes=55, + uptime_seconds=50, + ) + assert info.mac_str == "00:21:69:01:2d:af" + + +def test_pack_assignment_matches_captured_bytes(): + # This regression-pins the chr()->bytes([...]) fix: vendortools/nfutil.py's + # MkAssignment raises struct.error on this exact call (see sampledata.txt), + # since struct.pack's 'c' format needs a length-1 bytes object, not chr()'s str. + packet = _pack_assignment( + SEQ, + MAC, + IpType.DYNAMIC, + "192.168.0.229", + "255.255.255.0", + "192.168.0.1", + ) + expected = ( + b"Z\x02vw\x00!i\x01-\xaf\x00\x00\x00\x00\x00\x00\xc0\xa8\x00\xe5" + b"\xff\xff\xff\x00\xc0\xa8\x00\x01" + b"\x00" * 32 + ) + assert len(expected) == 60 + assert packet == expected + + +def test_unpack_assignment_reply_matches_captured_bytes(): + # This datagram is captured mid-session in sampledata.txt as a reply to + # an earlier Assignment request (both used seq 30327): header id 0x03 = + # ASSIGNMENT_REPLY, followed by a single 0x00 result byte = NF_SUCCESS. + reply = b"Z\x03vw\x00!i\x01-\xaf\x00\x00\x00\x00\x00\x00" + assert _unpack_assignment_reply(reply) == NfResult.SUCCESS + + +def test_pack_and_unpack_test_reply_roundtrip(): + # sampledata.txt has no clean captured TestReply (the byte string shown + # there after a MkTest() call is actually a stale AssignmentReply still + # sitting in the receive socket's buffer from an earlier command in that + # REPL session -- see test_unpack_assignment_reply_matches_captured_bytes + # above). So this is a synthetic round-trip instead of a captured fixture. + text = "Prologix GPIB-ETHERNET Sim 1.0" + body = text.encode("ascii").ljust(32, b"\x00") + reply = _pack_header(NfCommand.TEST_REPLY, SEQ, MAC) + body + assert len(reply) == _HEADER_LEN + 32 + assert _unpack_test_reply(reply) == text diff --git a/tests/test_transport_ethernet.py b/tests/test_transport_ethernet.py new file mode 100644 index 0000000..0bd50a8 --- /dev/null +++ b/tests/test_transport_ethernet.py @@ -0,0 +1,101 @@ +import time + +from progpib.exceptions import TransportClosed, TransportTimeout +from progpib.transport import EthernetTransport + + +def test_read_line_reassembles_split_recv(mock_server): + def handler(conn): + conn.sendall(b"hello, wo") + time.sleep(0.05) + conn.sendall(b"rld\n") + time.sleep(0.2) + + server = mock_server(handler) + transport = EthernetTransport(server.host, server.port) + transport.open() + try: + line = transport.read_line(timeout=2.0) + assert line == b"hello, world" + finally: + transport.close() + + +def test_read_line_strips_crlf(mock_server): + def handler(conn): + conn.sendall(b"line one\r\n") + time.sleep(0.2) + + server = mock_server(handler) + transport = EthernetTransport(server.host, server.port) + transport.open() + try: + assert transport.read_line(timeout=2.0) == b"line one" + finally: + transport.close() + + +def test_read_line_times_out_with_no_data(mock_server): + def handler(conn): + time.sleep(0.5) + + server = mock_server(handler) + transport = EthernetTransport(server.host, server.port) + transport.open() + try: + try: + transport.read_line(timeout=0.1) + assert False, "expected TransportTimeout" + except TransportTimeout: + pass + finally: + transport.close() + + +def test_read_until_quiet_stops_after_idle_gap(mock_server): + def handler(conn): + conn.sendall(b"abc") + time.sleep(0.05) + conn.sendall(b"def") + time.sleep(0.5) # long gap -- read_until_quiet should stop before this + conn.sendall(b"ghi") + + server = mock_server(handler) + transport = EthernetTransport(server.host, server.port) + transport.open() + try: + data = transport.read_until_quiet(timeout=2.0, idle=0.2) + assert data == b"abcdef" + finally: + transport.close() + + +def test_read_until_quiet_returns_empty_with_no_data(mock_server): + def handler(conn): + time.sleep(0.3) + + server = mock_server(handler) + transport = EthernetTransport(server.host, server.port) + transport.open() + try: + assert transport.read_until_quiet(timeout=0.15) == b"" + finally: + transport.close() + + +def test_recv_raises_closed_on_peer_disconnect(mock_server): + def handler(conn): + conn.close() + + server = mock_server(handler) + transport = EthernetTransport(server.host, server.port) + transport.open() + try: + time.sleep(0.1) + try: + transport.read_line(timeout=1.0) + assert False, "expected TransportClosed" + except TransportClosed: + pass + finally: + transport.close()