EmonioConnection

Eta-utility includes a connection class designed to read data from an Emonio device. To ensure proper communication, the Emonio must be connected to the same network as the machine running the Eta-utility. Additionally, because the EmonioConnection uses the Modbus protocol, it is essential that the Modbus Server on the Emonio is set to Enabled.

This configuration can be done in the Emonio App by navigating to Settings -> Modbus Server. The IP address and port of the Emonio can also be easily found within the app.

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

class eta_utility.connectors.EmonioConnection(url: str, *, nodes: Nodes[NodeEmonio] | 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 attribute of the node. Possible values are a, b, c or abc, with abc being the default.

read(nodes: Nodes[NodeEmonio] | 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[NodeEmonio, Any]) → None[source]: Warning

Not implemented: Writing to Emonio nodes is not supported.

subscribe(handler: SubscriptionHandler, nodes: Nodes[NodeEmonio] | 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)

Examples

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
return 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
return connection.read()

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