From 08f4a98189d63c48db0eada5a2633844105c773d Mon Sep 17 00:00:00 2001 From: Lutz Bender Date: Wed, 16 Apr 2025 08:07:04 +0200 Subject: [PATCH 1/3] bugfix --- .../devices/sonnen/sonnenbatterie/bat.py | 50 +++++++++++-------- .../devices/sonnen/sonnenbatterie/config.py | 2 +- .../devices/sonnen/sonnenbatterie/counter.py | 26 +++++----- .../sonnenbatterie/counter_consumption.py | 42 +++++++++------- .../devices/sonnen/sonnenbatterie/device.py | 8 +-- .../devices/sonnen/sonnenbatterie/inverter.py | 22 ++++---- 6 files changed, 82 insertions(+), 68 deletions(-) diff --git a/packages/modules/devices/sonnen/sonnenbatterie/bat.py b/packages/modules/devices/sonnen/sonnenbatterie/bat.py index aa075eac4b..98558d0466 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/bat.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/bat.py @@ -31,18 +31,18 @@ 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'] 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): + def __read_rest_api_1(self): return req.get_http_session().get('http://' + self.__device_address + ':7979/rest/devices/battery', timeout=5).json() - def __update_variant_0(self) -> BatState: + def __update_rest_api_1(self) -> BatState: # Auslesen einer Sonnenbatterie Eco 4 über die integrierte JSON-API des Batteriesystems - battery_state = self.__read_variant_0() + battery_state = self.__read_rest_api_1() battery_soc = int(battery_state["M05"]) battery_export_power = int(battery_state["M34"]) battery_import_power = int(battery_state["M35"]) @@ -52,14 +52,24 @@ def __update_variant_0(self) -> BatState: soc=battery_soc ) - def __read_variant_1(self, api: str = "v1", target: str = "status") -> Dict: + def __read_json_api(self, api_version: str = "v1", target: str = "status") -> Dict: + """ + Reads data from the Sonnenbatterie JSON API. + + Args: + api (str): The API version to use ("v1" or "v2"). Defaults to "v1". + target (str): The target 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.__device_address}/api/{api}/{target}", + f"http://{self.__device_address}/api/{api_version}/{target}", timeout=5, - headers={"Auth-Token": self.__api_v2_token} if api == "v2" else None + headers={"auth-token": self.__api_v2_token} if api_version == "v2" else {} ).json() - def __update_variant_1(self, api: str = "v1") -> BatState: + def __update_json_api(self, api_version: str = "v1") -> BatState: # Auslesen einer Sonnenbatterie 8 oder 10 über die integrierte JSON-API v1/v2 des Batteriesystems ''' example data: @@ -97,7 +107,7 @@ def __update_variant_1(self, api: str = "v1") -> BatState: "NVM_REINIT_STATUS": 0 } ''' - battery_state = self.__read_variant_1(api) + battery_state = self.__read_json_api(api_version=api_version, target="status") battery_power = -battery_state["Pac_total_W"] log.debug('Speicher Leistung: ' + str(battery_power)) battery_soc = battery_state["USOC"] @@ -113,7 +123,7 @@ def __update_variant_1(self, api: str = "v1") -> BatState: 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") + return self.__read_json_api("v2", "configurations") def __set_json_api_v2_configurations(self, configuration: Dict) -> None: if self.__device_variant != 3: @@ -136,7 +146,7 @@ def __set_json_api_v2_setpoint(self, power_limit: int) -> None: headers={"Auth-Token": self.__api_v2_token, "Content-Type": "application/json"} ) - def __read_variant_2_element(self, element: str) -> str: + def __read_rest_api_2_element(self, element: str) -> str: response = req.get_http_session().get( 'http://' + self.__device_address + ':7979/rest/devices/battery/' + element, timeout=5) @@ -145,9 +155,9 @@ def __read_variant_2_element(self, element: str) -> str: 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_soc = int(float(self.__read_rest_api_2_element(element="M05"))) + battery_export_power = int(float(self.__read_rest_api_2_element(element="M01"))) + battery_import_power = int(float(self.__read_rest_api_2_element(element="M02"))) battery_power = battery_import_power - battery_export_power return BatState( power=battery_power, @@ -157,20 +167,20 @@ def __update_variant_2(self) -> BatState: def update(self) -> None: log.debug("Variante: " + str(self.__device_variant)) if self.__device_variant == 0: - state = self.__update_variant_0() + state = self.__update_rest_api_1() elif self.__device_variant == 1: - state = self.__update_variant_1() + state = self.__update_json_api(api_version="v1") elif self.__device_variant == 2: state = self.__update_variant_2() elif self.__device_variant == 3: - state = self.__update_variant_1("v2") + state = self.__update_json_api(api_version="v2") else: - raise ValueError("Unbekannte Variante: " + str(self.__device_variant)) + raise ValueError("Unbekannte API: " + str(self.__device_variant)) self.store.set(state) 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!") + raise ValueError("Leistungsvorgabe wird nur für '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: @@ -188,7 +198,7 @@ def set_power_limit(self, power_limit: Optional[int]) -> None: 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.__device_variant == 3 and self.__api_v2_token is not None 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..51e029b667 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/counter.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/counter.py @@ -35,14 +35,14 @@ def initialize(self) -> None: 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"): + def __read_json_api(self, api_version: str = "v1"): return req.get_http_session().get( - "http://" + self.__device_address + "/api/" + api + "/status", + "http://" + self.__device_address + "/api/" + api_version + "/status", timeout=5, - headers={"Auth-Token": self.__api_v2_token} if api == "v2" else None + headers={"Auth-Token": self.__api_v2_token} if api_version == "v2" else None ).json() - def __update_variant_1(self, api: str = "v1") -> CounterState: + def __update_json_api(self, api_version: str = "v1") -> CounterState: # Auslesen einer Sonnenbatterie 8 oder 10 über die integrierte JSON-API v1/v2 des Batteriesystems ''' example data: @@ -80,7 +80,7 @@ def __update_variant_1(self, api: str = "v1") -> CounterState: "NVM_REINIT_STATUS": 0 } ''' - counter_state = self.__read_variant_1(api) + counter_state = self.__read_json_api(api_version=api_version) grid_power = -counter_state["GridFeedIn_W"] log.debug('EVU Leistung: ' + str(grid_power)) # Es wird nur eine Spannung ausgegeben @@ -97,17 +97,17 @@ def __update_variant_1(self, api: str = "v1") -> CounterState: exported=exported, ) - def __read_variant_2_element(self, element: str) -> str: + def __read_rest_api_2(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: + def __update_rest_api_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_import_power = int(float(self.__read_rest_api_2(element="M39"))) + grid_export_power = int(float(self.__read_rest_api_2(element="M38"))) grid_power = grid_import_power - grid_export_power imported, exported = self.sim_counter.sim_count(grid_power) return CounterState( @@ -121,13 +121,13 @@ def update(self) -> None: if self.__device_variant == 0: log.debug("Die Variante '0' bietet keine EVU Daten!") elif self.__device_variant == 1: - state = self.__update_variant_1() + state = self.__update_json_api(api_version="v1") elif self.__device_variant == 2: - state = self.__update_variant_2() + state = self.__update_rest_api_2() elif self.__device_variant == 3: - state = self.__update_variant_1("v2") + state = self.__update_json_api(api_version="v2") else: - raise ValueError("Unbekannte Variante: " + str(self.__device_variant)) + raise ValueError("Unbekannte API: " + str(self.__device_variant)) self.store.set(state) diff --git a/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py b/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py index 3b46066f35..0d5621dc07 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py @@ -1,8 +1,7 @@ #!/usr/bin/env python3 import logging -from typing import Dict, Optional, Union +from typing import Optional, Any, TypedDict -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 @@ -14,20 +13,25 @@ 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'] 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): + def __read_json_api_v2(self): result = req.get_http_session().get( "http://" + self.__device_address + "/api/v2/powermeter", timeout=5, @@ -38,7 +42,7 @@ def __read_variant_3(self): return channel raise ValueError("No consumption channel found") - def __update_variant_3(self) -> CounterState: + def __update_json_api_v2(self) -> CounterState: # Auslesen einer Sonnenbatterie 8 oder 10 über die integrierte JSON-API v2 des Batteriesystems ''' example data: @@ -91,7 +95,7 @@ def __update_variant_3(self) -> CounterState: } ] ''' - counter_state = self.__read_variant_3() + counter_state = self.__read_json_api_v2() return CounterState( power=counter_state["w_total"], powers=[counter_state[f"w_l{phase}"] for phase in range(1, 4)], @@ -103,12 +107,12 @@ def __update_variant_3(self) -> CounterState: 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() + if self.__device_variant == 3: + state = self.__update_json_api_v2() + elif self.__device_variant in [0, 1, 2]: + log.debug("Die ausgewählte API bietet keine Verbrauchsdaten!") else: - raise ValueError("Unbekannte Variante: " + str(self.__device_variant)) + raise ValueError("Unbekannte API: " + str(self.__device_variant)) self.store.set(state) diff --git a/packages/modules/devices/sonnen/sonnenbatterie/device.py b/packages/modules/devices/sonnen/sonnenbatterie/device.py index 0d0064ae8b..d9cc18b987 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/device.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/device.py @@ -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..36cfaacb57 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/inverter.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/inverter.py @@ -35,14 +35,14 @@ def initialize(self) -> None: 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"): + def __read_json_api(self, api_version: str = "v1"): return req.get_http_session().get( - "http://" + self.__device_address + "/api/" + api + "/status", + "http://" + self.__device_address + "/api/" + api_version + "/status", timeout=5, - headers={"Auth-Token": self.__api_v2_token} if api == "v2" else None + headers={"Auth-Token": self.__api_v2_token} if api_version == "v2" else None ).json() - def __update_variant_1(self, api: str = "v1") -> InverterState: + def __update_json_api(self, api_version: str = "v1") -> InverterState: # Auslesen einer Sonnenbatterie 8 oder 10 über die integrierte JSON-API v1/v2 des Batteriesystems ''' example data: @@ -80,7 +80,7 @@ def __update_variant_1(self, api: str = "v1") -> InverterState: "NVM_REINIT_STATUS": 0 } ''' - inverter_state = self.__read_variant_1(api) + inverter_state = self.__read_json_api(api_version=api_version) pv_power = -inverter_state["Production_W"] log.debug('Speicher PV Leistung: ' + str(pv_power)) _, exported = self.sim_counter.sim_count(pv_power) @@ -89,15 +89,15 @@ def __update_variant_1(self, api: str = "v1") -> InverterState: power=pv_power ) - def __read_variant_2_element(self, element: str) -> str: + def __read_rest_api_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: + def __update_rest_api_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"))) + pv_power = -int(float(self.__read_rest_api_2_element(element="M03"))) log.debug('Speicher PV Leistung: ' + str(pv_power)) _, exported = self.sim_counter.sim_count(pv_power) return InverterState( @@ -110,11 +110,11 @@ def update(self) -> None: if self.__device_variant == 0: log.debug("Die Variante '0' bietet keine PV Daten!") elif self.__device_variant == 1: - state = self.__update_variant_1() + state = self.__update_json_api(api_version="v1") elif self.__device_variant == 2: - state = self.__update_variant_2() + state = self.__update_rest_api_2() elif self.__device_variant == 3: - state = self.__update_variant_1("v2") + state = self.__update_json_api(api_version="v2") else: raise ValueError("Unbekannte Variante: " + str(self.__device_variant)) self.store.set(state) From dbc4ba00fb228e85dba2da8c1da387d135e8cc75 Mon Sep 17 00:00:00 2001 From: Lutz Bender Date: Wed, 16 Apr 2025 11:16:10 +0200 Subject: [PATCH 2/3] refactor module --- .../devices/sonnen/sonnenbatterie/api.py | 398 ++++++++++++++++++ .../devices/sonnen/sonnenbatterie/bat.py | 185 +------- .../devices/sonnen/sonnenbatterie/counter.py | 112 +---- .../sonnenbatterie/counter_consumption.py | 97 +---- .../devices/sonnen/sonnenbatterie/device.py | 2 +- .../devices/sonnen/sonnenbatterie/inverter.py | 111 +---- 6 files changed, 466 insertions(+), 439 deletions(-) create mode 100644 packages/modules/devices/sonnen/sonnenbatterie/api.py diff --git a/packages/modules/devices/sonnen/sonnenbatterie/api.py b/packages/modules/devices/sonnen/sonnenbatterie/api.py new file mode 100644 index 0000000000..4343c659a3 --- /dev/null +++ b/packages/modules/devices/sonnen/sonnenbatterie/api.py @@ -0,0 +1,398 @@ +#!/usr/bin/env python3 +from typing import Dict, Optional +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 JsonApi(): + def __init__(self, host: str, api_version: str = "v1", auth_token: Optional[str] = None) -> None: + self.host = host + self.api_version = api_version + self.auth_token = auth_token + if self.api_version == "v2" and self.auth_token is None: + raise ValueError("API v2 requires an auth_token.") + self.headers = {"auth-token": auth_token} if api_version == "v2" else {} + + 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 == "v2" and self.auth_token is not None + + 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}/{endpoint}", + timeout=5, + headers=self.headers + ).json() + + 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. + 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(endpoint="status") + pv_power = -inverter_state["Production_W"] + _, 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 JSON API. + Returns: + CounterState: The updated grid counter state. + 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(endpoint="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_consumption_counter(self) -> CounterState: + """ + Updates the consumption counter state by reading data from the JSON API. + Returns: + CounterState: The updated consumption counter state. + 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 + } + ] + """ + result = self.read(endpoint="powermeter") + for channel in result: + if channel["direction"] == "consumption": + return CounterState(power=channel["w_total"], + powers=[channel[f"w_l{phase}"] for phase in range(1, 4)], + currents=[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_imported"], + exported=channel["kwh_exported"]) + raise ValueError("No consumption data found in the response.") + + 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. + 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 + } + """ + battery_state = self.read(endpoint="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 get_configurations(self) -> Dict: + if self.api_version != "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 != "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 != "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 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 98558d0466..e15510a474 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, RestApi1, RestApi2 +from modules.devices.sonnen.sonnenbatterie.config import SonnenbatterieBatSetup log = logging.getLogger(__name__) @@ -32,173 +32,28 @@ 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 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_rest_api_1(self): - return req.get_http_session().get('http://' + self.__device_address + ':7979/rest/devices/battery', - timeout=5).json() - - def __update_rest_api_1(self) -> BatState: - # Auslesen einer Sonnenbatterie Eco 4 über die integrierte JSON-API des Batteriesystems - battery_state = self.__read_rest_api_1() - 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_json_api(self, api_version: str = "v1", target: str = "status") -> Dict: - """ - Reads data from the Sonnenbatterie JSON API. - - Args: - api (str): The API version to use ("v1" or "v2"). Defaults to "v1". - target (str): The target 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.__device_address}/api/{api_version}/{target}", - timeout=5, - headers={"auth-token": self.__api_v2_token} if api_version == "v2" else {} - ).json() - - def __update_json_api(self, api_version: 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_json_api(api_version=api_version, target="status") - 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_json_api("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_rest_api_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_rest_api_2_element(element="M05"))) - battery_export_power = int(float(self.__read_rest_api_2_element(element="M01"))) - battery_import_power = int(float(self.__read_rest_api_2_element(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_rest_api_1() - elif self.__device_variant == 1: - state = self.__update_json_api(api_version="v1") + 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_json_api(api_version="v2") + self.api = RestApi2(host=self.__device_address) else: - raise ValueError("Unbekannte API: " + str(self.__device_variant)) - self.store.set(state) + self.api = JsonApi(host=self.__device_address, + api_version="v2" if self.__device_variant == 3 else "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 '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 and self.__api_v2_token is not None + return self.api.power_limit_controllable() component_descriptor = ComponentDescriptor(configuration_factory=SonnenbatterieBatSetup) diff --git a/packages/modules/devices/sonnen/sonnenbatterie/counter.py b/packages/modules/devices/sonnen/sonnenbatterie/counter.py index 51e029b667..c92af963ba 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 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_json_api(self, api_version: str = "v1"): - return req.get_http_session().get( - "http://" + self.__device_address + "/api/" + api_version + "/status", - timeout=5, - headers={"Auth-Token": self.__api_v2_token} if api_version == "v2" else None - ).json() - - def __update_json_api(self, api_version: 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_json_api(api_version=api_version) - 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_rest_api_2(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_rest_api_2(self) -> CounterState: - # Auslesen einer Sonnenbatterie Eco 6 über die integrierte REST-API des Batteriesystems - grid_import_power = int(float(self.__read_rest_api_2(element="M39"))) - grid_export_power = int(float(self.__read_rest_api_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="v2" if self.__device_variant == 3 else "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_json_api(api_version="v1") - elif self.__device_variant == 2: - state = self.__update_rest_api_2() - elif self.__device_variant == 3: - state = self.__update_json_api(api_version="v2") - else: - raise ValueError("Unbekannte API: " + 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 0d5621dc07..50b8e76875 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py @@ -1,13 +1,13 @@ #!/usr/bin/env python3 import logging -from typing import Optional, Any, TypedDict +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.store import get_counter_value_store + +from modules.devices.sonnen.sonnenbatterie.api import JsonApi from modules.devices.sonnen.sonnenbatterie.config import SonnenbatterieConsumptionCounterSetup log = logging.getLogger(__name__) @@ -28,92 +28,19 @@ 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_json_api_v2(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_json_api_v2(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_json_api_v2() - 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="v2", + auth_token=self.__api_v2_token) def update(self) -> None: - log.debug("Variante: " + str(self.__device_variant)) - if self.__device_variant == 3: - state = self.__update_json_api_v2() - elif self.__device_variant in [0, 1, 2]: - log.debug("Die ausgewählte API bietet keine Verbrauchsdaten!") - else: - raise ValueError("Unbekannte API: " + 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 d9cc18b987..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 diff --git a/packages/modules/devices/sonnen/sonnenbatterie/inverter.py b/packages/modules/devices/sonnen/sonnenbatterie/inverter.py index 36cfaacb57..f0e81d6d91 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 +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_json_api(self, api_version: str = "v1"): - return req.get_http_session().get( - "http://" + self.__device_address + "/api/" + api_version + "/status", - timeout=5, - headers={"Auth-Token": self.__api_v2_token} if api_version == "v2" else None - ).json() - - def __update_json_api(self, api_version: 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_json_api(api_version=api_version) - 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_rest_api_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_rest_api_2(self) -> InverterState: - # Auslesen einer Sonnenbatterie Eco 6 über die integrierte REST-API des Batteriesystems - pv_power = -int(float(self.__read_rest_api_2_element(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="v2" if self.__device_variant == 3 else "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_json_api(api_version="v1") - elif self.__device_variant == 2: - state = self.__update_rest_api_2() - elif self.__device_variant == 3: - state = self.__update_json_api(api_version="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) From b5983252dd68a9dd40511253ac58973a71cf9345 Mon Sep 17 00:00:00 2001 From: Lutz Bender Date: Wed, 16 Apr 2025 13:36:05 +0200 Subject: [PATCH 3/3] refactor api --- .../devices/sonnen/sonnenbatterie/api.py | 426 ++++++++---------- .../devices/sonnen/sonnenbatterie/bat.py | 4 +- .../devices/sonnen/sonnenbatterie/counter.py | 4 +- .../sonnenbatterie/counter_consumption.py | 4 +- .../devices/sonnen/sonnenbatterie/inverter.py | 4 +- 5 files changed, 197 insertions(+), 245 deletions(-) diff --git a/packages/modules/devices/sonnen/sonnenbatterie/api.py b/packages/modules/devices/sonnen/sonnenbatterie/api.py index 4343c659a3..a774953267 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/api.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/api.py @@ -1,5 +1,6 @@ #!/usr/bin/env python3 -from typing import Dict, Optional +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 @@ -17,7 +18,7 @@ def power_limit_controllable(self) -> bool: """ return False - def read(self, device_endpoint: str = 'battery') -> dict: + def __read(self, device_endpoint: str = 'battery') -> dict: """ Reads data from the Sonnenbatterie REST API. Args: @@ -35,7 +36,7 @@ def update_battery(self, sim_counter: SimCounter) -> BatState: Returns: BatState: The updated battery state. """ - battery_state = self.read(device_endpoint="battery") + 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"]) @@ -59,7 +60,7 @@ def power_limit_controllable(self) -> bool: """ return False - def read_element(self, device: str, element: str) -> str: + def __read_element(self, device: str, element: str) -> str: """ Reads a specific element from the Sonnenbatterie REST API v2. Args: @@ -80,7 +81,7 @@ def update_inverter(self, sim_counter: SimCounter) -> InverterState: Returns: InverterState: The updated inverter state. """ - pv_power = -int(float(self.read_element(device="battery", element="M03"))) + 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) @@ -91,8 +92,8 @@ def update_grid_counter(self, sim_counter: SimCounter) -> CounterState: 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_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, @@ -105,9 +106,9 @@ def update_battery(self, sim_counter: SimCounter) -> BatState: 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_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, @@ -116,264 +117,149 @@ def update_battery(self, sim_counter: SimCounter) -> BatState: exported=exported) +class JsonApiVersion(Enum): + V1 = "v1" + V2 = "v2" + + class JsonApi(): - def __init__(self, host: str, api_version: str = "v1", auth_token: Optional[str] = None) -> None: + 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 == "v2" and self.auth_token is None: + 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 == "v2" else {} + self.headers = {"auth-token": auth_token} if api_version == JsonApiVersion.V2 else {} - 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 == "v2" and self.auth_token is not None - - def read(self, endpoint: str = "status") -> Dict: + 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}/{endpoint}", + f"http://{self.host}/api/{self.api_version.value}/{endpoint}", timeout=5, headers=self.headers ).json() - def update_inverter(self, sim_counter: SimCounter) -> InverterState: + def __read_status(self) -> StatusDict: """ - Updates the inverter state by reading data from the JSON API. + Reads the status data from the JSON API. Returns: - InverterState: The updated inverter state. - 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(endpoint="status") - pv_power = -inverter_state["Production_W"] - _, exported = sim_counter.sim_count(pv_power) - return InverterState(exported=exported, - power=pv_power) - - def update_grid_counter(self, sim_counter: SimCounter) -> CounterState: + StatusDict: The status data as a dictionary. """ - Updates the grid counter state by reading data from the JSON API. - Returns: - CounterState: The updated grid counter state. - 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(endpoint="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) + return self.__read(endpoint="status") - def update_consumption_counter(self) -> CounterState: + def __read_power_meter(self, direction: Optional[PowerMeterDirection] = None) -> List[ChannelDict]: """ - Updates the consumption counter state by reading data from the JSON API. + 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: - CounterState: The updated consumption counter state. - 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 - } - ] - """ - result = self.read(endpoint="powermeter") - for channel in result: - if channel["direction"] == "consumption": - return CounterState(power=channel["w_total"], - powers=[channel[f"w_l{phase}"] for phase in range(1, 4)], - currents=[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_imported"], - exported=channel["kwh_exported"]) - raise ValueError("No consumption data found in the response.") + 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 update_battery(self, sim_counter: SimCounter) -> BatState: + def __counter_state_from_channel(self, channel: ChannelDict, inverted: bool = False) -> CounterState: """ - Updates the battery state by reading data from the JSON API. + Converts a channel dictionary to a CounterState object. + Args: + channel (ChannelDict): The channel data as a dictionary. Returns: - InverterState: The updated battery state. - 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 - } - """ - battery_state = self.read(endpoint="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) + 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 != "v2": + 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") + return self.__read(endpoint="configurations") - def set_configurations(self, configuration: Dict) -> None: - if self.api_version != "v2": + 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 != "v2": + 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: @@ -383,16 +269,82 @@ def update_set_point(self, power_limit: int) -> None: 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"] + 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"}) + 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) + 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 e15510a474..0705216c59 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/bat.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/bat.py @@ -8,7 +8,7 @@ from modules.common.simcount import SimCounter from modules.common.store import get_bat_value_store -from modules.devices.sonnen.sonnenbatterie.api import JsonApi, RestApi1, RestApi2 +from modules.devices.sonnen.sonnenbatterie.api import JsonApi, JsonApiVersion, RestApi1, RestApi2 from modules.devices.sonnen.sonnenbatterie.config import SonnenbatterieBatSetup @@ -43,7 +43,7 @@ def initialize(self) -> None: self.api = RestApi2(host=self.__device_address) else: self.api = JsonApi(host=self.__device_address, - api_version="v2" if self.__device_variant == 3 else "v1", + 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: diff --git a/packages/modules/devices/sonnen/sonnenbatterie/counter.py b/packages/modules/devices/sonnen/sonnenbatterie/counter.py index c92af963ba..ab4c154c52 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/counter.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/counter.py @@ -8,7 +8,7 @@ from modules.common.simcount import SimCounter from modules.common.store import get_counter_value_store -from modules.devices.sonnen.sonnenbatterie.api import JsonApi, RestApi2 +from modules.devices.sonnen.sonnenbatterie.api import JsonApi, RestApi2, JsonApiVersion from modules.devices.sonnen.sonnenbatterie.config import SonnenbatterieCounterSetup log = logging.getLogger(__name__) @@ -42,7 +42,7 @@ def initialize(self) -> None: self.api = RestApi2(host=self.__device_address) else: self.api = JsonApi(host=self.__device_address, - api_version="v2" if self.__device_variant == 3 else "v1", + 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: diff --git a/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py b/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py index 50b8e76875..36f0de4bcf 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/counter_consumption.py @@ -7,7 +7,7 @@ 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 +from modules.devices.sonnen.sonnenbatterie.api import JsonApi, JsonApiVersion from modules.devices.sonnen.sonnenbatterie.config import SonnenbatterieConsumptionCounterSetup log = logging.getLogger(__name__) @@ -36,7 +36,7 @@ def initialize(self) -> None: self.store = get_counter_value_store(self.component_config.id) self.fault_state = FaultState(ComponentInfo.from_component_config(self.component_config)) self.api = JsonApi(host=self.__device_address, - api_version="v2", + api_version=JsonApiVersion.V2, auth_token=self.__api_v2_token) def update(self) -> None: diff --git a/packages/modules/devices/sonnen/sonnenbatterie/inverter.py b/packages/modules/devices/sonnen/sonnenbatterie/inverter.py index f0e81d6d91..446420297e 100644 --- a/packages/modules/devices/sonnen/sonnenbatterie/inverter.py +++ b/packages/modules/devices/sonnen/sonnenbatterie/inverter.py @@ -8,7 +8,7 @@ from modules.common.simcount import SimCounter from modules.common.store import get_inverter_value_store -from modules.devices.sonnen.sonnenbatterie.api import JsonApi, RestApi2 +from modules.devices.sonnen.sonnenbatterie.api import JsonApi, RestApi2, JsonApiVersion from modules.devices.sonnen.sonnenbatterie.config import SonnenbatterieInverterSetup log = logging.getLogger(__name__) @@ -42,7 +42,7 @@ def initialize(self) -> None: self.api = RestApi2(host=self.__device_address) else: self.api = JsonApi(host=self.__device_address, - api_version="v2" if self.__device_variant == 3 else "v1", + 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: