EmonioConnection

Eta-utility provides a connection class to read data from an Emonio device. The emonio has to be in the same network as the computer running eta-utility. Since the EmonioConnection is implmented using the modbus protocol, the Emonio needs to have its Modbus Server configured to Enabled.

This can be done in the Emonio App under Settings -> Modbus Server. The ip address and port of the emonio can also easily be found in the app.

See the Emonio documentation for more information: https://wiki.emonio.de/de/Emonio_P3

EmonioConnection

class eta_utility.connectors.EmonioConnection(url: str, *, nodes: Nodes | None = None, check_error: bool = True)[source]

Thin wrapper class for the emonio that uses a modbus TCP Connection. Internally the emonio nodes are converted to modbus nodes with fixed parameters, expect for the name, url and channel. If nodes have specified a phase, the connection will check if the phase is connected. Additionally, the connection will check for emonio errors and warnings (max. every minute).

When creating a NodeEmonio the parameter (and resulting modbus channel) is set by the name of the node (case insensitive). See Available Emonio Nodes for for possible parameter names. Alternatively, the address (modbus channel) can be set manually. The phase is set by the phase parameter of the node. Possible values are “a”, “b”, “c” or “abc”, with “abc” being the default.

See the emonio documentation for more information: https://wiki.emonio.de/de/Emonio_P3

read(nodes: Nodes | None = None) → pd.DataFrame[source]

Read the values of the selected nodes. If nodes is None, all previously selected nodes will be read.

Parameters:: nodes – List of nodes to read from.
Returns:: Dataframe with the read values.

write(values: Mapping[Node, Any]) → None[source]: Warning

Not implemented: Writing to emonio nodes is not supported.

subscribe(handler: SubscriptionHandler, nodes: Nodes | None = None, interval: TimeStep = 1) → None[source]

Subscribe to the selected nodes. The Modbus connection does all the handling.

Parameters:

handler – The handler to subscribe.
nodes – List of nodes to subscribe to.
interval – The interval in seconds to read the values.

close_sub() → None[source]: Close the subscription of the modus connection.

check_warnings_and_errors() → None[source]: Calls the error and warning check if the last check was more than a minute ago.

Available Emonio Nodes
Allowed Names	Address
‘VRMS’, ‘V_RMS’, ‘Voltage’, ‘V’, ‘Spannung’	0
‘IRMS’, ‘I_RMS’, ‘Current’, ‘I’, ‘Strom’	2
‘WATT’, ‘Power’, ‘W’, ‘Leistung’, ‘Wirkleistung’	4
‘VAR’, ‘Reactive Power’, ‘VAR’, ‘Blindleistung’	6
‘VA’, ‘Apparent Power’, ‘VA’, ‘Scheinleistung’	8
‘FREQ’, ‘Frequency’, ‘Hz’, ‘Frequenz’	10
‘KWH’, ‘Energy’, ‘kWh’, ‘Energie’	12
‘PF’, ‘Power Factor’, ‘PF’, ‘Leistungsfaktor’	14
‘VRMS MIN’, ‘VRMS_MIN’, ‘Voltage Min’, ‘V Min’, ‘Spannung Min’	20
‘VRMS MAX’, ‘VRMS_MAX’, ‘Voltage Max’, ‘V Max’, ‘Spannung Max’	22
‘IRMS MIN’, ‘IRMS_MIN’, ‘Current Min’, ‘I Min’, ‘Strom Min’	24
‘IRMS MAX’, ‘IRMS_MAX’, ‘Current Max’, ‘I Max’, ‘Strom Max’	26
‘WATT MIN’, ‘WATT_MIN’, ‘Power Min’, ‘W Min’, ‘Leistung Min’	28
‘WATT MAX’, ‘WATT_MAX’, ‘Power Max’, ‘W Max’, ‘Leistung Max’	30
‘Temp’, ‘degree’, ‘Temperature’, ‘°C’, ‘Temperatur’	500
‘Impulse’, ‘Impuls’	800

Note

For reading MIN and MAX values, the phase must be specified. “abc” is not allowed.

class eta_utility.connectors.emonio.NodeModbusFactory(url: str)[source]

The NodeModbusFactory is a factory class that creates NodeModbus objects with fixed parameters, expect: name, url and mb_channel.

Has to be initialized with the url of the emonio.

It’s a helper class for the EmonioConnection to create its modbus nodes. It also can be used to manually create a NodeModbus object, which has to be read with a ModbusConnection. (not recommended, use the EmonioConnection instead)

get_default_node(name: str, channel: int) → NodeModbus[source]

Create a modbus node for reading emonio values.

Parameters:

name – Name of the node.
channel – Modbus channel of the node. (Emonio address)

get_discrete_input_node(name: str, channel: int) → NodeModbus[source]

Create a modbus node for reading the connection status of the emonio phases.

Parameters:

name – Name of the node.
channel – Modbus channel of the node. (Emonio address)

get_warnings_errors_node(name: str, channel: int) → NodeModbus[source]

Create a modbus node for reading the error and warning registers of the emonio.

Parameters:

name – Name of the node.
channel – Modbus channel of the node. (Emonio address)

NodeEmonio

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

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

parameter_map = {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']}

phase_map = {'a': 0, 'abc': 300, 'b': 100, 'c': 200}

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).

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’.

Examples Usage

This example demonstrates how to create a LiveConnect from a dictionary to read data from an Emonio device. Alternatively, the LiveConnection can be created from a JSON file, with the JSON having the same structure as the dictionary.

url has to be replaced with the IP address and port of the Emonio device, e.g. "192.168.178.123:502".

from eta_utility.connectors.live_connect import LiveConnect

live = {
    "system": [
        {
            "name": "emonio",
            "servers": {"ac_supply": {"url": url, "protocol": "emonio"}},
            "nodes": [
                {"name": "V_RMS", "server": "ac_supply"},
                {"name": "I_RMS", "server": "ac_supply", "phase": "a"},
            ],
        }
    ]
}
# Create the connection object with classmethod from_dict
connection = LiveConnect.from_dict(None, None, 1, 10, **live)

# Read the values of the nodes we defined in the dictionary
result = connection.read("V_RMS", "I_RMS")

Here we create the Emonio nodes manually and read them with the EmonioConnection class.

from eta_utility.connectors import EmonioConnection

voltage_node = NodeEmonio("V_RMS", url, "emonio")
current_node = NodeEmonio("I_RMS", url, "emonio", phase="a")

# Initialize the connection object with both nodes
connection = EmonioConnection.from_node([voltage_node, current_node])

# Read values of selected nodes
if isinstance(connection, EmonioConnection):
    result = connection.read()
else:
    raise TypeError("The connection must be an ModbusConnection.")

The NodeModbusFactory can be used to create modbus nodes directly. But this is not recommended, as the EmonioConnection class is more convenient and has extra error checking.

from eta_utility.connectors import ModbusConnection
from eta_utility.connectors.emonio import NodeModbusFactory

factory = NodeModbusFactory(url)

# V_RMS for all phases
voltage_node = factory.get_default_node("Spannung", 300)
# I_RMS for phase a
current_node = factory.get_default_node("Strom", 2)

connection = ModbusConnection.from_node([voltage_node, current_node])

if isinstance(connection, ModbusConnection):
    result = connection.read()
else:
    raise TypeError("The connection must be an ModbusConnection.")