SolarEdge Power Limit

#!/usr/bin/env python3
"""
SolarEdge Modbus TCP helper for reading power-control settings and curtailing production.

Default target: 192.168.XXX.XXX:1502, unit/slave id 1.

Typical usage:
  python3 solaredge_control.py status
  python3 solaredge_control.py standby --yes       # set Active Power Limit to 0%
  python3 solaredge_control.py production --yes    # set Active Power Limit to 100%
  python3 solaredge_control.py limit 25 --yes
  python3 solaredge_control.py power             # show L_AC_VA

Status modes:
  status          reads reliable control registers plus live apparent AC power
  status --full   probes optional/export registers too; this may cause timeout messages on some firmware
  probe           tests a few candidate register/count combinations for troubleshooting

Notes:
  - 0xF001 Active Power Limit is a dynamic command. It is not the same as SetApp Maintenance > Standby Mode.
  - Setting 0xF001 to 0% is standby-like PV curtailment; setting it to 100% resumes normal production limit.
  - Dynamic limits may be lost after inverter restart, so check status after a reboot.
"""

from __future__ import annotations

import argparse
import logging
import math
import struct
import sys
import time
from dataclasses import dataclass
from typing import Callable, Optional, Sequence

# Keep pymodbus from printing scary warnings for optional registers that simply do not exist/respond.
logging.basicConfig(level=logging.WARNING)
for logger_name in ("pymodbus", "pymodbus.logging", "pymodbus.transaction", "pymodbus.client"):
    logging.getLogger(logger_name).setLevel(logging.CRITICAL)

try:
    from pymodbus.client import ModbusTcpClient
except ImportError as exc:  # pragma: no cover
    print("pymodbus is not installed. Install it with: pip install pymodbus", file=sys.stderr)
    raise SystemExit(2) from exc


DEFAULT_HOST = "192.168.XXX.XXX"
DEFAULT_PORT = 1502
DEFAULT_UNIT = 1

REG_APPARENT_POWER_VA = 40087  # L_AC_VA, apparent AC power, signed int16, unit VA

# SolarEdge Power Control registers. Use the addresses directly, e.g. 0xF001 == 61441.
REG_RRCR_STATE = 0xF000
REG_ACTIVE_POWER_LIMIT = 0xF001
REG_COSPHI_DYNAMIC = 0xF002
REG_COMMIT_POWER_CONTROL = 0xF100

# Firmware/examples differ around these two settings, so v3 treats them as optional/fallback probes.
REG_REACTIVE_POWER_CONFIG_A = 0xF103
REG_REACTIVE_POWER_CONFIG_B = 0xF104
REG_ADVANCED_POWER_CONTROL = 0xF142

REG_ENH_DYNAMIC_CONTROL = 0xF300
REG_MAX_ACTIVE_POWER = 0xF304
REG_DYNAMIC_ACTIVE_POWER_LIMIT = 0xF322

REG_EXPORT_MODE_E = 0xE000
REG_EXPORT_LIMIT_MODE_E = 0xE001
REG_EXPORT_SITE_LIMIT_E = 0xE002
REG_EXPORT_MODE_F = 0xF700
REG_EXPORT_LIMIT_MODE_F = 0xF701
REG_EXPORT_SITE_LIMIT_F = 0xF702


@dataclass(frozen=True)
class RegisterValue:
    name: str
    address: int
    value: object
    unit: str = ""
    note: str = ""


class SolarEdgeModbus:
    def __init__(self, host: str, port: int, unit: int, timeout: float = 3.0, retries: int = 0) -> None:
        self.host = host
        self.port = port
        self.unit = unit
        # Different pymodbus versions accept slightly different constructor args.
        try:
            self.client = ModbusTcpClient(host=host, port=port, timeout=timeout, retries=retries)
        except TypeError:
            self.client = ModbusTcpClient(host=host, port=port, timeout=timeout)

    def __enter__(self) -> "SolarEdgeModbus":
        if not self.client.connect():
            raise RuntimeError(f"Could not connect to {self.host}:{self.port}")
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        self.client.close()

    def _read_holding(self, address: int, count: int):
        try:
            return self.client.read_holding_registers(address=address, count=count, slave=self.unit)
        except TypeError:
            return self.client.read_holding_registers(address=address, count=count, unit=self.unit)

    def _write_register(self, address: int, value: int):
        try:
            return self.client.write_register(address=address, value=value, slave=self.unit)
        except TypeError:
            return self.client.write_register(address=address, value=value, unit=self.unit)

    def _write_registers(self, address: int, values: Sequence[int]):
        try:
            return self.client.write_registers(address=address, values=list(values), slave=self.unit)
        except TypeError:
            return self.client.write_registers(address=address, values=list(values), unit=self.unit)

    @staticmethod
    def _ensure_ok(result, action: str) -> None:
        if result is None:
            raise RuntimeError(f"{action}: no response")
        if hasattr(result, "isError") and result.isError():
            raise RuntimeError(f"{action}: Modbus error: {result}")

    def read_regs(self, address: int, count: int) -> list[int]:
        result = self._read_holding(address, count)
        self._ensure_ok(result, f"read 0x{address:04X}/{count}")
        return list(result.registers)

    def write_u16(self, address: int, value: int) -> None:
        result = self._write_register(address, value & 0xFFFF)
        self._ensure_ok(result, f"write 0x{address:04X}")

    def write_u32(self, address: int, value: int) -> None:
        data = struct.pack(">I", value & 0xFFFFFFFF)
        regs = list(struct.unpack(">HH", data))
        result = self._write_registers(address=address, values=regs)
        self._ensure_ok(result, f"write 0x{address:04X}")

    def write_float32(self, address: int, value: float) -> None:
        data = struct.pack(">f", float(value))
        regs = list(struct.unpack(">HH", data))
        result = self._write_registers(address=address, values=regs)
        self._ensure_ok(result, f"write 0x{address:04X}")

    @staticmethod
    def decode_u16(regs: list[int]) -> int:
        return regs[0]

    @staticmethod
    def decode_i16(regs: list[int]) -> int:
        value = regs[0]
        return value - 0x10000 if value & 0x8000 else value

    @staticmethod
    def decode_u32(regs: list[int]) -> int:
        return struct.unpack(">I", struct.pack(">HH", regs[0], regs[1]))[0]

    @staticmethod
    def decode_i32(regs: list[int]) -> int:
        return struct.unpack(">i", struct.pack(">HH", regs[0], regs[1]))[0]

    @staticmethod
    def decode_float32(regs: list[int]) -> float:
        return struct.unpack(">f", struct.pack(">HH", regs[0], regs[1]))[0]

    @staticmethod
    def decode_first_u16(regs: list[int]) -> int:
        # Useful for inverters that only answer F142 when two registers are requested.
        return regs[0]

    def try_read(
        self,
        name: str,
        address: int,
        count: int,
        decoder: Callable[[list[int]], object],
        unit: str = "",
        note: str = "",
        show_error: bool = False,
    ) -> RegisterValue:
        try:
            regs = self.read_regs(address, count)
            return RegisterValue(name=name, address=address, value=decoder(regs), unit=unit, note=note)
        except Exception as exc:
            msg = f"unavailable ({exc})" if show_error else "unavailable"
            return RegisterValue(name=name, address=address, value=msg, note=note)

    def try_read_candidates(
        self,
        name: str,
        candidates: Sequence[tuple[int, int, Callable[[list[int]], object], str]],
        show_error: bool = False,
    ) -> RegisterValue:
        errors: list[str] = []
        for address, count, decoder, note in candidates:
            try:
                regs = self.read_regs(address, count)
                return RegisterValue(name=name, address=address, value=decoder(regs), note=note)
            except Exception as exc:
                errors.append(f"0x{address:04X}/{count}: {exc}")
        msg = "unavailable"
        if show_error and errors:
            msg = "unavailable (" + "; ".join(errors) + ")"
        return RegisterValue(name=name, address=candidates[0][0], value=msg, note="tried fallback addresses")

    def read_status(self, full: bool = False, show_errors: bool = False) -> list[RegisterValue]:
        # Start with registers that your inverter has already answered reliably.
        values: list[RegisterValue] = [
            self.try_read("Apparent Power (L_AC_VA)", REG_APPARENT_POWER_VA, 1, self.decode_i16, "VA", "", show_error=show_errors),
            self.try_read("RRCR state", REG_RRCR_STATE, 1, self.decode_u16, show_error=show_errors),
            self.try_read("Active Power Limit", REG_ACTIVE_POWER_LIMIT, 1, self.decode_u16, "%", show_error=show_errors),
            self.try_read("Commit Power Control status", REG_COMMIT_POWER_CONTROL, 1, self.decode_i16, show_error=show_errors),
            self.try_read("Enhanced Dynamic Power Control", REG_ENH_DYNAMIC_CONTROL, 1, self.decode_u16, show_error=show_errors),
            self.try_read("Dynamic Active Power Limit", REG_DYNAMIC_ACTIVE_POWER_LIMIT, 2, self.decode_float32, "%", show_error=show_errors),
        ]

        # These are useful settings, but some firmware rejects one-register reads here.
        # v3 tries the variants that are most likely to respond based on your previous output.
        values.extend([
            self.try_read_candidates(
                "ReactivePwrConfig",
                [
                    (REG_REACTIVE_POWER_CONFIG_B, 1, self.decode_u16, "fallback 0xF104/u16"),
                    (REG_REACTIVE_POWER_CONFIG_B, 2, self.decode_i32, "fallback 0xF104/i32"),
                ],
                show_error=show_errors,
            ),
            self.try_read_candidates(
                "AdvancedPwrControlEn",
                [
                    (REG_ADVANCED_POWER_CONTROL, 2, self.decode_first_u16, "read 2 regs, show first word"),
                    (REG_ADVANCED_POWER_CONTROL, 1, self.decode_u16, "read 1 reg"),
                ],
                show_error=show_errors,
            ),
        ])

        if full:
            values.extend([
                self.try_read("Max Active Power", REG_MAX_ACTIVE_POWER, 2, self.decode_float32, "W", show_error=show_errors),
                self.try_read("Export Control Mode E000", REG_EXPORT_MODE_E, 1, self.decode_u16, show_error=show_errors),
                self.try_read("Export Limit Mode E001", REG_EXPORT_LIMIT_MODE_E, 1, self.decode_u16, show_error=show_errors),
                self.try_read("Export Site Limit E002", REG_EXPORT_SITE_LIMIT_E, 2, self.decode_float32, "W", show_error=show_errors),
                self.try_read("Export Control Mode F700", REG_EXPORT_MODE_F, 1, self.decode_u16, show_error=show_errors),
                self.try_read("Export Limit Mode F701", REG_EXPORT_LIMIT_MODE_F, 1, self.decode_u16, show_error=show_errors),
                self.try_read("Export Site Limit F702", REG_EXPORT_SITE_LIMIT_F, 2, self.decode_float32, "W", show_error=show_errors),
            ])
        return values

    def read_operating_limit(self) -> int:
        return self.decode_u16(self.read_regs(REG_ACTIVE_POWER_LIMIT, 1))

    def read_apparent_power_va(self) -> int:
        #  Register at 40087, decoded as signed 16-bit.
        return self.decode_i16(self.read_regs(REG_APPARENT_POWER_VA, 1))

    def set_active_power_limit(self, percent: int) -> None:
        if not 0 <= percent <= 100:
            raise ValueError("Active power limit must be between 0 and 100 percent")
        self.write_u16(REG_ACTIVE_POWER_LIMIT, percent)

    def enable_basic_power_control(self, reactive_register: int = REG_REACTIVE_POWER_CONFIG_A) -> int:
        """Experimental: enable static power-control settings and commit."""
        self.write_u16(REG_ADVANCED_POWER_CONTROL, 1)
        self.write_u32(reactive_register, 4)
        self.write_u16(REG_COMMIT_POWER_CONTROL, 1)
        time.sleep(10)
        return self.decode_i16(self.read_regs(REG_COMMIT_POWER_CONTROL, 1))

    def probe(self, show_errors: bool = False) -> list[RegisterValue]:
        tests = [
            ("40087 Apparent Power L_AC_VA i16", REG_APPARENT_POWER_VA, 1, self.decode_i16),
            ("F001 Active Power Limit u16", REG_ACTIVE_POWER_LIMIT, 1, self.decode_u16),
            ("F103 Reactive candidate i32", REG_REACTIVE_POWER_CONFIG_A, 2, self.decode_i32),
            ("F104 Reactive candidate u16", REG_REACTIVE_POWER_CONFIG_B, 1, self.decode_u16),
            ("F104 Reactive candidate i32", REG_REACTIVE_POWER_CONFIG_B, 2, self.decode_i32),
            ("F142 Advanced candidate u16", REG_ADVANCED_POWER_CONTROL, 1, self.decode_u16),
            ("F142 Advanced candidate first-of-2", REG_ADVANCED_POWER_CONTROL, 2, self.decode_first_u16),
            ("F300 Enhanced Dynamic Control u16", REG_ENH_DYNAMIC_CONTROL, 1, self.decode_u16),
            ("F322 Dynamic Active Limit float32", REG_DYNAMIC_ACTIVE_POWER_LIMIT, 2, self.decode_float32),
        ]
        return [
            self.try_read(name, address, count, decoder, show_error=show_errors)
            for name, address, count, decoder in tests
        ]


def format_value(value: object) -> str:
    if isinstance(value, float):
        if math.isnan(value):
            return "nan"
        if abs(value) < 1000:
            return f"{value:.3f}"
        return f"{value:.1f}"
    return str(value)


def print_values(title: str, values: list[RegisterValue]) -> None:
    print(f"\n{title}")
    print("-" * 88)
    for item in values:
        suffix = f" {item.unit}" if item.unit else ""
        note = f" [{item.note}]" if item.note else ""
        print(f"0x{item.address:04X}  {item.name:<38} {format_value(item.value)}{suffix}{note}")
    print("-" * 88)


def print_interpretation(full: bool = False) -> None:
    print("Interpretation:")
    print("  Active Power Limit 0%   = standby-like curtailment / no PV production")
    print("  Active Power Limit 100% = normal production limit")
    print("  Apparent Power is the live L_AC_VA value")
    print("  If standby/production works, F103/F104/F142 do not need to be changed.")
    if not full:
        print("  Use 'status --full' only when you want to probe optional/export registers.")


def require_yes(args: argparse.Namespace, action: str) -> None:
    if not getattr(args, "yes", False):
        raise SystemExit(
            f"Refusing to {action} without --yes. Re-run with --yes after checking this is allowed for your installation."
        )


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Read and control SolarEdge inverter power-control settings over Modbus TCP")
    parser.add_argument("--host", default=DEFAULT_HOST, help=f"Inverter IP address, default {DEFAULT_HOST}")
    parser.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"Modbus TCP port, default {DEFAULT_PORT}")
    parser.add_argument("--unit", type=int, default=DEFAULT_UNIT, help=f"Modbus unit/slave id, default {DEFAULT_UNIT}")
    parser.add_argument("--timeout", type=float, default=3.0, help="Modbus timeout in seconds, default 3")
    parser.add_argument("--retries", type=int, default=0, help="Modbus retry count where supported by pymodbus, default 0")
    parser.add_argument("--yes", action="store_true", help="Actually perform write operations")

    def add_yes(subparser: argparse.ArgumentParser) -> None:
        subparser.add_argument("--yes", action="store_true", default=argparse.SUPPRESS, help="Actually perform write operations")

    sub = parser.add_subparsers(dest="command", required=True)

    p_status = sub.add_parser("status", help="Read current control/status registers")
    p_status.add_argument("--full", action="store_true", help="Also probe optional/export registers")
    p_status.add_argument("--show-errors", action="store_true", help="Show full Modbus errors for unavailable registers")

    p_probe = sub.add_parser("probe", help="Test candidate register/count combinations")
    p_probe.add_argument("--show-errors", action="store_true", help="Show full Modbus errors for unavailable registers")

    sub.add_parser("power", help="Read live Apparent Power / L_AC_VA ")

    p_limit = sub.add_parser("limit", help="Set Active Power Limit percentage, 0..100")
    p_limit.add_argument("percent", type=int)
    add_yes(p_limit)

    p_standby = sub.add_parser("standby", help="Alias for: limit 0")
    add_yes(p_standby)

    p_production = sub.add_parser("production", help="Alias for: limit 100")
    add_yes(p_production)

    p_enable = sub.add_parser("enable-control", help="Experimental: set AdvancedPwrControlEn=1 and ReactivePwrConfig=4, then commit")
    p_enable.add_argument("--reactive-register", choices=["F103", "F104"], default="F103", help="ReactivePwrConfig base register to write, default F103")
    add_yes(p_enable)

    return parser


def print_after_write(se: SolarEdgeModbus) -> None:
    try:
        limit = se.read_operating_limit()
        print(f"Read-back Active Power Limit: {limit}%")
    except Exception as exc:
        print(f"Write was sent, but read-back failed: {exc}")


def main(argv: Optional[list[str]] = None) -> int:
    args = build_parser().parse_args(argv)

    try:
        with SolarEdgeModbus(args.host, args.port, args.unit, timeout=args.timeout, retries=args.retries) as se:
            if args.command == "status":
                values = se.read_status(full=args.full, show_errors=args.show_errors)
                title = "SolarEdge Modbus power-control status"
                if args.full:
                    title += " (full probe)"
                print_values(title, values)
                print_interpretation(full=args.full)
                return 0

            if args.command == "probe":
                print_values("SolarEdge Modbus register probe", se.probe(show_errors=args.show_errors))
                return 0

            if args.command == "power":
                value = se.read_apparent_power_va()
                print(f"⚡ Apparent Power (L_AC_VA): {value} VA")
                return 0

            if args.command == "limit":
                require_yes(args, f"set Active Power Limit to {args.percent}%")
                se.set_active_power_limit(args.percent)
                print(f"Set Active Power Limit to {args.percent}%")
                print_after_write(se)
                return 0

            if args.command == "standby":
                require_yes(args, "set standby-like curtailment")
                se.set_active_power_limit(0)
                print("Set Active Power Limit to 0% (standby-like curtailment)")
                print_after_write(se)
                return 0

            if args.command == "production":
                require_yes(args, "resume production")
                se.set_active_power_limit(100)
                print("Set Active Power Limit to 100% (normal production limit)")
                print_after_write(se)
                return 0

            if args.command == "enable-control":
                require_yes(args, "enable and commit power control")
                reactive_register = REG_REACTIVE_POWER_CONFIG_A if args.reactive_register == "F103" else REG_REACTIVE_POWER_CONFIG_B
                status = se.enable_basic_power_control(reactive_register=reactive_register)
                print(f"Commit Power Control status after commit: {status}")
                return 0

    except KeyboardInterrupt:
        print("Interrupted", file=sys.stderr)
        return 130
    except Exception as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1

    return 1


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