Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add /setup endpoint #223

Merged
merged 17 commits into from
Dec 7, 2021
3 changes: 3 additions & 0 deletions .pylintrc
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,9 @@ good-names=id,tb
[LOGGING]
logging-format-style=new

[FORMAT]
max-line-length=140
tetienne marked this conversation as resolved.
Show resolved Hide resolved

[MESSAGES CONTROL]
disable=
logging-fstring-interpolation,
Expand Down
53 changes: 52 additions & 1 deletion pyhoma/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
OverkizServer,
Place,
Scenario,
Setup,
State,
)

Expand All @@ -61,6 +62,9 @@ async def refresh_listener(invocation: dict[str, Any]) -> None:
await invocation["args"][0].register_event_listener()


# pylint: disable=too-many-instance-attributes


class TahomaClient:
"""Interface class for the Overkiz API"""

Expand All @@ -84,6 +88,7 @@ def __init__(
self.password = password
self.server = server

self.setup: Setup | None = None
tetienne marked this conversation as resolved.
Show resolved Hide resolved
self.devices: list[Device] = []
self.gateways: list[Gateway] = []
tetienne marked this conversation as resolved.
Show resolved Hide resolved
self.event_listener_id: str | None = None
Expand Down Expand Up @@ -226,6 +231,42 @@ async def nexity_login(self) -> str:

return token["token"]

@backoff.on_exception(
backoff.expo,
(NotAuthenticatedException, ServerDisconnectedError),
max_tries=2,
on_backoff=relogin,
)
async def get_setup(self, refresh: bool = False) -> Setup:
"""
Get all data about the connected user setup
-> gateways data (serial number, activation state, ...): <gateways/gateway>
-> setup location: <location>
-> house places (rooms and floors): <place>
-> setup devices: <devices>

A gateway may be in different modes (mode) regarding to the activated functions (functions).
A house may be composed of several floors and rooms. The house, floors and rooms are viewed as a place.
Devices in the house are grouped by type called uiClass. Each device has an associated widget.
The widget is used to control or to know the device state, whatever the device protocol (controllable): IO, RTS, X10, ... .
A device can be either an actuator (type=1) or a sensor (type=2).
Data of one or several devices can be also get by setting the device(s) url as request parameter.

Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load)
"""
if self.setup and not refresh:
return self.setup

response = await self.__get("setup")
setup = Setup(**humps.decamelize(response))

# Cache response
self.setup = setup
self.gateways = setup.gateways
self.devices = setup.devices
iMicknl marked this conversation as resolved.
Show resolved Hide resolved

return setup

@backoff.on_exception(
backoff.expo,
(NotAuthenticatedException, ServerDisconnectedError),
Expand All @@ -235,13 +276,18 @@ async def nexity_login(self) -> str:
async def get_devices(self, refresh: bool = False) -> list[Device]:
"""
List devices
Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load)
"""
if self.devices and not refresh:
return self.devices

response = await self.__get("setup/devices")
devices = [Device(**d) for d in humps.decamelize(response)]

# Cache response
self.devices = devices
if self.setup:
self.setup.devices = devices
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So basically here, you don’t cache the devices, but sync with the /setup entry point?

Copy link
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What do you mean? What we do here is overwrite the /setup response with the latest devices response.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK it’s what I can sync :D

By the way, do we use this feature (cache) somewhere?

Copy link
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@tetienne not really I think, thus I was also thinking to remove it... However, it doesn't have a lot of extra code and can help other people to make sure they don't call expensive endpoints all the time. :-)


return devices

Expand All @@ -253,14 +299,19 @@ async def get_devices(self, refresh: bool = False) -> list[Device]:
)
async def get_gateways(self, refresh: bool = False) -> list[Gateway]:
"""
List gateways
Get every gateways of a connected user setup
Per-session rate-limit : 1 calls per 1d period for this particular operation (bulk-load)
"""
if self.gateways and not refresh:
return self.gateways

response = await self.__get("setup/gateways")
gateways = [Gateway(**g) for g in humps.decamelize(response)]

# Cache response
self.gateways = gateways
if self.setup:
self.setup.gateways = gateways

return gateways

iMicknl marked this conversation as resolved.
Show resolved Hide resolved
Expand Down
159 changes: 157 additions & 2 deletions pyhoma/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,13 +22,124 @@


def obfuscate_id(id: str | None) -> str:
return re.sub(r"\d+-", "****-", str(id))
return re.sub(r"(SETUP)?\d+-", "****-", str(id))


def obfuscate_email(email: str | None) -> str:
return re.sub(r"(.).*@.*(.\..*)", r"\1****@****\2", str(email))


def mask(input: str) -> str:
return re.sub(r"[a-zA-Z0-9_.-]*", "*", str(input))
iMicknl marked this conversation as resolved.
Show resolved Hide resolved


@attr.s(auto_attribs=True, init=False, slots=True, kw_only=True)
class Setup:
creation_time: str
last_update_time: str
id: str = attr.ib(repr=obfuscate_id, default=None)
location: Location
gateways: list[Gateway]
devices: list[Device]
zones: list[Zone]
reseller_delegation_type: str
oid: str
root_place: Place
features: list[Feature] | None = None

def __init__(
self,
*,
creation_time: str,
last_update_time: str,
id: str = attr.ib(repr=obfuscate_id, default=None),
location: dict[str, Any],
gateways: list[dict[str, Any]],
devices: list[dict[str, Any]],
zones: list[dict[str, Any]],
reseller_delegation_type: str,
oid: str,
root_place: dict[str, Any],
features: list[dict[str, Any]] | None = None,
**_: Any,
) -> None:
self.id = id
self.creation_time = creation_time
self.last_update_time = last_update_time
self.location = Location(**location)
self.gateways = [Gateway(**g) for g in gateways]
self.devices = [Device(**d) for d in devices]
self.zones = [Zone(**z) for z in zones]
self.reseller_delegation_type = reseller_delegation_type
self.oid = oid
self.root_place = Place(**root_place)
self.features = [Feature(**f) for f in features] if features else None


@attr.s(auto_attribs=True, init=False, slots=True, kw_only=True)
class Location:
creation_time: str
last_update_time: str
city: str = attr.ib(repr=mask, default=None)
country: str = attr.ib(repr=mask, default=None)
postal_code: str = attr.ib(repr=mask, default=None)
address_line1: str = attr.ib(repr=mask, default=None)
address_line2: str = attr.ib(repr=mask, default=None)
timezone: str
longitude: str = attr.ib(repr=mask, default=None)
latitude: str = attr.ib(repr=mask, default=None)
twilight_mode: int
twilight_angle: str
twilight_city: str
summer_solstice_dusk_minutes: str
winter_solstice_dusk_minutes: str
twilight_offset_enabled: bool
dawn_offset: int
dusk_offset: int

def __init__(
self,
*,
creation_time: str,
last_update_time: str,
city: str = attr.ib(repr=mask, default=None),
country: str = attr.ib(repr=mask, default=None),
postal_code: str = attr.ib(repr=mask, default=None),
address_line1: str = attr.ib(repr=mask, default=None),
address_line2: str = attr.ib(repr=mask, default=None),
timezone: str,
longitude: str = attr.ib(repr=mask, default=None),
latitude: str = attr.ib(repr=mask, default=None),
twilight_mode: int,
twilight_angle: str,
twilight_city: str,
summer_solstice_dusk_minutes: str,
winter_solstice_dusk_minutes: str,
twilight_offset_enabled: bool,
dawn_offset: int,
dusk_offset: int,
**_: Any,
) -> None:
self.creation_time = creation_time
self.last_update_time = last_update_time
self.city = city
self.country = country
self.postal_code = postal_code
self.address_line1 = address_line1
self.address_line2 = address_line2
self.timezone = timezone
self.longitude = longitude
self.latitude = latitude
self.twilight_mode = twilight_mode
self.twilight_angle = twilight_angle
self.twilight_city = twilight_city
self.summer_solstice_dusk_minutes = summer_solstice_dusk_minutes
self.winter_solstice_dusk_minutes = winter_solstice_dusk_minutes
self.twilight_offset_enabled = twilight_offset_enabled
self.dawn_offset = dawn_offset
self.dusk_offset = dusk_offset


@attr.s(auto_attribs=True, init=False, slots=True, kw_only=True)
class Device:
id: str = attr.ib(repr=False)
Expand Down Expand Up @@ -205,7 +316,6 @@ def __init__(self, name: str, parameters: list[str] | None = None, **_: Any):
dict.__init__(self, name=name, parameters=parameters)


# pylint: disable-msg=too-many-locals
@attr.s(auto_attribs=True, init=False, slots=True, kw_only=True)
class Event:
timestamp: int
Expand Down Expand Up @@ -509,6 +619,51 @@ def __init__(
self.sub_places = [Place(**p) for p in sub_places] if sub_places else []


@attr.s(auto_attribs=True, slots=True, kw_only=True)
class Feature:
name: str
source: str


@attr.s(auto_attribs=True, slots=True, kw_only=True)
class ZoneItem:
iMicknl marked this conversation as resolved.
Show resolved Hide resolved
item_type: str
device_oid: str
device_url: str


@attr.s(auto_attribs=True, init=False, slots=True, kw_only=True)
class Zone:
creation_time: str
last_update_time: str
label: str
type: int
items: list[ZoneItem] | None
external_oid: str
metadata: str
oid: str

def __init__(
self,
*,
last_update_time: str,
label: str,
type: int,
items: list[dict[str, Any]] | None,
external_oid: str,
metadata: str,
oid: str,
**_: Any,
) -> None:
self.last_update_time = last_update_time
self.label = label
self.type = type
self.items = [ZoneItem(**z) for z in items] if items else []
self.external_oid = external_oid
self.metadata = metadata
self.oid = oid


@attr.s(auto_attribs=True, slots=True, kw_only=True)
class OverkizServer:
"""Class to describe an Overkiz server."""
Expand Down
Loading