eta_utility.connectors.util module

class eta_utility.connectors.util.RetryWaiter[source]

Bases: object

Helper class which keeps track of waiting time before retrying a connection.

VALUES: Final[list[int]] = [0, 1, 3, 5, 5, 10, 20, 30, 40, 60]
tried() → None[source]

Register a retry with the RetryWaiter.

success() → None[source]

Register a successful connection with the RetryWaiter.

property wait_time: int

Return the time to wait for.

wait() → None[source]

Wait/sleep synchronously.

async wait_async() → None[source]

Wait/sleep asynchronously - must be awaited.

eta_utility.connectors.util.decode_modbus_value(value: Sequence[int], byteorder: str, type_: Callable | None = None, wordorder: str = 'big') → Any[source]

Method to decode incoming modbus values. Strings are always decoded as utf-8 values. If you do not want this behaviour specify ‘bytes’ as the data type.

Parameters:

  • value – Current value to be decoded into float.

  • byteorder – Byteorder for decoding i.e. ‘little’ or ‘big’ endian.

  • type_ – Type of the output value. See Python struct format character documentation <https://docs.python.org/3/library/struct.html#format-characters> for all possible format strings (default: f).

Returns:

Decoded value as a python type.

eta_utility.connectors.util.encode_bits(value: str | float | bytes, byteorder: str, bit_length: int, type_: Callable | None = None) → list[int][source]

Method to encode python data type to modbus value. This means an array of bytes to send to a modbus server.

Parameters:

  • value – Current value to be decoded into float.

  • byteorder – Byteorder for decoding i.e. ‘little’ or ‘big’ endian.

  • bit_length – Length of the value in bits.

  • type_ – Type of the output value. See Python struct format character documentation <https://docs.python.org/3/library/struct.html#format-characters> for all possible format strings (default: f).

Returns:

Decoded value as a python type.

eta_utility.connectors.util.bitarray_to_registers(bits: list[int | bool]) → list[int][source]

Convert a list of bits into a list of 16 bit ‘bytes’.

eta_utility.connectors.util.all_equal(iterable: Iterable[Any]) → bool[source]

Check if all values inside iterable are equal

Parameters:

iterable – python iterable

Returns:

True if all values are equal False elsewhere

class eta_utility.connectors.util.IntervalChecker[source]

Bases: object

Class for the subscription interval checking.

node_latest_values: dict[Node, list]

Dictionary that stores the value and the time for checking changes and the time interval

push(node: Node, value: Any | pd.Series | Sequence[Any], timestamp: datetime | pd.DatetimeIndex | TimeStep | None = None) → None[source]

Push value and time in dictionary for a node. If the value doesn’t change compared to the previous timestamp, the push is skipped.

Parameters:

  • node – Node to check.

  • value – Value from the subscription.

  • timestamp – Time of the incoming value of the node.

check_interval_connection() → bool | None[source]

Check the interval between old and new value. If no interval has been defined, the check interval is skipped.

Returns:

Boolean for the interval check.