eta_utility.connectors.node module

This module implements the node class, which is used to parametrize connections

class eta_utility.connectors.node.NodeMeta(name: str, bases: tuple, namespace: dict[str, Any], **kwargs: Any)

Bases: type

Metaclass to define all Node classes as frozen attr dataclasses.

class eta_utility.connectors.node.Node(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)

Bases: object

The node objects represents a single variable. Valid keyword arguments depend on the protocol.

name: str

Name for the node.

url: str

URL of the connection.

url_parsed: ParseResult

Parse result object of the URL (in case more post-processing is required).

protocol: str

Protocol of the connection.

usr: str | None

Username for login to the connection (default: None).

pwd: str | None

Password for login to the connection (default: None).

interval: str | None

Interval

dtype: Callable | None

Data type of the node (for value conversion). Note that strings will be interpreted as utf-8 encoded. If you do not want this behaviour, use ‘bytes’.

evolve(**kwargs: Any) → Node

Returns a new node instance by copying the current node and changing only specified keyword arguments.

This allows for seamless node instantiation with only a few changes.

Parameters:

kwargs – Keyword arguments to change.

Returns:

New instance of the node.

as_dict(filter_none: bool = False, **kwargs: Any) → dict[str, Any]

Return the attrs attribute values of node instance as a dict.

Parameters:

filter_none – Filter none values, defaults to False

Returns:

dict of attribute values

as_tuple(filter_none: bool = False, **kwargs: Any) → tuple[Any, ...]

Return the attrs attribute values of inst as a tuple.

Parameters:

filter_none – Filter none values, defaults to False

Returns:

tuple of attribute values

classmethod from_dict(dikt: Sequence[Mapping] | Mapping[str, Any], fail: bool = True) → list[Self]

Create nodes from a dictionary of node configurations. The configuration must specify the following fields for each node:

• Code (or name), URL, Protocol (i.e. modbus or opcua or eneffco). The URL should be a complete network location identifier. Alternatively it is possible to specify the location in two fields: IP and Port. These should only contain the respective parts (as in only an IP address and only the port number). The IP-Address should always be given without scheme (https://).

For local nodes no additional fields are required.

For Modbus nodes the following additional fields are required:

• ModbusRegisterType (or mb_register), ModbusSlave (or mb_slave), ModbusChannel (or mb_channel).

For OPC UA nodes the following additional fields are required:

• Identifier.

For EnEffCo nodes the code field must be present.

For EntsoE nodes the endpoint field must be present.

Parameters:

• dikt – Configuration dictionary.

• fail – Set this to false, if you would like to log errors instead of raising them.

Returns:

List of Node objects.

classmethod from_excel(path: Path, sheet_name: str, fail: bool = True) → list[Self]

Method to read out nodes from an Excel document. The document must specify the following fields:

• Code, IP, Port, Protocol (modbus or opcua or eneffco).

For Modbus nodes the following additional fields are required:

• ModbusRegisterType, ModbusByte, ModbusChannel.

For OPC UA nodes the following additional fields are required:

• Identifier.

For EnEffCo nodes the Code field must be present.

The IP-Address should always be given without scheme (https://).

Parameters:

• path – Path to Excel document.

• sheet_name – name of Excel sheet, which will be read out.

• fail – Set this to false, if you would like to log errors instead of raising them.

Returns:

List of Node objects.

classmethod get_eneffco_nodes_from_codes(code_list: Sequence[str], eneffco_url: str) → list[Self]

Utility function to retrieve Node objects from a list of EnEffCo Codes (Identifiers).

Deprecated since version v2.0.0: Use the from_ids function of the EnEffCoConnection Class instead.

Parameters:

• code_list – List of EnEffCo identifiers to create nodes from.

• eneffco_url – URL to the EnEffCo system.

Returns:

List of EnEffCo nodes.

class eta_utility.connectors.node.NodeLocal(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)

Bases: Node

Local Node (no specific protocol), useful for example to manually provide data to subscription handlers.

class eta_utility.connectors.node.NodeModbus(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)

Bases: Node

Node for the Modbus protocol.

mb_slave: int | None

Modbus Slave ID

mb_register: str

Modbus Register name. One of input, discrete_input, coils and holding. Note that only coils and holding can be written to.

mb_channel: int

Modbus Channel (Address of the value)

mb_bit_length: int

Length of the value in bits (default 32). This determines, how much data is read from the server. The value must be a multiple of 16.

mb_byteorder: str

Byteorder of values returned by modbus

mb_wordorder: str

Wordorder of values returned by modbus

class eta_utility.connectors.node.NodeOpcUa(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)

Bases: Node

Node for the OPC UA protocol.

opc_id: str | None

Node ID of the OPC UA Node.

opc_path_str: str | None

Path to the OPC UA node.

opc_ns: int | None

Namespace of the OPC UA Node.

opc_id_type: str

Type of the OPC UA Node ID Specification.

opc_name: str

Name of the OPC UA Node.

opc_path: list[NodeOpcUa]

Path to the OPC UA node in list representation. Nodes in this list can be used to access any parent objects.

evolve(**kwargs: Any) → Node

Returns a new node instance by copying the current node and changing only specified keyword arguments.

This allows for seamless node instantiation with only a few changes.

Adjusted attributes handling according to OpcUa node instantiation logic as in ‘__attrs_post_init__’.

Parameters:

kwargs – Keyword arguments to change.

Returns:

New instance of the node.

class eta_utility.connectors.node.NodeEnEffCo(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)

Bases: Node

Node for the EnEffCo API.

eneffco_code: str

EnEffCo datapoint code / ID.

class eta_utility.connectors.node.NodeEntsoE(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)

Bases: Node

Node for the EntsoE API (see ENTSO-E Transparency Platform API).

Available endpoint

Endpoint	Description
ActualGenerationPerType	Actual Generation Per Energy Type
Price	Price day ahead

Currently, there is only two endpoints available, due to the parameter managing required by the API documentation. The other possible endpoints are listed in

eta_utility.connectors.entso_e._ConnectionConfiguration._doc_types

Main bidding zone

Bidding Zone	Description
DEU-LUX	Deutschland-Luxemburg

The other possible bidding zones are listed in

eta_utility.connectors.entso_e._ConnectionConfiguration._bidding_zones

endpoint: str

REST endpoint.

bidding_zone: str

Bidding zone.

class eta_utility.connectors.node.NodeCumulocity(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)

Bases: Node

Node for the Cumulocity API.

device_id: str

measurement: str

fragment: str

class eta_utility.connectors.node.NodeWetterdienst(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)

Bases: Node

Basic Node for the Wetterdienst API. This class is not meant to be used directly, but to be subclassed by NodeWetterdienstObservation and NodeWetterdienstPrediction.

parameter: str

Parameter to read from wetterdienst (e.g HUMIDITY or TEMPERATURE_AIR_200)

station_id: str | None

The id of the weather station

latlon: str | None

latitude and longitude (not necessarily a weather station)

number_of_stations: int | None

Number of stations to be used for the query

class eta_utility.connectors.node.NodeWetterdienstObservation(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)

Bases: NodeWetterdienst

Node for the Wetterdienst API to get weather observations. For more information see: https://wetterdienst.readthedocs.io/en/latest/data/coverage/dwd/observation.html

interval: str

Redeclare interval attribute, but don’t allow it to be optional

static convert_interval_to_resolution(interval: int | str | timedelta) → str

class eta_utility.connectors.node.NodeWetterdienstPrediction(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)

Bases: NodeWetterdienst

Node for the Wetterdienst API to get weather predictions. For more information see: https://wetterdienst.readthedocs.io/en/latest/data/coverage/dwd/mosmix.html

mosmix_type: str

Type of the MOSMIX prediction. Either ‘SMALL’ or ‘LARGE’

class eta_utility.connectors.node.EmonioConstants

Bases: object

Dict constants for the Emonio API.

PARAMETER_MAP: Final[dict[int, list[str]]] = {0: ['VRMS', 'V_RMS', 'Voltage', 'V', 'Spannung'], 2: ['IRMS', 'I_RMS', 'Current', 'I', 'Strom'], 4: ['WATT', 'Power', 'W', 'Leistung', 'Wirkleistung'], 6: ['VAR', 'Reactive Power', 'VAR', 'Blindleistung'], 8: ['VA', 'Apparent Power', 'VA', 'Scheinleistung'], 10: ['FREQ', 'Frequency', 'Hz', 'Frequenz'], 12: ['KWH', 'Energy', 'kWh', 'Energie'], 14: ['PF', 'Power Factor', 'PF', 'Leistungsfaktor'], 20: ['VRMS MIN', 'VRMS_MIN', 'Voltage Min', 'V Min', 'Spannung Min'], 22: ['VRMS MAX', 'VRMS_MAX', 'Voltage Max', 'V Max', 'Spannung Max'], 24: ['IRMS MIN', 'IRMS_MIN', 'Current Min', 'I Min', 'Strom Min'], 26: ['IRMS MAX', 'IRMS_MAX', 'Current Max', 'I Max', 'Strom Max'], 28: ['WATT MIN', 'WATT_MIN', 'Power Min', 'W Min', 'Leistung Min'], 30: ['WATT MAX', 'WATT_MAX', 'Power Max', 'W Max', 'Leistung Max'], 500: ['Temp', 'degree', 'Temperature', '°C', 'Temperatur'], 800: ['Impulse', 'Impuls']}

Mapping of parameters to addresses

UPPER_CASED: Final[dict[int, list[str]]] = {0: ['VRMS', 'V_RMS', 'VOLTAGE', 'V', 'SPANNUNG'], 2: ['IRMS', 'I_RMS', 'CURRENT', 'I', 'STROM'], 4: ['WATT', 'POWER', 'W', 'LEISTUNG', 'WIRKLEISTUNG'], 6: ['VAR', 'REACTIVE POWER', 'VAR', 'BLINDLEISTUNG'], 8: ['VA', 'APPARENT POWER', 'VA', 'SCHEINLEISTUNG'], 10: ['FREQ', 'FREQUENCY', 'HZ', 'FREQUENZ'], 12: ['KWH', 'ENERGY', 'KWH', 'ENERGIE'], 14: ['PF', 'POWER FACTOR', 'PF', 'LEISTUNGSFAKTOR'], 20: ['VRMS MIN', 'VRMS_MIN', 'VOLTAGE MIN', 'V MIN', 'SPANNUNG MIN'], 22: ['VRMS MAX', 'VRMS_MAX', 'VOLTAGE MAX', 'V MAX', 'SPANNUNG MAX'], 24: ['IRMS MIN', 'IRMS_MIN', 'CURRENT MIN', 'I MIN', 'STROM MIN'], 26: ['IRMS MAX', 'IRMS_MAX', 'CURRENT MAX', 'I MAX', 'STROM MAX'], 28: ['WATT MIN', 'WATT_MIN', 'POWER MIN', 'W MIN', 'LEISTUNG MIN'], 30: ['WATT MAX', 'WATT_MAX', 'POWER MAX', 'W MAX', 'LEISTUNG MAX'], 500: ['TEMP', 'DEGREE', 'TEMPERATURE', '°C', 'TEMPERATUR'], 800: ['IMPULSE', 'IMPULS']}

Create dictionary with all upper cased parameters

PHASE_MAP: Final[dict[str, int]] = {'a': 0, 'abc': 300, 'b': 100, 'c': 200}

Mapping of phases to address offsets

class eta_utility.connectors.node.NodeEmonio(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)

Bases: Node

Node for the emonio. The parameter to read is specified by the name of the node. Available parameters are defined in the parameter_map class attribute. Additionally, the phase of the parameter can be specified, with ‘a’, ‘b’, ‘c’ or ‘abc’.

https://wiki.emonio.de/de/Emonio_P3

address: int

Modbus address of the parameter to read

phase: str

Phase of the parameter (a, b, c). If not set, all phases are read

class eta_utility.connectors.node.NodeForecastSolar(name: str, url: str, protocol: str, *args: Any, **kwargs: Any)

Bases: Node

Node for using the Forecast.Solar API.

Mandatory parameters are:

• The location of the forecast solar plane(s): latitude, longitude,

• Plane parameters: declination, azimuth and kwp.

Additionally api_key must be set for endpoints other than ‘estimate’, multiple planes or if requests capacity is exceeded.

For multiple planes, the parameters shall be passed as lists of the same length (e.g. [0, 30], [180, 180], [5, 5]).

api_key: str | None

API key for the Forecast.Solar API; string

endpoint: str

Endpoint in (estimate, history, clearsky), defaults to estimate; string

data: str | None

What data to query, i.e. only watts, watt hours, watt hours per period or watt hours per day; string

latitude: int

Latitude of plane location, -90 (south) … 90 (north); handled with a precision of 0.0001 or abt. 10 m

longitude: int

Longitude of plane location, -180 (west) … 180 (east); handled with a precision of 0.0001 or abt. 10 m

declination: int | list[int]

Plane declination, 0 (horizontal) … 90 (vertical) - always in relation to earth’s surface; integer

azimuth: int | list[int]

Plane azimuth, -180 … 180 (-180 = north, -90 = east, 0 = south, 90 = west, 180 = north); integer

kwp: float | list[float]

Installed modules power of plane in kilo watt; float

no_sun: int | None

Format of timestamps in the response, see API doc for values; string Forecast for full day or only sunrise to sunset, 0|1 (API defaults to 0); int

damping_morning: float | None

Damping factor for the morning (API defaults to 0.0)

damping_evening: float | None

Damping factor for the evening (API defaults to 0.0)

horizon: int | list[int] | None

Horizon information; string, (comma-separated list of numerics) See API doc

inverter: float | None

Maximum of inverter in kilowatts or kVA; float > 0

actual: float | None

Actual production until now; float >= 0