diff --git a/Makefile b/Makefile index ddbf243..d092ffa 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,7 @@ # Makefile ;) dist: clean + @pip3 install --upgrade build twine python -m build twine check dist/* diff --git a/README.md b/README.md index 907d47f..7d5e4ab 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,12 @@ This modules contains two distinct sets of APIs: - the original sonnen API (v1) - the new v2 API +Both are available in a synchronous and an asynchronous version: +- `sonnenbatterie` - SonnenAPI v1, synchronous +- `AsyncSonnenBatterie` - SonnenAPI v1, asynchronous +- `SonnenBatterieV2` - SonnenAPI v2, synchronous +- `AsyncSonnenBatterieV2` - SonnenAPI v2, asynchronous + The major difference between those two is the method to authenticate. With API v1 you need to login using a username and a password. The newer API requires a token that can be found in the web UI of your Sonnenbatterie. @@ -28,7 +34,9 @@ pip3 install sonnenbatterie ## Usage +### SonnenAPI v1 - sync ``` python +# API v1 from sonnenbatterie import sonnenbatterie sb_host = '192.168.1.2' @@ -44,12 +52,40 @@ print(sb.get_batterysystem()) # retrieve battery system data print(sb.get_inverter()) # retrieve inverter status print(sb.get_systemdata()) # retrieve system data print(sb.get_battery()) # get battery information +``` + +### SonnenAPI v1 - async +``` python +# API v1 +from sonnenbatterie import AsyncSonnenBatterie + +sb_host = '192.168.1.2' +sb_user = 'User' +sb_pass = 'Password' + +# Init class, establish connection +sb = AsyncSonnenBatterie(sb_host, sb_user, sb_pass) + +print(await sb.get_status()) # retrieve general information +print(await sb.get_powermeters()) # retrieive power meter details +print(await sb.get_batterysystem()) # retrieve battery system data +print(await sb.get_inverter()) # retrieve inverter status +print(await sb.get_systemdata()) # retrieve system data +print(await sb.get_battery()) # get battery information +# Async needs to close the connection! +await sb.logout() +``` + +### SonnenAPI v2 - sync +``` python # API v2 # can either be access directly, see below, or # via sb.sb2 (gets initialiazed automatically when creating a V1 object) from sonnebatterie2 import SonnenBatterieV2 + +sb_host = '192.168.1.2' sb_token = 'SeCrEtToKeN' # retrieve via Web UI of SonnenBatterie sb2 = SonnenBatterieV2(sb_host, sb_token) @@ -58,6 +94,29 @@ print(sb2.get_battery_module_data()) # get battery module data print(sb2.get_inverter_data()) # retrieve inverter data print(sb2.get_latest_data()) # get latest date from sonnenbatterie print(sb2_get_powermeter_data()) # get data from power meters -print(sb2.get_status_data()) # get overall status information +print(sb2.get_status()) # get overall status information print(sb2.get_io_data()) # get io status ``` + +### SonnenAPI v2 - async +``` python +# API v2 +# can either be access directly, see below, or +# via sb.sb2 (gets initialiazed automatically when creating a V1 object) + +from sonnebatterie2 import AsyncSonnenBatterieV2 + +sb_host = '192.168.1.2' +sb_token = 'SeCrEtToKeN' # retrieve via Web UI of SonnenBatterie + +sb2 = AsyncSonnenBatterieV2(sb_host, sb_token) +print(await sb2.get_configurations()) # retrieve configuration overview +print(await sb2.get_battery_module_data()) # get battery module data +print(await sb2.get_inverter_data()) # retrieve inverter data +print(await sb2.get_latest_data()) # get latest date from sonnenbatterie +print(await sb2_get_powermeter_data()) # get data from power meters +print(await sb2.get_status()) # get overall status information +print(await sb2.get_io_data()) # get io status +# Async needs to close the connection! +await sb.logout() +``` diff --git a/setup.cfg b/setup.cfg index 2767d97..63ec7f2 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,6 +1,6 @@ [metadata] name = sonnenbatterie -version = 0.4.0 +version = 0.5.0 author = Jan Weltmeyer description = "Access Sonnenbatterie REST API" long_description = file: README.md @@ -11,7 +11,7 @@ project_urls = Issue Tracker = https://github.com/weltmeyer/python_sonnenbatterie/issues classifiers = Development Status :: 4 - Beta - Programming Language :: Python :: 3.8 + Programming Language :: Python :: 3.10 Operating System :: OS Independent Environment :: Console License :: OSI Approved :: GNU General Public License v3 (GPLv3) @@ -21,9 +21,10 @@ package_dir = = . packages = find: include_package_data = True -python_requires = >= 3.8 +python_requires = >= 3.10 install_requires = requests + aiohttp [options.packages.find] where = . diff --git a/sonnenbatterie/__init__.py b/sonnenbatterie/__init__.py index 2aa0a37..19bc763 100644 --- a/sonnenbatterie/__init__.py +++ b/sonnenbatterie/__init__.py @@ -1 +1 @@ -from .sonnenbatterie import sonnenbatterie \ No newline at end of file +from .sonnenbatterie import sonnenbatterie, AsyncSonnenBatterie \ No newline at end of file diff --git a/sonnenbatterie/const.py b/sonnenbatterie/const.py index a497cb5..7d8ce37 100644 --- a/sonnenbatterie/const.py +++ b/sonnenbatterie/const.py @@ -1,6 +1,6 @@ -DEFAULT_BATTERY_LOGIN_TIMEOUT=120 -DEFAULT_CONNECT_TO_BATTERY_TIMEOUT=60 -DEFAULT_READ_FROM_BATTERY_TIMEOUT=60 +DEFAULT_BATTERY_LOGIN_TIMEOUT=10 +DEFAULT_CONNECT_TO_BATTERY_TIMEOUT=6 +DEFAULT_READ_FROM_BATTERY_TIMEOUT=6 SONNEN_OPERATING_MODE_UNKNOWN_NAME="Unknown Operating Mode" SONNEN_OPERATING_MODE_UNKNOWN="0" @@ -9,10 +9,10 @@ SONNEN_OPERATING_MODE_BATTERY_MODULE_EXTENSION_30_PERCENT_NAME="Battery-Module-Extension (30%)" SONNEN_OPERATING_MODE_TIME_OF_USE_NAME="Time-Of-Use" SONNEN_OPERATING_MODE_NAMES_TO_OPERATING_MODES = { - SONNEN_OPERATING_MODE_MANUAL_NAME :"1", - SONNEN_OPERATING_MODE_AUTOMATIC_SELF_CONSUMPTION_NAME:"2", - SONNEN_OPERATING_MODE_BATTERY_MODULE_EXTENSION_30_PERCENT_NAME:"6", - SONNEN_OPERATING_MODE_TIME_OF_USE_NAME:"10" + SONNEN_OPERATING_MODE_MANUAL_NAME : 1, + SONNEN_OPERATING_MODE_AUTOMATIC_SELF_CONSUMPTION_NAME: 2, + SONNEN_OPERATING_MODE_BATTERY_MODULE_EXTENSION_30_PERCENT_NAME: 6, + SONNEN_OPERATING_MODE_TIME_OF_USE_NAME: 10 } SONNEN_OPERATING_MODES_TO_OPERATING_MODE_NAMES = {v:k for k,v in SONNEN_OPERATING_MODE_NAMES_TO_OPERATING_MODES.items()} diff --git a/sonnenbatterie/sonnenbatterie.py b/sonnenbatterie/sonnenbatterie.py index 811dce9..64aefed 100644 --- a/sonnenbatterie/sonnenbatterie.py +++ b/sonnenbatterie/sonnenbatterie.py @@ -1,13 +1,20 @@ +import json import sys + +from sonnenbatterie2.sonnenbatterie2 import AsyncSonnenBatterieV2 + sys.path.append("..") import hashlib import requests +import aiohttp + from sonnenbatterie2 import SonnenBatterieV2 from .const import * class sonnenbatterie: + # noinspection HttpUrlsUsage def __init__(self,username,password,ipaddress): self.username=username self.password=password @@ -19,26 +26,21 @@ def __init__(self,username,password,ipaddress): self._batteryReadTimeout = DEFAULT_READ_FROM_BATTERY_TIMEOUT self._batteryRequestTimeout = (self._batteryConnectTimeout, self._batteryReadTimeout) - self._login() - + self.login() + self.token = None self.sb2 = SonnenBatterieV2(ip_address=self.ipaddress, api_token=self.token) - def _login(self): + def login(self): password_sha512 = hashlib.sha512(self.password.encode('utf-8')).hexdigest() req_challenge=requests.get(self.baseurl+'challenge', timeout=self._batteryLoginTimeout) req_challenge.raise_for_status() challenge=req_challenge.json() response=hashlib.pbkdf2_hmac('sha512',password_sha512.encode('utf-8'),challenge.encode('utf-8'),7500,64).hex() - #print(password_sha512) - #print(challenge) - #print(response) getsession=requests.post(self.baseurl+'session',{"user":self.username,"challenge":challenge,"response":response}, timeout=self._batteryLoginTimeout) getsession.raise_for_status() - #print(getsession.text) token=getsession.json()['authentication_token'] - #print(token) self.token=token def set_login_timeout(self, timeout:int = 120): @@ -68,7 +70,7 @@ def _get(self,what,isretry=False): headers={'Auth-Token': self.token}, timeout=self._batteryRequestTimeout ) if not isretry and response.status_code==401: - self._login() + self.login() return self._get(what,True) if response.status_code != 200: response.raise_for_status() @@ -82,7 +84,7 @@ def _put(self, what, payload, isretry=False): headers={'Auth-Token': self.token,'Content-Type': 'application/json'} , json=payload, timeout=self._batteryRequestTimeout ) if not isretry and response.status_code==401: - self._login() + self.login() return self._put(what, payload,True) if response.status_code != 200: response.raise_for_status() @@ -91,12 +93,11 @@ def _put(self, what, payload, isretry=False): def _post(self, what, isretry=False): # This is a synchronous call, you may need to wrap it in a thread or something for asynchronous operation url = self.baseurl+what - print("Posting "+url) response=requests.post(url, headers={'Auth-Token': self.token,'Content-Type': 'application/json'}, timeout=self._batteryRequestTimeout ) if not isretry and response.status_code==401: - self._login() + self.login() return self._post(what, True) if response.status_code != 200: response.raise_for_status() @@ -138,30 +139,34 @@ def get_latest_data(self): def get_configurations(self): return self.sb2.get_configurations() - + + # deprecate with 0.7.0 -> get_config_item def get_configuration(self, name): return self.sb2.get_config_item(name) + + def get_config_item(self, name): + return self.sb2.get_config_item(name) # these have special handling in some form, for example converting a mode as a number into a string def get_current_charge_level(self): return self.get_latest_data().get(SONNEN_LATEST_DATA_CHARGE_LEVEL) - def get_operating_mode(self): - return self.get_configuration(SONNEN_CONFIGURATION_OPERATING_MODE) + def get_operating_mode(self) -> int: + return int(self.get_config_item(SONNEN_CONFIGURATION_OPERATING_MODE)[SONNEN_CONFIGURATION_OPERATING_MODE]) def get_operating_mode_name(self): operating_mode = self.get_operating_mode() - return SONNEN_OPERATING_MODES_TO_OPERATING_MODE_NAMES.get(operating_mode[SONNEN_CONFIGURATION_OPERATING_MODE]) + return SONNEN_OPERATING_MODES_TO_OPERATING_MODE_NAMES.get(operating_mode) - def set_operating_mode(self, operating_mode): - return self.set_configuration(SONNEN_CONFIGURATION_OPERATING_MODE, operating_mode) + def set_operating_mode(self, operating_mode) -> int: + return int(self.set_configuration(SONNEN_CONFIGURATION_OPERATING_MODE, operating_mode)) - def set_operating_mode_by_name(self, operating_mode_name): + def set_operating_mode_by_name(self, operating_mode_name) -> int: return self.set_operating_mode(SONNEN_OPERATING_MODE_NAMES_TO_OPERATING_MODES.get(operating_mode_name)) def get_battery_reserve(self): - return self.get_configuration(SONNEN_CONFIGURATION_BACKUP_RESERVE) + return self.get_config_item(SONNEN_CONFIGURATION_BACKUP_RESERVE)[SONNEN_CONFIGURATION_BACKUP_RESERVE] def set_battery_reserve(self, reserve=5): reserve = int(reserve) @@ -184,3 +189,236 @@ def set_battery_reserve_relative_to_current_charge(self, offset=0, minimum_reser elif target_level > 100: target_level = 100 return self.set_battery_reserve(target_level) + + +""" AsyncSonnenbatterie """ + +class AsyncSonnenBatterie: + # noinspection HttpUrlsUsage + def __init__(self, username, password, ipaddress): + self.username = username + self.password = password + self.ipaddress = ipaddress + + self.baseurl = 'http://' + self.ipaddress + '/api/' + + self._timeout_total = DEFAULT_BATTERY_LOGIN_TIMEOUT + self._timeout_connect = DEFAULT_CONNECT_TO_BATTERY_TIMEOUT + self._timeout_read = DEFAULT_READ_FROM_BATTERY_TIMEOUT + + self._timeout = self._set_timeouts() + + self.token = None + self.sb2 = None + + self._session = None + + def _set_timeouts(self) -> aiohttp.ClientTimeout: + return aiohttp.ClientTimeout( + total=self._timeout_total, + connect=self._timeout_connect, + sock_connect=self._timeout_connect, + sock_read=self._timeout_read, + ) + + async def logout(self): + if self.sb2: + await self.sb2.logout() + + if self._session: + await self._session.close() + + async def login(self): + if self._session is None: + self._session = aiohttp.ClientSession() + + pw_sha512 = hashlib.sha512(self.password.encode('utf-8')).hexdigest() + req_challenge = await self._session.get( + self.baseurl+'challenge', + timeout=self._timeout, + ) + req_challenge.raise_for_status() + + challenge = await req_challenge.json() + response = hashlib.pbkdf2_hmac( + 'sha512', + pw_sha512.encode('utf-8'), + challenge.encode('utf-8'), + 7500, + 64 + ).hex() + + session = await self._session.post( + url = self.baseurl+'session', + data = {"user":self.username,"challenge":challenge,"response":response}, + timeout=self._timeout, + ) + session.raise_for_status() + token = await session.json() + self.token = token['authentication_token'] + + # Inisitalite async API v2 + if self.sb2 is None: + self.sb2 = AsyncSonnenBatterieV2(ip_address=self.ipaddress, api_token=self.token) + + + """ Base functions """ + + + async def _get(self, what, isretry=False) -> json: + if self.token is None: + await self.login() + + url = self.baseurl + what + response = await self._session.get( + url = url, + headers={'Auth-Token': self.token}, + timeout=self._timeout, + ) + + if not isretry and response.status == 401: + return await self._get(what, True) + + if response.status != 200: + response.raise_for_status() + + return await response.json() + + + async def _post(self, what, isretry=False) -> json: + if self.token is None: + await self.login() + + url = self.baseurl + what + response = await self._session.post( + url = url, + headers = {'Auth-Token': self.token, 'Content-Type': 'application/json'}, + timeout = self._timeout, + ) + + if not isretry and response.status == 401: + return await self._post(what, True) + if response.status != 200: + response.raise_for_status() + + return await response.json() + + + async def _put(self, what, payload, isretry=False) -> json: + if self.token is None: + await self.login() + + url = self.baseurl + what + response = await self._session.put( + url = url, + headers = {'Auth-Token': self.token, 'Content-Type': 'application/json'}, + json = payload, + timeout = self._timeout, + ) + + if not isretry and response.status == 401: + return await self._put(what, True) + + if response.status != 200: + response.raise_for_status() + + return await response.json() + + + """ API functions """ + + + async def get_powermeter(self) -> json: + result = await self._get(SONNEN_API_PATH_POWER_METER) + return result + + async def get_batterysystem(self) -> json: + return await self._get(SONNEN_API_PATH_BATTERY_SYSTEM) + + async def get_inverter(self) -> json: + return await self._get(SONNEN_API_PATH_INVERTER) + + async def get_systemdata(self) -> json: + return await self._get(SONNEN_API_PATH_SYSTEM_DATA) + + async def get_status(self) -> json: + return await self._get(SONNEN_API_PATH_STATUS) + + async def get_battery(self) -> json: + return await self._get(SONNEN_API_PATH_BATTERY) + + + """ API v2 calls """ + + + async def set_configuration(self, name, value) -> json: + if self.sb2 is None: + await self.login() + return await self.sb2.set_config_item(name, value) + + async def get_latest_data(self) -> json: + if self.sb2 is None: + await self.login() + return await self.sb2.get_latest_data() + + async def get_configurations(self) -> json: + if self.sb2 is None: + await self.login() + return await self.sb2.get_configurations() + + async def get_config_item(self, name) -> json: + if self.sb2 is None: + await self.login() + return await self.sb2.get_config_item(name) + + + """ Special functions """ + + # GET + async def get_current_charge_level(self) -> int: + result = await self.get_latest_data() + return result.get(SONNEN_LATEST_DATA_CHARGE_LEVEL) + + async def get_operating_mode(self) -> int: + result = await self.get_config_item(SONNEN_CONFIGURATION_OPERATING_MODE) + return int(result[SONNEN_CONFIGURATION_OPERATING_MODE]) + + async def get_operating_mode_name(self) -> str: + opmode = await self.get_operating_mode() + return SONNEN_OPERATING_MODES_TO_OPERATING_MODE_NAMES.get(opmode) + + async def get_battery_reserve(self) -> int: + result = await self.get_config_item(SONNEN_CONFIGURATION_BACKUP_RESERVE) + return int(result[SONNEN_CONFIGURATION_BACKUP_RESERVE]) + + #SET + async def set_operating_mode(self, operating_mode) -> int: + result = await self.set_configuration( + SONNEN_CONFIGURATION_OPERATING_MODE, + operating_mode + ) + return int(result[SONNEN_CONFIGURATION_OPERATING_MODE]) + + async def set_operating_mode_by_name(self, operating_mode_name) -> int: + result = await self.set_configuration( + SONNEN_CONFIGURATION_OPERATING_MODE, + SONNEN_OPERATING_MODE_NAMES_TO_OPERATING_MODES.get(operating_mode_name) + ) + return int(result[SONNEN_CONFIGURATION_OPERATING_MODE]) + + async def set_battery_reserve(self, battery_reserve) -> json: + reserve = int(battery_reserve) + if (reserve < 0) or (reserve > 100): + raise Exception(f"Reserve must be between 0 and 100, you spcified {reserve}") + return await self.set_configuration(SONNEN_CONFIGURATION_BACKUP_RESERVE, reserve) + + async def set_battery_reserve_relative_to_current_charge(self, offset=0, min_res=0) -> json: + current_level = await self.get_current_charge_level() + target_level = current_level + offset + if target_level < min_res: + target_level = min_res + if target_level > 100: + target_level = 100 + if target_level < 0: + target_level = 0 + return await self.set_battery_reserve(target_level) diff --git a/sonnenbatterie/timeofuse.py b/sonnenbatterie/timeofuse.py deleted file mode 100644 index 825af1a..0000000 --- a/sonnenbatterie/timeofuse.py +++ /dev/null @@ -1,220 +0,0 @@ -"""Process the time of use schedule stuff""" -from datetime import time, datetime -from typing import List - - -ATTR_TOU_START="start" -ATTR_TOU_STOP="stop" -ATTR_TOU_MAX_POWER="threshold_p_max" -TIME_FORMAT="%H:%M" - -MIDNIGHT=time() - -class timeofuse: - def __init__(self, start_time:time, stop_time:time, max_power=20000): - self.start_time = start_time - # for comparisson reasons we can't have the stop time be midnight as thaty's actually the earliest possible time - # so if it is provided as midnight make it the latest possible time - if (stop_time == MIDNIGHT): - stop_time = time.max - self.stop_time = stop_time - self.max_power = max_power - - def __eq__(self, other) -> bool : - if not isinstance(other, type(self)): - return False - return (self.start_time == other.start_time) and (self.stop_time == other.stop_time) and (self.max_power == other.max_power) - - def __hash__(self) -> int: - return hash(self.start_time) ^ hash(self.stop_time) ^ hash(self.max_power) - - def get_as_tou(self): - start_string = self.get_start_time_as_string() - tmp_stop_time = self.stop_time - # previously we had to munge midnight as an stop time to 23:59:59.999999 - # if that was done now undo it - if (tmp_stop_time == time.max): - tmp_stop_time = MIDNIGHT - - stop_string=tmp_stop_time.strftime(TIME_FORMAT) - max_power_string = str(self.max_power) - return {ATTR_TOU_START:start_string, ATTR_TOU_STOP:stop_string, ATTR_TOU_MAX_POWER:max_power_string} - - def get_as_string(self) -> str: - resp = "Start "+self.get_start_time_as_string()+", End "+self.get_stop_time_as_string(), "Max allowable power "+str(self.get_max_power()) - return resp - - def get_start_time_as_string(self) -> str: - return self._get_time_as_string(self.start_time) - - def get_stop_time_as_string(self) -> str: - tmp_stop_time = self.stop_time - # previously we had to munge midnight as an stop time to 23:59:59.999999 - # if that was done now undo it - if (tmp_stop_time == time.max): - tmp_stop_time = MIDNIGHT - return self._get_time_as_string(tmp_stop_time) - - def _get_time_as_string(self, timeobj:time) -> str: - return timeobj.strftime(TIME_FORMAT) - - def get_max_power(self) -> int: - return self.max_power - - def from_tou(tou): - # parse it out - start_time = datetime.strptime(tou.get(ATTR_TOU_START), TIME_FORMAT).time() - stop_time = datetime.strptime(tou.get(ATTR_TOU_STOP), TIME_FORMAT).time() - max_power= int(tou.get(ATTR_TOU_MAX_POWER)) - # build the resulting object - return timeofuse(start_time, stop_time, max_power) - - def is_overlapping(self, other): - # is our start time within the others time window ? - if (self.start_time>= other.start_time) and (self.start_time<= other.stop_time): - return True - - # is our end time within the others time window ? - if (self.stop_time>= other.start_time) and (self.stop_time<= other.stop_time): - return True - - - # is it's start time within the out time window ? - if (other.start_time>= self.start_time) and (other.start_time<= self.stop_time): - return True - - # is it's end time within the out time window ? - if (other.stop_time>= self.start_time) and (other.stop_time<= self.stop_time): - return True - - # no overlap - return False - - def create_time_of_use_entry(start_hour=23, start_min=30, stop_hour=5, stop_min=30, max_power=20000): - start_time = time(hour=start_hour, minute=start_min) - stop_time = time(hour=stop_hour, minute=stop_min) - return timeofuse(start_time, stop_time, int(max_power)) - -class timeofuseschedule: - def __init__(self): - self._schedule_entries = [] - - # adds the entry ensureing that it does not overlap with an existing entry - def _add_entry(self, entry): - if (entry.stop_time < entry.start_time): - raise Exception("End time cannot be before start time") - for i in self._schedule_entries: - if (i.is_overlapping(entry)): - raise Exception("Unable to add entry, overlaps with exisitngv entry") - self._schedule_entries.append(entry) - # maintains this as a sotred list based on the start time, this lets us compare properly - self._schedule_entries = sorted(self._schedule_entries, key=lambda entry: entry.start_time) - - # Add an entry, if it spans midnight split it into a before midnight and after midnight section - # Note that this IS NOT reversed on retrieving the saved entries - # Note that the change may result in a modified list order, so callers should AWAYS - # use the returned list or get a new one as that will reflect the current state of - # afairs - def add_entry(self, entry): - if (entry.start_time > entry.stop_time): - # this is a over midnight situation - self._add_entry(timeofuse(entry.start_time, time.max, entry.max_power)) - self._add_entry(timeofuse(time.min, entry.stop_time,entry.max_power)) - else: - self._add_entry(entry) - return self.get_as_tou_schedule() - - # Note that the change may result in a modified list order, so callers should AWAYS - # use the returned list or get a new one as that will reflect the current state of - # afairs - def delete_entry(self, entry_number): - self._schedule_entries.pop(entry_number) - return self.get_as_tou_schedule() - # removes and exisitng entry and adds a new one , this is really just a convenience - # If the new entry is rejected due to overlap then the deleted one IS NOT REPLACED - # Note that the change may result in a modified list order, so callers should AWAYS - # use the returned list or get a new one as that will reflect the current state of - # afairs - def remove_and_replace_entry(self, old_entry_nuber, new_entry): - self._schedule_entries.pop(old_entry_nuber) - return self.add_entry(new_entry) - - def get_as_tou_schedule(self)-> List[timeofuse]: - schedules = [] - for i in self._schedule_entries : - schedules.append(i.get_as_tou()) - return schedules - - def get_as_string(self) -> str: - result = "" - doneFirst = False - for entry in self._schedule_entries : - if (doneFirst) : - result = result +"," - else : - doneFirst = True - result = result + str(entry.get_as_string()) - return result - - # retained fore compatibility purposed, is not just a wrapper roung def load_tou_schedule - def load_tou_schedule(self, schedcule): - self.load_tou_schedule_from_json(schedcule) - - # replace the current tou schedule data with the new dicitonary data - def load_tou_schedule_from_json(self, json_schedule): - self._schedule_entries = [] - for entry in json_schedule: - tou_entry = timeofuse.from_tou(entry) - self.add_entry(tou_entry) - - # create a new timeofuseschedule with the provided array ofdictionary data - def build_from_json(json_schedule) : - tous = timeofuseschedule() - for entry in json_schedule: - tou_entry = timeofuse.from_tou(entry) - tous.add_entry(tou_entry) - return tous - - def entry_count(self) -> int: - return len(self._schedule_entries) - - - def get_tou_entry_count(self) -> int: - return len(self._schedule_entries) - - def get_tou_entry(self, i:int) -> timeofuse: - entrties = self.get_tou_entry_count() - if (i > entrties): - return None - else: - return self._schedule_entries[i] - - def __eq__(self, other) -> bool: - if not isinstance(other, type(self)): - return False - myEntryCount = self.entry_count() - otherEntryCount = other.entry_count() - # if there both zero length by definition they are equal - if (myEntryCount == 0) and (otherEntryCount==0): - return True - # different numbers of entries means different scheduled - if (myEntryCount != otherEntryCount): - return False - # for each entry - for i in range(0, myEntryCount): - myTou = self.get_tou_entry(i) - otherTou = other.get_tou_entry(i) - if (myTou != otherTou): - return False - # got to the end of the individual timeofuse entries and they arew all equal so ... - return True - - def __hash__(self) -> int: - myHash = 0 - myEntryCount = self.entry_count() - for i in range(0, myEntryCount): - myTou = self.get_tou_entry(i) - myTouHash = hash(myTou) - # adjust the hash based on the position in the order to allow for things in a differing order - myHash = myHash ^ (myTouHash + i) - return myHash \ No newline at end of file diff --git a/sonnenbatterie2/__init__.py b/sonnenbatterie2/__init__.py index 25e34ab..bca0e8e 100644 --- a/sonnenbatterie2/__init__.py +++ b/sonnenbatterie2/__init__.py @@ -1 +1 @@ -from .sonnenbatterie2 import SonnenBatterieV2 +from .sonnenbatterie2 import SonnenBatterieV2, AsyncSonnenBatterieV2 diff --git a/sonnenbatterie2/const.py b/sonnenbatterie2/const.py new file mode 100644 index 0000000..7d8ce37 --- /dev/null +++ b/sonnenbatterie2/const.py @@ -0,0 +1,36 @@ +DEFAULT_BATTERY_LOGIN_TIMEOUT=10 +DEFAULT_CONNECT_TO_BATTERY_TIMEOUT=6 +DEFAULT_READ_FROM_BATTERY_TIMEOUT=6 + +SONNEN_OPERATING_MODE_UNKNOWN_NAME="Unknown Operating Mode" +SONNEN_OPERATING_MODE_UNKNOWN="0" +SONNEN_OPERATING_MODE_MANUAL_NAME="Manual" +SONNEN_OPERATING_MODE_AUTOMATIC_SELF_CONSUMPTION_NAME="Automatic - Self-Consumption" +SONNEN_OPERATING_MODE_BATTERY_MODULE_EXTENSION_30_PERCENT_NAME="Battery-Module-Extension (30%)" +SONNEN_OPERATING_MODE_TIME_OF_USE_NAME="Time-Of-Use" +SONNEN_OPERATING_MODE_NAMES_TO_OPERATING_MODES = { + SONNEN_OPERATING_MODE_MANUAL_NAME : 1, + SONNEN_OPERATING_MODE_AUTOMATIC_SELF_CONSUMPTION_NAME: 2, + SONNEN_OPERATING_MODE_BATTERY_MODULE_EXTENSION_30_PERCENT_NAME: 6, + SONNEN_OPERATING_MODE_TIME_OF_USE_NAME: 10 +} +SONNEN_OPERATING_MODES_TO_OPERATING_MODE_NAMES = {v:k for k,v in SONNEN_OPERATING_MODE_NAMES_TO_OPERATING_MODES.items()} + +SONNEN_OPEATING_MODES=[SONNEN_OPERATING_MODE_MANUAL_NAME, SONNEN_OPERATING_MODE_AUTOMATIC_SELF_CONSUMPTION_NAME, SONNEN_OPERATING_MODE_BATTERY_MODULE_EXTENSION_30_PERCENT_NAME,SONNEN_OPERATING_MODE_TIME_OF_USE_NAME ] + +SONNEN_CONFIGURATION_OPERATING_MODE="EM_OperatingMode" +SONNEN_CONFIGURATION_TOU_SCHEDULE="EM_ToU_Schedule" +SONNEN_CONFIGURATION_BACKUP_RESERVE="EM_USOC" +SONNEN_LATEST_DATA_CHARGE_LEVEL="USOC" + +SONNEN_API_PATH_CONFIGURATIONS="v2/configurations" +SONNEN_API_PATH_LATEST_DATA="v2/latestdata" +SONNEN_API_PATH_POWER_METER="powermeter" +SONNEN_API_PATH_BATTERY_SYSTEM="battery_system" +SONNEN_API_PATH_INVERTER="inverter" +SONNEN_API_PATH_SYSTEM_DATA="system_data" +SONNEN_API_PATH_STATUS="v1/status" +SONNEN_API_PATH_BATTERY="battery" + +SONNEN_CHARGE_PATH="charge" +SONNEN_DISCHARGE_PATH="discharge" diff --git a/sonnenbatterie2/sonnenbatterie2.py b/sonnenbatterie2/sonnenbatterie2.py index 65cafa6..cb7cc75 100644 --- a/sonnenbatterie2/sonnenbatterie2.py +++ b/sonnenbatterie2/sonnenbatterie2.py @@ -1,11 +1,14 @@ import json +import aiohttp import requests -from sonnenbatterie.const import ( +from .const import ( DEFAULT_CONNECT_TO_BATTERY_TIMEOUT, DEFAULT_READ_FROM_BATTERY_TIMEOUT, - SONNEN_CONFIGURATION_TOU_SCHEDULE + SONNEN_CONFIGURATION_TOU_SCHEDULE, DEFAULT_BATTERY_LOGIN_TIMEOUT, SONNEN_LATEST_DATA_CHARGE_LEVEL, + SONNEN_CONFIGURATION_OPERATING_MODE, SONNEN_OPERATING_MODES_TO_OPERATING_MODE_NAMES, + SONNEN_CONFIGURATION_BACKUP_RESERVE, SONNEN_OPERATING_MODE_NAMES_TO_OPERATING_MODES ) from timeofuse import TimeofUseSchedule from timeofuse.timeofuse import create_time_of_use_schedule_from_json @@ -62,7 +65,6 @@ def _put(self, what, payload, isretry=False): return self._put(what, payload, isretry=True) if response.status_code != 200: - print("Aaargh", response.json()) response.raise_for_status() return response.json() @@ -101,6 +103,10 @@ def get_latest_data(self): def get_powermeter_data(self): return self._get("powermeter") + def get_status(self): + return self._get("status") + + # deprecate with 0.7.x -> get_status def get_status_data(self): return self._get("status") @@ -108,9 +114,9 @@ def get_io_data(self): return self._get("io") # POST API - def set_manual_flow(self, direction, watts): + def set_manual_flow(self, direction, watts) -> bool: response = self._post(f"setpoint/{direction}/{watts}") - return response.status_code == 201 + return response def charge_battery(self, watts:int): return self.set_manual_flow("charge", watts) @@ -119,11 +125,55 @@ def discharge_battery(self, watts:int): return self.set_manual_flow("discharge", watts) # PUT API - def set_config_item(self, item:str, value:str): + """ set item in configurations and return value as reported back + by the SonnenBatterieV2 API + """ + def set_config_item(self, item:str, value:str|int) -> str: payload = {str(item): str(value)} - return self._put(f"configurations", payload) + return self._put(f"configurations", payload)[item] # API Helpers + def get_current_charge_level(self): + return self.get_latest_data().get(SONNEN_LATEST_DATA_CHARGE_LEVEL) + + def get_operating_mode(self) -> int: + return int(self.get_config_item(SONNEN_CONFIGURATION_OPERATING_MODE)[SONNEN_CONFIGURATION_OPERATING_MODE]) + + def get_operating_mode_name(self): + operating_mode = self.get_operating_mode() + return SONNEN_OPERATING_MODES_TO_OPERATING_MODE_NAMES.get(operating_mode) + + def set_operating_mode(self, operating_mode) -> int: + return int(self.set_config_item(SONNEN_CONFIGURATION_OPERATING_MODE, operating_mode)) + + def set_operating_mode_by_name(self, operating_mode_name) -> int: + return int(self.set_operating_mode(SONNEN_OPERATING_MODE_NAMES_TO_OPERATING_MODES.get(operating_mode_name))) + + def get_battery_reserve(self): + return self.get_config_item(SONNEN_CONFIGURATION_BACKUP_RESERVE)[SONNEN_CONFIGURATION_BACKUP_RESERVE] + + def set_battery_reserve(self, reserve=5): + reserve = int(reserve) + if (reserve < 0) or (reserve > 100): + raise Exception(f"Reserve must be between 0 and 100, you specified {reserve}") + return self.set_config_item(SONNEN_CONFIGURATION_BACKUP_RESERVE, reserve) + + # set the reserve to the current battery level adjusted by the offset if provided + # (a positive offset means that the reserve will be set to more than the current level + # a negative offser means less than the current level) + # If the new reserve is less than the minimum reserve then use the minimum reserve + # the reserve will be tested to ensure it's >= 0 or <= 100 + def set_battery_reserve_relative_to_current_charge(self, offset=0, minimum_reserve=0): + current_level = self.get_current_charge_level() + target_level = current_level + offset + if target_level < minimum_reserve: + target_level = minimum_reserve + if target_level < 0: + target_level = 0 + elif target_level > 100: + target_level = 100 + return self.set_battery_reserve(target_level) + def get_tou_schedule_string(self) -> str: return self.get_config_item(SONNEN_CONFIGURATION_TOU_SCHEDULE)[SONNEN_CONFIGURATION_TOU_SCHEDULE] @@ -145,3 +195,213 @@ def clear_tou_schedule_string(self): def clear_tou_schedule_json(self): return self.set_tou_schedule_string("[]") + + +""" Async API calls """ + + +class AsyncSonnenBatterieV2: + def __init__(self, ip_address, api_token): + self._ip_address = ip_address + self._api_token = api_token + + # noinspection HttpUrlsUsage + self.base_url = f"http://{self._ip_address}/api/v2/" + + self._timeout_total = DEFAULT_BATTERY_LOGIN_TIMEOUT + self._timeout_connect = DEFAULT_CONNECT_TO_BATTERY_TIMEOUT + self._timeout_read = DEFAULT_READ_FROM_BATTERY_TIMEOUT + self._timeout = self._set_timeouts() + + self._session = None + + async def logout(self): + if self._session: + await self._session.close() + self._session = None + + def _set_timeouts(self) -> aiohttp.ClientTimeout: + return aiohttp.ClientTimeout( + total=self._timeout_total, + connect=self._timeout_connect, + sock_connect=self._timeout_connect, + sock_read=self._timeout_read, + ) + + + """ Base functions """ + + + async def _get(self, what, isretry=False) -> json: + if self._session is None: + self._session = aiohttp.ClientSession() + + url = self.base_url + what + response = await self._session.get( + url=url, + headers={'Auth-Token': self._api_token}, + timeout=self._timeout, + ) + + if not isretry and response.status == 401: + return await self._get(what, isretry=True) + if response.status != 200: + response.raise_for_status() + + return await response.json() + + + async def _put(self, what, payload, isretry=False): + if self._session is None: + self._session = aiohttp.ClientSession() + + url = self.base_url + what + response = await self._session.put( + url=url, + headers={'Auth-Token': self._api_token}, + json=payload, + timeout=self._timeout, + ) + + + if not isretry and response.status == 401: + return await self._put(what, payload, isretry=True) + if response.status != 200: + response.raise_for_status() + + return await response.json() + + + async def _post(self, what, isretry=False) -> json: + if self._session is None: + self._session = aiohttp.ClientSession() + + url = self.base_url + what + response = await self._session.post( + url=url, + headers={'Auth-Token': self._api_token}, + timeout=self._timeout, + ) + + if not isretry and response.status == 401: + return await self._post(what, isretry=True) + if response.status != 200: + response.raise_for_status() + + return await response.json() + + + """ API function """ + + # GET + async def get_config_item(self, item:str) -> json: + return await self._get(f"configurations/{item}") + + async def get_configurations(self) -> json: + return await self._get("configurations") + + async def get_battery_module_data(self) -> json: + return await self._get("battery") + + async def get_inverter_data(self) -> json: + return await self._get("inverter") + + async def get_latest_data(self) -> json: + return await self._get("latestdata") + + async def get_powermeter_data(self) -> json: + return await self._get("powermeter") + + async def get_status(self) -> json: + return await self._get("status") + + async def get_status_data(self) -> json: + return await self._get("status") + + async def get_io_data(self) -> json: + return await self._get("io") + + # POST + async def set_manual_flow(self, direction, watts) -> json: + return await self._post(f"setpoint/{direction}/{watts}") + + async def charge_battery(self, watts:int) -> json: + return await self.set_manual_flow("charge", watts) + + async def discharge_battery(self, watts:int) -> json: + return await self.set_manual_flow("discharge", watts) + + # PUT + async def set_config_item(self, item:str, value:str|int) -> json: + payload = {str(item): str(value)} + return await self._put(f"configurations", payload) + + # API Helpers + async def get_current_charge_level(self) -> int: + result = await self.get_latest_data() + return result.get(SONNEN_LATEST_DATA_CHARGE_LEVEL) + + async def get_operating_mode(self) -> int: + result = await self.get_config_item(SONNEN_CONFIGURATION_OPERATING_MODE) + return int(result[SONNEN_CONFIGURATION_OPERATING_MODE]) + + async def get_operating_mode_name(self) -> str: + opmode = await self.get_operating_mode() + return SONNEN_OPERATING_MODES_TO_OPERATING_MODE_NAMES.get(opmode) + + async def get_battery_reserve(self) -> int: + result = await self.get_config_item(SONNEN_CONFIGURATION_BACKUP_RESERVE) + return int(result[SONNEN_CONFIGURATION_BACKUP_RESERVE]) + + async def set_operating_mode(self, operating_mode) -> int: + result = await self.set_config_item( + SONNEN_CONFIGURATION_OPERATING_MODE, + operating_mode + ) + return int(result[SONNEN_CONFIGURATION_OPERATING_MODE]) + + async def set_operating_mode_by_name(self, operating_mode_name) -> int: + result = await self.set_config_item( + SONNEN_CONFIGURATION_OPERATING_MODE, + SONNEN_OPERATING_MODE_NAMES_TO_OPERATING_MODES.get(operating_mode_name) + ) + return int(result[SONNEN_CONFIGURATION_OPERATING_MODE]) + + async def set_battery_reserve(self, battery_reserve) -> json: + reserve = int(battery_reserve) + if (reserve < 0) or (reserve > 100): + raise Exception(f"Reserve must be between 0 and 100, you spcified {reserve}") + return await self.set_config_item(SONNEN_CONFIGURATION_BACKUP_RESERVE, reserve) + + async def set_battery_reserve_relative_to_current_charge(self, offset=0, min_res=0) -> json: + current_level = await self.get_current_charge_level() + target_level = current_level + offset + if target_level < min_res: + target_level = min_res + if target_level > 100: + target_level = 100 + if target_level < 0: + target_level = 0 + return await self.set_battery_reserve(target_level) + + async def get_tou_schedule_string(self) -> str: + return (await self.get_config_item(SONNEN_CONFIGURATION_TOU_SCHEDULE))[SONNEN_CONFIGURATION_TOU_SCHEDULE] + + async def get_tou_schedule_json(self) -> json: + return json.loads(await self.get_tou_schedule_string()) + + async def get_tou_schedule_object(self) -> json: + return create_time_of_use_schedule_from_json(await self.get_tou_schedule_json()) + + async def set_tou_schedule_string(self, schedule:str) -> json: + return await self.set_config_item(SONNEN_CONFIGURATION_TOU_SCHEDULE, schedule) + + async def set_tou_schedule_json(self, schedule:str|list) -> json: + # We need to convert object to valid json string + return await self.set_tou_schedule_string(json.dumps(schedule)) + + async def clear_tou_schedule_string(self) -> json: + return await self.set_tou_schedule_string("[]") + + async def clear_tou_schedule_json(self) -> json: + return await self.set_tou_schedule_string("[]") \ No newline at end of file diff --git a/test/login.py b/test/login.py index b9ed78b..eb6031a 100644 --- a/test/login.py +++ b/test/login.py @@ -1,3 +1,4 @@ -SONNEN_USERNAME="User" -SONNEN_PASSWORD="sonnenUser3552" -SONNEN_IP="192.168.1.1" +SONNEN_USERNAME = "User" +SONNEN_PASSWORD = "sonnenUser3552" +SONNEN_IP = "192.168.27.21" +SONNEN_TOKEN = "cd46d0f2-201e-4e82-882b-f590d9e754b8" \ No newline at end of file diff --git a/test/test_origional_code.py b/test/test_api_v1.py similarity index 51% rename from test/test_origional_code.py rename to test/test_api_v1.py index a03dc55..6c84972 100755 --- a/test/test_origional_code.py +++ b/test/test_api_v1.py @@ -1,17 +1,23 @@ #!/usr/bin/env python3 # I suspect that I haven't got to grips with the way phthon does things, but soppusely this will setup the path to allod for the sonnen batteri moduel to be in a separate location # To me having to do this for testing seems a horrendous hack +import asyncio import os import sys +import time script_path = os.path.realpath(os.path.dirname(__name__)) os.chdir(script_path) sys.path.append("..") from login import * from pprint import pprint -from sonnenbatterie.sonnenbatterie import sonnenbatterie +from sonnenbatterie.sonnenbatterie import sonnenbatterie, AsyncSonnenBatterie + + # this is based on the test code by rust dust -if __name__ == '__main__': + +def main(): + print("\nMain (sync)\n==========") sb = sonnenbatterie(SONNEN_USERNAME, SONNEN_PASSWORD, SONNEN_IP) print("\nPower:\n") pprint(sb.get_powermeter()) @@ -25,3 +31,28 @@ pprint(sb.get_status()) print("\nBattery:\n") pprint(sb.get_battery()) + +async def async_main(): + print("\n\nAsync main\n==========\n") + sb = AsyncSonnenBatterie(SONNEN_USERNAME, SONNEN_PASSWORD, SONNEN_IP) + print("\nPower:\n") + pprint(await sb.get_powermeter()) + print("\nBattery System:\n") + pprint(await sb.get_batterysystem()) + print("\nInverter:\n") + pprint(await sb.get_inverter()) + print("\nSystem Data:\n") + pprint(await sb.get_systemdata()) + print("\nStatus:\n") + pprint(await sb.get_status()) + print("\nBattery:\n") + pprint(await sb.get_battery()) + await sb.logout() + +if __name__ == '__main__': + start = time.time() + main() + print("\nTotal time sync:", time.time() - start) + start = time.time() + asyncio.run(async_main()) + print("\nTotal time async:", time.time() - start) diff --git a/test/test_api_v2.py b/test/test_api_v2.py new file mode 100755 index 0000000..b29615c --- /dev/null +++ b/test/test_api_v2.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +import asyncio +import os +import sys +import time + +script_path = os.path.realpath(os.path.dirname(__name__)) +os.chdir(script_path) +sys.path.append("..") + +from sonnenbatterie2.sonnenbatterie2 import SonnenBatterieV2, AsyncSonnenBatterieV2 + +from pprint import pprint + +from login import * + +def main(): + print("\nTesting SonnenBatterieV2 - Sync API\n") + sb2 = SonnenBatterieV2(SONNEN_IP, SONNEN_TOKEN) + pprint(sb2.get_configurations()) # retrieve configuration overview + pprint(sb2.get_battery_module_data()) # get battery module data + pprint(sb2.get_inverter_data()) # retrieve inverter data + pprint(sb2.get_latest_data()) # get latest date from sonnenbatterie + pprint(sb2.get_powermeter_data()) # get data from power meters + pprint(sb2.get_status()) # get overall status information + pprint(sb2.get_io_data()) # get io status + +async def async_main(): + print("\nTesting AsyncSonnenBatterieV2 - ASync API\n") + sb2 = AsyncSonnenBatterieV2(SONNEN_IP, SONNEN_TOKEN) + pprint(await sb2.get_configurations()) # retrieve configuration overview + pprint(await sb2.get_battery_module_data()) # get battery module data + pprint(await sb2.get_inverter_data()) # retrieve inverter data + pprint(await sb2.get_latest_data()) # get latest date from sonnenbatterie + pprint(await sb2.get_powermeter_data()) # get data from power meters + pprint(await sb2.get_status()) # get overall status information + pprint(await sb2.get_io_data()) # get io status + await sb2.logout() + +if __name__ == "__main__": + start = time.time() + main() + print("--- %s seconds sync ---" % (time.time() - start)) + start = time.time() + asyncio.run(async_main()) + print("--- %s seconds async ---" % (time.time() - start)) \ No newline at end of file diff --git a/test/test_battery_reserve.py b/test/test_battery_reserve.py index 35aac8c..e2f15c9 100755 --- a/test/test_battery_reserve.py +++ b/test/test_battery_reserve.py @@ -12,11 +12,11 @@ sb = sonnenbatterie(SONNEN_USERNAME, SONNEN_PASSWORD, SONNEN_IP) battery_reserve = sb.get_battery_reserve() current_charge = sb.get_current_charge_level() - print("\nConfigurations") + print("\nCurrent Configuration") pprint(sb.get_configurations()) print("\nCurrent charge level") pprint(sb.get_current_charge_level()) - print("\nBattery reserve") + print("\nCurrent battery reserve") pprint(sb.get_battery_reserve()) print("\nSetting absolute reserve to 7") pprint(sb.set_battery_reserve(7)) @@ -26,9 +26,9 @@ pprint(sb.set_battery_reserve_relative_to_current_charge()) print("\nBattery reserve with no offset against charge of "+str(current_charge)) pprint(sb.get_battery_reserve()) - pprint("\nSet relative limit (offset of -5)") + print("\nSet relative limit (offset of -5)") pprint(sb.set_battery_reserve_relative_to_current_charge(-5)) - print("\nBattery reserve with offset of -5 against charge of "+str(current_charge)) + print("\nBattery reserve with offset of -5 against charge of "+str(current_charge)) pprint(sb.get_battery_reserve()) print("\nSet relative limit (offset of 0 and minimum of 20)") pprint(sb.set_battery_reserve_relative_to_current_charge(-5, 20)) @@ -38,8 +38,9 @@ pprint(sb.set_battery_reserve_relative_to_current_charge(0, 80)) print("\nBattery reserve with offset of 0 against minimum charge of 80") pprint(sb.get_battery_reserve()) - print("\nGoing to battery reserve of 5") - pprint(sb.set_battery_reserve(5)) + print("\nGoing to battery reserve of 10") + pprint(sb.set_battery_reserve(10)) print("\nReset battery reserve") + pprint(sb.set_battery_reserve(battery_reserve)) + print("\nVerify battery reserve") pprint(sb.get_battery_reserve()) - diff --git a/test/test_battery_reserve_async.py b/test/test_battery_reserve_async.py new file mode 100755 index 0000000..e813244 --- /dev/null +++ b/test/test_battery_reserve_async.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +# I suspect that I haven't got to grips with the way phthon does things, but soppusely this will setup the path to allod for the sonnen batteri moduel to be in a separate location +# To me having to do this for testing seems a horrendous hack +import asyncio +import os, sys +script_path = os.path.realpath(os.path.dirname(__name__)) +os.chdir(script_path) +sys.path.append("..") +from login import * +from pprint import pprint +from sonnenbatterie.sonnenbatterie import AsyncSonnenBatterie + +async def main(): + sb = AsyncSonnenBatterie(SONNEN_USERNAME, SONNEN_PASSWORD, SONNEN_IP) + battery_reserve = await sb.get_battery_reserve() + current_charge = await sb.get_current_charge_level() + print("\nCurrent Configuration") + pprint(await sb.get_configurations()) + print("\nCurrent charge level") + pprint(await sb.get_current_charge_level()) + print("\nCurrent battery reserve") + pprint(await sb.get_battery_reserve()) + print("\nSetting absolute reserve to 7") + pprint(await sb.set_battery_reserve(7)) + print("\nUpdated battery reserve") + pprint(await sb.get_battery_reserve()) + print("\nSet relative limit (no offset)") + pprint(await sb.set_battery_reserve_relative_to_current_charge()) + print("\nBattery reserve with no offset against charge of "+str(current_charge)) + pprint(await sb.get_battery_reserve()) + print("\nSet relative limit (offset of -5)") + pprint(await sb.set_battery_reserve_relative_to_current_charge(-5)) + print("\nBattery reserve with offset of -5 against charge of "+str(current_charge)) + pprint(await sb.get_battery_reserve()) + print("\nSet relative limit (offset of 0 and minimum of 20)") + pprint(await sb.set_battery_reserve_relative_to_current_charge(-5, 20)) + print("\nBattery reserve with offset of 0 against minimum charge of 20") + pprint(await sb.get_battery_reserve()) + print("\nSet relative limit (offset of 0 and minimum of 80)") + pprint(await sb.set_battery_reserve_relative_to_current_charge(0, 80)) + print("\nBattery reserve with offset of 0 against minimum charge of 80") + pprint(await sb.get_battery_reserve()) + print("\nGoing to battery reserve of 10") + pprint(await sb.set_battery_reserve(10)) + print("\nReset battery reserve") + pprint(await sb.set_battery_reserve(battery_reserve)) + print("\nVerify battery reserve") + pprint(await sb.get_battery_reserve()) + await sb.logout() + +if __name__ == '__main__': + asyncio.run(main()) \ No newline at end of file diff --git a/test/test_operating_mode.py b/test/test_operating_mode.py index c10483e..37c268e 100755 --- a/test/test_operating_mode.py +++ b/test/test_operating_mode.py @@ -1,7 +1,8 @@ #!/usr/bin/env python3 # I suspect that I haven't got to grips with the way phthon does things, but soppusely this will setup the path to allod for the sonnen batteri moduel to be in a separate location # To me having to do this for testing seems a horrendous hack -import os, sys +import os +import sys script_path = os.path.realpath(os.path.dirname(__name__)) os.chdir(script_path) sys.path.append("..") @@ -9,7 +10,8 @@ from pprint import pprint from sonnenbatterie.sonnenbatterie import sonnenbatterie from sonnenbatterie.const import * -if __name__ == '__main__': + +def main(): sb = sonnenbatterie(SONNEN_USERNAME, SONNEN_PASSWORD, SONNEN_IP) operating_mode_num = sb.get_operating_mode() operating_mode_name = sb.get_operating_mode_name() @@ -25,7 +27,7 @@ pprint(sb.get_operating_mode()) print("\nNew Operating mode (name)") pprint(sb.get_operating_mode_name()) - print("Resetting operating mode num to "+operating_mode_num) + print(f"Resetting operating mode num to {operating_mode_num}") pprint(sb.set_operating_mode(operating_mode_num)) print("\nReset Operating mode (num)") pprint(sb.get_operating_mode()) @@ -37,6 +39,10 @@ pprint(sb.get_operating_mode()) print("\nNew Operating mode (name)") pprint(sb.get_operating_mode_name()) - print("Resetting operating mode num to "+operating_mode_num) + print(f"Resetting operating mode num to {operating_mode_num}") pprint(sb.set_operating_mode(operating_mode_num)) + print("\nVerifiying operating mode") + pprint(sb.get_operating_mode()) +if __name__ == '__main__': + main() diff --git a/test/test_operating_mode_api_v2.py b/test/test_operating_mode_api_v2.py new file mode 100755 index 0000000..dbba8a6 --- /dev/null +++ b/test/test_operating_mode_api_v2.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 +# I suspect that I haven't got to grips with the way phthon does things, but soppusely this will setup the path to allod for the sonnen batteri moduel to be in a separate location +# To me having to do this for testing seems a horrendous hack +import os +import sys +script_path = os.path.realpath(os.path.dirname(__name__)) +os.chdir(script_path) +sys.path.append("..") +from login import * +from pprint import pprint +from sonnenbatterie2.sonnenbatterie2 import SonnenBatterieV2 +from sonnenbatterie2.const import * + +def main(): + sb = SonnenBatterieV2(SONNEN_IP, SONNEN_TOKEN) + operating_mode_num = sb.get_operating_mode() + operating_mode_name = sb.get_operating_mode_name() + print("\nConfigurations") + pprint(sb.get_configurations()) + print("\nOperating mode (num)") + pprint(sb.get_operating_mode()) + print("\nOperating mode (name)") + pprint(sb.get_operating_mode_name()) + print("Setting operating mode num to 1") + pprint(sb.set_operating_mode(1)) + print("\nNew Operating mode (num)") + pprint(sb.get_operating_mode()) + print("\nNew Operating mode (name)") + pprint(sb.get_operating_mode_name()) + print(f"Resetting operating mode num to {operating_mode_num}") + pprint(sb.set_operating_mode(operating_mode_num)) + print("\nReset Operating mode (num)") + pprint(sb.get_operating_mode()) + print("\nReset Operating mode (name)") + pprint(sb.get_operating_mode_name()) + print("Setting operating mode name to "+SONNEN_OPERATING_MODE_AUTOMATIC_SELF_CONSUMPTION_NAME) + pprint(sb.set_operating_mode_by_name(SONNEN_OPERATING_MODE_AUTOMATIC_SELF_CONSUMPTION_NAME)) + print("\nNew Operating mode (num)") + pprint(sb.get_operating_mode()) + print("\nNew Operating mode (name)") + pprint(sb.get_operating_mode_name()) + print(f"Resetting operating mode num to {operating_mode_num}") + pprint(sb.set_operating_mode(operating_mode_num)) + print("\nVerifiying operating mode") + pprint(sb.get_operating_mode()) + +if __name__ == '__main__': + main() diff --git a/test/test_operating_mode_async.py b/test/test_operating_mode_async.py new file mode 100755 index 0000000..0ae8eed --- /dev/null +++ b/test/test_operating_mode_async.py @@ -0,0 +1,51 @@ +#!/usr/bin/env python3 +# I suspect that I haven't got to grips with the way phthon does things, but soppusely this will setup the path to allod for the sonnen batteri moduel to be in a separate location +# To me having to do this for testing seems a horrendous hack +import asyncio +import os +import sys +script_path = os.path.realpath(os.path.dirname(__name__)) +os.chdir(script_path) +sys.path.append("..") + +from login import * +from pprint import pprint +from sonnenbatterie import AsyncSonnenBatterie +from sonnenbatterie.const import * + +async def main(): + sb = AsyncSonnenBatterie(SONNEN_USERNAME, SONNEN_PASSWORD, SONNEN_IP) + operating_mode_num = await sb.get_operating_mode() + operating_mode_name = await sb.get_operating_mode_name() + print("\nConfigurations") + pprint(await sb.get_configurations()) + print("\nOperating mode (num)") + pprint(await sb.get_operating_mode()) + print("\nOperating mode (name)") + pprint(await sb.get_operating_mode_name()) + print("Setting operating mode num to 1") + pprint(await sb.set_operating_mode(1)) + print("\nNew Operating mode (num)") + pprint(await sb.get_operating_mode()) + print("\nNew Operating mode (name)") + pprint(await sb.get_operating_mode_name()) + print(f"Resetting operating mode num to {operating_mode_num}") + pprint(await sb.set_operating_mode(operating_mode_num)) + print("\nReset Operating mode (num)") + pprint(await sb.get_operating_mode()) + print("\nReset Operating mode (name)") + pprint(await sb.get_operating_mode_name()) + print("Setting operating mode name to "+SONNEN_OPERATING_MODE_AUTOMATIC_SELF_CONSUMPTION_NAME) + pprint(await sb.set_operating_mode_by_name(SONNEN_OPERATING_MODE_AUTOMATIC_SELF_CONSUMPTION_NAME)) + print("\nNew Operating mode (num)") + pprint(await sb.get_operating_mode()) + print("\nNew Operating mode (name)") + pprint(await sb.get_operating_mode_name()) + print(f"Resetting operating mode num to {operating_mode_num}") + pprint(await sb.set_operating_mode(operating_mode_num)) + print("\nVerifiying operating mode") + pprint(await sb.get_operating_mode()) + await sb.logout() + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/test/test_operating_mode_async_api_v2.py b/test/test_operating_mode_async_api_v2.py new file mode 100755 index 0000000..edd35de --- /dev/null +++ b/test/test_operating_mode_async_api_v2.py @@ -0,0 +1,51 @@ +#!/usr/bin/env python3 +# I suspect that I haven't got to grips with the way phthon does things, but soppusely this will setup the path to allod for the sonnen batteri moduel to be in a separate location +# To me having to do this for testing seems a horrendous hack +import asyncio +import os +import sys +script_path = os.path.realpath(os.path.dirname(__name__)) +os.chdir(script_path) +sys.path.append("..") + +from login import * +from pprint import pprint +from sonnenbatterie2 import AsyncSonnenBatterieV2 +from sonnenbatterie2.const import * + +async def main(): + sb = AsyncSonnenBatterieV2(SONNEN_IP, SONNEN_TOKEN) + operating_mode_num = await sb.get_operating_mode() + operating_mode_name = await sb.get_operating_mode_name() + print("\nConfigurations") + pprint(await sb.get_configurations()) + print("\nOperating mode (num)") + pprint(await sb.get_operating_mode()) + print("\nOperating mode (name)") + pprint(await sb.get_operating_mode_name()) + print("Setting operating mode num to 1") + pprint(await sb.set_operating_mode(1)) + print("\nNew Operating mode (num)") + pprint(await sb.get_operating_mode()) + print("\nNew Operating mode (name)") + pprint(await sb.get_operating_mode_name()) + print(f"Resetting operating mode num to {operating_mode_num}") + pprint(await sb.set_operating_mode(operating_mode_num)) + print("\nReset Operating mode (num)") + pprint(await sb.get_operating_mode()) + print("\nReset Operating mode (name)") + pprint(await sb.get_operating_mode_name()) + print("Setting operating mode name to "+SONNEN_OPERATING_MODE_AUTOMATIC_SELF_CONSUMPTION_NAME) + pprint(await sb.set_operating_mode_by_name(SONNEN_OPERATING_MODE_AUTOMATIC_SELF_CONSUMPTION_NAME)) + print("\nNew Operating mode (num)") + pprint(await sb.get_operating_mode()) + print("\nNew Operating mode (name)") + pprint(await sb.get_operating_mode_name()) + print(f"Resetting operating mode num to {operating_mode_num}") + pprint(await sb.set_operating_mode(operating_mode_num)) + print("\nVerifiying operating mode") + pprint(await sb.get_operating_mode()) + await sb.logout() + +if __name__ == '__main__': + asyncio.run(main()) diff --git a/test/test_retrieval_loop.py b/test/test_retrieval_loop.py index be51d6b..51c459a 100755 --- a/test/test_retrieval_loop.py +++ b/test/test_retrieval_loop.py @@ -13,7 +13,8 @@ os.chdir(script_path) sys.path.append("..") from sonnenbatterie.sonnenbatterie import sonnenbatterie -if __name__ == '__main__': + +def main(): print("Starting login") try: sb = sonnenbatterie(SONNEN_USERNAME, SONNEN_PASSWORD, SONNEN_IP) @@ -27,7 +28,7 @@ sb.set_request_connect_timeout(30) sb.set_request_read_timeout(30) counter = 0 - while True: + while True and counter < 5: counter = counter + 1 reserve = -1 current_level = -1 @@ -65,3 +66,6 @@ print("Loop "+str(counter)+", retrireved data : current_level "+str(current_level)+", reserve "+str(reserve)+", mode "+mode+", tou "+tou) time.sleep(15) + +if __name__ == '__main__': + main() \ No newline at end of file diff --git a/test/test_sbv2.py b/test/test_sbv2.py deleted file mode 100755 index c0d46cc..0000000 --- a/test/test_sbv2.py +++ /dev/null @@ -1,22 +0,0 @@ -#!/usr/bin/env python3 - -import os -import sys -from pprint import pprint - -from login import * -from sonnenbatterie import sonnenbatterie - -script_path = os.path.realpath(os.path.dirname(__name__)) -os.chdir(script_path) -sys.path.append("..") - -if __name__ == "__main__": - sb = sonnenbatterie(SONNEN_USERNAME, SONNEN_PASSWORD, SONNEN_IP) - pprint(sb.sb2.get_configurations()) # retrieve configuration overview - pprint(sb.sb2.get_battery_module_data()) # get battery module data - pprint(sb.sb2.get_inverter_data()) # retrieve inverter data - pprint(sb.sb2.get_latest_data()) # get latest date from sonnenbatterie - pprint(sb.sb2.get_powermeter_data()) # get data from power meters - pprint(sb.sb2.get_status_data()) # get overall status information - pprint(sb.sb2.get_io_data()) # get io status diff --git a/test/test_set_flow.py b/test/test_set_flow.py index 28cc5b2..9e829ef 100755 --- a/test/test_set_flow.py +++ b/test/test_set_flow.py @@ -10,34 +10,37 @@ from pprint import pprint from sonnenbatterie.sonnenbatterie import sonnenbatterie from sonnenbatterie.const import * -if __name__ == '__main__': + +def test_set_flow(): sb = sonnenbatterie(SONNEN_USERNAME, SONNEN_PASSWORD, SONNEN_IP) operating_mode_name = sb.get_operating_mode_name() current_flow = sb.get_status()["Pac_total_W"] - print("current total flow is "+str(current_flow)) - print("Setting operating mode name to "+SONNEN_OPERATING_MODE_MANUAL_NAME) + print(f"current total flow is {current_flow}") + print(f"Setting operating mode name to {SONNEN_OPERATING_MODE_MANUAL_NAME}") pprint(sb.set_operating_mode_by_name(SONNEN_OPERATING_MODE_MANUAL_NAME)) e = threading.Event() - print("Waiting 15 for things to settlt down") - e.wait(15) + print("Waiting 5 seconds for things to settle down") + e.wait(5) manual_flow=sb.get_status()["Pac_total_W"] - print("Flow on manual is"+str(manual_flow)) + print(f"Flow on manual is {manual_flow}") print("Setting a charge rate of 100") # set to a charge rate of 100 set_resp = sb.sb2.charge_battery(100) - print("response to set charge is "+str(set_resp)) - print("Waiting 15 for things to settle down") - e.wait(15) + print(f"response to set charge is {set_resp}") + print("Waiting 5 seconds for things to settle down") + e.wait(5) manual_flow=sb.get_status()["Pac_total_W"] - print("Flow with charge rate of 100 is"+str(manual_flow)) + print(f"Flow with charge rate of 100 is {manual_flow}") print("Setting a discharge rate of 100") # set to a charge rate of 100 set_resp = sb.sb2.discharge_battery(100) - print("response to set discharge is "+str(set_resp)) - print("Waiting 15 for things to settle down") - e.wait(15) + print(f"response to set discharge is {set_resp}") + print("Waiting 5 for things to settle down") + e.wait(5) manual_flow=sb.get_status()["Pac_total_W"] - print("Flow with discharge rate of 100 is"+str(manual_flow)) - print("Returning operating mode to origional mode of "+operating_mode_name) + print(f"Flow with discharge rate of 100 is {manual_flow}") + print(f"Returning operating mode to origional mode of {operating_mode_name}") pprint(sb.set_operating_mode_by_name(operating_mode_name)) +if __name__ == '__main__': + test_set_flow() \ No newline at end of file diff --git a/test/test_set_flow_async.py b/test/test_set_flow_async.py new file mode 100755 index 0000000..72b4357 --- /dev/null +++ b/test/test_set_flow_async.py @@ -0,0 +1,49 @@ +#!/usr/bin/env python3 +# I suspect that I haven't got to grips with the way phthon does things, but soppusely this will setup the path to allod for the sonnen batteri moduel to be in a separate location +# To me having to do this for testing seems a horrendous hack +import asyncio +import os, sys +import threading +script_path = os.path.realpath(os.path.dirname(__name__)) +os.chdir(script_path) +sys.path.append("..") + +from login import * +from pprint import pprint +from sonnenbatterie.sonnenbatterie import AsyncSonnenBatterie +from sonnenbatterie.const import * + +async def test_set_flow(): + sb = AsyncSonnenBatterie(SONNEN_USERNAME, SONNEN_PASSWORD, SONNEN_IP) + operating_mode_name = await sb.get_operating_mode_name() + current_flow = (await sb.get_status())["Pac_total_W"] + print(f"current total flow is {current_flow}") + print(f"Setting operating mode name to {SONNEN_OPERATING_MODE_MANUAL_NAME}") + pprint(await sb.set_operating_mode_by_name(SONNEN_OPERATING_MODE_MANUAL_NAME)) + e = threading.Event() + print("Waiting 5 seconds for things to settle down") + e.wait(5) + manual_flow = (await sb.get_status())["Pac_total_W"] + print(f"Flow on manual is {manual_flow}") + print("Setting a charge rate of 100") + # set to a charge rate of 100 + set_resp = await sb.sb2.charge_battery(100) + print(f"response to set charge is {set_resp}") + print("Waiting 5 seconds for things to settle down") + e.wait(5) + manual_flow = (await sb.get_status())["Pac_total_W"] + print(f"Flow with charge rate of 100 is {manual_flow}") + print("Setting a discharge rate of 100") + # set to a charge rate of 100 + set_resp = await sb.sb2.discharge_battery(100) + print(f"response to set discharge is {set_resp}") + print("Waiting 5 for things to settle down") + e.wait(5) + manual_flow = (await sb.get_status())["Pac_total_W"] + print(f"Flow with discharge rate of 100 is {manual_flow}") + print(f"Returning operating mode to origional mode of {operating_mode_name}") + pprint(await sb.set_operating_mode_by_name(operating_mode_name)) + await sb.logout() + +if __name__ == '__main__': + asyncio.run(test_set_flow()) \ No newline at end of file diff --git a/test/test_set_flow_v2.py b/test/test_set_flow_v2.py new file mode 100755 index 0000000..0007f94 --- /dev/null +++ b/test/test_set_flow_v2.py @@ -0,0 +1,46 @@ +#!/usr/bin/env python3 +# I suspect that I haven't got to grips with the way phthon does things, but soppusely this will setup the path to allod for the sonnen batteri moduel to be in a separate location +# To me having to do this for testing seems a horrendous hack +import os, sys +import threading +script_path = os.path.realpath(os.path.dirname(__name__)) +os.chdir(script_path) +sys.path.append("..") +from login import * +from pprint import pprint +from sonnenbatterie2.sonnenbatterie2 import SonnenBatterieV2 +from sonnenbatterie2.const import * + +def test_set_flow(): + sb = SonnenBatterieV2(SONNEN_IP, SONNEN_TOKEN) + operating_mode_name = sb.get_operating_mode_name() + current_flow = sb.get_status()["Pac_total_W"] + print(f"current total flow is {current_flow}") + print(f"Setting operating mode name to {SONNEN_OPERATING_MODE_MANUAL_NAME}") + pprint(sb.set_operating_mode_by_name(SONNEN_OPERATING_MODE_MANUAL_NAME)) + e = threading.Event() + print("Waiting 5 seconds for things to settle down") + e.wait(5) + manual_flow=sb.get_status()["Pac_total_W"] + print(f"Flow on manual is {manual_flow}") + print("Setting a charge rate of 100") + # set to a charge rate of 100 + set_resp = sb.charge_battery(100) + print(f"response to set charge is {set_resp}") + print("Waiting 5 seconds for things to settle down") + e.wait(5) + manual_flow=sb.get_status()["Pac_total_W"] + print(f"Flow with charge rate of 100 is {manual_flow}") + print("Setting a discharge rate of 100") + # set to a charge rate of 100 + set_resp = sb.discharge_battery(100) + print(f"response to set discharge is {set_resp}") + print("Waiting 5 for things to settle down") + e.wait(5) + manual_flow=sb.get_status()["Pac_total_W"] + print(f"Flow with discharge rate of 100 is {manual_flow}") + print(f"Returning operating mode to origional mode of {operating_mode_name}") + pprint(sb.set_operating_mode_by_name(operating_mode_name)) + +if __name__ == '__main__': + test_set_flow() \ No newline at end of file diff --git a/test/test_set_flow_v2_async.py b/test/test_set_flow_v2_async.py new file mode 100755 index 0000000..5b05f86 --- /dev/null +++ b/test/test_set_flow_v2_async.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 +# I suspect that I haven't got to grips with the way phthon does things, but soppusely this will setup the path to allod for the sonnen batteri moduel to be in a separate location +# To me having to do this for testing seems a horrendous hack +import asyncio +import os, sys +import threading +script_path = os.path.realpath(os.path.dirname(__name__)) +os.chdir(script_path) +sys.path.append("..") +from login import * +from pprint import pprint +from sonnenbatterie2.sonnenbatterie2 import AsyncSonnenBatterieV2 +from sonnenbatterie2.const import * + +async def test_set_flow(): + sb = AsyncSonnenBatterieV2(SONNEN_IP, SONNEN_TOKEN) + operating_mode_name = await sb.get_operating_mode_name() + current_flow = (await sb.get_status())["Pac_total_W"] + print(f"current total flow is {current_flow}") + print(f"Setting operating mode name to {SONNEN_OPERATING_MODE_MANUAL_NAME}") + pprint(await sb.set_operating_mode_by_name(SONNEN_OPERATING_MODE_MANUAL_NAME)) + e = threading.Event() + print("Waiting 5 seconds for things to settle down") + e.wait(5) + manual_flow = (await sb.get_status())["Pac_total_W"] + print(f"Flow on manual is {manual_flow}") + print("Setting a charge rate of 100") + # set to a charge rate of 100 + set_resp = await sb.charge_battery(100) + print(f"response to set charge is {set_resp}") + print("Waiting 5 seconds for things to settle down") + e.wait(5) + manual_flow = (await sb.get_status())["Pac_total_W"] + print(f"Flow with charge rate of 100 is {manual_flow}") + print("Setting a discharge rate of 100") + # set to a charge rate of 100 + set_resp = await sb.discharge_battery(100) + print(f"response to set discharge is {set_resp}") + print("Waiting 5 for things to settle down") + e.wait(5) + manual_flow = (await sb.get_status())["Pac_total_W"] + print(f"Flow with discharge rate of 100 is {manual_flow}") + print(f"Returning operating mode to origional mode of {operating_mode_name}") + pprint(await sb.set_operating_mode_by_name(operating_mode_name)) + await sb.logout() + +if __name__ == '__main__': + asyncio.run(test_set_flow()) \ No newline at end of file diff --git a/test/test_timeofuse.py b/test/test_timeofuse.py index 4f0f99f..3bd2108 100755 --- a/test/test_timeofuse.py +++ b/test/test_timeofuse.py @@ -14,13 +14,14 @@ from timeofuse import TimeofUseSchedule from timeofuse.timeofuse import create_time_of_use_entry from sonnenbatterie.const import SONNEN_OPERATING_MODE_TIME_OF_USE_NAME -if __name__ == '__main__': + +def test_timeofuse(): sb = sonnenbatterie(SONNEN_USERNAME, SONNEN_PASSWORD, SONNEN_IP) operating_mode_name = sb.get_operating_mode_name() - print("\nOrigional operating mode is "+operating_mode_name) - print("\nSetting operating mode name to "+SONNEN_OPERATING_MODE_TIME_OF_USE_NAME) + print(f"\nOriginal operating mode is {operating_mode_name}") + print(f"\nSetting operating mode name to {SONNEN_OPERATING_MODE_TIME_OF_USE_NAME}") pprint(sb.set_operating_mode_by_name(SONNEN_OPERATING_MODE_TIME_OF_USE_NAME)) - print("\nSet operating mode to "+sb.get_operating_mode_name()) + print(f"\nSet operating mode to {sb.get_operating_mode_name()}") tous = TimeofUseSchedule() print("\nExtract tou schedule as string") orig_battery_tou_string=sb.sb2.get_tou_schedule_string() @@ -49,10 +50,13 @@ print("\nExtract restored tou schedule as string") pprint(sb.sb2.get_tou_schedule_string()) ### Disabled since Sonnen seems to have changed the format - print("\nRestoring origional TOU schedule to "+orig_battery_tou_string) + print(f"\nRestoring origional TOU schedule to {orig_battery_tou_string}") sb.sb2.set_tou_schedule_json(orig_battery_tou_json) print("Sleeping for 60 seconds so you can check the change has been applied") time.sleep(60) - print("\nResetting operating mode name to "+operating_mode_name) + print(f"\nResetting operating mode name to {operating_mode_name}") pprint(sb.set_operating_mode_by_name(operating_mode_name)) - print("\nReset operating mode to "+sb.get_operating_mode_name()) + print(f"\nReset operating mode to {sb.get_operating_mode_name()}") + +if __name__ == '__main__': + test_timeofuse() \ No newline at end of file diff --git a/test/test_timeofuse_async.py b/test/test_timeofuse_async.py new file mode 100755 index 0000000..97aa2b3 --- /dev/null +++ b/test/test_timeofuse_async.py @@ -0,0 +1,64 @@ +#!/usr/bin/env python3 +# I suspect that I haven't got to grips with the way phthon does things, but soppusely this will setup the path to allod for the sonnen batteri moduel to be in a separate location +# To me having to do this for testing seems a horrendous hack +import asyncio +import os +import sys +import time +script_path = os.path.realpath(os.path.dirname(__name__)) +os.chdir(script_path) +sys.path.append("..") + +from login import * +from pprint import pprint +from sonnenbatterie import AsyncSonnenBatterie +from timeofuse import TimeofUseSchedule +from timeofuse.timeofuse import create_time_of_use_entry +from sonnenbatterie.const import SONNEN_OPERATING_MODE_TIME_OF_USE_NAME + +async def test_timeofuse(): + sb = AsyncSonnenBatterie(SONNEN_USERNAME, SONNEN_PASSWORD, SONNEN_IP) + operating_mode_name = await sb.get_operating_mode_name() + print(f"\nOriginal operating mode is {operating_mode_name}") + print(f"\nSetting operating mode name to {SONNEN_OPERATING_MODE_TIME_OF_USE_NAME}") + pprint(await sb.set_operating_mode_by_name(SONNEN_OPERATING_MODE_TIME_OF_USE_NAME)) + print(f"\nSet operating mode to {await sb.get_operating_mode_name()}") + tous = TimeofUseSchedule() + print("\nExtract tou schedule as string") + orig_battery_tou_string = await sb.sb2.get_tou_schedule_string() + orig_battery_tou_json = await sb.sb2.get_tou_schedule_json() + pprint(await sb.sb2.get_tou_schedule_string()) + print("\nExtract tou schedule as json objects") + pprint(await sb.sb2.get_tou_schedule_json()) + print("\nExtract tou schedule as text schedule") + pprint(await sb.sb2.get_tou_schedule_object()) + print("\nLoad tou schedule from json") + tous.load_tou_schedule_from_json(await sb.sb2.get_tou_schedule_json()) + print("\nLoaded tou schedule") + print(tous.get_as_tou_schedule()) + ### Disabled since Sonnen seems to have changed the format + print ("\nCreate a new TOU Schedule 10:00 - 11:00 and 14:00 - 15:00") + tous_new = TimeofUseSchedule() + tous_new.add_entry(create_time_of_use_entry(10,00,11,00)) + tous_new.add_entry(create_time_of_use_entry(14,00,15,00)) + print("Setting new TOU schedule to ") + print(tous_new.get_as_tou_schedule()) + await sb.sb2.set_tou_schedule_json(tous_new.get_as_tou_schedule()) + print("Sleeping for 60 seconds so you can check the change has been applied") + time.sleep(60) + print("\nExtract replaced tou schedule as objects") + pprint(await sb.sb2.get_tou_schedule_json()) + print("\nExtract restored tou schedule as string") + pprint(await sb.sb2.get_tou_schedule_string()) + ### Disabled since Sonnen seems to have changed the format + print(f"\nRestoring origional TOU schedule to {orig_battery_tou_string}") + await sb.sb2.set_tou_schedule_json(orig_battery_tou_json) + print("Sleeping for 60 seconds so you can check the change has been applied") + time.sleep(60) + print(f"\nResetting operating mode name to {operating_mode_name}") + pprint(await sb.set_operating_mode_by_name(operating_mode_name)) + print(f"\nReset operating mode to {await sb.get_operating_mode_name()}") + await sb.logout() + +if __name__ == '__main__': + asyncio.run(test_timeofuse()) \ No newline at end of file diff --git a/test/test_timeofuse_lib.py b/test/test_timeofuse_lib.py index d03158c..c0b623d 100755 --- a/test/test_timeofuse_lib.py +++ b/test/test_timeofuse_lib.py @@ -71,16 +71,16 @@ def do_test(description:str, test, expected_result): print ("\nExpected exception message "+str(e.args)) pprint(tous.get_as_tou_schedule()) - print ("\nAdding exacty match overlaping entry") + print ("\nAdding exact match overlaping entry") tou=create_time_of_use_entry(10,0,11,0) try: tous.add_entry(tou) - print ("\nOpps, no exception, tnis is a bug") + print ("\nOpps, no exception, this is a bug") except Exception as e: print ("\nExpected exception message "+str(e.args)) pprint(tous.get_as_tou_schedule()) - print ("\Building based on returned entry") + print ("\nBuilding based on returned entry") old_schedule = tous.get_as_tou_schedule() tous = TimeofUseSchedule() tous.load_tou_schedule(old_schedule) diff --git a/test/test_timeofuse_v2.py b/test/test_timeofuse_v2.py new file mode 100755 index 0000000..aec9801 --- /dev/null +++ b/test/test_timeofuse_v2.py @@ -0,0 +1,62 @@ +#!/usr/bin/env python3 +# I suspect that I haven't got to grips with the way phthon does things, but soppusely this will setup the path to allod for the sonnen batteri moduel to be in a separate location +# To me having to do this for testing seems a horrendous hack +import os +import sys +import time +script_path = os.path.realpath(os.path.dirname(__name__)) +os.chdir(script_path) +sys.path.append("..") + +from login import * +from pprint import pprint +from sonnenbatterie2 import SonnenBatterieV2 +from timeofuse import TimeofUseSchedule +from timeofuse.timeofuse import create_time_of_use_entry +from sonnenbatterie2.const import SONNEN_OPERATING_MODE_TIME_OF_USE_NAME + +def test_timeofuse(): + sb = SonnenBatterieV2(SONNEN_IP, SONNEN_TOKEN) + operating_mode_name = sb.get_operating_mode_name() + print(f"\nOriginal operating mode is {operating_mode_name}") + print(f"\nSetting operating mode name to {SONNEN_OPERATING_MODE_TIME_OF_USE_NAME}") + pprint(sb.set_operating_mode_by_name(SONNEN_OPERATING_MODE_TIME_OF_USE_NAME)) + print(f"\nSet operating mode to {sb.get_operating_mode_name()}") + tous = TimeofUseSchedule() + print("\nExtract tou schedule as string") + orig_battery_tou_string=sb.get_tou_schedule_string() + orig_battery_tou_json = sb.get_tou_schedule_json() + pprint(sb.get_tou_schedule_string()) + print("\nExtract tou schedule as json objects") + pprint(sb.get_tou_schedule_json()) + print("\nExtract tou schedule as text schedule") + pprint(sb.get_tou_schedule_object()) + print("\nLoad tou schedule from json") + tous.load_tou_schedule_from_json(sb.get_tou_schedule_json()) + print("\nLoaded tou schedule") + print(tous.get_as_tou_schedule()) + ### Disabled since Sonnen seems to have changed the format + print ("\nCreate a new TOU Schedule 10:00 - 11:00 and 14:00 - 15:00") + tous_new = TimeofUseSchedule() + tous_new.add_entry(create_time_of_use_entry(10,00,11,00)) + tous_new.add_entry(create_time_of_use_entry(14,00,15,00)) + print("Setting new TOU schedule to ") + print(tous_new.get_as_tou_schedule()) + sb.set_tou_schedule_json(tous_new.get_as_tou_schedule()) + print("Sleeping for 60 seconds so you can check the change has been applied") + time.sleep(60) + print("\nExtract replaced tou schedule as objects") + pprint(sb.get_tou_schedule_json()) + print("\nExtract restored tou schedule as string") + pprint(sb.get_tou_schedule_string()) + ### Disabled since Sonnen seems to have changed the format + print(f"\nRestoring origional TOU schedule to {orig_battery_tou_string}") + sb.set_tou_schedule_json(orig_battery_tou_json) + print("Sleeping for 60 seconds so you can check the change has been applied") + time.sleep(60) + print(f"\nResetting operating mode name to {operating_mode_name}") + pprint(sb.set_operating_mode_by_name(operating_mode_name)) + print(f"\nReset operating mode to {sb.get_operating_mode_name()}") + +if __name__ == '__main__': + test_timeofuse() \ No newline at end of file diff --git a/test/test_timeofuse_v2_async.py b/test/test_timeofuse_v2_async.py new file mode 100755 index 0000000..4e99646 --- /dev/null +++ b/test/test_timeofuse_v2_async.py @@ -0,0 +1,64 @@ +#!/usr/bin/env python3 +# I suspect that I haven't got to grips with the way phthon does things, but soppusely this will setup the path to allod for the sonnen batteri moduel to be in a separate location +# To me having to do this for testing seems a horrendous hack +import asyncio +import os +import sys +import time +script_path = os.path.realpath(os.path.dirname(__name__)) +os.chdir(script_path) +sys.path.append("..") + +from login import * +from pprint import pprint +from sonnenbatterie2 import AsyncSonnenBatterieV2 +from timeofuse import TimeofUseSchedule +from timeofuse.timeofuse import create_time_of_use_entry +from sonnenbatterie2.const import SONNEN_OPERATING_MODE_TIME_OF_USE_NAME + +async def test_timeofuse(): + sb = AsyncSonnenBatterieV2(SONNEN_IP, SONNEN_TOKEN) + operating_mode_name = await sb.get_operating_mode_name() + print(f"\nOriginal operating mode is {operating_mode_name}") + print(f"\nSetting operating mode name to {SONNEN_OPERATING_MODE_TIME_OF_USE_NAME}") + pprint(await sb.set_operating_mode_by_name(SONNEN_OPERATING_MODE_TIME_OF_USE_NAME)) + print(f"\nSet operating mode to {await sb.get_operating_mode_name()}") + tous = TimeofUseSchedule() + print("\nExtract tou schedule as string") + orig_battery_tou_string = await sb.get_tou_schedule_string() + orig_battery_tou_json = await sb.get_tou_schedule_json() + pprint(await sb.get_tou_schedule_string()) + print("\nExtract tou schedule as json objects") + pprint(await sb.get_tou_schedule_json()) + print("\nExtract tou schedule as text schedule") + pprint(await sb.get_tou_schedule_object()) + print("\nLoad tou schedule from json") + tous.load_tou_schedule_from_json(await sb.get_tou_schedule_json()) + print("\nLoaded tou schedule") + print(tous.get_as_tou_schedule()) + ### Disabled since Sonnen seems to have changed the format + print ("\nCreate a new TOU Schedule 10:00 - 11:00 and 14:00 - 15:00") + tous_new = TimeofUseSchedule() + tous_new.add_entry(create_time_of_use_entry(10,00,11,00)) + tous_new.add_entry(create_time_of_use_entry(14,00,15,00)) + print("Setting new TOU schedule to ") + print(tous_new.get_as_tou_schedule()) + await sb.set_tou_schedule_json(tous_new.get_as_tou_schedule()) + print("Sleeping for 60 seconds so you can check the change has been applied") + time.sleep(60) + print("\nExtract replaced tou schedule as objects") + pprint(await sb.get_tou_schedule_json()) + print("\nExtract restored tou schedule as string") + pprint(await sb.get_tou_schedule_string()) + ### Disabled since Sonnen seems to have changed the format + print(f"\nRestoring origional TOU schedule to {orig_battery_tou_string}") + await sb.set_tou_schedule_json(orig_battery_tou_json) + print("Sleeping for 60 seconds so you can check the change has been applied") + time.sleep(60) + print(f"\nResetting operating mode name to {operating_mode_name}") + pprint(await sb.set_operating_mode_by_name(operating_mode_name)) + print(f"\nReset operating mode to {await sb.get_operating_mode_name()}") + await sb.logout() + +if __name__ == '__main__': + asyncio.run(test_timeofuse()) \ No newline at end of file diff --git a/timeofuse/timeofuse.py b/timeofuse/timeofuse.py index e42aedc..251c844 100644 --- a/timeofuse/timeofuse.py +++ b/timeofuse/timeofuse.py @@ -116,7 +116,7 @@ def _add_entry(self, entry): raise Exception("End time cannot be before start time") for i in self._schedule_entries: if i.is_overlapping(entry): - raise Exception("Unable to add entry, overlaps with exisitngv entry") + raise Exception("Unable to add entry, overlaps with exisiting entry") self._schedule_entries.append(entry) # maintains this as a sotred list based on the start time, this lets us compare properly self._schedule_entries = sorted(self._schedule_entries, key=lambda l_entry: l_entry.start_time)