hvl_ccb.comm.tcp

TCP communication protocol. Makes use of the socket library.

class Tcp(*args, **kwargs)

Bases: TcpCommunication

Old name Tcp for TcpCommunication for keeping this import backwards compatibile

class TcpCommunication(configuration)

Bases: AsyncCommunicationProtocol

Tcp Communication Protocol.

close() → None

Close TCP connection.

static config_cls() → type[TcpCommunicationConfig]

Return the default configdataclass class.

Returns:

a reference to the default configdataclass class

property is_open: bool

Is the connection open?

Returns:

True for an open connection

open() → None

Open TCP connection.

read_bytes() → bytes

Read a single line as bytes from the TcpCommunication.

This method uses self.access_lock to ensure thread-safety.

Returns:

a single line as bytes containing the terminator, which can also be empty b””

write_bytes(data: bytes) → int

Write data as bytes to the TcpCommunication.

This method uses self.access_lock to ensure thread-safety.

Parameters:

data – data as bytes-string to be written

Returns:

number of bytes written

class TcpCommunicationConfig(terminator: bytes = b'\r\n', encoding: str = 'utf-8', encoding_error_handling: str = 'strict', wait_sec_read_text_nonempty: int | float = 0.5, default_n_attempts_read_text_nonempty: int = 10, host: str | IPv4Address | IPv6Address | None = None, port: int = 54321, bufsize: int = 1024, timeout: float | None = 0.2)

Bases: AsyncCommunicationProtocolConfig

Configuration dataclass for TcpCommunication.

bufsize: int = 1024

clean_values() → None

force_value(fieldname, value)

Forces a value to a dataclass field despite the class being frozen.

NOTE: you can define post_force_value method with same signature as this method to do extra processing after value has been forced on fieldname.

Parameters:

• fieldname – name of the field

• value – value to assign

host: str | IPv4Address | IPv6Address | None = None

classmethod keys() → Sequence[str]

Returns a list of all configdataclass fields key-names.

Returns:

a list of strings containing all keys.

classmethod optional_defaults() → dict[str, object]

Returns a list of all configdataclass fields, that have a default value assigned and may be optionally specified on instantiation.

Returns:

a list of strings containing all optional keys.

port: int = 54321

classmethod required_keys() → Sequence[str]

Returns a list of all configdataclass fields, that have no default value assigned and need to be specified on instantiation.

Returns:

a list of strings containing all required keys.

timeout: float | None = 0.2

exception TcpCommunicationError

Bases: CommunicationError

Error of the TcpCommunication