plesty.lib.traffic.apt ====================== .. py:module:: plesty.lib.traffic.apt .. autoapi-nested-parse:: Traffic manager for the binary Thorlabs APT protocol over a serial port. Attributes ---------- .. autoapisummary:: plesty.lib.traffic.apt._POLL_INTERVAL plesty.lib.traffic.apt._MAX_UNMATCHED plesty.lib.traffic.apt._PURGE_DWELL plesty.lib.traffic.apt.APT_VENDOR_ID plesty.lib.traffic.apt.APT_DESCRIPTION_HINT Classes ------- .. autoapisummary:: plesty.lib.traffic.apt.AptTrafficManager Functions --------- .. autoapisummary:: plesty.lib.traffic.apt._apt_candidates plesty.lib.traffic.apt._normalise_serial plesty.lib.traffic.apt._serial_matches plesty.lib.traffic.apt._describe Module Contents --------------- .. py:data:: _POLL_INTERVAL :value: 0.005 Delay between read attempts while waiting for a reply, in seconds. .. py:data:: _MAX_UNMATCHED :value: 64 Maximum number of retained unsolicited messages. .. py:data:: _PURGE_DWELL :value: 0.05 Settle time around buffer purges during the fresh-session cleanup, in seconds. .. py:data:: APT_VENDOR_ID :value: 1027 USB vendor id of the FTDI bridge every Thorlabs APT controller enumerates behind. This is the filter; the product id deliberately is not, because APT controllers differ by model and an allowlist would exclude the next one. .. py:data:: APT_DESCRIPTION_HINT :value: 'apt' Substring Windows reports for an APT controller ("APT USB Device Serial Port"). Only used to rank candidates when no serial number was given — the serial number is what actually identifies a controller. .. py:class:: AptTrafficManager(port: str, timeout: int = 5, baudrate: int = 115200, transport: Any | None = None) Bases: :py:obj:`plesty.lib.traffic.TrafficManager` Serial traffic manager speaking the binary APT protocol. Sends :class:`AptMessage` commands and matches device replies by message id and source/destination address. Unsolicited messages (for example periodic status updates) are retained in a bounded queue and consumed by the next matching request. APT devices enumerate as USB serial ports at 115200 baud, 8N1. Initialize with a serial port name and communication settings. :param port: The serial port to connect to (e.g. 'COM3' or '/dev/ttyUSB0'). :param timeout: Default reply timeout in seconds. :param baudrate: Serial baud rate; APT devices use 115200. :param transport: Optional pre-built serial-like object (used by simulators in tests); when given, ``open()`` uses it instead of opening a real serial port. .. py:attribute:: port .. py:attribute:: baudrate :value: 115200 .. py:attribute:: _transport :value: None .. py:attribute:: _recv_state .. py:attribute:: _unmatched :type: collections.deque[plesty.lib.traffic.apt_protocol.AptMessage] .. py:attribute:: command_log_level :value: 10 .. py:property:: is_open :type: bool Return True if the serial connection is open. .. py:method:: discover_instrument(address: str) -> list[str] | bool Resolve *address* to the serial port the stage is actually on. A COM number is not a stable identity: Windows assigns it per USB port, so replugging a stage renames it and a hardcoded port then opens whatever took its place — a real risk on a host carrying several APT devices. The controller's serial number does not move. Identification uses USB descriptor metadata only; no port is opened. That matters more than it sounds: probing ports to find out what they are means opening ports another device server holds, and on this platform that has hung processes uninterruptibly. :param address: A port name (``"COM9"``, ``"/dev/ttyUSB0"``) to use as-is; a controller serial number (``"55001327"``) to look up; or empty to find any APT controller present. :returns: Matching port names, most specific first. bool: ``False`` when nothing matched. :rtype: list[str] .. py:method:: _open(*args: Any, **kwargs: Any) -> None Open the serial connection (or adopt the injected transport). .. py:method:: _explain_open_error(exc: BaseException) -> str | None Say whether the port was held or forbidden rather than absent. On the K10CR1 rig the port was right and merely open in Kinesis; the bare ``Access is denied`` sent the user through driver checks instead. .. py:method:: _close() -> None Close the serial connection. .. py:method:: flush_stale(settle: float = _PURGE_DWELL) -> None Wait briefly, then discard buffered input and retained messages. Used at connect time to drop traffic a previous session left in flight (for example a still-streaming status-update subscription). :param settle: Time to wait for in-flight bytes to arrive, in seconds. .. py:method:: _write_message(message: plesty.lib.traffic.apt_protocol.AptMessage) -> None Write one message to the wire. .. py:method:: _pump() -> None Drain available serial input into the framing state machine. .. py:method:: _matches(request: plesty.lib.traffic.apt_protocol.AptMessage, reply_id: plesty.lib.traffic.apt_protocol.AptMessageId, candidate: plesty.lib.traffic.apt_protocol.AptMessage) -> bool :staticmethod: Return True if a received message answers the given request. .. py:method:: _take_match(request: plesty.lib.traffic.apt_protocol.AptMessage, reply_id: plesty.lib.traffic.apt_protocol.AptMessageId) -> plesty.lib.traffic.apt_protocol.AptMessage | None Remove and return the first retained message answering the request. .. py:method:: _wait_for(request: plesty.lib.traffic.apt_protocol.AptMessage, reply_id: plesty.lib.traffic.apt_protocol.AptMessageId, timeout: float) -> plesty.lib.traffic.apt_protocol.AptMessage | None Poll the wire until the requested reply arrives or the timeout expires. .. py:method:: _send_command(command: Any, timeout: float | None = None, reply_id: plesty.lib.traffic.apt_protocol.AptMessageId | None = None) -> Any Send an :class:`AptMessage`; wait for a reply when ``reply_id`` is given. :param command: The :class:`AptMessage` to send. :param timeout: Reply timeout in seconds; defaults to the instance timeout. :param reply_id: Message id of the expected reply, or ``None`` for a fire-and-forget command. :returns: The reply :class:`AptMessage` for requests, ``None`` for commands. .. py:method:: command(message: plesty.lib.traffic.apt_protocol.AptMessage) -> bool Send a fire-and-forget message; return True unless sending failed. .. py:method:: request(message: plesty.lib.traffic.apt_protocol.AptMessage, reply_id: plesty.lib.traffic.apt_protocol.AptMessageId, timeout: float | None = None) -> plesty.lib.traffic.apt_protocol.AptMessage | None Send a message and wait for the matching reply. :returns: The reply message, or ``None``/``False``-ish if the request failed. .. py:function:: _apt_candidates(ports: list[Any]) -> list[Any] Return the ports that could be an APT controller, best match first. APT controllers share the FTDI vendor id but differ in product id by model, so the vendor id is the filter and the product id is ignored — matching on a product id allowlist would silently exclude a controller the list has not met yet. That filter is deliberately loose: unrelated FTDI adapters pass it too. Ranking puts anything whose description mentions APT first, which is what Windows reports for these controllers, but the honest discriminator is the serial number and this ordering only decides which candidate is offered when no serial was given. :param ports: Entries from ``serial.tools.list_ports.comports()``. .. py:function:: _normalise_serial(value: str | None) -> str Return a serial number in a form two platforms can be compared by. Windows appends the FTDI channel letter to the device serial — the stage reported as ``…+55001327A`` has serial ``55001327`` — and pyserial passes that through, so a serial copied from Device Manager would never match one read on Linux. Case and the channel suffix are dropped from both sides. :param value: A serial number as reported by pyserial or typed by a user. .. py:function:: _serial_matches(requested: str, port: Any) -> bool Return ``True`` when *port* is the controller with the requested serial. :param requested: The serial number asked for. :param port: An entry from ``serial.tools.list_ports.comports()``. .. py:function:: _describe(ports: list[Any]) -> str Return a readable summary of candidate ports, for an error message. :param ports: Entries from ``serial.tools.list_ports.comports()``.