diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..fee5039 --- /dev/null +++ b/.gitignore @@ -0,0 +1,23 @@ +# IDEs & editors +/.idea/ + +# cache +*.py[cod] +__pycache__/ + +# virtualenv +/virtualenv/ +/venv/ + +# packaging +/daikinapi.egg-info/ +/dist/ +/build/ + +# testing +/.cache/ +/pytestdebug.log +/.pytest_cache/ +/.tox/ +/.coverage +/htmlcov/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..564f92b --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2021 Aarno Aukia + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..e372c75 --- /dev/null +++ b/README.md @@ -0,0 +1,16 @@ +# Balboa ControlMySpa Whirlpool cloud API + +## Usage + +see example.py for runnable example + +```python +from controlmyspa import ControlMySpa + +API = ControlMySpa("user@example.com", "myverysecretpassword") +pprint.pprint(API._info) +``` + +## References + +Based on the JavaScript library https://gitlab.com/VVlasy/controlmyspajs diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..6e7981c --- /dev/null +++ b/__init__.py @@ -0,0 +1,7 @@ +""" +Import class from file +""" + +from controlmyspa.controlmyspa import ControlMySpa + +__all__ = ["ControlMySpa"] diff --git a/controlmyspa.py b/controlmyspa.py new file mode 100644 index 0000000..97609ab --- /dev/null +++ b/controlmyspa.py @@ -0,0 +1,348 @@ +""" +Python module to get metrics from and control Balboa ControlMySpa whirlpools +""" + +import requests + + +class ControlMySpa: + """ + Class representing Balboa ControlMySpa whirlpools + """ + + _email = None + _password = None + + def __init__(self, email, password): + """ + Initialize connection to Balboa ControlMySpa cloud API + :param email: email address used to log in + :param password: password used to log in + """ + self._email = email + self._password = password + + # log in and fetch pool info + self._get_idm() + self._do_login() + self._get_whoami() + self._get_info() + + def _get_idm(self): + """ + Get URL and basic auth to log in to IDM + """ + response = requests.get("https://iot.controlmyspa.com/idm/tokenEndpoint") + response.raise_for_status() + self._idm = response.json() + return self._idm + + def _do_login(self): + """ + Log in and get API access tokens + """ + response = requests.post( + self._idm["_links"]["tokenEndpoint"]["href"], + data={ + "grant_type": "password", + "password": self._password, + "scope": "openid user_name", + "username": self._email, + }, + auth=(self._idm["mobileClientId"], self._idm["mobileClientSecret"]), + ) + response.raise_for_status() + self._token = response.json() + return self._token + + def _get_whoami(self): + """ + Get information about the logged in user, the owner + """ + response = requests.get( + self._idm["_links"]["whoami"]["href"], + headers={"Authorization": "Bearer " + self._token["access_token"]}, + ) + response.raise_for_status() + self._user = response.json() + return self._user + + def _get_info(self): + """ + Get all the details for the whirlpool of the logged in user + """ + response = requests.get( + "https://iot.controlmyspa.com/mobile/spas/search/findByUsername", + params={"username": self._email}, + headers={"Authorization": "Bearer " + self._token["access_token"]}, + ) + response.raise_for_status() + self._info = response.json() + return self._info + + @property + def current_temp(self): + """ + Get current pool temperature + """ + # update fresh info + # self._get_info() + if self._info["currentState"]["celsius"]: + return round( + (float(self._info["currentState"]["currentTemp"]) - 32) * 5 / 9, 1 + ) + return float(self._info["currentState"]["currentTemp"]) + + @property + def desired_temp(self): + """ + Get desired pool temperature + """ + # update fresh info + # self._get_info() + if self._info["currentState"]["celsius"]: + return round( + (float(self._info["currentState"]["desiredTemp"]) - 32) * 5 / 9, 1 + ) + return float(self._info["currentState"]["desiredTemp"]) + + @desired_temp.setter + def desired_temp(self, temperature): + """ + Set the desired temperature of the whirlpool + :param temperature: temperature, in celsius if the whirlpool is set to celsius or + in fahrenheit if the whirlpool is set to fahrenheit + """ + if self._info["currentState"]["celsius"]: + # convert to fahrenheit since the API always expects fahrenheit + temperature = round(temperature / 5 * 9 + 32, 1) + response = requests.post( + "https://iot.controlmyspa.com/mobile/control/" + + self._info["_id"] + + "/setDesiredTemp", + json={"desiredTemp": temperature}, + headers={"Authorization": "Bearer " + self._token["access_token"]}, + ) + response.raise_for_status() + # update the local info + self._get_info() + return response.json() + + @property + def temp_range(self): + """ + Get temp range HIGH (True) or LOW (False) + """ + # update fresh info + # self._get_info() + return self._info["currentState"]["tempRange"] == "HIGH" + + @temp_range.setter + def temp_range(self, temp_range=True): + """ + Set temp range HIGH or LOW + :param temp_range: True for HIGH, False for LOW + """ + response = requests.post( + "https://iot.controlmyspa.com/mobile/control/" + + self._info["_id"] + + "/setTempRange", + json={"desiredState": ("HIGH" if temp_range else "LOW")}, + headers={"Authorization": "Bearer " + self._token["access_token"]}, + ) + response.raise_for_status() + # update the local info + self._get_info() + return response.json() + + @property + def panel_lock(self): + """ + Get panel lock status, Locked = True, unlocked = False + """ + # update fresh info + # self._get_info() + return self._info["currentState"]["panelLock"] + + @panel_lock.setter + def panel_lock(self, lock=True): + """ + Set panel lock + :param lock: True for locked, False for unlocked + """ + response = requests.post( + "https://iot.controlmyspa.com/mobile/control/" + + self._info["_id"] + + "/setPanel", + json={"desiredState": ("LOCK_PANEL" if lock else "UNLOCK_PANEL")}, + headers={"Authorization": "Bearer " + self._token["access_token"]}, + ) + response.raise_for_status() + # update the local info + self._get_info() + return response.json() + + def get_jet(self, jet_number=0): + """ + get jet state HIGH = True, OFF = False + :param jet_number: My pool has jets 0, 1 and 2 + """ + # update fresh info + # self._get_info() + return [ + x["value"] == "HIGH" + for x in self._info["currentState"]["components"] + if x["componentType"] == "PUMP" and x["port"] == str(jet_number) + ][0] + + def set_jet(self, jet_number=0, state=False): + """ + Enable/disable jet + :param jet_number: My pool has jets 0, 1 and 2 + :param state: False to furn off, True to turn on + """ + response = requests.post( + "https://iot.controlmyspa.com/mobile/control/" + + self._info["_id"] + + "/setJetState", + json={ + "desiredState": ("HIGH" if state else "OFF"), + "deviceNumber": jet_number, + "originatorId": "optional-Jet", + }, + headers={"Authorization": "Bearer " + self._token["access_token"]}, + ) + response.raise_for_status() + # update the local info + self._get_info() + return response.json() + + @property + def jets(self): + """ + get an array of jets True/False (ON/OFF) status + """ + return [ + x["value"] == "HIGH" + for x in self._info["currentState"]["components"] + if x["componentType"] == "PUMP" + ] + + @jets.setter + def jets(self, array): + """ + set jets ON/OFF based on array of True/False + :param array: array of True/False + """ + for i, state in enumerate(array): + self.set_jet(i, state) + + def get_blower(self, blower_number=0): + """ + get blower state HIGH = True, OFF = False + :param blower_number: My pool has no blowers + """ + # update fresh info + # self._get_info() + return [ + x["value"] == "HIGH" + for x in self._info["currentState"]["components"] + if x["componentType"] == "BLOWER" and x["port"] == str(blower_number) + ][0] + + def set_blower(self, blower_number=0, state=False): + """ + Enable/disable jet + :param jet_number: My pool has blowers 0, 1 and 2 + :param state: False to furn off, True to turn on + """ + response = requests.post( + "https://iot.controlmyspa.com/mobile/control/" + + self._info["_id"] + + "/setBlowerState", + json={ + "desiredState": ("HIGH" if state else "OFF"), + "deviceNumber": blower_number, + "originatorId": "optional-Blower", + }, + headers={"Authorization": "Bearer " + self._token["access_token"]}, + ) + response.raise_for_status() + # update the local info + self._get_info() + return response.json() + + @property + def blowers(self): + """ + get an array of blowers True/False (ON/OFF) status + """ + return [ + x["value"] == "HIGH" + for x in self._info["currentState"]["components"] + if x["componentType"] == "BLOWER" + ] + + @blowers.setter + def blowers(self, array): + """ + set blowers ON/OFF based on array of True/False + :param array: array of True/False + """ + for i, state in enumerate(array): + self.set_blower(i, state) + + def get_light(self, light_number=0): + """ + get light state HIGH = True, OFF = False + :param light_number: My pool has light 0 + """ + # update fresh info + # self._get_info() + return [ + x["value"] == "HIGH" + for x in self._info["currentState"]["components"] + if x["componentType"] == "LIGHT" and x["port"] == str(light_number) + ][0] + + def set_light(self, light_number=0, state=False): + """ + Enable/disable light + :param jet_number: My pool has lights 0, 1 and 2 + :param state: False to furn off, True to turn on + """ + response = requests.post( + "https://iot.controlmyspa.com/mobile/control/" + + self._info["_id"] + + "/setLightState", + json={ + "desiredState": ("HIGH" if state else "OFF"), + "deviceNumber": light_number, + "originatorId": "optional-Light", + }, + headers={"Authorization": "Bearer " + self._token["access_token"]}, + ) + response.raise_for_status() + # update the local info + self._get_info() + return response.json() + + @property + def lights(self): + """ + get an array of lights True/False (ON/OFF) status + """ + return [ + x["value"] == "HIGH" + for x in self._info["currentState"]["components"] + if x["componentType"] == "LIGHT" + ] + + @lights.setter + def lights(self, array): + """ + set lights ON/OFF based on array of True/False + :param array: array of True/False + """ + for i, state in enumerate(array): + self.set_light(i, state) diff --git a/example.py b/example.py new file mode 100644 index 0000000..6d6d1f0 --- /dev/null +++ b/example.py @@ -0,0 +1,53 @@ +""" +Example usage of controlmyspa module + +use e.g. with "python example.py user@example.com myverysecretpassword" +""" +import argparse +import logging +import pprint + +from controlmyspa import ControlMySpa + +PARSER = argparse.ArgumentParser(description="Get metrics from Balboa Controlmyspa") +PARSER.add_argument( + "-v", "--verbose", help="enable debug logging", action="store_true", default=False, +) +PARSER.add_argument("email", help="email to log in to controlmyspa.com") +PARSER.add_argument("password", help="password to log in to controlmyspa.com") +ARGS = PARSER.parse_args() + +LOGFORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" + +if ARGS.verbose: + logging.basicConfig(level=logging.DEBUG, format=LOGFORMAT) +else: + logging.basicConfig(level=logging.INFO, format=LOGFORMAT) + logging.getLogger("requests.packages.urllib3.connectionpool").setLevel( + logging.WARNING + ) + +logging.debug("starting with arguments: %s", ARGS) + +API = ControlMySpa(ARGS.email, ARGS.password) +print("current temp", API.current_temp) +print("desired temp", API.desired_temp) + +# API.desired_temp = 36 if API.desired_temp == 37 else 37 + +print("temp range", API.temp_range) +print("panel lock", API.panel_lock) + +print("lights", API.lights) + +# toggle lights +# API.lights = [not x for x in API.lights] + +print("jets", API.jets) + +# toggle jets +# API.set_jet(0, not API.get_jet(0)) +# API.set_jet(1, not API.get_jet(1)) +# API.set_jet(2, not API.get_jet(2)) + +print("blowers", API.blowers) diff --git a/requirements.in b/requirements.in new file mode 100644 index 0000000..76c28c5 --- /dev/null +++ b/requirements.in @@ -0,0 +1,2 @@ +requests +responses diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..4073d9b --- /dev/null +++ b/requirements.txt @@ -0,0 +1,13 @@ +# +# This file is autogenerated by pip-compile +# To update, run: +# +# pip-compile requirements.in +# +certifi==2021.5.30 # via requests +charset-normalizer==2.0.4 # via requests +idna==3.2 # via requests +requests==2.26.0 +responses==0.14.0 +six==1.16.0 # via responses +urllib3==1.26.6 # via requests, responses diff --git a/setup.cfg b/setup.cfg new file mode 100644 index 0000000..9886c7c --- /dev/null +++ b/setup.cfg @@ -0,0 +1,6 @@ +[bdist_wheel] +# This flag says to generate wheels that support both Python 2 and Python +# 3. If your code will not run unchanged on both Python 2 and 3, you will +# need to generate separate wheels for each Python version that you +# support. +universal=1 diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..5a121f6 --- /dev/null +++ b/setup.py @@ -0,0 +1,42 @@ +""" +controlmyspa python module manifest +""" +from os.path import abspath, dirname, join +from setuptools import setup + +__version__ = "1.0.0" + + +def read_file(filename): + """Get the contents of a file""" + here = abspath(dirname(__file__)) + with open(join(here, filename), encoding="utf-8") as file: + return file.read() + + +setup( + name="controlmyspa", + version=__version__, + description="Get metrics and control Balboa Controlmyspa whirlpool", + long_description=read_file("README.md"), + long_description_content_type="text/markdown", + packages=["controlmyspa"], + package_dir={"controlmyspa": "."}, + keywords=["Balboa", "Controlmyspa", "Whirlpool", "API"], + classifiers=[ + "Intended Audience :: Developers", + "License :: OSI Approved :: BSD License", + "Programming Language :: Python :: 3.5", + "Programming Language :: Python :: 3.6", + "Programming Language :: Python :: 3.7", + "Programming Language :: Python :: 3.8", + "Programming Language :: Python :: 3.9", + ], + url="https://github.com/arska/controlmyspa", + author="Aarno Aukia", + author_email="aarno@aukia.com", + license="MIT", + python_requires=">=3.5", + extras_require={"dev": ["tox"]}, + install_requires=["requests>=2"], +) diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_controlmyspa.py b/tests/test_controlmyspa.py new file mode 100644 index 0000000..fa9e05f --- /dev/null +++ b/tests/test_controlmyspa.py @@ -0,0 +1,721 @@ +import datetime +import json +import unittest +import base64 +import pprint + +from controlmyspa import ControlMySpa +import responses + + +def suite(): + """Define all the tests of the module.""" + suite = unittest.TestSuite() + suite.addTest(unittest.makeSuite(ControlMySpaTestCase)) + return suite + + +class ControlMySpaTestCase(unittest.TestCase): + exampleusername = "example@example.com" + examplepassword = "password123" + + def setUp(self): + self.responses = responses.RequestsMock() + self.responses.start() + self.idm = { + "_links": { + "refreshEndpoint": { + "href": "https://idmqa.controlmyspa.com/oxauth/restv1/token" + }, + "tokenEndpoint": { + "href": "https://idmqa.controlmyspa.com/oxauth/restv1/token" + }, + "whoami": {"href": "https://iot.controlmyspa.com/mobile/auth/whoami"}, + }, + "mobileClientId": "@!1234.5678.9ABC.DEF0!1234!5678.9ABC!DEF0!1234.5678", + "mobileClientSecret": "mobile", + } + self.responses.add( + responses.GET, + "https://iot.controlmyspa.com/idm/tokenEndpoint", + status=200, + json=self.idm, + ) + self.token = { + "access_token": "12345678-9abc-def0-1234-56789abcdef0", + "expires_in": 14399, + "id_token": "ewogICJraWQiOiAiMTIzNDU2NzgtOWFiYy1kZWYwLTEyMzQtNTY3ODlhYmNkZWYwIiwKICAidHlwIjogIkpXVCIsCiAgImFsZyI6ICJSUzI1NiIKfQ.ewogICJpc3MiOiAiaHR0cHM6Ly9pZG1xYS5jb250cm9sbXlzcGEuY29tIiwKICAiYXVkIjogIkAhMTIzNC41Njc4LjlBQkMuREVGMCExMjM0ITU2NzguOUFCQyFERUYwITEyMzQuNTY3OCIsCiAgImV4cCI6IDE2MzE3MjI5MjQsCiAgImlhdCI6IDE2MzE2MzY1MjQsCiAgIm94T3BlbklEQ29ubmVjdFZlcnNpb24iOiAib3BlbmlkY29ubmVjdC0xLjAiLAogICJzdWIiOiAiQCExMjM0LjU2NzguOUFCQy5ERUYwITEyMzQhNTY3OC45QUJDIURFRjAhMTIzNC41Njc4Igp9.yaWUF4vghbDRqS7ceFK55NPbOQvQIO_F-FIZmQkdCO2XQKtRq_X2nNmqYVEf35YlAsB09AD7P-NSPD_NPPwJeD1v0EECWZdI7qGzaegX34eM7aJU0j3LHHKT28n68AZ9NsfOuoQDLUmHUtXLkPb3522iHqqWclfZLqMX_Ug5vWej9IujFsHPLct8_a4OR7Xt07yPriKPC__qjSl_qFVFeJwoC2bNSh8kUja1p7G7e_cqUTEydK7ZVQxkpqG_HLOBjY3IoJBkRal2Rsh8PtgUhE0SJJJlLuYUWAW2DpU6ceFTA1ocGjv1c7ShDoD2zCedgynKIvogkpbdnoBzkECyOA", + "refresh_token": "12345678-9abc-def0-1234-56789abcdef0", + "scope": "openid user_name", + "token_type": "bearer", + } + self.responses.add( + responses.POST, + "https://idmqa.controlmyspa.com/oxauth/restv1/token", + status=200, + json=self.token, + match=[ + responses.matchers.urlencoded_params_matcher( + { + "grant_type": "password", + "password": self.examplepassword, + "scope": "openid user_name", + "username": self.exampleusername, + } + ) + ], + ) + self.user = { + "_id": "0123456789abcdef01234567", + "_links": { + "logo": { + "href": "https://iot.controlmyspa.com/mobile/attachments/0123456789abcdef01234567" + }, + "self": { + "href": "https://iot.controlmyspa.com/mobile/users/0123456789abcdef01234567" + }, + "spa": { + "href": "https://iot.controlmyspa.com/mobile/spas/0123456789abcdef01234567{?projection}", + "templated": True, + }, + "user": { + "href": "https://iot.controlmyspa.com/mobile/users/0123456789abcdef01234567" + }, + }, + "active": True, + "address": { + "address1": "Streetaddress 123", + "city": "City ", + "country": "Country", + "zip": "12345", + }, + "dealerId": "0123456789abcdef01234567", + "dealerName": "MySpaDealer", + "deviceToken": "0123456789abcdef01234567", + "deviceType": "IOS", + "email": self.exampleusername, + "firstName": "Firstname", + "fullName": "Firstname Lastname", + "lastName": "Lastname", + "oemId": "0123456789abcdef01234567", + "oemName": "The Spa Producing Company Ltd", + "password": self.examplepassword, + "phone": "00123456789", + "roles": ["OWNER"], + "spaId": "0123456789abcdef01234567", + "username": self.exampleusername, + } + self.responses.add( + responses.GET, + "https://iot.controlmyspa.com/mobile/auth/whoami", + status=200, + json=self.user, + ) + self.info = { + "_id": "0123456789abcdef01234567", + "_links": { + "AC Current measurements": { + "href": "https://iot.controlmyspa.com/mobile/spas/0123456789abcdef01234567/measurements?measurementType=AC_CURRENT" + }, + "Ambient Temp measurements": { + "href": "https://iot.controlmyspa.com/mobile/spas/0123456789abcdef01234567/measurements?measurementType=AMBIENT_TEMP" + }, + "events": { + "href": "https://iot.controlmyspa.com/mobile/spas/0123456789abcdef01234567/events" + }, + "faultLogs": { + "href": "https://iot.controlmyspa.com/mobile/spas/0123456789abcdef01234567/faultLogs" + }, + "owner": { + "href": "https://iot.controlmyspa.com/mobile/users/0123456789abcdef01234567" + }, + "recipes": { + "href": "https://iot.controlmyspa.com/mobile/spas/0123456789abcdef01234567/recipes" + }, + "self": { + "href": "https://iot.controlmyspa.com/mobile/spas/0123456789abcdef01234567" + }, + "spa": { + "href": "https://iot.controlmyspa.com/mobile/spas/0123456789abcdef01234567{?projection}", + "templated": True, + }, + "spaTemplate": { + "href": "https://iot.controlmyspa.com/mobile/spaTemplates/0123456789abcdef01234567" + }, + "turnOffSpa": { + "href": "https://iot.controlmyspa.com/mobile/spas/0123456789abcdef01234567/recipes/0123456789abcdef01234567/run" + }, + "wifiStats": { + "href": "https://iot.controlmyspa.com/mobile/spas/0123456789abcdef01234567/wifiStats" + }, + }, + "buildNumber": "1101/101", + "currentState": { + "abdisplay": False, + "allSegsOn": False, + "bluetoothStatus": "NOT_PRESENT", + "celsius": True, + "cleanupCycle": False, + "components": [ + { + "componentType": "HEATER", + "materialType": "HEATER", + "name": "HEATER", + "port": "0", + "registeredTimestamp": "2021-09-14T17:35:17.430+0000", + "value": "OFF", + }, + { + "availableValues": ["OFF", "ON", "DISABLED"], + "componentType": "FILTER", + "durationMinutes": 120, + "hour": 20, + "materialType": "FILTER", + "name": "FILTER", + "port": "0", + "registeredTimestamp": "2021-09-14T17:35:17.430+0000", + "value": "ON", + }, + { + "availableValues": ["OFF", "ON", "DISABLED"], + "componentType": "FILTER", + "durationMinutes": 120, + "hour": 8, + "materialType": "FILTER", + "name": "FILTER", + "port": "1", + "registeredTimestamp": "2021-09-14T17:35:17.430+0000", + "value": "OFF", + }, + { + "availableValues": ["OFF", "ON"], + "componentType": "OZONE", + "materialType": "OZONE", + "name": "OZONE", + "registeredTimestamp": "2021-09-14T17:35:17.430+0000", + "value": "ON", + }, + { + "availableValues": ["OFF", "HIGH"], + "componentType": "PUMP", + "materialType": "PUMP", + "name": "PUMP", + "port": "0", + "registeredTimestamp": "2021-09-14T17:35:17.430+0000", + "targetValue": "OFF", + "value": "OFF", + }, + { + "availableValues": ["OFF", "HIGH"], + "componentType": "PUMP", + "materialType": "PUMP", + "name": "PUMP", + "port": "1", + "registeredTimestamp": "2021-09-14T17:35:17.430+0000", + "targetValue": "OFF", + "value": "OFF", + }, + { + "availableValues": ["OFF", "HIGH"], + "componentType": "PUMP", + "materialType": "PUMP", + "name": "PUMP", + "port": "2", + "registeredTimestamp": "2021-09-14T17:35:17.430+0000", + "targetValue": "OFF", + "value": "OFF", + }, + { + "availableValues": ["OFF", "HIGH"], + "componentType": "CIRCULATION_PUMP", + "materialType": "CIRCULATION_PUMP", + "name": "CIRCULATION_PUMP", + "registeredTimestamp": "2021-09-14T17:35:17.430+0000", + "value": "HIGH", + }, + { + "availableValues": ["OFF", "HIGH"], + "componentType": "LIGHT", + "materialType": "LIGHT", + "name": "LIGHT", + "port": "0", + "registeredTimestamp": "2021-09-14T17:35:17.446+0000", + "targetValue": "HIGH", + "value": "OFF", + }, + { + "_links": { + "component": { + "href": "https://iot.controlmyspa.com/mobile/components/0123456789abcdef01234567" + } + }, + "componentId": "0123456789abcdef01234567", + "componentType": "GATEWAY", + "materialType": "GATEWAY", + "name": "ControlMySpa Gateway", + "registeredTimestamp": "2021-09-14T17:35:17.446+0000", + "serialNumber": "12345***1234567890", + }, + { + "componentType": "CONTROLLER", + "materialType": "CONTROLLER", + "name": "CONTROLLER", + "registeredTimestamp": "2021-09-14T17:35:17.446+0000", + }, + ], + "controllerType": "NGSC", + "currentTemp": "100.4", + "demoMode": False, + "desiredTemp": "99.5", + "ecoMode": False, + "elapsedTimeDisplay": False, + "errorCode": 0, + "ethernetPluggedIn": True, + "heatExternallyDisabled": False, + "heaterCooling": False, + "heaterMode": "READY", + "hour": 20, + "invert": False, + "latchingMessage": False, + "lightCycle": False, + "messageSeverity": 0, + "military": True, + "minute": 35, + "offlineAlert": False, + "online": True, + "overrangeEnabled": False, + "panelLock": False, + "panelMode": "PANEL_MODE_NGSC", + "primaryTZLStatus": "TZL_NOT_PRESENT", + "primingMode": False, + "repeat": False, + "rs485AcquiredAddress": 16, + "rs485ConnectionActive": True, + "runMode": "Ready", + "secondaryFiltrationMode": "AWAY", + "secondaryTZLStatus": "TZL_NOT_PRESENT", + "settingsLock": False, + "setupParams": { + "drainModeEnabled": False, + "gfciEnabled": False, + "highRangeHigh": 104, + "highRangeLow": 80, + "lastUpdateTimestamp": "1970-01-01T00:00:03.436+0000", + "lowRangeHigh": 99, + "lowRangeLow": 50, + }, + "shouldShowAlert": False, + "soakMode": False, + "soundAlarm": False, + "spaOverheatDisabled": False, + "specialTimeouts": False, + "staleTimestamp": "2021-09-14T17:37:02.430+0000", + "stirring": False, + "swimSpaMode": "SWIM_MODE_OTHER", + "swimSpaModeChanging": False, + "systemInfo": { + "controllerSoftwareVersion": "M100_226 V43.0", + "currentSetup": 7, + "dipSwitches": [ + {"on": False, "slotNumber": 1}, + {"on": False, "slotNumber": 2}, + {"on": False, "slotNumber": 3}, + {"on": True, "slotNumber": 4}, + {"on": False, "slotNumber": 5}, + {"on": False, "slotNumber": 6}, + {"on": False, "slotNumber": 7}, + {"on": False, "slotNumber": 8}, + {"on": False, "slotNumber": 9}, + {"on": False, "slotNumber": 10}, + ], + "heaterPower": 3, + "lastUpdateTimestamp": "1970-01-01T00:00:06.449+0000", + "mfrSSID": 100, + "modelSSID": 226, + "swSignature": -148899849, + "versionSSID": 43, + }, + "targetDesiredTemp": "96.8", + "tempLock": False, + "tempRange": "HIGH", + "testMode": False, + "timeNotSet": False, + "tvLiftState": 0, + "uiCode": 0, + "uiSubCode": 0, + "updateIntervalSeconds": 0, + "uplinkTimestamp": "2021-09-14T17:35:17.430+0000", + "wifiUpdateIntervalSeconds": 0, + }, + "dealerId": "0123456789abcdef01234567", + "dealerName": "MySpaDealer", + "demo": False, + "manufacturedDate": "2021-09-07T14:27:06.851+0000", + "model": "Default Spa", + "oemId": "0123456789abcdef01234567", + "oemName": "The Spa Producing Company Ltd", + "online": True, + "owner": { + "_id": "0123456789abcdef01234567", + "_links": { + "address": { + "href": "https://iot.controlmyspa.com/mobile/addresses/0123456789abcdef01234567" + }, + "self": { + "href": "https://iot.controlmyspa.com/mobile/users/0123456789abcdef01234567" + }, + }, + "active": True, + "address": { + "address1": "Streetaddress 123", + "city": "City ", + "country": "Country", + "zip": "12345", + }, + "dealerId": "0123456789abcdef01234567", + "dealerName": "MySpaDealer", + "deviceToken": "0123456789abcdef01234567", + "deviceType": "IOS", + "email": self.exampleusername, + "firstName": "Firstname", + "fullName": "Firstname Lastname", + "lastName": "Lastname", + "oemId": "0123456789abcdef01234567", + "oemName": "The Spa Producing Company Ltd", + "password": self.examplepassword, + "phone": "00123456789", + "roles": ["OWNER"], + "spaId": "0123456789abcdef01234567", + "username": self.exampleusername, + }, + "p2pAPSSID": "CMS_SPA_12345***1234567890", + "productName": "Default Spa", + "registrationDate": "2021-09-06T13:13:29.705+0000", + "salesDate": "2021-09-07T14:27:18.117+0000", + "serialNumber": "12345***1234567890", + "sold": "true", + "templateId": "0123456789abcdef01234567", + "transactionCode": "A1B2C3D4", + } + self.responses.add( + responses.GET, + "https://iot.controlmyspa.com/mobile/spas/search/findByUsername?username=" + + self.exampleusername, + status=200, + json=self.info, + ) + + self.addCleanup(self.responses.stop) + self.addCleanup(self.responses.reset) + + def test_init_config(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + self.assertEqual(cms._email, self.exampleusername) + self.assertEqual(cms._password, self.examplepassword) + # there should have been 4 API calls + self.assertEqual(len(self.responses.calls), 4) + # test the basic auth of login + self.assertLessEqual( + { + "Authorization": "Basic " + + base64.b64encode( + ( + self.idm["mobileClientId"] + + ":" + + self.idm["mobileClientSecret"] + ).encode("ascii") + ).decode("ascii") + }.items(), + self.responses.calls[1].request.headers.items(), + ) + # test token authentication of whoami + self.assertLessEqual( + {"Authorization": "Bearer 12345678-9abc-def0-1234-56789abcdef0"}.items(), + self.responses.calls[2].request.headers.items(), + ) + # test token authentication of search + self.assertLessEqual( + {"Authorization": "Bearer 12345678-9abc-def0-1234-56789abcdef0"}.items(), + self.responses.calls[3].request.headers.items(), + ) + self.assertDictEqual(cms._idm, self.idm) + self.assertDictEqual(cms._token, self.token) + self.assertDictEqual(cms._user, self.user) + self.assertDictEqual(cms._info, self.info) + + def test_current_temp_get(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + self.assertEqual(cms.current_temp, 38) + + def test_desired_temp_get(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + self.assertEqual(cms.desired_temp, 37.5) + + def test_current_temp_get_fahrenheit(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + cms._info["currentState"]["celsius"] = False + self.assertEqual(cms.current_temp, 100.4) + + def test_desired_temp_get_fahrenheit(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + cms._info["currentState"]["celsius"] = False + self.assertEqual(cms.desired_temp, 99.5) + + def test_desired_temp_set(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + self.responses.add( + responses.POST, + "https://iot.controlmyspa.com/mobile/control/0123456789abcdef01234567/setDesiredTemp", + match=[responses.matchers.json_params_matcher({"desiredTemp": 96.8})], + json={}, + ) + cms.desired_temp = 36 + + def test_desired_temp_set_fahrenheit(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + cms._info["currentState"]["celsius"] = False + self.responses.add( + responses.POST, + "https://iot.controlmyspa.com/mobile/control/0123456789abcdef01234567/setDesiredTemp", + match=[responses.matchers.json_params_matcher({"desiredTemp": 96})], + json={}, + ) + cms.desired_temp = 96 + + def test_temp_range(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + # Default data in test set is "HIGH" + self.assertEqual(cms.temp_range, True) + cms._info["currentState"]["tempRange"] = "LOW" + self.assertEqual(cms.temp_range, False) + + def test_temp_range_set(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + self.responses.add( + responses.POST, + "https://iot.controlmyspa.com/mobile/control/0123456789abcdef01234567/setTempRange", + match=[responses.matchers.json_params_matcher({"desiredState": "HIGH"})], + json={}, + ) + cms.temp_range = True + self.responses.add( + responses.POST, + "https://iot.controlmyspa.com/mobile/control/0123456789abcdef01234567/setTempRange", + match=[responses.matchers.json_params_matcher({"desiredState": "LOW"})], + json={}, + ) + cms.temp_range = False + + def test_panel_lock(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + # Default data in test set is unlocked + self.assertEqual(cms.panel_lock, False) + cms._info["currentState"]["panelLock"] = True + self.assertEqual(cms.panel_lock, True) + + def test_panel_lock_set(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + self.responses.add( + responses.POST, + "https://iot.controlmyspa.com/mobile/control/0123456789abcdef01234567/setPanel", + match=[ + responses.matchers.json_params_matcher({"desiredState": "LOCK_PANEL"}) + ], + json={}, + ) + cms.panel_lock = True + self.responses.add( + responses.POST, + "https://iot.controlmyspa.com/mobile/control/0123456789abcdef01234567/setPanel", + match=[ + responses.matchers.json_params_matcher({"desiredState": "UNLOCK_PANEL"}) + ], + json={}, + ) + cms.panel_lock = False + + def test_jets(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + #  default all jets are off + self.assertEqual(cms.jets, [False, False, False]) + self.assertEqual(cms.get_jet(0), False) + self.assertEqual(cms.get_jet(1), False) + self.assertEqual(cms.get_jet(2), False) + # manually enable all pumps/jets + for component in cms._info["currentState"]["components"]: + if component["componentType"] == "PUMP": + component["value"] = "HIGH" + self.assertEqual(cms.jets, [True, True, True]) + self.assertEqual(cms.get_jet(0), True) + self.assertEqual(cms.get_jet(1), True) + self.assertEqual(cms.get_jet(2), True) + + def test_jets_set(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + self.responses.add( + responses.POST, + "https://iot.controlmyspa.com/mobile/control/0123456789abcdef01234567/setJetState", + match=[ + responses.matchers.json_params_matcher( + { + "desiredState": "HIGH", + "deviceNumber": 0, + "originatorId": "optional-Jet", + } + ) + ], + json={}, + ) + cms.set_jet(0, True) + cms.jets = [True] + self.responses.add( + responses.POST, + "https://iot.controlmyspa.com/mobile/control/0123456789abcdef01234567/setJetState", + match=[ + responses.matchers.json_params_matcher( + { + "desiredState": "OFF", + "deviceNumber": 0, + "originatorId": "optional-Jet", + } + ) + ], + json={}, + ) + cms.set_jet(0, False) + cms.jets = [False] + + def test_jets_empty(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + # test correct handling if there were no jets at all + cms._info["currentState"]["components"] = [ + x + for x in cms._info["currentState"]["components"] + if x["componentType"] != "PUMP" + ] + self.assertEqual(cms.jets, []) + + def test_blower_empty(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + #  default all blowers are off + self.assertEqual(cms.blowers, []) + + def test_blower(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + # my test data contains no blower. creating a synthetic one to be able to test + cms._info["currentState"]["components"].append({ + "availableValues": ["OFF", "HIGH"], + "componentType": "BLOWER", + "materialType": "BLOWER", + "name": "BLOWER", + "port": "0", + "registeredTimestamp": "2021-09-14T17:35:17.430+0000", + "targetValue": "OFF", + "value": "OFF", + }) + self.assertEqual(cms.blowers, [False]) + self.assertEqual(cms.get_blower(0), False) + # manually enable + for component in cms._info["currentState"]["components"]: + if component["componentType"] == "BLOWER": + component["value"] = "HIGH" + self.assertEqual(cms.blowers, [True]) + self.assertEqual(cms.get_blower(0), True) + + def test_blowers_set(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + self.responses.add( + responses.POST, + "https://iot.controlmyspa.com/mobile/control/0123456789abcdef01234567/setBlowerState", + match=[ + responses.matchers.json_params_matcher( + { + "desiredState": "HIGH", + "deviceNumber": 0, + "originatorId": "optional-Blower", + } + ) + ], + json={}, + ) + cms.set_blower(0, True) + cms.blowers = [True] + + self.responses.add( + responses.POST, + "https://iot.controlmyspa.com/mobile/control/0123456789abcdef01234567/setBlowerState", + match=[ + responses.matchers.json_params_matcher( + { + "desiredState": "OFF", + "deviceNumber": 0, + "originatorId": "optional-Blower", + } + ) + ], + json={}, + ) + cms.set_blower(0, False) + cms.blower = [False] + + def test_lights(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + #  default all jets are off + self.assertEqual(cms.lights, [False]) + self.assertEqual(cms.get_light(0), False) + # manually enable all lights + for component in cms._info["currentState"]["components"]: + if component["componentType"] == "LIGHT": + component["value"] = "HIGH" + self.assertEqual(cms.lights, [True]) + self.assertEqual(cms.get_light(0), True) + + def test_lights_set(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + self.responses.add( + responses.POST, + "https://iot.controlmyspa.com/mobile/control/0123456789abcdef01234567/setLightState", + match=[ + responses.matchers.json_params_matcher( + { + "desiredState": "HIGH", + "deviceNumber": 0, + "originatorId": "optional-Light", + } + ) + ], + json={}, + ) + cms.set_light(0, True) + cms.lights = [True] + + self.responses.add( + responses.POST, + "https://iot.controlmyspa.com/mobile/control/0123456789abcdef01234567/setLightState", + match=[ + responses.matchers.json_params_matcher( + { + "desiredState": "OFF", + "deviceNumber": 0, + "originatorId": "optional-Light", + } + ) + ], + json={}, + ) + cms.set_light(0, False) + cms.lights = [False] + + def test_lights_empty(self): + cms = ControlMySpa(self.exampleusername, self.examplepassword) + # test correct handling if there were no jets at all + cms._info["currentState"]["components"] = [ + x + for x in cms._info["currentState"]["components"] + if x["componentType"] != "LIGHT" + ] + self.assertEqual(cms.lights, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tox.ini b/tox.ini new file mode 100644 index 0000000..52abe28 --- /dev/null +++ b/tox.ini @@ -0,0 +1,78 @@ +[tox] +envlist = + clean + pylint + flake8 + py35 + py36 + py37 + py38 + py39 + pypy3 + report +skip_missing_interpreters = True +requires = + tox-pip-extensions +basepython = + py35: python3.5 + py36: python3.6 + py37: python3.7 + py38: python3.8 + py39: python3.9 + pypy3: pypy3 + +[testenv] +deps = + pytest + pytest-cov + responses +commands = + pip install -e . + pytest --cov --cov-append --ignore .tox +norecursedirs = .tox + +[testenv:clean] +deps = coverage +skip_install = true +commands = coverage erase + +[testenv:report] +passenv = TOXENV CI TRAVIS TRAVIS_* CODECOV_* +deps = + coverage + codecov +skip_install = true +commands = + coverage report --omit='.tox/*' + coverage html --omit='.tox/*' + codecov -e TOXENV + +[testenv:flake8] +basepython = python3.6 +deps = flake8 + flake8-isort + flake8-black + flake8-blind-except + flake8-builtins + flake8-docstrings + flake8-bugbear + flake8-mypy + pep8-naming + flake8-assertive + flake8-mock + flake8-bandit +commands = flake8 + +[testenv:pylint] +deps = + pylint + -rrequirements.txt +commands = pylint --disable=bad-continuation controlmyspa +# pylint known bug https://github.com/ambv/black/issues/48 +# https://stackoverflow.com/questions/17142236/how-do-i-make-pylint-recognize-twisted-and-ephem-members + +[flake8] +exclude = .tox,venv,*.egg*,.git,__pycache__,*.pyc*,build,dist +max-line-length = 88 +select = C,E,F,G,W,B,B902,B950 +ignore = E501,W503, BLK100