diff --git a/.gitignore b/.gitignore index 3ac3800..abb56a7 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,6 @@ **/.vscode **/.mypy_cache -**/config.yaml +**/config*.yaml **/.venv **/poetry.lock **/__pycache__ @@ -8,4 +8,5 @@ **/build **/*.egg-info -test2.py +test* +**/.wsl_venv diff --git a/.python-version b/.python-version index afad818..371cfe3 100644 --- a/.python-version +++ b/.python-version @@ -1 +1 @@ -3.11.0 +3.11.1 diff --git a/README.md b/README.md index a84316e..286970b 100644 --- a/README.md +++ b/README.md @@ -3,3 +3,29 @@ This is an async implementation of the v2 Philips Hue API. For a brief example of how to use, see [`here`](example.py) + +|[Events](phlyght/http.py#L53)|Desc| +|:--|:-:| +|on_ready|; +on_behavior_instance_update|; +on_behavior_script_update|; +on_bridge_update|; +on_bridge_home_update|; +on_button_update|; +on_device_power_update|; +on_entertainment_update|; +on_entertainment_configuration_update|; +on_geofence_client_update|; +on_geolocationa_update|; +on_light_update|; +on_grouped_light_update|; +on_light_level_update|; +on_motion_update|; +on_relative_rotary_update|; +on_room_update|; +on_scene_update|; +on_temperature_update|; +on_zigbee_connectivity_update|; +on_zigbee_device_discovery_update|; +on_zgb_connectivity_update|; +on_zone_update|; diff --git a/config_example.yaml b/config_example.yaml index ee81f4a..f7bb3f1 100644 --- a/config_example.yaml +++ b/config_example.yaml @@ -2,9 +2,22 @@ api_key: bridge_ip: 192.168.1.1 aliases: + light_levels: + 38b5b9f0-f754-4e70-8609-d40e52ef8f12: yd_parking_lumens + 9ea4f33d-0da1-4bc5-9b01-98396a3b7677: ktch_lumens + ef343aaf-0e77-4563-b8ec-bd0662990ceb: yd_fenced_lumens lights: - 00000000-0000-0000-0000-000000000000: entry - 11111111-1111-1111-1111-111111111111: bed - 22222222-2222-2222-2222-222222222222: kitchen - 33333333-3333-3333-3333-333333333333: bathroom - 44444444-4444-4444-4444-444444444444: footrest + 3c9d5ba7-ea43-45f8-a823-f3b32168776d: r_desk + 46588240-8205-4615-a2b5-27a8ecf9715f: p_strip + 930437b3-a0ef-4ade-be50-7cce032b9d0b: g_doors + 93a1533e-bfeb-4103-a150-180943a3ff3b: r_curtain + motions: + 3cf4bc43-3c73-43c9-ba03-d9b6af84785b: yd_fenced_motion + 59e03887-fd24-42eb-b7f0-f5b88eb20291: ktch_motion + rooms: + 3cecbcff-a654-48c4-a699-84882b087c17: porch + 448afef5-cccb-49d3-ab27-aa8e4c6e7dd4: toilet + temperatures: + 357431ce-7470-4722-8f48-4c45546dd53b: yd_parking_temp + 62762858-1c26-436e-a4fd-1f97382dc0a7: ktch_temp + 96d545e0-bf61-4d4d-9ada-f5e8fb7d75e5: yd_fenced_temp diff --git a/example.py b/example.py index 25f6ee0..f96e4d9 100644 --- a/example.py +++ b/example.py @@ -1,43 +1,90 @@ +from logging import basicConfig from asyncio import get_running_loop, sleep -from pathlib import Path from phlyght import HueEntsV2, Router, Attributes, _XY -from random import random +from random import random, randint +from rich import print +from loguru import logger + +basicConfig(filename="phlyght.log", filemode="a+", level=10, force=True) +logger.enable("INFO") +logger.enable("ERROR") +logger.enable("WARNING") +logger.enable("DEBUG") class HueRouter(Router): async def on_light_update(self, light: HueEntsV2.Light): + # print(light) return True async def on_button_update(self, button: HueEntsV2.Button): - if ( - button.id == self.ra_button.id - and button.button.last_event == "initial_press" - ): - for l in [ - self.curtain, - self.lamp, - self.desk, - self.lightbar_under, - self.lightbar_monitor, - ]: - l.color = Attributes.LightColor(xy=_XY(x=random(), y=random())) - if l.dimming.brightness < 30: - l.dimming = Attributes.Dimming(brightness=75.0) - else: - l.dimming = Attributes.Dimming(brightness=5.0) - await l.update() - return True + if button.id == self.r_button.id: + gradient = None + mirek = None + color = None + brightness = 100.0 + alternate_brightness = False + match button.button.last_event: + case "short_release": + color = Attributes.LightColor(xy=_XY(x=random(), y=random())) + mirek = randint(153, 500) + alternate_brightness = True + + case "long_release": + mirek = 500 + brightness = 100.0 + color = Attributes.LightColor(xy=_XY(x=0.99, y=0.99)) + gradient = Attributes.Gradient( + points=[ + Attributes.ColorPointColor( + color=Attributes.ColorPoint( + xy=Attributes.XY(x=0.99, y=0.88) + ) + ), + Attributes.ColorPointColor( + color=Attributes.ColorPoint( + xy=Attributes.XY(x=0.88, y=0.99) + ) + ), + ], + points_capable=2, + ) + case "repeat": + brightness = 50.0 + color = Attributes.LightColor(xy=_XY(x=random(), y=random())) + + for _light in [ + self.r_curtain, + self.r_lamp, + self.r_desk, + self.r_under, + self.r_monitor, + ]: + if color: + _light.color = color + + if brightness: + if alternate_brightness: + if _light.dimming.brightness < 30: + brightness = 75.0 + else: + brightness = 5.0 + + _light.dimming = Attributes.Dimming(brightness=brightness) + if mirek: + _light.color_temperature = Attributes.ColorTemp(mirek=mirek) + if gradient: + _light.gradient = gradient + await _light.update() async def on_grouped_light_update(self, grouped_light: HueEntsV2.GroupedLight): - return True + ... async def on_motion_update(self, motion: HueEntsV2.Motion): - return True + print(motion) - async def _shift( - self, light: HueEntsV2.Light - ): # this wont be ran unless the lines in on_ready are uncommented + async def _shift(self, light: HueEntsV2.Light): while get_running_loop().is_running(): # We can set the values by explicitly setting the attributes light.color = Attributes.LightColor(xy=_XY(x=random(), y=random())) @@ -45,7 +92,8 @@ class HueRouter(Router): await sleep(0.3) async def on_ready(self): # This will be called once, right after startup - await self.dump(Path("config.yaml")) + logger.info("Router is ready") + await self.dump("config.yaml") # don't use this if prone to seizures # # for l in [ # self.curtain, @@ -60,9 +108,5 @@ class HueRouter(Router): ... -router = HueRouter( - "TzPrxDf9hyW5oR5lvUaG2Zn4Hlbp2yFg7ue2ynzI", # Fill this in with [[YOUR API KEY]] otherwise it wont run - bridge_ip="https://192.168.1.1", # Your bridge IP here - max_cache_size=64, -) +router = HueRouter(max_cache_size=64) # , hue_api_key="", hue_bridge_ip="") router.run() diff --git a/phlyght/abc.py b/phlyght/abc.py index f25500b..e6cf57a 100644 --- a/phlyght/abc.py +++ b/phlyght/abc.py @@ -1,5 +1,6 @@ -from typing import Any +from typing import Any, Optional, Self from httpx import AsyncClient +from yaml import YAMLObject from .utils import ENDPOINT_METHOD @@ -58,9 +59,9 @@ class SubRouter(metaclass=RouterMeta): BASE_URI: str _api_path: str _client: AsyncClient - _bridge_ip: str + _bridge_host: str - def __new__(cls, hue_api_key: str): + def __new__(cls, **kwargs): if not hasattr(cls, "handlers"): cls.handlers: dict[str, type] = {} @@ -70,11 +71,11 @@ class SubRouter(metaclass=RouterMeta): return super().__new__(cls) - def __init__(self, hue_api_key: str): - self._hue_api_key = hue_api_key + def __init__(self, api_key: str, /): + self._api_key = api_key self._headers = { "User-Agent": "Python/HueClient", - "hue-application-key": self._hue_api_key, + "hue-application-key": self._api_key, } def __init_subclass__(cls, *_, **kwargs) -> None: @@ -84,3 +85,59 @@ class SubRouter(metaclass=RouterMeta): def __getattribute__(self, key) -> Any: return object.__getattribute__(self, key) + + +class YAMLConfig(YAMLObject, dict): + yaml_tag = "!YAMLConfig" + + def __setstate__(self, state): + for k, v in state.items(): + if isinstance(v, dict): + v = YAMLConfig(**v) + + setattr(self, k, v) + dict.__setitem__(self, k, v) + self.__dict__[k] = v + + return self + + def __setitem__(self, key, value): + if isinstance(value, dict): + value = YAMLConfig(**value) + + dict.__setitem__(self, key, value) + setattr(self, key, value) + self.__dict__[key] = value + + def __getattribute__(self, name): + return object.__getattribute__(self, name) + + def __getitem__(self, key): + return dict.__getitem__(self, key) + + def keys(self): + return dict.keys(self) + + def __iter__(self): + return dict.__iter__(self) + + def __get__(self, key, f=None) -> Self | Any | None: + if f: + return self + return getattr(self, key, None) + + def __init__(self, **data): + super().__init__() + for k, v in data.items(): + self.__dict__[k] = v + self.__setattr__(k, v) + dict.__setitem__(self, k, v) + + def items(self): + return dict.items(self) + + def __repr__(self): + return dict.__repr__(self) + + def __str__(self): + return dict.__str__(self) diff --git a/phlyght/http.py b/phlyght/http.py index 5db1181..6ae4c9d 100644 --- a/phlyght/http.py +++ b/phlyght/http.py @@ -1,21 +1,20 @@ from abc import abstractmethod from asyncio import gather, get_running_loop, new_event_loop, sleep import collections -from inspect import signature, Parameter -from io import BytesIO, StringIO -from os import PathLike +from io import StringIO from pathlib import Path -from typing import Any, Literal, Optional, TypeVar, Generic +from re import compile as re_compile +from typing import Any, Iterable, Literal, Optional, TypeVar, Generic from aiofiles import open as aio_open from httpx import AsyncClient, ConnectError, ConnectTimeout, _content from httpx._exceptions import ReadTimeout as HTTPxReadTimeout from httpcore._exceptions import ReadTimeout -from numpy import sort -from pydantic import BaseModel, Field +from pydantic import BaseConfig, BaseModel, Field +from rich import print from pydantic.generics import GenericModel -from yaml import Loader, load, Dumper, dump as yaml_dump +from yaml import Loader, load, dump as yaml_dump from .abc import SubRouter @@ -23,9 +22,7 @@ from .utils import ( IP_RE, LRU, MSG_RE_BYTES, - STR_FMT_RE, URL, - URL_TYPES, get_data_fields, get_url_args, ret_cls, @@ -35,13 +32,12 @@ from . import models from .models import HueEntsV2, Entity, UUID -try: - from ujson import dumps, loads, JSONDecodeError -except ImportError: - from json import dumps, loads, JSONDecodeError +from ujson import dumps, loads setattr(_content, "json_dumps", dumps) +UUID_CMP = re_compile(r"^[0-9a-f]{8}-(?:[0-9a-f]{4}-){3}[0-9a-f]{12}$") + try: from yarl import URL as UR except ImportError: @@ -113,9 +109,9 @@ def route(method, endpoint) -> Any: if _v := d.pop(k, None): url_args[k] = v(_v) - _match_bridge = IP_RE.search(self._bridge_ip) + _match_bridge = IP_RE.search(self._bridge_host) if not _match_bridge: - raise ValueError(f"Invalid bridge ip {self._bridge_ip}") + raise ValueError(f"Invalid bridge ip {self._bridge_host}") _url_base = f"https://{_match_bridge.group(1)}/" + f"{base_uri}".lstrip("/") if url_args: @@ -262,7 +258,7 @@ class HueAPIv2(SubRouter): @ret_cls(HueEntsV2.Device) @route("GET", "/resource/device") - async def get_devices(self, /): + async def get_devices(self, /) -> Iterable[HueEntsV2.Device]: ... @ret_cls(HueEntsV2.Device) @@ -324,9 +320,7 @@ class HueAPIv2(SubRouter): "GET", "/resource/entertainment_configuration/{entertainment_configuration_id}" ) async def get_entertainment_configuration( - self, - entertainment_configuration_id: UUID, - /, + self, entertainment_configuration_id: UUID, / ): ... @@ -345,9 +339,7 @@ class HueAPIv2(SubRouter): "/resource/entertainment_configuration/{entertainment_configuration_id}", ) async def delete_entertainment_configuration( - self, - entertainment_configuration_id: UUID, - /, + self, entertainment_configuration_id: UUID, / ): ... @@ -657,12 +649,12 @@ class HueAPIv2(SubRouter): async def set_zigbee_connectivity(self, zigbee_connectivity_id: UUID, /, **kwargs): ... - # @ret_cls(HueEntsV2.ZigbeeDeviceDiscovery) + @ret_cls(HueEntsV2.ZigbeeDeviceDiscovery) @route("GET", "/resource/zigbee_device_discovery") async def get_zigbee_device_discoveries(self, /): ... - # @ret_cls(HueEntsV2.ZigbeeDeviceDiscovery) + @ret_cls(HueEntsV2.ZigbeeDeviceDiscovery) @route("GET", " /resource/zigbee_device_discovery/{zigbee_device_discovery_id}") async def get_zigbee_device_discovery(self, zigbee_device_discovery_id: UUID, /): ... @@ -815,6 +807,11 @@ class HueAPIv2(SubRouter): class Router(HueAPIv2, HueEDK): class Aliases(BaseModel): + class Config(BaseConfig): + smart_union = True + use_enum_values = True + allow_population_by_field_name = True + behavior_instances: Optional[dict[str, HueEntsV2.BehaviorInstance]] = Field( default_factory=dict ) @@ -826,6 +823,9 @@ class Router(HueAPIv2, HueEDK): default_factory=dict ) buttons: Optional[dict[str, HueEntsV2.Button]] = Field(default_factory=dict) + device_powers: Optional[dict[str, HueEntsV2.DevicePower]] = Field( + default_factory=dict + ) entertainments: Optional[dict[str, HueEntsV2.Entertainment]] = Field( default_factory=dict ) @@ -854,44 +854,71 @@ class Router(HueAPIv2, HueEDK): zigbee_connectivities: Optional[ dict[str, HueEntsV2.ZigbeeConnectivity] ] = Field(default_factory=dict) + zigbee_device_discoveries: Optional[ + dict[str, HueEntsV2.ZigbeeDeviceDiscovery] + ] = Field(default_factory=dict) zgb_connectivities: Optional[dict[str, HueEntsV2.ZigbeeConnectivity]] = Field( default_factory=dict ) zones: Optional[dict[str, HueEntsV2.Zone]] = Field(default_factory=dict) + def items(self): + return ((k, getattr(self, k)) for k in self.__fields__.keys()) + + def keys(self): + return self.__fields__.keys() + + def __setitem__(self, key, value): + self.__setattr__(key, value) + + def __getitem__(self, key, default=None): + return self.__getattribute__(key) + def __json__(self): return dumps(dict(self), indent=4) - def __new__(cls, hue_api_key: str, bridge_ip: str = "", max_cache_size: int = 10): - cls = super().__new__(cls, hue_api_key) + # def get_entity(self, id: UUID | str): + + def __new__(cls, max_cache_size: int = 10, **kwargs): + cls = super().__new__(cls, **kwargs) return cls - def __init__( - self, - hue_api_key: Optional[str] = None, - bridge_ip: Optional[str] = None, - max_cache_size=10, - ): - with open("config.yaml", "r+") as f: - self.config = load(f, Loader=Loader) - if not hue_api_key and not self.config.api_key: - print( - "No API key provided, please fill out your hue API key in the config or Router.__init__" + def __init__(self, max_cache_size=10, **kwargs): + from .abc import YAMLConfig + + _config = Path("config_test.yaml") + if not _config.exists(): + _config.touch() + self.config = YAMLConfig( + api_key=kwargs.get("api_key", None), + bridge_host=kwargs.get("bridge_host", None), + aliases={}, ) - exit(1) - super().__init__(hue_api_key or self.config.api_key) + with _config.open("w+") as f: + yaml_dump(self.config, f, default_flow_style=False) + if not self.config.api_key: + print("Please fill out config.yaml with the api_key and bridge_host") + exit(1) + else: + with _config.open("r+") as f: + self.config = load(f, Loader=Loader) + super().__init__(kwargs.pop("api_key", None) or self.config.api_key or exit(1)) Entity.cache_client(self) self.cache = LRU(max_cache_size) self._client = AsyncClient(headers=self._headers, verify=False) self._subscription = None - self._bridge_ip = bridge_ip or self.config.bridge_ip + self._bridge_host = f"""https://{( + kwargs.pop("bridge_host", None) or self.config.bridge_host or exit(1) + )}""" self._tasks = [] + self._entities = self.Aliases() self.behavior_instances = {} self.behavior_scripts = {} self.bridges = {} self.bridge_homes = {} self.buttons = {} + self.device_powers = {} self.entertainments = {} self.entertainment_configurations = {} self.geofence_clients = {} @@ -904,6 +931,7 @@ class Router(HueAPIv2, HueEDK): self.scenes = {} self.temperatures = {} self.zigbee_connectivities = {} + self.zigbee_device_discoveries = {} self.zgb_connectivities = {} self.zones = {} @@ -940,8 +968,46 @@ class Router(HueAPIv2, HueEDK): alias = v.get(str(obj.id)) if alias: ob = obj.__class__(id=obj.id) + self._entities[k][alias] = ob getattr(self, k)[alias] = ob setattr(self, alias, ob) + else: + self._entities[k][obj.metadata.name] = await obj.get() + + cts = collections.Counter() + dn = {"devices"} + for device in await self.get_devices(): + for service in device.services: + pl = Entity.get_plural(service.rtype) + dn.add(pl) + if str(service.rid) in self.config["aliases"].get(pl, {}).keys(): + continue + cts[service.rtype] += 1 + dvc = await getattr(self, f"get_{service.rtype}")(service.rid) + if dvc: + self._entities[pl][ + f"{TYPE_CACHE[service.rtype].cfg_prefix}_{cts[service.rtype]}" + ] = dvc[0] + for k, v in self._entities.items(): + if k not in dn: + dn.add(k) + for itm in await getattr(self, f"get_{k}")(): + if str(itm.id) in self.config["aliases"].get(k, {}).keys(): + continue + if hasattr(itm, "metadata"): + nm = ( + ( + itm.metadata["name"] + if isinstance(itm.metadata, dict) + else itm.metadata.name + ) + .replace(" ", "_") + .replace("-", "_") + .lower() + ) + self._entities[k][nm] = itm + else: + self._entities[k][itm.id] = itm self.new_task(self._subscribe()) @@ -950,6 +1016,10 @@ class Router(HueAPIv2, HueEDK): except KeyboardInterrupt: await gather(*self._tasks) + async def dump_state(self): + async with aio_open("state.json", "w+") as f: + await f.write(dumps(self._entities, indent=4, sort_keys=True)) + def _parse_payload(self, payload: bytes): _match = MSG_RE_BYTES.search(payload) if not _match: @@ -962,14 +1032,10 @@ class Router(HueAPIv2, HueEDK): for _ent in _event["data"]: _event_id = _id.decode() _event_type = _event["type"] - _object = TYPE_CACHE[_ent["type"]](**_ent) - event = Event( - id=_event_id, - object=_object, - type=_event_type, - ) - - if hasattr(self, f"on_{event.object.type}_{event.type}"): + if hasattr(self, f"on_{_ent['type']}_{_event['type']}"): + _object = TYPE_CACHE[_ent["type"]](**_ent) + event = Event(id=_event_id, object=_object, type=_event_type) + # if hasattr(self, f"on_{event.object.type}_{event.type}"): _evs.append( _t := get_running_loop().create_task( getattr(self, f"on_{event.object.type}_{event.type}")( @@ -980,20 +1046,25 @@ class Router(HueAPIv2, HueEDK): self.cache.extend(*_evs) - async def dump(self, filename: Optional[Path | PathLike] = None): - devices = await self.get_devices() + async def dump(self, filename: Optional[Path | str] = None): aliases = {} - for device in devices: - for service in device.services: - aliases.setdefault(Entity.get_plural(service.rtype), {}) - aliases[Entity.get_plural(service.rtype)][ - str(service.rid) - ] = device.metadata.name + for key, sub_val in self._entities.items(): + defa = 0 + aliases.setdefault(key, {}) + aliases[key] = { + k + if not (aliased := UUID_CMP.match(str(k))) + else (v.cfg_prefix + str(defa := (defa + 1))): ( + str(v.id if not aliased else v.id) + ) + for k, v in sub_val.items() + } + cfg = { - "bridge_ip": self.config["bridge_ip"], - "api_key": self.config["api_key"], + "bridge_ip": self.config.get("bridge_ip", ""), + "api_key": self.config.get("api_key", ""), "aliases": { - k[0]: {v: vk for v, vk in sorted(k[1].items(), key=lambda i: i[1])} + k[0]: {vk: v for v, vk in sorted(k[1].items(), key=lambda i: i[1])} for k in sorted(aliases.items(), key=lambda k: k[0]) }, } @@ -1004,11 +1075,15 @@ class Router(HueAPIv2, HueEDK): file_path = filename else: file_path = Path("dump.yaml") - f = await aio_open(file_path, "w+") buf = StringIO() yaml_dump( - cfg, buf, default_flow_style=False, allow_unicode=True, sort_keys=False + cfg, + buf, + default_flow_style=False, + allow_unicode=True, + sort_keys=True, + canonical=False, ) await f.write(buf.getvalue()) await f.close() diff --git a/phlyght/models.py b/phlyght/models.py index 45cf52b..433c0ad 100644 --- a/phlyght/models.py +++ b/phlyght/models.py @@ -1,22 +1,10 @@ -from types import new_class -from typing import ( - Any, - Literal, - Optional, - Type, - ClassVar, - TypeVar, -) +from typing import Any, Literal, Optional, Type, ClassVar, TypeVar from uuid import UUID as _UUID, uuid4 -from typing import TypeVar, Generic from enum import Enum, auto -from httpx import AsyncClient from pydantic import BaseConfig, BaseModel, Field -from pydantic.main import ModelMetaclass, BaseModel -from pydantic.generics import GenericModel from pydantic.dataclasses import dataclass -from requests import delete +import ujson try: from ujson import dumps, loads @@ -53,20 +41,19 @@ def config_dumps(obj: Any) -> str: class Config(BaseConfig): - json_loads = loads - json_dumps = lambda *args, **kwargs: ( - d if isinstance(d := dumps(*args, **kwargs), str) else d.decode() - ) + json_loads = ujson.loads + # json_dumps = lambda *args, **kwargs: ( + # d if isinstance(d := dumps(*args, **kwargs), str) else d.decode() + # ) + json_dumps = ujson.dumps smart_union = True allow_mutations = True class HueConfig(BaseConfig): allow_population_by_field_name = True - json_loads = loads - json_dumps = lambda *args, **kwargs: ( - d if isinstance(d := dumps(*args, **kwargs), str) else d.decode() - ) + json_loads = ujson.loads + json_dumps = ujson.dumps smart_union = True allow_mutation = True @@ -140,6 +127,9 @@ class Entity(BaseModel): def __hash__(self) -> int: return hash(self.id) + def __json__(self): + return ujson.dumps(self) + class BaseAttribute(BaseModel): Config = HueConfig @@ -147,6 +137,7 @@ class BaseAttribute(BaseModel): def __init_subclass__(cls, *args, **kwargs): cls.Config = HueConfig + cls.__config__ = HueConfig class RoomType(Enum): @@ -196,7 +187,7 @@ class RoomType(Enum): OTHER = auto() def __json__(self): - return self.value + return f'"{self.value}"' class Archetype(Enum): @@ -362,8 +353,8 @@ class Attributes: level: str = "unknown" class Dimming(BaseAttribute): - brightness: Optional[float] = Field(default=100, gt=0, le=100) - min_dim_level: Optional[float] = Field(default=0, ge=0, le=100) + brightness: Optional[float] = Field(default=99.0, gt=0, le=100.0) + min_dim_level: Optional[float] = Field(default=1.0, ge=0, le=100.0) class DimmingDelta(BaseAttribute): action: Optional[Literal["up", "down", "stop"]] = "stop" @@ -428,11 +419,10 @@ class Attributes: ) class Motion(BaseAttribute): - motion: Optional[bool] = False - motion_valid: Optional[bool] = True + motion: Optional[bool] = Field(default=False) + motion_valid: Optional[bool] = Field(default=True) class On(BaseAttribute): - on: Optional[bool] = Field(default=True, alias="on") class Palette(BaseAttribute): @@ -447,7 +437,7 @@ class Attributes: default_factory=lambda: Attributes.ColorPoint() ) dimming: Optional["Attributes.Dimming"] = Field( - default_factory=lambda: Attributes.Dimming(brightness=100.0) + default_factory=lambda: Attributes.Dimming(brightness=99.0) ) class PaletteTemperature(BaseAttribute): @@ -591,6 +581,7 @@ class HueEntsV2: last_error: str = "none" metadata: dict[Literal["name"], str] = Field(default_factory=dict) migrated_from: str = "unknown" + cfg_prefix: ClassVar[str] = "bhv_inst_" class BehaviorScript(Entity): type: ClassVar[str] = "behavior_script" @@ -604,6 +595,7 @@ class HueEntsV2: metadata: dict[Any, Any] = Field(default_factory=dict) supported_features: list[str] = Field(default_factory=list) max_number_of_instances: int = Field(default=0, ge=0, le=255) + cfg_prefix: ClassVar[str] = "bhv_script_" class Bridge(Entity): type: ClassVar[str] = "bridge" @@ -611,6 +603,7 @@ class HueEntsV2: id_v1: str = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") bridge_id: str = "" time_zone: dict[str, str] = Field(default_factory=dict) + cfg_prefix: ClassVar[str] = "brdg_" class BridgeHome(Entity): type: ClassVar[str] = "bridge_home" @@ -618,6 +611,7 @@ class HueEntsV2: id_v1: str = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") services: list[Attributes.Identifier] = Field(default_factory=list) children: list[Attributes.Identifier] = Field(default_factory=list) + cfg_prefix: ClassVar[str] = "brdg_hm_" class Button(Entity): type: ClassVar[str] = "button" @@ -626,6 +620,7 @@ class HueEntsV2: owner: Attributes.Identifier = Field(default_factory=Attributes.Identifier) metadata: dict[Literal["control_id"], int] = Field(default_factory=dict) button: Attributes.Button = Field(default_factory=Attributes.Button) + cfg_prefix: ClassVar[str] = "btn_" class Device(Entity): type: ClassVar[str] = "device" @@ -636,6 +631,7 @@ class HueEntsV2: product_data: Attributes.ProductData = Field( default_factory=lambda: Attributes.ProductData() ) + cfg_prefix: ClassVar[str] = "dvc_" class DevicePower(Entity): type: ClassVar[str] = "device_power" @@ -645,6 +641,7 @@ class HueEntsV2: power_state: Attributes.PowerState = Field( default_factory=Attributes.PowerState ) + cfg_prefix: ClassVar[str] = "dvc_pwr_" class Entertainment(Entity): type: ClassVar[str] = "entertainment" @@ -657,6 +654,7 @@ class HueEntsV2: segments: Attributes.SegmentManager = Field( default_factory=Attributes.SegmentManager ) + cfg_prefix: ClassVar[str] = "ent_" class EntertainmentConfiguration(Entity): type: ClassVar[str] = "entertainment_configuration" @@ -679,18 +677,22 @@ class HueEntsV2: default_factory=Attributes.EntLocation ) light_services: list[Attributes.Identifier] = Field(default_factory=list) + cfg_prefix: ClassVar[str] = "ent_cfg_" class GeofenceClient(Entity): type: ClassVar[str] = "geofence_client" id: UUID id_v1: str = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") name: str = "" + is_at_home: Optional[bool] = True + cfg_prefix: ClassVar[str] = "geo_clnt_" class Geolocation(Entity): type: ClassVar[str] = "geolocation" id: UUID id_v1: str = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") is_configured: bool = False + cfg_prefix: ClassVar[str] = "geoloc_" class GroupedLight(Entity): type: ClassVar[str] = "grouped_light" @@ -698,15 +700,18 @@ class HueEntsV2: id_v1: str = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") on: Attributes.On = Field(default_factory=Attributes.On) alert: Attributes.Alert = Field(default_factory=Attributes.Alert) + cfg_prefix: ClassVar[str] = "grp_lt_" class Homekit(Entity): id: UUID type: ClassVar[str] = "resource" id_v1: str = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") status: Literal["paired", "pairing", "unpaired"] = "unpaired" + cfg_prefix: ClassVar[str] = "hm_kt_" class Light(Entity): - # id: UUID + type: ClassVar[str] = "light" + id: UUID id_v1: Optional[str] = Field( default="", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$", exclude=True ) @@ -734,7 +739,7 @@ class HueEntsV2: timed_effects: Attributes.TimedEffects = Field( default_factory=Attributes.TimedEffects ) - type: ClassVar[str] = "light" + cfg_prefix: ClassVar[str] = "l_" class LightLevel(Entity): type: ClassVar[str] = "light_level" @@ -745,6 +750,7 @@ class HueEntsV2: light: Attributes.LightLevelValue = Field( default_factory=Attributes.LightLevelValue ) + cfg_prefix: ClassVar[str] = "l_lvl_" class Motion(Entity): type: ClassVar[str] = "motion" @@ -753,6 +759,7 @@ class HueEntsV2: owner: Attributes.Identifier = Field(default_factory=Attributes.Identifier) enabled: bool = True motion: Attributes.Motion = Field(default_factory=Attributes.Motion) + cfg_prefix: ClassVar[str] = "mtn_" class RelativeRotary(Entity): id: UUID @@ -762,11 +769,13 @@ class HueEntsV2: relative_rotary: Attributes.RelativeRotary = Field( default_factory=Attributes.RelativeRotary ) + cfg_prefix: ClassVar[str] = "rel_rot_" class Resource(Entity): id: UUID type: ClassVar[str] = "device" id_v1: str = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") + cfg_prefix: ClassVar[str] = "res_" class Room(Entity): type: ClassVar[str] = "room" @@ -781,6 +790,7 @@ class HueEntsV2: default_factory=Attributes.Metadata ) children: Optional[list[Attributes.Identifier]] = Field(default_factory=list) + cfg_prefix: ClassVar[str] = "rm_" class Scene(Entity): id: UUID @@ -792,6 +802,7 @@ class HueEntsV2: speed: float = 0.0 auto_dynamic: bool = False type: ClassVar[str] = "scene" + cfg_prefix: ClassVar[str] = "scn_" class Temperature(Entity): type: ClassVar[str] = "temperature" @@ -800,6 +811,7 @@ class HueEntsV2: owner: Attributes.Identifier = Field(default_factory=Attributes.Identifier) enabled: bool = True temperature: Attributes.Temp = Field(default_factory=Attributes.Temp) + cfg_prefix: ClassVar[str] = "tmp_" class ZGPConnectivity(Entity): type: ClassVar[str] = "zgp_connectivity" @@ -815,12 +827,17 @@ class HueEntsV2: ] ] = "connected" source_id: str = "" + cfg_prefix: ClassVar[str] = "zgp_" class ZigbeeConnectivity(Entity): type: ClassVar[str] = "zigbee_connectivity" - id: UUID - id_v1: str = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") - owner: Attributes.Identifier = Field(default_factory=Attributes.Identifier) + id: Optional[UUID] + id_v1: Optional[str] = Field( + None, regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$" + ) + owner: Optional[Attributes.Identifier] = Field( + default_factory=Attributes.Identifier + ) status: Optional[ Literal[ "connected", @@ -829,7 +846,20 @@ class HueEntsV2: "unidirectional_incoming", ] ] = "connected" - mac_address: str = Field("", regex=r"^(?:[0-9a-fA-F]{2}(?:-|:)?){6}$") + mac_address: Optional[str] = Field("", regex=r"^(?:[0-9a-fA-F]{2}:?){6}") + cfg_prefix: ClassVar[str] = "zig_conn_" + + class ZigbeeDeviceDiscovery(Entity): + type: ClassVar[str] = "zigbee_device_discovery" + id: UUID + id_v1: Optional[str] = Field( + "", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$" + ) + owner: Optional[Attributes.Identifier] = Field( + default_factory=Attributes.Identifier + ) + status: Optional[Literal["active", "ready"]] = "ready" + cfg_prefix: ClassVar[str] = "zig_dev_" class Zone(Entity): type: ClassVar[str] = "zone" @@ -840,6 +870,7 @@ class HueEntsV2: ) metadata: Attributes.Metadata = Field(default_factory=Attributes.Metadata) children: list[Attributes.Identifier] = Field(default_factory=list) + cfg_prefix: ClassVar[str] = "zn_" for k, v in HueEntsV2.__dict__.items(): diff --git a/phlyght/utils.py b/phlyght/utils.py index 07c2ab2..0e2e8a8 100644 --- a/phlyght/utils.py +++ b/phlyght/utils.py @@ -40,7 +40,9 @@ __all__ = ( ENDPOINT_METHOD = re_compile(r"^(?=((?:get|set|create|delete)\w+))\1") STR_FMT_RE = re_compile(r"(?=(\{([^:]+)(?::([^}]+))?\}))\1") URL_TYPES = {"str": str, "int": int} -IP_RE = re_compile(r"(?=(?:(?<=[^0-9])|)((?:[0-9]{,3}\.){3}[0-9]{,3}))\1") +IP_RE = re_compile( + r"(?=(?:(?<=[^0-9])|^)((?:[a-z0-9]+\.)*[a-z0-9]+\.[a-z]{2,}(?:[0-9]{2,5})?|(?:[0-9]{,3}\.){3}[0-9]{,3}))\1" +) MSG_RE_BYTES = re_compile( rb"(?=((?P^: hi\n\n$)|^id:\s(?P[0-9]+:\d*?)\ndata:(?P[^$]+)\n\n))\1" @@ -114,14 +116,18 @@ def ret_cls(cls): ) kwargs.pop("base_uri", None) - ret = ret.get("data", []) + ret = ret.get("data", None) _rets = [] + if not ret: + return [] + if isinstance(ret, list): for r in ret: _rets.append(cls(**r)) else: return cls(**ret) + return _rets except JSONDecodeError: return [] diff --git a/pyproject.toml b/pyproject.toml index 3ce7551..2767975 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,7 +18,7 @@ include = ["phlyght"] [tool.poetry] name = "lights" -version = "1.0.0" +version = "1.0.1" description = "An async Python library for controlling Philips Hue lights." authors = ["Ra "] @@ -31,6 +31,8 @@ ujson = ">=5.6.0" rich = ">=12.6.0" aiofiles = ">=22.1.0" pyyaml = "^6.0" +loguru = "^0.6.0" +orjson = "^3.8.5" [tool.poetry.dev-dependencies] black = ">=22.10.0" @@ -39,6 +41,7 @@ black = ">=22.10.0" pycodestyle = "^2.10.0" pylint = "^2.15.7" mypy = "^0.991" +flake8 = "^6.0.0" [tool.poetry.group.linux.dependencies] uvloop = "^0.17.0" @@ -54,3 +57,11 @@ build-backend = "setuptools.build_meta" [tool.flake8] ignore = ["W503"] extras = ["E501", "E203"] + +[tool.pyright] +pythonVersion = "3.11.1" +pythonPlatform = "Linux" +include = [ "*.py" ] +ignore = ["reportGeneralTypeIssues"] +reportMissingImports = true +reportMissingTypeStubs = false