From f505cbf77b26919f02d50edfad032f05522b7e0b Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Sat, 13 Sep 2025 15:00:18 +0200 Subject: [PATCH 01/14] Proper python module --- soliscloud_api/__init__.py | 851 +------------------------------------ 1 file changed, 6 insertions(+), 845 deletions(-) diff --git a/soliscloud_api/__init__.py b/soliscloud_api/__init__.py index 7f0b168..409ca8b 100644 --- a/soliscloud_api/__init__.py +++ b/soliscloud_api/__init__.py @@ -3,850 +3,11 @@ For more information: https://github.com/hultenvp/soliscloud_api """ -from __future__ import annotations - -import hashlib -import hmac -import base64 -import asyncio -import re -from datetime import datetime -from datetime import timezone -from enum import Enum -from http import HTTPStatus -import json -from typing import Any -from throttler import throttle -from aiohttp import ClientError, ClientSession -import async_timeout +from .client import SoliscloudAPI # noqa: F401 +from .client import SoliscloudError, SoliscloudHttpError # noqa: F401 +from .client import SoliscloudTimeoutError, SoliscloudApiError # noqa: F401 +from .entities import Plant, Inverter, Collector # noqa: F401 # VERSION -VERSION = '1.2.0' -SUPPORTED_SPEC_VERSION = '2.0' -RESOURCE_PREFIX = '/v1/api/' - -VERB = "POST" - -# Endpoints -USER_STATION_LIST = RESOURCE_PREFIX + 'userStationList' -STATION_DETAIL = RESOURCE_PREFIX + 'stationDetail' -COLLECTOR_LIST = RESOURCE_PREFIX + 'collectorList' -COLLECTOR_DETAIL = RESOURCE_PREFIX + 'collectorDetail' -COLLECTOR_DAY = RESOURCE_PREFIX + 'collector/day' -INVERTER_LIST = RESOURCE_PREFIX + 'inverterList' -INVERTER_DETAIL = RESOURCE_PREFIX + 'inverterDetail' -STATION_DAY = RESOURCE_PREFIX + 'stationDay' -STATION_MONTH = RESOURCE_PREFIX + 'stationMonth' -STATION_YEAR = RESOURCE_PREFIX + 'stationYear' -STATION_ALL = RESOURCE_PREFIX + 'stationAll' -INVERTER_DAY = RESOURCE_PREFIX + 'inverterDay' -INVERTER_MONTH = RESOURCE_PREFIX + 'inverterMonth' -INVERTER_YEAR = RESOURCE_PREFIX + 'inverterYear' -INVERTER_ALL = RESOURCE_PREFIX + 'inverterAll' -INVERTER_SHELF_TIME = RESOURCE_PREFIX + 'inverter/shelfTime' -ALARM_LIST = RESOURCE_PREFIX + 'alarmList' -STATION_DETAIL_LIST = RESOURCE_PREFIX + 'stationDetailList' -INVERTER_DETAIL_LIST = RESOURCE_PREFIX + 'inverterDetailList' -STATION_DAY_ENERGY_LIST = RESOURCE_PREFIX + 'stationDayEnergyList' -STATION_MONTH_ENERGY_LIST = RESOURCE_PREFIX + 'stationMonthEnergyList' -STATION_YEAR_ENERGY_LIST = RESOURCE_PREFIX + 'stationYearEnergyList' -EPM_LIST = RESOURCE_PREFIX + 'epmList' -EPM_DETAIL = RESOURCE_PREFIX + 'epmDetail' -EPM_DAY = RESOURCE_PREFIX + 'epm/day' -EPM_MONTH = RESOURCE_PREFIX + 'epm/month' -EPM_YEAR = RESOURCE_PREFIX + 'epm/year' -EPM_ALL = RESOURCE_PREFIX + 'epm/all' -WEATHER_LIST = RESOURCE_PREFIX + 'weatherList' -WEATHER_DETAIL = RESOURCE_PREFIX + 'weatherDetail' - - -ONLY_INV_ID_OR_SN_ERR = \ - "Only pass one of inverter_id or inverter_sn as identifier" -INV_SN_ERR = "Pass inverter_sn as identifier" -ONLY_COL_ID_OR_SN_ERR = \ - "Only pass one of collector_id or collector_sn as identifier" -COL_SN_ERR = "Pass collector_sn as identifier" -ONLY_STN_ID_OR_SN_ERR = \ - "Only pass one of station_id or nmi_code as identifier" -PAGE_SIZE_ERR = "page_size must be <= 100" -WEATHER_SN_ERR = "Pass instrument_sn as identifier, \ -containing weather instrument serial" - - -class SoliscloudAPI(): - """Class with functions for reading data from the Soliscloud Portal.""" - - class SolisCloudError(Exception): - """ - Exception raised for timeouts during calls. - """ - - def __init__(self, message="SolisCloud API error"): - - self.message = message - super().__init__(self.message) - - class HttpError(SolisCloudError): - """ - Exception raised for HTTP errors during calls. - """ - - def __init__(self, statuscode, message=None): - self.statuscode = statuscode - self.message = message - if not message: - if statuscode == 408: - now = datetime.now().strftime("%d-%m-%Y %H:%M GMT") - self.message = f"Your system time is different from \ -server time, your time is {now}" - else: - self.message = f"Http status code: {statuscode}" - super().__init__(self.message) - - class TimeoutError(SolisCloudError): - """ - Exception raised for timeouts during calls. - """ - - def __init__(self, message="Timeout error occurred"): - - self.message = message - super().__init__(self.message) - - class ApiError(SolisCloudError): - """ - Exception raised for errors during API calls. - """ - - def __init__( - self, - message="Undefined API error occurred", - code="Unknown", - response=None - ): - - self.message = message - self.code = code - self.response = response - super().__init__(self.message) - - def __str__(self): - return f'API returned an error: {self.message}, \ -error code: {self.code}, response: {self.response}' - - def __init__(self, domain: str, session: ClientSession) -> None: - self._domain = domain.rstrip("/") - self._session: ClientSession = session - - class DateFormat(Enum): - DAY = 0 - MONTH = 1 - YEAR = 2 - - @property - def domain(self) -> str: - """ Domain name.""" - return self._domain - - @property - def session(self) -> ClientSession: - """ aiohttp client session ID.""" - return self._session - - @property - def spec_version(self) -> str: - """ supported version of the Soliscloud spec.""" - return SUPPORTED_SPEC_VERSION - - # All methods take key and secret as positional arguments followed by - # one or more keyword arguments - async def user_station_list( - self, key_id: str, secret: bytes, /, *, - page_no: int = 1, - page_size: int = 20, - nmi_code: str = None - ) -> dict[str, str]: - """Power station List""" - - if page_size > 100: - raise SoliscloudAPI.SolisCloudError(PAGE_SIZE_ERR) - - params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} - if nmi_code is not None: - params['nmiCode'] = nmi_code - return await self._get_records( - USER_STATION_LIST, key_id, secret, params - ) - - async def station_detail( - self, key_id: str, secret: bytes, /, *, - station_id: int, - nmi_code: str = None - ) -> dict[str, str]: - """Power station details""" - - params: dict[str, Any] = {'id': station_id} - if nmi_code is not None: - params['nmiCode'] = nmi_code - return await self._get_data(STATION_DETAIL, key_id, secret, params) - - async def collector_list( - self, key_id: str, secret: bytes, /, *, - page_no: int = 1, - page_size: int = 20, - station_id: int = None, - nmi_code: str = None - ) -> dict[str, str]: - """Datalogger list""" - - if page_size > 100: - raise SoliscloudAPI.SolisCloudError(PAGE_SIZE_ERR) - - params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} - if station_id is not None: - params['stationId'] = station_id - if nmi_code is not None: - params['nmiCode'] = nmi_code - return await self._get_records(COLLECTOR_LIST, key_id, secret, params) - - async def collector_detail( - self, key_id: str, secret: bytes, /, *, - collector_sn: int = None, - collector_id: str = None - ) -> dict[str, str]: - """Datalogger details""" - - params: dict[str, Any] = {} - if (collector_sn is not None and collector_id is None): - params['sn'] = collector_sn - elif (collector_sn is None and collector_id is not None): - params['id'] = collector_id - else: - raise SoliscloudAPI.SolisCloudError(ONLY_COL_ID_OR_SN_ERR) - return await self._get_data(COLLECTOR_DETAIL, key_id, secret, params) - - async def collector_day( - self, key_id: str, secret: bytes, /, *, - collector_sn: int = None, - time: str, - time_zone: int, - ) -> dict[str, str]: - """Datalogger day statistics""" - - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, time) - params: dict[str, Any] = { - 'time': time, - 'timeZone': time_zone - } - - if (collector_sn is None): - raise SoliscloudAPI.SolisCloudError(COL_SN_ERR) - params['sn'] = collector_sn - - return await self._get_data(COLLECTOR_DAY, key_id, secret, params) - - async def inverter_list( - self, key_id: str, secret: bytes, /, *, - page_no: int = 1, - page_size: int = 20, - station_id: str = None, - nmi_code: str = None - ) -> dict[str, str]: - """Inverter list""" - - if page_size > 100: - raise SoliscloudAPI.SolisCloudError(PAGE_SIZE_ERR) - - params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} - if station_id is not None: - # If not specified all inverters for all stations for key_id are - # returned - params['stationId'] = station_id - if nmi_code is not None: - params['nmiCode'] = nmi_code - return await self._get_records(INVERTER_LIST, key_id, secret, params) - - async def inverter_detail( - self, key_id: str, secret: bytes, /, *, - inverter_sn: int = None, - inverter_id: str = None - ) -> dict[str, str]: - """Inverter details""" - - params: dict[str, Any] = {} - if (inverter_sn is not None and inverter_id is None): - params['sn'] = inverter_sn - elif (inverter_sn is None and inverter_id is not None): - params['id'] = inverter_id - else: - raise SoliscloudAPI.SolisCloudError(ONLY_INV_ID_OR_SN_ERR) - return await self._get_data(INVERTER_DETAIL, key_id, secret, params) - - async def station_day( - self, key_id: str, secret: bytes, /, *, - currency: str, - time: str, - time_zone: int, - station_id: int = None, - nmi_code=None - ) -> dict[str, str]: - """Station daily graph""" - - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, time) - params: dict[str, Any] = { - 'money': currency, - 'time': time, - 'timeZone': time_zone - } - - if (station_id is not None and nmi_code is None): - params['id'] = station_id - elif (station_id is None and nmi_code is not None): - params['nmiCode'] = nmi_code - else: - raise SoliscloudAPI.SolisCloudError(ONLY_STN_ID_OR_SN_ERR) - - return await self._get_data(STATION_DAY, key_id, secret, params) - - async def station_month( - self, key_id: str, secret: bytes, /, *, - currency: str, - month: str, - station_id: int = None, - nmi_code=None - ) -> dict[str, str]: - """Station monthly graph""" - - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.MONTH, month) - params: dict[str, Any] = {'money': currency, 'month': month} - - if (station_id is not None and nmi_code is None): - params['id'] = station_id - elif (station_id is None and nmi_code is not None): - params['nmiCode'] = nmi_code - else: - raise SoliscloudAPI.SolisCloudError(ONLY_STN_ID_OR_SN_ERR) - - return await self._get_data(STATION_MONTH, key_id, secret, params) - - async def station_year( - self, key_id: str, secret: bytes, /, *, - currency: str, - year: str, - station_id: int = None, - nmi_code=None - ) -> dict[str, str]: - """Station yearly graph""" - - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.YEAR, year) - params: dict[str, Any] = {'money': currency, 'year': year} - - if (station_id is not None and nmi_code is None): - params['id'] = station_id - elif (station_id is None and nmi_code is not None): - params['nmiCode'] = nmi_code - else: - raise SoliscloudAPI.SolisCloudError(ONLY_STN_ID_OR_SN_ERR) - - return await self._get_data(STATION_YEAR, key_id, secret, params) - - async def station_all( - self, key_id: str, secret: bytes, /, *, - currency: str, - station_id: int = None, - nmi_code: str = None - ) -> dict[str, str]: - """Station cumulative graph""" - - params: dict[str, Any] = {'money': currency} - if (station_id is not None and nmi_code is None): - params['id'] = station_id - elif (station_id is None and nmi_code is not None): - params['nmiCode'] = nmi_code - else: - raise SoliscloudAPI.SolisCloudError(ONLY_STN_ID_OR_SN_ERR) - - return await self._get_data(STATION_ALL, key_id, secret, params) - - async def inverter_day( - self, key_id: str, secret: bytes, /, *, - currency: str, - time: str, - time_zone: int, - inverter_id: int = None, - inverter_sn: str = None - ) -> dict[str, str]: - """Inverter daily graph""" - - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, time) - params: dict[str, Any] = { - 'money': currency, - 'time': time, - 'timeZone': time_zone - } - - if (inverter_id is not None and inverter_sn is None): - params['id'] = inverter_id - elif (inverter_id is None and inverter_sn is not None): - params['sn'] = inverter_sn - else: - raise SoliscloudAPI.SolisCloudError(ONLY_INV_ID_OR_SN_ERR) - - return await self._get_data(INVERTER_DAY, key_id, secret, params) - - async def inverter_month( - self, key_id: str, secret: bytes, /, *, - currency: str, - month: str, - inverter_id: int = None, - inverter_sn: str = None - ) -> dict[str, str]: - """Inverter monthly graph""" - - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.MONTH, month) - params: dict[str, Any] = {'money': currency, 'month': month} - - if (inverter_id is not None and inverter_sn is None): - params['id'] = inverter_id - elif (inverter_id is None and inverter_sn is not None): - params['sn'] = inverter_sn - else: - raise SoliscloudAPI.SolisCloudError(ONLY_INV_ID_OR_SN_ERR) - - return await self._get_data(INVERTER_MONTH, key_id, secret, params) - - async def inverter_year( - self, key_id: str, secret: bytes, /, *, - currency: str, - year: str, - inverter_id: int = None, - inverter_sn: str = None - ) -> dict[str, str]: - """Inverter yearly graph""" - - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.YEAR, year) - params: dict[str, Any] = {'money': currency, 'year': year} - - if (inverter_id is not None and inverter_sn is None): - params['id'] = inverter_id - elif (inverter_id is None and inverter_sn is not None): - params['sn'] = inverter_sn - else: - raise SoliscloudAPI.SolisCloudError(ONLY_INV_ID_OR_SN_ERR) - - return await self._get_data(INVERTER_YEAR, key_id, secret, params) - - async def inverter_all( - self, key_id: str, secret: bytes, /, *, - currency: str, - inverter_id: int = None, - inverter_sn: str = None - ) -> dict[str, str]: - """Inverter cumulative graph""" - - params: dict[str, Any] = {'money': currency} - if (inverter_id is not None and inverter_sn is None): - params['id'] = inverter_id - elif (inverter_id is None and inverter_sn is not None): - params['sn'] = inverter_sn - else: - raise SoliscloudAPI.SolisCloudError(ONLY_INV_ID_OR_SN_ERR) - - return await self._get_data(INVERTER_ALL, key_id, secret, params) - - async def inverter_shelf_time( - self, key_id: str, secret: bytes, /, *, - page_no: int = 1, - page_size: int = 20, - inverter_sn: str = None - ) -> dict[str, str]: - """Inverter warranty information""" - - if page_size > 100: - raise SoliscloudAPI.SolisCloudError(PAGE_SIZE_ERR) - if inverter_sn is None: - raise SoliscloudAPI.SolisCloudError(INV_SN_ERR) - - params: dict[str, Any] = { - 'pageNo': page_no, - 'pageSize': page_size, - 'sn': inverter_sn} - - return await self._get_records( - INVERTER_SHELF_TIME, key_id, secret, params) - - async def alarm_list( - self, key_id: str, secret: bytes, /, *, - page_no: int = 1, - page_size: int = 20, - station_id: str = None, - device_sn: str = None, - begintime: str = None, - endtime: str = None, - nmi_code: str = None - ) -> dict[str, str]: - """Alarm check""" - - if page_size > 100: - raise SoliscloudAPI.SolisCloudError(PAGE_SIZE_ERR) - - params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} - if station_id is not None and device_sn is None: - params['stationId'] = station_id - elif station_id is None and device_sn is not None: - params['alarmDeviceSn'] = device_sn - else: - raise SoliscloudAPI.SolisCloudError( - "Only pass one of station_id or device_sn as identifier") - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, begintime) - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, endtime) - params['alarmBeginTime'] = begintime - params['alarmEndTime'] = endtime - if nmi_code is not None: - params['nmiCode'] = nmi_code - return await self._get_records(ALARM_LIST, key_id, secret, params) - - async def station_detail_list( - self, key_id: str, secret: bytes, /, *, - page_no: int = 1, - page_size: int = 20 - ) -> dict[str, str]: - """Batch acquire station details""" - - if page_size > 100: - raise SoliscloudAPI.SolisCloudError(PAGE_SIZE_ERR) - params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} - - return await self._get_records( - STATION_DETAIL_LIST, key_id, secret, params) - - async def inverter_detail_list( - self, key_id: str, secret: bytes, /, *, - page_no: int = 1, - page_size: int = 20 - ) -> dict[str, str]: - """Batch acquire inverter details""" - - if page_size > 100: - raise SoliscloudAPI.SolisCloudError(PAGE_SIZE_ERR) - params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} - - return await self._get_records( - INVERTER_DETAIL_LIST, key_id, secret, params) - - async def station_day_energy_list( - self, key_id: str, secret: bytes, /, *, - page_no: int = 1, - page_size: int = 20, - time: str - ) -> dict[str, str]: - """Batch acquire station daily generation""" - - if page_size > 100: - raise SoliscloudAPI.SolisCloudError(PAGE_SIZE_ERR) - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, time) - params: dict[str, Any] = { - 'pageNo': page_no, - 'pageSize': page_size, - 'time': time - } - - return await self._get_records( - STATION_DAY_ENERGY_LIST, key_id, secret, params) - - async def station_month_energy_list( - self, key_id: str, secret: bytes, /, *, - page_no: int = 1, - page_size: int = 20, - month: str - ) -> dict[str, str]: - """Batch acquire station monthly generation""" - - if page_size > 100: - raise SoliscloudAPI.SolisCloudError(PAGE_SIZE_ERR) - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.MONTH, month) - params: dict[str, Any] = { - 'pageNo': page_no, - 'pageSize': page_size, - 'time': month - } - - return await self._get_records( - STATION_MONTH_ENERGY_LIST, key_id, secret, params) - - async def station_year_energy_list( - self, key_id: str, secret: bytes, /, *, - page_no: int = 1, - page_size: int = 20, - year: str - ) -> dict[str, str]: - """Batch acquire station yearly generation""" - - if page_size > 100: - raise SoliscloudAPI.SolisCloudError(PAGE_SIZE_ERR) - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.YEAR, year) - params: dict[str, Any] = { - 'pageNo': page_no, - 'pageSize': page_size, - 'time': year - } - - return await self._get_records( - STATION_YEAR_ENERGY_LIST, key_id, secret, params) - - async def epm_list( - self, key_id: str, secret: bytes, /, *, - page_no: int = 1, - page_size: int = 20, - station_id: str = None - ) -> dict[str, str]: - """EPM list""" - - if page_size > 100: - raise SoliscloudAPI.SolisCloudError(PAGE_SIZE_ERR) - params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} - if station_id is not None: - params['stationId'] = station_id - - return await self._get_records(EPM_LIST, key_id, secret, params) - - async def epm_detail( - self, key_id: str, secret: bytes, /, *, - epm_sn: str - ) -> dict[str, str]: - """EPM details""" - - params: dict[str, Any] = {'sn': epm_sn} - - return await self._get_data(EPM_DETAIL, key_id, secret, params) - - async def epm_day( - self, key_id: str, secret: bytes, /, *, - searchinfo: str, - epm_sn: str, - time: str, - time_zone: int - ) -> dict[str, str]: - """EPM daily graph""" - - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, time) - params: dict[str, Any] = { - 'searchinfo': searchinfo, - 'sn': epm_sn, - 'time': time, - 'timezone': time_zone} - - return await self._get_data(EPM_DAY, key_id, secret, params) - - async def epm_month( - self, key_id: str, secret: bytes, /, *, - epm_sn: str, - month: str, - ) -> dict[str, str]: - """EPM monthly graph""" - - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.MONTH, month) - params: dict[str, Any] = {'sn': epm_sn, 'month': month} - - return await self._get_data(EPM_MONTH, key_id, secret, params) - - async def epm_year( - self, key_id: str, secret: bytes, /, *, - epm_sn: str, - year: str - ) -> dict[str, str]: - """EPM yearly graph""" - - SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.YEAR, year) - params: dict[str, Any] = {'sn': epm_sn, 'year': year} - - return await self._get_data(EPM_YEAR, key_id, secret, params) - - async def epm_all( - self, key_id: str, secret: bytes, /, *, - epm_sn: str - ) -> dict[str, str]: - """EPM cumulative graph""" - - params: dict[str, Any] = {'sn': epm_sn} - - return await self._get_data(EPM_ALL, key_id, secret, params) - - async def weather_list( - self, key_id: str, secret: bytes, /, *, - page_no: int = 1, - page_size: int = 20, - station_id: str = None, - nmi_code: str = None - ) -> dict[str, str]: - """Weather list""" - - if page_size > 100: - raise SoliscloudAPI.SolisCloudError(PAGE_SIZE_ERR) - - params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} - if station_id is not None: - # If not specified all inverters for all stations for key_id are - # returned - params['stationId'] = station_id - if nmi_code is not None: - params['nmiCode'] = nmi_code - return await self._get_records(WEATHER_LIST, key_id, secret, params) - - async def weather_detail( - self, key_id: str, secret: bytes, /, *, - instrument_sn: str = None - ) -> dict[str, str]: - """Inverter details""" - - params: dict[str, Any] = {} - if instrument_sn is None: - raise SoliscloudAPI.SolisCloudError(WEATHER_SN_ERR) - params['sn'] = instrument_sn - - return await self._get_data(WEATHER_DETAIL, key_id, secret, params) - - async def _get_records( - self, canonicalized_resource: str, key_id: str, secret: bytes, - params: dict[str, Any] - ): - """ - Return all records from call - """ - - header: dict[str, str] = SoliscloudAPI._prepare_header( - key_id, secret, params, canonicalized_resource) - - url = f"{self.domain}{canonicalized_resource}" - try: - result = await self._post_data_json(url, header, params) - if 'page' in result.keys(): - return result['page']['records'] - else: - return result['records'] - except KeyError as err: - raise SoliscloudAPI.ApiError("Malformed data", result) from err - - async def _get_data( - self, canonicalized_resource: str, key_id: str, secret: bytes, - params: dict[str, Any] - ): - """ - Return data from call - """ - - header: dict[str, str] = SoliscloudAPI._prepare_header( - key_id, secret, params, canonicalized_resource) - - url = f"{self.domain}{canonicalized_resource}" - result = await self._post_data_json(url, header, params) - - return result - - @staticmethod - def _now() -> datetime.datetime: - return datetime.now(timezone.utc) - - @staticmethod - def _prepare_header( - key_id: str, - secret: bytes, - body: dict[str, str], - canonicalized_resource: str - ) -> dict[str, str]: - content_md5 = base64.b64encode( - hashlib.md5(json.dumps(body, separators=( - ",", ":")).encode('utf-8')).digest() - ).decode('utf-8') - - content_type = "application/json" - - date = SoliscloudAPI._now().strftime("%a, %d %b %Y %H:%M:%S GMT") - - encrypt_str = ( - VERB + "\n" - + content_md5 + "\n" - + content_type + "\n" - + date + "\n" - + canonicalized_resource - ) - hmac_obj = hmac.new( - secret, - msg=encrypt_str.encode('utf-8'), - digestmod=hashlib.sha1 - ) - sign = base64.b64encode(hmac_obj.digest()) - authorization = "API " + key_id + ":" + sign.decode('utf-8') - - header: dict[str, str] = { - "Content-MD5": content_md5, - "Content-Type": content_type, - "Date": date, - "Authorization": authorization - } - return header - - @throttle(rate_limit=2, period=1.0) - async def _post_data_json( - self, - url: str, - header: dict[str, Any], - params: dict[str, Any] - ) -> dict[str, Any]: - """ Http-post data to specified domain/canonicalized_resource. """ - - resp = None - result = None - if self._session is None: - raise SoliscloudAPI.SolisCloudError( - "aiohttp.ClientSession not set") - try: - async with async_timeout.timeout(10): - resp = await SoliscloudAPI._do_post_aiohttp( - self._session, url, params, header) - - result = await resp.json() - if resp.status == HTTPStatus.OK: - if result['code'] != '0': - raise SoliscloudAPI.ApiError( - result['msg'], result['code']) - return result['data'] - else: - raise SoliscloudAPI.HttpError(resp.status) - except asyncio.TimeoutError as err: - if resp is not None: - await resp.release() - raise SoliscloudAPI.TimeoutError() from err - except ClientError as err: - if resp is not None: - await resp.release() - raise SoliscloudAPI.ApiError(err) - except (KeyError, TypeError) as err: - raise SoliscloudAPI.ApiError( - "Malformed server response", response=result) from err - - @staticmethod - async def _do_post_aiohttp( - session, - url: str, - params: dict[str, Any], - header: dict[str, Any] - ) -> dict[str, Any]: - """ Allows mocking for unit tests.""" - return await session.post(url, json=params, headers=header) - - @staticmethod - def _verify_date(format: SoliscloudAPI.DateFormat, date: str): - rex = re.compile("^[0-9]{4}-[0-9]{2}-[0-9]{2}$") - err = SoliscloudAPI.SolisCloudError( - "time must be in format YYYY-MM-DD") - if format == SoliscloudAPI.DateFormat.MONTH: - rex = re.compile("^[0-9]{4}-[0-9]{2}$") - err = SoliscloudAPI.SolisCloudError( - "month must be in format YYYY-MM") - elif format == SoliscloudAPI.DateFormat.YEAR: - rex = re.compile("^[0-9]{4}$") - err = SoliscloudAPI.SolisCloudError("year must be in format YYYY") - if not rex.match(date): - raise err - return +VERSION = '1.3.0' +__version__ = VERSION From 2b5d56310f11612e21067995164727a13a778158 Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Sat, 13 Sep 2025 15:01:37 +0200 Subject: [PATCH 02/14] New higher level types --- soliscloud_api/client.py | 824 +++++++++++++++++++++++++++++++++++++ soliscloud_api/entities.py | 414 +++++++++++++++++++ soliscloud_api/types.py | 626 ++++++++++++++++++++++++++++ 3 files changed, 1864 insertions(+) create mode 100644 soliscloud_api/client.py create mode 100644 soliscloud_api/entities.py create mode 100644 soliscloud_api/types.py diff --git a/soliscloud_api/client.py b/soliscloud_api/client.py new file mode 100644 index 0000000..7a53f84 --- /dev/null +++ b/soliscloud_api/client.py @@ -0,0 +1,824 @@ +from __future__ import annotations + +import hashlib +import hmac +import base64 +import asyncio +import re +import json +import async_timeout +from datetime import datetime, timezone +from http import HTTPStatus +from enum import Enum +from typing import Any +from throttler import throttle +from aiohttp import ClientError, ClientSession + +SUPPORTED_SPEC_VERSION = '2.0' + +VERB = "POST" +RESOURCE_PREFIX = '/v1/api/' + +# Endpoints +USER_STATION_LIST = RESOURCE_PREFIX + 'userStationList' +STATION_DETAIL = RESOURCE_PREFIX + 'stationDetail' +COLLECTOR_LIST = RESOURCE_PREFIX + 'collectorList' +COLLECTOR_DETAIL = RESOURCE_PREFIX + 'collectorDetail' +COLLECTOR_DAY = RESOURCE_PREFIX + 'collector/day' +INVERTER_LIST = RESOURCE_PREFIX + 'inverterList' +INVERTER_DETAIL = RESOURCE_PREFIX + 'inverterDetail' +STATION_DAY = RESOURCE_PREFIX + 'stationDay' +STATION_MONTH = RESOURCE_PREFIX + 'stationMonth' +STATION_YEAR = RESOURCE_PREFIX + 'stationYear' +STATION_ALL = RESOURCE_PREFIX + 'stationAll' +INVERTER_DAY = RESOURCE_PREFIX + 'inverterDay' +INVERTER_MONTH = RESOURCE_PREFIX + 'inverterMonth' +INVERTER_YEAR = RESOURCE_PREFIX + 'inverterYear' +INVERTER_ALL = RESOURCE_PREFIX + 'inverterAll' +INVERTER_SHELF_TIME = RESOURCE_PREFIX + 'inverter/shelfTime' +ALARM_LIST = RESOURCE_PREFIX + 'alarmList' +STATION_DETAIL_LIST = RESOURCE_PREFIX + 'stationDetailList' +INVERTER_DETAIL_LIST = RESOURCE_PREFIX + 'inverterDetailList' +STATION_DAY_ENERGY_LIST = RESOURCE_PREFIX + 'stationDayEnergyList' +STATION_MONTH_ENERGY_LIST = RESOURCE_PREFIX + 'stationMonthEnergyList' +STATION_YEAR_ENERGY_LIST = RESOURCE_PREFIX + 'stationYearEnergyList' +EPM_LIST = RESOURCE_PREFIX + 'epmList' +EPM_DETAIL = RESOURCE_PREFIX + 'epmDetail' +EPM_DAY = RESOURCE_PREFIX + 'epm/day' +EPM_MONTH = RESOURCE_PREFIX + 'epm/month' +EPM_YEAR = RESOURCE_PREFIX + 'epm/year' +EPM_ALL = RESOURCE_PREFIX + 'epm/all' +WEATHER_LIST = RESOURCE_PREFIX + 'weatherList' +WEATHER_DETAIL = RESOURCE_PREFIX + 'weatherDetail' + +ONLY_INV_ID_OR_SN_ERR = \ + "Only pass one of inverter_id or inverter_sn as identifier" +INV_SN_ERR = "Pass inverter_sn as identifier" +ONLY_COL_ID_OR_SN_ERR = \ + "Only pass one of collector_id or collector_sn as identifier" +COL_SN_ERR = "Pass collector_sn as identifier" +ONLY_STN_ID_OR_SN_ERR = \ + "Only pass one of station_id or nmi_code as identifier" +PAGE_SIZE_ERR = "page_size must be <= 100" +WEATHER_SN_ERR = "Pass instrument_sn as identifier, \ +containing weather instrument serial" + + +class SoliscloudError(Exception): + """General exception for SolisCloud API errors.""" + + +class SoliscloudHttpError(SoliscloudError): + """Exception raised for HTTP errors during calls.""" + + def __init__(self, statuscode, message=None): + self.statuscode = statuscode + if not message: + if statuscode == 408: + now = datetime.now().strftime("%d-%m-%Y %H:%M GMT") + message = f"Your system time is different from server time, your time is {now}" # noqa: E501 + else: + message = f"Http status code: {statuscode}" + super().__init__(message) + + +class SoliscloudTimeoutError(SoliscloudError): + """Exception raised for timeouts during calls.""" + + def __init__(self, message="Timeout error occurred"): + super().__init__(message) + + +class SoliscloudApiError(SoliscloudError): + """Exception raised for errors during API calls.""" + + def __init__( + self, message="Undefined API error occurred", code="Unknown", + response=None): + self.code = code + self.response = response + super().__init__(message) + + def __str__(self): + return f'API returned an error: {self.args[0]}, error code: {self.code}, response: {self.response}' # noqa: E501 + + +class SoliscloudAPI(): + """Class with functions for reading data from the Soliscloud Portal.""" + + def __init__(self, domain: str, session: ClientSession) -> None: + self._domain = domain.rstrip("/") + self._session: ClientSession = session + + class DateFormat(Enum): + DAY = 0 + MONTH = 1 + YEAR = 2 + + @property + def domain(self) -> str: + """ Domain name.""" + return self._domain + + @property + def session(self) -> ClientSession: + """ aiohttp client session ID.""" + return self._session + + @property + def spec_version(self) -> str: + """ supported version of the Soliscloud spec.""" + return SUPPORTED_SPEC_VERSION + + # All methods take key and secret as positional arguments followed by + # one or more keyword arguments + async def user_station_list( + self, key_id: str, secret: bytes, /, *, + page_no: int = 1, + page_size: int = 20, + nmi_code: str = None + ) -> dict[str, str]: + """Power station List""" + + if page_size > 100: + raise SoliscloudError(PAGE_SIZE_ERR) + + params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} + if nmi_code is not None: + params['nmiCode'] = nmi_code + return await self._get_records( + USER_STATION_LIST, key_id, secret, params + ) + + async def station_detail( + self, key_id: str, secret: bytes, /, *, + station_id: int, + nmi_code: str = None + ) -> dict[str, str]: + """Power station details""" + + params: dict[str, Any] = {'id': station_id} + if nmi_code is not None: + params['nmiCode'] = nmi_code + return await self._get_data(STATION_DETAIL, key_id, secret, params) + + async def collector_list( + self, key_id: str, secret: bytes, /, *, + page_no: int = 1, + page_size: int = 20, + station_id: int = None, + nmi_code: str = None + ) -> dict[str, str]: + """Datalogger list""" + + if page_size > 100: + raise SoliscloudError(PAGE_SIZE_ERR) + + params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} + if station_id is not None: + params['stationId'] = station_id + if nmi_code is not None: + params['nmiCode'] = nmi_code + return await self._get_records(COLLECTOR_LIST, key_id, secret, params) + + async def collector_detail( + self, key_id: str, secret: bytes, /, *, + collector_sn: int = None, + collector_id: str = None + ) -> dict[str, str]: + """Datalogger details""" + + params: dict[str, Any] = {} + if (collector_sn is not None and collector_id is None): + params['sn'] = collector_sn + elif (collector_sn is None and collector_id is not None): + params['id'] = collector_id + else: + raise SoliscloudError(ONLY_COL_ID_OR_SN_ERR) + return await self._get_data(COLLECTOR_DETAIL, key_id, secret, params) + + async def collector_day( + self, key_id: str, secret: bytes, /, *, + collector_sn: int = None, + time: str, + time_zone: int, + ) -> dict[str, str]: + """Datalogger day statistics""" + + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, time) + params: dict[str, Any] = { + 'time': time, + 'timeZone': time_zone + } + + if (collector_sn is None): + raise SoliscloudError(COL_SN_ERR) + params['sn'] = collector_sn + + return await self._get_data(COLLECTOR_DAY, key_id, secret, params) + + async def inverter_list( + self, key_id: str, secret: bytes, /, *, + page_no: int = 1, + page_size: int = 20, + station_id: str = None, + nmi_code: str = None + ) -> dict[str, str]: + """Inverter list""" + + if page_size > 100: + raise SoliscloudError(PAGE_SIZE_ERR) + + params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} + if station_id is not None: + # If not specified all inverters for all stations for key_id are + # returned + params['stationId'] = station_id + if nmi_code is not None: + params['nmiCode'] = nmi_code + return await self._get_records(INVERTER_LIST, key_id, secret, params) + + async def inverter_detail( + self, key_id: str, secret: bytes, /, *, + inverter_sn: int = None, + inverter_id: str = None + ) -> dict[str, str]: + """Inverter details""" + + params: dict[str, Any] = {} + if (inverter_sn is not None and inverter_id is None): + params['sn'] = inverter_sn + elif (inverter_sn is None and inverter_id is not None): + params['id'] = inverter_id + else: + raise SoliscloudError(ONLY_INV_ID_OR_SN_ERR) + return await self._get_data(INVERTER_DETAIL, key_id, secret, params) + + async def station_day( + self, key_id: str, secret: bytes, /, *, + currency: str, + time: str, + time_zone: int, + station_id: int = None, + nmi_code=None + ) -> dict[str, str]: + """Station daily graph""" + + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, time) + params: dict[str, Any] = { + 'money': currency, + 'time': time, + 'timeZone': time_zone + } + + if (station_id is not None and nmi_code is None): + params['id'] = station_id + elif (station_id is None and nmi_code is not None): + params['nmiCode'] = nmi_code + else: + raise SoliscloudError(ONLY_STN_ID_OR_SN_ERR) + + return await self._get_data(STATION_DAY, key_id, secret, params) + + async def station_month( + self, key_id: str, secret: bytes, /, *, + currency: str, + month: str, + station_id: int = None, + nmi_code=None + ) -> dict[str, str]: + """Station monthly graph""" + + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.MONTH, month) + params: dict[str, Any] = {'money': currency, 'month': month} + + if (station_id is not None and nmi_code is None): + params['id'] = station_id + elif (station_id is None and nmi_code is not None): + params['nmiCode'] = nmi_code + else: + raise SoliscloudError(ONLY_STN_ID_OR_SN_ERR) + + return await self._get_data(STATION_MONTH, key_id, secret, params) + + async def station_year( + self, key_id: str, secret: bytes, /, *, + currency: str, + year: str, + station_id: int = None, + nmi_code=None + ) -> dict[str, str]: + """Station yearly graph""" + + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.YEAR, year) + params: dict[str, Any] = {'money': currency, 'year': year} + + if (station_id is not None and nmi_code is None): + params['id'] = station_id + elif (station_id is None and nmi_code is not None): + params['nmiCode'] = nmi_code + else: + raise SoliscloudError(ONLY_STN_ID_OR_SN_ERR) + + return await self._get_data(STATION_YEAR, key_id, secret, params) + + async def station_all( + self, key_id: str, secret: bytes, /, *, + currency: str, + station_id: int = None, + nmi_code: str = None + ) -> dict[str, str]: + """Station cumulative graph""" + + params: dict[str, Any] = {'money': currency} + if (station_id is not None and nmi_code is None): + params['id'] = station_id + elif (station_id is None and nmi_code is not None): + params['nmiCode'] = nmi_code + else: + raise SoliscloudError(ONLY_STN_ID_OR_SN_ERR) + + return await self._get_data(STATION_ALL, key_id, secret, params) + + async def inverter_day( + self, key_id: str, secret: bytes, /, *, + currency: str, + time: str, + time_zone: int, + inverter_id: int = None, + inverter_sn: str = None + ) -> dict[str, str]: + """Inverter daily graph""" + + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, time) + params: dict[str, Any] = { + 'money': currency, + 'time': time, + 'timeZone': time_zone + } + + if (inverter_id is not None and inverter_sn is None): + params['id'] = inverter_id + elif (inverter_id is None and inverter_sn is not None): + params['sn'] = inverter_sn + else: + raise SoliscloudError(ONLY_INV_ID_OR_SN_ERR) + + return await self._get_data(INVERTER_DAY, key_id, secret, params) + + async def inverter_month( + self, key_id: str, secret: bytes, /, *, + currency: str, + month: str, + inverter_id: int = None, + inverter_sn: str = None + ) -> dict[str, str]: + """Inverter monthly graph""" + + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.MONTH, month) + params: dict[str, Any] = {'money': currency, 'month': month} + + if (inverter_id is not None and inverter_sn is None): + params['id'] = inverter_id + elif (inverter_id is None and inverter_sn is not None): + params['sn'] = inverter_sn + else: + raise SoliscloudError(ONLY_INV_ID_OR_SN_ERR) + + return await self._get_data(INVERTER_MONTH, key_id, secret, params) + + async def inverter_year( + self, key_id: str, secret: bytes, /, *, + currency: str, + year: str, + inverter_id: int = None, + inverter_sn: str = None + ) -> dict[str, str]: + """Inverter yearly graph""" + + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.YEAR, year) + params: dict[str, Any] = {'money': currency, 'year': year} + + if (inverter_id is not None and inverter_sn is None): + params['id'] = inverter_id + elif (inverter_id is None and inverter_sn is not None): + params['sn'] = inverter_sn + else: + raise SoliscloudError(ONLY_INV_ID_OR_SN_ERR) + + return await self._get_data(INVERTER_YEAR, key_id, secret, params) + + async def inverter_all( + self, key_id: str, secret: bytes, /, *, + currency: str, + inverter_id: int = None, + inverter_sn: str = None + ) -> dict[str, str]: + """Inverter cumulative graph""" + + params: dict[str, Any] = {'money': currency} + if (inverter_id is not None and inverter_sn is None): + params['id'] = inverter_id + elif (inverter_id is None and inverter_sn is not None): + params['sn'] = inverter_sn + else: + raise SoliscloudError(ONLY_INV_ID_OR_SN_ERR) + + return await self._get_data(INVERTER_ALL, key_id, secret, params) + + async def inverter_shelf_time( + self, key_id: str, secret: bytes, /, *, + page_no: int = 1, + page_size: int = 20, + inverter_sn: str = None + ) -> dict[str, str]: + """Inverter warranty information""" + + if page_size > 100: + raise SoliscloudError(PAGE_SIZE_ERR) + if inverter_sn is None: + raise SoliscloudError(INV_SN_ERR) + + params: dict[str, Any] = { + 'pageNo': page_no, + 'pageSize': page_size, + 'sn': inverter_sn} + + return await self._get_records( + INVERTER_SHELF_TIME, key_id, secret, params) + + async def alarm_list( + self, key_id: str, secret: bytes, /, *, + page_no: int = 1, + page_size: int = 20, + station_id: str = None, + device_sn: str = None, + begintime: str = None, + endtime: str = None, + nmi_code: str = None + ) -> dict[str, str]: + """Alarm check""" + + if page_size > 100: + raise SoliscloudError(PAGE_SIZE_ERR) + + params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} + if station_id is not None and device_sn is None: + params['stationId'] = station_id + elif station_id is None and device_sn is not None: + params['alarmDeviceSn'] = device_sn + else: + raise SoliscloudError( + "Only pass one of station_id or device_sn as identifier") + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, begintime) + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, endtime) + params['alarmBeginTime'] = begintime + params['alarmEndTime'] = endtime + if nmi_code is not None: + params['nmiCode'] = nmi_code + return await self._get_records(ALARM_LIST, key_id, secret, params) + + async def station_detail_list( + self, key_id: str, secret: bytes, /, *, + page_no: int = 1, + page_size: int = 20 + ) -> dict[str, str]: + """Batch acquire station details""" + + if page_size > 100: + raise SoliscloudError(PAGE_SIZE_ERR) + params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} + + return await self._get_records( + STATION_DETAIL_LIST, key_id, secret, params) + + async def inverter_detail_list( + self, key_id: str, secret: bytes, /, *, + page_no: int = 1, + page_size: int = 20 + ) -> dict[str, str]: + """Batch acquire inverter details""" + + if page_size > 100: + raise SoliscloudError(PAGE_SIZE_ERR) + params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} + + return await self._get_records( + INVERTER_DETAIL_LIST, key_id, secret, params) + + async def station_day_energy_list( + self, key_id: str, secret: bytes, /, *, + page_no: int = 1, + page_size: int = 20, + time: str + ) -> dict[str, str]: + """Batch acquire station daily generation""" + + if page_size > 100: + raise SoliscloudError(PAGE_SIZE_ERR) + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, time) + params: dict[str, Any] = { + 'pageNo': page_no, + 'pageSize': page_size, + 'time': time + } + + return await self._get_records( + STATION_DAY_ENERGY_LIST, key_id, secret, params) + + async def station_month_energy_list( + self, key_id: str, secret: bytes, /, *, + page_no: int = 1, + page_size: int = 20, + month: str + ) -> dict[str, str]: + """Batch acquire station monthly generation""" + + if page_size > 100: + raise SoliscloudError(PAGE_SIZE_ERR) + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.MONTH, month) + params: dict[str, Any] = { + 'pageNo': page_no, + 'pageSize': page_size, + 'time': month + } + + return await self._get_records( + STATION_MONTH_ENERGY_LIST, key_id, secret, params) + + async def station_year_energy_list( + self, key_id: str, secret: bytes, /, *, + page_no: int = 1, + page_size: int = 20, + year: str + ) -> dict[str, str]: + """Batch acquire station yearly generation""" + + if page_size > 100: + raise SoliscloudError(PAGE_SIZE_ERR) + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.YEAR, year) + params: dict[str, Any] = { + 'pageNo': page_no, + 'pageSize': page_size, + 'time': year + } + + return await self._get_records( + STATION_YEAR_ENERGY_LIST, key_id, secret, params) + + async def epm_list( + self, key_id: str, secret: bytes, /, *, + page_no: int = 1, + page_size: int = 20, + station_id: str = None + ) -> dict[str, str]: + """EPM list""" + + if page_size > 100: + raise SoliscloudError(PAGE_SIZE_ERR) + params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} + if station_id is not None: + params['stationId'] = station_id + + return await self._get_records(EPM_LIST, key_id, secret, params) + + async def epm_detail( + self, key_id: str, secret: bytes, /, *, + epm_sn: str + ) -> dict[str, str]: + """EPM details""" + + params: dict[str, Any] = {'sn': epm_sn} + + return await self._get_data(EPM_DETAIL, key_id, secret, params) + + async def epm_day( + self, key_id: str, secret: bytes, /, *, + searchinfo: str, + epm_sn: str, + time: str, + time_zone: int + ) -> dict[str, str]: + """EPM daily graph""" + + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.DAY, time) + params: dict[str, Any] = { + 'searchinfo': searchinfo, + 'sn': epm_sn, + 'time': time, + 'timezone': time_zone} + + return await self._get_data(EPM_DAY, key_id, secret, params) + + async def epm_month( + self, key_id: str, secret: bytes, /, *, + epm_sn: str, + month: str, + ) -> dict[str, str]: + """EPM monthly graph""" + + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.MONTH, month) + params: dict[str, Any] = {'sn': epm_sn, 'month': month} + + return await self._get_data(EPM_MONTH, key_id, secret, params) + + async def epm_year( + self, key_id: str, secret: bytes, /, *, + epm_sn: str, + year: str + ) -> dict[str, str]: + """EPM yearly graph""" + + SoliscloudAPI._verify_date(SoliscloudAPI.DateFormat.YEAR, year) + params: dict[str, Any] = {'sn': epm_sn, 'year': year} + + return await self._get_data(EPM_YEAR, key_id, secret, params) + + async def epm_all( + self, key_id: str, secret: bytes, /, *, + epm_sn: str + ) -> dict[str, str]: + """EPM cumulative graph""" + + params: dict[str, Any] = {'sn': epm_sn} + + return await self._get_data(EPM_ALL, key_id, secret, params) + + async def weather_list( + self, key_id: str, secret: bytes, /, *, + page_no: int = 1, + page_size: int = 20, + station_id: str = None, + nmi_code: str = None + ) -> dict[str, str]: + """Weather list""" + + if page_size > 100: + raise SoliscloudError(PAGE_SIZE_ERR) + + params: dict[str, Any] = {'pageNo': page_no, 'pageSize': page_size} + if station_id is not None: + # If not specified all inverters for all stations for key_id are + # returned + params['stationId'] = station_id + if nmi_code is not None: + params['nmiCode'] = nmi_code + return await self._get_records(WEATHER_LIST, key_id, secret, params) + + async def weather_detail( + self, key_id: str, secret: bytes, /, *, + instrument_sn: str = None + ) -> dict[str, str]: + """Inverter details""" + + params: dict[str, Any] = {} + if instrument_sn is None: + raise SoliscloudError(WEATHER_SN_ERR) + params['sn'] = instrument_sn + + return await self._get_data(WEATHER_DETAIL, key_id, secret, params) + + async def _get_records( + self, canonicalized_resource: str, key_id: str, secret: bytes, + params: dict[str, Any] + ): + """ + Return all records from call + """ + + header: dict[str, str] = SoliscloudAPI._prepare_header( + key_id, secret, params, canonicalized_resource) + + url = f"{self.domain}{canonicalized_resource}" + try: + result = await self._post_data_json(url, header, params) + if 'page' in result.keys(): + return result['page']['records'] + else: + return result['records'] + except KeyError as err: + raise SoliscloudApiError("Malformed data", result) from err + + async def _get_data( + self, canonicalized_resource: str, key_id: str, secret: bytes, + params: dict[str, Any] + ): + """ + Return data from call + """ + + header: dict[str, str] = SoliscloudAPI._prepare_header( + key_id, secret, params, canonicalized_resource) + + url = f"{self.domain}{canonicalized_resource}" + result = await self._post_data_json(url, header, params) + + return result + + @staticmethod + def _now() -> datetime.datetime: + return datetime.now(timezone.utc) + + @staticmethod + def _prepare_header( + key_id: str, + secret: bytes, + body: dict[str, str], + canonicalized_resource: str + ) -> dict[str, str]: + content_md5 = base64.b64encode( + hashlib.md5(json.dumps(body, separators=( + ",", ":")).encode('utf-8')).digest() + ).decode('utf-8') + + content_type = "application/json" + + date = SoliscloudAPI._now().strftime("%a, %d %b %Y %H:%M:%S GMT") + + encrypt_str = ( + VERB + "\n" + + content_md5 + "\n" + + content_type + "\n" + + date + "\n" + + canonicalized_resource + ) + hmac_obj = hmac.new( + secret, + msg=encrypt_str.encode('utf-8'), + digestmod=hashlib.sha1 + ) + sign = base64.b64encode(hmac_obj.digest()) + authorization = "API " + key_id + ":" + sign.decode('utf-8') + + header: dict[str, str] = { + "Content-MD5": content_md5, + "Content-Type": content_type, + "Date": date, + "Authorization": authorization + } + return header + + @throttle(rate_limit=2, period=1.0) + async def _post_data_json( + self, + url: str, + header: dict[str, Any], + params: dict[str, Any] + ) -> dict[str, Any]: + """ Http-post data to specified domain/canonicalized_resource. """ + + resp = None + result = None + if self._session is None: + raise SoliscloudError( + "aiohttp.ClientSession not set") + try: + async with async_timeout.timeout(10): + resp = await SoliscloudAPI._do_post_aiohttp( + self._session, url, params, header) + + result = await resp.json() + if resp.status == HTTPStatus.OK: + if result['code'] != '0': + raise SoliscloudApiError( + result['msg'], result['code']) + return result['data'] + else: + raise SoliscloudHttpError(resp.status) + except asyncio.TimeoutError as err: + if resp is not None: + await resp.release() + raise SoliscloudTimeoutError() from err + except ClientError as err: + if resp is not None: + await resp.release() + raise SoliscloudApiError(err) + except (KeyError, TypeError) as err: + raise SoliscloudApiError( + "Malformed server response", response=result) from err + + @staticmethod + async def _do_post_aiohttp( + session, + url: str, + params: dict[str, Any], + header: dict[str, Any] + ) -> dict[str, Any]: + """ Allows mocking for unit tests.""" + return await session.post(url, json=params, headers=header) + + @staticmethod + def _verify_date(format: SoliscloudAPI.DateFormat, date: str): + rex = re.compile("^[0-9]{4}-[0-9]{2}-[0-9]{2}$") + err = SoliscloudError( + "time must be in format YYYY-MM-DD") + if format == SoliscloudAPI.DateFormat.MONTH: + rex = re.compile("^[0-9]{4}-[0-9]{2}$") + err = SoliscloudError( + "month must be in format YYYY-MM") + elif format == SoliscloudAPI.DateFormat.YEAR: + rex = re.compile("^[0-9]{4}$") + err = SoliscloudError("year must be in format YYYY") + if not rex.match(date): + raise err + return diff --git a/soliscloud_api/entities.py b/soliscloud_api/entities.py new file mode 100644 index 0000000..1df1a4b --- /dev/null +++ b/soliscloud_api/entities.py @@ -0,0 +1,414 @@ +from __future__ import annotations + +from typing import Any +from datetime import datetime +from aiohttp import ClientSession + +from soliscloud_api import SoliscloudAPI, SoliscloudError +from soliscloud_api.types import EntityType, SolisDataFactory + +WHITELIST = ( + 'a', + 'b', + 'c' +) + + +def _normalize_to_list(data): + """Ensure input is always a list of dicts.""" + if isinstance(data, dict): + return [data] + return data + + +class SolisEntity(object): + def __init__( + self, + type: EntityType, + data: dict[str, Any] = None, + whitelist: list[str] = None + ): + self._data = None + if data is not None: + self._data = SolisDataFactory.create(type, data) + self._whitelist = whitelist + + def __str__(self) -> str: + out = f"timestamp data: {self.data_timestamp}" + return out + + def __eq__(self, other): + if isinstance(other, self.__class__): + return self._data == other._data + return NotImplemented + + @property + def data(self) -> dict: + return self._data + + @property + def data_timestamp(self) -> datetime: + return datetime.fromtimestamp(int(self._data['data_timestamp']) / 1e3) + + @classmethod + def initialize_from_data( + cls, + data: dict[str, Any] | list[dict[str, Any]], + /, **filters + ) -> list["SolisEntity"]: + """ + Generic method to parse data and return entity objects. + Filters can be passed as keyword arguments (e.g., plantId=..., id=...). + """ + return cls._do_initialize_from_data(data, **filters) + + @classmethod + def _do_initialize_from_data( + cls, + data: dict[str, Any] | list[dict[str, Any]], + /, **filters + ) -> list["SolisEntity"]: + + items = [] + data_list = _normalize_to_list(data) + for record in data_list: + if not isinstance(record, dict): + raise ValueError(f"Each {cls.__name__} record must be a dict") + if 'id' not in record: + raise ValueError(f"{cls.__name__} record missing 'id'") + match = True + for key, value in filters.items(): + # Map filter keys to record keys if needed + if value is not None and record.get(key) != value: + match = False + break + if match: + items.append(cls(record)) + return items + + +class Collector(SolisEntity): + def __init__(self, data: dict[str, Any]) -> None: + if not isinstance(data, dict): + raise ValueError("Collector data must be a dict") + if 'id' not in data: + raise ValueError("Collector data must include 'id'") + super().__init__(EntityType.COLLECTOR, data) + + def __str__(self): + out = f"collector id: {self._data['id']}\n" + return out + + @property + def collector_id(self): + '''Alias for 'id' attribute ''' + try: + return self._data['id'] + except AttributeError: + return None + + @classmethod + async def initialize_from_session( + cls, + session: ClientSession, + key_id: str, + secret: bytes, + /, *, + plant_id: int = None, + collector_sn: str = None, + collector_id: int = None + ) -> list[Collector]: + """Uses SolisCloudAPI to create Collector objects + Will return objects for all collectors under account if plant_id + is specified + Will return a list containing the Collector object(s) for collector_id + or collector_sn if specified. + First evaluates plant_id, then collector_id, then collector_sn. + + Args: + session (ClientSession): _description_ + key_id (str): _description_ + secret (bytes): _description_ + plant_id (int, optional): _description_. Defaults to None. + collector_sn (str, optional): _description_. Defaults to None. + collector_id (int, optional): _description_. Defaults to None. + + Returns: + list[Collector]: Returns a list of zero or more Collector objects + """ + collectors: list[Collector] = [] + try: + soliscloud = SoliscloudAPI( + 'https://soliscloud.com:13333', session) + collector_data = [] + if plant_id is not None: + collector_data = await soliscloud.collector_list( + key_id, secret, + station_id=plant_id) + elif collector_id is not None: + collector_data.append(await soliscloud.collector_detail( + key_id, secret, + collector_id=collector_id)) + elif collector_sn is not None: + collector_data.append(await soliscloud.collector_detail( + key_id, secret, + collector_sn=collector_sn)) + else: + return collectors + for record in collector_data: + if plant_id is None or record['stationId'] == plant_id: + collector = cls(record) + collectors.append(collector) + except SoliscloudError: + pass + return collectors + + @classmethod + def initialize_from_data( + cls, + data: dict[str, Any] | list[dict[str, Any]], + /, *, + collector_id: str = None, + station_id: str = None + ) -> list[Collector]: + """Parse data retrieved from collector_detail or collector_list + and return Collector objects. + collector_id can be used to filter. + + Args: + data (dict[str, Any] | list[dict[str, Any]]): + response from collector_detail() or collector_list() call. + collector_id (str, optional): _description_. Defaults to None. + + Returns: + list[Collector]: list of zero or more Collector objects + """ + return cls._do_initialize_from_data( + data, id=collector_id, stationId=station_id) + + +class Inverter(SolisEntity): + def __init__(self, data: dict[str, Any]) -> None: + if not isinstance(data, dict): + raise ValueError("Inverter data must be a dict") + if 'id' not in data: + raise ValueError("Inverter data must include 'id'") + super().__init__(EntityType.INVERTER, data) + self._collector = None + + def add_collector(self, collector: Collector) -> None: + self._collector = collector + + def __str__(self): + out = f" inverter id: {self._data['id']}, " + out += f"collector id: {self._collector.collector_id}" + return out + + @property + def inverter_id(self): + '''Alias for 'id' attribute ''' + try: + return self._data['id'] + except AttributeError: + return None + + @classmethod + async def initialize_from_session( + cls, + session: ClientSession, + key_id: str, + secret: bytes, + /, *, + plant_id: int = None, + inverter_id: int = None + ) -> list[Inverter]: + """Uses SolisCloudAPI to create Inverter objects + Works recursive and will also create underlying Collector objects. + plant_id and inverter_id can be used to filter. + Will return objects for all inverters under account if no plant_id + or inverter_id is specified + + Args: + session (ClientSession): http session to use + key_id (str): API key + secret (bytes): API secret + plant_id (int, optional): + Only inverters for matching plant_id will be created. + Defaults to None. + inverter_id (int, optional): + Only inverters for matching inverter_id will be created. + Defaults to None. + + Returns: + list[Inverter]: list of zero or more Inverter objects + """ + inverters: list[Inverter] = [] + try: + soliscloud = SoliscloudAPI( + 'https://soliscloud.com:13333', session) + inverter_data = [] + if inverter_id is not None: + inverter_data.append(await soliscloud.inverter_detail( + key_id, secret, + inverter_id=inverter_id)) + else: + inverter_data = await soliscloud.inverter_detail_list( + key_id, secret) + for record in inverter_data: + if inverter_id is None or record['id'] == inverter_id: + if plant_id is None or record['stationId'] == plant_id: + inverter = cls(record) + collectors = await Collector.initialize_from_session( + session, key_id, secret, + collector_id=inverter.data["collector_id"]) + inverter.add_collector(collectors[0]) + inverters.append(inverter) + except SoliscloudError: + pass + return inverters + + @classmethod + def initialize_from_data( + cls, + data: dict[str, Any] | list[dict[str, Any]], + /, *, + plant_id: int = None, + inverter_id: int = None, + id: int = None + ) -> list[Inverter]: + """Parse data retrieved from inverter_detail or inverter_detail_list + and return Inverter objects. + plant_id and inverter_id can be used to filter. + + Args: + data (dict[str, Any] | list[dict[str, Any]]): + response from inverter_detail() or inverter_detail_list() call. + plant_id (int, optional): + Only inverters for matching plant_id will be created. + Defaults to None. + inverter_id (int, optional): + Only inverters for matching inverter_id will be created. + Defaults to None. + Returns: + list[Inverter]: list of zero or more Inverter objects + """ + return cls._do_initialize_from_data( + data, stationId=plant_id, id=inverter_id) + + +class Plant(SolisEntity): + def __init__(self, data: dict[str, Any]) -> None: + if not isinstance(data, dict): + raise ValueError("Plant data must be a dict") + if 'id' not in data: + raise ValueError("Plant data must include 'id'") + super().__init__(EntityType.PLANT, data) + self._inverters: list[Inverter] = [] + + def __str__(self): + out = f"plant id: {self.plant_id}\n" + out += super().__str__() + out += "\ninverters: [\n" + str(*self.inverters) + "\n]\n" + if out[-4:] == "\n\n]\n": + out = out[:-4] + "]\n" + return out + + def add_inverters(self, inverters: list[Inverter]) -> None: + self._inverters = inverters + + def add_inverter(self, inverter: Inverter) -> None: + self._inverters.append(inverter) + + @property + def plant_id(self): + '''Alias for 'id' attribute ''' + try: + return self._data['id'] + except AttributeError: + return None + + @property + def inverters(self) -> list[Inverter]: + return self._inverters + + @classmethod + async def initialize_from_session( + cls, + session: ClientSession, + key_id: str, secret: bytes, + plant_id: int = None + ) -> list[Plant]: + """Uses SoliscloudAPI to create Plant objects + Works recursive and will also create underlying Inverter objects and + their Collector objects, building a whole object tree. + If no plant_id is specified then Plant objects will be created for all + plants under account. + + Args: + session (ClientSession): http session to use + key_id (str): API key + secret (bytes): API secret + plant_id (int, optional): plant ID to get data from. + Defaults to None. + + Returns: + list[Plant]: list of zero or more Plant objects + """ + + plants: list[Plant] = [] + try: + plants: list[Plant] = [] + plant_data: list[dict[str, Any]] = [] + soliscloud = SoliscloudAPI( + 'https://soliscloud.com:13333', session) + if plant_id is None: + plant_data = await soliscloud.station_detail_list( + key_id, secret) + else: + plant_data = [await soliscloud.station_detail( + key_id, secret, + station_id=plant_id)] + + for plant_record in plant_data: + plant = cls(plant_record) + plant.add_inverters(await Inverter.initialize_from_session( + session, key_id, secret, + plant_id=plant.plant_id)) + + plants.append(plant) + except SoliscloudError: + pass + return plants + + @classmethod + def initialize_from_data( + cls, + data: dict[str, Any] | list[dict[str, Any]], + /, *, + plant_id: int = None + ) -> list[Plant]: + """Parse data retrieved from stationDetail or stationDetailList and + return Plant objects + + Args: + data (dict[str, Any] | list[dict[str, Any]]): + response from station_detail_list() or station_detail() call + plant_id (int, optional): plant id to use. + Defaults to None. If not given a plant is created for all + records in the input data + + Returns: + list[Plant]: A list of zero or more Plant objects + """ + cls._do_initialize_from_data(data, id=plant_id) +# plants: list[Plant] = [] +# plant_data = _normalize_to_list(data) +# for record in plant_data: +# if not isinstance(record, dict): +# raise ValueError("Each plant/station record must be a dict") +# if 'id' not in record: +# raise ValueError("Plant/station record missing 'id'") +# if plant_id is None or record['id'] == plant_id: +# plant = cls(record) +# plants.append(plant) +# return plants diff --git a/soliscloud_api/types.py b/soliscloud_api/types.py new file mode 100644 index 0000000..c2e1f22 --- /dev/null +++ b/soliscloud_api/types.py @@ -0,0 +1,626 @@ +from __future__ import annotations + +import re + +from abc import ABC, abstractmethod +from enum import IntEnum +from collections import UserDict, UserList +from typing import Any +from datetime import datetime, tzinfo, timezone, timedelta + + +class EntityType(IntEnum): + PLANT = 1 + INVERTER = 2 + COLLECTOR = 3 + + +class State(IntEnum): + ONLINE = 1 + OFFLINE = 2 + ALARM = 3 + + +class InverterOfflineState(IntEnum): + NORMAL_OFFLINE = 0 + ABNORMAL_OFFLINE = 1 + + +class InverterType(IntEnum): + GRID = 1 + STORAGE = 2 + + +class PlantType(IntEnum): + GRID = 0 + ENERGY_STORAGE = 1 + AC_COUPLE = 2 + EPM = 3 + BUILT_IN_METER = 4 + EXTERNAL_METER = 5 + S5_OFFLINE_AND_PARALLEL_STORAGE = 6 + S5_GRID_AND_PARALLEL_STORAGE = 7 + GRID_AND_AC_COUPLE = 8 + OFFGRID_STORAGE = 9 + S6_GRID_AND_PARALLEL_STORAGE = 10 + S6_OFFLINE_AND_PARALLEL_STORAGE = 11 + + +class CollectorState(IntEnum): + ONLINE = 1 + OFFLINE = 2 + + +class UnitError(Exception): + """ + Exception raised for wrong unit assignment. + """ + + def __init__(self, unit, allowed_units): + + self.message = f"{unit} is invalid, allowed: {allowed_units}" + super().__init__(self.message) + + +class DimensionedType(UserDict, ABC): + + def __init__(self, value, unit=None): + try: + if unit not in self.__class__.UNITS: + raise UnitError(unit, self.__class__.UNITS) + except AttributeError: + pass + super().__init__({'value': value, 'unit': unit}) + + def __setitem__(self, key, value): + super().__setitem__(key, value) + + def __str__(self): + s = f"{self['value']}" + if self['unit'] is not None: + s += f" {self['unit']}" + return s + f" ({self.__class__.__name__})" + + def __eq__(self, other): + if isinstance(other, self.__class__): + return self.value == other.value and self.unit == other.unit + else: + return False + + def original(self): + d = {} + try: + d = { + 'value': self['original_value'], + 'unit': self['original_unit'] + } + except AttributeError: + d = {'value': self['value'], 'unit': self['unit']} + return d + + @property + def original_value(self): + try: + return self['original_value'] + except AttributeError: + return self['value'] + + @property + def original_unit(self): + try: + return self['original_unit'] + except AttributeError: + return self['unit'] + + @property + def value(self): + return self['value'] + + @property + def unit(self): + return self['unit'] + + @abstractmethod + def _normalize(self): + """ Normalize to default unit """ + + +class GenericType(DimensionedType): + def __init__(self, value, unit): + super().__init__(value, unit) + self._normalize() + + def _normalize(self): + pass + + +class EnergyType(DimensionedType): + UNITS = ('Wh', 'kWh', 'MWh') + DEFAULT = UNITS[1] + + def __init__(self, value, unit): + super().__init__(value, unit) + self._normalize() + + @property + def value_mwh(self): + return self['value']/1000 + + @property + def value_wh(self): + return self['value']*1000 + + def _normalize(self): + if self['unit'] == EnergyType.UNITS[1]: + return + self['original_value'] = self['value'] + self['original_unit'] = self['unit'] + if self['unit'] == EnergyType.UNITS[0]: + self['value'] = self['value']/1000 + elif self['unit'] == EnergyType.UNITS[2]: + self['value'] = self['value']*1000 + else: + pass + self['unit'] = EnergyType.UNITS[1] + + +class VoltageType(DimensionedType): + UNITS = ('V', 'kV') + DEFAULT = UNITS[0] + + def __init__(self, value, unit): + super().__init__(value, unit) + self._normalize() + + @property + def value_kv(self): + return self['value']/1000 + + def _normalize(self): + if self['unit'] == VoltageType.UNITS[0]: + return + self['original_value'] = self['value'] + self['original_unit'] = self['unit'] + if self['unit'] == VoltageType.UNITS[1]: + self['value'] = self['value']*1000 + else: + pass + self['unit'] = VoltageType.UNITS[0] + + +class CurrentType(DimensionedType): + UNITS = ('mA', 'A', 'kA') + DEFAULT = UNITS[1] + + def __init__(self, value, unit): + super().__init__(value, unit) + + @property + def value_a(self): + return self['value'] + + @property + def value_ka(self): + return self['value']/1000 + + def _normalize(self): + if self['unit'] == CurrentType.UNITS[1]: + return + self['original_value'] = self['value'] + self['original_unit'] = self['unit'] + if self['unit'] == CurrentType.UNITS[2]: + self['value'] = self['value']*1000 + elif self['unit'] == CurrentType.UNITS[0]: + self['value'] = self['value']/1000 + else: + pass + self['unit'] = CurrentType.UNITS[1] + + +class PowerType(DimensionedType): + UNITS = ['mW', 'W', 'kW'] + DEFAULT = UNITS[1] + + def __init__(self, value, unit): + super().__init__(value, unit) + self._normalize() + + @property + def value_w(self): + return self['value'] + + @property + def value_kw(self): + return self['value']/1000 + + def _normalize(self): + if self['unit'] == PowerType.UNITS[1]: + return + self['original_value'] = self['value'] + self['original_unit'] = self['unit'] + if self['unit'] == PowerType.UNITS[2]: + self['value'] = self['value']*1000 + elif self['unit'] == PowerType.UNITS[0]: + self['value'] = self['value']/1000 + else: + pass + self['unit'] = PowerType.UNITS[1] + + +class ReactivePowerType(DimensionedType): + UNITS = ('var', 'kvar') + DEFAULT = UNITS[0] + + def __init__(self, value, unit): + super().__init__(value, unit) + self._normalize() + + def _normalize(self): + if self['unit'] == ReactivePowerType.UNITS[0]: + return + self['original_value'] = self['value'] + self['original_unit'] = self['unit'] + if self['unit'] == ReactivePowerType.UNITS[1]: + self['value'] = self['value']*1000 + else: + pass + self['unit'] = ReactivePowerType.UNITS[0] + + +class ApparentPowerType(DimensionedType): + UNITS = ('VA', 'kVA') + DEFAULT = UNITS[0] + + def __init__(self, value, unit): + super().__init__(value, unit) + self._normalize() + + def _normalize(self): + if self['unit'] == ApparentPowerType.UNITS[0]: + return + self['original_value'] = self['value'] + self['original_unit'] = self['unit'] + if self['unit'] == ApparentPowerType.UNITS[1]: + self['value'] = self['value']*1000 + else: + pass + self['unit'] = ApparentPowerType.UNITS[0] + + +class FrequencyType(DimensionedType): + UNITS = ('Hz', 'kHz') + DEFAULT = UNITS[0] + + def __init__(self, value, unit): + super().__init__(value, unit) + self._normalize() + + @property + def value_hz(self): + return self['value'] + + def _normalize(self): + if self['unit'] == FrequencyType.UNITS[0]: + return + self['original_value'] = self['value'] + self['original_unit'] = self['unit'] + if self['unit'] == FrequencyType.UNITS[1]: + self['value'] = self['value']*1000 + else: + pass + self['unit'] = FrequencyType.UNITS[0] + + +class TemperatureType(DimensionedType): + UNITS = ('℃', 'F', 'K') + DEFAULT = UNITS[0] + + def __init__(self, value, unit): + super().__init__(value, unit) + self._normalize() + + @property + def value_celsius(self): + return self['value'] + + @property + def value_fahrenheit(self): + return (float(self['value']) * 9 / 5 + 32) + + @property + def value_kelvin(self): + return float(self['value'] + 273.15) + + def _normalize(self): + if self['unit'] == TemperatureType.UNITS[0]: + return + self['original_value'] = self['value'] + self['original_unit'] = self['unit'] + if self['unit'] == TemperatureType.UNITS[1]: + self['value'] = float(self['value'] - 32) * 5 / 9 + elif self['unit'] == TemperatureType.UNITS[2]: + self['value'] = float(self['value']) - 273.15 + + self['unit'] = TemperatureType.UNITS[0] + + +class CurrencyType(DimensionedType): + def __init__(self, value, unit): + super().__init__(value, unit) + self._normalize() + + def _normalize(self): + pass + + +class DateTimeType(): + def __init__(self, value, t_z: tzinfo = None): + self._timestamp = value + self._datetime = datetime.fromtimestamp(int(value) / 1e3, t_z) + + def date(self): + return self._datetime.date + + def time(self): + return self._datetime.time + + def timestamp(self): + return self._datetime.timestamp + + def timezone(self): + return self._datetime.tzname + + def __str__(self): + return str(self._datetime) + f" ({self.__class__.__name__})" + + def __eq__(self, other): + if isinstance(other, self.__class__): + return self._datetime == other._datetime + else: + return False + + +class EnumType(UserDict): + + def __init__(self, value: IntEnum): + if isinstance(value, IntEnum): + super().__init__( + {'value': value.value, 'name': value.name.title()}) + else: + raise TypeError(f"{value} not of type IntEnum") + + +class ListType(UserList): + def __init__(self, value): + if type(value) is list: + super().__init__(value) + self._normalize() + else: + raise TypeError(f"{value} not of type list") + + def __str__(self): + out = '[' + for p in self.data: + out += '\n' + r = f"{p}" + # indent all lines with 2 extra spaces + r = re.sub(r'(?<=^)(.)', r' \1', r, flags=re.MULTILINE) + out += r + ',' + # Remove trailing comma and newline from last element in list + if out[-1:] == ',': + out = out[0:-1] + '\n' + out += ']' + return out + + def _normalize(self): + for d in self.data: + if '_normalize' in dir(d): + d._normalize() + + +class DictType(UserDict): + def __str__(self) -> str: + out = "{\n" + for k in self.keys(): + out += f" {k}: {self[k]}\n" + out += "}" + return out + + +class SolisDataFactory: + """ Convert input dict from API into dict of values + Keys get normalized to snake_case + Dimensioned values get parsed, converted and/or normalized + Unused k/v pairs get dropped + Values may be again be dict or list + """ + + @staticmethod + def create(type: EntityType, input: dict[str, Any]) -> dict[str, Any]: + data: DictType = DictType() + keys = input.keys() + pattern = re.compile( + r""" + (? 1: + unit = s[-1] + except KeyError: + pass + try: + nr_of_strings = input['dcInputtype'] + 1 + except KeyError: + pass + d = SolisDataFactory._create_value( + type, new_key, value, unit) + # only create properties for available strings + if (new_key[0:4] in ['i_pv', 'u_pv', 'pow_', 'mppt']): + # find number at end of key string + r = re.findall(r'\d+', new_key) + # and dropping for string numbers not present + if len(r) > 0 and int(r[-1]) > nr_of_strings: + d = None + # print(f"Dropping: {new_key}") + if d is not None: + data[new_key] = d + + return data + + @staticmethod + def _key_is_unit(key: str) -> bool: + is_unit = False + is_unit |= key[-3:] == 'Str' + is_unit |= re.search('unit', key, re.IGNORECASE) is not None + return is_unit + + @staticmethod + def _create_value( + type: EntityType, + key: str, + value: Any, + unit: str = None + ) -> Any: + p = None + if isinstance(value, dict): + p = SolisDataFactory.create(type, value) + elif isinstance(value, list): + p = ListType([ + SolisDataFactory._create_value(type, key, x) for x in value + ]) + else: + p = SolisDataFactory._create_typed_value(type, key, value, unit) + return p + + @staticmethod + def _create_typed_value( + type: EntityType, + key: str, + value: Any, + unit: str = None + ) -> Any: + p = None + if unit is not None: + re_datetime = '_time$|_timestamp$|__date$' + match(unit): + case unit if unit in EnergyType.UNITS: + p = EnergyType(value, unit) + case unit if unit in VoltageType.UNITS: + p = VoltageType(value, unit) + case unit if unit in CurrentType.UNITS: + p = CurrentType(value, unit) + case unit if unit in PowerType.UNITS: + p = PowerType(value, unit) + case unit if unit in ReactivePowerType.UNITS: + p = ReactivePowerType(value, unit.lower()) + case unit if unit in ApparentPowerType.UNITS: + p = ApparentPowerType(value, unit) + case unit if unit in FrequencyType.UNITS: + p = FrequencyType(value, unit) + case _ if key[-11:] == 'temperature': + p = TemperatureType(value, unit) + case _ if re.search('_income', key, re.IGNORECASE) is not None: + p = CurrencyType(value, unit) + case _ if re.search(re_datetime, key, re.IGNORECASE)\ + is not None: + try: + regex = '.*UTC([+\-]?)(\d{2}):(\d{2})' # noqa: W605 + sign, hours, minutes = re.match(regex, unit).groups() + sign = -1 if sign == '-' else 1 + hours, minutes = int(hours), int(minutes) + tzinfo = timezone(sign * timedelta(hours=hours, minutes=minutes)) # noqa: E501 + p = DateTimeType(value, tzinfo) + except AttributeError: + p = DateTimeType(value) + case _: + p = GenericType(value, unit) + else: + match(key): + case 'state': + p = EnumType(State(value)) + case 'state_exception_flag': + p = EnumType(InverterOfflineState(value)) + case 'type': + match(type): + case EntityType.PLANT: + p = EnumType(PlantType(value)) + case EntityType.INVERTER: + p = EnumType(InverterType(value)) + case _: + p = value + case _ if re.search('_pec', key, re.IGNORECASE) is not None: + p = value + case _ if re.search('voltage', key, re.IGNORECASE) is not None: + p = VoltageType(value, VoltageType.DEFAULT) + case _ if re.search('upv', key, re.IGNORECASE) is not None: + p = VoltageType(value, VoltageType.DEFAULT) + case _ if key in ('u_a', 'u_b', 'u_c'): + p = VoltageType(value, VoltageType.DEFAULT) + case _ if re.search('current', key, re.IGNORECASE) is not None: + p = CurrentType(value, CurrentType.DEFAULT) + case _ if re.search('ipv', key, re.IGNORECASE) is not None: + p = CurrentType(value, CurrentType.DEFAULT) + case _ if key in ('i_a', 'i_b', 'i_c'): + p = CurrentType(value, CurrentType.DEFAULT) + case _ if key in ('p_a', 'p_b', 'p_c'): + p = PowerType(value, PowerType.DEFAULT) + case _ if re.search('looked_power', key, re.IGNORECASE)\ + is not None: + p = ApparentPowerType(value, ApparentPowerType.DEFAULT) + case _ if re.search('reactive_power', key, re.IGNORECASE)\ + is not None: + p = ReactivePowerType(value, ReactivePowerType.DEFAULT) + case _ if re.search(r"power_?\d{1,2}$", key, re.IGNORECASE)\ + is not None: + p = PowerType(value, PowerType.DEFAULT) + case _ if re.search(r"_pow$", key, re.IGNORECASE) is not None: + p = PowerType(value, PowerType.DEFAULT) + case _ if re.search(r"_time$", key, re.IGNORECASE) is not None: + p = DateTimeType(value) + case _ if re.search(r"_timestamp$", key, re.IGNORECASE)\ + is not None: + p = DateTimeType(value) + case _ if re.search(r"_date$", key, re.IGNORECASE) is not None: + p = DateTimeType(value) + case _: + if unit is not None: + p = GenericType(value, unit) + else: + p = value + return p From bffc4563382400dca504cb9cc99bf65ead1e12cd Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Sat, 13 Sep 2025 15:05:10 +0200 Subject: [PATCH 03/14] Fix names --- test/test_exceptions.py | 40 +++++++++++++++++++++------------------- 1 file changed, 21 insertions(+), 19 deletions(-) diff --git a/test/test_exceptions.py b/test/test_exceptions.py index ef40d2c..3c3eb0a 100644 --- a/test/test_exceptions.py +++ b/test/test_exceptions.py @@ -1,5 +1,11 @@ import pytest from soliscloud_api import SoliscloudAPI +from soliscloud_api import ( + SoliscloudError, + SoliscloudApiError, + SoliscloudHttpError, + SoliscloudTimeoutError, +) from datetime import datetime @@ -10,37 +16,33 @@ def api_instance(): def test_soliscloud_error(mocker): - err = SoliscloudAPI.SolisCloudError() - assert f"{err}" == 'SolisCloud API error' - err = SoliscloudAPI.SolisCloudError("TEST") + err = SoliscloudError() + assert f"{err}" == '' + err = SoliscloudError("TEST") assert f"{err}" == 'TEST' def test_api_error(mocker): - err = SoliscloudAPI.ApiError() - assert f"{err}" == 'API returned an error: \ -Undefined API error occurred, error code: Unknown, response: None' - err = SoliscloudAPI.ApiError("TEST") - assert f"{err}" == 'API returned an error: \ -TEST, error code: Unknown, response: None' - err = SoliscloudAPI.ApiError("TEST", 3, 1) - assert f"{err}" == 'API returned an error: \ -TEST, error code: 3, response: 1' + err = SoliscloudApiError() + assert f"{err}" == 'API returned an error: Undefined API error occurred, error code: Unknown, response: None' # noqa: E501 + err = SoliscloudApiError("TEST") + assert f"{err}" == 'API returned an error: TEST, error code: Unknown, response: None' # noqa: E501 + err = SoliscloudApiError("TEST", 3, 1) + assert f"{err}" == 'API returned an error: TEST, error code: 3, response: 1' # noqa: E501 def test_http_error(mocker): - err = SoliscloudAPI.HttpError(408) + err = SoliscloudHttpError(408) now = datetime.now().strftime("%d-%m-%Y %H:%M GMT") - assert f"{err}" == f'Your system time is different from server time, \ -your time is {now}' - err = SoliscloudAPI.HttpError(502) + assert f"{err}" == f'Your system time is different from server time, your time is {now}' # noqa: E501 + err = SoliscloudHttpError(502) assert f"{err}" == 'Http status code: 502' - err = SoliscloudAPI.HttpError(502, "TEST") + err = SoliscloudHttpError(502, "TEST") assert f"{err}" == 'TEST' def test_timeout_error(mocker): - err = SoliscloudAPI.TimeoutError() + err = SoliscloudTimeoutError() assert f"{err}" == 'Timeout error occurred' - err = SoliscloudAPI.TimeoutError("TEST") + err = SoliscloudTimeoutError("TEST") assert f"{err}" == 'TEST' From 4279adc4d3d035c78ff12ff0627ae9a4e5b92d4e Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Sat, 13 Sep 2025 15:06:31 +0200 Subject: [PATCH 04/14] Fix names --- test/test_private_methods.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/test/test_private_methods.py b/test/test_private_methods.py index b6c7b15..d32171e 100644 --- a/test/test_private_methods.py +++ b/test/test_private_methods.py @@ -5,6 +5,7 @@ from datetime import timezone from aiohttp import ClientError from soliscloud_api import SoliscloudAPI +from soliscloud_api import SoliscloudError, SoliscloudApiError from .const import KEY, SECRET, VALID_RESPONSE, VALID_RESPONSE_PAGED_RECORDS VALID_HEADER = { @@ -100,7 +101,7 @@ async def test_post_data_json_fail(api_instance, mocker): mocker.patch( 'soliscloud_api.SoliscloudAPI._do_post_aiohttp', return_value=HTTP_RESPONSE_KEYERROR) - with pytest.raises(SoliscloudAPI.ApiError): + with pytest.raises(SoliscloudApiError): await api_instance._post_data_json( "/TEST", VALID_HEADER, @@ -109,7 +110,7 @@ async def test_post_data_json_fail(api_instance, mocker): 'soliscloud_api.SoliscloudAPI._do_post_aiohttp', return_value=VALID_HTTP_RESPONSE, side_effect=asyncio.TimeoutError) - with pytest.raises(SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance._post_data_json( "/TEST", VALID_HEADER, @@ -118,7 +119,7 @@ async def test_post_data_json_fail(api_instance, mocker): 'soliscloud_api.SoliscloudAPI._do_post_aiohttp', return_value=VALID_HTTP_RESPONSE, side_effect=ClientError) - with pytest.raises(SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance._post_data_json( "/TEST", VALID_HEADER, From 737041a5a589bf6955a85cb88d45ce808e1557d3 Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Sat, 13 Sep 2025 15:09:11 +0200 Subject: [PATCH 05/14] Fix names --- test/test_public_methods.py | 100 ++++++++++++++++++++---------------- 1 file changed, 57 insertions(+), 43 deletions(-) diff --git a/test/test_public_methods.py b/test/test_public_methods.py index daea53f..faa506d 100644 --- a/test/test_public_methods.py +++ b/test/test_public_methods.py @@ -1,7 +1,21 @@ import pytest import soliscloud_api as api +from soliscloud_api import SoliscloudError +from soliscloud_api.client import ( + COLLECTOR_DETAIL, + COLLECTOR_LIST, + COLLECTOR_DAY, + ALARM_LIST, + EPM_LIST, + EPM_DETAIL, + EPM_DAY, + EPM_MONTH, + EPM_YEAR, + EPM_ALL, + WEATHER_LIST, + WEATHER_DETAIL, +) -# from soliscloud_api import * from .const import ( KEY, SECRET, @@ -68,7 +82,7 @@ async def test_collector_list_valid(api_instance, patched_api_paged): result = await api_instance.collector_list(KEY, SECRET) assert result == VALID_RESPONSE_PAGED_RECORDS patched_api_paged._get_records.assert_called_with( - api.COLLECTOR_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20}) + COLLECTOR_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20}) # All arguments filled result = await api_instance.collector_list( @@ -77,7 +91,7 @@ async def test_collector_list_valid(api_instance, patched_api_paged): page_no=4, page_size=100, station_id=1000, nmi_code=NMI) assert result == VALID_RESPONSE_PAGED_RECORDS patched_api_paged._get_records.assert_called_with( - api.COLLECTOR_LIST, + COLLECTOR_LIST, KEY, SECRET, { 'pageNo': 4, @@ -89,7 +103,7 @@ async def test_collector_list_valid(api_instance, patched_api_paged): @pytest.mark.asyncio async def test_collector_list_invalid_page_size(api_instance): - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.collector_list(KEY, SECRET, page_size=101) @@ -100,19 +114,19 @@ async def test_collector_detail_valid(api_instance, patched_api): KEY, SECRET, collector_sn=1000) assert result == VALID_RESPONSE patched_api._get_data.assert_called_with( - api.COLLECTOR_DETAIL, KEY, SECRET, {'sn': 1000}) + COLLECTOR_DETAIL, KEY, SECRET, {'sn': 1000}) result = await api_instance.collector_detail( KEY, SECRET, collector_id='1000') assert result == VALID_RESPONSE patched_api._get_data.assert_called_with( - api.COLLECTOR_DETAIL, KEY, SECRET, {'id': '1000'}) + COLLECTOR_DETAIL, KEY, SECRET, {'id': '1000'}) @pytest.mark.asyncio async def test_collector_detail_invalid_params(api_instance): # ID and SN together - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.collector_detail( KEY, SECRET, collector_sn=1000, collector_id='1000') @@ -126,29 +140,29 @@ async def test_collector_day_valid(api_instance, patched_api_list): collector_sn=1000, time='2023-01-01', time_zone=1) assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.COLLECTOR_DAY, + COLLECTOR_DAY, KEY, SECRET, {'sn': 1000, 'time': '2023-01-01', 'timeZone': 1}) @pytest.mark.asyncio async def test_collector_day_invalid_params(api_instance): - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.collector_day( KEY, SECRET, collector_sn=None, time='2023-01-01', time_zone=1) - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.collector_day( KEY, SECRET, collector_sn='1000', time='2023', time_zone=1) - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.collector_day( KEY, SECRET, collector_sn='1000', time='2023+01-01', time_zone=1) - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.collector_day( KEY, SECRET, collector_sn='1000', time='2023-01+01', time_zone=1) @@ -162,7 +176,7 @@ async def test_alarm_list_valid(api_instance, patched_api_records): station_id='1000', begintime='2022-01-01', endtime='2023-01-01') assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.ALARM_LIST, + ALARM_LIST, KEY, SECRET, { 'pageNo': 1, @@ -176,7 +190,7 @@ async def test_alarm_list_valid(api_instance, patched_api_records): device_sn='1000', begintime='2022-01-01', endtime='2023-01-01') assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.ALARM_LIST, + ALARM_LIST, KEY, SECRET, { 'pageNo': 1, @@ -195,7 +209,7 @@ async def test_alarm_list_valid(api_instance, patched_api_records): nmi_code=NMI) assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.ALARM_LIST, + ALARM_LIST, KEY, SECRET, { 'pageNo': 4, @@ -209,7 +223,7 @@ async def test_alarm_list_valid(api_instance, patched_api_records): @pytest.mark.asyncio async def test_alarm_list_invalid_params(api_instance): # Wrong page size - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.alarm_list( KEY, SECRET, page_size=1000, @@ -217,12 +231,12 @@ async def test_alarm_list_invalid_params(api_instance): begintime='2022-01-01', endtime='2023-01-01') # No Id and no Sn - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.alarm_list( KEY, SECRET, begintime='2022-01-01', endtime='2023-01-01') # Both Id and Sn - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.alarm_list( KEY, SECRET, begintime='2022-01-01', @@ -230,36 +244,36 @@ async def test_alarm_list_invalid_params(api_instance): station_id='1000', device_sn='sn') # Illegal begin time - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.alarm_list( KEY, SECRET, begintime='2022', endtime='2023-01-01', station_id='1000') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.alarm_list( KEY, SECRET, begintime='2022+01-01', endtime='2023-01-01', station_id='1000') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.alarm_list( KEY, SECRET, begintime='2022-01+01', endtime='2023-01-01', station_id='1000') # Illegal end time - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.alarm_list( KEY, SECRET, begintime='2022-01-01', endtime='2023', station_id='1000') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.alarm_list( KEY, SECRET, begintime='2022-01-01', endtime='2023+01-01', station_id='1000') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.alarm_list( KEY, SECRET, begintime='2022-01-01', endtime='2023-01+01', station_id='1000') @@ -271,7 +285,7 @@ async def test_epm_list_valid(api_instance, patched_api_paged): result = await api_instance.epm_list(KEY, SECRET) assert result == VALID_RESPONSE_PAGED_RECORDS patched_api_paged._get_records.assert_called_with( - api.EPM_LIST, + EPM_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20}) @@ -280,7 +294,7 @@ async def test_epm_list_valid(api_instance, patched_api_paged): page_no=4, page_size=30, station_id='1000') assert result == VALID_RESPONSE_PAGED_RECORDS patched_api_paged._get_records.assert_called_with( - api.EPM_LIST, + EPM_LIST, KEY, SECRET, {'pageNo': 4, 'pageSize': 30, 'stationId': '1000'}) @@ -288,7 +302,7 @@ async def test_epm_list_valid(api_instance, patched_api_paged): @pytest.mark.asyncio async def test_epm_list_invalid_params(api_instance): # Wrong page_size - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.epm_list(KEY, SECRET, page_size=1000) @@ -298,7 +312,7 @@ async def test_epm_detail(api_instance, patched_api): result = await api_instance.epm_detail(KEY, SECRET, epm_sn='sn') assert result == VALID_RESPONSE patched_api._get_data.assert_called_with( - api.EPM_DETAIL, + EPM_DETAIL, KEY, SECRET, {'sn': 'sn'}) @@ -311,7 +325,7 @@ async def test_epm_day_valid(api_instance, patched_api): searchinfo='info', epm_sn='sn', time='2023-01-01', time_zone=1) assert result == VALID_RESPONSE patched_api._get_data.assert_called_with( - api.EPM_DAY, + EPM_DAY, KEY, SECRET, { 'searchinfo': 'info', @@ -324,17 +338,17 @@ async def test_epm_day_valid(api_instance, patched_api): @pytest.mark.asyncio async def test_epm_day_invalid_params(api_instance): # Wrong time format - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.epm_day( KEY, SECRET, searchinfo='info', epm_sn='sn', time='2023', time_zone=1) - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.epm_day( KEY, SECRET, searchinfo='info', epm_sn='sn', time='2023+01-01', time_zone=1) - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.epm_day( KEY, SECRET, searchinfo='info', epm_sn='sn', time='2023-01+01', time_zone=1) @@ -347,7 +361,7 @@ async def test_epm_month_valid(api_instance, patched_api_list): KEY, SECRET, epm_sn='sn', month='2023-01') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.EPM_MONTH, + EPM_MONTH, KEY, SECRET, {'sn': 'sn', 'month': '2023-01'}) @@ -355,12 +369,12 @@ async def test_epm_month_valid(api_instance, patched_api_list): @pytest.mark.asyncio async def test_epm_month_invalid_params(api_instance): # Wrong time format - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.epm_month( KEY, SECRET, epm_sn='sn', month='2023') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.epm_month( KEY, SECRET, epm_sn='sn', month='2023+01') @@ -374,7 +388,7 @@ async def test_epm_year_valid(api_instance, patched_api_list): epm_sn='sn', year='2023') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.EPM_YEAR, + EPM_YEAR, KEY, SECRET, {'sn': 'sn', 'year': '2023'}) @@ -382,7 +396,7 @@ async def test_epm_year_valid(api_instance, patched_api_list): @pytest.mark.asyncio async def test_epm_year_invalid_params(api_instance): # Wrong time format - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.epm_year( KEY, SECRET, epm_sn='sn', year='22023') @@ -394,7 +408,7 @@ async def test_epm_all_valid(api_instance, patched_api_list): result = await api_instance.epm_all(KEY, SECRET, epm_sn='sn') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.EPM_ALL, KEY, SECRET, {'sn': 'sn'}) + EPM_ALL, KEY, SECRET, {'sn': 'sn'}) @pytest.mark.asyncio @@ -403,7 +417,7 @@ async def test_weather_list_valid(api_instance, patched_api_paged): result = await api_instance.weather_list(KEY, SECRET) assert result == VALID_RESPONSE_PAGED_RECORDS patched_api_paged._get_records.assert_called_with( - api.WEATHER_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20}) + WEATHER_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20}) # All arguments filled result = await api_instance.weather_list( @@ -411,7 +425,7 @@ async def test_weather_list_valid(api_instance, patched_api_paged): page_no=4, page_size=100, station_id=1000, nmi_code=NMI) assert result == VALID_RESPONSE_PAGED_RECORDS patched_api_paged._get_records.assert_called_with( - api.WEATHER_LIST, + WEATHER_LIST, KEY, SECRET, { 'pageNo': 4, @@ -422,7 +436,7 @@ async def test_weather_list_valid(api_instance, patched_api_paged): @pytest.mark.asyncio async def test_weather_list_invalid_page_size(api_instance): - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.weather_list(KEY, SECRET, page_size=101) @@ -432,12 +446,12 @@ async def test_weather_detail_valid(api_instance, patched_api): result = await api_instance.weather_detail(KEY, SECRET, instrument_sn='sn') assert result == VALID_RESPONSE patched_api._get_data.assert_called_with( - api.WEATHER_DETAIL, KEY, SECRET, {'sn': 'sn'}) + WEATHER_DETAIL, KEY, SECRET, {'sn': 'sn'}) @pytest.mark.asyncio async def test_weather_detail_invalid_params(api_instance): - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.weather_detail( KEY, SECRET, instrument_sn=None) From 3c74e4a6aa6d08f2559a11ac6e2135a4518b8725 Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Sat, 13 Sep 2025 15:10:15 +0200 Subject: [PATCH 06/14] Fix names --- test/test_public_methods_inverter.py | 82 ++++++++++++++++------------ 1 file changed, 46 insertions(+), 36 deletions(-) diff --git a/test/test_public_methods_inverter.py b/test/test_public_methods_inverter.py index 8ffc6e1..2b3cf81 100644 --- a/test/test_public_methods_inverter.py +++ b/test/test_public_methods_inverter.py @@ -1,7 +1,17 @@ import pytest import soliscloud_api as api +from soliscloud_api import SoliscloudError +from soliscloud_api.client import ( + INVERTER_LIST, + INVERTER_DETAIL, + INVERTER_DAY, + INVERTER_MONTH, + INVERTER_YEAR, + INVERTER_ALL, + INVERTER_DETAIL_LIST, + INVERTER_SHELF_TIME +) -# from soliscloud_api import * from .const import ( KEY, SECRET, @@ -68,7 +78,7 @@ async def test_inverter_list_valid(api_instance, patched_api_paged): result = await api_instance.inverter_list(KEY, SECRET) assert result == VALID_RESPONSE_PAGED_RECORDS patched_api_paged._get_records.assert_called_with( - api.INVERTER_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20}) + INVERTER_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20}) # All arguments filled result = await api_instance.inverter_list( @@ -76,7 +86,7 @@ async def test_inverter_list_valid(api_instance, patched_api_paged): page_no=4, page_size=100, station_id=1000, nmi_code=NMI) assert result == VALID_RESPONSE_PAGED_RECORDS patched_api_paged._get_records.assert_called_with( - api.INVERTER_LIST, + INVERTER_LIST, KEY, SECRET, { 'pageNo': 4, @@ -88,7 +98,7 @@ async def test_inverter_list_valid(api_instance, patched_api_paged): @pytest.mark.asyncio async def test_inverter_list_invalid_page_size(api_instance): - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_list(KEY, SECRET, page_size=101) @@ -98,20 +108,20 @@ async def test_inverter_detail_valid(api_instance, patched_api): result = await api_instance.inverter_detail(KEY, SECRET, inverter_sn=1000) assert result == VALID_RESPONSE patched_api._get_data.assert_called_with( - api.INVERTER_DETAIL, KEY, SECRET, {'sn': 1000}) + INVERTER_DETAIL, KEY, SECRET, {'sn': 1000}) result = await api_instance.inverter_detail( KEY, SECRET, inverter_id='1000') assert result == VALID_RESPONSE patched_api._get_data.assert_called_with( - api.INVERTER_DETAIL, KEY, SECRET, {'id': '1000'}) + INVERTER_DETAIL, KEY, SECRET, {'id': '1000'}) @pytest.mark.asyncio async def test_inverter_detail_invalid_params(api_instance): # ID and SN together - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_detail( KEY, SECRET, inverter_sn=1000, inverter_id='1000') @@ -125,7 +135,7 @@ async def test_inverter_day_valid(api_instance, patched_api_list): currency='EUR', time='2023-01-01', time_zone=1, inverter_id='1000') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.INVERTER_DAY, + INVERTER_DAY, KEY, SECRET, {'money': 'EUR', 'time': '2023-01-01', 'timeZone': 1, 'id': '1000'}) @@ -134,7 +144,7 @@ async def test_inverter_day_valid(api_instance, patched_api_list): currency='EUR', time='2023-01-01', time_zone=1, inverter_sn='sn') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.INVERTER_DAY, + INVERTER_DAY, KEY, SECRET, {'money': 'EUR', 'time': '2023-01-01', 'timeZone': 1, 'sn': 'sn'}) @@ -142,12 +152,12 @@ async def test_inverter_day_valid(api_instance, patched_api_list): @pytest.mark.asyncio async def test_inverter_day_invalid_params(api_instance): # ID and SN together - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_day( KEY, SECRET, currency='EUR', time='2023-01-01', time_zone=1) - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_day( KEY, SECRET, currency='EUR', @@ -156,17 +166,17 @@ async def test_inverter_day_invalid_params(api_instance): inverter_id='1000', inverter_sn='sn') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_day( KEY, SECRET, currency='EUR', time='2023', time_zone=1, inverter_id='1000') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_day( KEY, SECRET, currency='EUR', time='2023+01-01', time_zone=1, inverter_id='1000') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_day( KEY, SECRET, currency='EUR', time='2023-01+01', time_zone=1, inverter_id='1000') @@ -180,7 +190,7 @@ async def test_inverter_month_valid(api_instance, patched_api_list): currency='EUR', month='2023-01', inverter_id='1000') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.INVERTER_MONTH, + INVERTER_MONTH, KEY, SECRET, {'money': 'EUR', 'month': '2023-01', 'id': '1000'}) @@ -189,7 +199,7 @@ async def test_inverter_month_valid(api_instance, patched_api_list): currency='EUR', month='2023-01', inverter_sn='sn') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.INVERTER_MONTH, + INVERTER_MONTH, KEY, SECRET, {'money': 'EUR', 'month': '2023-01', 'sn': 'sn'}) @@ -197,12 +207,12 @@ async def test_inverter_month_valid(api_instance, patched_api_list): @pytest.mark.asyncio async def test_inverter_month_invalid_params(api_instance): # ID and SN together - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_month( KEY, SECRET, currency='EUR', month='2023-01') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_month( KEY, SECRET, currency='EUR', @@ -210,12 +220,12 @@ async def test_inverter_month_invalid_params(api_instance): inverter_id='1000', inverter_sn='sn') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_month( KEY, SECRET, currency='EUR', month='2023', inverter_id='1000') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_month( KEY, SECRET, currency='EUR', month='2023+01', inverter_id='1000') @@ -229,7 +239,7 @@ async def test_inverter_year_valid(api_instance, patched_api_list): currency='EUR', year='2023', inverter_id='1000') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.INVERTER_YEAR, + INVERTER_YEAR, KEY, SECRET, {'money': 'EUR', 'year': '2023', 'id': '1000'}) @@ -238,7 +248,7 @@ async def test_inverter_year_valid(api_instance, patched_api_list): currency='EUR', year='2023', inverter_sn='sn') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.INVERTER_YEAR, + INVERTER_YEAR, KEY, SECRET, {'money': 'EUR', 'year': '2023', 'sn': 'sn'}) @@ -246,17 +256,17 @@ async def test_inverter_year_valid(api_instance, patched_api_list): @pytest.mark.asyncio async def test_inverter_year_invalid_params(api_instance): # ID and SN together - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_year( KEY, SECRET, currency='EUR', year='2023') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_year( KEY, SECRET, currency='EUR', year='2023', inverter_id='1000', inverter_sn='sn') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_year( KEY, SECRET, currency='EUR', year='22023', inverter_id='1000') @@ -270,7 +280,7 @@ async def test_inverter_all_valid(api_instance, patched_api_list): currency='EUR', inverter_id='1000') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.INVERTER_ALL, + INVERTER_ALL, KEY, SECRET, {'money': 'EUR', 'id': '1000'}) @@ -279,7 +289,7 @@ async def test_inverter_all_valid(api_instance, patched_api_list): currency='EUR', inverter_sn='sn') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.INVERTER_ALL, + INVERTER_ALL, KEY, SECRET, {'money': 'EUR', 'sn': 'sn'}) @@ -287,10 +297,10 @@ async def test_inverter_all_valid(api_instance, patched_api_list): @pytest.mark.asyncio async def test_inverter_all_invalid_params(api_instance): # ID and SN together - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_all(KEY, SECRET, currency='EUR') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_all( KEY, SECRET, currency='EUR', inverter_id='1000', inverter_sn='sn') @@ -302,14 +312,14 @@ async def test_inverter_detail_list_valid(api_instance, patched_api_records): result = await api_instance.inverter_detail_list(KEY, SECRET) assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.INVERTER_DETAIL_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20}) + INVERTER_DETAIL_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20}) result = await api_instance.inverter_detail_list( KEY, SECRET, page_no=4, page_size=30) assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.INVERTER_DETAIL_LIST, + INVERTER_DETAIL_LIST, KEY, SECRET, {'pageNo': 4, 'pageSize': 30}) @@ -317,7 +327,7 @@ async def test_inverter_detail_list_valid(api_instance, patched_api_records): @pytest.mark.asyncio async def test_inverter_detail_list_invalid_params(api_instance): # Wrong page_size - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_detail_list(KEY, SECRET, page_size=1000) @@ -329,7 +339,7 @@ async def test_inverter_shelf_time(api_instance, patched_api_records): inverter_sn='sn') assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.INVERTER_SHELF_TIME, + INVERTER_SHELF_TIME, KEY, SECRET, {'pageNo': 1, 'pageSize': 20, 'sn': 'sn'}) @@ -339,7 +349,7 @@ async def test_inverter_shelf_time(api_instance, patched_api_records): page_no=50, page_size=50, inverter_sn='sn') assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.INVERTER_SHELF_TIME, + INVERTER_SHELF_TIME, KEY, SECRET, {'pageNo': 50, 'pageSize': 50, 'sn': 'sn'}) @@ -347,11 +357,11 @@ async def test_inverter_shelf_time(api_instance, patched_api_records): @pytest.mark.asyncio async def test_inverter_shelf_time_invalid_params(api_instance): # Wrong page_size - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_shelf_time( KEY, SECRET, page_size=1000, inverter_sn='sn') # Wrong page_size - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.inverter_shelf_time(KEY, SECRET, inverter_sn=None) From 56538d78602bcd8bb01c707e8592841b03032b53 Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Sat, 13 Sep 2025 15:12:04 +0200 Subject: [PATCH 07/14] Fix names --- test/test_public_methods_station.py | 103 +++++++++++++++------------- 1 file changed, 57 insertions(+), 46 deletions(-) diff --git a/test/test_public_methods_station.py b/test/test_public_methods_station.py index 2f4894c..d650c37 100644 --- a/test/test_public_methods_station.py +++ b/test/test_public_methods_station.py @@ -1,7 +1,18 @@ import pytest import soliscloud_api as api +from soliscloud_api import SoliscloudError +from soliscloud_api.client import ( + USER_STATION_LIST, + STATION_DETAIL, + STATION_DAY, + STATION_MONTH, + STATION_YEAR, + STATION_ALL, + STATION_DETAIL_LIST, + STATION_DAY_ENERGY_LIST, + STATION_MONTH_ENERGY_LIST, + STATION_YEAR_ENERGY_LIST) -# from soliscloud_api import * from .const import ( KEY, SECRET, @@ -68,7 +79,7 @@ async def test_user_station_list_valid(api_instance, patched_api_paged): result = await api_instance.user_station_list(KEY, SECRET) assert result == VALID_RESPONSE_PAGED_RECORDS patched_api_paged._get_records.assert_called_with( - api.USER_STATION_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20}) + USER_STATION_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20}) assert result == VALID_RESPONSE_PAGED_RECORDS # All arguments filled @@ -77,14 +88,14 @@ async def test_user_station_list_valid(api_instance, patched_api_paged): page_no=4, page_size=100, nmi_code=NMI) assert result == VALID_RESPONSE_PAGED_RECORDS patched_api_paged._get_records.assert_called_with( - api.USER_STATION_LIST, + USER_STATION_LIST, KEY, SECRET, {'pageNo': 4, 'pageSize': 100, 'nmiCode': 'nmi_code'}) @pytest.mark.asyncio async def test_user_station_list_invalid_page_size(api_instance): - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.user_station_list(KEY, SECRET, page_size=101) @@ -94,7 +105,7 @@ async def test_station_detail_valid(api_instance, patched_api): result = await api_instance.station_detail(KEY, SECRET, station_id=1000) assert result == VALID_RESPONSE patched_api._get_data.assert_called_with( - api.STATION_DETAIL, KEY, SECRET, {'id': 1000}) + STATION_DETAIL, KEY, SECRET, {'id': 1000}) # All arguments filled result = await api_instance.station_detail( @@ -102,7 +113,7 @@ async def test_station_detail_valid(api_instance, patched_api): station_id=1000, nmi_code=NMI) assert result == VALID_RESPONSE patched_api._get_data.assert_called_with( - api.STATION_DETAIL, + STATION_DETAIL, KEY, SECRET, {'id': 1000, 'nmiCode': 'nmi_code'}) @@ -115,7 +126,7 @@ async def test_station_day_valid(api_instance, patched_api_list): currency='EUR', time='2023-01-01', time_zone=1, station_id='1000') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.STATION_DAY, + STATION_DAY, KEY, SECRET, {'money': 'EUR', 'time': '2023-01-01', 'timeZone': 1, 'id': '1000'}) @@ -124,7 +135,7 @@ async def test_station_day_valid(api_instance, patched_api_list): currency='EUR', time='2023-01-01', time_zone=1, nmi_code=NMI) assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.STATION_DAY, + STATION_DAY, KEY, SECRET, {'money': 'EUR', 'time': '2023-01-01', 'timeZone': 1, 'nmiCode': NMI}) @@ -132,12 +143,12 @@ async def test_station_day_valid(api_instance, patched_api_list): @pytest.mark.asyncio async def test_station_day_invalid_params(api_instance): # ID and SN together - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_day( KEY, SECRET, currency='EUR', time='2023-01-01', time_zone=1) - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_day( KEY, SECRET, currency='EUR', @@ -146,17 +157,17 @@ async def test_station_day_invalid_params(api_instance): station_id='1000', nmi_code=NMI) - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_day( KEY, SECRET, currency='EUR', time='2023', time_zone=1, station_id='1000') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_day( KEY, SECRET, currency='EUR', time='2023+01-01', time_zone=1, station_id='1000') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_day( KEY, SECRET, currency='EUR', time='2023-01+01', time_zone=1, station_id='1000') @@ -170,7 +181,7 @@ async def test_station_month_valid(api_instance, patched_api_list): currency='EUR', month='2023-01', station_id='1000') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.STATION_MONTH, + STATION_MONTH, KEY, SECRET, {'money': 'EUR', 'month': '2023-01', 'id': '1000'}) @@ -179,7 +190,7 @@ async def test_station_month_valid(api_instance, patched_api_list): currency='EUR', month='2023-01', nmi_code=NMI) assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.STATION_MONTH, + STATION_MONTH, KEY, SECRET, {'money': 'EUR', 'month': '2023-01', 'nmiCode': NMI}) @@ -187,22 +198,22 @@ async def test_station_month_valid(api_instance, patched_api_list): @pytest.mark.asyncio async def test_station_month_invalid_params(api_instance): # ID and SN together - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_month( KEY, SECRET, currency='EUR', month='2023-01') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_month( KEY, SECRET, currency='EUR', month='2023-01', station_id='1000', nmi_code=NMI) - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_month( KEY, SECRET, currency='EUR', month='2023', station_id='1000') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_month( KEY, SECRET, currency='EUR', month='2023+01', station_id='1000') @@ -216,7 +227,7 @@ async def test_station_year_valid(api_instance, patched_api_list): currency='EUR', year='2023', station_id='1000') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.STATION_YEAR, + STATION_YEAR, KEY, SECRET, {'money': 'EUR', 'year': '2023', 'id': '1000'}) @@ -225,7 +236,7 @@ async def test_station_year_valid(api_instance, patched_api_list): currency='EUR', year='2023', nmi_code=NMI) assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.STATION_YEAR, + STATION_YEAR, KEY, SECRET, {'money': 'EUR', 'year': '2023', 'nmiCode': NMI}) @@ -233,17 +244,17 @@ async def test_station_year_valid(api_instance, patched_api_list): @pytest.mark.asyncio async def test_station_year_invalid_params(api_instance): # ID and SN together - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_year( KEY, SECRET, currency='EUR', year='2023') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_year( KEY, SECRET, currency='EUR', year='2023', station_id='1000', nmi_code=NMI) - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_year( KEY, SECRET, currency='EUR', year='22023', station_id='1000') @@ -257,23 +268,23 @@ async def test_station_all_valid(api_instance, patched_api_list): currency='EUR', station_id='1000') assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.STATION_ALL, KEY, SECRET, {'money': 'EUR', 'id': '1000'}) + STATION_ALL, KEY, SECRET, {'money': 'EUR', 'id': '1000'}) result = await api_instance.station_all( KEY, SECRET, currency='EUR', nmi_code=NMI) assert result == VALID_RESPONSE_LIST patched_api_list._get_data.assert_called_with( - api.STATION_ALL, KEY, SECRET, {'money': 'EUR', 'nmiCode': NMI}) + STATION_ALL, KEY, SECRET, {'money': 'EUR', 'nmiCode': NMI}) @pytest.mark.asyncio async def test_station_all_invalid_params(api_instance): # ID and SN together - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_all(KEY, SECRET, currency='EUR') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_all( KEY, SECRET, currency='EUR', station_id='1000', nmi_code=NMI) @@ -285,7 +296,7 @@ async def test_station_detail_list_valid(api_instance, patched_api_records): result = await api_instance.station_detail_list(KEY, SECRET) assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.STATION_DETAIL_LIST, + STATION_DETAIL_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20}) @@ -294,7 +305,7 @@ async def test_station_detail_list_valid(api_instance, patched_api_records): page_no=4, page_size=30) assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.STATION_DETAIL_LIST, + STATION_DETAIL_LIST, KEY, SECRET, {'pageNo': 4, 'pageSize': 30}) @@ -302,7 +313,7 @@ async def test_station_detail_list_valid(api_instance, patched_api_records): @pytest.mark.asyncio async def test_station_detail_list_invalid_params(api_instance): # Wrong page_size - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_detail_list(KEY, SECRET, page_size=1000) @@ -315,7 +326,7 @@ async def test_station_day_energy_list_valid( time='2023-01-01') assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.STATION_DAY_ENERGY_LIST, + STATION_DAY_ENERGY_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20, 'time': '2023-01-01'}) @@ -324,7 +335,7 @@ async def test_station_day_energy_list_valid( page_no=4, page_size=30, time='2023-01-01') assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.STATION_DAY_ENERGY_LIST, + STATION_DAY_ENERGY_LIST, KEY, SECRET, {'pageNo': 4, 'pageSize': 30, 'time': '2023-01-01'}) @@ -332,22 +343,22 @@ async def test_station_day_energy_list_valid( @pytest.mark.asyncio async def test_station_day_energy_list_invalid_params(api_instance): # Wrong page_size - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_day_energy_list( KEY, SECRET, page_size=1000, time='2023-01-01') # Wrong time format - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_day_energy_list( KEY, SECRET, time='2023') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_day_energy_list( KEY, SECRET, time='2023+01-01') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_day_energy_list( KEY, SECRET, time='2023-01+01') @@ -362,7 +373,7 @@ async def test_station_month_energy_list_valid( month='2023-01') assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.STATION_MONTH_ENERGY_LIST, + STATION_MONTH_ENERGY_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20, 'time': '2023-01'}) @@ -371,7 +382,7 @@ async def test_station_month_energy_list_valid( page_no=4, page_size=30, month='2023-01') assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.STATION_MONTH_ENERGY_LIST, + STATION_MONTH_ENERGY_LIST, KEY, SECRET, {'pageNo': 4, 'pageSize': 30, 'time': '2023-01'}) @@ -379,17 +390,17 @@ async def test_station_month_energy_list_valid( @pytest.mark.asyncio async def test_station_month_energy_list_invalid_params(api_instance): # Wrong page_size - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_month_energy_list( KEY, SECRET, page_size=1000, month='2023-01') # Wrong month format - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_month_energy_list( KEY, SECRET, month='2023') - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_month_energy_list( KEY, SECRET, month='2023+01') @@ -404,7 +415,7 @@ async def test_station_year_energy_list_valid( year='2023') assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.STATION_YEAR_ENERGY_LIST, + STATION_YEAR_ENERGY_LIST, KEY, SECRET, {'pageNo': 1, 'pageSize': 20, 'time': '2023'}) @@ -413,7 +424,7 @@ async def test_station_year_energy_list_valid( page_no=4, page_size=30, year='2023') assert result == VALID_RESPONSE_RECORDS patched_api_records._get_records.assert_called_with( - api.STATION_YEAR_ENERGY_LIST, + STATION_YEAR_ENERGY_LIST, KEY, SECRET, {'pageNo': 4, 'pageSize': 30, 'time': '2023'}) @@ -421,12 +432,12 @@ async def test_station_year_energy_list_valid( @pytest.mark.asyncio async def test_station_year_energy_list_invalid_params(api_instance): # Wrong page_size - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_year_energy_list( KEY, SECRET, page_size=1000, year='2023') # Wrong year format - with pytest.raises(api.SoliscloudAPI.SolisCloudError): + with pytest.raises(SoliscloudError): await api_instance.station_year_energy_list( KEY, SECRET, year='22023') From 64e2556131aa2d073ea873b0d4e656892f3aa3ee Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Sat, 13 Sep 2025 15:39:13 +0200 Subject: [PATCH 08/14] Add and improve helpers --- soliscloud_api/helpers.py | 53 +++++++++++++++++++++++++++++++++------ 1 file changed, 45 insertions(+), 8 deletions(-) diff --git a/soliscloud_api/helpers.py b/soliscloud_api/helpers.py index 5818d95..c866de3 100644 --- a/soliscloud_api/helpers.py +++ b/soliscloud_api/helpers.py @@ -4,7 +4,7 @@ """ from __future__ import annotations -from soliscloud_api import SoliscloudAPI +from soliscloud_api import SoliscloudAPI, SoliscloudError class Helpers(): @@ -15,10 +15,28 @@ async def get_station_ids( nmi=None ) -> tuple: """ - Parses response from get_station_list and returns all station id's + Calls user_station_list and returns all station id's from the response + as a tuple of int's or None on error. + If nmi is given then the nmi_code parameter is used in the call + (Australian accounts require nmi) """ - response = await api.user_station_list( - key, secret, page_no=1, page_size=100, nmi_code=nmi) + try: + response = await api.user_station_list( + key, secret, page_no=1, page_size=100, nmi_code=nmi) + return Helpers.get_station_ids_from_response(response) + except SoliscloudError: + return None + + @staticmethod + def get_station_ids_from_response( + response: dict[str, str], + ) -> tuple: + """ + Takes response from user_station_list and returns all station id's + from the response as a tuple of int's. + """ + if response is None: + return None stations = () for element in response: stations = stations + (int(element['id']),) @@ -31,13 +49,32 @@ async def get_inverter_ids( nmi=None ) -> tuple: """ - Parses response from get_inverter_list and returns all inverter id's + Calls inverter_list and returns all inverter id's from response as a + tuple of int's or None on error. + (Australian accounts require nmi) + If a station_id is given then a list of inverters for that station_id + is returned, else all inverter id's for all stations + """ + try: + response = await api.inverter_list( + key, secret, page_no=1, page_size=100, + station_id=station_id, nmi_code=nmi) + return Helpers.get_inverter_ids_from_response(response) + except SoliscloudError: + return None + + @staticmethod + def get_inverter_ids_from_response( + response: dict[str, str], + ) -> tuple: + """ + Takes response from inverter_list and returns all inverter id's from + response as a tuple of int's. If a station_id is given then a list of inverters for that station_id is returned, else all inverter id's for all stations """ - response = await api.inverter_list( - key, secret, page_no=1, page_size=100, - station_id=station_id, nmi_code=nmi) + if response is None: + return None inverters = () for element in response: inverters = inverters + (int(element['id']),) From 9447a6fa7d91d20cd35ffd9b92e728d561bc974c Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Fri, 26 Sep 2025 15:36:48 +0200 Subject: [PATCH 09/14] Improve example code --- example.py | 87 ++++++++++++++++++++++++++++++++---------------------- 1 file changed, 51 insertions(+), 36 deletions(-) diff --git a/example.py b/example.py index c620068..93cf7b9 100644 --- a/example.py +++ b/example.py @@ -1,12 +1,18 @@ -"""Example application.""" +"""Example application. +Demonstrates usage of the SoliscloudAPI class to interact with the Soliscloud API. +It also shows how to use the Helpers class for common tasks. +Fill in your own API key and secret in config.json before running. +""" + import asyncio import logging import json from aiohttp import ClientSession -from soliscloud_api import SoliscloudAPI -from soliscloud_api.helpers import Helpers +from soliscloud_api import SoliscloudAPI, SoliscloudError, \ + SoliscloudHttpError, SoliscloudTimeoutError, SoliscloudApiError, \ + Helpers logging.basicConfig(level=logging.DEBUG) @@ -29,52 +35,54 @@ async def main(): soliscloud = SoliscloudAPI( 'https://soliscloud.com:13333', websession) - # Retrieves list of Stations, a.k.a. plants, - # containing the inverters. + # Retrieve list of Stations, a.k.a. plants station_list = await soliscloud.user_station_list( - api_key, api_secret, page_no=1, page_size=100) - # Australian accounts require nmi, uncomment if required. - # (NOT TESTED!) - # station_list = await soliscloud.user_station_list( - # api_key, api_secret, page_no=1, - # page_size=100, nmi_code=api_nmi) + api_key, api_secret, page_no=1, page_size=100, + # Australian accounts require nmi, uncomment if required. + # (NOT TESTED!) + # nmi_code=api_nmi + ) station_list_json = json.dumps(station_list, indent=2) - # Use helper class as alternative + # Use helper class as alternative to directly retrieve station id's station_ids = await Helpers.get_station_ids( soliscloud, api_key, api_secret) - # Get inverters for all stations + # Get inverter data for all stations inverter_list = await soliscloud.inverter_list( - api_key, api_secret, page_no=1, page_size=100) - # Australian accounts require nmi, uncomment if required. - # (NOT TESTED!) - # inverter_list = await soliscloud.inverter_list( - # api_key, api_secret, page_no=1, - # page_size=100, nmi_code=api_nmi) + api_key, api_secret, page_no=1, page_size=100, + # Australian accounts require nmi, uncomment if required. + # (NOT TESTED!) + # nmi_code=api_nmi + ) inverter_list_json = json.dumps(inverter_list, indent=2) - # Use helper class as alternative - inverter_ids = await Helpers.get_inverter_ids( - soliscloud, api_key, api_secret) + # Use helper class to retrieve all inverter id's + inverter_ids = Helpers.get_inverter_ids_from_response( + inverter_list) - inverter_detail = await soliscloud.inverter_detail( - api_key, api_secret, inverter_id=inverter_ids[0]) - inverter_detail_json = json.dumps(inverter_detail, indent=2) + # Get detailed data for all inverters + idl = await soliscloud.inverter_detail_list(api_key, api_secret) + idl_json = json.dumps(idl, indent=2) + + # Use serial number of first inverter to get details + # (should be same as first in inverter_list) + inverter_x = await soliscloud.inverter_detail( + api_key, api_secret, inverter_sn=inverter_list[0]['sn']) + inverter_detail_json = json.dumps(inverter_x, indent=2) # Get data collectors for all stations collector_list = await soliscloud.collector_list( - api_key, api_secret, page_no=1, page_size=100) - # Australian accounts require nmi, uncomment if required. - # (NOT TESTED!) - # collector_list = await soliscloud.collector_list( - # api_key, api_secret, page_no=1, - # page_size=100, nmi_code=api_nmi) + api_key, api_secret, page_no=1, page_size=100, + # Australian accounts require nmi, uncomment if required. + # (NOT TESTED!) + # nmi_code=api_nmi + ) collector_list_json = json.dumps(collector_list, indent=2) except ( - SoliscloudAPI.SolisCloudError, - SoliscloudAPI.HttpError, - SoliscloudAPI.TimeoutError, - SoliscloudAPI.ApiError, + SoliscloudError, + SoliscloudHttpError, + SoliscloudTimeoutError, + SoliscloudApiError, ) as error: print(f"Error: {error}") else: @@ -88,7 +96,9 @@ async def main(): print(f"{inverter_list_json}") print("InverterDetails call success:") - print(f"{inverter_detail_json}") + print(inverter_detail_json) + for key in ['iA', 'uA', 'pA', 'iB', 'uB', 'pB', 'iC', 'uC', 'pC']: + print(f"{key}: {inverter_x[key]}") print("Helper call success:") print(f"{inverter_ids}") @@ -96,6 +106,11 @@ async def main(): print("CollectorList call success:") print(f"{collector_list_json}") + print("InverterDetailList call success:") + print(idl_json) + for key in ['iAc1', 'uAc1', 'pA', 'iB', 'uB', 'pB', 'iC', 'uC', 'pC', 'apparentPower']: + print(f"{key}: {idl[0][key]}") + loop = asyncio.new_event_loop() loop.run_until_complete(main()) loop.close() From f4c990ccadb99016766cf68b181a90c7ac16b874 Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Fri, 26 Sep 2025 15:47:14 +0200 Subject: [PATCH 10/14] Pretty --- example.py | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/example.py b/example.py index 93cf7b9..af799c7 100644 --- a/example.py +++ b/example.py @@ -1,6 +1,7 @@ -"""Example application. -Demonstrates usage of the SoliscloudAPI class to interact with the Soliscloud API. -It also shows how to use the Helpers class for common tasks. +""" +Example application. +Demonstrates usage of the SoliscloudAPI class to interact with the Soliscloud +API. It also shows how to use the Helpers class for common tasks. Fill in your own API key and secret in config.json before running. """ @@ -108,7 +109,18 @@ async def main(): print("InverterDetailList call success:") print(idl_json) - for key in ['iAc1', 'uAc1', 'pA', 'iB', 'uB', 'pB', 'iC', 'uC', 'pC', 'apparentPower']: + for key in [ + 'iAc1', + 'uAc1', + 'pA', + 'iB', + 'uB', + 'pB', + 'iC', + 'uC', + 'pC', + 'apparentPower' + ]: print(f"{key}: {idl[0][key]}") loop = asyncio.new_event_loop() From 0e548fd32b2ed7267a8f7d0932b27e16e30f6d36 Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Fri, 26 Sep 2025 16:19:13 +0200 Subject: [PATCH 11/14] Bump major version --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 13509ea..ba8faea 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "soliscloud-api" -version = "1.2.0" +version = "2.0.0" authors = [ { name="Peter van Hulten", email="peter.vanhulten@gmx.net" }, ] From ef5de15e2c77e4aca6a9af1be1b867c10711f9b0 Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Fri, 26 Sep 2025 16:38:15 +0200 Subject: [PATCH 12/14] Create example_entities.py --- example_entities.py | 85 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 85 insertions(+) create mode 100644 example_entities.py diff --git a/example_entities.py b/example_entities.py new file mode 100644 index 0000000..4fab25a --- /dev/null +++ b/example_entities.py @@ -0,0 +1,85 @@ +""" +Example application. +Demonstrates the use of entity classes as a higher level interface to the +Soliscloud API. Fill in your own API key and secret in config.json before +running. +By using the entity classes you can more easily access the data +attributes of the plant, inverters and collectors. + +There are 2 ways to create the entity classes: +1. Use the class method 'initialize_from_data' to create entity instances + from data retrieved from the API. It will create one or more entity + instances from the data passed in. This method will not retrieve any data + from the API, so you need to do that yourself first. It will not + recursively create all entities, i.e. plants, inverters and collectors. + You need to call it separately for each entity class. +2. Use the class method 'initialize_from_session' to create entity instances + by passing in the aiohttp session and API credentials. This method will + retrieve the data from the API and create the entity instances. It + will recursively create all entities, i.e. plants, inverters + and collectors when called on the Plant class and inverters and collectors + when called on the Inverter class. +""" +from __future__ import annotations + +import asyncio +import logging +import json + +from aiohttp import ClientSession + +from soliscloud_api import SoliscloudError, SoliscloudHttpError, \ + SoliscloudTimeoutError, SoliscloudApiError +from soliscloud_api import Plant, Inverter, Collector # noqa: F401 +from soliscloud_api import SoliscloudAPI as api + + +logging.basicConfig(level=logging.DEBUG) + + +async def main(): + """Run main function.""" + # Put your own key and secret in the config.json file + with open('config.json', 'r') as file: + data = json.load(file) + + api_key = data['key'] + api_secret = bytearray(data['secret'], 'utf-8') + # Australian accounts require nmi, uncomment if required. + # (NOT TESTED!) + # api_nmi = data['nmi'] + + async with ClientSession() as websession: + try: + the_api = api('https://soliscloud.com:13333', websession) + data = await the_api.inverter_list( + api_key, api_secret, page_no=1, page_size=100) + # use plant_id of your own plant here to get inverters for that plant + # use inverter_id to only get one specific inverter + inverters = Inverter.initialize_from_data( + data, plant_id='your plant id goes here') + plants = await Plant.initialize_from_session( + websession, api_key, api_secret) + except ( + SoliscloudError, + SoliscloudHttpError, + SoliscloudTimeoutError, + SoliscloudApiError, + ) as error: + print(f"Error: {error}") + else: + p = plants[0] + print(f"Plant id: {p.plant_id}") + print(f"Plant name: {p.data['station_name']}") + print(f"Number of inverters: {len(p.inverters)}") + for inv in inverters: + print(f"Inverter id: {inv.inverter_id}") + # The attributes if the inverter are in inv.data + # If an attribute has a unit then the value is of + # dimensioned type + print(f"Total energy: {inv.data['etotal']}") + print(f"Total energy value: {inv.data['etotal'].value}") + print(f"Total energy unit: {inv.data['etotal'].unit}") +loop = asyncio.new_event_loop() +loop.run_until_complete(main()) +loop.close() From 62542bc041456d19e3113a9f1f4847b6a1caf37a Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Fri, 26 Sep 2025 16:43:23 +0200 Subject: [PATCH 13/14] Update minor version --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index ba8faea..be170e1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "soliscloud-api" -version = "2.0.0" +version = "1.3.0" authors = [ { name="Peter van Hulten", email="peter.vanhulten@gmx.net" }, ] From db4e54a066532b8d5870edacc68289083dc7be7b Mon Sep 17 00:00:00 2001 From: hultenvp <61835400+hultenvp@users.noreply.github.com> Date: Fri, 26 Sep 2025 16:51:09 +0200 Subject: [PATCH 14/14] Ignore complexity error for now --- .github/workflows/python-app.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/python-app.yml b/.github/workflows/python-app.yml index f1d8539..264adba 100644 --- a/.github/workflows/python-app.yml +++ b/.github/workflows/python-app.yml @@ -34,7 +34,7 @@ jobs: flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide # --exit-zero removed to enforce zero lint errors - flake8 . --extend-ignore=E128,E124,F403,F405 --count --max-complexity=10 --max-line-length=127 --statistics + flake8 . --extend-ignore=C901,E128,E124,F403,F405 --count --max-complexity=10 --max-line-length=127 --statistics - name: Test with pytest run: | pytest --capture=tee-sys -v