plesty.lib.traffic

Abstract traffic manager base class and common interface for Plesty devices.

Submodules

Classes

`Preflight`	What one transport found when it looked for an instrument.
`Reach`	Whether the instrument can be reached, as far as the transport knows.
`TrafficManager`	Abstract base class for managing communication with device resources.

Functions

handle_error(→ None)

Handle a caught error by logging and optionally re-raising it.

Package Contents

class plesty.lib.traffic.Preflight

What one transport found when it looked for an instrument.

Variables:

protocol – The transport that produced this — "apt", "visa", "serial", "tcp".
address – The address that was asked for; empty means “any instrument this transport can drive”.
reach – Whether the address resolved. See Reach.
driver – What the OS did with the device, where it could be asked.
resolved – The concrete addresses to open, best match first. Empty unless reach is Reach.READY.
seen – Everything the transport enumerated, matching or not. This is the half of a failure that says what to try instead.
backend – The library doing the enumerating, where there is a choice of one — the VISA backend, for instance. Empty where there is not.
reason – One sentence saying what is wrong and what to do about it. Empty when reach is Reach.READY. Carries no serial number, so it is safe wherever to_dict() is.
stalled – Whether the probe itself never finished. A bus with a wedged device on it stalls enumeration, and the instrument asked for may be perfectly healthy behind it, so this failure means “power-cycle the bus” where every other one means “the address is wrong”.
name – What this result is of, when it is one of several: the sub-device name in a rig, empty for a result standing alone.
parts – The preflights this one is composed of. A rig’s parts are its sub-devices; a sub-device’s part is what its own transport reported from the machine it runs on. This is what makes the three layers one answer instead of three: an experiment asking its rig gets back the bus the stage is missing from, not merely “no answer”.

protocol: str

address: str

reach: Reach

driver: Driver

resolved: tuple[str, Ellipsis] = ()

seen: tuple[Candidate, Ellipsis] = ()

backend: str = ''

reason: str = ''

stalled: bool = False

name: str = ''

parts: tuple[Preflight, Ellipsis] = ()

property overall: Reach

The reach of this result and everything under it, worst first.

One unreachable part makes the whole unreachable: a rig missing a stage is not a rig. An unknown part leaves the whole unknown, because a layer that could not look has not established that anything is there — and saying “ready” on its behalf is exactly the claim that let a server bind against an instrument that was never there.

Return type:: Reach

property ready: bool

Whether this and every part of it resolved.

Return type:: bool

property matched: str | None

The address that will be opened, or None when nothing resolved.

Return type:: str | None

problems() → dict[str, str]

Return the named parts that are unreachable, and why.

The shape the device layer answered in before it spoke this one — sub-device name to a one-line problem — kept because it is what an operator prompt and a log line actually want, and because a caller that only needs “what is broken” should not have to walk a tree.

Return type:: dict[str, str]

summary() → str

Return the operator-facing lines, most useful first.

This is the one place a serial number is printed: the reader is the person at the bench, and telling them which serials are present is most of the diagnosis when the one they asked for is not.

Parts are indented under the result they belong to, so a rig prints as one page: the sub-device that failed, and under it what its own transport saw on the machine it runs on.

Return type:: str

to_dict() → dict[str, Any]

Return the publishable mapping of this result.

Every serial number is dropped, here and in each candidate: this is the projection for anything written to a file in a repository that is world-readable. Port names, resource strings and host names remain — the module’s own configuration already carries them — so a caller publishing an artifact still passes it through that artifact’s own projection, as plesty.lib.test.report_artifact does.

Return type:: dict[str, Any]

classmethod from_dict(document: dict[str, Any]) → Preflight

Rebuild a result from to_dict().

This is the wire format: a device server answers its own transport preflight as a mapping, and the composite that asked rebuilds it and hangs it under the sub-device it belongs to. What comes back has been through to_dict(), so it carries no serial numbers and the candidates it rebuilds have none either.

Parameters:: document (dict[str, Any]) – A mapping as to_dict() produces. Missing keys take their defaults, so a server running an older version — which answers fewer of them — still produces a usable result.
Return type:: Preflight

raise_for_reach() → Preflight

Return self, raising when this or any part of it is unreachable.

Reach.UNKNOWN does not raise. A layer that could not look has not found a problem; it has found nothing, and refusing on that would refuse every transport that cannot enumerate.

Raises:: PreflightError – overall is Reach.UNREACHABLE; the message is summary(), parts included.
Return type:: Preflight

static _scrub(text: str, serials: list[str]) → str

Return text with every serial this report saw taken out of it.

A resolved VISA address is a resource string with the unit’s serial in it, and the address asked for may be the same string. Both are removed by the serials the enumeration reported, which is the only way to know what in an opaque address string is the identifying part.

Parameters:

text (str) – The address or resolved name to clean.
serials (list[str]) – Every serial number seen carries.

Return type:

str

_describe() → str

Return the candidates as one line, serial numbers included.

Return type:: str

class plesty.lib.traffic.Reach

Bases: enum.StrEnum

Whether the instrument can be reached, as far as the transport knows.

Initialize self. See help(type(self)) for accurate signature.

READY = 'ready': The address resolved to something this transport can open.

UNREACHABLE = 'unreachable': Nothing matched, and Preflight.reason says what to do about it.

UNKNOWN = 'unknown'

This transport cannot tell without opening the connection.

Not a failure. A transport with no enumeration to offer answers this, and a caller that refuses to serve on it would refuse every such transport.

plesty.lib.traffic.handle_error(err: Any, error_msg: str | None = None, silent: bool = False, func_name: str = 'Unknown Function') → None

Handle a caught error by logging and optionally re-raising it.

Parameters:

err (Any)
error_msg (str | None)
silent (bool)
func_name (str)

Return type:

None

class plesty.lib.traffic.TrafficManager(address: str, timeout: float = 5)

Bases: abc.ABC

Abstract base class for managing communication with device resources.

Defines common interface for opening, closing, and sending commands. The private methods (_open, _send_command, _close) are implemented by subclasses while the public methods (open, send_command, close) provide consistent resource management and error handling.

Initialize the TrafficManager with the given address and timeout.

A float because a sub-second window is a legitimate ask on a serial bus, where a port sweep pays the timeout per silent port. Transports that count in whole units pass an int, which is a float as far as the type system is concerned, so nothing that already worked changes.

Parameters:

address (str) – The address of the resource.
timeout (float, optional) – The timeout for operations. Defaults to 5. Units are the transport’s: seconds for serial, milliseconds for VISA.

_resource_in_use: dict[str, Any]

PROTOCOL: str = 'unknown'

address

timeout = 5

inst = None

command_log_level: int = 20

timeout_cap: int | None = None

property is_open: bool

Return True if the connection is currently open.

Return type:: bool

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

Report what this transport can see of the instrument, without opening it.

The base class enumerates nothing, so it answers UNKNOWN: this transport cannot tell whether the instrument is there until the connection is opened. That is not a failure, and a caller must not treat it as one — see raise_for_reach().

A transport that can enumerate overrides this and says what it found.

Parameters:: address (str | None) – The address to look for; None uses this manager’s own, and an empty string means “any instrument this transport can drive”.
Return type:: preflight.Preflight

abstractmethod _open(*args, **kwargs)

Open the connection to the resource.

This method should be implemented by subclasses to handle the specific details of opening the connection.

Parameters:

args – Positional arguments for the specific implementation.
kwargs – Keyword arguments for the specific implementation.

open(*args, **kwargs) → bool

Open the connection, delegating to _open and registering the resource.

Return type:: bool

_explain_open_error(exc: BaseException) → str | None

Return guidance for an open failure the user cannot read off exc.

The base class knows nothing about transports, so it returns None and open() reports the bare failure. A subclass whose backend reports the wrong port and a held port with the same exception type overrides this to say which one it was and what to do about it; the text is appended to the raised message, not buried in the cause chain.

Parameters:: exc (BaseException) – The exception _open() raised.
Return type:: str | None

abstractmethod _send_command(command: str, timeout=None, *args, **kwargs) → Any

Send a command to the resource and return the response.

Parameters:

command (str) – The command to send.
timeout – Timeout in seconds; defaults to the instance’s timeout.
*args – Additional positional arguments for subclass implementations.
**kwargs – Additional keyword arguments for subclass implementations.

Returns:

The response from the resource, or False if the command failed.

Return type:

Any

orphan(command: str) → bool

Transmit command and never read what comes back.

Not a write. A write is a command that sets something, and plenty of instruments acknowledge one — those replies are read, and must be, or the acknowledgement becomes the stale answer handed to the next command. This reads nothing whatever the command means, which is the one thing send_command() cannot express and exactly the condition the buffer-drain gate has to create: a reply left in the instrument’s buffer, as a timed-out query leaves one.

Deliberately narrow. Racing a timeout is the alternative and it is not a lever on every driver — NI-VISA does not enforce short timeouts on some instruments, so the gate silently tested nothing there.

Returns:: True when the command was transmitted, False when this transport cannot transmit without reading — in which case the drain gate skips and says so rather than reporting a pass.
Parameters:: command (str)
Return type:: bool

send_command(command: str, timeout=None, *args, **kwargs) → Any

Send a command, delegating to _send_command with error handling.

Parameters:: command (str)
Return type:: Any

abstractmethod _close()

Close the connection to the resource.

Subclasses must implement this to handle the specific close logic.

close() → bool

Close the connection, unregister the resource, and return success status.

The resource is unregistered even when _close fails. Otherwise a transport that throws on the way out — a clear refused by an unresponsive instrument, a close on a dead handle — leaves the address registered for the life of the process, and every later open is rejected with “already in use” against a resource nothing is holding.

Return type:: bool

__enter__(): Open the resource connection when entering the context manager.

__exit__(exc_type, exc_val, exc_tb): Close the resource connection when exiting the context manager.