First revision

README.md

@@ -1 +1,7 @@
-# python-tahoma-api
+# Somfy TaHoma
+
+An updated and async version of the original [tahoma-api](https://github.com/philklei/tahoma-api) by [@philklei](https://github.com/philklei). The aim of this wrapper is to offer an easy to consume Python wrapper for the internal API's used by tahomalink.com.
+
+Somfy TaHoma has an official API, which can be consumed via the [Somfy-open-api](https://github.com/tetienne/somfy-open-api). Unfortunately only a few device classes are supported via the official API, thus the need for this wrapper.
+
+This package is written for the Home Assistant [ha-tahoma](https://github.com/iMicknl/ha-tahoma) integration, but could be used by any Python project interacting with Somfy TaHoma devices.

setup.py

@@ -8,7 +8,7 @@
     version="0.1.0",
     author="Mick Vleeshouwer",
     author_email="mick@imick.nl",
-    description="Python wrapper to interact with SagemCom F@st routers via internal API's.",
+    description="Python wrapper to interact with internal Somfy TaHama API",
     long_description=long_description,
     long_description_content_type="text/markdown",
     url="https://github.com/iMicknl/python-tahoma-api",

tahoma_api/client.py

@@ -10,13 +10,15 @@
 import json
 
 from .exceptions import *
+from .models import *
+
+API_URL = "https://tahomalink.com/enduser-mobile-web/enduserAPI/"  # /doc for API doc
 
-API_URL = 'https://tahomalink.com/enduser-mobile-web/enduserAPI/'  # /doc for API doc
 
 class TahomaClient(object):
     """ Interface class for the Tahoma API """
 
-    def __init__(self, username, password):
+    def __init__(self, username, password, api_url=API_URL):
         """
         Constructor
 
@@ -26,77 +28,135 @@ def __init__(self, username, password):
 
         self.username = username
         self.password = password
+        self.api_url = api_url
 
-        self.__roles = []
+        self.__cookies = None
+        self.__devices = None
+        self.__roles = None
 
     async def login(self):
-
-        payload = {
-            'userId': self.username,
-            'userPassword': self.password
-        }
+
+        payload = {"userId": self.username, "userPassword": self.password}
 
         async with aiohttp.ClientSession() as session:
-            async with session.post(API_URL + 'login', data=payload) as response:
+            async with session.post(self.api_url + "login", data=payload) as response:
 
                 result = await response.json()
 
                 # 401
                 # {'errorCode': 'AUTHENTICATION_ERROR', 'error': 'Bad credentials'}
+                # {'errorCode': 'AUTHENTICATION_ERROR', 'error': 'Your setup cannot be accessed through this application'}
+                if response.status == 401:
+                    if result["errorCode"] == "AUTHENTICATION_ERROR":
+
+                        if "Too many requests" in result["error"]:
+                            print(result["error"])
+                            raise Exception
+
+                        if (
+                            "Your setup cannot be accessed through this application"
+                            in result["error"]
+                        ):
+                            print(result["error"])
+
+                        if "Bad credentials" in result["error"]:
+                            print(result["error"])
+
+                        print(result["error"])
+
+                        return False  # todo throw error
 
                 # 401
                 # {'errorCode': 'AUTHENTICATION_ERROR', 'error': 'Too many requests, try again later : login with xxx@xxx.tld'}
                 # TODO Add retry logic on too many requests + for debug, log requests + timespans
 
                 # 200
                 # {'success': True, 'roles': [{'name': 'ENDUSER'}]}
-                if (response.status is 200):
-                    if result['success'] == True:
-                        self.__roles = result['roles']
+                if response.status == 200:
+                    if result["success"] == True:
+                        self.__roles = result["roles"]
                         self.__cookies = response.cookies
 
                         return True
 
+                # Temp fallbacks
                 print(response.status)
                 print(result)
 
-    async def get_devices(self):
-
-        cookies = self.__cookies
+    async def get_devices(self, refresh=False) -> List[Device]:
+        if self.__devices and refresh == False:
+            return self._devices
 
-        async with aiohttp.ClientSession() as session:
-            async with session.get(API_URL + 'setup/devices', cookies=cookies) as response:
+        response = await self.__make_http_request("GET", "setup/devices")
 
-                print(response.status)
-                print(response)
-
-                result = await response.json()
+        devices = [Device(**d) for d in response]
+        self.__devices = devices
 
-                print(result)
-                # 401
-                # {'errorCode': 'AUTHENTICATION_ERROR', 'error': 'Bad credentials'}
+        return devices
 
-                # {'success': True, 'roles': [{'name': 'ENDUSER'}]}
-                if (response.status is 200):
-                    if result["success"] == True:
-                        print(result)
+    async def register_event_listener(self) -> str:
+        """ 
+        Register a new setup event listener on the current session and return a new listener id. 
+        Only one listener may be registered on a given session. 
+        Registering an new listener will invalidate the previous one if any. 
+        Note that registering an event listener drastically reduces the session timeout : listening sessions are expected to call the /events/{listenerId}/fetch API on a regular basis.
+        """
+        response = await self.__make_http_request("POST", "events/register")
+        listener_id = response.get("id")
 
-                # TODO Save cookies
+        return listener_id
 
-    async def get_states(self):
-
-        cookies = self.__cookies
+    async def fetch_event_listener(self, listener_id: str) -> List[Any]:
+        """
+        Fetch new events from a registered event listener. Fetched events are removed from the listener buffer. Return an empty response if no event is available.
+        Per-session rate-limit : 1 calls per 1 SECONDS period for this particular operation (polling)
+        """
+        response = await self.__make_http_request("POST", f"events/{listener_id}/fetch")
 
-        async with aiohttp.ClientSession() as session:
-            async with session.get(API_URL + 'setup/devices/states', cookies=cookies) as response:
+        return response
 
-                print(response.status)                
-                result = await response.json()
+    async def execute_action_group(
+        self, actions: [Command], label="python-tahoma-api", priority=False
+    ) -> List[Any]:
+        """
+        Execute a non-persistent action group
+        The executed action group does not have to be persisted on the server before use.
+        Per-session rate-limit : 50 calls per 24 HOURS period for all operations of the same category (exec)
+        """
+        payload = {"label": label, "actions": actions}
+        response = await self.__make_http_request("POST", f"exec/apply", payload)
 
-                print(result)
-
+        return response
 
+    apply_actions
 
-
+    async def __make_http_request(
+        self, method: str, endpoint: str, payload: Optional[Any] = None
+    ) -> Any:
+        """Make a request to the TaHoma API"""
+        cookies = self.__cookies
+        supported_methods = ["GET", "POST"]
+
+        if method not in supported_methods:
+            raise Exception
 
+        async with aiohttp.ClientSession() as session:
+            if method == "GET":
+                async with session.get(
+                    self.api_url + endpoint, cookies=cookies
+                ) as response:
+                    result = await response.json()
+
+            if method == "POST":
+                async with session.post(
+                    self.api_url + endpoint, cookies=cookies, data=payload
+                ) as response:
+                    result = await response.json()
+
+        if response.status == 200:
+            return result
+
+        if response.status > 400 and response.status < 500:
+            # implement retry logic
+            print("TODO")
 

tahoma_api/models.py

@@ -0,0 +1,79 @@
+from typing import Any, Dict, List, Optional, Union
+
+# TODO Rewrite camelCase to snake_case
+class Device:
+    __slots__ = (
+        "id",
+        "creationTime",
+        "lastUpdateTime",
+        "label",
+        "deviceURL",
+        "shortcut",
+        "controllableName",
+        "definition",
+        "states",
+        "dataProperties",
+        "available",
+        "enabled",
+        "widgetName",
+        "widget",
+        "uiClass",
+        "qualifiedName",
+        "type",
+    )
+
+    def __init__(
+        self,
+        *,
+        label: str,
+        deviceURL: str,
+        controllableName: str,
+        # definition: Dict[List[Any]],
+        states: List[Dict[str, Any]],
+        dataProperties: Optional[List[Dict[str, Any]]] = None,
+        widgetName: Optional[str] = None,
+        uiClass: str,
+        qualifiedName: Optional[str] = None,
+        type: str,
+        **kwargs: Any
+    ):
+        self.id = deviceURL
+        self.deviceURL = deviceURL
+        self.label = label
+        self.controllableName = controllableName
+        self.states = [State(**s) for s in states]
+
+
+class StateDefinition:
+    __slots__ = (
+        "qualifiedName",
+        "type",
+        "values",
+    )
+
+    def __init__(
+        self, qualifiedName: str, type: str, values: Optional[str], **kwargs: Any
+    ):
+        self.qualifiedName = qualifiedName
+        self.type = type
+        self.values = values
+
+
+class CommandDefinition:
+    __slots__ = (
+        "commandName",
+        "nparams",
+    )
+
+    def __init__(self, commandName: str, nparams: int, **kwargs: Any):
+        self.commandName = commandName
+        self.nparams = nparams
+
+
+class State:
+    __slots__ = "name", "value", "type"
+
+    def __init__(self, name: str, value: str, type: str, **kwargs: Any):
+        self.name = name
+        self.value = value
+        self.type = type

test.py

@@ -1,18 +1,27 @@
 import asyncio
 from tahoma_api import TahomaClient
+import time
 
 username = ""
 password = ""
 
+
 async def main():
     client = TahomaClient(username, password)
-    
+
     try:
         login = await client.login()
-        devices = await client.get_states()
+        devices = await client.get_devices()
+
+        for device in devices:
+            print(f"{device.label} ({device.id})")
 
-        print(devices)
+        listener_id = await client.register_event_listener()
 
+        while True:
+            events = await client.fetch_event_listener(listener_id)
+            print(events)
+            time.sleep(2)
 
     except Exception as exception:
         print(exception)