Traffic Manager

TrafficManager is the communication layer between your device class and the physical transport (Serial, VISA, TCP/IP, and others).

It standardizes connection handling and command dispatch so device implementations can focus on device behavior instead of low-level I/O details.

What It Does

A traffic manager is responsible for:

Opening and closing a connection to a device resource.
Sending one command and returning one response.
Applying timeout behavior.
Providing consistent error handling through a shared interface.
Supporting context manager usage (with ...) for safe cleanup.

In PlestyLib, the abstract base class is plesty.lib.traffic.TrafficManager.

Lifecycle and Interface

The public API is:

open(...)
send_command(command, timeout=None, ...)
close()

When you implement a custom transport, you override three private hooks:

_open(...)
_send_command(command, timeout=None, ...)
_close()

The base class wraps these hooks with common behaviors:

In-use resource protection (same address cannot be opened by multiple manager instances).
Error capture and reporting.
__enter__ and __exit__ support.

Preflight: is the instrument there at all?

preflight(address=None) asks what the transport can see without opening anything, and returns a plesty.lib.traffic.preflight.Preflight: whether the address resolved (reach), to what (resolved), what else is present (seen), and — where the transport can ask the device node — whether the operating system bound a driver at all (driver). A module calls it before it serves, so a server does not bind against an instrument that was never there.

report = manager.preflight()
if report.reach is Reach.UNREACHABLE:
    raise SystemExit(report.summary())   # the operator's copy, serials included
logging.info(report.reason or f"instrument at {report.matched}")

Three things to know when you consume it:

Reach.UNKNOWN is not a failure. A transport with nothing to enumerate answers it, and refusing to serve on it would refuse every such transport. raise_for_reach() follows that rule: only UNREACHABLE raises.
summary() is for the operator and prints serial numbers. to_dict() is for anything written to a file, and carries none — not as a field, and not embedded in a resource string. A hub repository is world-readable.
discover_instrument on the APT and VISA transports is this same lookup in the shape those callers already take; it did not change signature.

Overriding it is optional: a transport that cannot enumerate inherits the UNKNOWN answer and needs no code.

This is the transport level of a check that spans three layers — the experiment asks its rig, the rig asks each device server, and a server asks its own transport. Preflight has the whole picture, including how the result nests and why unknown is not a failure.

Built-in Implementations

Common built-in traffic managers include:

SerialTrafficManager for serial ports.
VisaTrafficManager for VISA resources.
TCPIPTrafficManager for TCP/IP connections.

AptTrafficManager for Thorlabs APT devices — binary frames over a serial port rather than terminated text, so it takes and returns AptMessage objects (plesty.lib.traffic.apt_protocol) instead of strings. plesty.lib.device.base_apt_device.BaseAptDevice builds on it, and plesty.lib.sim.apt.AptMotorSimulator answers real protocol frames when no motor is attached.

The address is a port name, the controller’s serial number, or d2xx:<serial>. On Windows, Thorlabs Kinesis binds a controller to FTDI’s D2XX driver, which gives it no COM port at all. Such a controller is opened through that driver instead, by calling FTDI’s own ftd2xx.dll, which comes with Kinesis or FTDI’s driver package; nothing extra is installed in Python:

Address	Opens
`COM9`, `/dev/ttyUSB0`	that port
`55001327` (serial number)	the COM port carrying it, else the D2XX device with it
`d2xx:55001327`	the D2XX device only
empty	any APT COM port, else the one D2XX device whose serial starts with the device’s `SERIAL_PREFIX`

Preflight lists D2XX devices next to the COM ports, as d2xx:<serial>. Opening one needs Kinesis closed: it holds the device while it runs.

Demo 1: Use the Built-in SerialTrafficManager

from plesty.lib.traffic.serial import SerialTrafficManager


manager = SerialTrafficManager(
	port="/dev/ttyUSB0",   # Example on Linux
	baudrate=9600,
	timeout=5,
	write_termination="\r",
	read_termination="\r\n",
)

try:
	opened = manager.open(parity="none", stopbits="one", bytesize=8)
	if not opened:
		raise RuntimeError("Failed to open serial port")

	# Command format depends on your device protocol.
	response = manager.send_command("MEAS:VOLT?")
	print("Response:", response)

finally:
	manager.close()

Notes for SerialTrafficManager:

It appends write_termination automatically.
It reads until read_termination.
Current response logic treats lines containing OK as successful.

Demo 2: Implement Your Own Serial Traffic Manager

Use this pattern if your serial device protocol differs from the default behavior.

from typing import Any
import serial

from plesty.lib.traffic import TrafficManager


class MySerialTrafficManager(TrafficManager):
	def __init__(self, port: str, baudrate: int = 115200, timeout: int = 3):
		super().__init__(address=port, timeout=timeout)
		self.port = port
		self.baudrate = baudrate

	def _open(self) -> None:
		self.inst = serial.Serial(
			port=self.port,
			baudrate=self.baudrate,
			timeout=self.timeout,
			parity=serial.PARITY_NONE,
			stopbits=serial.STOPBITS_ONE,
			bytesize=serial.EIGHTBITS,
		)
		self.inst.reset_input_buffer()
		self.inst.reset_output_buffer()

	def _send_command(self, command: str, timeout=None) -> Any:
		if timeout is not None:
			self.inst.timeout = timeout

		wire = (command + "\n").encode("utf-8")
		self.inst.write(wire)

		# Read one full line response for this example protocol.
		raw = self.inst.readline()
		if not raw:
			return False

		return raw.decode("utf-8", errors="replace").strip()

	def _close(self) -> None:
		self.inst.close()

Then use it through the standard public API:

tm = MySerialTrafficManager("/dev/ttyUSB0", baudrate=115200)

with tm:
	print(tm.send_command("*IDN?"))

Integration with Device Classes

In a synchronized device implementation, traffic managers are usually called inside low-level query and write paths:

Device receives a standardized operation.
Command solver builds the protocol command string.
Traffic manager sends/receives bytes or text.
Command solver decodes response into standardized value.

This separation keeps each layer simple and testable.

Best Practices

Keep transport-specific logic inside traffic managers only.
Keep protocol formatting in command solvers, not in traffic managers.
Always use with manager: or try/finally to guarantee close().
Use explicit timeout values suitable for your device.
Validate port/resource settings early in _open().