diff --git a/pyproject.toml b/pyproject.toml index 2ac2002c..fdb2049a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,7 +23,7 @@ "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", ] - dependencies = ["pyvisa>=1.15.0", "typing_extensions"] + dependencies = ["pyvisa>=1.17.0", "typing_extensions"] dynamic = ["version"] [project.optional-dependencies] diff --git a/pyvisa_py/gpib.py b/pyvisa_py/gpib.py index 86b884a8..b7726ff4 100644 --- a/pyvisa_py/gpib.py +++ b/pyvisa_py/gpib.py @@ -9,68 +9,31 @@ import ctypes # Used for missing bindings not ideal from bisect import bisect -from typing import Any, Iterator, List, Tuple, Type, Union +from collections.abc import Iterator +from typing import Any from pyvisa import attributes, constants from pyvisa.constants import ResourceAttribute, StatusCode -from pyvisa.rname import GPIBInstr, GPIBIntfc, parse_resource_name +from pyvisa.rname import GPIBInstr, GPIBIntfc -from . import gpib_constants, prologix +from . import gpib_constants from .common import LOGGER -from .sessions import Session, UnavailableSession, UnknownAttribute, VISARMSession +from .sessions import Session, UnavailableSession, UnknownAttribute # NOTE dummy implementation that is overwritten when a GPIB library is found # Allow to provide session listing even when no GPIB library is available. -def _find_listeners() -> Iterator[Tuple[int, int, int]]: +def _find_listeners() -> Iterator[tuple[int, int, int]]: yield from () -@Session.register(constants.InterfaceType.gpib, "INSTR") -class GPIBSessionDispatch(Session): - """Dispatch to the proper class based on prologix._PrologixIntfcSession.boards. +# Dispatch of ``GPIB::INSTR`` resources (native vs Prologix vs NI-ENET/100 +# bridge) is owned by :mod:`pyvisa_py.gpib_dispatch`, which registers the +# native driver as a backend resolver. This module therefore no longer +# registers a ``(gpib, "INSTR")`` class itself. - Uses the __new__ method to intercept the creation of the instance of a - GPIB session. If parsed.board is found in - prologix._PrologixIntfcSession.boards, create an instance of - prologix.PrologixInstrSession, otherwise create an instance of GPIBSession. - """ - - def __new__( # type: ignore[misc] - cls, - resource_manager_session: VISARMSession, - resource_name: str, - parsed=None, - open_timeout: int | None = None, - ) -> Session: - newcls: Type - - if parsed is None: - parsed = parse_resource_name(resource_name) - - if parsed.board in prologix._PrologixIntfcSession.boards: - newcls = prologix.PrologixInstrSession - else: - newcls = GPIBSession - - return newcls(resource_manager_session, resource_name, parsed, open_timeout) - - @staticmethod - def list_resources() -> List[str]: - return [ - "GPIB%d::%d::INSTR" % (board, pad) - if sad == gpib_constants.sad.NO_SAD - else "GPIB%d::%d::%d::INSTR" - % (board, pad, sad - gpib_constants.sad.FIRST_SAD) - for board, pad, sad in _find_listeners() - ] - - # FIXME when PROLOGIX gain the ability to list resources, we should - # include PROLOGIX results. - - -def make_unavailable(msg: str) -> Type: +def make_unavailable(msg: str) -> type[Any]: """Creates a fake session class that raises a ValueError if instantiated. We can't use Session.register_unavailable() because we need to be able to @@ -83,7 +46,7 @@ def make_unavailable(msg: str) -> Type: Returns ------- - Type[Session] + type[Session] Fake session. """ @@ -155,7 +118,7 @@ def _inner(self): _patch_Gpib() -def _find_boards() -> Iterator[Tuple[int, int]]: +def _find_boards() -> Iterator[tuple[int, int]]: """Find GPIB board addresses.""" for board in range(16): try: @@ -164,7 +127,7 @@ def _find_boards() -> Iterator[Tuple[int, int]]: LOGGER.debug("GPIB board %i error in _find_boards(): %s", board, repr(e)) -def _find_listeners() -> Iterator[Tuple[int, int, int]]: # type: ignore[no-redef] +def _find_listeners() -> Iterator[tuple[int, int, int]]: # type: ignore[no-redef] """Find GPIB listeners.""" for board, boardpad in _find_boards(): for i in range(31): @@ -329,7 +292,7 @@ class _GPIBCommon(Session): # Override parsed to take into account the fact that this class is only used # for a specific kind of resource - parsed: Union[GPIBIntfc, GPIBInstr] + parsed: GPIBIntfc | GPIBInstr #: Bus wide controller. controller: Gpib @@ -384,7 +347,7 @@ def after_parsing(self) -> None: def _get_timeout( self, attribute: constants.ResourceAttribute - ) -> Tuple[int, StatusCode]: + ) -> tuple[int, StatusCode]: if self.interface: gpib_timeout = self.interface.ask(gpib_constants.ask.IbaTMO) if gpib_timeout and gpib_timeout < len(TIMETABLE): @@ -440,7 +403,7 @@ def close(self) -> StatusCode: self.controller.close() return StatusCode.success - def read(self, count: int) -> Tuple[bytes, StatusCode]: + def read(self, count: int) -> tuple[bytes, StatusCode]: """Reads data from device or interface synchronously. Corresponds to viRead function of the VISA library. @@ -467,7 +430,7 @@ def read(self, count: int) -> Tuple[bytes, StatusCode]: return self._read(reader, count, checker, False, None, False, gpib.GpibError) - def write(self, data: bytes) -> Tuple[int, StatusCode]: + def write(self, data: bytes) -> tuple[int, StatusCode]: """Writes data to device or interface synchronously. Corresponds to viWrite function of the VISA library. @@ -561,7 +524,7 @@ def gpib_control_ren(self, mode: constants.RENLineOperation) -> StatusCode: return constants.StatusCode.success - def _get_attribute(self, attribute: ResourceAttribute) -> Tuple[Any, StatusCode]: + def _get_attribute(self, attribute: ResourceAttribute) -> tuple[Any, StatusCode]: """Get the value for a given VISA attribute for this session. Use to implement custom logic for attributes. @@ -738,7 +701,7 @@ class GPIBSession(_GPIBCommon): # type: ignore[no-redef] parsed: GPIBInstr @staticmethod - def list_resources() -> List[str]: + def list_resources() -> list[str]: return [ "GPIB%d::%d::INSTR" % (board, pad) if sad == gpib_constants.sad.NO_SAD @@ -791,7 +754,7 @@ def assert_trigger(self, protocol: constants.TriggerProtocol) -> StatusCode: except gpib.GpibError as e: return convert_gpib_error(e, self.interface.ibsta(), "assert trigger") - def read_stb(self) -> Tuple[int, StatusCode]: + def read_stb(self) -> tuple[int, StatusCode]: """Read the device status byte.""" try: return self.interface.serial_poll(), StatusCode.success @@ -800,7 +763,7 @@ def read_stb(self) -> Tuple[int, StatusCode]: def _get_attribute( self, attribute: constants.ResourceAttribute - ) -> Tuple[Any, StatusCode]: + ) -> tuple[Any, StatusCode]: """Get the value for a given VISA attribute for this session. Use to implement custom logic for attributes. GPIB::INSTR have the @@ -896,10 +859,10 @@ class GPIBInterface(_GPIBCommon): parsed: GPIBIntfc @staticmethod - def list_resources() -> List[str]: + def list_resources() -> list[str]: return ["GPIB%d::INTFC" % board for board, pad in _find_boards()] - def gpib_command(self, command_bytes: bytes) -> Tuple[int, StatusCode]: + def gpib_command(self, command_bytes: bytes) -> tuple[int, StatusCode]: """Write GPIB command byte on the bus. Corresponds to viGpibCommand function of the VISA library. @@ -1002,7 +965,7 @@ def gpib_pass_control( status = gpib_lib.ibpct(did) return convert_gpib_status(status) - def _get_attribute(self, attribute: ResourceAttribute) -> Tuple[Any, StatusCode]: + def _get_attribute(self, attribute: ResourceAttribute) -> tuple[Any, StatusCode]: """Get the value for a given VISA attribute for this session. Use to implement custom logic for attributes. GPIB::INTFC have the diff --git a/pyvisa_py/gpib_dispatch.py b/pyvisa_py/gpib_dispatch.py new file mode 100644 index 00000000..1027c6dc --- /dev/null +++ b/pyvisa_py/gpib_dispatch.py @@ -0,0 +1,223 @@ +# -*- coding: utf-8 -*- +"""Central dispatch for GPIB INSTR resources across pyvisa-py backends. + +pyvisa-py can serve ``GPIB::...::INSTR`` resources through several +mutually exclusive backends — native linux-gpib / gpib-ctypes, a Prologix +controller, or an NI GPIB-ENET/100 bridge. The session registry holds a +single class per ``(InterfaceType.gpib, "INSTR")`` slot. + +This module owns the ``(gpib, "INSTR")`` slot once and resolves the +concrete session class at open time by consulting an ordered list of +backend resolvers. :func:`register_builtin_backends` imports each in-tree +backend defensively and registers a resolver for it; a backend whose +import fails is simply absent, so the remaining backends keep working +regardless of import order. The module has no hard dependency on any GPIB +library, so it can own the slot even when ``gpib.py`` fails to import. + +:copyright: 2026 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. + +""" + +from collections.abc import Callable +from typing import cast + +from pyvisa import constants, rname +from pyvisa.constants import StatusCode +from pyvisa.rname import GPIBInstr +from pyvisa.typing import VISARMSession + +from .common import LOGGER +from .sessions import OpenError, Session, UnavailableSession + +#: A resolver inspects a parsed resource name and returns the session class +#: that should handle it, or ``None`` if the resource is not served by this +#: backend (e.g. the board number is not registered to it). Resources reach +#: this dispatcher only via the ``(gpib, "INSTR")`` slot, so the parsed name +#: is always a :class:`~pyvisa.rname.GPIBInstr`. +GPIBInstrResolver = Callable[[GPIBInstr], type[Session] | None] + +#: Registered backends as ``(priority, label, resolver)``. Lower priority +#: values are consulted first; :func:`register_backend` keeps the list +#: sorted. The sort is stable, so insertion order breaks ties — import +#: order therefore only affects equal-priority backends, never the +#: correctness of the fallback chain. +_GPIB_INSTR_BACKENDS: list[tuple[int, str, GPIBInstrResolver]] = [] + + +def register_backend( + resolver: GPIBInstrResolver, priority: int, label: str = "" +) -> None: + """Register a backend resolver for ``GPIB::INSTR`` dispatch. + + Parameters + ---------- + resolver : GPIBInstrResolver + Callable returning the session class for a parsed resource, or + ``None`` if this backend does not serve it. + priority : int + Lower values are consulted first. A catch-all fallback (e.g. the + native driver, which claims any board) should use a high value so + it never shadows a more specific backend. + label : str + Optional human-readable name, used only for debugging. + + """ + _GPIB_INSTR_BACKENDS.append((priority, label, resolver)) + _GPIB_INSTR_BACKENDS.sort(key=lambda item: item[0]) + + +@Session.register(constants.InterfaceType.gpib, "INSTR") +class GPIBInstrDispatch(Session): + """Resolve a ``GPIB::INSTR`` resource to the right backend session. + + The class is a dispatch shim: its :meth:`__new__` returns an instance + of a *different* class (the backend session), so ``__init__`` is never + run on the dispatcher itself. + + """ + + def __new__( # type: ignore[misc] + cls, + resource_manager_session: VISARMSession, + resource_name: str, + parsed: rname.ResourceName | None = None, + open_timeout: int | None = None, + ) -> Session: + if parsed is None: + parsed = rname.parse_resource_name(resource_name) + # Registered only for (gpib, INSTR), so the parsed name is a GPIBInstr. + gpib_parsed = cast(GPIBInstr, parsed) + + for _priority, _label, resolve in _GPIB_INSTR_BACKENDS: + newcls = resolve(gpib_parsed) + if newcls is not None: + return newcls( + resource_manager_session, resource_name, parsed, open_timeout + ) + + raise OpenError(StatusCode.error_resource_not_found) + + @staticmethod + def list_resources() -> list[str]: + """List native GPIB::INSTR resources found on local boards. + + Only the native linux-gpib / gpib-ctypes listeners are enumerated: + Prologix has no listing support, and NI GPIB-ENET/100 instruments + are discovered at the INTFC level. Returns an empty list when no + native GPIB library is installed. + + """ + try: + from .gpib import _find_listeners + except Exception: + return [] + return [ + "GPIB%d::%d::INSTR" % (board, pad) + if sad == 0 + else "GPIB%d::%d::%d::INSTR" % (board, pad, sad - 0x60) + for board, pad, sad in _find_listeners() + ] + + +# --- Built-in backend wiring ------------------------------------------------ +# Priorities: lower values are consulted first. A backend bound to specific +# boards (Prologix, NI-ENET/100 bridge) must out-rank the native driver, +# which is the catch-all fallback and therefore comes last. +_PRIORITY_BRIDGE = 10 +_PRIORITY_PROLOGIX = 20 +_PRIORITY_NATIVE = 100 + +#: Guards :func:`register_builtin_backends` so repeated calls (e.g. one per +#: ResourceManager) do not append duplicate resolvers. +_builtins_registered = False + + +def _board_resolver(boards: dict, session_cls: type[Session]) -> GPIBInstrResolver: + """Build a resolver that claims a resource only for registered boards. + + The ``boards`` mapping is captured by reference and queried at dispatch + time, so boards registered after wiring (the normal case — INTFC + sessions populate it on open) are still seen. + + """ + + def resolve(parsed: GPIBInstr) -> type[Session] | None: + return session_cls if parsed.board in boards else None + + return resolve + + +def _make_native_unavailable(exc: Exception) -> type[Session]: + """Return an unavailable session explaining the missing GPIB library. + + ``gpib.py`` raises on import when neither linux-gpib nor gpib-ctypes is + present. Routing unmatched boards here preserves the previous + behaviour: a clear, actionable error at open time instead of a generic + "resource not found". + + """ + + class _NativeGPIBUnavailable(UnavailableSession): + session_issue = ( + "Please install linux-gpib (Linux) or gpib-ctypes (Windows, Linux) " + "to use native GPIB::INSTR resources.\n%s" % exc + ) + + return _NativeGPIBUnavailable + + +def register_builtin_backends() -> None: + """Wire pyvisa-py's in-tree GPIB::INSTR backends into the dispatcher. + + Each backend is imported defensively: a backend whose import fails is + simply skipped (native gpib is additionally represented by an + unavailable session so its install hint survives). The call is + idempotent. + + """ + global _builtins_registered + if _builtins_registered: + return + + # NI GPIB-ENET/100 bridge: claims boards bound to a bridge INTFC. Highest + # precedence so a board registered to a bridge is never shadowed. + try: + from . import nienet100 + except Exception as e: + LOGGER.debug("NI-ENET/100 GPIB::INSTR backend not registered: %s", e) + else: + register_backend( + _board_resolver( + nienet100._NIEnet100IntfcSession.boards, + nienet100.NIEnet100InstrSession, + ), + priority=_PRIORITY_BRIDGE, + label="ni-enet100", + ) + + # Prologix controller: claims boards bound to a Prologix interface. + try: + from . import prologix + except Exception as e: # pragma: no cover - prologix import is robust + LOGGER.debug("Prologix GPIB::INSTR backend not registered: %s", e) + else: + register_backend( + _board_resolver( + prologix._PrologixIntfcSession.boards, prologix.PrologixInstrSession + ), + priority=_PRIORITY_PROLOGIX, + label="prologix", + ) + + # Native linux-gpib / gpib-ctypes: catch-all for any board not claimed + # above. + try: + from .gpib import GPIBSession + except Exception as e: + native_cls: type[Session] = _make_native_unavailable(e) + else: + native_cls = GPIBSession + register_backend(lambda parsed: native_cls, priority=_PRIORITY_NATIVE, label="gpib") + + _builtins_registered = True diff --git a/pyvisa_py/highlevel.py b/pyvisa_py/highlevel.py index 9e54335a..0da1f853 100644 --- a/pyvisa_py/highlevel.py +++ b/pyvisa_py/highlevel.py @@ -98,6 +98,26 @@ class PyVisaLibrary(highlevel.VisaLibraryBase): except Exception as e: LOGGER.debug("GPIBSession was not imported %s." % e) + try: + from . import nienet100 + + LOGGER.debug("NIEnet100TCPIPIntfcSession was correctly imported.") + except Exception as e: + LOGGER.debug("NIEnet100TCPIPIntfcSession was not imported %s." % e) + + # Own the (gpib, INSTR) slot from a dependency-free module so dispatch + # works even when gpib.py fails to import, and wire the available GPIB + # backends (NI GPIB-ENET/100 / Prologix / native) into it. Imported + # last, but the central dispatcher makes the result independent of + # backend import order. + try: + from . import gpib_dispatch + + gpib_dispatch.register_builtin_backends() + LOGGER.debug("GPIB::INSTR dispatcher was correctly registered.") + except Exception as e: + LOGGER.debug("GPIB::INSTR dispatcher was not registered %s." % e) + @staticmethod def get_library_paths() -> Iterable[LibraryPath]: """List a dummy library path to allow to create the library.""" diff --git a/pyvisa_py/nienet100.py b/pyvisa_py/nienet100.py new file mode 100644 index 00000000..2fd21ce6 --- /dev/null +++ b/pyvisa_py/nienet100.py @@ -0,0 +1,478 @@ +# -*- coding: utf-8 -*- +"""Sessions for NI GPIB-ENET/100 Ethernet-to-GPIB bridges. + +The bridge speaks a proprietary TCP protocol on ports 5000 (main) and +5015 (companion) — see :mod:`pyvisa_py.protocols.nienet100`. +This module wires that protocol into pyvisa-py as two session types: + +- ``NI-ENET100-TCPIP::::INTFC`` — binds board number ``n`` to the + given box and keeps a connection open as a connectivity sentinel. +- ``GPIB::[::]::INSTR`` — routed to this module when board + ``n`` was registered as a NIENET100 board (the dispatch hook lives in + :mod:`pyvisa_py.gpib_dispatch`). Each INSTR session owns its own TCP + connection to the box; the spec recommends per-resource TCP sessions over + sharing one connection with multi-PAD bracket switching. + +:copyright: 2026 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. + +""" + +from typing import Any, ClassVar + +from pyvisa import attributes, constants, rname +from pyvisa.constants import ResourceAttribute, StatusCode +from pyvisa.typing import VISARMSession + +from . import gpib_constants +from .common import LOGGER +from .protocols import nienet100 +from .sessions import OpenError, Session, UnknownAttribute + + +class _NIEnet100IntfcSession(Session): + """Common base for NI GPIB-ENET/100 INTFC sessions. + + Holds the class-level ``boards`` registry that the GPIB dispatch hook + in :mod:`pyvisa_py.gpib_dispatch` consults to route ``GPIB::*::INSTR`` + resources through the appropriate bridge. + + The INTFC owns its own :class:`~pyvisa_py.protocols.nienet100.EnetConnection` + for the session lifetime. The connection acts as a connectivity sentinel + (the box rejects Device-I/O on stale sessions, so an open socket is a + reliable health signal). INSTR sessions do **not** share this connection; + they each open their own. + + """ + + #: Maps board number (as parsed string) -> INTFC session instance. + #: Populated on open, cleared on close. The GPIB dispatch hook reads + #: this to find the bridge for a given ``GPIB::*::INSTR`` resource. + #: Key is a string to mirror :class:`rname.GPIBInstr.board`, so dispatch + #: lookups with ``parsed.board`` match without conversion. + boards: ClassVar[dict[str, "_NIEnet100IntfcSession"]] = {} + + #: The long-lived connection to the bridge. ``None`` before + #: ``after_parsing`` runs successfully and after ``close``. + interface: nienet100.EnetConnection | None + + def _get_attribute(self, attribute: ResourceAttribute) -> tuple[Any, StatusCode]: + raise UnknownAttribute(attribute) + + def _set_attribute( + self, attribute: ResourceAttribute, attribute_state: Any + ) -> StatusCode: + raise UnknownAttribute(attribute) + + def gpib_send_ifc(self) -> StatusCode: + """Pulse the GPIB IFC (Interface Clear) line via the bridge. + + Maps ``viGpibSendIFC`` to the board-level ibsic verb (``0x1c`` on the + main socket; see :func:`pyvisa_py.protocols.nienet100._ibsic`). IFC is + a board operation, so it lives on the INTFC session — mirroring where + the linux-gpib backend places it. The INTFC opens a board session + (:meth:`EnetConnection.open_board_session`), which leaves the box in + the online state the box requires to accept IFC. + + """ + if self.interface is None: + return StatusCode.error_connection_lost + try: + self.interface.ibsic() + except nienet100.NIEnet100IOError as e: + return _map_iberr_to_status(e.err) + return StatusCode.success + + def close(self) -> StatusCode: + # Always deregister; if open partially failed there may be no entry. + board = getattr(self.parsed, "board", None) + if board is not None: + self.boards.pop(board, None) + if self.interface is not None: + try: + self.interface.close() + except Exception as e: + LOGGER.debug("error closing NI GPIB-ENET/100 connection: %s", e) + self.interface = None + return StatusCode.success + + +@Session.register(constants.InterfaceType.ni_enet100_tcpip, "INTFC") +class NIEnet100TCPIPIntfcSession(_NIEnet100IntfcSession): + """Session for ``NI-ENET100-TCPIP::::INTFC`` resources.""" + + # Override parsed to take into account the fact that this class is only + # used for a specific kind of resource. + parsed: rname.NIEnet100TCPIPIntfc # type: ignore[name-defined] + + @classmethod + def get_low_level_info(cls) -> str: + return "via pure-Python NI GPIB-ENET/100 protocol" + + @staticmethod + def list_resources() -> list[str]: + """Discover bridges on the local broadcast domain and emit resource + strings for each one found. + + Each discovered bridge gets a distinct board number (0-indexed by + sort order of IP) so that ``open_resource`` calls on the returned + strings can all succeed against the same host. Cross-subnet + bridges do not surface here — for those the user must supply the + resource string explicitly with the box IP and an arbitrary board + number. + + Returns an empty list (rather than raising) on any discovery + error — typically a bind conflict or a missing broadcast route. + + """ + # Local import keeps the top-level imports tidy and isolates the + # UDP code path from sessions that never call list_resources. + from .protocols import nienet100_discovery + + try: + boxes = nienet100_discovery.discover(timeout=1.0) + except OSError as e: + LOGGER.debug("NI GPIB-ENET/100 discovery failed: %s", e) + return [] + return [ + "NI-ENET100-TCPIP%d::%s::INTFC" % (i, box.ip) for i, box in enumerate(boxes) + ] + + def __init__( + self, + resource_manager_session: VISARMSession, + resource_name: str, + parsed: rname.ResourceName | None = None, + open_timeout: int | None = None, + ) -> None: + self.interface = None + super().__init__(resource_manager_session, resource_name, parsed, open_timeout) + + def after_parsing(self) -> None: + # pyvisa's open_timeout nominally bounds lock acquisition; like VXI-11 + # we use it as the connection/open timeout, matching the genuine + # driver's own connect timeout. ``None`` or 0 (VI_TMO_IMMEDIATE, the + # pyvisa default) means "use a sane default": a literal ~1 ms socket + # timeout is far too short for the multi-frame board open, so fall + # back to 10 s rather than ``max(0, 0.001)``. + if not self.open_timeout: + connect_timeout_s = 10.0 + else: + connect_timeout_s = max(self.open_timeout / 1000.0, 0.001) + + host = self.parsed.host_address + try: + self.interface = nienet100.EnetConnection( + host, + open_timeout=connect_timeout_s, + # Operational socket timeout: honour the session timeout + # attribute when set, else the connect fallback. + timeout=self.timeout if self.timeout else connect_timeout_s, + ) + # Board-level open (not the device open()): leaves the box online + # so a board-level ibsic (gpib_send_ifc) is accepted. A bare open() + # wedges the box on ibsic. + self.interface.open_board_session() + except Exception as e: + LOGGER.exception( + "Failed to open NI GPIB-ENET/100 at %s for board %s", + host, + self.parsed.board, + ) + if self.interface is not None: + try: + self.interface.close() + except Exception: + pass + self.interface = None + raise OpenError() from e + + self.boards[self.parsed.board] = self + + self.attrs[ResourceAttribute.interface_number] = self.parsed.board + self.attrs[ResourceAttribute.tcpip_address] = host + self.attrs[ResourceAttribute.tcpip_hostname] = host + self.attrs[ResourceAttribute.tcpip_port] = nienet100.PORT_MAIN + + +class NIEnet100InstrSession(Session): + """Session for ``GPIB::[::]::INSTR`` routed through a + GPIB-ENET/100 bridge. + + This class is **not** decorated with ``@Session.register`` directly. + The central dispatcher in :mod:`pyvisa_py.gpib_dispatch` owns the + ``(gpib, "INSTR")`` slot and instantiates this class when the + resource's board number is registered to a NI GPIB-ENET/100 (see + :attr:`_NIEnet100IntfcSession.boards`); otherwise another backend + resolver (Prologix or native linux-gpib / gpib-ctypes) handles it. + + Each INSTR session owns its own :class:`EnetConnection` and its own + bracket — INSTRs do not share TCP sockets with the INTFC or with one + another. This mirrors the wire spec's recommended pattern and lets + multiple instruments on the same bridge operate without a shared + cross-resource lock. + + """ + + # We don't decorate this class with Session.register() because we don't + # want it to be registered in the _session_classes array, but we still + # need to define session_type to make the set_attribute machinery work. + session_type = (constants.InterfaceType.gpib, "INSTR") + + # Override parsed to take into account the fact that this class is only + # used for a specific kind of resource. + parsed: rname.GPIBInstr + + #: The per-session bridge connection. ``None`` before ``after_parsing`` + #: succeeds and after ``close``. Bracket lifecycle is tracked inside + #: the connection itself, so ``close()`` releases any open bracket + #: even when ``after_parsing`` fails mid-way (e.g., a wire error after + #: Frame F was acked). + interface: nienet100.EnetConnection | None + + def __init__( + self, + resource_manager_session: VISARMSession, + resource_name: str, + parsed: rname.ResourceName | None = None, + open_timeout: int | None = None, + ) -> None: + self.interface = None + #: Holds the tail of a wire message for which the caller's max-count was + #: too small to consume. The next read() drains this before going + #: back to the wire, so no bytes are lost between calls. + self._read_buffer = bytearray() + super().__init__(resource_manager_session, resource_name, parsed, open_timeout) + + def after_parsing(self) -> None: + try: + intfc = _NIEnet100IntfcSession.boards[self.parsed.board] + except KeyError as e: + raise OpenError() from e + + # ``None`` or 0 (VI_TMO_IMMEDIATE, the pyvisa default) means "use a + # sane default": a literal ~1 ms timeout makes the TCP connect flaky, + # so fall back to 10 s rather than ``max(0, 0.001)``. + if not self.open_timeout: + connect_timeout_s = 10.0 + else: + connect_timeout_s = max(self.open_timeout / 1000.0, 0.001) + + pad = int(self.parsed.primary_address) + sad_raw = self.parsed.secondary_address + sad = int(sad_raw) + 0x60 if sad_raw is not None else 0 + + # intfc.parsed is a NIEnet100TCPIPIntfc at runtime, but pyvisa does + # not yet export that rname type (see the name-defined ignore on the + # INTFC class), so mypy only sees the ResourceName base here. + host = intfc.parsed.host_address # type: ignore[attr-defined] + try: + self.interface = nienet100.EnetConnection( + host, + open_timeout=connect_timeout_s, + timeout=self.timeout if self.timeout else connect_timeout_s, + ) + self.interface.open() + self.interface.open_gpib_session( + primary_address=pad, + secondary_address=sad, + tmo_code=nienet100.seconds_to_tmo_code(self.timeout) + if self.timeout + else gpib_constants.timeout.T10s, + ) + # Set the socket recv-timeout ceiling for the tmo_code just used; + # the code itself is already applied, so this does not reopen. + self._sync_read_timeout() + except Exception as e: + LOGGER.exception( + "Failed to open GPIB-ENET/100 session to %s pad=%d sad=%d", + host, + pad, + sad, + ) + self._cleanup_interface() + raise OpenError() from e + + self.attrs[ResourceAttribute.interface_number] = self.parsed.board + self.attrs[ResourceAttribute.gpib_primary_address] = pad + self.attrs[ResourceAttribute.gpib_secondary_address] = ( + (sad - 0x60) if sad else constants.VI_NO_SEC_ADDR + ) + self.attrs[ResourceAttribute.tcpip_address] = host + self.attrs[ResourceAttribute.tcpip_hostname] = host + self.attrs[ResourceAttribute.tcpip_port] = nienet100.PORT_MAIN + for name in ("SEND_END_EN", "TERMCHAR", "TERMCHAR_EN"): + attribute = getattr(constants, "VI_ATTR_" + name) + self.attrs[attribute] = attributes.AttributesByID[attribute].default + + def _cleanup_interface(self) -> None: + if self.interface is not None: + # close() handles bracket cleanup internally based on the + # connection's own _bracket_open flag — no need to gate the + # call here. + try: + self.interface.close() + except Exception as e: + LOGGER.debug("error closing GPIB-ENET/100 connection: %s", e) + self.interface = None + # Discard any buffered-but-unread response so it cannot leak into a + # subsequent session that reuses this object. + self._read_buffer.clear() + + def close(self) -> StatusCode: + self._cleanup_interface() + return StatusCode.success + + # --- I/O ------------------------------------------------------------ + + def write(self, data: bytes) -> tuple[int, StatusCode]: + if self.interface is None: + return 0, StatusCode.error_connection_lost + try: + written = self.interface.ibwrt(data) + except nienet100.NIEnet100IOError as e: + return 0, _map_iberr_to_status(e.err) + return written, StatusCode.success + + def read(self, count: int) -> tuple[bytes, StatusCode]: + if self.interface is None: + return b"", StatusCode.error_connection_lost + + # The wire-level ibrd always reads a whole message (until EOI/EOS). + # We cache it here and hand it out in <= count-byte slices, keeping + # any remainder for the next call — this is what lets a caller read + # a response one byte at a time without losing the tail. + if not self._read_buffer: + # The effective read timeout is the bracket's tmo_code, set from + # self.timeout in _set_timeout; this per-call tmo_ms has no effect. + # We still send a non-zero value because the genuine NI driver does + # and tmo_ms=0 means "no override" on the wire. + if self.timeout is None: + tmo_ms = nienet100.DEFAULT_IBRD_TMO_MS + else: + tmo_ms = max(int(self.timeout * 1000), 1) + try: + self._read_buffer.extend(self.interface.ibrd(tmo_ms=tmo_ms)) + except nienet100.NIEnet100IOError as e: + return b"", _map_iberr_to_status(e.err) + + # More remains buffered than requested: hand back exactly `count` + # bytes and keep the rest. Per VISA this is success_max_count_read. + if count < len(self._read_buffer): + chunk = bytes(self._read_buffer[:count]) + del self._read_buffer[:count] + return chunk, StatusCode.success_max_count_read + + # The caller's count covers the rest of the message; drain the + # buffer and report end-of-message (termination char or success). + chunk = bytes(self._read_buffer) + self._read_buffer.clear() + + term_char, _ = self.get_attribute(ResourceAttribute.termchar) + term_en, _ = self.get_attribute(ResourceAttribute.termchar_enabled) + if term_en and term_char is not None and chunk and chunk[-1] == term_char: + return chunk, StatusCode.success_termination_character_read + return chunk, StatusCode.success + + def clear(self) -> StatusCode: + if self.interface is None: + return StatusCode.error_connection_lost + # Drop any buffered-but-unread response: after a device clear the old + # tail is stale and must not leak into the next read(). + self._read_buffer.clear() + try: + self.interface.ibclr() + except nienet100.NIEnet100IOError as e: + return _map_iberr_to_status(e.err) + return StatusCode.success + + def assert_trigger(self, protocol: constants.TriggerProtocol) -> StatusCode: + if protocol != constants.VI_TRIG_PROT_DEFAULT: + return StatusCode.error_nonsupported_operation + if self.interface is None: + return StatusCode.error_connection_lost + try: + self.interface.ibtrg() + except nienet100.NIEnet100IOError as e: + return _map_iberr_to_status(e.err) + return StatusCode.success + + def read_stb(self) -> tuple[int, StatusCode]: + if self.interface is None: + return 0, StatusCode.error_connection_lost + try: + stb = self.interface.ibrsp() + except nienet100.NIEnet100IOError as e: + return 0, _map_iberr_to_status(e.err) + return stb, StatusCode.success + + def gpib_control_ren(self, mode: constants.RENLineOperation) -> StatusCode: + # ibloc covers VI_GPIB_REN_DEASSERT_GTL (Go-To-Local). The remaining + # six REN modes (assert/deassert REN, with optional address and/or + # LLO) need an ibsre verb that drives the REN line; that wire frame + # is not yet reverse-engineered, so they currently surface as + # error_nonsupported_operation (the documented pyvisa contract for + # backends that cannot honour a given REN mode). + if mode != constants.VI_GPIB_REN_DEASSERT_GTL: + return StatusCode.error_nonsupported_operation + if self.interface is None: + return StatusCode.error_connection_lost + try: + self.interface.ibloc() + except nienet100.NIEnet100IOError as e: + return _map_iberr_to_status(e.err) + return StatusCode.success + + # --- timeout plumbing ---------------------------------------------- + + def _set_timeout(self, attribute: ResourceAttribute, value: int) -> StatusCode: + status = super()._set_timeout(attribute, value) + self._sync_read_timeout() + return status + + def _sync_read_timeout(self) -> None: + """Push the current session timeout onto the wire. + + The read timeout is the discrete ``tmo_code`` carried in the open + Frame A; the per-call ibrd ``tmo_ms`` has no effect (verified against + hardware). The 'P 03' IbcTMO property is rejected once a bracket is + open, so a change is applied by reopening the bracket + (``EnetConnection.set_read_timeout_code``). The socket recv-timeout is + set a little above the box's own timeout so the box surfaces its EABO + first; ``seconds_to_tmo_code`` rounds *up*, so the ceiling is based on + the code's real duration (``TIMETABLE``) rather than ``self.timeout``. + + """ + if self.interface is None: + return + tmo_code = ( + nienet100.seconds_to_tmo_code(self.timeout) + if self.timeout + else gpib_constants.timeout.T10s + ) + self.interface.set_read_timeout_code(tmo_code) + box_timeout = nienet100.TIMETABLE[tmo_code] + self.interface.set_socket_timeout(box_timeout * 1.25 + 1.0) + + def _get_attribute(self, attribute: ResourceAttribute) -> tuple[Any, StatusCode]: + raise UnknownAttribute(attribute) + + def _set_attribute( + self, attribute: ResourceAttribute, attribute_state: Any + ) -> StatusCode: + raise UnknownAttribute(attribute) + + +def _map_iberr_to_status(iberr: int) -> StatusCode: + """Translate a wire-level iberr code into a pyvisa StatusCode.""" + if iberr == gpib_constants.error.EABO: + return StatusCode.error_timeout + if iberr == gpib_constants.error.ENOL: + return StatusCode.error_no_listeners + if iberr == gpib_constants.error.ECIC: + return StatusCode.error_not_cic + if iberr == gpib_constants.error.EARG: + return StatusCode.error_invalid_mode + if iberr == gpib_constants.error.ESAC: + return StatusCode.error_nonsupported_operation + return StatusCode.error_system_error diff --git a/pyvisa_py/protocols/nienet100.py b/pyvisa_py/protocols/nienet100.py new file mode 100644 index 00000000..6c21f222 --- /dev/null +++ b/pyvisa_py/protocols/nienet100.py @@ -0,0 +1,1070 @@ +# -*- coding: utf-8 -*- +"""Python implementation of the NI GPIB-ENET/100 wire protocol. + +This module talks the proprietary TCP protocol of the National Instruments +GPIB-ENET/100 Ethernet-to-GPIB bridge. It is **not** compatible with the +older GPIB-ENET (10 MBit/s, libnienet target), which uses a similar frame +layout but different verb opcodes and a single-step open. + +The wire format was reverse-engineered; the frame layout, verb opcodes and +status fields are documented inline alongside the constants and packers in +this module. + +All multi-byte fields are big-endian (network byte order). + +:copyright: 2026 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. + +""" + +import logging +import socket +import struct +from collections.abc import Callable +from typing import TYPE_CHECKING + +from .. import gpib_constants + +LOGGER = logging.getLogger("pyvisa_py.protocols.nienet100") + +#: Main TCP port (synchronous request/response). +PORT_MAIN = 5000 + +#: Companion socket TCP port (hello-only, mandatory for FW >= A8). Also the +#: asynchronous/SRQ event channel that ibwait polls. +PORT_COMPANION = 5015 + +#: Fixed length of every command frame sent to the box. +COMMAND_FRAME_SIZE = 12 + +#: Fixed length of every status header received from the box. +STATUS_HEADER_SIZE = 12 + +#: Fixed length of every payload chunk header in a read stream. +CHUNK_HEADER_SIZE = 4 + + +#: Discrete timeout values in seconds, indexed by TMO code. ``None`` = disabled. +TIMETABLE: tuple = ( + None, # infinite + 10e-6, + 30e-6, + 100e-6, + 300e-6, + 1e-3, + 3e-3, + 10e-3, + 30e-3, + 100e-3, + 300e-3, + 1.0, + 3.0, + 10.0, + 30.0, + 100.0, + 300.0, + 1000.0, +) + + +def seconds_to_tmo_code(timeout: float) -> int: + """Round a timeout (in seconds) up to the closest discrete TMO code. + + Values larger than ``TIMETABLE[-1]`` are clamped to the largest code. + ``None`` or ``0`` map to the disabled (infinite) code. + + """ + if not timeout: + return gpib_constants.timeout.TNONE + for code in range(1, len(TIMETABLE)): + if TIMETABLE[code] >= timeout * 0.999: + return code + return gpib_constants.timeout.T1000s + + +# --- Chunk header flags (read stream after a status header) ----------------- + +CHUNK_FLAG_DATA = 0 # data chunk; ``length`` bytes of payload follow +CHUNK_FLAG_END = 1 # END marker; ``length`` must be 0, read complete +CHUNK_FLAG_SIGNAL = 2 # out-of-band signal byte (1 byte follows), defensively skip + + +# --- 12-byte command-frame layout ------------------------------------------- +# Byte: 0 1 2-3 4-5 6-7 8-11 +# +-----+-----+--------+---------+--------+---------+ +# | id | b1 | ushort | ushort | ushort | ulong | 12 B, big-endian +# +-----+-----+--------+---------+--------+---------+ + +_COMMAND_FRAME_FMT = "!BBHHHL" + + +def pack_command( + cmd_id: int, + b1: int = 0, + w1: int = 0, + w2: int = 0, + w3: int = 0, + dw: int = 0, +) -> bytes: + """Build a 12-byte command frame. + + All fields default to 0. Unused fields **must** stay zero — the box + accepts non-zeroed buffers only inconsistently. + + """ + return struct.pack(_COMMAND_FRAME_FMT, cmd_id, b1, w1, w2, w3, dw) + + +# --- 12-byte status-header layout ------------------------------------------- +# Byte: 0-1 2-3 4-7 8-11 +# +-----+-----+-----------+---------+ +# | sta | err | 4 padding | count | 12 B, big-endian +# +-----+-----+-----------+---------+ + +_STATUS_HEADER_FMT = "!HH4xL" + + +def parse_status_header(buf: bytes) -> tuple[int, int, int]: + """Decode a 12-byte status header into ``(sta, err, cnt)``. + + ``err`` is only meaningful when the ERR bit is set; otherwise it may + carry sentinel values such as 0xFFFF that the caller must ignore. + + """ + if len(buf) != STATUS_HEADER_SIZE: + raise ValueError( + "status header must be exactly %d bytes, got %d" + % (STATUS_HEADER_SIZE, len(buf)) + ) + return struct.unpack(_STATUS_HEADER_FMT, buf) + + +def parse_chunk_header(buf: bytes) -> tuple[int, int]: + """Decode a 4-byte chunk header into ``(flags, length)``.""" + if len(buf) != CHUNK_HEADER_SIZE: + raise ValueError( + "chunk header must be exactly %d bytes, got %d" + % (CHUNK_HEADER_SIZE, len(buf)) + ) + return struct.unpack("!HH", buf) + + +# --- Chunk reader ----------------------------------------------------------- +# Every response from the bridge is framed as a sequence of chunks: each chunk +# is a 4-byte header (flags, length) followed by ``length`` bytes of data. +# Status headers are themselves data chunks with length=12; payload bytes for +# ibrd / ibrsp are additional data chunks; the end-of-stream marker is its +# own chunk (flags=1, length=0). The chunk stream is stateful: chunks may be +# split across TCP segments and several chunks may arrive in one segment. +# Callers drive recv via the ``read_exactly`` callable so this layer is +# socket-agnostic and testable. + + +def read_chunks_until_end(read_exactly: Callable[[int], bytes]) -> bytes: + """Consume a chunk stream until the END marker (flags=1). + + Tolerates out-of-band signal chunks (flags=2) by reading and discarding + their single payload byte. Unknown flag values with ``length==0`` are + treated as end-of-stream terminators with a warning — hardware has + been observed to use flag 0x0004 on timeouts (and other terminal + conditions the wire spec does not enumerate). The caller's + subsequent status-header read carries the real outcome (e.g. the ERR + bit + iberr=EABO for a timeout). Unknown flags carrying a non-zero + length still raise :class:`NIEnet100ProtocolError` because we cannot + stay frame-aligned without knowing how to consume the data. + + Parameters + ---------- + read_exactly : Callable[[int], bytes] + Reader returning exactly ``n`` bytes or raising on short read / + timeout. Pass a bound socket helper here. + + Returns + ------- + bytes + Concatenated payload of all data chunks (END chunk excluded). + + """ + payload = bytearray() + while True: + flags, length = parse_chunk_header(read_exactly(CHUNK_HEADER_SIZE)) + if flags == CHUNK_FLAG_DATA: + if length: + payload.extend(read_exactly(length)) + elif flags == CHUNK_FLAG_END: + if length != 0: + raise NIEnet100ProtocolError( + "END chunk has non-zero length %d" % length + ) + return bytes(payload) + elif flags == CHUNK_FLAG_SIGNAL: + # Defensive: per spec, a signal chunk carries exactly 1 OOB byte + # which we log and skip. Never observed in practice. + signal_byte = read_exactly(1) + LOGGER.debug("NI-ENET/100 signal byte received: 0x%02x", signal_byte[0]) + elif length == 0: + # Undocumented zero-length flag — treat as a terminal marker so + # the caller's status-header read can report the real outcome + # instead of us crashing on a flag we have not characterized. + LOGGER.warning( + "treating unknown chunk flag 0x%04x (length=0) as end-of-stream", + flags, + ) + return bytes(payload) + else: + raise NIEnet100ProtocolError( + "unknown chunk flag 0x%04x with non-zero length %d " + "(cannot stay aligned, aborting)" % (flags, length) + ) + + +def read_one_data_chunk(read_exactly: Callable[[int], bytes]) -> bytes: + """Read exactly one data chunk and return its payload. + + Use this for verbs whose response carries a single fixed-size data chunk + and may omit the END marker (e.g. ``ibrsp`` returns a single STB byte). + Signal chunks (flags=2) are still tolerated. + + """ + while True: + flags, length = parse_chunk_header(read_exactly(CHUNK_HEADER_SIZE)) + if flags == CHUNK_FLAG_DATA: + return read_exactly(length) if length else b"" + elif flags == CHUNK_FLAG_SIGNAL: + signal_byte = read_exactly(1) + LOGGER.debug("NI-ENET/100 signal byte received: 0x%02x", signal_byte[0]) + else: + raise NIEnet100ProtocolError( + "expected data chunk, got flags=0x%04x length=%d" % (flags, length) + ) + + +def read_status_chunk(read_exactly: Callable[[int], bytes]) -> tuple[int, int, int]: + """Read a chunk-wrapped 12-byte status header and parse it. + + The bridge wraps every status header in the standard chunk framing + (``[flags=0, length=12][12B body]``) — there is no such thing as a + "raw" 12-byte status read on the wire. Callers must use this helper + rather than reading 12 bytes directly, or successive operations will + accumulate a 4-byte misalignment per status read. + + """ + body = read_one_data_chunk(read_exactly) + if len(body) != STATUS_HEADER_SIZE: + raise NIEnet100ProtocolError( + "expected %d-byte status chunk body, got %d bytes" + % (STATUS_HEADER_SIZE, len(body)) + ) + return parse_status_header(body) + + +class NIEnet100Error(Exception): + """Base exception for NI GPIB-ENET/100 protocol errors.""" + + +class NIEnet100ProtocolError(NIEnet100Error): + """The peer sent something we cannot parse (bad magic, bad chunk flag).""" + + +class NIEnet100IOError(NIEnet100Error): + """The box returned a status header with the ERR bit set. + + Attributes + ---------- + sta : int + Raw ``ibsta`` bitmask from the status header. + err : int + Raw ``iberr`` code from the status header. + + """ + + def __init__(self, sta: int, err: int, operation: str = ""): + self.sta = sta + self.err = err + msg = "NI-ENET/100 operation %r failed: sta=0x%04x err=%d" % ( + operation, + sta, + err, + ) + super().__init__(msg) + + +# --- Connection ------------------------------------------------------------- +# The box uses parallel TCP sockets per session. The main socket (5000) +# carries all synchronous Device-I/O. The companion socket (5015) is +# mandatory on every firmware shipped in the last ~20 years; it carries the +# hello frame, must stay open for the session lifetime, and doubles as the +# async/SRQ event channel that ibwait polls. No other sockets are used — +# every verb (including the board-level ibsic) runs on main or companion. + + +def _u32_from_ip(ip: str) -> int: + """Convert dotted-quad IP to a 32-bit integer in host order. + + The result is meant to be re-emitted via ``struct.pack('!L', ...)``, + which puts the high byte first on the wire — matching the box's + convention (e.g. 192.0.2.5 -> ``c0 00 02 05``). + + """ + return int.from_bytes(socket.inet_aton(ip), "big") + + +class EnetConnection: + """Synchronous TCP transport to a single GPIB-ENET/100 box. + + :meth:`open` connects the main socket (port 5000) and the companion + socket (port 5015) and sends the mandatory companion hello frame. The + companion socket doubles as the asynchronous/SRQ event channel: ibwait + polls it (see :meth:`ibwait`). No other sockets are used. + + The class is **not** thread-safe. Concurrent calls into a single + instance (e.g. one thread issuing ibwrt while another polls ibwait) + will interleave bytes on the sockets and corrupt the protocol state. + + Parameters + ---------- + host : str + Box IP or hostname. + open_timeout : float + Per-socket connect timeout in seconds. + timeout : float | None + Per-operation socket timeout in seconds applied after connect. + ``None`` means blocking without timeout. + + Attributes + ---------- + host : str + The host string passed at construction time. + main : socket.socket + The synchronous main socket. + companion : socket.socket + The companion socket; kept open for the session lifetime and used as + the async/SRQ event channel that :meth:`ibwait` polls. + + """ + + if TYPE_CHECKING: + # GPIB verbs are implemented as module-level functions and bound onto + # the class at the end of this file. Declare their signatures here so + # type checkers see the public ``conn.ibXXX`` API; the runtime + # implementations come from the assignments below. + def ibwrt(self, data: bytes) -> int: ... + + def ibrd(self, tmo_ms: int = ...) -> bytes: ... + + def ibclr(self) -> None: ... + + def ibtrg(self) -> None: ... + + def ibloc(self) -> None: ... + + def ibrsp(self) -> int: ... + + def ibwait(self, mask: int) -> int: ... + + def ibsic(self) -> None: ... + + def set_io_timeout(self, tmo_code: int) -> None: ... + + def set_read_timeout_code(self, tmo_code: int) -> None: ... + + def transact_main_status( + self, operation: str = ... + ) -> tuple[int, int, int]: ... + + #: Companion-hello flag word for device-mode sessions (single resource). + COMPANION_FLAGS_DEVICE = 2 + + #: Companion-hello flag word for board-mode sessions (INTFC). The genuine + #: NI board open sends 0 here, vs :data:`COMPANION_FLAGS_DEVICE` for a + #: device session. + COMPANION_FLAGS_BOARD = 0 + + def __init__( + self, + host: str, + open_timeout: float = 10.0, + timeout: float | None = 10.0, + ) -> None: + self.host = host + self._open_timeout = open_timeout + self._timeout = timeout + self.main: socket.socket | None = None + self.companion: socket.socket | None = None + # Tracks whether a Frame F bracket-open has been acked by the box + # without a matching Frame X close yet. Owned by _transact_bracket + # so failures between bracket-open and the session-layer marker + # (e.g., Frame G of open_gpib_session) still trigger a bracket + # close on the way out — otherwise the bridge leaks a session + # slot per such failure. + self._bracket_open: bool = False + # Target address and read-timeout code of the current device bracket. + # Remembered so the read timeout can be changed mid-session by + # reopening the bracket with a fresh Frame A (the box rejects the + # 'P 03' IbcTMO setter while a bracket is open). ``None`` until an + # open_gpib_session has run. + self._pad: int = 0 + self._sad: int = 0 + self._read_tmo_code: int | None = None + + # --- lifecycle ------------------------------------------------------ + + def open(self) -> None: + """Open main and companion sockets and send the companion hello.""" + self.main = self._connect(PORT_MAIN) + try: + self.companion = self._connect(PORT_COMPANION) + except Exception: + self.main.close() + self.main = None + raise + try: + self._send_companion_hello() + except Exception: + self.close() + raise + + def close(self) -> None: + """Close every open socket. Idempotent. + + If a Frame F bracket is currently open on the box, best-effort + sends the matching ``X 00 01`` bracket-close before tearing the + sockets down — otherwise the bridge leaks the session slot. This + runs unconditionally so error paths in higher layers cannot skip + it. Errors during cleanup are logged and swallowed so socket + teardown always runs. + + """ + if self._bracket_open and self.main is not None: + try: + self.close_gpib_session() + except Exception as e: + LOGGER.debug("bracket close during teardown failed: %s", e) + + # Close in reverse open-order so the box sees the companion socket + # disappear before main. The box does not require a goodbye frame. + for attr in ("companion", "main"): + sock = getattr(self, attr, None) + if sock is not None: + try: + sock.close() + except OSError as e: + LOGGER.debug("error closing %s socket: %s", attr, e) + setattr(self, attr, None) + + def set_socket_timeout(self, timeout: float | None) -> None: + """Apply ``timeout`` (in seconds) to all currently open sockets. + + Use ``None`` for blocking without timeout. The value is cached so a + socket opened later picks up the same setting. + + """ + self._timeout = timeout + for sock in (self.main, self.companion): + if sock is not None: + sock.settimeout(timeout) + + # --- low-level helpers --------------------------------------------- + + def _connect(self, port: int) -> socket.socket: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(self._open_timeout) + try: + sock.connect((self.host, port)) + except Exception: + sock.close() + raise + sock.settimeout(self._timeout) + return sock + + @staticmethod + def _recv_exactly(sock: socket.socket, n: int) -> bytes: + """Read exactly ``n`` bytes from ``sock`` or raise.""" + buf = bytearray() + while len(buf) < n: + chunk = sock.recv(n - len(buf)) + if not chunk: + raise NIEnet100Error( + "connection closed by peer after %d/%d bytes" % (len(buf), n) + ) + buf.extend(chunk) + return bytes(buf) + + def recv_main_exactly(self, n: int) -> bytes: + """Read exactly ``n`` bytes from the main socket or raise. + + At DEBUG log level the bytes received are hex-dumped — invaluable + for diagnosing wire-protocol surprises against real hardware. Set + ``--log-cli-level=DEBUG`` on pytest to see the dumps, or attach a + handler to ``pyvisa_py.protocols.nienet100`` in your own code. + + """ + if self.main is None: + raise NIEnet100Error("main socket is not open") + data = self._recv_exactly(self.main, n) + LOGGER.debug("← main: %s", data.hex()) + return data + + def send_main(self, data: bytes) -> None: + """Send ``data`` on the main socket in a single ``sendall``. + + At DEBUG log level the bytes sent are hex-dumped — see + :meth:`recv_main_exactly` for details. + + """ + if self.main is None: + raise NIEnet100Error("main socket is not open") + LOGGER.debug("→ main: %s", data.hex()) + self.main.sendall(data) + + def read_status_main(self) -> tuple[int, int, int]: + """Read a chunk-wrapped status header from the main socket and parse it.""" + return read_status_chunk(self.recv_main_exactly) + + def transact_main(self, frame: bytes, operation: str = "") -> tuple[int, int, int]: + """Send a command frame and read the status header on the main socket. + + Raises :class:`NIEnet100IOError` if the status header has the ERR + bit set. Returns ``(sta, err, cnt)`` on success. + + """ + self.send_main(frame) + sta, err, cnt = self.read_status_main() + if sta & gpib_constants.status.ERR: + raise NIEnet100IOError(sta, err, operation) + return sta, err, cnt + + # --- companion socket ---------------------------------------------- + + def _send_companion_hello(self, flags: int = COMPANION_FLAGS_DEVICE) -> None: + """Send the 'U 02' hello on the companion socket and read the status. + + Sub-op layout: ``55 02 [htons(flags)] 00 00 [htons(port)] [ip:4]``. + ``port``/``ip`` are the **main** socket's ``getsockname()`` — the box + uses them to link this companion (the async/SRQ event channel) back to + the main session, so an ibwait poll on the companion is answered. The + genuine NI software reports the main socket's address here too; + reporting the companion's own port (as earlier revisions did) leaves + the event channel unlinked and ibwait never returns. + + ``flags`` is :data:`COMPANION_FLAGS_DEVICE` for a device (INSTR) + session and :data:`COMPANION_FLAGS_BOARD` for a board (INTFC) session + — the genuine NI software uses different values for each. + + """ + if self.companion is None: + raise NIEnet100Error("companion socket is not open") + if self.main is None: + raise NIEnet100Error("main socket is not open") + main_ip, main_port = self.main.getsockname() + frame = pack_command( + cmd_id=0x55, # 'U' + b1=0x02, + w1=flags, + w2=0, + w3=main_port, + dw=_u32_from_ip(main_ip), + ) + self.companion.sendall(frame) + companion = self.companion + sta, err, _cnt = read_status_chunk(lambda n: self._recv_exactly(companion, n)) + if sta & gpib_constants.status.ERR: + raise NIEnet100IOError(sta, err, "companion hello") + + # --- GPIB-session open / close (Frames A-G of the spec) ----------- + + #: Default Frame-C board flags. Sets HS488 marker + an EOI/EOS bit and + #: leaves everything else off — the conservative baseline for a generic + #: instrument session. + DEFAULT_BOARD_FLAGS = 0x1801 + + #: Default Frame-E event-queue depth. + DEFAULT_EVENT_QUEUE_DEPTH = 0x0B + + def open_board_session( + self, + board_flags: int = DEFAULT_BOARD_FLAGS, + tmo_code: int = gpib_constants.timeout.T10s, + ) -> None: + """Open a board-level (INTFC) session and leave the box online. + + The board counterpart to :meth:`open` + :meth:`open_gpib_session` + (which open a *device* session). This connects the sockets itself and + replays the genuine NI board-open, leaving the box online and + Controller-In-Charge-capable — the state in which it accepts a + board-level :meth:`ibsic` (Interface Clear). A bare :meth:`open` + (companion hello only) is **not** enough. + + Unlike a device session this opens no operation bracket and arms no + async-notify. Wire sequence (main socket unless noted):: + + 0b ............................ identify (reply: firmware banner) + 50 05 00 ...................... Mode/PPC = 0 + 07 00 [flags] 00 00 00 00 [tmo] SetConfig (non-SC): flags + timeout + 50 10 01 ...................... Online = 1 + 55 02 .. (companion socket) ... board companion hello (flags=0) + 12 01 ......................... board verb 0x12 + 06 1f 0f00 .................... board verb 0x06 + + The ``0x12`` / ``0x06`` verbs are not yet decoded; they are replayed + byte-exact from the genuine NI sequence because the box wedges on an + ibsic issued without them. + + """ + self.main = self._connect(PORT_MAIN) + try: + # identify: the reply is a single data chunk (firmware banner), not + # a status chunk; read and discard it. + self.send_main(pack_command(0x0B)) + read_one_data_chunk(self.recv_main_exactly) + self.transact_main(_pack_property_set(0x05, 0x00), "board open Mode") + self.transact_main( + pack_command(0x07, 0x00, w1=board_flags, dw=tmo_code << 24), + "board open SetConfig", + ) + self.transact_main(_pack_property_set(0x10, 0x01), "board open Online") + self.companion = self._connect(PORT_COMPANION) + self._send_companion_hello(flags=self.COMPANION_FLAGS_BOARD) + self.transact_main(pack_command(0x12, 0x01), "board open verb 0x12") + self.transact_main( + pack_command(0x06, 0x1F, w1=0x0F00), "board open verb 0x06" + ) + except Exception: + self.close() + raise + + def open_gpib_session( + self, + primary_address: int, + secondary_address: int = 0, + tmo_code: int = gpib_constants.timeout.T10s, + board_flags: int = DEFAULT_BOARD_FLAGS, + event_queue_depth: int = DEFAULT_EVENT_QUEUE_DEPTH, + mode_byte: int = 0, + ) -> None: + """Run the seven-frame open sequence on the main socket. + + After ``open()`` (which establishes main + companion sockets and + sends the companion hello), this method makes the bus ready for + Device-I/O against the given primary/secondary address. The bracket + opened by Frame F stays open until :meth:`close_gpib_session`. + + Parameters + ---------- + primary_address : int + Target GPIB primary address (0-30). + secondary_address : int + Target GPIB secondary address (0 means none). + tmo_code : int + NI-488.2 timeout code (see ``TIMETABLE``). Default is the 10 s + code, matching NI's measurement-equipment default. + board_flags : int + Frame-C bitmask. Default ``0x1801`` is the standard + single-instrument baseline. + event_queue_depth : int + Frame-E event-queue depth. Default ``0x0b`` (= 11). + mode_byte : int + Frame-B mode byte. ``0`` is standard. + + """ + # Remember the target and timeout so a later timeout change can + # rebuild Frame A for the bracket reopen (see set_read_timeout_code). + self._pad = primary_address + self._sad = secondary_address + self._read_tmo_code = tmo_code + + # Frame A: SetConfig with SC bit and target address. + # Wire bytes: 07 02 00 01 [PAD] [SAD] 00 00 [tmo] 00 04 00 + frame_a = pack_command( + cmd_id=0x07, + b1=0x02, + w1=0x0001, + w2=(primary_address << 8) | (secondary_address & 0xFF), + w3=0, + dw=(tmo_code << 24) | 0x0400, + ) + self.transact_main(frame_a, "open Frame A SetConfig SC") + + # Frame B: Property 'Mode' (PPC, idx 0x05). + # Wire bytes: 50 05 [mode_byte] 00*9 + self.transact_main(_pack_property_set(0x05, mode_byte), "open Frame B PPC") + + # Frame C: SetConfig (non-SC variant) with board flags. + # Wire bytes: 07 00 [htons(flags)] 00 00 00 00 [tmo] 00 00 00 + frame_c = pack_command( + cmd_id=0x07, + b1=0x00, + w1=board_flags, + w2=0, + w3=0, + dw=tmo_code << 24, + ) + self.transact_main(frame_c, "open Frame C SetConfig non-SC") + + # Frame D: Property 'Online' (PP2, idx 0x10) with value 1. + # Wire bytes: 50 10 01 00*9 + self.transact_main(_pack_property_set(0x10, 0x01), "open Frame D online") + + # Frame E: Property 'Event-Queue depth' (idx 0x15). + # Wire bytes: 50 15 [depth] 00*9 + self.transact_main( + _pack_property_set(0x15, event_queue_depth), + "open Frame E event-queue depth", + ) + + # Frame F: Operation-bracket open ('X', idx 0x58). + # Wire bytes: 58 01 01 00*9 + self._transact_bracket(enter=True) + + # Frame G: arm the async/SRQ notify channel ('N 01', idx 0x4e) on the + # main socket. This is what lets a later ibwait poll on the companion + # socket return SRQ events for this session. + # Wire bytes: 4e 01 00*10 + self.transact_main(pack_command(0x4E, 0x01), "open Frame G async-notify arm") + + def close_gpib_session(self) -> None: + """Close the operation bracket opened by :meth:`open_gpib_session`. + + Sockets are not closed here — call :meth:`close` for that. No-op + when no bracket is currently open, so callers can invoke this on + any cleanup path without first probing state. + + """ + if not self._bracket_open or self.main is None: + return + try: + self._transact_bracket(enter=False) + except (NIEnet100Error, OSError) as e: + LOGGER.debug("error closing GPIB bracket: %s", e) + # Clear the flag even when the wire transact failed, so the + # subsequent close() does not re-attempt on a wedged socket. + self._bracket_open = False + + def _transact_bracket(self, enter: bool) -> None: + # Wire bytes: 58 [01|00] 01 00*9 + frame = struct.pack("!BBB9x", 0x58, 0x01 if enter else 0x00, 0x01) + self.transact_main(frame, "bracket %s" % ("open" if enter else "close")) + # Flip the flag only after the box acked the frame, so a failing + # open does not leave us thinking we owe a close, and vice-versa. + self._bracket_open = enter + + +def _pack_property_set(prop_idx: int, value_byte: int) -> bytes: + """Build a 'P' property-set frame (0x50). + + Wire layout: ``50 [prop_idx] [value_byte] 00*9``. + + """ + return struct.pack("!BBB9x", 0x50, prop_idx, value_byte) + + +# --- Device-level verbs ----------------------------------------------------- +# These methods are added to EnetConnection via assignment below. They cover +# the pyvisa-Resource API surface: write, read, clear, trigger, serial poll, +# local-lockout release, the I/O timeout setter, ibwait (SRQ poll on the +# companion socket), and the board-level ibsic on the main socket. + + +def _ibwrt(self: EnetConnection, data: bytes) -> int: + """Write ``data`` to the addressed device. + + Wire layout: ``62 00 00 00 [htonl(byte_count):4] 00 00 00 00`` followed + immediately by the raw payload in the same ``sendall``. The payload is + sent unpadded, exactly ``byte_count`` bytes. + + Returns the number of bytes the box reports as transferred. + + """ + byte_count = len(data) + # Frame: 62 00 00 00 [htonl(byte_count):4] 00 00 00 00 + header = struct.pack("!BBHL4x", 0x62, 0x00, 0x0000, byte_count) + self.send_main(header + data) + _sta, _err, cnt = self.transact_main_status("ibwrt") + return cnt + + +#: Default per-call read timeout in milliseconds when the caller passes +#: nothing. ``tmo_ms=0`` on the wire is **not** "use the session default" +#: — the bridge interprets it as "do not wait" and returns immediately +#: with cnt=0 and no END marker. 10 s matches NI's default IbcTMO for +#: measurement equipment and is the conservative choice for callers that +#: do not know better. +DEFAULT_IBRD_TMO_MS = 10_000 + + +def _ibrd(self: EnetConnection, tmo_ms: int = DEFAULT_IBRD_TMO_MS) -> bytes: + """Read one message from the addressed device. + + The wire-level read pulls bytes until the device signals end-of-message + (EOI/EOS) or the bridge's per-read timeout fires. The box does not + take a maximum-byte argument — callers that want to truncate must do + so after the fact. + + Wire layout: ``16 00 00 00 [htonl(tmo_ms):4] 00 00 00 00``. ``tmo_ms`` + is the bridge-side timeout for waiting on device data — **not** a + session default fallback. ``0`` makes the bridge return immediately + with no data; pass a positive value (the default is 10 s) to give + the device time to respond. + + Response shape depends on whether the device returned data within + the timeout: + + - **With data**: preliminary status chunk, then one + or more data chunks (each a chunk-wrapped block of device bytes), + then an END marker (flags=1 length=0), then the final status + chunk whose ``cnt`` equals the total payload length. + - **Without data** (timeout or no response): preliminary status + chunk, then the final status chunk directly — no END marker, no + intermediate data chunks. The parser distinguishes the two paths + by inspecting the body of each candidate-data chunk: a 12-byte + chunk whose body parses as a status header with CMPL/ERR/END/TIMO + bits set is the final status, not data. + + """ + # Frame: 16 00 00 00 [htonl(tmo_ms):4] 00 00 00 00 + frame = struct.pack("!BBHL4x", 0x16, 0x00, 0x0000, tmo_ms) + self.send_main(frame) + # Preliminary status (typically sta=0x0100 cnt=0, err may be 0xFFFF). + sta_p, err_p, _ = self.read_status_main() + if sta_p & gpib_constants.status.ERR: + raise NIEnet100IOError(sta_p, err_p, "ibrd preliminary") + + payload = bytearray() + _status_bits = ( + gpib_constants.status.CMPL + | gpib_constants.status.ERR + | gpib_constants.status.END + | gpib_constants.status.TIMO + ) + while True: + flags, length = parse_chunk_header(self.recv_main_exactly(CHUNK_HEADER_SIZE)) + + if flags == CHUNK_FLAG_END: + # Spec path: END marker, followed by the final status chunk. + if length != 0: + raise NIEnet100ProtocolError( + "END chunk has non-zero length %d" % length + ) + sta_f, err_f, _cnt = self.read_status_main() + if sta_f & gpib_constants.status.ERR: + raise NIEnet100IOError(sta_f, err_f, "ibrd final") + return bytes(payload) + + if flags == CHUNK_FLAG_SIGNAL: + self.recv_main_exactly(1) + continue + + if flags != CHUNK_FLAG_DATA: + if length == 0: + LOGGER.warning( + "treating unknown chunk flag 0x%04x (length=0) as end-of-stream", + flags, + ) + return bytes(payload) + raise NIEnet100ProtocolError( + "unknown chunk flag 0x%04x with non-zero length %d" % (flags, length) + ) + + body = self.recv_main_exactly(length) if length else b"" + + # No-data path: the bridge sends the final status as a length-12 + # data chunk without a preceding END marker. Detect by parsing + # the body as a status header and checking for CMPL/ERR/END/TIMO + # bits. Real device data of exactly 12 bytes whose first u16 is + # one of those status values is in principle ambiguous, but the + # leading 0x0100/0x8100/etc. patterns are rare enough in raw + # instrument data that this heuristic is reliable in practice. + if length == STATUS_HEADER_SIZE: + sta_c, err_c, _cnt_c = parse_status_header(body) + if sta_c & _status_bits: + if sta_c & gpib_constants.status.ERR: + raise NIEnet100IOError(sta_c, err_c, "ibrd final") + return bytes(payload) + + payload.extend(body) + + +def _ibclr(self: EnetConnection) -> None: + """Clear the addressed device. + + Wire layout: ``04 00*11``. + + """ + self.transact_main(pack_command(0x04), "ibclr") + + +def _ibtrg(self: EnetConnection) -> None: + """Assert the device trigger on the addressed device. + + Wire layout: ``20 00*11``. + + """ + self.transact_main(pack_command(0x20), "ibtrg") + + +def _ibloc(self: EnetConnection) -> None: + """Send go-to-local to the addressed device. + + Wire layout: ``10 00*11``. + + """ + self.transact_main(pack_command(0x10), "ibloc") + + +def _ibrsp(self: EnetConnection) -> int: + """Serial-poll the addressed device and return the status byte (STB). + + Wire layout (request): ``19 00*11``. The response is **one** data + chunk whose length is 13: the first 12 bytes are the standard status + header and the trailing byte is the STB. The STB is always that trailing + byte; the header's ``cnt`` field is not reliably 1 — a serial poll taken + right after an SRQ wait reports ``cnt=0`` with a perfectly valid STB — + so the STB is read by position, not gated on ``cnt``. + + """ + self.send_main(pack_command(0x19)) + chunk = read_one_data_chunk(self.recv_main_exactly) + if len(chunk) < STATUS_HEADER_SIZE + 1: + raise NIEnet100ProtocolError( + "ibrsp chunk has %d bytes, expected at least %d" + % (len(chunk), STATUS_HEADER_SIZE + 1) + ) + sta, err, _cnt = parse_status_header(chunk[:STATUS_HEADER_SIZE]) + if sta & gpib_constants.status.ERR: + raise NIEnet100IOError(sta, err, "ibrsp") + return chunk[STATUS_HEADER_SIZE] + + +def _set_io_timeout(self: EnetConnection, tmo_code: int) -> None: + """Set the wire-level I/O timeout via the IbcTMO property (idx 0x03). + + ``tmo_code`` is a discrete NI-488.2 timeout index, not milliseconds — + use :func:`seconds_to_tmo_code` to convert. Only valid *before* the + operation bracket is open: the box rejects this property with EARG + once a bracket is open. Use :meth:`set_read_timeout_code` to change + the timeout mid-session. + + """ + self.transact_main(_pack_property_set(0x03, tmo_code), "set IbcTMO") + + +def _set_read_timeout_code(self: EnetConnection, tmo_code: int) -> None: + """Change the device read timeout to ``tmo_code`` mid-session. + + The read timeout is the discrete NI-488.2 code carried in the open + Frame A. It cannot be changed with the 'P 03' IbcTMO property while a + bracket is open (the box answers EARG), so a change is applied by closing + the bracket, re-sending Frame A (SetConfig SC) with the same target + address and the new ``tmo_code``, and reopening the bracket — the wire + sequence the genuine driver uses for a target-address switch. + + No-op when the code is unchanged. When no bracket is open the new code is + just recorded and takes effect at the next :meth:`open_gpib_session`. + + """ + if tmo_code == self._read_tmo_code: + return + if not self._bracket_open or self.main is None: + self._read_tmo_code = tmo_code + return + self._transact_bracket(enter=False) + frame_a = pack_command( + cmd_id=0x07, + b1=0x02, + w1=0x0001, + w2=(self._pad << 8) | (self._sad & 0xFF), + w3=0, + dw=(tmo_code << 24) | 0x0400, + ) + self.transact_main(frame_a, "read-timeout change Frame A SetConfig SC") + self._transact_bracket(enter=True) + self._read_tmo_code = tmo_code + + +def _transact_main_status( + self: EnetConnection, operation: str = "" +) -> tuple[int, int, int]: + """Read a status header on the main socket and raise on error. + + Sibling of :meth:`EnetConnection.transact_main` for verbs that have + already sent their frame (and any payload) via :meth:`send_main`. + + """ + sta, err, cnt = self.read_status_main() + if sta & gpib_constants.status.ERR: + raise NIEnet100IOError(sta, err, operation) + return sta, err, cnt + + +def _ibsic(self: EnetConnection) -> None: + """Pulse the GPIB IFC (Interface Clear) line on the bridge. + + Send-Interface-Clear is a bare ``0x1c`` command frame on the **main** + socket — no sub-op and no address payload. Verified against the genuine + NI software. The box replies with a status whose ``sta`` is ``CMPL|CIC|ATN`` + (the bridge becomes Controller-In-Charge with ATN asserted after the IFC). + Wire layout:: + + 1c 00 00 00 00 00 00 00 00 00 00 00 + + """ + self.transact_main(pack_command(0x1C), "ibsic") + + +def _ibwait(self: EnetConnection, mask: int) -> int: + """Issue one ibwait round-trip on the companion socket and return ``sta``. + + Sends a single ``0x22`` poll frame carrying ``mask`` (a 16-bit ibsta + bitmask of the events the caller is interested in — typically + the RQS bit for SRQ, OR'd with the TIMO bit so the box's own IbcTMO + surfaces as a timeout event). The box **blocks** and replies with a + 12-byte status header whose ``sta`` is the result: + + sta = conn.ibwait(gpib_constants.status.RQS | gpib_constants.status.TIMO) + if sta & gpib_constants.status.RQS: + stb = conn.ibrsp() # acknowledge RQS + elif sta & gpib_constants.status.TIMO: + ... # no SRQ within IbcTMO + + The poll goes on the companion socket (the box's event channel, linked + to this session by the companion hello). The async arm was already sent + as Frame G ('N 01') of :meth:`open_gpib_session`, so no extra setup is + needed here. + + Wire layout: ``22 00 [htons(mask):2] 00*8``. + + .. note:: + + With no matching event the box only answers after its IbcTMO, so the + companion socket timeout must exceed the session timeout — abandoning + a still-pending poll wedges the bridge. Callers should size the + socket timeout above the wire timeout (see + :meth:`set_socket_timeout`). + + """ + if self.companion is None: + raise NIEnet100Error("companion socket is not open") + self.companion.sendall(pack_command(cmd_id=0x22, b1=0x00, w1=mask)) + companion = self.companion + sta, err, _cnt = read_status_chunk(lambda n: self._recv_exactly(companion, n)) + if sta & gpib_constants.status.ERR: + raise NIEnet100IOError(sta, err, "ibwait") + return sta + + +# Attach verbs to EnetConnection. Keeping them as module-level functions +# makes the wire-bytes-per-verb mapping straightforward to read in this +# file; binding them here gives users the familiar `conn.ibwrt(...)` API. +# The signatures are declared under TYPE_CHECKING in the class body, so the +# assignments below need to silence mypy's method-reassignment guard. +EnetConnection.ibwrt = _ibwrt # type: ignore[method-assign] +EnetConnection.ibrd = _ibrd # type: ignore[method-assign] +EnetConnection.ibclr = _ibclr # type: ignore[method-assign] +EnetConnection.ibtrg = _ibtrg # type: ignore[method-assign] +EnetConnection.ibloc = _ibloc # type: ignore[method-assign] +EnetConnection.ibrsp = _ibrsp # type: ignore[method-assign] +EnetConnection.ibwait = _ibwait # type: ignore[method-assign] +EnetConnection.ibsic = _ibsic # type: ignore[method-assign] +EnetConnection.set_io_timeout = _set_io_timeout # type: ignore[method-assign] +EnetConnection.set_read_timeout_code = _set_read_timeout_code # type: ignore[method-assign] +EnetConnection.transact_main_status = _transact_main_status # type: ignore[method-assign] diff --git a/pyvisa_py/protocols/nienet100_discovery.py b/pyvisa_py/protocols/nienet100_discovery.py new file mode 100644 index 00000000..b87c7cea --- /dev/null +++ b/pyvisa_py/protocols/nienet100_discovery.py @@ -0,0 +1,296 @@ +# -*- coding: utf-8 -*- +"""NI GPIB-ENET/100 UDP discovery protocol. + +A small UDP protocol exposes the bridge's IP, MAC, serial number, +hostname, subnet, and gateway without requiring a TCP session — useful +for populating ``list_resources()`` and for finding a freshly +factory-reset box that does not yet have a known IP. Default port is +**44515** (broadcast); the **44516** unicast variant works across +subnets when the box IP is known in advance. All frames are exactly +184 bytes wrapped in an ``ED ... NI`` magic sandwich. + +This module only handles frame encoding/decoding; the UDP socket loop +lives in :func:`discover` (added in a later commit). + +:copyright: 2026 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. + +""" + +import logging +import socket +import struct +import time +from dataclasses import dataclass + +LOGGER = logging.getLogger("pyvisa_py.protocols.nienet100_discovery") + +#: Default UDP port for broadcast discovery on the local LAN. +PORT_BROADCAST = 44515 + +#: UDP port for the cross-subnet unicast variant (Box-IP must be known). +PORT_UNICAST = 44516 + +#: Fixed total length of every discovery frame. +FRAME_SIZE = 0xB8 # 184 + +#: First two bytes of a valid frame. +MAGIC_HEAD = b"ED" + +#: Last two bytes of a valid frame. +MAGIC_TAIL = b"NI" + +#: Protocol version byte (offset 0x05). The reverse-engineered firmware +#: emits and accepts this value; other values have not been observed. +PROTOCOL_VERSION = 0x02 + + +# --- Op-codes (offset 0x02) ------------------------------------------------- + +OP_DISCOVER = 0x01 # request: probe the LAN for bridges +OP_OK = 0x08 # response: discovery answer (box is idle/ready) +OP_BUSY = 0x09 # response: box is busy with another client + +#: Op-codes that are valid responses to a discovery probe. +_RESPONSE_OPS = frozenset({OP_OK, OP_BUSY}) + + +# --- Frame field offsets (named here to keep pack/parse readable) ----------- + +_OFF_OPCODE = 0x02 +_OFF_VERSION = 0x05 +_OFF_NONCE = 0x06 +_OFF_SERIAL = 0x0A +_OFF_MAC = 0x0E +_OFF_MODEL = 0x1C +_OFF_HOSTNAME = 0x3C +_OFF_COMMENT = 0x5C +_OFF_IP = 0x9E +_OFF_SUBNET = 0xA2 +_OFF_GATEWAY = 0xAA +_OFF_TAIL = 0xB6 + +_LEN_MODEL = 32 +_LEN_HOSTNAME = 32 +_LEN_COMMENT = 64 + + +@dataclass(frozen=True) +class BoxInfo: + """Parsed contents of a GPIB-ENET/100 discovery reply. + + All strings have been decoded from the null-terminated ASCII byte + blobs the box ships. Empty fields surface as the empty string. + + """ + + #: Box IP in dotted-quad form (e.g. ``"192.0.2.5"``). + ip: str + + #: MAC address as colon-separated lowercase hex (e.g. ``"00:80:2f:1a:2b:3c"``). + mac: str + + #: Box serial number (32-bit unsigned, big-endian on the wire). + serial: int + + #: Hardware model string reported by the box, e.g. ``"GPIB-ENET/100"``. + model: str + + #: User-assigned hostname; empty if never configured. + hostname: str + + #: Subnet mask in dotted-quad form. + subnet: str + + #: Default gateway in dotted-quad form. + gateway: str + + #: Free-form comment configured via NI MAX / EthernetConfig; usually empty. + comment: str + + #: Echo of the probe's transaction nonce. Useful for matching replies to + #: a specific probe in targeted (non-broadcast) operations. + nonce: int + + #: Raw response op-code: ``OP_OK`` (0x08) or ``OP_BUSY`` (0x09). + op_code: int + + @property + def is_busy(self) -> bool: + """``True`` when the box reported itself as busy with another client.""" + return self.op_code == OP_BUSY + + +def pack_discovery_request(nonce: int = 0) -> bytes: + """Build a 184-byte discovery probe frame. + + All fields default to zero; only the magic sandwich, op-code, protocol + version, and (optionally) the caller-supplied nonce are set. The nonce + is echoed in the box's reply and lets callers correlate replies to + probes when several probes are in flight. + + """ + buf = bytearray(FRAME_SIZE) + buf[0:2] = MAGIC_HEAD + buf[_OFF_OPCODE] = OP_DISCOVER + buf[_OFF_VERSION] = PROTOCOL_VERSION + struct.pack_into("!L", buf, _OFF_NONCE, nonce & 0xFFFFFFFF) + buf[_OFF_TAIL : _OFF_TAIL + 2] = MAGIC_TAIL + return bytes(buf) + + +def parse_discovery_response(buf: bytes) -> BoxInfo | None: + """Parse a 184-byte discovery reply into a :class:`BoxInfo`. + + Returns ``None`` for any frame that fails validation — wrong length, + bad magic, or non-response op-code. Returning ``None`` (rather than + raising) is intentional: the broadcast listener will receive arbitrary + foreign UDP datagrams that should be silently discarded. + + """ + if len(buf) != FRAME_SIZE: + return None + if bytes(buf[0:2]) != MAGIC_HEAD: + return None + if bytes(buf[_OFF_TAIL : _OFF_TAIL + 2]) != MAGIC_TAIL: + return None + op_code = buf[_OFF_OPCODE] + if op_code not in _RESPONSE_OPS: + return None + + nonce = struct.unpack_from("!L", buf, _OFF_NONCE)[0] + serial = struct.unpack_from("!L", buf, _OFF_SERIAL)[0] + mac = ":".join("%02x" % b for b in buf[_OFF_MAC : _OFF_MAC + 6]) + model = _cstring(buf[_OFF_MODEL : _OFF_MODEL + _LEN_MODEL]) + hostname = _cstring(buf[_OFF_HOSTNAME : _OFF_HOSTNAME + _LEN_HOSTNAME]) + comment = _cstring(buf[_OFF_COMMENT : _OFF_COMMENT + _LEN_COMMENT]) + ip = socket.inet_ntoa(bytes(buf[_OFF_IP : _OFF_IP + 4])) + subnet = socket.inet_ntoa(bytes(buf[_OFF_SUBNET : _OFF_SUBNET + 4])) + gateway = socket.inet_ntoa(bytes(buf[_OFF_GATEWAY : _OFF_GATEWAY + 4])) + + return BoxInfo( + ip=ip, + mac=mac, + serial=serial, + model=model, + hostname=hostname, + subnet=subnet, + gateway=gateway, + comment=comment, + nonce=nonce, + op_code=op_code, + ) + + +def _cstring(buf: bytes) -> str: + """Decode a null-terminated ASCII field from a fixed-size buffer. + + Bytes past the first NUL are ignored. Non-ASCII bytes are replaced + with U+FFFD; the box should only ship ASCII but a robust parser does + not crash on rubbish. + + """ + end = buf.find(b"\x00") + if end < 0: + end = len(buf) + return bytes(buf[:end]).decode("ascii", errors="replace") + + +# --- UDP discovery loop ----------------------------------------------------- + + +def discover( + timeout: float = 1.0, + broadcast_addr: str = "255.255.255.255", + port: int = PORT_BROADCAST, + deduplicate: bool = True, +) -> list[BoxInfo]: + """Send a discovery probe and collect bridge responses. + + Opens a UDP socket bound to ``('', port)`` (with ``SO_REUSEADDR`` and + ``SO_BROADCAST`` set), sends one probe to ``(broadcast_addr, port)``, + and reads replies until ``timeout`` seconds elapse. The bridge sends + replies as broadcasts to port ``PORT_BROADCAST`` (per the wire spec), + so receiving them requires a socket bound to that port — ephemeral + binds will miss them. + + Parameters + ---------- + timeout : float + Total time budget for collecting replies. The bind, sendto, and + recvfrom calls all complete within this budget. + broadcast_addr : str + Where to send the probe. The default LAN-wide broadcast covers + the host's default broadcast domain. Pass a subnet broadcast + (e.g. ``"192.0.2.255"``) to limit the scan, or a unicast IP + combined with ``port=PORT_UNICAST`` for the cross-subnet variant + when the box IP is already known. + port : int + UDP port to send to and to bind for replies. Defaults to + ``PORT_BROADCAST`` (44515); use ``PORT_UNICAST`` (44516) for the + cross-subnet variant. + deduplicate : bool + Collapse duplicate replies (same MAC) into a single ``BoxInfo``. + Boxes commonly answer once per host network interface that hears + the probe; without dedup the same box appears more than once. + + Returns + ------- + list[BoxInfo] + Sorted by IP address. Empty if no replies arrived within + ``timeout``. + + """ + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1) + try: + sock.bind(("", port)) + except OSError as e: + # Port may be held by another process or blocked by firewall. + # Surface a meaningful error rather than a recvfrom hang. + raise OSError( + "could not bind UDP port %d for discovery: %s" % (port, e) + ) from e + + sock.sendto(pack_discovery_request(), (broadcast_addr, port)) + + found: dict[str, BoxInfo] = {} + deadline = time.monotonic() + timeout + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + sock.settimeout(remaining) + try: + data, addr = sock.recvfrom(2048) + except socket.timeout: + break + except ConnectionResetError as e: + # Windows surfaces an ICMP "port unreachable" for a prior + # sendto as WSAECONNRESET on the next recvfrom. One such + # rejection does not mean discovery is done — keep listening + # until the timeout actually expires. + LOGGER.debug("recvfrom raised %s; continuing", e) + continue + info = parse_discovery_response(data) + if info is None: + # Foreign UDP traffic or our own probe — silently skip. + continue + key = info.mac if deduplicate else "%s|%d" % (info.mac, len(found)) + found[key] = info + LOGGER.debug( + "discovery reply from %s: ip=%s mac=%s model=%r busy=%s", + addr, + info.ip, + info.mac, + info.model, + info.is_busy, + ) + return sorted( + found.values(), + key=lambda b: tuple(int(part) for part in b.ip.split(".")), + ) + finally: + sock.close() diff --git a/pyvisa_py/testsuite/nienet100_assisted_tests/__init__.py b/pyvisa_py/testsuite/nienet100_assisted_tests/__init__.py new file mode 100644 index 00000000..39f2e741 --- /dev/null +++ b/pyvisa_py/testsuite/nienet100_assisted_tests/__init__.py @@ -0,0 +1,72 @@ +# -*- coding: utf-8 -*- +"""Hardware-gated tests for the NI GPIB-ENET/100 driver. + +These tests only run when a real GPIB-ENET/100 bridge (and optionally an +instrument on its bus) is reachable. Configuration is via environment +variables: + +* ``PYVISA_TEST_NIENET100_HOST`` (required) + IP or hostname of the bridge. Without it every test in this package + skips cleanly. + +* ``PYVISA_TEST_GPIB_PAD`` (required for instrument tests) + Primary GPIB address (0-30) of the instrument under test. + +* ``PYVISA_TEST_GPIB_SAD`` (optional) + Secondary GPIB address (0-30). Defaults to none. + +* ``PYVISA_TEST_IDN_VENDOR`` (optional) + A substring that must appear in the instrument's ``*IDN?`` response + — gives the IDN round-trip test a meaningful assertion when set. + +* ``PYVISA_TEST_GPIB_TERM`` (optional) + Write/read termination as a Python escape sequence, e.g. ``\\n`` (the + default) or ``\\r\\n``. Many older GPIB instruments do not accept + ``\\r\\n`` (pyvisa's library default), so the assisted tests use + ``\\n`` unless this variable overrides it. + +:copyright: 2026 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. + +""" + +import os + +import pytest + +#: Bridge IP/hostname, or ``None`` when not configured. +HOST: str | None = os.environ.get("PYVISA_TEST_NIENET100_HOST") or None + +#: Instrument primary address as int, or ``None`` when not configured. +_pad_env = os.environ.get("PYVISA_TEST_GPIB_PAD") +PAD: int | None = int(_pad_env) if _pad_env else None + +#: Instrument secondary address as int, or ``None`` when not configured. +_sad_env = os.environ.get("PYVISA_TEST_GPIB_SAD") +SAD: int | None = int(_sad_env) if _sad_env else None + +#: Optional substring that must appear in the ``*IDN?`` response. +IDN_VENDOR: str | None = os.environ.get("PYVISA_TEST_IDN_VENDOR") or None + +#: Write/read termination string for the assisted instrument tests. +#: Defaults to ``\n`` rather than pyvisa's library default of ``\r\n`` +#: because many GPIB instruments reject the ``\r``. +TERM: str = ( + (os.environ.get("PYVISA_TEST_GPIB_TERM") or "\\n").encode().decode("unicode_escape") +) + + +#: Skip a test when no bridge is configured. +require_bridge = pytest.mark.skipif( + HOST is None, + reason="set PYVISA_TEST_NIENET100_HOST to a reachable bridge IP", +) + +#: Skip a test when no instrument primary address is configured. +require_instrument = pytest.mark.skipif( + HOST is None or PAD is None, + reason=( + "set PYVISA_TEST_NIENET100_HOST and PYVISA_TEST_GPIB_PAD to enable " + "instrument-level tests" + ), +) diff --git a/pyvisa_py/testsuite/nienet100_assisted_tests/test_session.py b/pyvisa_py/testsuite/nienet100_assisted_tests/test_session.py new file mode 100644 index 00000000..a14d5022 --- /dev/null +++ b/pyvisa_py/testsuite/nienet100_assisted_tests/test_session.py @@ -0,0 +1,250 @@ +# -*- coding: utf-8 -*- +"""Session-level hardware tests for the NI GPIB-ENET/100 driver. + +Drives the bridge through the full pyvisa stack: a ``ResourceManager`` +opens the ``NI-ENET100-TCPIP::INTFC`` interface, then a +``GPIB::::INSTR`` is routed by the central dispatcher in +``pyvisa_py.gpib_dispatch`` to ``NIEnet100InstrSession``. This requires the +``InterfaceType.ni_enet100_tcpip`` and ``NIEnet100TCPIPIntfc`` additions +in upstream pyvisa (available since pyvisa 1.17.0). + +See the package ``__init__`` docstring for environment-variable setup. + +:copyright: 2026 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. + +""" + +from collections.abc import Iterator +from typing import cast + +import pytest + +import pyvisa +from pyvisa import constants +from pyvisa.errors import VisaIOError +from pyvisa.resources import MessageBasedResource +from pyvisa_py import nienet100 as _ni + +from . import HOST, IDN_VENDOR, PAD, SAD, TERM, require_bridge, require_instrument + +# --- fixtures -------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def rm() -> Iterator[pyvisa.ResourceManager]: + """Module-scoped pyvisa ResourceManager bound to the @py backend.""" + manager = pyvisa.ResourceManager("@py") + try: + yield manager + finally: + manager.close() + + +@pytest.fixture(scope="module") +def intfc(rm: pyvisa.ResourceManager): + """Open the bridge INTFC once per module so all INSTR tests share it. + + Binding the INTFC to board 0 also registers it in the dispatch table + so subsequent ``GPIB0::*::INSTR`` opens route through the bridge. + + """ + resource = "NI-ENET100-TCPIP0::%s::INTFC" % HOST + session = rm.open_resource(resource) + try: + yield session + finally: + session.close() + + +@pytest.fixture +def inst(rm: pyvisa.ResourceManager, intfc): + """Per-test INSTR session against the configured PAD/SAD on board 0. + + Depends on ``intfc`` so the bridge binding is in place before the + GPIB dispatch hook fires. The session timeout is set to 3 s by + default; tests that need a different value override it directly. + + """ + # Guaranteed non-None by require_instrument, but assert so the type + # checker can narrow the env-configured Optionals. + assert PAD is not None + if SAD is None: + resource = "GPIB0::%d::INSTR" % PAD + else: + resource = "GPIB0::%d::%d::INSTR" % (PAD, SAD) + session = cast(MessageBasedResource, rm.open_resource(resource)) + session.timeout = 3000 + session.write_termination = TERM + session.read_termination = TERM + try: + yield session + finally: + session.close() + + +# --- bridge-only tests (no instrument required) --------------------------- + + +@require_bridge +def test_list_resources_includes_bridge(rm: pyvisa.ResourceManager): + """Discovery via the resource manager should surface our bridge. + + The default pyvisa query is ``?*::INSTR``; pass ``?*::INTFC`` so our + bridge resource (an INTFC, not an INSTR) is not filtered out. Match + by resolved IP because discovery emits IPs while ``HOST`` may be a + hostname. + + """ + import socket as _socket + + assert HOST is not None # require_bridge guarantees this + host_ip = _socket.gethostbyname(HOST) + resources = rm.list_resources("?*::INTFC") + matches = [r for r in resources if host_ip in r and "NI-ENET100" in r] + assert matches, "no NI-ENET100 resource for %r (%s) in rm.list_resources() = %r" % ( + HOST, + host_ip, + resources, + ) + + +@require_bridge +def test_intfc_open_registers_board(intfc): + """Opening the INTFC must register board 0 in the dispatch table so + GPIB0::*::INSTR resolves to the bridge. + + Board keys mirror ``rname.GPIBInstr.board`` (a string), so the lookup + that the GPIB dispatch hook does with ``parsed.board`` matches. + + """ + boards = _ni._NIEnet100IntfcSession.boards + assert "0" in boards, "INTFC did not register board 0: boards=%r" % ( + list(boards.keys()), + ) + + +@require_bridge +def test_send_ifc_via_pyvisa(intfc): + """viGpibSendIFC pulses Interface Clear on the board session. + + pyvisa's NIEnet100TCPIPIntfc is a bare Resource with no ``send_ifc`` + convenience, so go through the library function directly. A clean IFC + returns ``StatusCode.success`` (the bridge becomes CIC with ATN); the + board-open the INTFC session performs is the state the box needs to + accept IFC. Repeated to confirm it does not wedge. + + """ + for _ in range(3): + status = intfc.visalib.gpib_send_ifc(intfc.session) + assert status == constants.StatusCode.success, "gpib_send_ifc returned %r" % ( + status, + ) + + +# --- instrument tests (need PYVISA_TEST_GPIB_PAD) ------------------------- + + +@require_instrument +def test_idn_query_via_pyvisa(inst): + """*IDN? through the standard pyvisa Resource.query() API.""" + response = inst.query("*IDN?") + assert response, "*IDN? returned empty string" + if IDN_VENDOR: + assert IDN_VENDOR.lower() in response.lower(), "IDN_VENDOR=%r not in %r" % ( + IDN_VENDOR, + response, + ) + + +@require_instrument +def test_idn_query_small_chunks_stress_read_buffer(inst): + """Read the *IDN? response one byte per backend read to exercise the + session's intermediate read buffer. + + Setting ``chunk_size = 1`` forces pyvisa to call the session ``read`` + with ``count == 1`` repeatedly, so the whole-message response cached by + ``ibrd`` is handed out in single-byte slices (the + ``success_max_count_read`` path in ``NIEnet100InstrSession.read``). The + reassembled string must match a normal one-shot query. + + """ + full = inst.query("*IDN?") + inst.chunk_size = 1 + chunked = inst.query("*IDN?") + assert chunked == full, "byte-wise read %r != one-shot read %r" % ( + chunked, + full, + ) + + +@require_instrument +def test_clear_discards_unread_response(inst): + """viClear drops bytes left unread from a previous response. + + Read only a few bytes of one *IDN? reply, leaving the remainder in the + session's intermediate buffer. An ordinary write does not discard that + tail -- like the other message-based backends, the buffer persists across + writes so a response can be read one byte at a time. ``clear()`` is the + VISA operation that flushes it, after which a fresh query returns a clean + response identical to a normal one-shot query. + + """ + expected = inst.query("*IDN?") + + inst.write("*IDN?") + partial = inst.read_bytes(3) + assert len(partial) == 3, "expected a 3-byte partial read, got %r" % (partial,) + + inst.clear() + + again = inst.query("*IDN?") + assert again == expected, "clear did not discard the buffered tail: %r != %r" % ( + again, + expected, + ) + + +@require_instrument +def test_clear_via_pyvisa(inst): + """Resource.clear() must complete without raising.""" + inst.clear() + + +@require_instrument +def test_read_stb_via_pyvisa(inst): + """Resource.read_stb() must return a byte-sized integer.""" + stb = inst.read_stb() + assert 0 <= stb <= 0xFF + + +@require_instrument +def test_assert_trigger_via_pyvisa(inst): + """Resource.assert_trigger() must accept the default protocol.""" + inst.assert_trigger() + + +@require_instrument +def test_timeout_raises_visa_error_timeout(inst): + """A read with no preceding write hits the box timeout and surfaces + as VisaIOError with StatusCode.error_timeout. + + """ + inst.timeout = 200 # 200 ms — short enough that the test is brisk + with pytest.raises(VisaIOError) as excinfo: + inst.read() + assert excinfo.value.error_code == constants.StatusCode.error_timeout, ( + "expected error_timeout, got %r" % excinfo.value.error_code + ) + + +@require_instrument +def test_repeated_query_keeps_session_healthy(inst): + """Three back-to-back queries must all succeed — guards against + accidental state leakage between operations (e.g. unread bytes left + in the chunk stream or a bracket that closed mid-test). + + """ + for _ in range(3): + response = inst.query("*IDN?") + assert response diff --git a/pyvisa_py/testsuite/nienet100_assisted_tests/test_wire.py b/pyvisa_py/testsuite/nienet100_assisted_tests/test_wire.py new file mode 100644 index 00000000..c740f1b7 --- /dev/null +++ b/pyvisa_py/testsuite/nienet100_assisted_tests/test_wire.py @@ -0,0 +1,243 @@ +# -*- coding: utf-8 -*- +"""Wire-level hardware tests for the NI GPIB-ENET/100 driver. + +Drives :class:`~pyvisa_py.protocols.nienet100.EnetConnection` against a +real bridge. Does **not** go through pyvisa-py's session layer, so these +tests pass without requiring the upstream ``InterfaceType.ni_enet100_tcpip`` +addition in pyvisa. Useful for bringing up new hardware and for catching +wire-protocol regressions independently of the session layer. + +See the package ``__init__`` docstring for environment-variable setup. + +:copyright: 2026 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. + +""" + +import os +import socket +import time +from collections.abc import Iterator + +import pytest + +from pyvisa_py import gpib_constants +from pyvisa_py.protocols import nienet100, nienet100_discovery + +from . import ( + HOST, + IDN_VENDOR, + PAD, + SAD, + require_bridge, + require_instrument, +) + +#: The cross-subnet unicast variant (port 44516) needs a probe-source that +#: is NOT on the same subnet as the bridge. Same-subnet probes on 44516 +#: tend to receive no reply (the box answers on the broadcast path). Set +#: PYVISA_TEST_NIENET100_CROSS_SUBNET=1 to opt in when you have a +#: cross-subnet host available. +require_cross_subnet = pytest.mark.skipif( + os.environ.get("PYVISA_TEST_NIENET100_CROSS_SUBNET") != "1", + reason=( + "cross-subnet unicast (port 44516) needs a probe source on a " + "different subnet; set PYVISA_TEST_NIENET100_CROSS_SUBNET=1 to enable" + ), +) + + +def _resolve_host_ip() -> str: + """Resolve ``HOST`` to a dotted-quad IP for comparison against + discovery results, which always carry the bridge's IP. Falls back to + ``HOST`` as-is when DNS resolution fails (e.g. NetBIOS-only names on + a locked-down Windows box) so the test surfaces a meaningful diff + rather than a gaierror. + + """ + assert HOST is not None # callers run only under require_bridge + try: + return socket.gethostbyname(HOST) + except socket.gaierror: + return HOST + + +# --- bridge-only tests (no instrument required) --------------------------- + + +@require_bridge +def test_discovery_finds_configured_bridge(): + """The configured bridge must surface in a broadcast scan.""" + expected_ip = _resolve_host_ip() + boxes = nienet100_discovery.discover(timeout=2.0) + assert boxes, "no bridges replied to broadcast" + ips = [b.ip for b in boxes] + assert expected_ip in ips, ( + "configured bridge %r (resolved to %r) not in discovered set %r" + % (HOST, expected_ip, ips) + ) + + +@require_bridge +@require_cross_subnet +def test_unicast_discovery_against_configured_bridge(): + """Unicast probe to the known IP on the cross-subnet port should return + that box. Skipped by default — see ``require_cross_subnet`` for why. + + """ + assert HOST is not None # require_bridge guarantees this + expected_ip = _resolve_host_ip() + boxes = nienet100_discovery.discover( + timeout=2.0, + broadcast_addr=HOST, + port=nienet100_discovery.PORT_UNICAST, + ) + ips = [b.ip for b in boxes] + assert expected_ip in ips, "unicast probe to %r (resolved to %r) returned %r" % ( + HOST, + expected_ip, + ips, + ) + + +@require_bridge +def test_open_and_close_main_companion(): + """Main + companion sockets and companion hello must round-trip.""" + assert HOST is not None # require_bridge guarantees this + conn = nienet100.EnetConnection(HOST, open_timeout=5.0, timeout=5.0) + conn.open() + try: + assert conn.main is not None + assert conn.companion is not None + finally: + conn.close() + assert conn.main is None and conn.companion is None + + +# --- instrument tests (need PYVISA_TEST_GPIB_PAD) ------------------------- + + +@pytest.fixture +def opened_session() -> Iterator[nienet100.EnetConnection]: + """Yield an EnetConnection with the full open sequence done. + + Cleans up sockets unconditionally even if the test body raises so a + failing test does not leave stale state on the bridge. + + """ + assert HOST is not None and PAD is not None # require_instrument guards + conn = nienet100.EnetConnection(HOST, open_timeout=5.0, timeout=5.0) + conn.open() + try: + conn.open_gpib_session( + primary_address=PAD, + secondary_address=SAD or 0, + tmo_code=gpib_constants.timeout.T3s, + ) + except Exception: + conn.close() + raise + try: + yield conn + finally: + try: + conn.close_gpib_session() + finally: + conn.close() + + +@require_instrument +def test_idn_query_round_trip(opened_session: nienet100.EnetConnection): + """*IDN? must return non-empty bytes; if a vendor substring was + configured, it must appear in the response. + + """ + written = opened_session.ibwrt(b"*IDN?\n") + assert written == 6 + response = opened_session.ibrd() + assert response, "*IDN? returned empty payload" + text = response.decode("ascii", errors="replace") + if IDN_VENDOR: + assert IDN_VENDOR.lower() in text.lower(), "IDN_VENDOR=%r not in %r" % ( + IDN_VENDOR, + text, + ) + + +@require_instrument +def test_clear_round_trip(opened_session: nienet100.EnetConnection): + """ibclr against the addressed device must complete without error.""" + opened_session.ibclr() + + +@require_instrument +def test_read_stb_round_trip(opened_session: nienet100.EnetConnection): + """ibrsp must return a single STB byte (0-255).""" + stb = opened_session.ibrsp() + assert 0 <= stb <= 0xFF + + +@require_instrument +def test_trigger_round_trip(opened_session: nienet100.EnetConnection): + """ibtrg must complete without error. The bus instrument may or may + not actually trigger anything — we only assert the verb is accepted. + + """ + opened_session.ibtrg() + + +@require_instrument +def test_timeout_surfaces_as_iberr_eabo( + opened_session: nienet100.EnetConnection, +): + """A read with no preceding write times out and surfaces as + NIEnet100IOError with iberr=EABO (6). + + The read timeout is governed by the session's tmo_code (the fixture + opens with T3s), not the per-call ibrd ``tmo_ms``. + + """ + started = time.monotonic() + with pytest.raises(nienet100.NIEnet100IOError) as excinfo: + opened_session.ibrd(tmo_ms=200) + assert excinfo.value.err == gpib_constants.error.EABO, ( + "expected EABO (timeout), got iberr=%d" % excinfo.value.err + ) + elapsed = time.monotonic() - started + # T3s means the box waits ~3 s before EABO: assert it actually waited + # (not an instant no-device error) and did not run to a longer code. + assert 1.0 < elapsed < 6.0, ( + "timeout took %.1fs, expected ~3 s (the session's T3s code)" % elapsed + ) + + +@require_instrument +def test_ibwait_round_trip(opened_session: nienet100.EnetConnection): + """ibwait must report an asserted Service Request (RQS). + + Arms a deterministic SRQ via the IEEE 488.2 status model: ``*SRE 16`` + enables SRQ-on-MAV and ``*IDN?`` queues a response, so MAV — and thus + RQS — is set. ibwait (a 0x22 poll on the companion event channel) then + returns immediately with RQS asserted, and a serial poll reads a status + byte that carries the RQS bit. The queued response is drained + afterwards so it cannot leak into a later test. + + """ + conn = opened_session + conn.ibwrt(b"*CLS") + conn.ibwrt(b"*SRE 16") # enable SRQ on MAV (message available) + conn.ibwrt(b"*IDN?") # queue a response -> MAV -> SRQ asserts + + sta = conn.ibwait(gpib_constants.status.RQS | gpib_constants.status.TIMO) + assert sta & gpib_constants.status.RQS, ( + "ibwait did not report RQS: sta=0x%04x" % sta + ) + + stb = conn.ibrsp() + assert stb & 0x40, "serial-poll STB lacks the RQS bit: 0x%02x" % stb + + # Drain the queued *IDN? response so it cannot leak into a later test. + try: + conn.ibrd() + except nienet100.NIEnet100IOError: + pass diff --git a/pyvisa_py/testsuite/test_gpib_dispatch.py b/pyvisa_py/testsuite/test_gpib_dispatch.py new file mode 100644 index 00000000..75b7c684 --- /dev/null +++ b/pyvisa_py/testsuite/test_gpib_dispatch.py @@ -0,0 +1,239 @@ +"""Tests for the central GPIB::INSTR dispatch helper. + +These exercise the routing logic in isolation: which backend session class +a parsed resource resolves to, in what order backends are consulted, and +the no-match behaviour. No real GPIB hardware or backend session is +involved — backends are registered as plain resolver callables returning +sentinel classes. + +:copyright: 2026 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. + +""" + +import pytest + +from pyvisa.constants import InterfaceType +from pyvisa_py import gpib_dispatch +from pyvisa_py.sessions import OpenError, Session + + +@pytest.fixture +def clean_registry(): + """Isolate the module-global backend list around each test. + + The dispatcher keeps its resolvers in a module-level list; save it, + clear it for the test, and restore it afterwards so tests do not leak + into one another or into real registrations. + + """ + saved = list(gpib_dispatch._GPIB_INSTR_BACKENDS) + saved_flag = gpib_dispatch._builtins_registered + gpib_dispatch._GPIB_INSTR_BACKENDS.clear() + gpib_dispatch._builtins_registered = False + try: + yield gpib_dispatch._GPIB_INSTR_BACKENDS + finally: + gpib_dispatch._GPIB_INSTR_BACKENDS[:] = saved + gpib_dispatch._builtins_registered = saved_flag + + +class _Recorder: + """Sentinel backend session that records its constructor arguments.""" + + def __init__(self, rm, name, parsed, open_timeout): + self.args = (rm, name, parsed, open_timeout) + + +def test_first_matching_backend_wins(clean_registry): + class A(_Recorder): + pass + + class B(_Recorder): + pass + + gpib_dispatch.register_backend(lambda p: None, priority=0, label="none") + gpib_dispatch.register_backend(lambda p: A, priority=10, label="a") + gpib_dispatch.register_backend(lambda p: B, priority=20, label="b") + + obj = gpib_dispatch.GPIBInstrDispatch( + object(), "GPIB0::1::INSTR", parsed=object(), open_timeout=None + ) + assert isinstance(obj, A) + + +def test_no_backend_matches_raises_openerror(clean_registry): + gpib_dispatch.register_backend(lambda p: None, priority=0) + + with pytest.raises(OpenError): + gpib_dispatch.GPIBInstrDispatch(object(), "GPIB0::1::INSTR", parsed=object()) + + +def test_empty_registry_raises_openerror(clean_registry): + with pytest.raises(OpenError): + gpib_dispatch.GPIBInstrDispatch(object(), "GPIB0::1::INSTR", parsed=object()) + + +def test_priority_then_insertion_order(clean_registry): + order = [] + + def probe(tag): + def resolve(parsed): + order.append(tag) + return None + + return resolve + + # Register out of order and with a tie at priority 10; the dispatcher + # must consult them by ascending priority, breaking the tie by + # insertion order (first-registered first). + gpib_dispatch.register_backend(probe("p10-first"), priority=10) + gpib_dispatch.register_backend(probe("p0"), priority=0) + gpib_dispatch.register_backend(probe("p10-second"), priority=10) + + with pytest.raises(OpenError): + gpib_dispatch.GPIBInstrDispatch(object(), "x", parsed=object()) + + assert order == ["p0", "p10-first", "p10-second"] + + +def test_constructor_args_forwarded_unchanged(clean_registry): + captured = {} + + class Fake: + def __init__(self, rm, name, parsed, open_timeout): + captured.update(rm=rm, name=name, parsed=parsed, open_timeout=open_timeout) + + gpib_dispatch.register_backend(lambda p: Fake, priority=0) + + rm, name, parsed, timeout = object(), "GPIB0::5::INSTR", object(), 7 + gpib_dispatch.GPIBInstrDispatch(rm, name, parsed=parsed, open_timeout=timeout) + + assert captured == { + "rm": rm, + "name": name, + "parsed": parsed, + "open_timeout": timeout, + } + + +def test_resource_name_parsed_when_not_supplied(clean_registry): + seen = {} + + def resolve(parsed): + seen["parsed"] = parsed + return None + + gpib_dispatch.register_backend(resolve, priority=0) + + with pytest.raises(OpenError): + gpib_dispatch.GPIBInstrDispatch(object(), "GPIB0::5::INSTR") + + # The dispatcher parsed the string into a resource object for us. + assert seen["parsed"] is not None + assert str(seen["parsed"].board) == "0" + + +class _Parsed: + """Minimal stand-in for a parsed GPIB resource carrying only ``board``.""" + + def __init__(self, board): + self.board = board + + +def test_board_resolver_claims_only_registered_boards(): + class Sentinel: + pass + + boards = {"1": object()} + resolve = gpib_dispatch._board_resolver(boards, Sentinel) + + assert resolve(_Parsed("1")) is Sentinel + assert resolve(_Parsed("2")) is None + + +def test_board_resolver_sees_boards_registered_after_wiring(): + class Sentinel: + pass + + boards = {} + resolve = gpib_dispatch._board_resolver(boards, Sentinel) + + # Board not present yet ... + assert resolve(_Parsed("0")) is None + # ... an INTFC session registers it later, by reference. + boards["0"] = object() + assert resolve(_Parsed("0")) is Sentinel + + +def test_native_unavailable_raises_actionable_error_on_open(): + cls = gpib_dispatch._make_native_unavailable(ImportError("no gpib lib")) + + with pytest.raises(ValueError) as info: + cls(object(), "GPIB0::1::INSTR", _Parsed("0"), None) + assert "gpib-ctypes" in str(info.value) + + +def test_register_builtin_backends_is_idempotent_and_ordered(clean_registry): + gpib_dispatch.register_builtin_backends() + after_first = list(clean_registry) + gpib_dispatch.register_builtin_backends() + after_second = list(clean_registry) + + # Idempotent: a second call adds nothing. + assert after_first == after_second + + priorities = [priority for priority, _label, _resolve in after_first] + labels = [label for _priority, label, _resolve in after_first] + + # Native is the catch-all and must be wired last (highest priority value). + assert priorities == sorted(priorities) + assert priorities[-1] == gpib_dispatch._PRIORITY_NATIVE + assert labels[-1] == "gpib" + # Pure-Python backends are always available and out-rank native; the + # bridge out-ranks Prologix. + assert labels.index("ni-enet100") < labels.index("prologix") < labels.index("gpib") + assert dict(zip(labels, priorities))["ni-enet100"] == gpib_dispatch._PRIORITY_BRIDGE + + +def test_bridge_out_ranks_prologix_for_shared_board(clean_registry): + # A board registered to *both* a bridge and Prologix must resolve to the + # bridge, because the bridge resolver is wired with higher precedence. + class Bridge(_Recorder): + pass + + class Prlgx(_Recorder): + pass + + bridge_boards = {"0": object()} + prologix_boards = {"0": object()} + gpib_dispatch.register_backend( + gpib_dispatch._board_resolver(bridge_boards, Bridge), + priority=gpib_dispatch._PRIORITY_BRIDGE, + label="ni-enet100", + ) + gpib_dispatch.register_backend( + gpib_dispatch._board_resolver(prologix_boards, Prlgx), + priority=gpib_dispatch._PRIORITY_PROLOGIX, + label="prologix", + ) + + obj = gpib_dispatch.GPIBInstrDispatch( + object(), "GPIB0::1::INSTR", parsed=_Parsed("0") + ) + assert isinstance(obj, Bridge) + + +def test_central_dispatcher_owns_gpib_instr_slot(): + # Importing gpib_dispatch registers it as the single owner of the slot. + assert ( + Session.get_session_class(InterfaceType.gpib, "INSTR") + is gpib_dispatch.GPIBInstrDispatch + ) + + +def test_list_resources_is_callable_and_returns_list(): + # Returns native listeners when a GPIB library is present, otherwise an + # empty list — never raises, regardless of platform. + result = gpib_dispatch.GPIBInstrDispatch.list_resources() + assert isinstance(result, list) diff --git a/pyvisa_py/testsuite/test_nienet100.py b/pyvisa_py/testsuite/test_nienet100.py new file mode 100644 index 00000000..69f42711 --- /dev/null +++ b/pyvisa_py/testsuite/test_nienet100.py @@ -0,0 +1,629 @@ +# -*- coding: utf-8 -*- +"""Offline unit tests for the NI GPIB-ENET/100 wire protocol primitives. + +These tests cover :mod:`pyvisa_py.protocols.nienet100` without touching the +network: frame pack/unpack, chunk parsing, IP/TMO conversion helpers, and +the device verbs driven against scripted in-memory peers over a Unix +``socketpair``. Hardware-gated integration tests against a real bridge +live elsewhere. + +:copyright: 2026 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. + +""" + +import io +import socket +import struct +import threading + +import pytest + +from pyvisa_py import gpib_constants +from pyvisa_py.protocols import nienet100 + +# --- frame pack / unpack ---------------------------------------------------- + + +def test_pack_command_zeroes_unset_fields(): + frame = nienet100.pack_command(0x04) + assert len(frame) == nienet100.COMMAND_FRAME_SIZE + assert frame == b"\x04\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00" + + +def test_pack_command_layout(): + # Frame A (SetConfig SC) skeleton: 07 02 00 01 [PAD=16] [SAD=0] 00 00 + # [tmo=0x0d] 00 04 00. PAD/SAD pack into the w2 ushort as (PAD<<8)|SAD. + frame = nienet100.pack_command( + cmd_id=0x07, + b1=0x02, + w1=0x0001, + w2=(16 << 8) | 0, + w3=0, + dw=(gpib_constants.timeout.T10s << 24) | 0x0400, + ) + assert frame.hex() == "07020001" + "10000000" + "0d000400" + + +def test_parse_status_header_ok(): + raw = struct.pack("!HH4xL", gpib_constants.status.CMPL, 0x0000, 42) + sta, err, cnt = nienet100.parse_status_header(raw) + assert sta == gpib_constants.status.CMPL + assert err == 0 + assert cnt == 42 + + +def test_parse_status_header_err_sentinel(): + # err=0xFFFF is a documented sentinel that callers must ignore unless + # the ERR bit is set. + raw = struct.pack("!HH4xL", gpib_constants.status.CMPL, 0xFFFF, 0) + sta, err, cnt = nienet100.parse_status_header(raw) + assert sta == gpib_constants.status.CMPL + assert err == 0xFFFF + assert cnt == 0 + + +def test_parse_status_header_rejects_wrong_size(): + with pytest.raises(ValueError): + nienet100.parse_status_header(b"\x00" * 11) + + +def test_parse_chunk_header(): + flags, length = nienet100.parse_chunk_header(b"\x00\x01\x00\x40") + assert flags == 1 + assert length == 0x40 + + +# --- chunk reader ----------------------------------------------------------- + + +def _reader(blob: bytes): + """Build a `read_exactly`-style callable backed by an in-memory blob.""" + stream = io.BytesIO(blob) + + def read_exactly(n: int) -> bytes: + data = stream.read(n) + if len(data) != n: + raise EOFError("short read") + return data + + return read_exactly + + +def _chunk(flags: int, payload: bytes) -> bytes: + return struct.pack("!HH", flags, len(payload)) + payload + + +def test_read_chunks_until_end_concatenates(): + blob = _chunk(0, b"ABC") + _chunk(0, b"DE") + _chunk(1, b"") + assert nienet100.read_chunks_until_end(_reader(blob)) == b"ABCDE" + + +def test_read_chunks_until_end_tolerates_signal_chunk(): + # signal-byte chunks (flags=2) carry 1 OOB byte; the reader logs and + # skips them rather than treating them as protocol errors. + blob = _chunk(0, b"X") + b"\x00\x02\x00\x00\xff" + _chunk(0, b"Y") + _chunk(1, b"") + assert nienet100.read_chunks_until_end(_reader(blob)) == b"XY" + + +def test_read_chunks_until_end_treats_unknown_zero_length_flag_as_terminator(): + # Real hardware emits flag 0x0004 (length=0) on timeouts. Treating it + # as end-of-stream (rather than raising) lets the caller's subsequent + # status-header read carry the real error code. + blob = _chunk(0, b"PARTIAL") + b"\x00\x04\x00\x00" + assert nienet100.read_chunks_until_end(_reader(blob)) == b"PARTIAL" + + +def test_read_chunks_until_end_rejects_unknown_flag_with_non_zero_length(): + # Non-zero length on an unknown flag would desync the byte stream + # (we cannot tell how to consume the payload), so it still raises. + with pytest.raises(nienet100.NIEnet100ProtocolError): + nienet100.read_chunks_until_end(_reader(b"\x00\x99\x00\x05" + b"XXXXX")) + + +def test_read_chunks_until_end_rejects_end_with_payload(): + with pytest.raises(nienet100.NIEnet100ProtocolError): + nienet100.read_chunks_until_end(_reader(b"\x00\x01\x00\x05XXXXX")) + + +def test_read_one_data_chunk_returns_first_payload(): + blob = _chunk(0, b"\x42") + _chunk(1, b"") + assert nienet100.read_one_data_chunk(_reader(blob)) == b"\x42" + + +def test_read_one_data_chunk_skips_signal_chunks(): + blob = b"\x00\x02\x00\x00\xff" + _chunk(0, b"OK") + assert nienet100.read_one_data_chunk(_reader(blob)) == b"OK" + + +# --- conversion helpers ----------------------------------------------------- + + +@pytest.mark.parametrize( + "ip, want_hex", + [ + ("0.0.0.0", 0x00000000), + ("127.0.0.1", 0x7F000001), + ("192.0.2.5", 0xC0000205), + ("255.255.255.255", 0xFFFFFFFF), + ], +) +def test_u32_from_ip(ip: str, want_hex: int): + assert nienet100._u32_from_ip(ip) == want_hex + + +@pytest.mark.parametrize( + "seconds, want", + [ + (None, gpib_constants.timeout.TNONE), + (0, gpib_constants.timeout.TNONE), + (10e-6, gpib_constants.timeout.T10us), + (1e-3, gpib_constants.timeout.T1ms), + (1.0, gpib_constants.timeout.T1s), + (10.0, gpib_constants.timeout.T10s), + # 0.5 s rounds up to 1 s (next discrete value) + (0.5, gpib_constants.timeout.T1s), + # 5 s rounds up to 10 s + (5.0, gpib_constants.timeout.T10s), + # Clamp to the largest available code + (5000.0, gpib_constants.timeout.T1000s), + ], +) +def test_seconds_to_tmo_code(seconds, want: int): + assert nienet100.seconds_to_tmo_code(seconds) == want + + +# --- IO errors -------------------------------------------------------------- + + +def test_iberr_exception_carries_fields(): + e = nienet100.NIEnet100IOError( + gpib_constants.status.ERR | gpib_constants.status.CMPL, + gpib_constants.error.ENOL, + "ibwrt", + ) + assert e.sta == gpib_constants.status.ERR | gpib_constants.status.CMPL + assert e.err == gpib_constants.error.ENOL + assert "ibwrt" in str(e) + + +# --- end-to-end against a scripted peer over socketpair -------------------- +# Only runs on platforms with socket.socketpair (Unix; Windows 3.5+). + + +ScriptStep = tuple[str, bytes] # ("send", payload) or ("recv", payload) + + +def _run_scripted_peer(script: list[ScriptStep]): + """Return (client_sock, thread). The thread plays ``script`` on the peer. + + On ``recv`` steps the thread asserts the exact bytes the client sent; + on ``send`` steps it pushes bytes to the client. + + """ + a, b = socket.socketpair() + + def play(): + try: + for direction, payload in script: + if direction == "recv": + got = bytearray() + while len(got) < len(payload): + chunk = b.recv(len(payload) - len(got)) + if not chunk: + raise AssertionError( + "peer disconnected after %d/%d bytes" + % (len(got), len(payload)) + ) + got.extend(chunk) + assert bytes(got) == payload, "client sent %r, expected %r" % ( + bytes(got), + payload, + ) + elif direction == "send": + b.sendall(payload) + else: + raise AssertionError("bad direction %r" % direction) + finally: + b.close() + + t = threading.Thread(target=play, daemon=True) + t.start() + return a, t + + +def _wrap_status(body: bytes) -> bytes: + """Wrap a 12-byte status body in the on-wire chunk header. + + Every status header the bridge emits is delivered as a data chunk + (flags=0, length=12); the scripted peer must therefore prepend the + chunk header for the bridge driver to read aligned. + + """ + return struct.pack("!HH", 0, 12) + body + + +def _status_ok(cnt: int = 0) -> bytes: + body = struct.pack("!HH4xL", gpib_constants.status.CMPL, 0, cnt) + return _wrap_status(body) + + +def _make_bound_connection(client_sock: socket.socket) -> nienet100.EnetConnection: + """Build an EnetConnection bound to ``client_sock`` as its main socket, + skipping the normal connect/companion-hello path so individual verbs can + be tested in isolation.""" + conn = nienet100.EnetConnection.__new__(nienet100.EnetConnection) + conn.main = client_sock + conn.companion = None + conn.host = "test-peer" + conn._open_timeout = 1.0 + conn._timeout = 1.0 + return conn + + +def _make_empty_connection() -> nienet100.EnetConnection: + """Build an EnetConnection with no sockets bound, for tests that drive + socket-lifecycle methods via a monkey-patched ``_connect``.""" + conn = nienet100.EnetConnection.__new__(nienet100.EnetConnection) + conn.main = None + conn.companion = None + conn.host = "test-peer" + conn._open_timeout = 1.0 + conn._timeout = 1.0 + conn._bracket_open = False + return conn + + +def _bound_inet_socket() -> socket.socket: + """Return an AF_INET socket bound to an ephemeral loopback port. + + The 'O'/'U' verbs embed the main socket's ``getsockname()`` address in + the wire frame, so tests that exercise them need a main socket whose + ``getsockname()`` returns a real ``(ip, port)`` tuple. ``socket. + socketpair()`` yields AF_UNIX sockets on Unix whose ``getsockname()`` + is the empty string, so it cannot stand in for the main socket here. + The socket is only queried for its address (never read/written), so a + bind without connect is enough. + """ + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.bind(("127.0.0.1", 0)) + return sock + + +def test_set_read_timeout_code_reopens_bracket(): + # Changing the read timeout mid-session must close the bracket, re-send + # Frame A (SetConfig SC) with the new tmo_code for the same target, and + # reopen the bracket -- the box rejects 'P 03' while a bracket is open. + # Arbitrary target address; this is an offline wire test. + pad, sad = 5, 0 + new_code = gpib_constants.timeout.T1s + bracket_close = struct.pack("!BBB9x", 0x58, 0x00, 0x01) + frame_a = nienet100.pack_command( + cmd_id=0x07, + b1=0x02, + w1=0x0001, + w2=(pad << 8) | (sad & 0xFF), + w3=0, + dw=(new_code << 24) | 0x0400, + ) + bracket_open = struct.pack("!BBB9x", 0x58, 0x01, 0x01) + script = [ + ("recv", bracket_close), + ("send", _status_ok()), + ("recv", frame_a), + ("send", _status_ok()), + ("recv", bracket_open), + ("send", _status_ok()), + ] + sock, t = _run_scripted_peer(script) + conn = _make_bound_connection(sock) + conn._bracket_open = True + conn._pad, conn._sad = pad, sad + conn._read_tmo_code = gpib_constants.timeout.T10s + try: + conn.set_read_timeout_code(new_code) + assert conn._read_tmo_code == new_code + assert conn._bracket_open is True + finally: + sock.close() + t.join(timeout=1.0) + + +def test_set_read_timeout_code_noop_when_unchanged(): + # Same code -> returns immediately without touching the socket. + conn = _make_empty_connection() + conn._read_tmo_code = gpib_constants.timeout.T3s + conn.set_read_timeout_code(gpib_constants.timeout.T3s) + assert conn._read_tmo_code == gpib_constants.timeout.T3s + + +def test_set_read_timeout_code_records_when_no_bracket(): + # No open bracket -> just remember the code for the next open, no wire I/O. + conn = _make_empty_connection() + conn._read_tmo_code = gpib_constants.timeout.T10s + conn.set_read_timeout_code(gpib_constants.timeout.T1s) + assert conn._read_tmo_code == gpib_constants.timeout.T1s + + +def test_ibwrt_sends_header_and_payload_combined(): + # Odd-length payload must be sent UNPADDED (count=5, 5 payload bytes) — + # padding makes the box reject the frame. Mirrors the NI capture. + payload = b"HELLO" + expected = struct.pack("!BBHL4x", 0x62, 0x00, 0x0000, len(payload)) + payload + script = [("recv", expected), ("send", _status_ok(cnt=5))] + sock, t = _run_scripted_peer(script) + conn = _make_bound_connection(sock) + try: + assert conn.ibwrt(payload) == 5 + finally: + sock.close() + t.join(timeout=2.0) + + +def test_ibrd_with_data_consumes_prelim_data_end_and_final(): + """Spec path: preliminary status, data chunks, END marker, final status.""" + expected_frame = struct.pack( + "!BBHL4x", 0x16, 0x00, 0x0000, nienet100.DEFAULT_IBRD_TMO_MS + ) + script = [ + ("recv", expected_frame), + ("send", _status_ok()), # preliminary status + ("send", _chunk(0, b"WORLD\n")), + ("send", _chunk(1, b"")), # END + ( + "send", + _wrap_status( + struct.pack( + "!HH4xL", + gpib_constants.status.END | gpib_constants.status.CMPL, + 0xFFFF, + 6, + ) + ), + ), # final status + ] + sock, t = _run_scripted_peer(script) + conn = _make_bound_connection(sock) + try: + assert conn.ibrd() == b"WORLD\n" + finally: + sock.close() + t.join(timeout=2.0) + + +def test_ibrd_no_data_path_accepts_final_status_without_end_marker(): + """No-data path: the bridge sends preliminary + final without an + intervening END marker. The parser must recognize the second + length-12 chunk as the final status by inspecting its body.""" + expected_frame = struct.pack("!BBHL4x", 0x16, 0x00, 0x0000, 100) + script = [ + ("recv", expected_frame), + ("send", _status_ok()), # preliminary + ("send", _status_ok()), # final directly, no END between + ] + sock, t = _run_scripted_peer(script) + conn = _make_bound_connection(sock) + try: + assert conn.ibrd(tmo_ms=100) == b"" + finally: + sock.close() + t.join(timeout=2.0) + + +def test_ibrd_no_data_path_propagates_error_status(): + """No-data path with an error final status (``ERR`` bit set) must raise.""" + expected_frame = struct.pack("!BBHL4x", 0x16, 0x00, 0x0000, 100) + error_status = _wrap_status( + struct.pack( + "!HH4xL", + gpib_constants.status.ERR | gpib_constants.status.CMPL, + gpib_constants.error.EABO, + 0, + ) + ) + script = [ + ("recv", expected_frame), + ("send", _status_ok()), # preliminary + ("send", error_status), # final with timeout + ] + sock, t = _run_scripted_peer(script) + conn = _make_bound_connection(sock) + try: + with pytest.raises(nienet100.NIEnet100IOError) as excinfo: + conn.ibrd(tmo_ms=100) + assert excinfo.value.err == gpib_constants.error.EABO + finally: + sock.close() + t.join(timeout=2.0) + + +@pytest.mark.parametrize("cnt", [1, 0]) +def test_ibrsp_returns_stb_from_combined_chunk(cnt: int): + # The bridge packs the 12-byte status header and the 1-byte STB into one + # chunk with length=13; the STB is always the trailing byte. cnt is not + # reliably 1 (a serial poll right after an SRQ wait reports cnt=0 with a + # valid STB), so the STB must be read by position regardless of cnt. + status_body = struct.pack("!HH4xL", gpib_constants.status.CMPL, 0, cnt) + response = _chunk(0, status_body + b"\x42") + script = [ + ("recv", nienet100.pack_command(0x19)), + ("send", response), + ] + sock, t = _run_scripted_peer(script) + conn = _make_bound_connection(sock) + try: + assert conn.ibrsp() == 0x42 + finally: + sock.close() + t.join(timeout=2.0) + + +def test_ibclr_raises_iberr_on_error_status(): + script = [ + ("recv", nienet100.pack_command(0x04)), + ( + "send", + _wrap_status( + struct.pack( + "!HH4xL", + gpib_constants.status.ERR | gpib_constants.status.CMPL, + gpib_constants.error.ENOL, + 0, + ) + ), + ), + ] + sock, t = _run_scripted_peer(script) + conn = _make_bound_connection(sock) + try: + with pytest.raises(nienet100.NIEnet100IOError) as excinfo: + conn.ibclr() + assert excinfo.value.err == gpib_constants.error.ENOL + finally: + sock.close() + t.join(timeout=2.0) + + +@pytest.mark.parametrize( + "opcode, method", + [ + (0x04, "ibclr"), + (0x20, "ibtrg"), + (0x10, "ibloc"), + ], +) +def test_simple_verbs_roundtrip(opcode, method): + script = [("recv", nienet100.pack_command(opcode)), ("send", _status_ok())] + sock, t = _run_scripted_peer(script) + conn = _make_bound_connection(sock) + try: + getattr(conn, method)() + finally: + sock.close() + t.join(timeout=2.0) + + +def test_set_io_timeout_sends_property_set_frame(): + expected = struct.pack("!BBB9x", 0x50, 0x03, gpib_constants.timeout.T10s) + script = [("recv", expected), ("send", _status_ok())] + sock, t = _run_scripted_peer(script) + conn = _make_bound_connection(sock) + try: + conn.set_io_timeout(gpib_constants.timeout.T10s) + finally: + sock.close() + t.join(timeout=2.0) + + +# --- ibwait (B2) ----------------------------------------------------------- + + +def test_ibwait_sends_mask_and_returns_sta(): + # ibwait polls with 0x22 on the companion socket (the event channel). + mask = gpib_constants.status.RQS | gpib_constants.status.TIMO + expected_frame = nienet100.pack_command(cmd_id=0x22, b1=0x00, w1=mask) + response_sta = gpib_constants.status.RQS | gpib_constants.status.CMPL + script = [ + ("recv", expected_frame), + ("send", _wrap_status(struct.pack("!HH4xL", response_sta, 0xFFFF, 0))), + ] + companion_sock, t = _run_scripted_peer(script) + try: + conn = _make_bound_connection(socket.socket()) # main present, unused here + conn.companion = companion_sock + sta = conn.ibwait(mask) + assert sta == response_sta + finally: + companion_sock.close() + t.join(timeout=2.0) + + +def test_ibwait_raises_on_error_status(): + script = [ + ( + "recv", + nienet100.pack_command(cmd_id=0x22, b1=0x00, w1=gpib_constants.status.RQS), + ), + ( + "send", + _wrap_status( + struct.pack( + "!HH4xL", + gpib_constants.status.ERR | gpib_constants.status.CMPL, + gpib_constants.error.EARG, + 0, + ) + ), + ), + ] + companion_sock, t = _run_scripted_peer(script) + try: + conn = _make_bound_connection(socket.socket()) + conn.companion = companion_sock + with pytest.raises(nienet100.NIEnet100IOError) as excinfo: + conn.ibwait(gpib_constants.status.RQS) + assert excinfo.value.err == gpib_constants.error.EARG + finally: + companion_sock.close() + t.join(timeout=2.0) + + +# --- ibsic (B3) ----------------------------------------------------------- + + +def test_ibsic_sends_1c_on_main(): + # IFC is a bare 0x1c command on the main socket (verified against the + # genuine NI software); the box replies CMPL|CIC|ATN. + expected = nienet100.pack_command(0x1C) + reply_sta = ( + gpib_constants.status.CMPL + | gpib_constants.status.CIC + | gpib_constants.status.ATN + ) + main_sock, t = _run_scripted_peer( + [ + ("recv", expected), + ("send", _wrap_status(struct.pack("!HH4xL", reply_sta, 0, 0))), + ] + ) + try: + conn = _make_bound_connection(main_sock) + conn.ibsic() + finally: + main_sock.close() + t.join(timeout=2.0) + + +def test_close_drops_all_sockets_without_extra_frames(): + # close() must not emit anything on the companion socket; it just tears + # the sockets down. A companion peer read therefore sees only EOF. + main_a = _bound_inet_socket() + companion_a, companion_b = socket.socketpair() + try: + conn = _make_empty_connection() + conn.main = main_a + conn.companion = companion_a + conn.close() + assert conn.main is None + assert conn.companion is None + companion_b.settimeout(2.0) + assert companion_b.recv(64) == b"", "close() unexpectedly sent a frame" + finally: + main_a.close() + companion_b.close() + + +def test_close_swallows_socket_errors(): + # Sockets already closed before teardown: close() logs and proceeds + # without raising. + main_a = _bound_inet_socket() + try: + fake_companion = socket.socket() + fake_companion.close() # already closed + conn = _make_empty_connection() + conn.main = main_a + conn.companion = fake_companion + conn.close() # must not raise + assert conn.companion is None + finally: + main_a.close() diff --git a/pyvisa_py/testsuite/test_nienet100_discovery.py b/pyvisa_py/testsuite/test_nienet100_discovery.py new file mode 100644 index 00000000..fbdd956f --- /dev/null +++ b/pyvisa_py/testsuite/test_nienet100_discovery.py @@ -0,0 +1,298 @@ +# -*- coding: utf-8 -*- +"""Offline unit tests for the NI GPIB-ENET/100 UDP discovery protocol. + +Pack/parse helpers run pure offline. The discover() loop is exercised +via socket.socket patching so the test fixtures do not depend on +broadcast-capable network interfaces or on port 44515 being free. + +:copyright: 2026 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. + +""" + +import socket +import struct +from unittest import mock + +import pytest + +from pyvisa_py.protocols import nienet100_discovery as d + +# --- frame builders -------------------------------------------------------- + + +def _build_response( + ip: str = "192.0.2.5", + mac_bytes: bytes = bytes.fromhex("00802f1a2b3c"), + serial: int = 1234567, + model: str = "GPIB-ENET/100", + hostname: str = "lab-vna", + comment: str = "by the door", + subnet: str = "255.255.255.0", + gateway: str = "192.0.2.1", + nonce: int = 0, + op_code: int = d.OP_OK, +) -> bytes: + """Build a valid 184-byte discovery reply frame for testing.""" + buf = bytearray(d.FRAME_SIZE) + buf[0:2] = d.MAGIC_HEAD + buf[0x02] = op_code + buf[0x05] = d.PROTOCOL_VERSION + struct.pack_into("!L", buf, 0x06, nonce & 0xFFFFFFFF) + struct.pack_into("!L", buf, 0x0A, serial & 0xFFFFFFFF) + buf[0x0E : 0x0E + 6] = mac_bytes + encoded_model = model.encode("ascii") + buf[0x1C : 0x1C + len(encoded_model)] = encoded_model + encoded_host = hostname.encode("ascii") + buf[0x3C : 0x3C + len(encoded_host)] = encoded_host + encoded_comment = comment.encode("ascii") + buf[0x5C : 0x5C + len(encoded_comment)] = encoded_comment + buf[0x9E : 0x9E + 4] = socket.inet_aton(ip) + buf[0xA2 : 0xA2 + 4] = socket.inet_aton(subnet) + buf[0xAA : 0xAA + 4] = socket.inet_aton(gateway) + buf[0xB6:0xB8] = d.MAGIC_TAIL + return bytes(buf) + + +# --- pack_discovery_request ------------------------------------------------ + + +def test_pack_discovery_request_layout(): + frame = d.pack_discovery_request() + assert len(frame) == d.FRAME_SIZE + assert frame[0:2] == d.MAGIC_HEAD + assert frame[0x02] == d.OP_DISCOVER + assert frame[0x05] == d.PROTOCOL_VERSION + assert frame[0xB6:0xB8] == d.MAGIC_TAIL + + +def test_pack_discovery_request_nonce_is_big_endian(): + frame = d.pack_discovery_request(nonce=0xDEADBEEF) + assert frame[0x06:0x0A] == bytes.fromhex("deadbeef") + + +def test_pack_discovery_request_zeroes_unset_fields(): + frame = d.pack_discovery_request(nonce=0xABCD) + # Everything outside the small set of always-on fields and the nonce + # must remain zero. + set_offsets = {0, 1, 2, 5, 6, 7, 8, 9, 0xB6, 0xB7} + for i, byte in enumerate(frame): + if i not in set_offsets: + assert byte == 0, "byte %d should be 0, got 0x%02x" % (i, byte) + + +# --- parse_discovery_response (happy paths) -------------------------------- + + +def test_parse_response_extracts_all_fields(): + info = d.parse_discovery_response( + _build_response( + ip="192.0.2.5", + mac_bytes=bytes.fromhex("00802fdeadbe"), + serial=99, + model="GPIB-ENET/100", + hostname="vna-01", + comment="rack 3, shelf 2", + subnet="255.255.255.128", + gateway="192.0.2.129", + nonce=0x12345678, + op_code=d.OP_OK, + ) + ) + assert info is not None + assert info.ip == "192.0.2.5" + assert info.mac == "00:80:2f:de:ad:be" + assert info.serial == 99 + assert info.model == "GPIB-ENET/100" + assert info.hostname == "vna-01" + assert info.comment == "rack 3, shelf 2" + assert info.subnet == "255.255.255.128" + assert info.gateway == "192.0.2.129" + assert info.nonce == 0x12345678 + assert info.op_code == d.OP_OK + assert info.is_busy is False + + +def test_parse_response_busy_flag(): + info = d.parse_discovery_response(_build_response(op_code=d.OP_BUSY)) + assert info is not None + assert info.op_code == d.OP_BUSY + assert info.is_busy is True + + +def test_parse_response_handles_empty_strings(): + info = d.parse_discovery_response(_build_response(hostname="", comment="")) + assert info is not None + assert info.hostname == "" + assert info.comment == "" + + +def test_parse_response_truncates_at_first_null(): + # Spec strings are null-terminated within their fixed-size buffer; + # anything past the first NUL must be ignored even if it looks like + # ASCII (which the box can leave behind from a prior longer value). + buf = bytearray(_build_response(hostname="short")) + # Put garbage after the NUL terminator that follows "short" + buf[0x3C + 6 : 0x3C + 16] = b"GARBAGE!!!" + info = d.parse_discovery_response(bytes(buf)) + assert info is not None + assert info.hostname == "short" + + +# --- parse_discovery_response (validation) -------------------------------- + + +@pytest.mark.parametrize( + "mutator,reason", + [ + (lambda b: b[:100], "too short"), + (lambda b: b + b"\x00", "too long"), + (lambda b: b"AA" + b[2:], "bad head magic"), + (lambda b: b[:0xB6] + b"AA", "bad tail magic"), + ( + lambda b: b[:0x02] + bytes([d.OP_DISCOVER]) + b[0x03:], + "probe op-code, not response", + ), + (lambda b: b[:0x02] + b"\xff" + b[0x03:], "unknown op-code"), + ], +) +def test_parse_response_rejects_invalid_frames(mutator, reason): + bad = mutator(_build_response()) + assert d.parse_discovery_response(bad) is None, "should reject: %s" % reason + + +def test_parse_response_rejects_truly_foreign_traffic(): + # Random UDP traffic from other devices on the LAN must not raise. + assert d.parse_discovery_response(b"") is None + assert d.parse_discovery_response(b"hello") is None + assert d.parse_discovery_response(b"\x00" * 184) is None + + +# --- discover() with patched socket layer --------------------------------- + + +def _make_fake_socket(recv_script): + """Return a Mock socket whose recvfrom plays back ``recv_script``. + + Each script item is either ``bytes`` (returned as ``(data, ("peer", 0))``) + or an Exception (raised). After the script is exhausted, recvfrom raises + socket.timeout to terminate the discover loop. + + """ + fake = mock.MagicMock(spec=socket.socket) + + iterator = iter(recv_script) + + def fake_recvfrom(_bufsize): + try: + item = next(iterator) + except StopIteration: + raise socket.timeout from None + if isinstance(item, BaseException): + raise item + return item, ("203.0.113.1", d.PORT_BROADCAST) + + fake.recvfrom.side_effect = fake_recvfrom + return fake + + +def test_discover_returns_boxes_sorted_by_ip(): + fake = _make_fake_socket( + [ + _build_response(ip="192.0.2.99"), + _build_response( + ip="192.0.2.5", + mac_bytes=bytes.fromhex("00802fdeadbe"), + ), + _build_response( + ip="192.0.2.50", + mac_bytes=bytes.fromhex("00802fcafe01"), + ), + ] + ) + with mock.patch("socket.socket", return_value=fake): + boxes = d.discover(timeout=1.0) + assert [b.ip for b in boxes] == ["192.0.2.5", "192.0.2.50", "192.0.2.99"] + + +def test_discover_deduplicates_by_mac(): + # Same box answers twice (e.g. multi-homed host). With dedup we get one + # entry; with dedup=False we get both. + mac = bytes.fromhex("00802fdeadbe") + fake = _make_fake_socket( + [ + _build_response(ip="192.0.2.5", mac_bytes=mac), + _build_response(ip="192.0.2.5", mac_bytes=mac), + ] + ) + with mock.patch("socket.socket", return_value=fake): + boxes = d.discover(timeout=1.0) + assert len(boxes) == 1 + + +def test_discover_can_disable_deduplication(): + mac = bytes.fromhex("00802fdeadbe") + fake = _make_fake_socket( + [ + _build_response(ip="192.0.2.5", mac_bytes=mac), + _build_response(ip="192.0.2.5", mac_bytes=mac), + ] + ) + with mock.patch("socket.socket", return_value=fake): + boxes = d.discover(timeout=1.0, deduplicate=False) + assert len(boxes) == 2 + + +def test_discover_skips_foreign_datagrams(): + fake = _make_fake_socket( + [ + b"random udp from another device", # rejected by parse + _build_response(ip="192.0.2.5"), + ] + ) + with mock.patch("socket.socket", return_value=fake): + boxes = d.discover(timeout=1.0) + assert [b.ip for b in boxes] == ["192.0.2.5"] + + +def test_discover_tolerates_connection_reset(): + # Windows surfaces ICMP port-unreachable on UDP as ConnectionResetError; + # one such error must not abort the whole scan. + fake = _make_fake_socket( + [ + ConnectionResetError(10054, "WSAECONNRESET"), + _build_response(ip="192.0.2.5"), + ] + ) + with mock.patch("socket.socket", return_value=fake): + boxes = d.discover(timeout=1.0) + assert [b.ip for b in boxes] == ["192.0.2.5"] + + +def test_discover_returns_empty_list_on_timeout(): + fake = _make_fake_socket([]) # immediate timeout + with mock.patch("socket.socket", return_value=fake): + boxes = d.discover(timeout=0.05) + assert boxes == [] + + +def test_discover_sends_probe_to_broadcast_destination(): + fake = _make_fake_socket([]) + with mock.patch("socket.socket", return_value=fake): + d.discover(timeout=0.01, broadcast_addr="203.0.113.255", port=44777) + # One sendto call with the probe frame addressed to the configured + # broadcast + port. + assert fake.sendto.call_count == 1 + sent_bytes, dest = fake.sendto.call_args.args + assert dest == ("203.0.113.255", 44777) + assert len(sent_bytes) == d.FRAME_SIZE + assert sent_bytes[0:2] == d.MAGIC_HEAD + assert sent_bytes[0x02] == d.OP_DISCOVER + + +def test_discover_wraps_bind_error_as_oserror(): + fake = mock.MagicMock(spec=socket.socket) + fake.bind.side_effect = OSError(98, "Address already in use") + with mock.patch("socket.socket", return_value=fake): + with pytest.raises(OSError, match="could not bind UDP port"): + d.discover(timeout=0.1, port=44788) diff --git a/pyvisa_py/testsuite/test_nienet100_session.py b/pyvisa_py/testsuite/test_nienet100_session.py new file mode 100644 index 00000000..d3fea242 --- /dev/null +++ b/pyvisa_py/testsuite/test_nienet100_session.py @@ -0,0 +1,59 @@ +# -*- coding: utf-8 -*- +"""Offline unit tests for the NI GPIB-ENET/100 session layer. + +These cover :mod:`pyvisa_py.nienet100` (the pyvisa Session classes) without +touching the network or a real bridge: the session is built with +``__new__`` and a fake interface so only the session-to-interface plumbing is +exercised. Hardware-gated session tests live in ``nienet100_assisted_tests``. + +:copyright: 2026 by PyVISA-py Authors, see AUTHORS for more details. +:license: MIT, see LICENSE for more details. + +""" + +from pyvisa.constants import StatusCode +from pyvisa_py import gpib_constants, nienet100 as ni +from pyvisa_py.protocols import nienet100 as proto + + +class _FakeInterface: + """Minimal stand-in for EnetConnection that records ibsic calls.""" + + def __init__(self, raise_err: int | None = None) -> None: + self.ibsic_calls = 0 + self._raise_err = raise_err + + def ibsic(self) -> None: + self.ibsic_calls += 1 + if self._raise_err is not None: + raise proto.NIEnet100IOError( + gpib_constants.status.ERR, self._raise_err, "ibsic" + ) + + +def _make_intfc_session(interface) -> ni._NIEnet100IntfcSession: + # Bypass Session.__init__ (which needs a live ResourceManager); gpib_send_ifc + # only touches self.interface. + session = ni._NIEnet100IntfcSession.__new__(ni._NIEnet100IntfcSession) + session.interface = interface + return session + + +def test_gpib_send_ifc_delegates_to_ibsic(): + fake = _FakeInterface() + session = _make_intfc_session(fake) + assert session.gpib_send_ifc() == StatusCode.success + assert fake.ibsic_calls == 1 + + +def test_gpib_send_ifc_maps_wire_error(): + # A wire-level iberr (here ECIC = not controller-in-charge) must surface as + # the matching VISA status, not raise. + fake = _FakeInterface(raise_err=gpib_constants.error.ECIC) + session = _make_intfc_session(fake) + assert session.gpib_send_ifc() == StatusCode.error_not_cic + + +def test_gpib_send_ifc_without_interface_is_connection_lost(): + session = _make_intfc_session(None) + assert session.gpib_send_ifc() == StatusCode.error_connection_lost diff --git a/pyvisa_py/testsuite/test_sessions.py b/pyvisa_py/testsuite/test_sessions.py index 3c645073..c854ccea 100644 --- a/pyvisa_py/testsuite/test_sessions.py +++ b/pyvisa_py/testsuite/test_sessions.py @@ -24,6 +24,7 @@ def test_sessions(self): (InterfaceType.tcpip, "INSTR"), (InterfaceType.tcpip, "SOCKET"), (InterfaceType.prlgx_tcpip, "INTFC"), + (InterfaceType.ni_enet100_tcpip, "INTFC"), (InterfaceType.gpib, "INSTR"), ] exp_missing = []