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:

1. Opening and closing a connection to a device resource.

2. Sending one command and returning one response.

3. Applying timeout behavior.

4. Providing consistent error handling through a shared interface.

5. Supporting context manager usage (with ...) for safe cleanup.

In PlestyLib, the abstract base class is plestylib.traffic.TrafficManager.

Lifecycle and Interface

The public API is:

1. open(...)

2. send_command(command, timeout=None, ...)

3. close()

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

1. _open(...)

2. _send_command(command, timeout=None, ...)

3. _close()

The base class wraps these hooks with common behaviors:

1. In-use resource protection (same address cannot be opened by multiple manager instances).

2. Error capture and reporting.

3. __enter__ and __exit__ support.

Built-in Implementations

Common built-in traffic managers include:

1. SerialTrafficManager for serial ports.

2. VisaTrafficManager for VISA resources.

3. TcpIpTrafficManager for TCP/IP connections.

Demo 1: Use the Built-in SerialTrafficManager

from plestylib.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:

1. It appends write_termination automatically.

2. It reads until read_termination.

3. 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 plestylib.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:

1. Device receives a standardized operation.

2. Command solver builds the protocol command string.

3. Traffic manager sends/receives bytes or text.

4. Command solver decodes response into standardized value.

This separation keeps each layer simple and testable.

Best Practices

1. Keep transport-specific logic inside traffic managers only.

2. Keep protocol formatting in command solvers, not in traffic managers.

3. Always use with manager: or try/finally to guarantee close().

4. Use explicit timeout values suitable for your device.

5. Validate port/resource settings early in _open().