Skip to content

Commit

Permalink
- none working 'mein-senec.de' Portal addon (v01)
Browse files Browse the repository at this point in the history
  • Loading branch information
marq24 committed Sep 2, 2023
1 parent 0af96f6 commit fef0990
Showing 1 changed file with 168 additions and 2 deletions.
170 changes: 168 additions & 2 deletions custom_components/senec/pysenec_ha/__init__.py
View file
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@

import aiohttp
import logging
import xmltodict
Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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()

0 comments on commit fef0990

Please sign in to comment.