Groesz.org/progpib
1
0
Fork 0
Code Issues 3 Pull Requests Releases Wiki Activity

Replace progpib.py stub with a proper installable package

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
S Groesz
2026-08-24 18:43:08 -05:00
parent 6664760477
commit a9a79b6dd7
19 changed files with 1723 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

134
tests/conftest.py Normal file
View File

@@ -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

84
tests/test_controller.py Normal file
View File

@@ -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)

97
tests/test_discovery_codec.py Normal file
View File

@@ -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

101
tests/test_transport_ethernet.py Normal file
View File

@@ -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()