From fef099094f74a844eae93bae84975d0bf8675a8d Mon Sep 17 00:00:00 2001 From: marq24 Date: Sat, 2 Sep 2023 08:10:10 +0200 Subject: [PATCH] - none working 'mein-senec.de' Portal addon (v01) --- .../senec/pysenec_ha/__init__.py | 170 +++++++++++++++++- 1 file changed, 168 insertions(+), 2 deletions(-) diff --git a/custom_components/senec/pysenec_ha/__init__.py b/custom_components/senec/pysenec_ha/__init__.py index 50b6964..41986a3 100644 --- a/custom_components/senec/pysenec_ha/__init__.py +++ b/custom_components/senec/pysenec_ha/__init__.py @@ -1,3 +1,4 @@ + import aiohttp import logging import xmltodict @@ -852,12 +853,12 @@ def wallbox_energy(self) -> float: def fan_inv_lv(self) -> bool: if hasattr(self, '_raw') and "FAN_SPEED" in self._raw and "INV_LV" in self._raw["FAN_SPEED"]: return self._raw["FAN_SPEED"]["INV_LV"] + @property def fan_inv_hv(self) -> bool: if hasattr(self, '_raw') and "FAN_SPEED" in self._raw and "INV_HV" in self._raw["FAN_SPEED"]: return self._raw["FAN_SPEED"]["INV_HV"] - async def update(self): await self.read_senec_v31() @@ -929,7 +930,7 @@ async def read_senec_v31(self): "L3_CHARGING_CURRENT": "", "EV_CONNECTED": "" }, - "FAN_SPEED":{}, + "FAN_SPEED": {}, } async with self.websession.post(self.url, json=form, ssl=False) as res: @@ -1329,3 +1330,168 @@ def ownconsumedpower(self) -> float: def derating(self) -> float: if (hasattr(self, '_derating')): return self._derating + + +class MeinSenecWebPortal: + + def __init__(self, user, pwd, websession): + self.websession: aiohttp.websession = websession + + # SENEC API + self._SENEC_USERNAME = user + self._SENEC_PASSWORD = pwd + + # https://documenter.getpostman.com/view/10329335/UVCB9ihW#17e2c6c6-fe5e-4ca9-bc2f-dca997adaf90 + self._SENEC_CLASSIC_AUTH_URL = "https://app-gateway-prod.senecops.com/v1/senec/login" + self._SENEC_CLASSIC_API_OVERVIEW_URL = "https://app-gateway-prod.senecops.com/v1/senec/anlagen" + + + self._SENEC_AUTH_URL = "https://mein-senec.de/auth/login" + #"https://mein-senec.de/endkunde/oauth2/authorization/endkunde-portal" + self._SENEC_API_OVERVIEW_URL = "https://mein-senec.de/endkunde/api/status/getstatusoverview.php?anlageNummer=0" + self._SENEC_API_URL_START = "https://mein-senec.de/endkunde/api/status/getstatus.php?type=" + self._SENEC_API_URL_END = "&period=all&anlageNummer=0" + + # can be used in all api calls, names come from senec website + self._API_KEYS = [ + "accuimport", # what comes OUT OF the accu + "accuexport", # what goes INTO the accu + "gridimport", # what comes OUT OF the grid + "gridexport", # what goes INTO the grid + "powergenerated", # power produced + "consumption" # power used + ] + + # can only be used in some api calls, names come from senec website + self._API_KEYS_EXTRA = [ + "acculevel" # accu level + ] + + # WEBDATA STORAGE + self._energy_entities = {} + self._power_entities = {} + self._battery_entities = {} + self._isAuthenticated = False + + async def authenticateClassic(self, doUpdate:bool): + auth_payload = { + "username": self._SENEC_USERNAME, + "password": self._SENEC_PASSWORD + } + async with self.websession.post(self._SENEC_CLASSIC_AUTH_URL, json=auth_payload) as res: + res.raise_for_status() + if res.status == 200: + r_json = await res.json() + if "token" in r_json: + self._token = r_json["token"] + self._isAuthenticated = True + _LOGGER.info("Login successful") + if doUpdate: + self.updateClassic() + else: + _LOGGER.warning("Login failed with Code " + str(res.status)) + + async def updateClassic(self): + _LOGGER.debug("***** updateClassic(self) ********") + if self._isAuthenticated: + await self.getSystemOverviewClassic() + else: + await self.authenticateClassic(True) + + async def getSystemOverviewClassic(self): + headers = {"Authorization": self._token} + async with self.websession.get(self._SENEC_CLASSIC_API_OVERVIEW_URL, headers=headers) as res: + res.raise_for_status() + if res.status == 200: + r_json = await res.json() + print(str(r_json)) + else: + self._isAuthenticated = False + await self.update() + + async def _debugLogout(self): + async with self.websession.get("https://mein-senec.de/endkunde/logout") as res: + res.raise_for_status() + print(str(res)) + async def authenticate(self, doUpdate:bool): + auth_payload = { + "username": self._SENEC_USERNAME, + "password": self._SENEC_PASSWORD + } + async with self.websession.post(self._SENEC_AUTH_URL, data=auth_payload) as res: + res.raise_for_status() + if res.status == 200: + r_json = await res.text() + print(r_json) + self._isAuthenticated = True + _LOGGER.info("Login successful") + if doUpdate: + self.update() + else: + _LOGGER.warning("Login failed with Code " + str(res.status)) + + async def update(self): + _LOGGER.debug("***** update(self) ********") + + if self._isAuthenticated: + await self.update_now_kW_stats() + await self.update_full_kWh_stats() + + _LOGGER.debug("Results:") + _LOGGER.debug("********* energy_entities ***************") + for key in self._energy_entities: + _LOGGER.debug(str(key) + ": " + str(self._energy_entities[key])) + + _LOGGER.debug("********* power_entities *****************") + for key in self._power_entities: + _LOGGER.debug(str(key) + ": " + str(self._power_entities[key])) + + _LOGGER.debug("********* battery_entities *****************") + for key in self._battery_entities: + _LOGGER.debug(str(key) + ": " + str(self._battery_entities[key])) + else: + await self.authenticate(True) + + async def update_now_kW_stats(self): + _LOGGER.debug("***** update_now_kW_stats(self) ********") + + # grab NOW and TODAY stats + async with self.websession.get(self._SENEC_API_OVERVIEW_URL) as res: + res.raise_for_status() + if res.status == 200: + r_json = await res.json() + print(str(r_json)) + for key in (self._API_KEYS + self._API_KEYS_EXTRA): + if (key != "acculevel"): + value_now = r_json[key]["now"] + entity_now_name = str(key + "_now") + self._power_entities[entity_now_name] = value_now + + value_today = r_json[key]["today"] + entity_today_name = str(key + "_today") + self._energy_entities[entity_today_name] = value_today + else: + value_now = r_json[key]["now"] + entity_now_name = str(key + "_now") + self._battery_entities[entity_now_name] = value_now + # value_today = r_json[key]["today"] + # entity_today_name = str(key + "_today") + # self._battery_entities[entity_today_name]=value_today + else: + self._isAuthenticated = False + await self.update() + + async def update_full_kWh_stats(self): + # grab TOTAL stats + for key in self._API_KEYS: + api_url = self._SENEC_API_URL_START + key + self._SENEC_API_URL_END + async with self.websession.get(api_url) as res: + res.raise_for_status() + if res.status == 200: + r_json = await res.json() + value = r_json["fullkwh"] + entity_name = str(key + "_total") + self._energy_entities[entity_name] = value + else: + self._isAuthenticated = False + await self.update()