update example

This commit is contained in:
ra 2023-02-05 22:18:22 -08:00
parent b02d538560
commit a000316fae
10 changed files with 408 additions and 144 deletions

5
.gitignore vendored
View File

@ -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

View File

@ -1 +1 @@
3.11.0
3.11.1

View File

@ -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|;

View File

@ -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

View File

@ -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()

View File

@ -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)

View File

@ -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()

View File

@ -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():

View File

@ -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<hello>^: hi\n\n$)|^id:\s(?P<id>[0-9]+:\d*?)\ndata:(?P<data>[^$]+)\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 []

View File

@ -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 <ra@tcp.direct>"]
@ -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