Powerplay! 3 Phasen 40+ Eingänge! Enecess Home Energy Monitor im Test

/

Software aus dem Video
Warnung:

Software KI generiert und nur zur Anschauung.
Keine Haftung, kein Support. Nutzung auf eigenes Risiko.

#!/usr/bin/env python3
"""
ecoMain Modbus TCP live power display.

Continuously reads:
    1000  Host Main Channel L1 Active Power
    1002  Host Main Channel L2 Active Power
    1004  Host Main Channel L3 Active Power

    1008  Host Branch Channel 1 Active Power
    1010  Host Branch Channel 2 Active Power
    1012  Host Branch Channel 3 Active Power

    1028  Slave 1 Branch Channel 1 Active Power
    1030  Slave 1 Branch Channel 2 Active Power

The screen is updated in place instead of scrolling.

Install:
    pip install pymodbus

Run:
    python ecomain_live_power.py 192.168.178.99

Optional:
    python ecomain_live_power.py 192.168.178.99 --interval 0.5
"""

from __future__ import annotations

import argparse
import os
import sys
import time
from datetime import datetime

from pymodbus.client import ModbusTcpClient


MODBUS_PORT = 502
DEVICE_ID = 255


def enable_ansi_windows() -> None:
    """Enable ANSI terminal escape sequences on Windows when possible."""
    if os.name != "nt":
        return

    try:
        import ctypes

        kernel32 = ctypes.windll.kernel32
        handle = kernel32.GetStdHandle(-11)  # STD_OUTPUT_HANDLE

        mode = ctypes.c_uint()
        if kernel32.GetConsoleMode(handle, ctypes.byref(mode)):
            ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x0004
            kernel32.SetConsoleMode(
                handle,
                mode.value | ENABLE_VIRTUAL_TERMINAL_PROCESSING,
            )
    except Exception:
        pass


def clear_screen() -> None:
    """Clear the terminal once."""
    sys.stdout.write("\033[2J\033[H")
    sys.stdout.flush()


def move_cursor_home() -> None:
    """Move cursor to top-left without clearing/scrolling."""
    sys.stdout.write("\033[H")
    sys.stdout.flush()


def hide_cursor() -> None:
    sys.stdout.write("\033[?25l")
    sys.stdout.flush()


def show_cursor() -> None:
    sys.stdout.write("\033[?25h")
    sys.stdout.flush()


def decode_int32_low_word_first(registers: list[int]) -> int:
    if len(registers) != 2:
        raise ValueError("Expected exactly 2 registers for Int32")

    low = registers[0] & 0xFFFF
    high = registers[1] & 0xFFFF

    value = (high << 16) | low

    if value & 0x80000000:
        value -= 0x100000000

    return value


def decode_power(registers: list[int]) -> float:
    """Decode ecoMain active power and apply the 0.01 W scale."""
    return decode_int32_low_word_first(registers) * 0.01


def read_power_block(
    client: ModbusTcpClient,
    start_address: int,
) -> tuple[float, float, float]:
    """
    Read three consecutive Int32 values (6 Modbus registers).

    1000..1005 -> Main L1, L2, L3
    1008..1013 -> Branch 1, 2, 3
    """
    response = client.read_holding_registers(
        start_address,
        count=6,
        device_id=DEVICE_ID,
    )

    if response.isError():
        code = getattr(response, "exception_code", None)
        raise RuntimeError(
            f"Modbus error at : "
            f"exception_code="
        )

    r = response.registers

    return (
        decode_power(r[0:2]),
        decode_power(r[2:4]),
        decode_power(r[4:6]),
    )


def read_two_power_channels(
    client: ModbusTcpClient,
    start_address: int,
) -> tuple[float, float]:
    """Read two consecutive Int32 power channels (4 Modbus registers)."""
    response = client.read_holding_registers(
        start_address,
        count=4,
        device_id=DEVICE_ID,
    )

    if response.isError():
        code = getattr(response, "exception_code", None)
        raise RuntimeError(
            f"Modbus error at : "
            f"exception_code="
        )

    r = response.registers

    return (
        decode_power(r[0:2]),
        decode_power(r[2:4]),
    )


def connect_client(host: str, timeout: float) -> ModbusTcpClient:
    client = ModbusTcpClient(
        host=host,
        port=MODBUS_PORT,
        timeout=timeout,
    )

    if not client.connect():
        client.close()
        raise ConnectionError(
            f"Could not connect to "
        )

    return client


def format_power(value: float | None) -> str:
    if value is None:
        return "---"
    return f" W"


def render_screen(
    host: str,
    interval: float,
    connected: bool,
    values: dict[str, float | None],
    last_update: datetime | None,
    error: str = "",
) -> None:
    """
    Draw the complete dashboard from the top-left.

    Each line is padded to erase leftover characters from the previous frame.
    """
    terminal_width = 78

    def line(text: str = "") -> str:
        return text[:terminal_width].ljust(terminal_width)

    status = "CONNECTED" if connected else "DISCONNECTED"
    updated = (
        last_update.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]
        if last_update
        else "---"
    )

    rows = [
        line("ecoMain Modbus TCP - Live Active Power"),
        line("=" * terminal_width),
        line(f"Device:       "),
        line(f"Device ID:    "),
        line(f"Status:       "),
        line(f"Last update:  "),
        line(f"Interval:      s"),
        line(),
        line("MAIN CHANNEL"),
        line("-" * terminal_width),
        line(f"  L1  [1000] : "),
        line(f"  L2  [1002] : "),
        line(f"  L3  [1004] : "),
        line(),
        line("HOST BRANCH CHANNELS"),
        line("-" * terminal_width),
        line(f"  CH1 [1008] : "),
        line(f"  CH2 [1010] : "),
        line(f"  CH3 [1012] : "),
        line(),
        line("EXTERNAL SUBDEVICE - SLAVE 1"),
        line("-" * terminal_width),
        line(f"  CH1 [1028] : "),
        line(f"  CH2 [1030] : "),
        line(),
        line(f"Error: " if error else "Error: none"),
        line(),
        line("Press Ctrl+C to stop."),
    ]

    move_cursor_home()
    sys.stdout.write("\n".join(rows))
    sys.stdout.flush()


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Live in-place ecoMain active-power display."
    )

    parser.add_argument(
        "host",
        help="ecoMain IP address or hostname",
    )

    parser.add_argument(
        "--interval",
        type=float,
        default=1.0,
        help="Polling interval in seconds (default: 1.0)",
    )

    parser.add_argument(
        "--timeout",
        type=float,
        default=3.0,
        help="Modbus timeout in seconds (default: 3.0)",
    )

    args = parser.parse_args()

    if args.interval <= 0:
        parser.error("--interval must be greater than 0")

    enable_ansi_windows()
    clear_screen()
    hide_cursor()

    values = {
        "main_l1": None,
        "main_l2": None,
        "main_l3": None,
        "branch_1": None,
        "branch_2": None,
        "branch_3": None,
        "slave1_ch1": None,
        "slave1_ch2": None,
    }

    client: ModbusTcpClient | None = None
    connected = False
    last_update: datetime | None = None
    error_message = ""

    try:
        while True:
            cycle_start = time.monotonic()

            try:
                if client is None or not connected:
                    if client is not None:
                        try:
                            client.close()
                        except Exception:
                            pass

                    client = connect_client(args.host, args.timeout)
                    connected = True
                    error_message = ""

                main_l1, main_l2, main_l3 = read_power_block(
                    client,
                    1000,
                )

                branch_1, branch_2, branch_3 = read_power_block(
                    client,
                    1008,
                )

                # External subdevice (Slave 1):
                # 1028 = Slave 1 Branch Channel 1 Active Power
                # 1030 = Slave 1 Branch Channel 2 Active Power
                slave1_ch1, slave1_ch2 = read_two_power_channels(
                    client,
                    1028,
                )

                values.update(
                    {
                        "main_l1": main_l1,
                        "main_l2": main_l2,
                        "main_l3": main_l3,
                        "branch_1": branch_1,
                        "branch_2": branch_2,
                        "branch_3": branch_3,
                        "slave1_ch1": slave1_ch1,
                        "slave1_ch2": slave1_ch2,
                    }
                )

                connected = True
                last_update = datetime.now()
                error_message = ""

            except Exception as exc:
                connected = False
                error_message = str(exc)

                if client is not None:
                    try:
                        client.close()
                    except Exception:
                        pass

                client = None

            render_screen(
                host=args.host,
                interval=args.interval,
                connected=connected,
                values=values,
                last_update=last_update,
                error=error_message,
            )

            elapsed = time.monotonic() - cycle_start
            time.sleep(max(0.0, args.interval - elapsed))

    except KeyboardInterrupt:
        return 0

    finally:
        if client is not None:
            try:
                client.close()
            except Exception:
                pass

        show_cursor()
        sys.stdout.write("\n")
        sys.stdout.flush()


if __name__ == "__main__":
    raise SystemExit(main())