plesty.lib.traffic.apt

Traffic manager for the binary Thorlabs APT protocol over a serial port.

Attributes

_POLL_INTERVAL	Delay between read attempts while waiting for a reply, in seconds.
_MAX_UNMATCHED	Maximum number of retained unsolicited messages.
_PURGE_DWELL	Settle time around buffer purges during the fresh-session cleanup, in seconds.
APT_VENDOR_ID	USB vendor id of the FTDI bridge every Thorlabs APT controller enumerates
APT_DESCRIPTION_HINT	Substring Windows reports for an APT controller ("APT USB Device Serial

Classes

AptTrafficManager

Serial traffic manager speaking the binary APT protocol.

Functions

_apt_candidates(→ list[Any])	Return the ports that could be an APT controller, best match first.
_normalise_serial(→ str)	Return a serial number in a form two platforms can be compared by.
_serial_matches(→ bool)	Return True when port is the controller with the requested serial.
_describe(→ str)	Return a readable summary of candidate ports, for an error message.

Module Contents

plesty.lib.traffic.apt._POLL_INTERVAL = 0.005

Delay between read attempts while waiting for a reply, in seconds.

plesty.lib.traffic.apt._MAX_UNMATCHED = 64

Maximum number of retained unsolicited messages.

plesty.lib.traffic.apt._PURGE_DWELL = 0.05

Settle time around buffer purges during the fresh-session cleanup, in seconds.

plesty.lib.traffic.apt.APT_VENDOR_ID = 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.

plesty.lib.traffic.apt.APT_DESCRIPTION_HINT = '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.

class plesty.lib.traffic.apt.AptTrafficManager(port: str, timeout: int = 5, baudrate: int = 115200, transport: Any | None = None)

Bases: plesty.lib.traffic.TrafficManager

Serial traffic manager speaking the binary APT protocol.

Sends 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.

Parameters:

• port (str) – The serial port to connect to (e.g. ‘COM3’ or ‘/dev/ttyUSB0’).

• timeout (int) – Default reply timeout in seconds.

• baudrate (int) – Serial baud rate; APT devices use 115200.

• transport (Any | None) – Optional pre-built serial-like object (used by simulators in tests); when given, open() uses it instead of opening a real serial port.

port

baudrate = 115200

_transport = None

_recv_state

_unmatched: collections.deque[plesty.lib.traffic.apt_protocol.AptMessage]

command_log_level = 10

property is_open: bool

Return True if the serial connection is open.

Return type:

bool

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.

Parameters:

address (str) – 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.

Return type:

list[str]

_open(*args: Any, **kwargs: Any) → None

Open the serial connection (or adopt the injected transport).

Parameters:

• args (Any)

• kwargs (Any)

Return type:

None

_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.

Parameters:

exc (BaseException)

Return type:

str | None

_close() → None

Close the serial connection.

Return type:

None

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

Parameters:

settle (float) – Time to wait for in-flight bytes to arrive, in seconds.

Return type:

None

_write_message(message: plesty.lib.traffic.apt_protocol.AptMessage) → None

Write one message to the wire.

Parameters:

message (plesty.lib.traffic.apt_protocol.AptMessage)

Return type:

None

_pump() → None

Drain available serial input into the framing state machine.

Return type:

None

static _matches(request: plesty.lib.traffic.apt_protocol.AptMessage, reply_id: plesty.lib.traffic.apt_protocol.AptMessageId, candidate: plesty.lib.traffic.apt_protocol.AptMessage) → bool

Return True if a received message answers the given request.

Parameters:

• request (plesty.lib.traffic.apt_protocol.AptMessage)

• reply_id (plesty.lib.traffic.apt_protocol.AptMessageId)

• candidate (plesty.lib.traffic.apt_protocol.AptMessage)

Return type:

bool

_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.

Parameters:

• request (plesty.lib.traffic.apt_protocol.AptMessage)

• reply_id (plesty.lib.traffic.apt_protocol.AptMessageId)

Return type:

plesty.lib.traffic.apt_protocol.AptMessage | None

_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.

Parameters:

• request (plesty.lib.traffic.apt_protocol.AptMessage)

• reply_id (plesty.lib.traffic.apt_protocol.AptMessageId)

• timeout (float)

Return type:

plesty.lib.traffic.apt_protocol.AptMessage | None

_send_command(command: Any, timeout: float | None = None, reply_id: plesty.lib.traffic.apt_protocol.AptMessageId | None = None) → Any

Send an AptMessage; wait for a reply when reply_id is given.

Parameters:

• command (Any) – The AptMessage to send.

• timeout (float | None) – Reply timeout in seconds; defaults to the instance timeout.

• reply_id (plesty.lib.traffic.apt_protocol.AptMessageId | None) – Message id of the expected reply, or None for a fire-and-forget command.

Returns:

The reply AptMessage for requests, None for commands.

Return type:

Any

command(message: plesty.lib.traffic.apt_protocol.AptMessage) → bool

Send a fire-and-forget message; return True unless sending failed.

Parameters:

message (plesty.lib.traffic.apt_protocol.AptMessage)

Return type:

bool

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.

Parameters:

• message (plesty.lib.traffic.apt_protocol.AptMessage)

• reply_id (plesty.lib.traffic.apt_protocol.AptMessageId)

• timeout (float | None)

Return type:

plesty.lib.traffic.apt_protocol.AptMessage | None

plesty.lib.traffic.apt._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.

Parameters:

ports (list[Any]) – Entries from serial.tools.list_ports.comports().

Return type:

list[Any]

plesty.lib.traffic.apt._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.

Parameters:

value (str | None) – A serial number as reported by pyserial or typed by a user.

Return type:

str

plesty.lib.traffic.apt._serial_matches(requested: str, port: Any) → bool

Return True when port is the controller with the requested serial.

Parameters:

• requested (str) – The serial number asked for.

• port (Any) – An entry from serial.tools.list_ports.comports().

Return type:

bool

plesty.lib.traffic.apt._describe(ports: list[Any]) → str

Return a readable summary of candidate ports, for an error message.

Parameters:

ports (list[Any]) – Entries from serial.tools.list_ports.comports().

Return type:

str