eta_utility.util module

eta_utility.util.get_logger(name: str | None = None, level: int = 10, log_format: str = 'simple') → logging.Logger

Get eta_utility specific logger.

This function initializes and configures the eta_utility’s logger with the specified logging level and format. By default, this logger will not propagate to the root logger, ensuring that eta_utility’s logs remain isolated unless otherwise configured.

Using this function is optional. The logger can be accessed and customized manually after retrieval.

Parameters:

• level – Logging level (lower is more verbose between 10 - Debugging and 40 - Errors).

• log_format – Format of the log output. One of: simple, logname, time. (default: simple).

Returns:

The eta_utility logger.

eta_utility.util.log_add_filehandler(filename: Path | None = None, level: int = 1, log_format: str = 'time') → logging.Logger

Add a file handler to the logger to save the log output.

Parameters:

• filename – File path where logger is stored.

• level – Logging level (higher is more verbose between 0 - no output and 4 - debug).

• log_format – Format of the log output. One of: simple, logname, time. (default: time).

Returns:

The FileHandler logger.

eta_utility.util.log_add_streamhandler(level: int = 10, log_format: str = 'simple', stream: io.TextIOBase | Any = <_io.TextIOWrapper name='<stdout>' mode='w' encoding='utf-8'>) → logging.Logger

Add a stream handler to the logger to show the log output.

Parameters:

• level – Logging level (lower is more verbose between 10 - Debugging and 40 - Errors).

• format – Format of the log output. One of: simple, logname, time. (default: time).

Returns:

The eta_utility logger with an attached StreamHandler

eta_utility.util.json_import(path: Path) → list[Any] | dict[str, Any]

Extend standard JSON import to allow ‘//’ comments in JSON files.

Parameters:

path – Path to JSON file.

Returns:

Parsed dictionary.

eta_utility.util.toml_import(path: Path) → dict[str, Any]

Import a TOML file and return the parsed dictionary.

Parameters:

path – Path to TOML file.

Returns:

Parsed dictionary.

eta_utility.util.yaml_import(path: Path) → dict[str, Any]

Import a YAML file and return the parsed dictionary.

Parameters:

path – Path to YAML file.

Returns:

Parsed dictionary.

eta_utility.util.url_parse(url: str | None, scheme: str = '') → tuple[ParseResult, str | None, str | None]

Extend parsing of URL strings to find passwords and remove them from the original URL.

Parameters:

url – URL string to be parsed.

Returns:

Tuple of ParseResult object and two strings for username and password.

eta_utility.util.dict_get_any(dikt: dict[str, Any], *names: str, fail: bool = True, default: Any = None) → Any

Get any of the specified items from dictionary, if any are available. The function will return the first value it finds, even if there are multiple matches.

Parameters:

• dikt – Dictionary to get values from.

• names – Item names to look for.

• fail – Flag to determine, if the function should fail with a KeyError, if none of the items are found. If this is False, the function will return the value specified by ‘default’.

• default – Value to return, if none of the items are found and ‘fail’ is False.

Returns:

Value from dictionary.

Raise:

KeyError, if none of the requested items are available and fail is True.

eta_utility.util.dict_pop_any(dikt: dict[str, Any], *names: str, fail: bool = True, default: Any = None) → Any

Pop any of the specified items from dictionary, if any are available. The function will return the first value it finds, even if there are multiple matches. This function removes the found values from the dictionary!

Parameters:

• dikt – Dictionary to pop values from.

• names – Item names to look for.

• fail – Flag to determine, if the function should fail with a KeyError, if none of the items are found. If this is False, the function will return the value specified by ‘default’.

• default – Value to return, if none of the items are found and ‘fail’ is False.

Returns:

Value from dictionary.

Raise:

KeyError, if none of the requested items are available and fail is True.

eta_utility.util.dict_search(dikt: dict[str, str], val: str) → str

Function to get key of _psr_types dictionary, given value. Raise ValueError in case of value not specified in data.

Parameters:

• val – value to search

• data – dictionary to search for value

Returns:

key of the dictionary

eta_utility.util.deep_mapping_update(source: Any, overrides: Mapping[str, str | Mapping[str, Any]]) → dict[str, str | Mapping[str, Any]]

Perform a deep update of a nested dictionary or similar mapping.

Parameters:

• source – Original mapping to be updated.

• overrides – Mapping with new values to integrate into the new mapping.

Returns:

New Mapping with values from the source and overrides combined.

eta_utility.util.csv_export(path: Path, data: Mapping[str, Any] | Sequence[Mapping[str, Any] | Any] | pd.DataFrame, names: Sequence[str] | None = None, index: Sequence[int] | pd.DatetimeIndex | None = None, *, sep: str = ';', decimal: str = '.') → None

Export data to CSV file.

Parameters:

• path – Directory path to export data.

• data – Data to be saved.

• names – Field names used when data is a Matrix without column names.

• index – Optional sequence to set an index

• sep – Separator to use between the fields.

• decimal – Sign to use for decimal points.

eta_utility.util.replace_decimal_str(value: str | float, decimal: str = '.') → str

Replace the decimal sign in a string.

Parameters:

• value – The value to replace in.

• decimal – New decimal sign.

eta_utility.util.ensure_timezone(dt_value: datetime) → datetime

Helper function to check if datetime has timezone and if not assign local time zone.

Parameters:

dt_value – Datetime object

Returns:

datetime object with timezone information

eta_utility.util.round_timestamp(dt_value: datetime, interval: float = 1, ensure_tz: bool = True) → datetime

Helper method for rounding date time objects to specified interval in seconds. The method will also add local timezone information is None in datetime and if ensure_timezone is True.

Parameters:

• dt_value – Datetime object to be rounded

• interval – Interval in seconds to be rounded to

• ensure_tz – Boolean value to ensure or not timezone info in datetime

Returns:

Rounded datetime object

eta_utility.util.deprecated(message: str) → Callable

This is a decorator which can be used to mark functions or classes as deprecated. It will result in a warning being emitted when the function or class is used.

Parameters:

message – Message to be displayed when the function is called.

class eta_utility.util.KeyCertPair(key: PrivateKey, cert: x509.Certificate)

Bases: ABC

KeyCertPair is a wrapper for an RSA private key file and a corresponding x509 certificate. Implementations provide a contextmanager “tempfiles”, which provides access to the certificate files and the properties key and cert, which contain the RSA key and certificate information.

property key: PrivateKey

RSA private key for the certificate.

property cert: Certificate

x509 certificate information.

abstract tempfiles() → Generator

Accessor for temporary certificate files.

abstract property key_path: str

Path to the key file.

abstract property cert_path: str

Path to the certificate file.

class eta_utility.util.SelfsignedKeyCertPair(common_name: str, passphrase: str | None = None, country: str | None = None, province: str | None = None, city: str | None = None, organization: str | None = None)

Bases: KeyCertPair

Self signed key and certificate pair for use with the connectors.

Parameters:

• common_name – Common name the certificate should be valid for.

• passphrase – Pass phrase for encryption of the private key.

• country – Country code for the certificate owner, for example “DE” or “US”.

• province – Province name of the certificate owner. Empty by default.

• city – City name of the certificate owner. Empty by default.

• organization – Name of the certificate owner’s organization. “OPC UA Client” by default.

generate_cert(common_name: str, country: str | None = None, province: str | None = None, city: str | None = None, organization: str | None = None) → tuple[rsa.RSAPrivateKey, x509.Certificate]

Generate a self signed key and certificate pair for use with the connectors.

Parameters:

• common_name – Common name the certificate should be valid for.

• country – Alpha-2 country code for the certificate owner. Empty by default.

• province – Province name of the certificate owner. Empty by default.

• city – City name of the certificate owner. Empty by default.

• organization – Name of the certificate owner’s organization. “OPC UA Client” by default.

Returns:

Tuple of RSA private key and x509 certificate.

store_tempfile() → tuple[str, str]

Store the key and certificate as named temporary files. The function returns the names of the two files.

Returns:

Tuple of name of the key file and name of the certificate file.

tempfiles() → Generator

Accessor for temporary certificate files.

property key_path: str

Path to the key file.

property cert_path: str

Path to the certificate file.

class eta_utility.util.PEMKeyCertPair(key_path: Path, cert_path: Path, passphrase: str | None)

Bases: KeyCertPair

Load a PEM formatted key and certificate pair from files.

Parameters:

• key_path – Path to the PEM formatted RSA private key file.

• cert_path – Path to the PEM formatted certificate file.

• passphrase – Pass phrase for encryption of the private key.

tempfiles() → Generator

Accessor for temporary certificate files.

property key_path: str

Path to the key file.

property cert_path: str

Path to the certificate file.

class eta_utility.util.Suppressor

Bases: TextIOBase

Context manager to suppress standard output.

write(x: Any) → int

Write string to stream. Returns the number of characters written (which is always equal to the length of the string).