openWB Forum

Code: Alles auswählen

bat.py ersetzen in Solaredge

Code: Alles auswählen

der Controll Mode 4 muss gesetzt werden , ich habe die Funktion auf dem Script jetzt entfernt , weil ich der Meinung bin das openwb keine Konfiguration ändern sollte am Fremdsystem , der Modus 4 Remote 0xE004 muss durch den User aktiv gesetzt werden 

Code: Alles auswählen

#!/usr/bin/env python3
"""
Dieses Skript ermöglicht die optionale Steuerung von SolarEdge-Batterien über Modbus. 
Für eine Fernsteuerung der Batterie muss der Wert des Registers StorageConf_CtrlMode (0xE004) auf "4" (Remote) gesetzt sein. 
Die Steuerung erfolgt über das Setzen von Leistungsbegrenzungen, Abfrage des Ladezustands (SOC) und der Batterieleistung.
"""

import logging
from typing import Optional, Dict, Union

from pymodbus.payload import BinaryPayloadDecoder, BinaryPayloadBuilder
from pymodbus.constants import Endian
from dataclass_utils import dataclass_from_dict
from modules.common import modbus
from modules.common.abstract_device import AbstractBat
from modules.common.component_state import BatState
from modules.common.component_type import ComponentDescriptor
from modules.common.fault_state import ComponentInfo, FaultState
from modules.common.simcount import SimCounter
from modules.common.store import get_bat_value_store
from modules.devices.solaredge.solaredge.config import SolaredgeBatSetup

# Initialisiere Logger für die Debug- und Fehlerausgabe
log = logging.getLogger(__name__)


class SolaredgeConfig:
    """
    Konfigurationsklasse für SolarEdge-Registeradressen. 
    Diese Klasse zentralisiert Adressen, um Änderungen einfach umzusetzen.
    """

    def __init__(self):
        # Register für Leistungsbegrenzung
        self.power_limit_register = 0xE010  # Maximal erlaubte Lade-/Entladeleistung
        # Register für den Ladezustand der Batterie (State of Charge - SOC)
        self.soc_register = 0xE184  # SOC in Prozent
        # Register für aktuelle Batterieleistung
        self.power_register = 0xE174  # Momentanleistung der Batterie


class SolaredgeBat(AbstractBat):
    """
    Klasse zur Steuerung eines SolarEdge-Speichersystems.
    Ermöglicht Modbus-Kommunikation zum Lesen und Schreiben von Parametern.
    """

    def __init__(self,
                 device_id: int,
                 component_config: Union[Dict, SolaredgeBatSetup],
                 tcp_client: modbus.ModbusTcpClient_,
                 config: SolaredgeConfig) -> None:
        """
        Initialisiert die SolarEdge-Batterie-Klasse.

        :param device_id: Eindeutige ID des Geräts für die Kommunikation.
        :param component_config: Konfigurationseinstellungen für die Batterie.
        :param tcp_client: Modbus-TCP-Client für die Kommunikation.
        :param config: Instanz der Konfigurationsklasse mit Registeradressen.
        """
        self.__device_id = device_id  # Geräte-ID
        self.component_config = dataclass_from_dict(SolaredgeBatSetup, component_config)
        self.__tcp_client = tcp_client  # Modbus-TCP-Client für Lese-/Schreiboperationen
        self.config = config  # SolarEdge-Registerkonfiguration
        self.sim_counter = SimCounter(self.__device_id, self.component_config.id, prefix="speicher")
        self.store = get_bat_value_store(self.component_config.id)  # Interner Zustandsspeicher
        self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config))

    def set_power_limit(self, power_limit: Optional[int]) -> None:
        """
        Setzt die Leistungsbegrenzung für das Laden/Entladen.

        :param power_limit: Die gewünschte Begrenzung in Watt.
                            - Ein positiver Wert aktiviert die Steuerung.
                            - None oder -1 setzt die Begrenzung zurück (keine Limitierung).
        """
        unit = self.component_config.configuration.modbus_id
        register = self.config.power_limit_register

        try:
            if power_limit is None:
                power_limit = -1
                log.debug("Keine Leistungsbegrenzung angegeben. Begrenzung wird deaktiviert (-1).")

            # Versuch, den Wert direkt als INT32 zu schreiben
            try:
                current_limit = self.__tcp_client.read_holding_registers(register, ModbusDataType.INT_32, unit=unit)
                if current_limit == power_limit:
                    log.debug(f"Aktuelle Leistungsbegrenzung ({current_limit} W) entspricht dem Sollwert ({power_limit} W). Kein Schreibvorgang erforderlich.")
                    return
                self.__tcp_client.write_registers(register, power_limit, ModbusDataType.INT_32, unit=unit)
                log.info(f"Leistungsbegrenzung erfolgreich auf {power_limit} W gesetzt.")
            except Exception as e:
                log.warning(f"Direkte INT32-Verarbeitung fehlgeschlagen: {e}. Wechsel zu manueller Dekodierung.")

                # Fallback auf manuelle Zerlegung und Schreiben
                builder = BinaryPayloadBuilder(byteorder=Endian.Little, wordorder=Endian.Little)
                builder.add_32bit_int(power_limit)
                registers = builder.to_registers()
                self.__tcp_client.write_registers(register, registers, unit=unit)
                log.info(f"Leistungsbegrenzung erfolgreich (Fallback) auf {power_limit} W gesetzt.")
        except Exception as e:
            log.error(f"Fehler beim Setzen der Leistungsbegrenzung: {e}")
            raise

    def get_power_limit(self) -> Optional[int]:
        """
        Liest die aktuell eingestellte Leistungsbegrenzung.

        :return: Der aktuelle Wert der Leistungsbegrenzung in Watt (INT32).
        """
        unit = self.component_config.configuration.modbus_id
        register = self.config.power_limit_register

        try:
            # Direkt INT32 lesen
            try:
                current_limit = self.__tcp_client.read_holding_registers(register, ModbusDataType.INT_32, unit=unit)
                log.debug(f"Aktuelle Leistungsbegrenzung aus Register {register}: {current_limit} W")
                return current_limit
            except Exception as e:
                log.warning(f"Direkte INT32-Verarbeitung fehlgeschlagen: {e}. Wechsel zu manueller Dekodierung.")

                # Fallback auf manuelle Dekodierung
                response = self.__tcp_client.read_holding_registers(register, count=2, unit=unit)
                if response.isError():
                    raise Exception(f"Fehler beim Lesen der Leistungsbegrenzung aus Register {register}.")
                decoder = BinaryPayloadDecoder.fromRegisters(
                    response.registers, byteorder=Endian.Little, wordorder=Endian.Little
                )
                current_limit = decoder.decode_32bit_int()
                log.debug(f"Aktuelle Leistungsbegrenzung (Fallback) aus Register {register}: {current_limit} W")
                return current_limit
        except Exception as e:
            log.error(f"Fehler beim Lesen der Leistungsbegrenzung: {e}")
            raise


# Beschreibung der Konfigurationskomponente
component_descriptor = ComponentDescriptor(configuration_factory=SolaredgeBatSetup)