Split login logic

README.md

@@ -7,23 +7,11 @@
 
 A fully async and easy to use API client for the (internal) OverKiz API. You can use this client to interact with smart devices connected to the OverKiz platform, used by various vendors like Somfy TaHoma and Atlantic Cozytouch.
 
-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 OverKiz hubs.
-
-> 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 API client.
+This package is written for the Home Assistant [Overkiz](https://www.home-assistant.io/integrations/overkiz/) integration, but could be used by any Python project interacting with OverKiz hubs.
 
 ## Supported hubs
 
-- Atlantic Cozytouch
-- Bouygues Flexom
-- Hitachi Hi Kumo
-- Nexity Eugénie
-- Rexel Energeasy Connect
-- Simu (LiveIn2)
-- Somfy Connexoon IO
-- Somfy Connexoon RTS
-- Somfy TaHoma
-- Somfy TaHoma Switch
-- Thermor Cozytouch
+See [pyoverkiz/const.py](./pyoverkiz/const.py)
 
 ## Installation
 
@@ -33,18 +21,35 @@ pip install pyoverkiz
 
 ## Getting started
 
+### API Documentation
+
+A subset of the API is [documented and maintened](https://somfy-developer.github.io/Somfy-TaHoma-Developer-Mode) by Somfy.
+
+### Cloud API
+
 ```python
 import asyncio
 import time
 
-from pyoverkiz.const import SUPPORTED_SERVERS
-from pyoverkiz.client import OverkizClient
+from aiohttp import ClientSession
+
+from pyoverkiz.clients.overkiz import OverkizClient
+from pyoverkiz.const import Server
+from pyoverkiz.overkiz import Overkiz
 
 USERNAME = ""
 PASSWORD = ""
 
+
 async def main() -> None:
-    async with OverkizClient(USERNAME, PASSWORD, server=SUPPORTED_SERVERS["somfy_europe"]) as client:
+
+    async with ClientSession() as session:
+        client = Overkiz.create_client(
+            server=Server.SOMFY_EUROPE,
+            username=USERNAME,
+            password=PASSWORD,
+            session=session
+        )
         try:
             await client.login()
         except Exception as exception:  # pylint: disable=broad-except
@@ -63,9 +68,72 @@ async def main() -> None:
 
             time.sleep(2)
 
+
 asyncio.run(main())
 ```
 
+### Local API or Developer mode
+
+
+See https://github.com/Somfy-Developer/Somfy-TaHoma-Developer-Mode#getting-started
+
+For the moment, only Somfy TaHoma Switch, TaHoma V2 and Connexoon hubs from Somfy Europe can enabled this mode. Not all the devices are returned. You can have more details [here](https://github.com/Somfy-Developer/Somfy-TaHoma-Developer-Mode/issues/20).
+
+
+```python
+import asyncio
+import time
+
+from aiohttp import ClientSession
+
+from pyoverkiz.clients.overkiz import OverkizClient
+from pyoverkiz.const import Server
+from pyoverkiz.overkiz import Overkiz
+
+USERNAME = ""
+PASSWORD = ""
+
+
+async def main() -> None:
+
+    async with ClientSession() as session:
+        client = Overkiz.create_client(
+            Server.SOMFY_EUROPE, USERNAME, PASSWORD, session
+        )
+        try:
+            await client.login()
+        except Exception as exception:  # pylint: disable=broad-except
+            print(exception)
+            return
+
+        gateways = await client.get_gateways()
+        token = await client.generate_local_token(gateways[0].id)
+        await client.activate_local_token(gateways[0].id, token, "pyoverkiz")
+
+        domain = f"gateway-{gateways[0].id}.local"
+        local_client: OverkizClient = Overkiz.create_client(
+            Server.SOMFY_DEV_MODE, domain, token, session
+        )
+
+        devices = await local_client.get_devices()
+
+        for device in devices:
+            print(f"{device.label} ({device.id}) - {device.controllable_name}")
+            print(f"{device.widget} - {device.ui_class}")
+
+        await local_client.register_event_listener()
+
+        while True:
+            events = await local_client.fetch_events()
+            print(events)
+
+            time.sleep(2)
+
+
+asyncio.run(main())
+
+```
+
 ## Development
 
 ### Installation

mypy.ini

@@ -3,18 +3,22 @@ python_version = 3.8
 show_error_codes = true
 follow_imports = silent
 ignore_missing_imports = true
+local_partial_types = true
 strict_equality = true
+no_implicit_optional = true
 warn_incomplete_stub = true
 warn_redundant_casts = true
 warn_unused_configs = true
 warn_unused_ignores = true
+enable_error_code = ignore-without-code, redundant-self, truthy-iterable
+disable_error_code = annotation-unchecked
+strict_concatenate = false
 check_untyped_defs = true
 disallow_incomplete_defs = true
 disallow_subclassing_any = true
 disallow_untyped_calls = true
-#disallow_untyped_decorators = true
+disallow_untyped_decorators = true
 disallow_untyped_defs = true
-no_implicit_optional = true
 warn_return_any = true
 warn_unreachable = true
 

pyoverkiz/clients/atlantic_cozytouch.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from aiohttp import FormData
+from attr import define, field
+
+from pyoverkiz.clients.overkiz import OverkizClient
+from pyoverkiz.exceptions import (
+    CozyTouchBadCredentialsException,
+    CozyTouchServiceException,
+)
+
+COZYTOUCH_ATLANTIC_API = "https://apis.groupe-atlantic.com"
+COZYTOUCH_CLIENT_ID = (
+    "Q3RfMUpWeVRtSUxYOEllZkE3YVVOQmpGblpVYToyRWNORHpfZHkzNDJVSnFvMlo3cFNKTnZVdjBh"
+)
+
+
+@define(kw_only=True)
+class AtlanticCozytouchClient(OverkizClient):
+
+    username: str
+    password: str = field(repr=lambda _: "***")
+
+    async def _login(self) -> bool:
+        """
+        Authenticate and create an API session allowing access to the other operations.
+        Caller must provide one of [userId+userPassword, userId+ssoToken, accessToken, jwt]
+        """
+
+        async with self.session.post(
+            COZYTOUCH_ATLANTIC_API + "/token",
+            data=FormData(
+                {
+                    "grant_type": "password",
+                    "username": "GA-PRIVATEPERSON/" + self.username,
+                    "password": self.password,
+                }
+            ),
+            headers={
+                "Authorization": f"Basic {COZYTOUCH_CLIENT_ID}",
+                "Content-Type": "application/x-www-form-urlencoded",
+            },
+        ) as response:
+            token = await response.json()
+
+            # {'error': 'invalid_grant',
+            # 'error_description': 'Provided Authorization Grant is invalid.'}
+            if "error" in token and token["error"] == "invalid_grant":
+                raise CozyTouchBadCredentialsException(token["error_description"])
+
+            if "token_type" not in token:
+                raise CozyTouchServiceException("No CozyTouch token provided.")
+
+        # Request JWT
+        async with self.session.get(
+            COZYTOUCH_ATLANTIC_API + "/magellan/accounts/jwt",
+            headers={"Authorization": f"Bearer {token['access_token']}"},
+        ) as response:
+            jwt = await response.text()
+
+            if not jwt:
+                raise CozyTouchServiceException("No JWT token provided.")
+
+            jwt = jwt.strip('"')  # Remove surrounding quotes
+
+        payload = {"jwt": jwt}
+
+        post_response = await self.post("login", data=payload)
+
+        return "success" in post_response

pyoverkiz/clients/default.py

@@ -0,0 +1,19 @@
+from attr import define, field
+
+from pyoverkiz.clients.overkiz import OverkizClient
+
+
+@define(kw_only=True)
+class DefaultClient(OverkizClient):
+
+    username: str
+    password: str = field(repr=lambda _: "***")
+
+    async def _login(self) -> bool:
+        """
+        Authenticate and create an API session allowing access to the other operations.
+        Caller must provide one of [userId+userPassword, userId+ssoToken, accessToken, jwt]
+        """
+        payload = {"userId": self.username, "userPassword": self.password}
+        response = await self.post("login", data=payload)
+        return "success" in response

pyoverkiz/clients/nexity.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import asyncio
+from typing import cast
+
+import boto3
+from attr import define, field
+from botocore.config import Config
+from warrant_lite import WarrantLite
+
+from pyoverkiz.clients.overkiz import OverkizClient
+from pyoverkiz.exceptions import NexityBadCredentialsException, NexityServiceException
+
+NEXITY_API = "https://api.egn.prd.aws-nexity.fr"
+NEXITY_COGNITO_CLIENT_ID = "3mca95jd5ase5lfde65rerovok"
+NEXITY_COGNITO_USER_POOL = "eu-west-1_wj277ucoI"
+NEXITY_COGNITO_REGION = "eu-west-1"
+
+
+@define(kw_only=True)
+class NexityClient(OverkizClient):
+
+    username: str
+    password: str = field(repr=lambda _: "***")
+
+    async def _login(self) -> bool:
+        """
+        Authenticate and create an API session allowing access to the other operations.
+        Caller must provide one of [userId+userPassword, userId+ssoToken, accessToken, jwt]
+        """
+
+        loop = asyncio.get_event_loop()
+
+        def _get_client() -> boto3.session.Session.client:
+            return boto3.client(
+                "cognito-idp", config=Config(region_name=NEXITY_COGNITO_REGION)
+            )
+
+        # Request access token
+        client = await loop.run_in_executor(None, _get_client)
+
+        aws = WarrantLite(
+            username=self.username,
+            password=self.password,
+            pool_id=NEXITY_COGNITO_USER_POOL,
+            client_id=NEXITY_COGNITO_CLIENT_ID,
+            client=client,
+        )
+
+        try:
+            tokens = await loop.run_in_executor(None, aws.authenticate_user)
+        except Exception as error:
+            raise NexityBadCredentialsException() from error
+
+        async with self.session.get(
+            NEXITY_API + "/deploy/api/v1/domotic/token",
+            headers={
+                "Authorization": tokens["AuthenticationResult"]["IdToken"],
+            },
+        ) as response:
+            token = await response.json()
+
+            if "token" not in token:
+                raise NexityServiceException("No Nexity SSO token provided.")
+
+            sso_token = cast(str, token["token"])
+
+        user_id = self.username.replace("@", "_-_")  # Replace @ for _-_
+        payload = {"ssoToken": sso_token, "userId": user_id}
+
+        post_response = await self.post("login", data=payload)
+
+        return "success" in post_response