finish basic implementation

Author: marcelveldt

blueconnect/__init__.py

@@ -1,191 +1,72 @@
 """Unofficial python library for the Blue Riiot Blue Connect API."""
 
-
 import asyncio
-import time
-import json
 import logging
+from typing import List
 
-import aiohttp
-from aws_request_signer import AwsRequestSigner
+from .api import BlueConnectApi
+from .models import (
+    BlueDevice,
+    SwimmingPool,
+    SwimmingPoolFeedMessage,
+    SwimmingPoolMeasurement,
+    TemperatureUnit,
+)
 
-AWS_REGION = "eu-west-1"
-AWS_SERVICE = "execute-api"
-BASE_HEADERS = {
-    "User-Agent": "BlueConnect/3.2.1",
-    "Accept-Language": "en-DK;q=1.0, da-DK;q=0.9",
-    "Accept": "*/*"
-}
-BASE_URL = "https://api.riiotlabs.com/prod/"
 LOGGER = logging.getLogger()
 
 
-class BlueConnectApi():
-    """Class that holds the connection to the Blue Connect API."""
-    _username = None
-    _password = None
-    _language = None
-    _token_info = {}
-    _loop = None
-    _http_session = None
-    _user_info = {}
-    _swimming_pool_info = {}
-    _swimming_pool_status = {}
-    _swimming_pool_feed = {}
-    _swimming_pool_ph = {}
-    _swimming_pool_orp = {}
-    _swimming_pool_temp = {}
-    _swimming_pool_device = {}
+class BlueConnectSimpleAPI:
+    """Class that provides a common structure around the Blue Connect API for a single/main swimming pool."""
 
     def __init__(self, username: str, password: str, language: str = "en") -> None:
         """Inititialize the api connection, a valid username and password must be provided."""
-        self._username = username
-        self._password = password
+        self._api = BlueConnectApi(username, password)
         self._language = language
-        self._loop = asyncio.get_event_loop()
-        self._http_session = aiohttp.ClientSession(
-            loop=self._loop, connector=aiohttp.TCPConnector()
-        )
-
-    def close(self):
-        """Close connection to the api."""
-        asyncio.create_task(self.close_async())
+        self._temperature_unit = None
+        self._pool_info = None
+        self._pool_feed_message = None
+        self._pool_blue_device = None
+        self._pool_measurements = []
 
-    async def close_async(self):
+    def close(self) -> None:
         """Close connection to the api."""
-        await self._http_session.close()
+        self._api.close()
 
-    async def fetch_data(self):
+    async def fetch_data(self) -> None:
         """Fetch latest state from API."""
-        self._user_info = await self.get_user_info()
-        self._swimming_pool_info = await self.get_swimming_pool_info(self.main_swimming_pool_id)
-        self._swimming_pool_status = await self.get_swimming_pool_status(self.main_swimming_pool_id)
-        self._swimming_pool_feed = await self.get_swimming_pool_feed(self.main_swimming_pool_id)
-        self._swimming_pool_device = await self.get_swimming_pool_blue_device(self.main_swimming_pool_id)
-        # get levels from status task infos
-        for task in self.swimming_pool_status["tasks"]:
-            if task["task_identifier"].startswith("ORP_"):
-                self._swimming_pool_orp = json.loads(task["data"])
-            if task["task_identifier"].startswith("TEMPERATURE_"):
-                self._swimming_pool_temp = json.loads(task["data"])
-            if task["task_identifier"].startswith("PH_"):
-                self._swimming_pool_ph = json.loads(task["data"])
-
-    @property
-    def main_swimming_pool_id(self):
-        """Return ID of the main swimming pool."""
-        return self.user_preferences.get("main_swimming_pool_id")
-
-    @property
-    def swimming_pool_name(self):
-        """Return name of the main swimming pool."""
-        return self.swimming_pool_info.get("name")
-
-    @property
-    def temperature_unit(self):
-        """Return temperature_unit preference of the logged in user."""
-        return self.user_preferences.get("display_temperature_unit")
-
-    @property
-    def swimming_pool_info(self):
-        """Return info for the main swimming pool."""
-        return self._swimming_pool_info
-
-    @property
-    def swimming_pool_status(self):
-        """Return info for the main swimming pool."""
-        return self._swimming_pool_status
-
-    @property
-    def swimming_pool_feed(self):
-        """Return status feed for the main swimming pool."""
-        return self._swimming_pool_feed
+        user_info = await self._api.get_user()
+        main_pool_id = user_info.user_preferences.main_swimming_pool_id
+        self._temperature_unit = user_info.user_preferences.display_temperature_unit
+        self._pool_info = await self._api.get_swimming_pool(main_pool_id)
+        self._pool_feed_message = (await self._api.get_swimming_pool_feed(main_pool_id, self._language)).current_message
+        blue_devices = await self._api.get_swimming_pool_blue_devices(main_pool_id)
+        self._pool_blue_device = blue_devices[0] if blue_devices else None
+        if self._pool_blue_device:
+            self._pool_measurements = (await self._api.get_last_measurements(
+                main_pool_id, self._pool_blue_device.serial)).measurements
 
     @property
-    def swimming_pool_device(self):
-        """Return Blue Connect device info for the main swimming pool."""
-        return self._swimming_pool_device
+    def pool(self) -> SwimmingPool:
+        """Return full details of the (main) swimming pool."""
+        return self._pool_info
 
     @property
-    def user_info(self):
-        """Return info about logged in user."""
-        return self._user_info
+    def temperature_unit(self) -> TemperatureUnit:
+        """Return temperature unit of the temperature measurements."""
+        return self._temperature_unit
 
     @property
-    def user_preferences(self):
-        """Return preferences of logged in user."""
-        return self._user_info.get("userPreferences", {})
+    def feed_message(self) -> SwimmingPoolFeedMessage:
+        """Return the (latest) feed/health message for the (main) swimming pool."""
+        return self._pool_feed_message
 
     @property
-    def swimming_pool_ph(self):
-        """Return current PH info of the main swimming pool."""
-        return self._swimming_pool_ph
+    def blue_device(self) -> BlueDevice:
+        """Return Blue Connect device info for the (main) swimming pool."""
+        return self._pool_blue_device
 
     @property
-    def swimming_pool_orp(self):
-        """Return current ORP info of the main swimming pool."""
-        return self._swimming_pool_orp
-
-    @property
-    def swimming_pool_temp(self):
-        """Return current Temperature info of the main swimming pool."""
-        return self._swimming_pool_temp
-
-    async def get_user_info(self):
-        """Retrieve details of logged in user."""
-        return await self.__get_data("user")
-
-    async def get_swimming_pool_info(self, swimming_pool_id: str):
-        """Retrieve details for a specific swimming pool."""
-        return await self.__get_data(f"swimming_pool/{swimming_pool_id}")
-
-    async def get_swimming_pool_status(self, swimming_pool_id: str):
-        """Retrieve status for a specific swimming pool."""
-        return await self.__get_data(f"swimming_pool/{swimming_pool_id}/status")
-
-    async def get_swimming_pool_blue_device(self, swimming_pool_id: str):
-        """Retrieve Blue device info for a specific swimming pool."""
-        return await self.__get_data(f"swimming_pool/{swimming_pool_id}/blue")
-
-    async def get_swimming_pool_feed(self, swimming_pool_id: str):
-        """Retrieve feed for a specific swimming pool, defaults to user's main swimming pool."""
-        return await self.__get_data(f"swimming_pool/{swimming_pool_id}/feed?lang={self._language}")
-
-    async def __get_credentials(self):
-        """Retrieve auth credentials by logging in with username/password."""
-        if self._token_info and self._token_info["expires"] > time.time():
-            # return cached credentials if still valid
-            return self._token_info["credentials"]
-        # perform log-in to get credentials
-        url = BASE_URL + "user/login"
-        async with self._http_session.post(
-            url, json={"email": self._username, "password": self._password}
-        ) as response:
-            if response.status != 200:
-                LOGGER.exception(await response.text())
-                return None
-            result = await response.json()
-            self._token_info = result
-            self._token_info["expires"] = time.time() + 3500
-            return result["credentials"]
-
-    async def __get_data(self, endpoint, params={}):
-        """Get data from api."""
-        url = BASE_URL + endpoint
-        headers = BASE_HEADERS.copy()
-        # sign the request
-        creds = await self.__get_credentials()
-        if not creds:
-            return None
-        request_signer = AwsRequestSigner(
-            AWS_REGION, creds["access_key"], creds["secret_key"], AWS_SERVICE
-        )
-        headers.update(
-            request_signer.sign_with_headers("GET", url, headers)
-        )
-        headers["X-Amz-Security-Token"] = creds["session_token"]
-        async with self._http_session.get(
-            url, headers=headers, params=params, verify_ssl=False
-        ) as response:
-            assert response.status == 200
-            return await response.json()
+    def measurements(self) -> List[SwimmingPoolMeasurement]:
+        """Return all last/current measurements for the (main) swimming pool."""
+        return self._pool_measurements

blueconnect/api.py

@@ -0,0 +1,139 @@
+"""Unofficial python library for the Blue Riiot Blue Connect API."""
+
+import asyncio
+import logging
+import time
+from typing import List, Optional
+
+import aiohttp
+from aws_request_signer import AwsRequestSigner
+
+from .models import (
+    BlueDevice,
+    SwimmingPool,
+    SwimmingPoolFeed,
+    SwimmingPoolLastMeasurements,
+    SwimmingPoolStatus,
+    User,
+)
+
+AWS_REGION = "eu-west-1"
+AWS_SERVICE = "execute-api"
+BASE_HEADERS = {
+    "User-Agent": "BlueConnect/3.2.1",
+    "Accept-Language": "en-DK;q=1.0, da-DK;q=0.9",
+    "Accept": "*/*",
+}
+BASE_URL = "https://api.riiotlabs.com/prod/"
+LOGGER = logging.getLogger()
+
+
+class BlueConnectApi:
+    """Class that holds the connection to the Blue Connect API."""
+
+    def __init__(self, username: str, password: str) -> None:
+        """Inititialize the api connection, a valid username and password must be provided."""
+        self._username = username
+        self._password = password
+        self._http_session = aiohttp.ClientSession(
+            connector=aiohttp.TCPConnector()
+        )
+        self._token_info = {}
+
+    def close(self) -> None:
+        """Close connection to the api."""
+        if asyncio.get_event_loop().is_running():
+            asyncio.create_task(self.close_async())
+        else:
+            asyncio.get_event_loop().run_until_complete(self.close_async())
+
+    async def close_async(self) -> None:
+        """Close connection to the api."""
+        await self._http_session.close()
+
+    async def get_user(self) -> User:
+        """Retrieve details of logged-in user."""
+        data = await self.__get_data(f"user")
+        return User.from_json(data)
+
+    async def get_blue_device(self, blue_device_serial: str) -> BlueDevice:
+        """Retrieve details for a specific blue device."""
+        data = await self.__get_data(f"blue/{blue_device_serial}")
+        return BlueDevice.from_json(data)
+
+    async def get_swimming_pools(self) -> List[SwimmingPool]:
+        """Retrieve all swimming pools."""
+        data = await self.__get_data(f"swimming_pool")
+        return [SwimmingPool.from_json(item["swimming_pool"]) for item in data["data"]]
+
+    async def get_swimming_pool(self, swimming_pool_id: str) -> SwimmingPool:
+        """Retrieve details for a specific swimming pool."""
+        data = await self.__get_data(f"swimming_pool/{swimming_pool_id}")
+        return SwimmingPool.from_json(data)
+
+    async def get_swimming_pool_status(self, swimming_pool_id: str) -> SwimmingPoolStatus:
+        """Retrieve status for a specific swimming pool."""
+        data = await self.__get_data(f"swimming_pool/{swimming_pool_id}/status")
+        return SwimmingPoolStatus.from_json(data)
+
+    async def get_swimming_pool_blue_devices(self, swimming_pool_id: str) -> List[BlueDevice]:
+        """Retrieve Blue devices for a specific swimming pool."""
+        data = await self.__get_data(f"swimming_pool/{swimming_pool_id}/blue")
+        result = []
+        for item in data["data"]:
+            blue_device = await self.get_blue_device(item["blue_device_serial"])
+            result.append(blue_device)
+        return result
+
+    async def get_swimming_pool_feed(self, swimming_pool_id: str, language: str = "en") -> SwimmingPoolFeed:
+        """Retrieve feed for a specific swimming pool."""
+        data = await self.__get_data(
+            f"swimming_pool/{swimming_pool_id}/feed?lang={language}"
+        )
+        return SwimmingPoolFeed.from_json(data)
+
+    async def get_last_measurements(self, swimming_pool_id: str, blue_device_serial: str) -> SwimmingPoolLastMeasurements:
+        """Retrieve last measurements for a specific swimming pool."""
+        data = await self.__get_data(
+            f"swimming_pool/{swimming_pool_id}/blue/{blue_device_serial}/lastMeasurements?mode=blue_and_strip"
+        )
+        return SwimmingPoolLastMeasurements.from_json(data)
+
+    async def __get_credentials(self) -> dict:
+        """Retrieve auth credentials by logging in with username/password."""
+        if self._token_info and self._token_info["expires"] > time.time():
+            # return cached credentials if still valid
+            return self._token_info["credentials"]
+        # perform log-in to get credentials
+        url = BASE_URL + "user/login"
+        async with self._http_session.post(
+            url, json={"email": self._username, "password": self._password}
+        ) as response:
+            if response.status != 200:
+                error_msg = await response.text()
+                raise Exception("Error logging in user: %s" % error_msg)
+            result = await response.json()
+            self._token_info = result
+            self._token_info["expires"] = time.time() + 3500
+            return result["credentials"]
+
+    async def __get_data(self, endpoint: str, params: dict = None) -> Optional[dict]:
+        """Get data from api."""
+        if params is None:
+            params = {}
+        url = BASE_URL + endpoint
+        headers = BASE_HEADERS.copy()
+        # sign the request
+        creds = await self.__get_credentials()
+        request_signer = AwsRequestSigner(
+            AWS_REGION, creds["access_key"], creds["secret_key"], AWS_SERVICE
+        )
+        headers.update(request_signer.sign_with_headers("GET", url, headers))
+        headers["X-Amz-Security-Token"] = creds["session_token"]
+        async with self._http_session.get(
+            url, headers=headers, params=params, verify_ssl=False
+        ) as response:
+            if response.status != 200:
+                error_msg = await response.text()
+                raise Exception("Error while retrieving data for endpoint %s: %s" % (endpoint, error_msg))
+            return await response.json()