97 lines
4.9 KiB
Python
97 lines
4.9 KiB
Python
import logging
|
|
import aiohttp
|
|
from homeassistant.config_entries import ConfigEntry
|
|
_LOGGER = logging.getLogger(__name__)
|
|
API_URL = "https://api.sunlitsolar.de/rest"
|
|
isMock = False # Set to True for testing with mock data
|
|
|
|
class SunlitResponseCode:
|
|
"""Enum for Sunlit Solar API response codes."""
|
|
SUCCESS = 0
|
|
# ERROR = 1
|
|
INVALID_CREDENTIALS = 100003
|
|
# USER_NOT_FOUND = 3
|
|
# SERVER_ERROR = 4
|
|
|
|
def get_message_from_response(response):
|
|
"""Extracts the message from the Sunlit Solar API response."""
|
|
if 'message' in response:
|
|
for key, value in response["message"].items():
|
|
return value
|
|
return response['message']
|
|
else:
|
|
return "Unbekannter Fehler"
|
|
|
|
async def async_login(username, password):
|
|
"""Asynchronous login to the Sunlit Solar API."""
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(API_URL+"/user/login", json={'account': username, 'password': password, 'isMock': False}) as response:
|
|
response.raise_for_status()
|
|
return await response.json()
|
|
except aiohttp.ClientError as e:
|
|
_LOGGER.error("Fehler beim Abrufen der Daten: %s", e)
|
|
raise
|
|
|
|
async def async_get_device_list(entry: ConfigEntry, session: aiohttp.ClientSession):
|
|
try:
|
|
token = entry.data["access_token"]
|
|
familys = []
|
|
# get family-id -> get devices from family -> get device statistics from device id
|
|
async with session.post(API_URL+"/family/list", headers={"Authorization": f"Bearer {token}"}, json={'isMock': isMock}) as response:
|
|
response.raise_for_status()
|
|
resp_familys = await response.json()
|
|
if resp_familys['code'] != SunlitResponseCode.SUCCESS:
|
|
_LOGGER.error("Fehler beim Abrufen der Familien: %s", get_message_from_response(resp_familys))
|
|
else:
|
|
familys = resp_familys['content']
|
|
if familys:
|
|
# Get devices for each family
|
|
for f_key,family in enumerate(familys):
|
|
print(family['id'])
|
|
async with session.post(API_URL+"/v1.2/device/list", headers={"Authorization": f"Bearer {token}"}, json={
|
|
"familyId": family['id'],
|
|
"deviceType": "ALL",
|
|
"isMock": isMock,
|
|
"size": 20,
|
|
"page": 0
|
|
}) as response:
|
|
response.raise_for_status()
|
|
resp_devices = await response.json()
|
|
if resp_devices['code'] != SunlitResponseCode.SUCCESS:
|
|
_LOGGER.error("Fehler beim Abrufen der Geräte: %s", get_message_from_response(resp_devices))
|
|
else:
|
|
# todo: implement pagination
|
|
familys[f_key]['devices'] = resp_devices['content']['content']
|
|
if 'spaces' not in familys[f_key]:
|
|
familys[f_key]['spaces'] = {}
|
|
if familys[f_key]['devices'] and len(familys[f_key]['devices']) > 0:
|
|
# Get statistics for each device
|
|
for d_key,device in enumerate(family['devices']):
|
|
async with session.post(API_URL+"/v1.1/statistics/static/device", headers={"Authorization": f"Bearer {token}"}, json={
|
|
"deviceId": device['deviceId'],
|
|
"isMock": isMock,
|
|
}) as response:
|
|
response.raise_for_status()
|
|
resp_statistics = await response.json()
|
|
if resp_statistics['code'] != SunlitResponseCode.SUCCESS:
|
|
_LOGGER.error("Fehler beim Abrufen der Statistiken: %s", get_message_from_response(resp_statistics))
|
|
else:
|
|
familys[f_key]['devices'][d_key]['statistics'] = resp_statistics['content']
|
|
|
|
if 'spaceId' in device and device['spaceId'] not in familys[f_key]['spaces']:
|
|
async with session.post(API_URL+"/v1.5/space/index", headers={"Authorization": f"Bearer {token}"}, json={
|
|
"spaceId": device['spaceId'],
|
|
"isMock": isMock,
|
|
}) as response:
|
|
response.raise_for_status()
|
|
resp_space = await response.json()
|
|
if resp_space['code'] != SunlitResponseCode.SUCCESS:
|
|
_LOGGER.error("Fehler beim Abrufen des Raums: %s", get_message_from_response(resp_space))
|
|
else:
|
|
familys[f_key]['spaces'][device['spaceId']] = resp_space['content']
|
|
|
|
return familys
|
|
except aiohttp.ClientError as e:
|
|
_LOGGER.error("Fehler beim Abrufen der Daten: %s", e)
|
|
raise |