From aba11fc416e6094141875e79157033b639f1e2a1 Mon Sep 17 00:00:00 2001 From: maxhov <14804474+maxhov@users.noreply.github.com> Date: Mon, 7 Jul 2025 12:46:02 +0200 Subject: [PATCH] Make SAJ eSolar Air work with the new elekeeper portal --- custom_components/saj_esolar_air/__init__.py | 5 +- .../saj_esolar_air/config_flow.py | 30 +- custom_components/saj_esolar_air/esolar.py | 530 ++++++++++-------- custom_components/saj_esolar_air/sensor.py | 240 ++++---- .../saj_esolar_air/test_esolar.py | 36 ++ 5 files changed, 463 insertions(+), 378 deletions(-) create mode 100644 custom_components/saj_esolar_air/test_esolar.py diff --git a/custom_components/saj_esolar_air/__init__.py b/custom_components/saj_esolar_air/__init__.py index 6275eae..af227e1 100644 --- a/custom_components/saj_esolar_air/__init__.py +++ b/custom_components/saj_esolar_air/__init__.py @@ -523,7 +523,8 @@ def get_data( if "error" in plant_info: raise UnknownError(plant_info["error"]) - if plant_info.get("status") != "success": + if len(plant_info) == 0: _LOGGER.exception("Unexpected response: %s", plant_info) raise UnknownError - return cast(ESolarResponse, plant_info) + + return plant_info diff --git a/custom_components/saj_esolar_air/config_flow.py b/custom_components/saj_esolar_air/config_flow.py index f1814b4..5c7140e 100644 --- a/custom_components/saj_esolar_air/config_flow.py +++ b/custom_components/saj_esolar_air/config_flow.py @@ -6,11 +6,9 @@ import requests import voluptuous as vol - from homeassistant import config_entries from homeassistant.const import CONF_REGION, CONF_PASSWORD, CONF_USERNAME from homeassistant.core import HomeAssistant, callback -from homeassistant.data_entry_flow import FlowResult from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers import config_validation as cv @@ -20,7 +18,7 @@ CONF_PV_GRID_DATA, DOMAIN, ) -from .esolar import esolar_web_autenticate, web_get_plant +from .esolar import esolar_web_authenticate, web_get_plant_list CONF_TITLE = "SAJ eSolar" @@ -43,10 +41,10 @@ def __init__(self) -> None: self.plant_list: dict[str, Any] = {} def auth_and_get_solar_plants(self, region: str, username: str, password: str) -> bool: - """Download and list availablse inverters.""" + """Download and list available inverters.""" try: - session = esolar_web_autenticate(region, username, password) - self.plant_list = web_get_plant(region, session).get("plantList") + session = esolar_web_authenticate(region, username, password) + self.plant_list = web_get_plant_list(region, session) except requests.exceptions.HTTPError: _LOGGER.error("Login: HTTPError") return False @@ -82,13 +80,13 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): VERSION = 1 def __init__(self): - """Set up the the config flow.""" + """Set up the config flow.""" self.sites = {} self.data = {} async def async_step_user( self, user_input: dict[str, Any] | None = None - ) -> FlowResult: + ): """Handle the initial step. Username and password.""" if user_input is None: return self.async_show_form( @@ -107,7 +105,7 @@ async def async_step_user( _LOGGER.exception("Unexpected exception") errors["base"] = "unknown" else: - self.sites = [site["plantname"] for site in info["plant_list"]] + self.sites = [site["plantName"] for site in info["plant_list"]] if len(self.sites) == 1: return self.async_create_entry( title=CONF_TITLE, @@ -129,7 +127,7 @@ async def async_step_user( async def async_step_sites( self, user_input: dict[str, Any] | None = None - ) -> FlowResult: + ): """Handle the second step. Select which sites to use.""" errors = {} @@ -159,21 +157,17 @@ async def async_step_sites( @callback def async_get_options_flow( config_entry: config_entries.ConfigEntry, - ) -> config_entries.OptionsFlow: + ) -> OptionsFlowHandler: """Create the options flow.""" - return OptionsFlowHandler(config_entry) + return OptionsFlowHandler() class OptionsFlowHandler(config_entries.OptionsFlow): """Handle a options flow for eSolar.""" - def __init__(self, config_entry: config_entries.ConfigEntry) -> None: - """Initialize options flow.""" - self.config_entry = config_entry - async def async_step_init( self, user_input: dict[str, Any] | None = None - ) -> FlowResult: + ): """Manage the options.""" if user_input is not None: user_input.update( @@ -196,7 +190,7 @@ async def async_step_init( vol.Required( CONF_PV_GRID_DATA, default=self.config_entry.options.get(CONF_PV_GRID_DATA), - ): bool, + ): bool } ), ) diff --git a/custom_components/saj_esolar_air/esolar.py b/custom_components/saj_esolar_air/esolar.py index 552f6a8..da917eb 100644 --- a/custom_components/saj_esolar_air/esolar.py +++ b/custom_components/saj_esolar_air/esolar.py @@ -1,10 +1,15 @@ """ESolar Cloud Platform data fetchers.""" -import calendar import datetime -from datetime import timedelta +import hashlib import logging +import random +import string import requests +from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import padding +from cryptography.hazmat.primitives.ciphers import algorithms, Cipher, modes +from requests import HTTPError, Timeout, RequestException _LOGGER = logging.getLogger(__name__) @@ -19,284 +24,331 @@ ) -def base_url(region): - if (region == "eu"): - return "https://fopapp.saj-electric.com/sajAppApi/api" - elif (region == "in"): - return "https://intopapp.saj-electric.com/sajAppApi/api" - else: - raise ValueError("Region not set. Please run Configure again") - def base_url_web(region): - if (region == "eu"): - return "https://fop.saj-electric.com/saj" - elif (region == "in"): - return "https://intop.saj-electric.com/saj" + if region == "eu": + return "https://eop.saj-electric.com/dev-api/api/v1" + elif region == "in": + return "https://iop.saj-electric.com/dev-api/api/v1" else: raise ValueError("Region not set. Please run Configure again") -def add_months(sourcedate, months): - """SAJ eSolar Helper Function - Adds a months to input.""" - month = sourcedate.month - 1 + months - year = sourcedate.year + month // 12 - month = month % 12 + 1 - day = min(sourcedate.day, calendar.monthrange(year, month)[1]) - return datetime.date(year, month, day) - - -def add_years(source_date, years): - """SAJ eSolar Helper Function - Adds a years to input.""" - try: - return source_date.replace(year=source_date.year + years) - except ValueError: - return source_date + ( - datetime.date(source_date.year + years, 1, 1) - - datetime.date(source_date.year, 1, 1) - ) - - -def get_esolar_data(region, username, password, plant_list=None, use_pv_grid_attributes=True): +def get_esolar_data(region, username, password, plant_list=None, + use_pv_grid_attributes=True): """SAJ eSolar Data Update.""" if BASIC_TEST: return get_esolar_data_static_h1_r5( region, username, password, plant_list, use_pv_grid_attributes ) - try: - plant_info = None - session = esolar_web_autenticate(region, username, password) - plant_info = web_get_plant(region, session, plant_list) - web_get_plant_details(region, session, plant_info) - web_get_plant_detailed_chart(region, session, plant_info) - web_get_device_page_list(region, session, plant_info, use_pv_grid_attributes) - - except requests.exceptions.HTTPError as errh: - raise requests.exceptions.HTTPError(errh) - except requests.exceptions.ConnectionError as errc: - raise requests.exceptions.ConnectionError(errc) - except requests.exceptions.Timeout as errt: - raise requests.exceptions.Timeout(errt) - except requests.exceptions.RequestException as errr: - raise requests.exceptions.RequestException(errr) - except ValueError as errv: - raise ValueError(errv) from errv + token = esolar_web_authenticate(region, username, password) + plant_info = web_get_plant_list(region, token, plant_list) + web_get_plant_info(region, token, plant_info) + web_get_plant_grid_overview_info(region, token, plant_info) + # TODO: Needs to be determined if this is still relevant. Not sure what it needs to do + # web_get_device_page_list(region, token, plant_info, + # use_pv_grid_attributes) return plant_info -def esolar_web_autenticate(region, username, password): +def encrypt_password(password, + encryption_key="ec1840a7c53cf0709eb784be480379b6"): + """Encrypt the password using AES-128-CBC. The key is hardcoded and can be found in the web portal.""" + padder = padding.PKCS7(algorithms.AES.block_size).padder() + padded_data = padder.update(password.encode()) + padder.finalize() + + # Create cipher + cipher = Cipher( + algorithms.AES(bytes.fromhex(encryption_key)), + modes.ECB(), + backend=default_backend() + ) + encryptor = cipher.encryptor() + return (encryptor.update(padded_data) + encryptor.finalize()).hex() + + +def generate_signature(params: dict[str, int | str], + signing_key="ktoKRLgQPjvNyUZO8lVc9kU1Bsip6XIe"): + """Generate the signature for the API request. The signing key is hardcoded and can be found in the web portal.""" + output = '&'.join(f"{k}={v}" for k, v in sorted(params.items())) + output += f"&key={signing_key}" + md5 = hashlib.md5(output.encode()).hexdigest() + return hashlib.sha1(md5.encode()).hexdigest().upper() + + +def generate_random(length=32): + characters = string.ascii_letters + string.digits # combines uppercase, lowercase and numbers + return ''.join(random.choice(characters) for _ in range(length)) + + +def esolar_web_authenticate(region, username, password): """Authenticate the user to the SAJ's WEB Portal.""" if BASIC_TEST: return True - try: - session = requests.Session() - response = session.post( - base_url_web(region) + "/login", - data={ - "lang": "en", - "username": username, - "password": password, - "rememberMe": "true", - }, - timeout=WEB_TIMEOUT, - ) + session = requests.Session() + lang = "en" + project_name = "elekeeper" + client_id = "esolar-monitor-admin" + client_date = "2025-07-06" + timestamp = int(datetime.datetime.now().timestamp() * 1000) + rnd = generate_random() + + response = session.post( + base_url_web(region) + "/sys/login", + headers={"Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.3"}, + data={ + "lang": lang, + "username": username, + "password": encrypt_password(password), + "rememberMe": "false", + "loginType": "1", + "appProjectName": project_name, + "random": rnd, + "clientDate": client_date, + "timeStamp": timestamp, + "clientId": client_id, + "signParams": "appProjectName,clientDate,lang,timeStamp,random,clientId", + "signature": generate_signature({"appProjectName": project_name, + "clientDate": client_date, + "clientId": client_id, + "lang": lang, + "random": rnd, + "timeStamp": timestamp}), + }, + timeout=WEB_TIMEOUT, + ) + try: response.raise_for_status() + return response.json()["data"]["token"] + except: + raise ValueError(response.content) - if response.status_code != 200: - raise ValueError(f"Login failed, returned {response.status_code}") - return session - - except requests.exceptions.HTTPError as errh: - raise requests.exceptions.HTTPError(errh) - except requests.exceptions.ConnectionError as errc: - raise requests.exceptions.ConnectionError(errc) - except requests.exceptions.Timeout as errt: - raise requests.exceptions.Timeout(errt) - except requests.exceptions.RequestException as errr: - raise requests.exceptions.RequestException(errr) - - -def web_get_plant(region, session, requested_plant_list=None): +def web_get_plant_list(region, token, requested_plant_list=None): """Retrieve the platUid from WEB Portal using web_authenticate.""" - if session is None: - raise ValueError("Missing session identifier trying to obain plants") + if token is None: + raise ValueError("Missing token trying to obtain plants") if BASIC_TEST: return web_get_plant_static_h1_r5() - try: - output_plant_list = [] - response = session.post( - base_url_web(region) + "/monitor/site/getUserPlantList", - data={ - "pageNo": "", - "pageSize": "", - "orderByIndex": "", - "officeId": "", - "clientDate": datetime.date.today().strftime("%Y-%m-%d"), - "runningState": "", - "selectInputType": "", - "plantName": "", - "deviceSn": "", - "type": "", - "countryCode": "", - "isRename": "", - "isTimeError": "", - "systemPowerLeast": "", - "systemPowerMost": "", - }, - timeout=WEB_TIMEOUT, - ) + output_plant_list = [] + session = requests.Session() + page_size = 100 + lang = "en" + project_name = "elekeeper" + client_id = "esolar-monitor-admin" + client_date = "2025-07-07" + timestamp = int(datetime.datetime.now().timestamp() * 1000) + rnd = generate_random() + sign_params = "pageSize,pageNo,searchOfficeIdArr,appProjectName,clientDate,lang,timeStamp,random,clientId" + + response = session.get( + base_url_web(region) + "/monitor/plant/getPlantList", + headers={"Authorization": "Bearer " + token, + "Accept": "application/json", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.3"}, + params={ + "pageSize": page_size, + "pageNo": 1, + "searchOfficeIdArr": 1, + "appProjectName": project_name, + "clientDate": client_date, + "lang": lang, + "timeStamp": timestamp, + "random": rnd, + "clientId": client_id, + "signParams": sign_params, + "signature": generate_signature({ + "pageSize": page_size, + "pageNo": 1, + "searchOfficeIdArr": 1, + "appProjectName": project_name, + "clientDate": client_date, + "clientId": client_id, + "lang": lang, + "random": rnd, + "timeStamp": timestamp}), + }, + timeout=WEB_TIMEOUT, + ) + try: response.raise_for_status() - plant_list = response.json() + response = response.json() + plant_list = response["data"]["list"] if requested_plant_list is not None: - for plant in plant_list["plantList"]: - if plant["plantname"] in requested_plant_list: + for plant in plant_list: + if plant["plantName"] in requested_plant_list: output_plant_list.append(plant) - return {"status": plant_list["status"], "plantList": output_plant_list} - + return output_plant_list return plant_list - - except requests.exceptions.HTTPError as errh: - raise requests.exceptions.HTTPError(errh) - except requests.exceptions.ConnectionError as errc: - raise requests.exceptions.ConnectionError(errc) - except requests.exceptions.Timeout as errt: - raise requests.exceptions.Timeout(errt) - except requests.exceptions.RequestException as errr: - raise requests.exceptions.RequestException(errr) + except: + raise ValueError(response.content) -def web_get_plant_details(region, session, plant_info): +def web_get_plant_info(region, token, plants): """Retrieve platUid from the WEB Portal using web_authenticate.""" - if session is None: - raise ValueError("Missing session identifier trying to obain plants") + if token is None: + raise ValueError("Missing token trying to obtain plant details") try: - device_list = [] - for plant in plant_info["plantList"]: - response = session.post( - base_url_web(region) + "/monitor/site/getPlantDetailInfo", - data={ - "plantuid": plant["plantuid"], - "clientDate": datetime.date.today().strftime("%Y-%m-%d"), + for plant in plants: + plant_uid = plant["plantUid"] + lang = "en" + project_name = "elekeeper" + client_id = "esolar-monitor-admin" + client_date = "2025-07-07" + timestamp = int(datetime.datetime.now().timestamp() * 1000) + rnd = generate_random() + sign_params = "plantUid,appProjectName,clientDate,lang,timeStamp,random,clientId" + + session = requests.Session() + response = session.get( + base_url_web(region) + "/monitor/plant/getOnePlantInfo", + headers={"Authorization": "Bearer " + token, + "Accept": "application/json", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.3"}, + params={ + "plantUid": plant_uid, + "appProjectName": project_name, + "clientId": client_id, + "clientDate": client_date, + "lang": lang, + "timeStamp": timestamp, + "random": rnd, + "signParams": sign_params, + "signature": generate_signature({ + "plantUid": plant_uid, + "appProjectName": project_name, + "clientDate": client_date, + "clientId": client_id, + "lang": lang, + "timeStamp": timestamp, + "random": rnd + }) }, timeout=WEB_TIMEOUT, ) - response.raise_for_status() - plant_detail = response.json() - plant.update(plant_detail) - for device in plant_detail["plantDetail"]["snList"]: - device_list.append(device) - - except requests.exceptions.HTTPError as errh: - raise requests.exceptions.HTTPError(errh) - except requests.exceptions.ConnectionError as errc: - raise requests.exceptions.ConnectionError(errc) - except requests.exceptions.Timeout as errt: - raise requests.exceptions.Timeout(errt) - except requests.exceptions.RequestException as errr: - raise requests.exceptions.RequestException(errr) - - -def web_get_plant_detailed_chart(region, session, plant_info): + try: + response.raise_for_status() + plant_detail = response.json()["data"] + plant.update(plant_detail) + except: + raise ValueError(response.content) + + except HTTPError as errh: + raise HTTPError(errh) + except ConnectionError as errc: + raise ConnectionError(errc) + except Timeout as errt: + raise Timeout(errt) + except RequestException as errr: + raise RequestException(errr) + + +def web_get_plant_grid_overview_info(region, token, plants): """Retrieve the kitList from the WEB Portal with web_authenticate.""" - if session is None: - raise ValueError("Missing session identifier trying to obain plants") - - try: - today = datetime.date.today() - previous_chart_day = today - timedelta(days=1) - next_chart_day = today + timedelta(days=1) - chart_day = today.strftime("%Y-%m-%d") - previous_chart_month = add_months(today, -1).strftime("%Y-%m") - next_chart_month = add_months(today, 1).strftime("%Y-%m") - chart_month = today.strftime("%Y-%m") - previous_chart_year = add_years(today, -1).strftime("%Y") - next_chart_year = add_years(today, 1).strftime("%Y") - chart_year = today.strftime("%Y") - epochmilliseconds = round( - int( - ( - datetime.datetime.utcnow() - datetime.datetime(1970, 1, 1) - ).total_seconds() - * 1000 + if token is None: + raise ValueError("Missing token trying to obtain plants") + + for plant in plants: + # bean = [] + peak_pow = [] + for inverter in plant["deviceSnList"]: + plant_uid = plant["plantUid"] + lang = "en" + project_name = "elekeeper" + client_id = "esolar-monitor-admin" + client_date = "2025-07-07" + refresh = int(datetime.datetime.now().timestamp() * 1000) + timestamp = int(datetime.datetime.now().timestamp() * 1000) + rnd = generate_random() + sign_params = "plantUid,refresh,appProjectName,clientDate,lang,timeStamp,random,clientId" + + session = requests.Session() + response = session.get( + base_url_web( + region) + "/monitor/home/getPlantGridOverviewInfo", + headers={"Authorization": "Bearer " + token, + "Accept": "application/json", + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.3"}, + params={ + "plantUid": plant_uid, + "appProjectName": project_name, + "clientId": client_id, + "clientDate": client_date, + "lang": lang, + "refresh": refresh, + "random": rnd, + "timeStamp": timestamp, + "signParams": sign_params, + "signature": generate_signature({ + "plantUid": plant_uid, + "appProjectName": project_name, + "clientDate": client_date, + "clientId": client_id, + "lang": lang, + "timeStamp": timestamp, + "refresh": refresh, + "random": rnd + }) + }, + timeout=WEB_TIMEOUT, ) - ) - client_date = datetime.date.today().strftime("%Y-%m-%d") - - for plant in plant_info["plantList"]: - # - # NOTE : This URL now takes a sinle inverter, but it should somehow take a list - # - # deviceSnArr={plant['plantDetail']['snList'][0] <<== Is correct if there is only one inverter in the system - # - bean = [] - peak_pow = [] - for inverter in plant["plantDetail"]["snList"]: - if plant["type"] == 3: - # Battery system - url = f"{base_url_web(region)}/monitor/site/getPlantDetailChart2?plantuid={plant['plantuid']}&chartDateType=1&energyType=0&clientDate={client_date}&deviceSnArr=&chartCountType=2&previousChartDay={previous_chart_day}&nextChartDay={next_chart_day}&chartDay={chart_day}&previousChartMonth={previous_chart_month}&nextChartMonth={next_chart_month}&chartMonth={chart_month}&previousChartYear={previous_chart_year}&nextChartYear={next_chart_year}&chartYear={chart_year}&elecDevicesn={inverter}&_={epochmilliseconds}" - else: - # Normal system - url = f"{base_url_web(region)}/monitor/site/getPlantDetailChart2?plantuid={plant['plantuid']}&chartDateType=1&energyType=0&clientDate={client_date}&deviceSnArr={inverter}&chartCountType=2&previousChartDay={previous_chart_day}&nextChartDay={next_chart_day}&chartDay={chart_day}&previousChartMonth={previous_chart_month}&nextChartMonth={next_chart_month}&chartMonth={chart_month}&previousChartYear={previous_chart_year}&nextChartYear={next_chart_year}&chartYear={chart_year}&elecDevicesn=&_={epochmilliseconds}" - - _LOGGER.debug("Fetching URL : %s", url) - response = session.post(url, timeout=WEB_TIMEOUT) + + try: response.raise_for_status() - plant_chart = response.json() + overview_info = response.json()["data"] + if VERBOSE_DEBUG: _LOGGER.debug( - "\n.../getPlantDetailChart2\n------------------------\n%s", - plant_chart, + "\n.../getPlantGridOverviewInfo\n------------------------\n%s", + overview_info, ) - if (plant_chart["type"]) == 0: - tmp = {} - tmp.update({"devicesn": inverter}) - tmp.update({"peakPower": plant_chart["peakPower"]}) - peak_pow.append(tmp) - plant.update({"peakList": peak_pow}) - # plant.update({"peakPower": plant_chart["peakPower"]}) - elif (plant_chart["type"]) == 1: - plant_chart["viewBean"].update({"devicesn": inverter}) - bean.append(plant_chart["viewBean"]) - plant.update({"beanList": bean}) - - except requests.exceptions.HTTPError as errh: - raise requests.exceptions.HTTPError(errh) - except requests.exceptions.ConnectionError as errc: - raise requests.exceptions.ConnectionError(errc) - except requests.exceptions.Timeout as errt: - raise requests.exceptions.Timeout(errt) - except requests.exceptions.RequestException as errr: - raise requests.exceptions.RequestException(errr) - - -def web_get_device_page_list(region, session, plant_info, use_pv_grid_attributes): - """Retrieve the platUid from the WEB Portal with web_authenticate.""" - if session is None: - raise ValueError("Missing session identifier trying to obain plants") + # if (overview_info["type"]) == 0: + peak_pow.append({ + "devicesn": inverter, + "peakPower": overview_info["peakPower"], + }) + plant.update({"peakList": peak_pow}) + plant.update({"overviewInfo": overview_info}) + # TODO: Not sure how to fix this and if this is (still) relevant + # elif (overview_info["type"]) == 1: + # overview_info["viewBean"].update({"devicesn": inverter}) + # bean.append(overview_info["viewBean"]) + # plant.update({"beanList": bean}) + except: + raise ValueError(response.content) + + +def web_get_device_page_list(region, token, plants, + use_pv_grid_attributes): + """Retrieve the plantUid from the WEB Portal with web_authenticate.""" + if token is None: + raise ValueError("Missing token trying to obtain plants") headers = { "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", } try: - for plant in plant_info["plantList"]: - _LOGGER.debug("Plant UID: %s", plant["plantuid"]) + for plant in plants: + _LOGGER.debug("Plant UID: %s", plant["plantUid"]) _LOGGER.debug("Plant Type: %s", plant["type"]) chart_month = datetime.date.today().strftime("%Y-%m") url = f"{base_url_web(region)}/cloudMonitor/device/findDevicePageList" - payload = f"officeId=1&pageNo=&pageSize=&orderName=1&orderType=2&plantuid={plant['plantuid']}&deviceStatus=&localDate={datetime.date.today().strftime('%Y-%m-%d')}&localMonth={chart_month}" + payload = f"officeId=1&pageNo=&pageSize=&orderName=1&orderType=2&plantuid={plant['plantUid']}&deviceStatus=&localDate={datetime.date.today().strftime('%Y-%m-%d')}&localMonth={chart_month}" _LOGGER.debug("Fetching URL : %s", url) _LOGGER.debug("Fetching Payload: %s", payload) + session = requests.Session() response = session.post( url, headers=headers, data=payload, timeout=WEB_TIMEOUT ) @@ -304,7 +356,8 @@ def web_get_device_page_list(region, session, plant_info, use_pv_grid_attributes device_list = response.json()["list"] if VERBOSE_DEBUG: _LOGGER.debug( - "\n.../findDevicePageList\n----------------------\n%s", device_list + "\n.../findDevicePageList\n----------------------\n%s", + device_list ) kit = [] @@ -323,17 +376,20 @@ def web_get_device_page_list(region, session, plant_info, use_pv_grid_attributes response.raise_for_status() find_rawdata_page_list = response.json() _LOGGER.debug( - "Result length : %s", len(find_rawdata_page_list["list"]) + "Result length : %s", + len(find_rawdata_page_list["list"]) ) if len(find_rawdata_page_list["list"]) > 0: device.update( - {"findRawdataPageList": find_rawdata_page_list["list"][0]} + {"findRawdataPageList": + find_rawdata_page_list["list"][0]} ) else: device.update({"findRawdataPageList": None}) - if VERBOSE_DEBUG and len(find_rawdata_page_list["list"]) > 0: + if VERBOSE_DEBUG and len( + find_rawdata_page_list["list"]) > 0: _LOGGER.debug( "\n.../findRawdataPageList\n-----------------------\n%s", find_rawdata_page_list["list"][0], @@ -342,15 +398,7 @@ def web_get_device_page_list(region, session, plant_info, use_pv_grid_attributes # Fetch battery for H1 system (UNTESTED CODE) if plant["type"] == 3: _LOGGER.debug("Fetching storage information") - epochmilliseconds = round( - int( - ( - datetime.datetime.utcnow() - - datetime.datetime(1970, 1, 1) - ).total_seconds() - * 1000 - ) - ) + epochmilliseconds = datetime.datetime.now().timestamp() * 1000 url = f"{base_url_web(region)}/monitor/site/getStoreOrAcDevicePowerInfo" payload = f"plantuid={plant['plantuid']}&devicesn={device['devicesn']}&_={epochmilliseconds}" _LOGGER.debug("Fetching URL : %s", url) @@ -371,11 +419,11 @@ def web_get_device_page_list(region, session, plant_info, use_pv_grid_attributes plant.update({"kitList": kit}) - except requests.exceptions.HTTPError as errh: - raise requests.exceptions.HTTPError(errh) - except requests.exceptions.ConnectionError as errc: - raise requests.exceptions.ConnectionError(errc) - except requests.exceptions.Timeout as errt: - raise requests.exceptions.Timeout(errt) - except requests.exceptions.RequestException as errr: - raise requests.exceptions.RequestException(errr) + except HTTPError as errh: + raise HTTPError(errh) + except ConnectionError as errc: + raise ConnectionError(errc) + except Timeout as errt: + raise Timeout(errt) + except RequestException as errr: + raise RequestException(errr) diff --git a/custom_components/saj_esolar_air/sensor.py b/custom_components/saj_esolar_air/sensor.py index b418c58..b2701e5 100644 --- a/custom_components/saj_esolar_air/sensor.py +++ b/custom_components/saj_esolar_air/sensor.py @@ -2,8 +2,8 @@ from __future__ import annotations import datetime -from datetime import timedelta import logging +from datetime import timedelta from homeassistant.components.sensor import ( SensorDeviceClass, @@ -17,7 +17,7 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.update_coordinator import CoordinatorEntity -from . import ESolarCoordinator, ESolarResponse +from . import ESolarCoordinator from .const import ( B_B_LOAD, B_BACKUP_POWER_W, @@ -120,7 +120,7 @@ async def async_setup_entry( """Set up the eSolar sensor.""" coordinator: ESolarCoordinator = hass.data[DOMAIN][entry.entry_id] entities: list[ESolarSensor] = [] - esolar_data: ESolarResponse = coordinator.data + esolar_data = coordinator.data my_plants = entry.options.get(CONF_MONITORED_SITES) use_inverter_sensors = entry.options.get(CONF_INVERTER_SENSORS) use_pv_grid_attributes = entry.options.get(CONF_PV_GRID_DATA) @@ -129,98 +129,98 @@ async def async_setup_entry( return for enabled_plant in my_plants: - for plant in esolar_data["plantList"]: - if plant["plantname"] != enabled_plant: + for plant in esolar_data: + if plant["plantName"] != enabled_plant: continue _LOGGER.debug( - "Setting up ESolarSensorPlant sensor for %s", plant["plantname"] + "Setting up ESolarSensorPlant sensor for %s", plant["plantName"] ) entities.append( - ESolarSensorPlant(coordinator, plant["plantname"], plant["plantuid"]) + ESolarSensorPlant(coordinator, plant["plantName"], plant["plantUid"]) ) if plant["type"] == 0: _LOGGER.debug( "Setting up ESolarSensorPlantTotalEnergy sensor for %s", - plant["plantname"], + plant["plantName"], ) entities.append( ESolarSensorPlantTotalEnergy( - coordinator, plant["plantname"], plant["plantuid"] + coordinator, plant["plantName"], plant["plantUid"] ) ) elif plant["type"] == 3: _LOGGER.debug( "Setting up ESolarSensorPlantBatterySellEnergy sensor for %s", - plant["plantname"], + plant["plantName"], ) entities.append( ESolarSensorPlantBatterySellEnergy( - coordinator, plant["plantname"], plant["plantuid"] + coordinator, plant["plantName"], plant["plantUid"] ) ) _LOGGER.debug( "Setting up ESolarSensorPlantBatteryBuyEnergy sensor for %s", - plant["plantname"], + plant["plantName"], ) entities.append( ESolarSensorPlantBatteryBuyEnergy( - coordinator, plant["plantname"], plant["plantuid"] + coordinator, plant["plantName"], plant["plantUid"] ) ) _LOGGER.debug( "Setting up ESolarSensorPlantBatteryChargeEnergy sensor for %s", - plant["plantname"], + plant["plantName"], ) entities.append( ESolarSensorPlantBatteryChargeEnergy( - coordinator, plant["plantname"], plant["plantuid"] + coordinator, plant["plantName"], plant["plantUid"] ) ) _LOGGER.debug( "Setting up ESolarSensorPlantBatteryDischargeEnergy sensor for %s", - plant["plantname"], + plant["plantName"], ) entities.append( ESolarSensorPlantBatteryDischargeEnergy( - coordinator, plant["plantname"], plant["plantuid"] + coordinator, plant["plantName"], plant["plantUid"] ) ) _LOGGER.debug( "Setting up ESolarSensorPlantBatterySoC sensor for %s", - plant["plantname"], + plant["plantName"], ) entities.append( ESolarSensorPlantBatterySoC( - coordinator, plant["plantname"], plant["plantuid"] + coordinator, plant["plantName"], plant["plantUid"] ) ) if use_inverter_sensors: - for inverter in plant["plantDetail"]["snList"]: + for inverter in plant["deviceSnList"]: _LOGGER.debug( "Setting up ESolarInverterEnergyTotal sensor for %s and inverter %s", - plant["plantname"], + plant["plantName"], inverter, ) entities.append( ESolarInverterEnergyTotal( coordinator, - plant["plantname"], - plant["plantuid"], + plant["plantName"], + plant["plantUid"], inverter, ) ) _LOGGER.debug( "Setting up ESolarInverterPower sensor for %s and inverter %s", - plant["plantname"], + plant["plantName"], inverter, ) entities.append( ESolarInverterPower( coordinator, - plant["plantname"], - plant["plantuid"], + plant["plantName"], + plant["plantUid"], inverter, use_pv_grid_attributes, ) @@ -228,14 +228,14 @@ async def async_setup_entry( if plant["type"] == 3: _LOGGER.debug( "Setting up ESolarInverterBatterySoC sensor for %s and inverter %s", - plant["plantname"], + plant["plantName"], inverter, ) entities.append( ESolarInverterBatterySoC( coordinator, - plant["plantname"], - plant["plantuid"], + plant["plantName"], + plant["plantUid"], inverter, ) ) @@ -302,12 +302,12 @@ def __init__(self, coordinator: ESolarCoordinator, plant_name, plant_uid) -> Non async def async_update(self) -> None: """Get the latest data and updates the states.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup static attributes self._attr_available = True - self._attr_extra_state_attributes[P_NAME] = plant["plantname"] - self._attr_extra_state_attributes[P_UID] = plant["plantuid"] + self._attr_extra_state_attributes[P_NAME] = plant["plantName"] + self._attr_extra_state_attributes[P_UID] = plant["plantUid"] self._attr_extra_state_attributes[P_ADR] = ( plant["address"] + " " + plant["country"] ) @@ -319,7 +319,7 @@ async def async_update(self) -> None: self._attr_extra_state_attributes[P_TYPE] = P_TYPE_BLEND if plant["type"] == 3: self._attr_extra_state_attributes[P_TYPE] = P_TYPE_AC_COUPLING - self._attr_extra_state_attributes[P_POWER] = float(plant["systempower"]) + self._attr_extra_state_attributes[P_POWER] = float(plant["systemPower"]) self._attr_extra_state_attributes[P_CURRENCY] = plant["currency"] # Setup state @@ -335,22 +335,30 @@ async def async_update(self) -> None: @property def native_value(self) -> str | None: """Return sensor state.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + value = None + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup dynamic attributes - if (plant["plantDetail"]["type"]) == 0: - self._attr_extra_state_attributes[P_INCOME] = plant["plantDetail"][ - "income" - ] + if (plant["type"]) == 0: + self._attr_extra_state_attributes[P_INCOME] = plant["overviewInfo"]["totalIncome"] else: self._attr_extra_state_attributes[P_INCOME] = None - self._attr_extra_state_attributes[P_CO2] = plant["plantDetail"][ - "totalReduceCo2" - ] - self._attr_extra_state_attributes[P_TREES] = plant["plantDetail"][ - "totalPlantTreeNum" - ] - self._attr_extra_state_attributes[P_TOTAL_E] = plant["totalElectricity"] + + try: + self._attr_extra_state_attributes[P_CO2] = plant[ + "totalReduceCo2" + ] + except KeyError: + self._attr_extra_state_attributes[P_CO2] = None + + try: + self._attr_extra_state_attributes[P_TREES] = plant[ + "totalPlantTreeNum" + ] + except KeyError: + self._attr_extra_state_attributes[P_TREES] = None + + self._attr_extra_state_attributes[P_TOTAL_E] = plant["totalEnergy"] # Setup state if plant["runningState"] == 1: @@ -359,8 +367,6 @@ def native_value(self) -> str | None: value = "Alarm" elif plant["runningState"] == 3: value = "Offline" - else: - value = None return value @@ -397,27 +403,27 @@ def __init__(self, coordinator: ESolarCoordinator, plant_name, plant_uid) -> Non async def async_update(self) -> None: """Get the latest data and updates the states.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup static attributes self._attr_available = True - self._attr_extra_state_attributes[P_NAME] = plant["plantname"] - self._attr_extra_state_attributes[P_UID] = plant["plantuid"] + self._attr_extra_state_attributes[P_NAME] = plant["plantName"] + self._attr_extra_state_attributes[P_UID] = plant["plantUid"] # Setup state - self._attr_native_value = float(plant["totalElectricity"]) + self._attr_native_value = float(plant["totalEnergy"]) @property def native_value(self) -> float | None: """Return sensor state.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup dynamic attributes self._attr_extra_state_attributes[P_TODAY_E] = float( - plant["todayElectricity"] + plant["todayEnergy"] ) self._attr_extra_state_attributes[P_CURRENT_POWER] = float( - plant["nowPower"] + plant["powerNow"] ) if plant["type"] == 0: peak_power = float(0.0) @@ -429,7 +435,7 @@ def native_value(self) -> float | None: self._attr_extra_state_attributes[P_PEAK_POWER] = None # Setup state - value = float(plant["totalElectricity"]) + value = float(plant["totalEnergy"]) return value @@ -463,27 +469,27 @@ def __init__(self, coordinator: ESolarCoordinator, plant_name, plant_uid) -> Non async def async_update(self) -> None: """Get the latest data and updates the states.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup static attributes self._attr_available = True - self._attr_extra_state_attributes[P_NAME] = plant["plantname"] - self._attr_extra_state_attributes[P_UID] = plant["plantuid"] + self._attr_extra_state_attributes[P_NAME] = plant["plantName"] + self._attr_extra_state_attributes[P_UID] = plant["plantUid"] # Setup state - if plant["plantDetail"]["totalBuyElec"] is not None: + if plant["totalBuyElec"] is not None: self._attr_native_value = float( - plant["plantDetail"]["totalBuyElec"] + plant["totalBuyElec"] ) @property def native_value(self) -> float | None: """Return sensor state.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup state - if plant["plantDetail"]["totalBuyElec"] is not None: - value = float(plant["plantDetail"]["totalBuyElec"]) + if plant["totalBuyElec"] is not None: + value = float(plant["totalBuyElec"]) return value @@ -517,27 +523,27 @@ def __init__(self, coordinator: ESolarCoordinator, plant_name, plant_uid) -> Non async def async_update(self) -> None: """Get the latest data and updates the states.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup static attributes self._attr_available = True - self._attr_extra_state_attributes[P_NAME] = plant["plantname"] - self._attr_extra_state_attributes[P_UID] = plant["plantuid"] + self._attr_extra_state_attributes[P_NAME] = plant["plantName"] + self._attr_extra_state_attributes[P_UID] = plant["plantUid"] # Setup state - if plant["plantDetail"]["totalSellElec"] is not None: + if plant["totalSellElec"] is not None: self._attr_native_value = float( - plant["plantDetail"]["totalSellElec"] + plant["totalSellElec"] ) @property def native_value(self) -> float | None: """Return sensor state.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup state - if plant["plantDetail"]["totalSellElec"] is not None: - value = float(plant["plantDetail"]["totalSellElec"]) + if plant["totalSellElec"] is not None: + value = float(plant["totalSellElec"]) return value @@ -571,12 +577,12 @@ def __init__(self, coordinator: ESolarCoordinator, plant_name, plant_uid) -> Non async def async_update(self) -> None: """Get the latest data and updates the states.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup static attributes self._attr_available = True - self._attr_extra_state_attributes[P_NAME] = plant["plantname"] - self._attr_extra_state_attributes[P_UID] = plant["plantuid"] + self._attr_extra_state_attributes[P_NAME] = plant["plantName"] + self._attr_extra_state_attributes[P_UID] = plant["plantUid"] # Setup state charge = float(0) @@ -588,8 +594,8 @@ async def async_update(self) -> None: @property def native_value(self) -> float | None: """Return sensor state.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup state charge = float(0) if "beanList" in plant and plant["beanList"] is not None: @@ -629,12 +635,12 @@ def __init__(self, coordinator: ESolarCoordinator, plant_name, plant_uid) -> Non async def async_update(self) -> None: """Get the latest data and updates the states.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup static attributes self._attr_available = True - self._attr_extra_state_attributes[P_NAME] = plant["plantname"] - self._attr_extra_state_attributes[P_UID] = plant["plantuid"] + self._attr_extra_state_attributes[P_NAME] = plant["plantName"] + self._attr_extra_state_attributes[P_UID] = plant["plantUid"] # Setup state discharge = float(0) @@ -646,8 +652,8 @@ async def async_update(self) -> None: @property def native_value(self) -> float | None: """Return sensor state.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup state discharge = float(0) if "beanList" in plant and plant["beanList"] is not None: @@ -688,16 +694,16 @@ async def async_update(self) -> None: """Get the latest data and updates the states.""" installed_power = float(0) available_power = float(0) - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] != self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] != self._plant_name: continue # Setup static attributes self._attr_available = True - self._attr_extra_state_attributes[P_NAME] = plant["plantname"] - self._attr_extra_state_attributes[P_UID] = plant["plantuid"] + self._attr_extra_state_attributes[P_NAME] = plant["plantName"] + self._attr_extra_state_attributes[P_UID] = plant["plantUid"] # Setup state - for inverter in plant["plantDetail"]["snList"]: + for inverter in plant["deviceSnList"]: if "kitList" not in plant or plant["kitList"] is None: continue for kit in plant["kitList"]: @@ -716,11 +722,11 @@ def native_value(self) -> float | None: """Return sensor state.""" installed_power = float(0) available_power = float(0) - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] != self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] != self._plant_name: continue # Setup state - for inverter in plant["plantDetail"]["snList"]: + for inverter in plant["deviceSnList"]: if "kitList" not in plant or plant["kitList"] is None: continue for kit in plant["kitList"]: @@ -782,12 +788,12 @@ def __init__( async def async_update(self) -> None: """Get the latest data and updates the states.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup static attributes self._attr_available = True - self._attr_extra_state_attributes[P_NAME] = plant["plantname"] - self._attr_extra_state_attributes[P_UID] = plant["plantuid"] + self._attr_extra_state_attributes[P_NAME] = plant["plantName"] + self._attr_extra_state_attributes[P_UID] = plant["plantUid"] if "kitList" not in plant or plant["kitList"] is None: continue for kit in plant["kitList"]: @@ -818,8 +824,8 @@ async def async_update(self) -> None: def native_value(self) -> float | None: """Return sensor state.""" value = None - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] != self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] != self._plant_name: continue if "kitList" in plant and plant["kitList"] is not None: for kit in plant["kitList"]: @@ -929,12 +935,12 @@ def __init__( async def async_update(self) -> None: """Get the latest data and updates the states.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] == self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] == self._plant_name: # Setup static attributes self._attr_available = True - self._attr_extra_state_attributes[P_NAME] = plant["plantname"] - self._attr_extra_state_attributes[P_UID] = plant["plantuid"] + self._attr_extra_state_attributes[P_NAME] = plant["plantName"] + self._attr_extra_state_attributes[P_UID] = plant["plantUid"] if "kitList" in plant and plant["kitList"] is not None: for kit in plant["kitList"]: if kit["devicesn"] == self.inverter_sn: @@ -950,8 +956,8 @@ async def async_update(self) -> None: def native_value(self) -> float | None: """Return sensor state.""" value = None - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] != self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] != self._plant_name: continue # Setup dynamic attributes if "kitList" not in plant or plant["kitList"] is None: @@ -1129,13 +1135,13 @@ def __init__( async def async_update(self) -> None: """Get the latest data and updates the states.""" - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] != self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] != self._plant_name: continue # Setup static attributes self._attr_available = True - self._attr_extra_state_attributes[P_NAME] = plant["plantname"] - self._attr_extra_state_attributes[P_UID] = plant["plantuid"] + self._attr_extra_state_attributes[P_NAME] = plant["plantName"] + self._attr_extra_state_attributes[P_UID] = plant["plantUid"] if "kitList" not in plant or plant["kitList"] is None: continue for kit in plant["kitList"]: @@ -1157,8 +1163,8 @@ def native_value(self) -> float | None: """Return sensor state.""" # Setup state value = None - for plant in self._coordinator.data["plantList"]: - if plant["plantname"] != self._plant_name: + for plant in self._coordinator.data: + if plant["plantName"] != self._plant_name: continue if "kitList" not in plant or plant["kitList"] is None: continue diff --git a/custom_components/saj_esolar_air/test_esolar.py b/custom_components/saj_esolar_air/test_esolar.py new file mode 100644 index 0000000..2e75984 --- /dev/null +++ b/custom_components/saj_esolar_air/test_esolar.py @@ -0,0 +1,36 @@ +import unittest + +from custom_components.saj_esolar_air.esolar import get_esolar_data, \ + generate_signature, encrypt_password + + +class TestGetEsolarData(unittest.TestCase): + + def test_get_esolar_data_success(self): + region = "eu" + username = "user" + password = "password" + result = get_esolar_data(region, username, password) + self.assertIsNotNone(result) + + + def test_verify_hash(self): + password = "passwd" + encrypted_password = encrypt_password(password) + self.assertEqual(encrypted_password, "a8055cb2a374c3d0f44d95c36d90073a") + + + def test_signature(self): + data = { + "appProjectName": "elekeeper", + "clientDate": "2025-07-06", + "clientId": "esolar-monitor-admin", + "lang": "en", + "random": "BGZWCi5XnfCkxw2X2Gw8y364QBkm6636", + "timeStamp": "1751834849930" + } + signature = generate_signature(data) + self.assertEqual(signature, "4E0F8614295B5D4802DAE4D0ED8AFA2EB70F09D1") + +if __name__ == "__main__": + unittest.main()