hvl_ccb.comm.base

Module with base classes for communication protocols.

class AsyncCommunicationProtocol(config)

Bases: CommunicationProtocol

Abstract base class for asynchronous communication protocols

static config_cls() → type[AsyncCommunicationProtocolConfig]

Return the default configdataclass class.

Returns:

a reference to the default configdataclass class

read() → str

Read a single line of text as str from the communication.

Returns:

text as str including the terminator, which can also be empty “”

read_all(n_attempts_max: int | None = None, attempt_interval_sec: int | float | None = None) → str | None

Read all lines of text from the connection till nothing is left to read.

Parameters:

• n_attempts_max – Amount of attempts how often a non-empty text is tried to be read

• attempt_interval_sec – time between the reading attempts

Returns:

A multi-line str including the terminator internally

abstract read_bytes() → bytes

Read a single line as bytes from the communication.

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””

read_nonempty(n_attempts_max: int | None = None, attempt_interval_sec: int | float | None = None) → str | None

Try to read a non-empty single line of text as str from the communication. If the host does not reply or reply with white space only, it will return None.

Returns:

a non-empty text as a str or None in case of an empty string

Parameters:

• n_attempts_max – Amount of attempts how often a non-empty text is tried to be read

• attempt_interval_sec – time between the reading attempts

read_text() → str

Read one line of text from the serial port. The input buffer may hold additional data afterwards, since only one line is read.

NOTE: backward-compatibility proxy for read method; to be removed in v1.0

Returns:

String read from the serial port; ‘’ if there was nothing to read.

Raises:

SerialCommunicationIOError – when communication port is not opened

read_text_nonempty(n_attempts_max: int | None = None, attempt_interval_sec: int | float | None = None) → str | None

Reads from the serial port, until a non-empty line is found, or the number of attempts is exceeded.

NOTE: backward-compatibility proxy for read method; to be removed in v1.0

Attention: in contrast to read_text, the returned answer will be stripped of a whitespace newline terminator at the end, if such terminator is set in the initial configuration (default).

Parameters:

• n_attempts_max – maximum number of read attempts

• attempt_interval_sec – time between the reading attempts

Returns:

String read from the serial port; ‘’ if number of attempts is exceeded or serial port is not opened.

write(text: str)

Write text as str to the communication.

Parameters:

text – test as a str to be written

abstract write_bytes(data: bytes) → int

Write data as bytes to the communication.

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

Parameters:

data – data as bytes-string to be written

Returns:

number of bytes written

write_text(text: str)

Write text to the serial port. The text is encoded and terminated by the configured terminator.

NOTE: backward-compatibility proxy for read method; to be removed in v1.0

Parameters:

text – Text to send to the port.

Raises:

SerialCommunicationIOError – when communication port is not opened

class AsyncCommunicationProtocolConfig(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)

Bases: object

Base configuration data class for asynchronous communication protocols

clean_values()

default_n_attempts_read_text_nonempty: int = 10

default number of attempts to read a non-empty text

encoding: str = 'utf-8'

Standard encoding of the connection. Typically this is utf-8, but can also be latin-1 or something from here: https://docs.python.org/3/library/codecs.html#standard-encodings

encoding_error_handling: str = 'strict'

Encoding error handling scheme as defined here: https://docs.python.org/3/library/codecs.html#error-handlers By default strict error handling that raises UnicodeError.

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

is_configdataclass = True

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.

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.

terminator: bytes = b'\r\n'

The terminator character. Typically this is b'\r\n' or b'\n', but can also be b'\r' or other combinations. This defines the end of a single line.

wait_sec_read_text_nonempty: int | float = 0.5

time to wait between attempts of reading a non-empty text

exception CommunicationError

Bases: CCBError

class CommunicationProtocol(config)

Bases: ConfigurationMixin, ABC

Communication protocol abstract base class.

Specifies the methods to implement for communication protocol, as well as implements some default settings and checks.

access_lock

Access lock to use with context manager when accessing the communication protocol (thread safety)

abstract close()

Close the communication protocol

abstract open()

Open communication protocol

class NullCommunicationProtocol(config)

Bases: CommunicationProtocol

Communication protocol that does nothing.

close() → None

Void close function.

static config_cls() → type[EmptyConfig]

Empty configuration

Returns:

EmptyConfig

open() → None

Void open function.

class SyncCommunicationProtocol(config)

Bases: AsyncCommunicationProtocol, ABC

Abstract base class for synchronous communication protocols with query()

static config_cls() → type[SyncCommunicationProtocolConfig]

Return the default configdataclass class.

Returns:

a reference to the default configdataclass class

query(command: str, n_attempts_max: int | None = None, attempt_interval_sec: int | float | None = None) → str | None

Send a command to the interface and handle the status message. Possibly raises an exception.

Parameters:

• command – Command to send

• n_attempts_max – Amount of attempts how often a non-empty text is tried to be read as answer

• attempt_interval_sec – time between the reading attempts

Returns:

Answer from the interface, which can be None instead of an empty reply

class SyncCommunicationProtocolConfig(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)

Bases: AsyncCommunicationProtocolConfig

Base configuration data class for synchronous communication protocols