a POC with aiohttp using senec app api

senec_app_api_aiohttp.py

'''
Logs in to Senec app api and reads all specified information from SENEC APP API
'''

import json
import logging
import aiohttp
import asyncio

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

async def main():
    trace_config = aiohttp.TraceConfig()
    trace_config.on_request_redirect.append(on_request_redirect)
    trace_config.on_request_end.append(on_request_end)
    #async with aiohttp.ClientSession(trace_configs=[trace_config]) as session:
    async with aiohttp.ClientSession() as session:
        w = SenecWebGrabber(session)
        await w.update()

async def on_request_redirect(session, trace_config_ctx, params):
    logger.info('--- on_request_redirect ---')
    #logger.debug(str(params.url) + " [" + str(params.response.status)+"] > "+str(params.response.headers["Location"]))
    #_LOGGER.debug(params)

async def on_request_end(session, trace_config_ctx, params):
    logger.info('--- on END ---')
    #logger.debug(str(params.url))
    #logger.debug(params)

class SenecWebGrabber:
    def __init__(self, websession) -> None:

        #SENEC API
        self._SENEC_USERNAME = "" # your senec login username
        self._SENEC_PASSWORD = "" # your senec password
        self._SENEC_AUTH_URL = "https://app-gateway-prod.senecops.com/v1/senec/login"
        self._SENEC_ANLAGEN_URL = "https://app-gateway-prod.senecops.com/v1/senec/anlagen"

        self._device_ids = []

        #WEBDATA STORAGE
        self._energy_entities = {}
        self._power_entities = {}
        self._battery_entities = {}
        self._isAuthenticated = False

        #WEBSESSION
        self._session = websession

    async def authenticate(self):
        logger.debug("***** authenticate(self) ********")
        auth_payload = {
            "username" : self._SENEC_USERNAME,
            "password" : self._SENEC_PASSWORD
        }
        async with self._session.post(self._SENEC_AUTH_URL, json=auth_payload) as r:
            if r.status == 200:
                logger.info("Login successful")
                self._isAuthenticated=True
                data = await r.json()
                self._session.headers["Authorization"] = data["token"]
                await self.get_devices()

                await self.update()
            else:
                logger.info("Login failed with Code " + str(r.status))

    async def get_devices(self):
        logger.debug("***** get_devices(self) ********")
        async with self._session.get(self._SENEC_ANLAGEN_URL) as r:
            if r.status == 200:
                for device in await r.json():
                    self._device_ids.append(device["id"])
            else:
                logger.info("Login to get devices with Code " + str(r.status))

    async def update(self):
        logger.debug("***** update(self) ********")

        if self._isAuthenticated:
            await self.update_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()

    async def update_stats(self):
        logger.debug("***** update_stats(self) ********")
        # just use first device id for now
        device_id = self._device_ids[0]
        async with self._session.get(f"{self._SENEC_ANLAGEN_URL}/{device_id}/dashboard") as r:
            if r.status==200:
                data = await r.json()
                for key, val in data["heute"].items():
                    if val["einheit"] == "Wh":
                        self._energy_entities[key] = val["wert"]
                    elif val["einheit"] == "%":
                        self._battery_entities[key] = val["wert"]
            else:
                self._isAuthenticated=False
                await self.update()

asyncio.run(main())