plesty.lib.traffic.tcp_ip

TCP/IP traffic manager for Plesty device communication.

Classes

TCPIPTrafficManager

A TrafficManager implementation for managing TCP/IP communication.

Functions

_endpoint(→ str)

Return host:port for a resolved address, whatever family it is.

Module Contents

class plesty.lib.traffic.tcp_ip.TCPIPTrafficManager(host, port, timeout=5, write_termination='\r\n', read_termination='\r\n')

Bases: plesty.lib.traffic.TrafficManager

A TrafficManager implementation for managing TCP/IP communication.

This class provides methods to open, close, and communicate with instruments over TCP/IP.

Initialize with host, port, timeout, and termination characters.

PROTOCOL = 'tcp'

host

port

write_termination = Multiline-String

Show Value

"""
"""

read_termination = Multiline-String

Show Value

"""
"""

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

Report whether the host resolves and the port accepts a connection.

There is nothing to enumerate on a network, so the two questions a network answers are asked directly. They separate the three failures that a connection error reports as one: a name that does not resolve (wrong host, or no DNS), a host that refuses the port (the instrument’s software is not running, or listens elsewhere), and a host that says nothing at all (powered off, or a firewall dropping the packets).

The connection made here is opened and immediately closed, before any protocol traffic. It is the one preflight that touches the instrument, because on a network there is no descriptor to read instead — and a TCP handshake is what the transport would do anyway.

Parameters:: address (str | None) – "host:port" to check instead of this manager’s own; None uses the configured host and port.
Return type:: plesty.lib.traffic.preflight.Preflight

_open(*args, **kwargs)

Open the TCP/IP connection to the instrument.

Parameters:

args – Positional arguments for the specific implementation (not used in this case).
kwargs – Keyword arguments for the specific implementation (not used in this case).

_close()

Close the connection to the resource.

Subclasses must implement this to handle the specific close logic.

receive_one(timeout: float | None = None) → str | bool

We leave this function public for direct use in some cases where the caller needs more control over the receiving process (e.g. in some requests, two responses might be expected, one for instant feedback and one for the final report when the operation is complete).

Parameters:: timeout (float | None)
Return type:: str | bool

orphan(command: str) → bool

Transmit command without receiving its answer.

Parameters:: command (str) – A command the peer will answer.
Returns:: True once transmitted, False when the socket is not open.
Return type:: bool

_send_command(command: str, timeout=None, expect_response=True)

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

plesty.lib.traffic.tcp_ip._endpoint(sockaddr: tuple[Any, Ellipsis]) → str

Return host:port for a resolved address, whatever family it is.

getaddrinfo types the address as a family-dependent tuple: a string host for IPv4 and IPv6, and bytes for the link-layer families nothing here asks for. Decoding the one case that cannot appear costs a line and keeps the annotation honest.

Parameters:: sockaddr (tuple[Any, Ellipsis]) – The fifth element of a getaddrinfo entry.
Return type:: str