Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 71 additions & 0 deletions docs/getting-started.md
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,77 @@ Use a cloud server when you want to connect through the vendor’s public API. U
asyncio.run(main())
```

=== "Somfy (multi-account cloud)"

Use `Server.SOMFY` with `UsernamePasswordCredentials` when a single Somfy
account owns or is invited to **multiple sites (homes)** — the "multi
account sign-in" feature of the TaHoma app. Unlike the region-specific
`Server.SOMFY_EUROPE`/`SOMFY_AMERICA`/`SOMFY_OCEANIA` servers, `Server.SOMFY`
is region-agnostic: it discovers every site on the account and resolves the
correct regional endpoint for the one you select.

```python
import asyncio

from pyoverkiz.auth.credentials import UsernamePasswordCredentials
from pyoverkiz.client import OverkizClient
from pyoverkiz.enums import Server


async def main() -> None:
async with OverkizClient(
server=Server.SOMFY,
credentials=UsernamePasswordCredentials("you@example.com", "password"),
) as client:
await client.login() # auto-selects a sole site

# Accounts with more than one site must select one explicitly.
gateways = await client.discover_gateways()
if len(gateways) > 1:
client.select_gateway(gateways[0].gateway_id)

# Client is now scoped to the selected site and ready to use.
setup = await client.get_setup()
print(f"{len(setup.devices)} device(s)")

asyncio.run(main())
```

Each `GatewayCandidate` from `discover_gateways()` carries a human-readable
`label` (the site name) and `home_id`, so a multi-site UI can let the user
pick before calling `select_gateway`.

**Resume without a password.** After selecting a site, call
`client.to_credentials()` to snapshot the session as `SomfyTokenCredentials`
(a refresh token scoped to the selected site). Persist it and pass it back on
the next run to log in without the password grant, token exchange, or
discovery. The refresh token rotates, so supply an `on_token_refresh`
callback to re-persist it.

```python
import asyncio

from pyoverkiz.auth.credentials import SomfyTokenCredentials
from pyoverkiz.client import OverkizClient
from pyoverkiz.enums import Server


async def persist(refresh_token: str) -> None:
# Store the rotated refresh token for next time.
...


async def main(stored: SomfyTokenCredentials) -> None:
async with OverkizClient(server=Server.SOMFY, credentials=stored) as client:
await client.login() # no network round trips
setup = await client.get_setup()
print(f"{len(setup.devices)} device(s)")


# `stored` is what you persisted earlier via:
# stored = client.to_credentials(on_token_refresh=persist)
```

=== "Somfy (local)"

Local authentication requires a token generated via the official mobile app. For details on obtaining a token, refer to [Somfy TaHoma Developer Mode](https://github.com/Somfy-Developer/Somfy-TaHoma-Developer-Mode).
Expand Down
19 changes: 17 additions & 2 deletions pyoverkiz/auth/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,12 @@
from __future__ import annotations

import datetime
from collections.abc import Mapping
from collections.abc import Awaitable, Callable, Mapping
from dataclasses import dataclass, field
from typing import Any, Protocol, runtime_checkable
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable

if TYPE_CHECKING:
from pyoverkiz.auth.credentials import SomfyTokenCredentials


@dataclass(slots=True)
Expand Down Expand Up @@ -66,6 +69,7 @@ class GatewayCandidate:
home_id: str | None = None
label: str | None = None
external_id: str | None = None
country: str | None = None


@runtime_checkable
Expand All @@ -81,3 +85,14 @@ def select_gateway(self, gateway_id: str) -> None:
@property
def selected_gateway(self) -> str | None:
"""Return the currently selected gateway id, or None."""


@runtime_checkable
class SupportsSessionResume(Protocol):
"""Optional capability: snapshot the session for later resume without re-login."""

def to_credentials(
self,
on_token_refresh: Callable[[str], Awaitable[None]] | None = None,
) -> SomfyTokenCredentials:
"""Return resume credentials for the current session."""
90 changes: 90 additions & 0 deletions pyoverkiz/auth/bob.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
"""Models for the Somfy BOB back-office site directory.
A separate service from the Overkiz enduser API, with its own payload shapes
and casing, so it gets its own small cattrs converter rather than sharing
``pyoverkiz.converter``.
"""

from __future__ import annotations

from dataclasses import dataclass, field

import cattrs
from cattrs.gen import make_dict_structure_fn, override

from pyoverkiz.auth.base import GatewayCandidate


@dataclass(slots=True)
class BobGateway:
"""A gateway entry under a sub-site."""

gateway_id: str


@dataclass(slots=True)
class BobSubSite:
"""A sub-site (setup) grouping one or more gateways."""

external_id: str | None = None
gateways: list[BobGateway] = field(default_factory=list)


@dataclass(slots=True)
class BobSite:
"""A site (home) the account owns or was invited to."""

site_oid: str
name: str | None = None
country: str | None = None
sub_sites: list[BobSubSite] = field(default_factory=list)


@dataclass(slots=True)
class BobSitesResponse:
"""The ``/sites`` listing, flattened on demand to gateway candidates."""

results: list[BobSite] = field(default_factory=list)

def gateway_candidates(self) -> list[GatewayCandidate]:
"""Flatten the site -> sub-site -> gateway tree into candidates."""
return [
GatewayCandidate(
gateway_id=gateway.gateway_id,
home_id=site.site_oid,
label=site.name,
external_id=sub.external_id,
country=site.country,
)
for site in self.results
for sub in site.sub_sites
for gateway in sub.gateways
]


def _make_bob_converter() -> cattrs.Converter:
# Converter (not GenConverter) so unknown BOB keys are dropped for forward-compat.
c = cattrs.Converter()
c.register_structure_hook(
BobGateway,
make_dict_structure_fn(BobGateway, c, gateway_id=override(rename="gatewayId")),
)
c.register_structure_hook(
BobSubSite,
make_dict_structure_fn(
BobSubSite, c, external_id=override(rename="externalOID")
),
)
c.register_structure_hook(
BobSite,
make_dict_structure_fn(
BobSite,
c,
site_oid=override(rename="siteOID"),
sub_sites=override(rename="subSites"),
),
)
return c


bob_converter = _make_bob_converter()
16 changes: 16 additions & 0 deletions pyoverkiz/auth/credentials.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,22 @@ class LocalTokenCredentials(TokenCredentials):
"""Credentials using a local API token."""


@dataclass(slots=True)
class SomfyTokenCredentials(Credentials):
"""Resume credentials for a previously-selected Somfy site (skips login + discovery).
Persist the ``refresh_token`` plus the site's ``site_oid`` and ``region``.
The refresh token rotates, so supply ``on_token_refresh`` to re-persist it;
otherwise a later reload fails.
"""

refresh_token: str = field(repr=False)
site_oid: str
region: str
gateway_id: str | None = None
on_token_refresh: Callable[[str], Awaitable[None]] | None = None


@dataclass(slots=True)
class RexelOAuthCodeCredentials(Credentials):
"""Credentials using Rexel OAuth2 authorization code with PKCE."""
Expand Down
14 changes: 14 additions & 0 deletions pyoverkiz/auth/factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
LocalTokenCredentials,
RexelOAuthCodeCredentials,
RexelTokenCredentials,
SomfyTokenCredentials,
TokenCredentials,
UsernamePasswordCredentials,
)
Expand All @@ -24,6 +25,7 @@
RexelAuthStrategy,
RexelTokenAuthStrategy,
SessionLoginStrategy,
SomfyAccountAuthStrategy,
SomfyAuthStrategy,
)
from pyoverkiz.enums import APIType, Server
Expand Down Expand Up @@ -62,6 +64,18 @@ def build_auth_strategy(
ssl_context,
)

if server == Server.SOMFY:
# Resume from a persisted site-scoped refresh token, or fresh login
# from username/password.
if not isinstance(credentials, SomfyTokenCredentials):
credentials = _ensure_credentials(credentials, UsernamePasswordCredentials)
return SomfyAccountAuthStrategy(
credentials,
session,
server_config,
ssl_context,
)

if server in {
Server.ATLANTIC_COZYTOUCH,
Server.THERMOR_COZYTOUCH,
Expand Down
Loading
Loading