diff --git a/README.md b/README.md index 3033de79..e0c830b2 100644 --- a/README.md +++ b/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 diff --git a/mypy.ini b/mypy.ini index 13ea0941..0bb1e81d 100644 --- a/mypy.ini +++ b/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 diff --git a/pyoverkiz/clients/atlantic_cozytouch.py b/pyoverkiz/clients/atlantic_cozytouch.py new file mode 100644 index 00000000..7b6c7248 --- /dev/null +++ b/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 diff --git a/pyoverkiz/clients/default.py b/pyoverkiz/clients/default.py new file mode 100644 index 00000000..5ca9145d --- /dev/null +++ b/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 diff --git a/pyoverkiz/clients/nexity.py b/pyoverkiz/clients/nexity.py new file mode 100644 index 00000000..05f26e7b --- /dev/null +++ b/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 diff --git a/pyoverkiz/client.py b/pyoverkiz/clients/overkiz.py similarity index 58% rename from pyoverkiz/client.py rename to pyoverkiz/clients/overkiz.py index a573aaf0..4c9a665f 100644 --- a/pyoverkiz/client.py +++ b/pyoverkiz/clients/overkiz.py @@ -1,51 +1,30 @@ -""" Python wrapper for the OverKiz API """ +""" Python wrapper for the OverKiz API""" + from __future__ import annotations -import asyncio -import datetime import urllib.parse -from collections.abc import Mapping +from abc import ABC, abstractmethod from json import JSONDecodeError -from types import TracebackType -from typing import Any, cast +from typing import Any, Mapping, cast import backoff -import boto3 import humps -from aiohttp import ClientResponse, ClientSession, FormData, ServerDisconnectedError -from botocore.config import Config -from warrant_lite import WarrantLite - -from pyoverkiz.const import ( - COZYTOUCH_ATLANTIC_API, - COZYTOUCH_CLIENT_ID, - NEXITY_API, - NEXITY_COGNITO_CLIENT_ID, - NEXITY_COGNITO_REGION, - NEXITY_COGNITO_USER_POOL, - SOMFY_API, - SOMFY_CLIENT_ID, - SOMFY_CLIENT_SECRET, - SUPPORTED_SERVERS, -) +from aiohttp import ClientResponse, ClientSession, ServerDisconnectedError +from attr import define +from attrs import field + from pyoverkiz.exceptions import ( AccessDeniedToGatewayException, BadCredentialsException, - CozyTouchBadCredentialsException, - CozyTouchServiceException, InvalidCommandException, InvalidEventListenerIdException, InvalidTokenException, MaintenanceException, MissingAuthorizationTokenException, - NexityBadCredentialsException, - NexityServiceException, NoRegisteredEventListenerException, NotAuthenticatedException, NotSuchTokenException, SessionAndBearerInSameRequestException, - SomfyBadCredentialsException, - SomfyServiceException, TooManyAttemptsBannedException, TooManyConcurrentRequestsException, TooManyExecutionsException, @@ -61,7 +40,6 @@ Gateway, HistoryExecution, LocalToken, - OverkizServer, Place, Scenario, Setup, @@ -79,284 +57,81 @@ async def refresh_listener(invocation: Mapping[str, Any]) -> None: await invocation["args"][0].register_event_listener() -# pylint: disable=too-many-instance-attributes, too-many-branches - - -class OverkizClient: - """Interface class for the Overkiz API""" +@define(kw_only=True) +class OverkizClient(ABC): + """Abstract class for the Overkiz API""" - username: str - password: str - server: OverkizServer - setup: Setup | None - devices: list[Device] - gateways: list[Gateway] - event_listener_id: str | None + name: str + endpoint: str + manufacturer: str session: ClientSession - - _refresh_token: str | None = None - _expires_in: datetime.datetime | None = None - _access_token: str | None = None - - def __init__( - self, - username: str, - password: str, - server: OverkizServer, - token: str | None = None, - session: ClientSession | None = None, - ) -> None: - """ - Constructor - - :param username: the username - :param password: the password - :param server: OverkizServer - :param session: optional ClientSession - """ - - self.username = username - self.password = password - self.server = server - self._access_token = token - - self.setup: Setup | None = None - self.devices: list[Device] = [] - self.gateways: list[Gateway] = [] - self.event_listener_id: str | None = None - - self.session = session if session else ClientSession() - - async def __aenter__(self) -> OverkizClient: - return self - - async def __aexit__( - self, - exc_type: type[BaseException] | None, - exc_value: BaseException | None, - traceback: TracebackType | None, - ) -> None: - await self.close() - - async def close(self) -> None: - """Close the session.""" - if self.event_listener_id: - await self.unregister_event_listener() - - await self.session.close() - - async def login( + configuration_url: str | None + event_listener_id: str | None = field(default=None, init=False) + setup: Setup | None = field(default=None, init=False) + devices: list[Device] = field(factory=list, init=False) + gateways: list[Gateway] = field(factory=list, init=False) + _ssl: bool = field(default=True, init=False) + + @abstractmethod + async def _login( self, - register_event_listener: bool | None = True, ) -> bool: + """Login to the server.""" + + async def login(self, register_event_listener: bool = True) -> bool: """ Authenticate and create an API session allowing access to the other operations. Caller must provide one of [userId+userPassword, userId+ssoToken, accessToken, jwt] """ - # Local authentication - if "/enduser-mobile-web/1/enduserAPI/" in self.server.endpoint: - if register_event_listener: - await self.register_event_listener() - else: - # Call a simple endpoint to verify if our token is correct - await self.get_gateways() - - return True - - # Somfy TaHoma authentication using access_token - if self.server == SUPPORTED_SERVERS["somfy_europe"]: - await self.somfy_tahoma_get_access_token() - + if await self._login(): if register_event_listener: await self.register_event_listener() - - return True - - # CozyTouch authentication using jwt - if self.server == SUPPORTED_SERVERS["atlantic_cozytouch"]: - jwt = await self.cozytouch_login() - payload = {"jwt": jwt} - - # Nexity authentication using ssoToken - elif self.server == SUPPORTED_SERVERS["nexity"]: - sso_token = await self.nexity_login() - user_id = self.username.replace("@", "_-_") # Replace @ for _-_ - payload = {"ssoToken": sso_token, "userId": user_id} - - # Regular authentication using userId+userPassword - else: - payload = {"userId": self.username, "userPassword": self.password} - - response = await self.__post("login", data=payload) - - if response.get("success"): - if register_event_listener: - await self.register_event_listener() - return True - return False - async def somfy_tahoma_get_access_token(self) -> str: - """ - Authenticate via Somfy identity and acquire access_token. - """ - # Request access token - async with self.session.post( - SOMFY_API + "/oauth/oauth/v2/token", - data=FormData( - { - "grant_type": "password", - "username": self.username, - "password": self.password, - "client_id": SOMFY_CLIENT_ID, - "client_secret": SOMFY_CLIENT_SECRET, - } - ), - headers={ - "Content-Type": "application/x-www-form-urlencoded", - }, - ) as response: - token = await response.json() - - # { "message": "error.invalid.grant", "data": [], "uid": "xxx" } - if "message" in token and token["message"] == "error.invalid.grant": - raise SomfyBadCredentialsException(token["message"]) - - if "access_token" not in token: - raise SomfyServiceException("No Somfy access token provided.") + @property + def _headers(self) -> dict[str, str]: + return {} - self._access_token = cast(str, token["access_token"]) - self._refresh_token = token["refresh_token"] - self._expires_in = datetime.datetime.now() + datetime.timedelta( - seconds=token["expires_in"] - 5 - ) - - return self._access_token - - async def refresh_token(self) -> None: - """ - Update the access and the refresh token. The refresh token will be valid 14 days. - """ - if self.server != SUPPORTED_SERVERS["somfy_europe"]: - return - - if not self._refresh_token: - raise ValueError("No refresh token provided. Login method must be used.") + async def get( + self, + path: str, + ) -> Any: + """Make a GET request to the OverKiz API""" - # &grant_type=refresh_token&refresh_token=REFRESH_TOKEN - # Request access token - async with self.session.post( - SOMFY_API + "/oauth/oauth/v2/token", - data=FormData( - { - "grant_type": "refresh_token", - "refresh_token": self._refresh_token, - "client_id": SOMFY_CLIENT_ID, - "client_secret": SOMFY_CLIENT_SECRET, - } - ), - headers={ - "Content-Type": "application/x-www-form-urlencoded", - }, + async with self.session.get( + f"{self.endpoint}{path}", headers=self._headers, ssl=self._ssl ) as response: - token = await response.json() - # { "message": "error.invalid.grant", "data": [], "uid": "xxx" } - if "message" in token and token["message"] == "error.invalid.grant": - raise SomfyBadCredentialsException(token["message"]) - - if "access_token" not in token: - raise SomfyServiceException("No Somfy access token provided.") + await self.check_response(response) + return await response.json() - self._access_token = cast(str, token["access_token"]) - self._refresh_token = token["refresh_token"] - self._expires_in = datetime.datetime.now() + datetime.timedelta( - seconds=token["expires_in"] - 5 - ) + async def post( + self, + path: str, + payload: JSON | None = None, + data: JSON | None = None, + ) -> Any: + """Make a POST request to the OverKiz API""" - async def cozytouch_login(self) -> str: - """ - Authenticate via CozyTouch identity and acquire JWT token. - """ - # Request access token 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']}"}, + f"{self.endpoint}{path}", + data=data, + json=payload, + headers=self._headers, + ssl=self._ssl, ) as response: - jwt = await response.text() - - if not jwt: - raise CozyTouchServiceException("No JWT token provided.") - - jwt = jwt.strip('"') # Remove surrounding quotes - - return jwt - - async def nexity_login(self) -> str: - """ - Authenticate via Nexity identity and acquire SSO token. - """ - 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 + await self.check_response(response) + return await response.json() - id_token = tokens["AuthenticationResult"]["IdToken"] + async def delete( + self, + path: str, + ) -> None: + """Make a DELETE request to the OverKiz API""" - async with self.session.get( - NEXITY_API + "/deploy/api/v1/domotic/token", - headers={ - "Authorization": id_token, - }, + async with self.session.delete( + f"{self.endpoint}{path}", headers=self._headers, ssl=self._ssl ) as response: - token = await response.json() - - if "token" not in token: - raise NexityServiceException("No Nexity SSO token provided.") - - return cast(str, token["token"]) + await self.check_response(response) @backoff.on_exception( backoff.expo, @@ -384,7 +159,7 @@ async def get_setup(self, refresh: bool = False) -> Setup: if self.setup and not refresh: return self.setup - response = await self.__get("setup") + response = await self.get("setup") setup = Setup(**humps.decamelize(response)) @@ -411,7 +186,7 @@ async def get_diagnostic_data(self) -> JSON: This data will be masked to not return any confidential or PII data. """ - response = await self.__get("setup") + response = await self.get("setup") return obfuscate_sensitive_data(response) @@ -429,7 +204,7 @@ async def get_devices(self, refresh: bool = False) -> list[Device]: if self.devices and not refresh: return self.devices - response = await self.__get("setup/devices") + response = await self.get("setup/devices") devices = [Device(**d) for d in humps.decamelize(response)] # Cache response @@ -453,7 +228,7 @@ async def get_gateways(self, refresh: bool = False) -> list[Gateway]: if self.gateways and not refresh: return self.gateways - response = await self.__get("setup/gateways") + response = await self.get("setup/gateways") gateways = [Gateway(**g) for g in humps.decamelize(response)] # Cache response @@ -473,7 +248,7 @@ async def get_execution_history(self) -> list[HistoryExecution]: """ List execution history """ - response = await self.__get("history/executions") + response = await self.get("history/executions") execution_history = [HistoryExecution(**h) for h in humps.decamelize(response)] return execution_history @@ -485,7 +260,7 @@ async def get_device_definition(self, deviceurl: str) -> JSON | None: """ Retrieve a particular setup device definition """ - response: dict = await self.__get( + response: dict = await self.get( f"setup/devices/{urllib.parse.quote_plus(deviceurl)}" ) @@ -498,7 +273,7 @@ async def get_state(self, deviceurl: str) -> list[State]: """ Retrieve states of requested device """ - response = await self.__get( + response = await self.get( f"setup/devices/{urllib.parse.quote_plus(deviceurl)}/states" ) state = [State(**s) for s in humps.decamelize(response)] @@ -512,24 +287,7 @@ async def refresh_states(self) -> None: """ Ask the box to refresh all devices states for protocols supporting that operation """ - await self.__post("setup/devices/states/refresh") - - @backoff.on_exception(backoff.expo, TooManyConcurrentRequestsException, max_tries=5) - 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.__post("events/register") - listener_id = cast(str, response.get("id")) - self.event_listener_id = listener_id - - return listener_id + await self.post("setup/devices/states/refresh") @backoff.on_exception(backoff.expo, TooManyConcurrentRequestsException, max_tries=5) @backoff.on_exception( @@ -548,19 +306,34 @@ async def fetch_events(self) -> list[Event]: Per-session rate-limit : 1 calls per 1 SECONDS period for this particular operation (polling) """ - await self._refresh_token_if_expired() - response = await self.__post(f"events/{self.event_listener_id}/fetch") + response = await self.post(f"events/{self.event_listener_id}/fetch") events = [Event(**e) for e in humps.decamelize(response)] return events + @backoff.on_exception(backoff.expo, TooManyConcurrentRequestsException, max_tries=5) + 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.post("events/register") + listener_id = cast(str, response.get("id")) + self.event_listener_id = listener_id + + return listener_id + async def unregister_event_listener(self) -> None: """ Unregister an event listener. API response status is always 200, even on unknown listener ids. """ - await self._refresh_token_if_expired() - await self.__post(f"events/{self.event_listener_id}/unregister") + await self.post(f"events/{self.event_listener_id}/unregister") self.event_listener_id = None @backoff.on_exception( @@ -568,7 +341,7 @@ async def unregister_event_listener(self) -> None: ) async def get_current_execution(self, exec_id: str) -> Execution: """Get an action group execution currently running""" - response = await self.__get(f"exec/current/{exec_id}") + response = await self.get(f"exec/current/{exec_id}") execution = Execution(**humps.decamelize(response)) return execution @@ -578,7 +351,7 @@ async def get_current_execution(self, exec_id: str) -> Execution: ) async def get_current_executions(self) -> list[Execution]: """Get all action groups executions currently running""" - response = await self.__get("exec/current") + response = await self.get("exec/current") executions = [Execution(**e) for e in humps.decamelize(response)] return executions @@ -588,7 +361,7 @@ async def get_current_executions(self) -> list[Execution]: ) async def get_api_version(self) -> str: """Get the API version (local only)""" - response = await self.__get("apiVersion") + response = await self.get("apiVersion") return cast(str, response["protocolVersion"]) @@ -618,7 +391,7 @@ async def execute_command( ) async def cancel_command(self, exec_id: str) -> None: """Cancel a running setup-level execution""" - await self.__delete(f"/exec/current/setup/{exec_id}") + await self.delete(f"/exec/current/setup/{exec_id}") @backoff.on_exception( backoff.expo, NotAuthenticatedException, max_tries=2, on_backoff=relogin @@ -634,7 +407,7 @@ async def execute_commands( "label": label, "actions": [{"deviceURL": device_url, "commands": commands}], } - response: dict = await self.__post("exec/apply", payload) + response: dict = await self.post("exec/apply", payload) return cast(str, response["execId"]) @backoff.on_exception( @@ -645,7 +418,7 @@ async def execute_commands( ) async def get_scenarios(self) -> list[Scenario]: """List the scenarios""" - response = await self.__get("actionGroups") + response = await self.get("actionGroups") return [Scenario(**scenario) for scenario in response] @backoff.on_exception( @@ -656,7 +429,7 @@ async def get_scenarios(self) -> list[Scenario]: ) async def get_places(self) -> Place: """List the places""" - response = await self.__get("setup/places") + response = await self.get("setup/places") places = Place(**humps.decamelize(response)) return places @@ -671,7 +444,7 @@ async def generate_local_token(self, gateway_id: str) -> str: Generates a new token Access scope : Full enduser API access (enduser/*) """ - response = await self.__get(f"config/{gateway_id}/local/tokens/generate") + response = await self.get(f"config/{gateway_id}/local/tokens/generate") return cast(str, response["token"]) @@ -688,7 +461,7 @@ async def activate_local_token( Create a token Access scope : Full enduser API access (enduser/*) """ - response = await self.__post( + response = await self.post( f"config/{gateway_id}/local/tokens", {"label": label, "token": token, "scope": scope}, ) @@ -708,7 +481,7 @@ async def get_local_tokens( Get all gateway tokens with the given scope Access scope : Full enduser API access (enduser/*) """ - response = await self.__get(f"config/{gateway_id}/local/tokens/{scope}") + response = await self.get(f"config/{gateway_id}/local/tokens/{scope}") local_tokens = [LocalToken(**lt) for lt in humps.decamelize(response)] return local_tokens @@ -724,7 +497,7 @@ async def delete_local_token(self, gateway_id: str, uuid: str) -> bool: Delete a token Access scope : Full enduser API access (enduser/*) """ - await self.__delete(f"config/{gateway_id}/local/tokens/{uuid}") + await self.delete(f"config/{gateway_id}/local/tokens/{uuid}") return True @@ -733,7 +506,7 @@ async def delete_local_token(self, gateway_id: str, uuid: str) -> bool: ) async def execute_scenario(self, oid: str) -> str: """Execute a scenario""" - response = await self.__post(f"exec/{oid}") + response = await self.post(f"exec/{oid}") return cast(str, response["execId"]) @backoff.on_exception( @@ -744,57 +517,15 @@ async def execute_scenario(self, oid: str) -> str: ) async def execute_scheduled_scenario(self, oid: str, timestamp: int) -> str: """Execute a scheduled scenario""" - response = await self.__post(f"exec/schedule/{oid}/{timestamp}") + response = await self.post(f"exec/schedule/{oid}/{timestamp}") return cast(str, response["triggerId"]) - async def __get(self, path: str) -> Any: - """Make a GET request to the OverKiz API""" - headers = {} - - await self._refresh_token_if_expired() - if self._access_token: - headers["Authorization"] = f"Bearer {self._access_token}" - - async with self.session.get( - f"{self.server.endpoint}{path}", - headers=headers, - ) as response: - await self.check_response(response) - return await response.json() - - async def __post( - self, path: str, payload: JSON | None = None, data: JSON | None = None - ) -> Any: - """Make a POST request to the OverKiz API""" - headers = {} - - if path != "login" and self._access_token: - await self._refresh_token_if_expired() - headers["Authorization"] = f"Bearer {self._access_token}" - - async with self.session.post( - f"{self.server.endpoint}{path}", data=data, json=payload, headers=headers - ) as response: - await self.check_response(response) - return await response.json() - - async def __delete(self, path: str) -> None: - """Make a DELETE request to the OverKiz API""" - headers = {} - - await self._refresh_token_if_expired() - - if self._access_token: - headers["Authorization"] = f"Bearer {self._access_token}" - - async with self.session.delete( - f"{self.server.endpoint}{path}", headers=headers - ) as response: - await self.check_response(response) - @staticmethod async def check_response(response: ClientResponse) -> None: """Check the response returned by the OverKiz API""" + + # pylint: disable=too-many-branches + if response.status in [200, 204]: return @@ -875,15 +606,3 @@ async def check_response(response: ClientResponse) -> None: raise AccessDeniedToGatewayException(message) raise Exception(message if message else result) - - async def _refresh_token_if_expired(self) -> None: - """Check if token is expired and request a new one.""" - if ( - self._expires_in - and self._refresh_token - and self._expires_in <= datetime.datetime.now() - ): - await self.refresh_token() - - if self.event_listener_id: - await self.register_event_listener() diff --git a/pyoverkiz/clients/somfy.py b/pyoverkiz/clients/somfy.py new file mode 100644 index 00000000..7d725dba --- /dev/null +++ b/pyoverkiz/clients/somfy.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +import datetime +from typing import Any, cast + +from aiohttp import FormData +from attr import define, field + +from pyoverkiz.clients.overkiz import OverkizClient +from pyoverkiz.exceptions import SomfyBadCredentialsException, SomfyServiceException +from pyoverkiz.types import JSON + +SOMFY_API = "https://accounts.somfy.com" +SOMFY_CLIENT_ID = "0d8e920c-1478-11e7-a377-02dd59bd3041_1ewvaqmclfogo4kcsoo0c8k4kso884owg08sg8c40sk4go4ksg" +SOMFY_CLIENT_SECRET = "12k73w1n540g8o4cokg0cw84cog840k84cwggscwg884004kgk" + + +@define(kw_only=True) +class SomfyClient(OverkizClient): + + username: str + password: str = field(repr=lambda _: "***") + _access_token: str | None = None + _refresh_token: str | None = None + _expires_in: datetime.datetime | None = None + + @property + def _headers(self) -> dict[str, str]: + return {"Authorization": f"Bearer {self._access_token}"} + + async def get(self, path: str) -> Any: + """Make a GET request to the OverKiz API""" + + await self._refresh_token_if_expired() + return await super().get(path) + + async def post( + self, + path: str, + payload: JSON | None = None, + data: JSON | None = None, + ) -> Any: + """Make a POST request to the OverKiz API""" + if path != "login": + await self._refresh_token_if_expired() + return await super().post(path, payload=payload, data=data) + + async def delete(self, path: str) -> None: + """Make a DELETE request to the OverKiz API""" + await self._refresh_token_if_expired() + return await super().delete(path) + + 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( + SOMFY_API + "/oauth/oauth/v2/token", + data=FormData( + { + "grant_type": "password", + "username": self.username, + "password": self.password, + "client_id": SOMFY_CLIENT_ID, + "client_secret": SOMFY_CLIENT_SECRET, + } + ), + headers={ + "Content-Type": "application/x-www-form-urlencoded", + }, + ) as response: + token = await response.json() + + # { "message": "error.invalid.grant", "data": [], "uid": "xxx" } + if "message" in token and token["message"] == "error.invalid.grant": + raise SomfyBadCredentialsException(token["message"]) + + if "access_token" not in token: + raise SomfyServiceException("No Somfy access token provided.") + + self._access_token = cast(str, token["access_token"]) + self._refresh_token = token["refresh_token"] + self._expires_in = datetime.datetime.now() + datetime.timedelta( + seconds=token["expires_in"] - 5 + ) + + return True + + async def refresh_token(self) -> None: + """ + Update the access and the refresh token. The refresh token will be valid 14 days. + """ + + if not self._refresh_token: + raise ValueError("No refresh token provided. Login method must be used.") + + # &grant_type=refresh_token&refresh_token=REFRESH_TOKEN + # Request access token + async with self.session.post( + SOMFY_API + "/oauth/oauth/v2/token", + data=FormData( + { + "grant_type": "refresh_token", + "refresh_token": self._refresh_token, + "client_id": SOMFY_CLIENT_ID, + "client_secret": SOMFY_CLIENT_SECRET, + } + ), + headers={ + "Content-Type": "application/x-www-form-urlencoded", + }, + ) as response: + token = await response.json() + # { "message": "error.invalid.grant", "data": [], "uid": "xxx" } + if "message" in token and token["message"] == "error.invalid.grant": + raise SomfyBadCredentialsException(token["message"]) + + if "access_token" not in token: + raise SomfyServiceException("No Somfy access token provided.") + + self._access_token = cast(str, token["access_token"]) + self._refresh_token = token["refresh_token"] + self._expires_in = datetime.datetime.now() + datetime.timedelta( + seconds=token["expires_in"] - 5 + ) + + async def _refresh_token_if_expired(self) -> None: + """Check if token is expired and request a new one.""" + if ( + self._expires_in + and self._refresh_token + and self._expires_in <= datetime.datetime.now() + ): + await self.refresh_token() + + if self.event_listener_id: + await self.register_event_listener() diff --git a/pyoverkiz/clients/somfy_local.py b/pyoverkiz/clients/somfy_local.py new file mode 100644 index 00000000..b1136d17 --- /dev/null +++ b/pyoverkiz/clients/somfy_local.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +from attr import define, field + +from pyoverkiz.clients.overkiz import OverkizClient + + +@define(kw_only=True) +class SomfyLocalClient(OverkizClient): + + _ssl: bool = field(default=False, init=False) + token: str = field(repr=lambda _: "***") + + async def _login(self) -> bool: + """There is no login needed for Somfy Local API""" + raise NotImplementedError("There is no login needed for the Somfy Local API") + + @property + def _headers(self) -> dict[str, str]: + return {"Authorization": f"Bearer {self.token}"} diff --git a/pyoverkiz/const.py b/pyoverkiz/const.py index 48478e97..0ca9d572 100644 --- a/pyoverkiz/const.py +++ b/pyoverkiz/const.py @@ -1,104 +1,22 @@ from __future__ import annotations -from pyoverkiz.models import OverkizServer +from enum import Enum, unique -COZYTOUCH_ATLANTIC_API = "https://apis.groupe-atlantic.com" -COZYTOUCH_CLIENT_ID = ( - "Q3RfMUpWeVRtSUxYOEllZkE3YVVOQmpGblpVYToyRWNORHpfZHkzNDJVSnFvMlo3cFNKTnZVdjBh" -) -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" - -SOMFY_API = "https://accounts.somfy.com" -SOMFY_CLIENT_ID = "0d8e920c-1478-11e7-a377-02dd59bd3041_1ewvaqmclfogo4kcsoo0c8k4kso884owg08sg8c40sk4go4ksg" -SOMFY_CLIENT_SECRET = "12k73w1n540g8o4cokg0cw84cog840k84cwggscwg884004kgk" - -SUPPORTED_SERVERS: dict[str, OverkizServer] = { - "atlantic_cozytouch": OverkizServer( - name="Atlantic Cozytouch", - endpoint="https://ha110-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Atlantic", - configuration_url=None, - ), - "brandt": OverkizServer( - name="Brandt Smart Control", - endpoint="https://ha3-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Brandt", - configuration_url=None, - ), - "flexom": OverkizServer( - name="Flexom", - endpoint="https://ha108-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Bouygues", - configuration_url=None, - ), - "hexaom_hexaconnect": OverkizServer( - name="Hexaom HexaConnect", - endpoint="https://ha5-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Hexaom", - configuration_url=None, - ), - "hi_kumo_asia": OverkizServer( - name="Hitachi Hi Kumo (Asia)", - endpoint="https://ha203-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Hitachi", - configuration_url=None, - ), - "hi_kumo_europe": OverkizServer( - name="Hitachi Hi Kumo (Europe)", - endpoint="https://ha117-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Hitachi", - configuration_url=None, - ), - "hi_kumo_oceania": OverkizServer( - name="Hitachi Hi Kumo (Oceania)", - endpoint="https://ha203-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Hitachi", - configuration_url=None, - ), - "nexity": OverkizServer( - name="Nexity Eugénie", - endpoint="https://ha106-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Nexity", - configuration_url=None, - ), - "rexel": OverkizServer( - name="Rexel Energeasy Connect", - endpoint="https://ha112-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Rexel", - configuration_url="https://utilisateur.energeasyconnect.com/user/#/zone/equipements", - ), - "simu_livein2": OverkizServer( # alias of https://tahomalink.com - name="SIMU (LiveIn2)", - endpoint="https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Somfy", - configuration_url=None, - ), - "somfy_europe": OverkizServer( # alias of https://tahomalink.com - name="Somfy (Europe)", - endpoint="https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Somfy", - configuration_url="https://www.tahomalink.com", - ), - "somfy_america": OverkizServer( - name="Somfy (North America)", - endpoint="https://ha401-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Somfy", - configuration_url=None, - ), - "somfy_oceania": OverkizServer( - name="Somfy (Oceania)", - endpoint="https://ha201-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Somfy", - configuration_url=None, - ), - "ubiwizz": OverkizServer( - name="Ubiwizz", - endpoint="https://ha129-1.overkiz.com/enduser-mobile-web/enduserAPI/", - manufacturer="Decelect", - configuration_url=None, - ), -} +@unique +class Server(str, Enum): + ATLANTIC_COZYTOUCH = "atlantic_cozytouch" + BRANDT = "brandt" + FLEXOM = "flexom" + HEXAOM_HEXACONNECT = "hexaom_hexaconnect" + HI_KUMO_ASIA = "hi_kumo_asia" + HI_KUMO_EUROPE = "hi_kumo_europe" + HI_KUMO_OCEANIA = "hi_kumo_oceania" + NEXITY = "nexity" + REXEL = "rexel" + SIMU_LIVEIN2 = "simu_livein2" + SOMFY_EUROPE = "somfy_europe" + SOMFY_DEV_MODE = "somfy_dev_mode" + SOMFY_AMERICA = "somfy_america" + SOMFY_OCEANIA = "somfy_oceania" + UBIWIZZ = "ubiwizz" diff --git a/pyoverkiz/enums/execution.py b/pyoverkiz/enums/execution.py index f5feb482..201c9030 100644 --- a/pyoverkiz/enums/execution.py +++ b/pyoverkiz/enums/execution.py @@ -15,7 +15,7 @@ class ExecutionType(str, Enum): RAW_TRIGGER_GATEWAY = "Raw trigger (Gateway)" @classmethod - def _missing_(cls, value): # type: ignore + def _missing_(cls, value): # type: ignore[no-untyped-def] _LOGGER.warning(f"Unsupported value {value} has been returned for {cls}") return cls.UNKNOWN @@ -33,7 +33,7 @@ class ExecutionState(str, Enum): QUEUED_SERVER_SIDE = "QUEUED_SERVER_SIDE" @classmethod - def _missing_(cls, value): # type: ignore + def _missing_(cls, value): # type: ignore[no-untyped-def] _LOGGER.warning(f"Unsupported value {value} has been returned for {cls}") return cls.UNKNOWN @@ -56,6 +56,6 @@ class ExecutionSubType(str, Enum): TIME_TRIGGER = "TIME_TRIGGER" @classmethod - def _missing_(cls, value): # type: ignore + def _missing_(cls, value): # type: ignore[no-untyped-def] _LOGGER.warning(f"Unsupported value {value} has been returned for {cls}") return cls.UNKNOWN diff --git a/pyoverkiz/enums/gateway.py b/pyoverkiz/enums/gateway.py index 41476034..c5a487ba 100644 --- a/pyoverkiz/enums/gateway.py +++ b/pyoverkiz/enums/gateway.py @@ -51,7 +51,7 @@ class GatewayType(IntEnum): TAHOMA_RAIL_DIN_S = 108 @classmethod - def _missing_(cls, value): # type: ignore + def _missing_(cls, value): # type: ignore[no-untyped-def] _LOGGER.warning(f"Unsupported value {value} has been returned for {cls}") return cls.UNKNOWN @@ -85,7 +85,7 @@ class GatewaySubType(IntEnum): # TAHOMA_BOX_C_IO = 12 That’s probably 17, but tahomalink.com says it’s 12 @classmethod - def _missing_(cls, value): # type: ignore + def _missing_(cls, value): # type: ignore[no-untyped-def] _LOGGER.warning(f"Unsupported value {value} has been returned for {cls}") return cls.UNKNOWN diff --git a/pyoverkiz/enums/general.py b/pyoverkiz/enums/general.py index 42dc4355..f36c24ec 100644 --- a/pyoverkiz/enums/general.py +++ b/pyoverkiz/enums/general.py @@ -234,7 +234,7 @@ class FailureType(IntEnum): TIME_OUT_ON_COMMAND_PROGRESS = 20003 @classmethod - def _missing_(cls, value): # type: ignore + def _missing_(cls, value): # type: ignore[no-untyped-def] _LOGGER.warning(f"Unsupported value {value} has been returned for {cls}") return cls.UNKNOWN @@ -416,6 +416,6 @@ class EventName(str, Enum): ZONE_UPDATED = "ZoneUpdatedEvent" @classmethod - def _missing_(cls, value): # type: ignore + def _missing_(cls, value): # type: ignore[no-untyped-def] _LOGGER.warning(f"Unsupported value {value} has been returned for {cls}") return cls.UNKNOWN diff --git a/pyoverkiz/enums/protocol.py b/pyoverkiz/enums/protocol.py index 40146371..3ebfbe2a 100644 --- a/pyoverkiz/enums/protocol.py +++ b/pyoverkiz/enums/protocol.py @@ -42,6 +42,6 @@ class Protocol(str, Enum): RTN = "rtn" @classmethod - def _missing_(cls, value): # type: ignore + def _missing_(cls, value): # type: ignore[no-untyped-def] _LOGGER.warning(f"Unsupported protocol {value} has been returned for {cls}") return cls.UNKNOWN diff --git a/pyoverkiz/enums/ui.py b/pyoverkiz/enums/ui.py index 0f55e7a1..f022e353 100644 --- a/pyoverkiz/enums/ui.py +++ b/pyoverkiz/enums/ui.py @@ -76,7 +76,7 @@ class UIClass(str, Enum): UNKNOWN = "unknown" @classmethod - def _missing_(cls, value): # type: ignore + def _missing_(cls, value): # type: ignore[no-untyped-def] _LOGGER.warning(f"Unsupported value {value} has been returned for {cls}") return cls.UNKNOWN @@ -408,6 +408,6 @@ class UIWidget(str, Enum): UNKNOWN = "unknown" @classmethod - def _missing_(cls, value): # type: ignore + def _missing_(cls, value): # type: ignore[no-untyped-def] _LOGGER.warning(f"Unsupported value {value} has been returned for {cls}") return cls.UNKNOWN diff --git a/pyoverkiz/models.py b/pyoverkiz/models.py index b8abfcc7..a7956e58 100644 --- a/pyoverkiz/models.py +++ b/pyoverkiz/models.py @@ -779,16 +779,6 @@ def __init__( self.oid = oid -@define(kw_only=True) -class OverkizServer: - """Class to describe an Overkiz server.""" - - name: str - endpoint: str - manufacturer: str - configuration_url: str | None - - @define(kw_only=True) class LocalToken: label: str diff --git a/pyoverkiz/overkiz.py b/pyoverkiz/overkiz.py new file mode 100644 index 00000000..d7f1268b --- /dev/null +++ b/pyoverkiz/overkiz.py @@ -0,0 +1,160 @@ +"""Main entropoint for the Overkiz API client.""" +from __future__ import annotations + +from typing import Callable + +from aiohttp import ClientSession + +from pyoverkiz.clients.atlantic_cozytouch import AtlanticCozytouchClient +from pyoverkiz.clients.default import DefaultClient +from pyoverkiz.clients.nexity import NexityClient +from pyoverkiz.clients.overkiz import OverkizClient +from pyoverkiz.clients.somfy import SomfyClient +from pyoverkiz.clients.somfy_local import SomfyLocalClient +from pyoverkiz.const import Server + +SUPPORTED_SERVERS: dict[Server, Callable[[str, str, ClientSession], OverkizClient]] = { + Server.ATLANTIC_COZYTOUCH: lambda username, password, session: AtlanticCozytouchClient( + name="Atlantic Cozytouch", + endpoint="https://ha110-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Atlantic", + configuration_url=None, + session=session, + username=username, + password=password, + ), + Server.BRANDT: lambda username, password, session: DefaultClient( + name="Brandt Smart Control", + endpoint="https://ha3-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Brandt", + configuration_url=None, + session=session, + username=username, + password=password, + ), + Server.FLEXOM: lambda username, password, session: DefaultClient( + name="Flexom", + endpoint="https://ha108-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Bouygues", + configuration_url=None, + session=session, + username=username, + password=password, + ), + Server.HEXAOM_HEXACONNECT: lambda username, password, session: DefaultClient( + name="Hexaom HexaConnect", + endpoint="https://ha5-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Hexaom", + configuration_url=None, + session=session, + username=username, + password=password, + ), + Server.HI_KUMO_ASIA: lambda username, password, session: DefaultClient( + name="Hitachi Hi Kumo (Asia)", + endpoint="https://ha203-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Hitachi", + configuration_url=None, + session=session, + username=username, + password=password, + ), + Server.HI_KUMO_EUROPE: lambda username, password, session: DefaultClient( + name="Hitachi Hi Kumo (Europe)", + endpoint="https://ha117-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Hitachi", + configuration_url=None, + session=session, + username=username, + password=password, + ), + Server.HI_KUMO_OCEANIA: lambda username, password, session: DefaultClient( + name="Hitachi Hi Kumo (Oceania)", + endpoint="https://ha203-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Hitachi", + configuration_url=None, + session=session, + username=username, + password=password, + ), + Server.NEXITY: lambda username, password, session: NexityClient( + name="Nexity Eugénie", + endpoint="https://ha106-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Nexity", + configuration_url=None, + session=session, + username=username, + password=password, + ), + Server.REXEL: lambda username, password, session: DefaultClient( + name="Rexel Energeasy Connect", + endpoint="https://ha112-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Rexel", + configuration_url="https://utilisateur.energeasyconnect.com/user/#/zone/equipements", + session=session, + username=username, + password=password, + ), + Server.SIMU_LIVEIN2: lambda username, password, session: DefaultClient( # alias of https://tahomalink.com + name="SIMU (LiveIn2)", + endpoint="https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Somfy", + configuration_url=None, + session=session, + username=username, + password=password, + ), + Server.SOMFY_EUROPE: lambda username, password, session: SomfyClient( # alias of https://tahomalink.com + name="Somfy (Europe)", + endpoint="https://ha101-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Somfy", + configuration_url="https://www.tahomalink.com", + session=session, + username=username, + password=password, + ), + Server.SOMFY_DEV_MODE: lambda domain, token, session: SomfyLocalClient( + name="Somfy Developer Mode (Local API)", + endpoint=f"https://{domain}:8443/enduser-mobile-web/1/enduserAPI/", + manufacturer="Somfy", + configuration_url=None, + session=session, + token=token, + ), + Server.SOMFY_AMERICA: lambda username, password, session: DefaultClient( + name="Somfy (North America)", + endpoint="https://ha401-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Somfy", + configuration_url=None, + session=session, + username=username, + password=password, + ), + Server.SOMFY_OCEANIA: lambda username, password, session: DefaultClient( + name="Somfy (Oceania)", + endpoint="https://ha201-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Somfy", + configuration_url=None, + session=session, + username=username, + password=password, + ), + Server.UBIWIZZ: lambda username, password, session: DefaultClient( + name="Ubiwizz", + endpoint="https://ha129-1.overkiz.com/enduser-mobile-web/enduserAPI/", + manufacturer="Decelect", + configuration_url=None, + session=session, + username=username, + password=password, + ), +} + + +class Overkiz: + @staticmethod + def create_client( + server: Server, username: str, password: str, session: ClientSession + ) -> OverkizClient: + """Get the client for the given server""" + return SUPPORTED_SERVERS[server](username, password, session) diff --git a/tests/test_client.py b/tests/test_client.py index db999a85..cd80bec6 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -6,9 +6,9 @@ import pytest from pytest import fixture -from pyoverkiz.client import OverkizClient -from pyoverkiz.const import SUPPORTED_SERVERS +from pyoverkiz.const import Server from pyoverkiz.enums import DataType +from pyoverkiz.overkiz import Overkiz CURRENT_DIR = os.path.dirname(os.path.abspath(__file__)) @@ -16,7 +16,9 @@ class TestOverkizClient: @fixture def client(self): - return OverkizClient("username", "password", SUPPORTED_SERVERS["somfy_europe"]) + return Overkiz.create_client( + Server.SOMFY_EUROPE, "foo", "pass", aiohttp.ClientSession() + ) @pytest.mark.asyncio async def test_get_devices_basic(self, client):