plesty.lib.traffic.apt

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

A controller with no serial port — bound to FTDI’s D2XX driver, as Thorlabs Kinesis leaves it on Windows — is reached through plesty.lib.traffic.apt_d2xx instead, under the address d2xx:<serial>.

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
`APT_HARDWARE_ID`	PnP hardware id the USB bus is asked about when no port matched. The vendor

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.
`_find_d2xx`(→ plesty.lib.traffic.apt_d2xx.D2xxDevice \| None)	Return the D2XX device carrying serial, or `None`.
`_d2xx_hint`(→ str)	Return the sentence pointing at the D2XX driver, where it would help.
`_d2xx_reason`(→ str)	Say why a `d2xx:<serial>` address found nothing.
`_reason`(→ str)	Compose the one line that says what is wrong and what to do about it.
`_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.

plesty.lib.traffic.apt.APT_HARDWARE_ID = 'USB\\VID_0403': PnP hardware id the USB bus is asked about when no port matched. The vendor id only, for the same reason the port filter uses it: APT controllers differ in product id by model, and an allowlist would exclude the next one. It is the looser question — an unrelated FTDI adapter answers it too — but it is asked only once every port has already failed to match, where the alternative is saying nothing at all about the driver.

class plesty.lib.traffic.apt.AptTrafficManager(port: str, timeout: int = 5, baudrate: int = 115200, transport: Any | None = None, serial_prefix: str | 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 — or, bound to the D2XX driver, as no port at all, and are then opened through D2XX.

Initialize with a serial port name and communication settings.

Parameters:

port (str) – The serial port to connect to (e.g. ‘COM3’ or ‘/dev/ttyUSB0’), a controller serial number, or d2xx:<serial>.
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.
serial_prefix (str | None) – Overrides SERIAL_PREFIX for this manager.

PROTOCOL = 'apt'

SERIAL_PREFIX = ''

port

baudrate = 115200

_transport = None

serial_prefix = ''

_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

preflight(address: str | None = None) → plesty.lib.traffic.preflight.Preflight

Report which port the stage is on, or why none of them is it.

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.

Where nothing matched, the USB bus is asked whether an FTDI device is there at all. A controller whose driver never bound has no port, so the serial layer sees exactly what it sees for a controller that is unplugged — and the two need different fixes.

Parameters:: address (str | None) – A port name ("COM9", "/dev/ttyUSB0") to use as-is; a controller serial number ("55001327") to look up; None for this manager’s own; or empty for any APT controller present.
Return type:: plesty.lib.traffic.preflight.Preflight

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.

The lookup itself is preflight(); this is the answer in the shape callers already take, and the reason a failure still reaches the log for anyone reading one.

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, the D2XX device, 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. A D2XX device that is listed but will not open is held the same way.

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._find_d2xx(serial: str, devices: list[plesty.lib.traffic.apt_d2xx.D2xxDevice]) → plesty.lib.traffic.apt_d2xx.D2xxDevice | None

Return the D2XX device carrying serial, or None.

Parameters:

serial (str) – The serial number asked for.
devices (list[plesty.lib.traffic.apt_d2xx.D2xxDevice]) – What the D2XX driver lists.

Return type:

plesty.lib.traffic.apt_d2xx.D2xxDevice | None

plesty.lib.traffic.apt._d2xx_hint() → str

Return the sentence pointing at the D2XX driver, where it would help.

On Windows a controller Kinesis has bound to the D2XX driver has no port; without FTDI’s D2XX library that looks exactly like no controller at all.

Return type:: str

plesty.lib.traffic.apt._d2xx_reason(address: str, devices: list[plesty.lib.traffic.apt_d2xx.D2xxDevice]) → str

Say why a d2xx:<serial> address found nothing.

Parameters:

address (str) – The address asked for.
devices (list[plesty.lib.traffic.apt_d2xx.D2xxDevice]) – What the D2XX driver lists.

Return type:

str

plesty.lib.traffic.apt._reason(address: str, ports: list[Any], candidates: list[Any], driver: plesty.lib.traffic.preflight.Driver, d2xx: list[plesty.lib.traffic.apt_d2xx.D2xxDevice] | None = None) → str

Compose the one line that says what is wrong and what to do about it.

Carries no serial number: it travels into to_dict(), and the serials present are printed by summary() where the operator is the only reader.

Parameters:

address (str) – The address that was asked for, empty for “any”.
ports (list[Any]) – Every serial port present.
candidates (list[Any]) – The APT-capable ports among them.
driver (plesty.lib.traffic.preflight.Driver) – What the USB bus said, where it was asked.
d2xx (list[plesty.lib.traffic.apt_d2xx.D2xxDevice] | None) – The controllers the D2XX driver lists.

Return type:

str

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