diff --git a/packages/modules/devices/sonnen/sonnenbatterie/api.py b/packages/modules/devices/sonnen/sonnenbatterie/api.py new file mode 100644 index 0000000000..a774953267 --- /dev/null +++ b/packages/modules/devices/sonnen/sonnenbatterie/api.py @@ -0,0 +1,350 @@ +#!/usr/bin/env python3 +from enum import Enum +from typing import Dict, List, Optional, TypedDict +from modules.common import req +from modules.common.component_state import BatState, CounterState, InverterState +from modules.common.simcount import SimCounter + + +class RestApi1(): + def __init__(self, host: str) -> None: + self.host = host + + def power_limit_controllable(self) -> bool: + """ + Checks if the power limit is controllable via the REST API. + Returns: + bool: True if controllable, False otherwise. + """ + return False + + def __read(self, device_endpoint: str = 'battery') -> dict: + """ + Reads data from the Sonnenbatterie REST API. + Args: + device_endpoint (str): The device to read data from. Defaults to 'battery'. + Returns: + dict: The JSON response from the API. + """ + return req.get_http_session().get( + f'http://{self.host}:7979/rest/devices/{device_endpoint}', + timeout=5).json() + + def update_battery(self, sim_counter: SimCounter) -> BatState: + """ + Updates the battery state by reading data from the REST API. + Returns: + BatState: The updated battery state. + """ + battery_state = self.__read(device_endpoint="battery") + battery_soc = int(battery_state["M05"]) + battery_export_power = int(battery_state["M34"]) + battery_import_power = int(battery_state["M35"]) + battery_power = battery_import_power - battery_export_power + imported, exported = sim_counter.sim_count(battery_power) + return BatState(power=battery_power, + soc=battery_soc, + imported=imported, + exported=exported) + + +class RestApi2(): + def __init__(self, host: str) -> None: + self.host = host + + def power_limit_controllable(self) -> bool: + """ + Checks if the power limit is controllable via the REST API. + Returns: + bool: True if controllable, False otherwise. + """ + return False + + def __read_element(self, device: str, element: str) -> str: + """ + Reads a specific element from the Sonnenbatterie REST API v2. + Args: + device (str): The device to read data from. + element (str): The specific element to read. + Returns: + str: The value of the specified element. + """ + response = req.get_http_session().get( + f'http://{self.host}:7979/rest/devices/{device}/{element}', + timeout=5) + response.encoding = 'utf-8' + return response.text.strip(" \n\r") + + def update_inverter(self, sim_counter: SimCounter) -> InverterState: + """ + Updates the inverter state by reading data from the REST API v2. + Returns: + InverterState: The updated inverter state. + """ + pv_power = -int(float(self.__read_element(device="battery", element="M03"))) + _, exported = sim_counter.sim_count(pv_power) + return InverterState(exported=exported, + power=pv_power) + + def update_grid_counter(self, sim_counter: SimCounter) -> CounterState: + """ + Updates the grid counter state by reading data from the REST API v2. + Returns: + CounterState: The updated grid counter state. + """ + grid_import_power = -int(float(self.__read_element(device="battery", element="M39"))) + grid_export_power = -int(float(self.__read_element(device="battery", element="M38"))) + grid_power = grid_import_power - grid_export_power + imported, exported = sim_counter.sim_count(grid_power) + return CounterState(power=grid_power, + imported=imported, + exported=exported) + + def update_battery(self, sim_counter: SimCounter) -> BatState: + """ + Updates the battery state by reading data from the REST API v2. + Returns: + BatState: The updated battery state. + """ + battery_soc = int(float(self.__read_element(device="battery", element="M05"))) + battery_export_power = int(float(self.__read_element(device="battery", element="M01"))) + battery_import_power = int(float(self.__read_element(device="battery", element="M02"))) + battery_power = battery_import_power - battery_export_power + imported, exported = sim_counter.sim_count(battery_power) + return BatState(power=battery_power, + soc=battery_soc, + imported=imported, + exported=exported) + + +class JsonApiVersion(Enum): + V1 = "v1" + V2 = "v2" + + +class JsonApi(): + class PowerMeterDirection(Enum): + PRODUCTION = "production" + CONSUMPTION = "consumption" + + class StatusDict(TypedDict): + Apparent_output: int + BackupBuffer: str + BatteryCharging: bool + BatteryDischarging: bool + Consumption_Avg: int + Consumption_W: int + Fac: float + FlowConsumptionBattery: bool + FlowConsumptionGrid: bool + FlowConsumptionProduction: bool + FlowGridBattery: bool + FlowProductionBattery: bool + FlowProductionGrid: bool + GridFeedIn_W: int + IsSystemInstalled: int + OperatingMode: str + Pac_total_W: int + Production_W: int + RSOC: int + RemainingCapacity_Wh: int + Sac1: int + Sac2: int + Sac3: int + SystemStatus: str + Timestamp: str + USOC: int + Uac: float + Ubat: float + + class ChannelDict(TypedDict): + a_l1: int + a_l2: int + a_l3: int + channel: int + deviceid: int + direction: str + error: int + kwh_exported: float + kwh_imported: float + v_l1_l2: float + v_l1_n: float + v_l2_l3: float + v_l2_n: float + v_l3_l1: float + v_l3_n: float + va_total: float + var_total: float + w_l1: float + w_l2: float + w_l3: float + w_total: float + + def __init__(self, + host: str, + api_version: JsonApiVersion = JsonApiVersion.V1, + auth_token: Optional[str] = None) -> None: + self.host = host + self.api_version = api_version + self.auth_token = auth_token + if self.api_version == JsonApiVersion.V2 and self.auth_token is None: + raise ValueError("API v2 requires an auth_token.") + self.headers = {"auth-token": auth_token} if api_version == JsonApiVersion.V2 else {} + + def __read(self, endpoint: str = "status") -> Dict: + """ + Reads data from the Sonnenbatterie JSON API. + Args: + endpoint (str): The endpoint to fetch data from. Defaults to "status". + Returns: + Dict: The JSON response from the API as a dictionary. + """ + return req.get_http_session().get( + f"http://{self.host}/api/{self.api_version.value}/{endpoint}", + timeout=5, + headers=self.headers + ).json() + + def __read_status(self) -> StatusDict: + """ + Reads the status data from the JSON API. + Returns: + StatusDict: The status data as a dictionary. + """ + return self.__read(endpoint="status") + + def __read_power_meter(self, direction: Optional[PowerMeterDirection] = None) -> List[ChannelDict]: + """ + Reads the power meter data from the JSON API. + Args: + direction (Optional[PowerMeterDirection]): The direction of the power meter data. + If None, all data is returned. Defaults to None. + Returns: + List[ChannelDict]: The power meter data as a list of dictionaries. + """ + data = self.__read(endpoint="powermeter") + if direction is not None: + data = [item for item in data if item["direction"] == direction.value] + if len(data) == 0: + raise ValueError(f"No data found for direction: {direction.value}") + return data + + def __counter_state_from_channel(self, channel: ChannelDict, inverted: bool = False) -> CounterState: + """ + Converts a channel dictionary to a CounterState object. + Args: + channel (ChannelDict): The channel data as a dictionary. + Returns: + CounterState: The converted CounterState object. + """ + return CounterState(power=-channel["w_total"] if inverted else channel["w_total"], + powers=[-channel[f"w_l{phase}"] for phase in range(1, 4)] if inverted else + [channel[f"w_l{phase}"] for phase in range(1, 4)], + currents=[-channel[f"a_l{phase}"] for phase in range(1, 4)] if inverted else + [channel[f"a_l{phase}"] for phase in range(1, 4)], + voltages=[channel[f"v_l{phase}_n"] for phase in range(1, 4)], + imported=channel["kwh_exported"] if inverted else channel["kwh_imported"], + exported=channel["kwh_imported"] if inverted else channel["kwh_exported"]) + + def __get_configurations(self) -> Dict: + if self.api_version != JsonApiVersion.V2: + raise ValueError("Diese Methode erfordert die JSON API v2!") + return self.__read(endpoint="configurations") + + def __set_configurations(self, configuration: Dict) -> None: + if self.api_version != JsonApiVersion.V2: + raise ValueError("Diese Methode erfordert die JSON API v2!") + req.get_http_session().put(f"http://{self.host}/api/v2/configurations", + json=configuration, + headers={"Auth-Token": self.auth_token}) + + def __update_set_point(self, power_limit: int) -> None: + if self.api_version != JsonApiVersion.V2: + raise ValueError("Diese Methode erfordert die JSON API v2!") + command = "charge" + if power_limit < 0: + command = "discharge" + power_limit = -power_limit + req.get_http_session().post(f"http://{self.host}/api/v2/setpoint/{command}/{power_limit}", + headers={"Auth-Token": self.auth_token, + "Content-Type": "application/json"}) + + def power_limit_controllable(self) -> bool: + """ + Checks if the power limit is controllable via the JSON API. + Returns: + bool: True if controllable, False otherwise. + """ + return self.api_version == JsonApiVersion.V2 and self.auth_token is not None + + def update_battery(self, sim_counter: SimCounter) -> BatState: + """ + Updates the battery state by reading data from the JSON API. + Returns: + InverterState: The updated battery state. + """ + battery_state = self.__read_status() + battery_power = -battery_state["Pac_total_W"] + battery_soc = battery_state["USOC"] + imported, exported = sim_counter.sim_count(battery_power) + return BatState(power=battery_power, + soc=battery_soc, + imported=imported, + exported=exported) + + def update_grid_counter(self, sim_counter: SimCounter) -> CounterState: + """ + Updates the grid counter state by reading data from the JSON API. + Returns: + CounterState: The updated grid counter state. + """ + counter_state = self.__read_status() + grid_power = -counter_state["GridFeedIn_W"] + grid_voltage = counter_state["Uac"] + grid_frequency = counter_state["Fac"] + imported, exported = sim_counter.sim_count(grid_power) + return CounterState(power=grid_power, + voltages=[grid_voltage]*3, + frequency=grid_frequency, + imported=imported, + exported=exported) + + def update_inverter(self, sim_counter: SimCounter) -> InverterState: + """ + Updates the inverter state by reading data from the JSON API. + Returns: + InverterState: The updated inverter state. + """ + if self.api_version == JsonApiVersion.V1: + inverter_state = self.__read_status() + pv_power = -inverter_state["Production_W"] + _, exported = sim_counter.sim_count(pv_power) + return InverterState(exported=exported, + power=pv_power) + else: + return self.__counter_state_from_channel( + self.__read_power_meter(direction=self.PowerMeterDirection.PRODUCTION)[0], + inverted=True) + + def update_consumption_counter(self) -> CounterState: + """ + Updates the consumption counter state by reading data from the JSON API. + Returns: + CounterState: The updated consumption counter state. + """ + return self.__counter_state_from_channel( + self.__read_power_meter(direction=self.PowerMeterDirection.CONSUMPTION)[0]) + + def set_power_limit(self, power_limit: Optional[int]) -> None: + if self.power_limit_controllable() is False: + raise ValueError("Leistungsvorgabe wird nur für 'JSON-API v2' unterstützt!") + operating_mode = self.__get_configurations()["EM_OperatingMode"] + if power_limit is None: + # Keine Leistungsvorgabe, Betriebsmodus "Eigenverbrauch" aktivieren + if operating_mode == "1": + self.__set_configurations({"EM_OperatingMode": "2"}) + else: + # Leistungsvorgabe, Betriebsmodus "Manuell" aktivieren + if operating_mode == "2": + self.__set_configurations({"EM_OperatingMode": "1"}) + self.__update_set_point(power_limit) diff --git a/packages/modules/devices/sonnen/sonnenbatterie/bat.py b/packages/modules/devices/sonnen/sonnenbatterie/bat.py index aa075eac4b..0705216c59 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/bat.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/bat.py @@ -1,15 +1,15 @@ #!/usr/bin/env python3 -from typing import Any, TypedDict, Dict, Optional import logging +from typing import Any, TypedDict, Optional -from modules.devices.sonnen.sonnenbatterie.config import SonnenbatterieBatSetup -from modules.common.store import get_bat_value_store -from modules.common.simcount import SimCounter -from modules.common.fault_state import ComponentInfo, FaultState -from modules.common.component_type import ComponentDescriptor -from modules.common.component_state import BatState from modules.common.abstract_device import AbstractBat -from modules.common import req +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.sonnen.sonnenbatterie.api import JsonApi, JsonApiVersion, RestApi1, RestApi2 +from modules.devices.sonnen.sonnenbatterie.config import SonnenbatterieBatSetup log = logging.getLogger(__name__) @@ -31,164 +31,29 @@ def initialize(self) -> None: self.__device_id: int = self.kwargs['device_id'] self.__device_address: str = self.kwargs['device_address'] self.__device_variant: int = self.kwargs['device_variant'] - self.__api_v2_token: Optional[str] = self.kwargs.get('api_v2_token') + self.__api_v2_token: Optional[str] = self.kwargs['device_api_v2_token'] + if self.__device_variant not in [0, 1, 2, 3]: + raise ValueError("Unbekannte API: " + str(self.__device_variant)) self.sim_counter = SimCounter(self.__device_id, self.component_config.id, prefix="speicher") self.store = get_bat_value_store(self.component_config.id) self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config)) - - def __read_variant_0(self): - return req.get_http_session().get('http://' + self.__device_address + ':7979/rest/devices/battery', - timeout=5).json() - - def __update_variant_0(self) -> BatState: - # Auslesen einer Sonnenbatterie Eco 4 über die integrierte JSON-API des Batteriesystems - battery_state = self.__read_variant_0() - battery_soc = int(battery_state["M05"]) - battery_export_power = int(battery_state["M34"]) - battery_import_power = int(battery_state["M35"]) - battery_power = battery_import_power - battery_export_power - return BatState( - power=battery_power, - soc=battery_soc - ) - - def __read_variant_1(self, api: str = "v1", target: str = "status") -> Dict: - return req.get_http_session().get( - f"http://{self.__device_address}/api/{api}/{target}", - timeout=5, - headers={"Auth-Token": self.__api_v2_token} if api == "v2" else None - ).json() - - def __update_variant_1(self, api: str = "v1") -> BatState: - # Auslesen einer Sonnenbatterie 8 oder 10 über die integrierte JSON-API v1/v2 des Batteriesystems - ''' - example data: - { - "Apparent_output": 225, - "BackupBuffer": "0", - "BatteryCharging": false, - "BatteryDischarging": false, - "Consumption_Avg": 2114, - "Consumption_W": 2101, - "Fac": 49.97200393676758, - "FlowConsumptionBattery": false, - "FlowConsumptionGrid": true, - "FlowConsumptionProduction": false, - "FlowGridBattery": false, - "FlowProductionBattery": false, - "FlowProductionGrid": false, - "GridFeedIn_W": -2106, - "IsSystemInstalled": 1, - "OperatingMode": "2", - "Pac_total_W": -5, - "Production_W": 0, - "RSOC": 6, - "RemainingCapacity_Wh": 2377, - "Sac1": 75, - "Sac2": 75, - "Sac3": 75, - "SystemStatus": "OnGrid", - "Timestamp": "2021-12-13 07:54:48", - "USOC": 0, - "Uac": 231, - "Ubat": 48, - "dischargeNotAllowed": true, - "generator_autostart": false, - "NVM_REINIT_STATUS": 0 - } - ''' - battery_state = self.__read_variant_1(api) - battery_power = -battery_state["Pac_total_W"] - log.debug('Speicher Leistung: ' + str(battery_power)) - battery_soc = battery_state["USOC"] - log.debug('Speicher SoC: ' + str(battery_soc)) - imported, exported = self.sim_counter.sim_count(battery_power) - return BatState( - power=battery_power, - soc=battery_soc, - imported=imported, - exported=exported - ) - - def __get_json_api_v2_configurations(self) -> Dict: - if self.__device_variant != 3: - raise ValueError("JSON API v2 wird nur für Variante 3 unterstützt!") - return self.__read_variant_1("v2", "configurations") - - def __set_json_api_v2_configurations(self, configuration: Dict) -> None: - if self.__device_variant != 3: - raise ValueError("JSON API v2 wird nur für Variante 3 unterstützt!") - req.get_http_session().put( - f"http://{self.__device_address}/api/v2/configurations", - json=configuration, - headers={"Auth-Token": self.__api_v2_token} - ) - - def __set_json_api_v2_setpoint(self, power_limit: int) -> None: - if self.__device_variant != 3: - raise ValueError("JSON API v2 wird nur für Variante 3 unterstützt!") - command = "charge" - if power_limit < 0: - command = "discharge" - power_limit = -power_limit - req.get_http_session().post( - f"http://{self.__device_address}/api/v2/setpoint/{command}/{power_limit}", - headers={"Auth-Token": self.__api_v2_token, "Content-Type": "application/json"} - ) - - def __read_variant_2_element(self, element: str) -> str: - response = req.get_http_session().get( - 'http://' + self.__device_address + ':7979/rest/devices/battery/' + element, - timeout=5) - response.encoding = 'utf-8' - return response.text.strip(" \n\r") - - def __update_variant_2(self) -> BatState: - # Auslesen einer Sonnenbatterie Eco 6 über die integrierte REST-API des Batteriesystems - battery_soc = int(float(self.__read_variant_2_element("M05"))) - battery_export_power = int(float(self.__read_variant_2_element("M01"))) - battery_import_power = int(float(self.__read_variant_2_element("M02"))) - battery_power = battery_import_power - battery_export_power - return BatState( - power=battery_power, - soc=battery_soc - ) - - def update(self) -> None: - log.debug("Variante: " + str(self.__device_variant)) if self.__device_variant == 0: - state = self.__update_variant_0() - elif self.__device_variant == 1: - state = self.__update_variant_1() + self.api = RestApi1(host=self.__device_address) elif self.__device_variant == 2: - state = self.__update_variant_2() - elif self.__device_variant == 3: - state = self.__update_variant_1("v2") + self.api = RestApi2(host=self.__device_address) else: - raise ValueError("Unbekannte Variante: " + str(self.__device_variant)) - self.store.set(state) + self.api = JsonApi(host=self.__device_address, + api_version=JsonApiVersion.V2 if self.__device_variant == 3 else JsonApiVersion.V1, + auth_token=self.__api_v2_token if self.__device_variant == 3 else None) + + def update(self) -> None: + self.store.set(self.api.update_battery(sim_counter=self.sim_counter)) def set_power_limit(self, power_limit: Optional[int]) -> None: - if self.__device_variant != 3: - raise ValueError("Leistungsvorgabe wird nur für Variante 'JSON-API v2' unterstützt!") - operating_mode = self.__get_json_api_v2_configurations()["EM_OperatingMode"] - log.debug(f"Betriebsmodus: aktuell: {operating_mode}") - if power_limit is None: - # Keine Leistungsvorgabe, Betriebsmodus "Eigenverbrauch" aktivieren - if operating_mode == "1": - log.debug("Keine Leistungsvorgabe, aktiviere normale Steuerung durch den Speicher") - self.__set_json_api_v2_configurations({"EM_OperatingMode": "2"}) - else: - # Leistungsvorgabe, Betriebsmodus "Manuell" aktivieren - if operating_mode == "2": - log.debug(f"Leistungsvorgabe: {power_limit}, aktiviere manuelle Steuerung durch openWB") - self.__set_json_api_v2_configurations({"EM_OperatingMode": "1"}) - log.debug(f"Setze Leistungsvorgabe auf: {power_limit}") - self.__set_json_api_v2_setpoint(power_limit) + self.api.set_power_limit(power_limit=power_limit) def power_limit_controllable(self) -> bool: - # Leistungsvorgabe ist nur für Variante 3 (JSON-API v2) möglich - return self.__device_variant == 3 + return self.api.power_limit_controllable() component_descriptor = ComponentDescriptor(configuration_factory=SonnenbatterieBatSetup) diff --git a/packages/modules/devices/sonnen/sonnenbatterie/config.py b/packages/modules/devices/sonnen/sonnenbatterie/config.py index bf9054ed23..469e4f23d9 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/config.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/config.py @@ -57,7 +57,7 @@ def __init__(self): pass -class SonnenbatterieConsumptionCounterSetup(ComponentSetup[SonnenbatterieCounterConfiguration]): +class SonnenbatterieConsumptionCounterSetup(ComponentSetup[SonnenbatterieConsumptionCounterConfiguration]): def __init__(self, name: str = "SonnenBatterie Verbrauchs-Zähler", type: str = "counter_consumption", diff --git a/packages/modules/devices/sonnen/sonnenbatterie/counter.py b/packages/modules/devices/sonnen/sonnenbatterie/counter.py index a91d6358f3..ab4c154c52 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/counter.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/counter.py @@ -1,14 +1,14 @@ #!/usr/bin/env python3 import logging -from typing import Optional, TypedDict, Any +from typing import Any, TypedDict, Optional -from modules.common import req from modules.common.abstract_device import AbstractCounter -from modules.common.component_state import CounterState 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_counter_value_store + +from modules.devices.sonnen.sonnenbatterie.api import JsonApi, RestApi2, JsonApiVersion from modules.devices.sonnen.sonnenbatterie.config import SonnenbatterieCounterSetup log = logging.getLogger(__name__) @@ -30,105 +30,23 @@ def initialize(self) -> None: self.__device_id: int = self.kwargs['device_id'] self.__device_address: str = self.kwargs['device_address'] self.__device_variant: int = self.kwargs['device_variant'] - self.__api_v2_token: Optional[str] = self.kwargs.get('api_v2_token') + self.__api_v2_token: Optional[str] = self.kwargs['device_api_v2_token'] + if self.__device_variant == 0: + raise ValueError("Die API 'Rest-API 1' bietet keine EVU Daten!") + if self.__device_variant not in [1, 2, 3]: + raise ValueError("Unbekannte API: " + str(self.__device_variant)) self.sim_counter = SimCounter(self.__device_id, self.component_config.id, prefix="bezug") self.store = get_counter_value_store(self.component_config.id) self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config)) - - def __read_variant_1(self, api: str = "v1"): - return req.get_http_session().get( - "http://" + self.__device_address + "/api/" + api + "/status", - timeout=5, - headers={"Auth-Token": self.__api_v2_token} if api == "v2" else None - ).json() - - def __update_variant_1(self, api: str = "v1") -> CounterState: - # Auslesen einer Sonnenbatterie 8 oder 10 über die integrierte JSON-API v1/v2 des Batteriesystems - ''' - example data: - { - "Apparent_output": 225, - "BackupBuffer": "0", - "BatteryCharging": false, - "BatteryDischarging": false, - "Consumption_Avg": 2114, - "Consumption_W": 2101, - "Fac": 49.97200393676758, - "FlowConsumptionBattery": false, - "FlowConsumptionGrid": true, - "FlowConsumptionProduction": false, - "FlowGridBattery": false, - "FlowProductionBattery": false, - "FlowProductionGrid": false, - "GridFeedIn_W": -2106, - "IsSystemInstalled": 1, - "OperatingMode": "2", - "Pac_total_W": -5, - "Production_W": 0, - "RSOC": 6, - "RemainingCapacity_Wh": 2377, - "Sac1": 75, - "Sac2": 75, - "Sac3": 75, - "SystemStatus": "OnGrid", - "Timestamp": "2021-12-13 07:54:48", - "USOC": 0, - "Uac": 231, - "Ubat": 48, - "dischargeNotAllowed": true, - "generator_autostart": false, - "NVM_REINIT_STATUS": 0 - } - ''' - counter_state = self.__read_variant_1(api) - grid_power = -counter_state["GridFeedIn_W"] - log.debug('EVU Leistung: ' + str(grid_power)) - # Es wird nur eine Spannung ausgegeben - grid_voltage = counter_state["Uac"] - log.debug('EVU Spannung: ' + str(grid_voltage)) - grid_frequency = counter_state["Fac"] - log.debug('EVU Netzfrequenz: ' + str(grid_frequency)) - imported, exported = self.sim_counter.sim_count(grid_power) - return CounterState( - power=grid_power, - voltages=[grid_voltage]*3, - frequency=grid_frequency, - imported=imported, - exported=exported, - ) - - def __read_variant_2_element(self, element: str) -> str: - response = req.get_http_session().get( - 'http://' + self.__device_address + ':7979/rest/devices/battery/' + element, - timeout=5) - response.encoding = 'utf-8' - return response.text.strip(" \n\r") - - def __update_variant_2(self) -> CounterState: - # Auslesen einer Sonnenbatterie Eco 6 über die integrierte REST-API des Batteriesystems - grid_import_power = int(float(self.__read_variant_2_element("M39"))) - grid_export_power = int(float(self.__read_variant_2_element("M38"))) - grid_power = grid_import_power - grid_export_power - imported, exported = self.sim_counter.sim_count(grid_power) - return CounterState( - power=grid_power, - imported=imported, - exported=exported, - ) + if self.__device_variant == 2: + self.api = RestApi2(host=self.__device_address) + else: + self.api = JsonApi(host=self.__device_address, + api_version=JsonApiVersion.V2 if self.__device_variant == 3 else JsonApiVersion.V1, + auth_token=self.__api_v2_token if self.__device_variant == 3 else None) def update(self) -> None: - log.debug("Variante: " + str(self.__device_variant)) - if self.__device_variant == 0: - log.debug("Die Variante '0' bietet keine EVU Daten!") - elif self.__device_variant == 1: - state = self.__update_variant_1() - elif self.__device_variant == 2: - state = self.__update_variant_2() - elif self.__device_variant == 3: - state = self.__update_variant_1("v2") - else: - raise ValueError("Unbekannte Variante: " + str(self.__device_variant)) - self.store.set(state) + self.store.set(self.api.update_grid_counter(sim_counter=self.sim_counter)) component_descriptor = ComponentDescriptor(configuration_factory=SonnenbatterieCounterSetup) diff --git a/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py b/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py index 3b46066f35..36f0de4bcf 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py @@ -1,115 +1,46 @@ #!/usr/bin/env python3 import logging -from typing import Dict, Optional, Union +from typing import Any, TypedDict, Optional -from dataclass_utils import dataclass_from_dict -from modules.common import req from modules.common.abstract_device import AbstractCounter -from modules.common.component_state import CounterState from modules.common.component_type import ComponentDescriptor from modules.common.fault_state import ComponentInfo, FaultState from modules.common.store import get_counter_value_store + +from modules.devices.sonnen.sonnenbatterie.api import JsonApi, JsonApiVersion from modules.devices.sonnen.sonnenbatterie.config import SonnenbatterieConsumptionCounterSetup log = logging.getLogger(__name__) +class KwargsDict(TypedDict): + device_address: str + device_variant: int + api_v2_token: Optional[str] + + class SonnenbatterieConsumptionCounter(AbstractCounter): - def __init__(self, - device_address: str, - device_variant: int, - api_v2_token: Optional[str], - component_config: Union[Dict, SonnenbatterieConsumptionCounterSetup]) -> None: - self.__device_address = device_address - self.__device_variant = device_variant - self.__api_v2_token = api_v2_token - self.component_config = dataclass_from_dict(SonnenbatterieConsumptionCounterSetup, component_config) + def __init__(self, component_config: SonnenbatterieConsumptionCounterSetup, **kwargs: Any) -> None: + self.component_config = component_config + self.kwargs: KwargsDict = kwargs + + def initialize(self) -> None: + self.__device_address: str = self.kwargs['device_address'] + self.__device_variant: int = self.kwargs['device_variant'] + self.__api_v2_token: Optional[str] = self.kwargs['device_api_v2_token'] + if self.__device_variant in [0, 1, 2]: + raise ValueError("Die ausgewählte API bietet keine Verbrauchsdaten!") + if self.__device_variant != 3: + raise ValueError("Unbekannte API: " + str(self.__device_variant)) + self.store = get_counter_value_store(self.component_config.id) self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config)) - - def __read_variant_3(self): - result = req.get_http_session().get( - "http://" + self.__device_address + "/api/v2/powermeter", - timeout=5, - headers={"Auth-Token": self.__api_v2_token} - ).json() - for channel in result: - if channel["direction"] == "consumption": - return channel - raise ValueError("No consumption channel found") - - def __update_variant_3(self) -> CounterState: - # Auslesen einer Sonnenbatterie 8 oder 10 über die integrierte JSON-API v2 des Batteriesystems - ''' - example data: - [ - { - "a_l1": 0, - "a_l2": 0, - "a_l3": 0, - "channel": 1, - "deviceid": 4, - "direction": "production", - "error": -1, - "kwh_exported": 0, - "kwh_imported": 0, - "v_l1_l2": 0, - "v_l1_n": 0, - "v_l2_l3": 0, - "v_l2_n": 0, - "v_l3_l1": 0, - "v_l3_n": 0, - "va_total": 0, - "var_total": 0, - "w_l1": 0, - "w_l2": 0, - "w_l3": 0, - "w_total": 0 - }, - { - "a_l1": 0, - "a_l2": 0, - "a_l3": 0, - "channel": 2, - "deviceid": 4, - "direction": "consumption", - "error": -1, - "kwh_exported": 0, - "kwh_imported": 0, - "v_l1_l2": 0, - "v_l1_n": 0, - "v_l2_l3": 0, - "v_l2_n": 0, - "v_l3_l1": 0, - "v_l3_n": 0, - "va_total": 0, - "var_total": 0, - "w_l1": 0, - "w_l2": 0, - "w_l3": 0, - "w_total": 0 - } - ] - ''' - counter_state = self.__read_variant_3() - return CounterState( - power=counter_state["w_total"], - powers=[counter_state[f"w_l{phase}"] for phase in range(1, 4)], - currents=[counter_state[f"a_l{phase}"] for phase in range(1, 4)], - voltages=[counter_state[f"v_l{phase}_n"] for phase in range(1, 4)], - imported=counter_state["kwh_imported"], - exported=counter_state["kwh_exported"] - ) + self.api = JsonApi(host=self.__device_address, + api_version=JsonApiVersion.V2, + auth_token=self.__api_v2_token) def update(self) -> None: - log.debug("Variante: " + str(self.__device_variant)) - if self.__device_variant in [0, 1, 2]: - log.debug("Diese Variante bietet keine Verbrauchsdaten!") - elif self.__device_variant == 3: - state = self.__update_variant_3() - else: - raise ValueError("Unbekannte Variante: " + str(self.__device_variant)) - self.store.set(state) + self.store.set(self.api.update_consumption_counter()) component_descriptor = ComponentDescriptor(configuration_factory=SonnenbatterieConsumptionCounterSetup) diff --git a/packages/modules/devices/sonnen/sonnenbatterie/device.py b/packages/modules/devices/sonnen/sonnenbatterie/device.py index 0d0064ae8b..5ec5f3cc1f 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/device.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/device.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -""" Modul zum Auslesen von sonnenBatterie Speichern. +""" Modul zum Auslesen von SonnenBatterie Speichern. """ import logging @@ -34,10 +34,10 @@ def create_evu_counter_component(component_config: SonnenbatterieCounterSetup): device_api_v2_token=device_config.configuration.api_v2_token) def create_consumption_counter_component(component_config: SonnenbatterieConsumptionCounterSetup): - return SonnenbatterieConsumptionCounter(device_config.configuration.ip_address, - device_config.configuration.variant, - device_config.configuration.api_v2_token, - component_config) + return SonnenbatterieConsumptionCounter(component_config, + device_address=device_config.configuration.ip_address, + device_variant=device_config.configuration.variant, + device_api_v2_token=device_config.configuration.api_v2_token) def create_inverter_component(component_config: SonnenbatterieInverterSetup): return SonnenbatterieInverter(component_config, diff --git a/packages/modules/devices/sonnen/sonnenbatterie/inverter.py b/packages/modules/devices/sonnen/sonnenbatterie/inverter.py index f2f584a98e..446420297e 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/inverter.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/inverter.py @@ -1,15 +1,15 @@ #!/usr/bin/env python3 -from typing import Any, Optional, TypedDict import logging -from modules.devices.sonnen.sonnenbatterie.config import SonnenbatterieInverterSetup -from modules.common.store import get_inverter_value_store -from modules.common.simcount import SimCounter -from modules.common.fault_state import ComponentInfo, FaultState -from modules.common.component_type import ComponentDescriptor -from modules.common.component_state import InverterState +from typing import Any, TypedDict, Optional + from modules.common.abstract_device import AbstractInverter -from modules.common import req +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_inverter_value_store +from modules.devices.sonnen.sonnenbatterie.api import JsonApi, RestApi2, JsonApiVersion +from modules.devices.sonnen.sonnenbatterie.config import SonnenbatterieInverterSetup log = logging.getLogger(__name__) @@ -30,94 +30,23 @@ def initialize(self) -> None: self.__device_id: int = self.kwargs['device_id'] self.__device_address: str = self.kwargs['device_address'] self.__device_variant: int = self.kwargs['device_variant'] - self.__api_v2_token: Optional[str] = self.kwargs.get('api_v2_token') + self.__api_v2_token: Optional[str] = self.kwargs['device_api_v2_token'] + if self.__device_variant == 0: + raise ValueError("Die Variante '0' bietet keine PV Daten!") + if self.__device_variant not in [1, 2, 3]: + raise ValueError("Unbekannte API: " + str(self.__device_variant)) self.sim_counter = SimCounter(self.__device_id, self.component_config.id, prefix="pv") self.store = get_inverter_value_store(self.component_config.id) self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config)) - - def __read_variant_1(self, api: str = "v1"): - return req.get_http_session().get( - "http://" + self.__device_address + "/api/" + api + "/status", - timeout=5, - headers={"Auth-Token": self.__api_v2_token} if api == "v2" else None - ).json() - - def __update_variant_1(self, api: str = "v1") -> InverterState: - # Auslesen einer Sonnenbatterie 8 oder 10 über die integrierte JSON-API v1/v2 des Batteriesystems - ''' - example data: - { - "Apparent_output": 225, - "BackupBuffer": "0", - "BatteryCharging": false, - "BatteryDischarging": false, - "Consumption_Avg": 2114, - "Consumption_W": 2101, - "Fac": 49.97200393676758, - "FlowConsumptionBattery": false, - "FlowConsumptionGrid": true, - "FlowConsumptionProduction": false, - "FlowGridBattery": false, - "FlowProductionBattery": false, - "FlowProductionGrid": false, - "GridFeedIn_W": -2106, - "IsSystemInstalled": 1, - "OperatingMode": "2", - "Pac_total_W": -5, - "Production_W": 0, - "RSOC": 6, - "RemainingCapacity_Wh": 2377, - "Sac1": 75, - "Sac2": 75, - "Sac3": 75, - "SystemStatus": "OnGrid", - "Timestamp": "2021-12-13 07:54:48", - "USOC": 0, - "Uac": 231, - "Ubat": 48, - "dischargeNotAllowed": true, - "generator_autostart": false, - "NVM_REINIT_STATUS": 0 - } - ''' - inverter_state = self.__read_variant_1(api) - pv_power = -inverter_state["Production_W"] - log.debug('Speicher PV Leistung: ' + str(pv_power)) - _, exported = self.sim_counter.sim_count(pv_power) - return InverterState( - exported=exported, - power=pv_power - ) - - def __read_variant_2_element(self, element: str) -> str: - response = req.get_http_session().get('http://' + self.__device_address + - ':7979/rest/devices/battery/' + element, timeout=5) - response.encoding = 'utf-8' - return response.text.strip(" \n\r") - - def __update_variant_2(self) -> InverterState: - # Auslesen einer Sonnenbatterie Eco 6 über die integrierte REST-API des Batteriesystems - pv_power = -int(float(self.__read_variant_2_element("M03"))) - log.debug('Speicher PV Leistung: ' + str(pv_power)) - _, exported = self.sim_counter.sim_count(pv_power) - return InverterState( - exported=exported, - power=pv_power - ) + if self.__device_variant == 2: + self.api = RestApi2(host=self.__device_address) + else: + self.api = JsonApi(host=self.__device_address, + api_version=JsonApiVersion.V2 if self.__device_variant == 3 else JsonApiVersion.V1, + auth_token=self.__api_v2_token if self.__device_variant == 3 else None) def update(self) -> None: - log.debug("Variante: " + str(self.__device_variant)) - if self.__device_variant == 0: - log.debug("Die Variante '0' bietet keine PV Daten!") - elif self.__device_variant == 1: - state = self.__update_variant_1() - elif self.__device_variant == 2: - state = self.__update_variant_2() - elif self.__device_variant == 3: - state = self.__update_variant_1("v2") - else: - raise ValueError("Unbekannte Variante: " + str(self.__device_variant)) - self.store.set(state) + self.store.set(self.api.update_inverter(sim_counter=self.sim_counter)) component_descriptor = ComponentDescriptor(configuration_factory=SonnenbatterieInverterSetup)