easier object management, dump entities, events
This commit is contained in:
parent
9e23aef973
commit
b02d538560
|
|
@ -1,3 +1,4 @@
|
|||
# running Router.dump() on initial run will fill this all out
|
||||
api_key:
|
||||
bridge_ip: 192.168.1.1
|
||||
aliases:
|
||||
|
|
|
|||
76
example.py
76
example.py
|
|
@ -1,54 +1,68 @@
|
|||
from asyncio import get_running_loop, sleep
|
||||
|
||||
from pathlib import Path
|
||||
from phlyght import HueEntsV2, Router, Attributes, _XY
|
||||
from rich import print
|
||||
from random import random
|
||||
|
||||
try:
|
||||
from uvloop import install
|
||||
|
||||
install()
|
||||
except ImportError:
|
||||
...
|
||||
|
||||
|
||||
class HueRouter(Router):
|
||||
async def on_light_update(self, light: HueEntsV2.Light):
|
||||
print(f"A light was updated: {light}")
|
||||
return True
|
||||
|
||||
async def on_button_update(self, button: HueEntsV2.Button):
|
||||
print(f"A button was pressed: {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
|
||||
|
||||
async def _shift(self, light: HueEntsV2.Light):
|
||||
async def on_grouped_light_update(self, grouped_light: HueEntsV2.GroupedLight):
|
||||
return True
|
||||
|
||||
async def on_motion_update(self, motion: HueEntsV2.Motion):
|
||||
return True
|
||||
|
||||
async def _shift(
|
||||
self, light: HueEntsV2.Light
|
||||
): # this wont be ran unless the lines in on_ready are uncommented
|
||||
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()))
|
||||
light.dimming = Attributes.Dimming(brightness=100.0, min_dim_level=0.2)
|
||||
# We can modify the attributes of the lights and send the light object as the parameter to set_light()
|
||||
await self.set_light(
|
||||
light.id,
|
||||
light,
|
||||
)
|
||||
await light.update()
|
||||
await sleep(0.3)
|
||||
# Could potentially get more in / at a faster update rate but its getting pretty close to making the bridge unresponsive at this rate
|
||||
|
||||
async def on_ready(self): # This will be called once, right after startup
|
||||
for light in [
|
||||
self.entry,
|
||||
self.bed,
|
||||
self.kitchen,
|
||||
self.bathroom,
|
||||
self.footrest,
|
||||
]:
|
||||
# These are all aliases defined in the config, accessible as an attribute using the name on the router
|
||||
self.new_task(self._shift(light))
|
||||
await sleep(0.5)
|
||||
await self.dump(Path("config.yaml"))
|
||||
# don't use this if prone to seizures
|
||||
# # for l in [
|
||||
# self.curtain,
|
||||
# self.lamp,
|
||||
# self.desk,
|
||||
# self.lightbar_under,
|
||||
# self.lightbar_monitor,
|
||||
# ]:
|
||||
# # These are all aliases defined in the config, accessible as an attribute using the name on the router
|
||||
# self.new_task(self._shift(l))
|
||||
# await sleep(0.5)
|
||||
...
|
||||
|
||||
|
||||
router = HueRouter(
|
||||
"Your API Key",
|
||||
bridge_ip="https://192.168.1.1",
|
||||
"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.run()
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
from .http import Router
|
||||
from .models import Archetype, HueEntsV2, Attributes, RoomType, Entity, HueEntsV1, _XY
|
||||
from .abc import RouterMeta, SubRouter
|
||||
|
||||
__all__ = (
|
||||
"Router",
|
||||
|
|
@ -10,4 +11,6 @@ __all__ = (
|
|||
"HueEntsV2",
|
||||
"HueEntsV1",
|
||||
"_XY",
|
||||
"RouterMeta",
|
||||
"SubRouter",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,86 @@
|
|||
from typing import Any
|
||||
from httpx import AsyncClient
|
||||
|
||||
from .utils import ENDPOINT_METHOD
|
||||
|
||||
|
||||
class RouterMeta(type):
|
||||
@classmethod
|
||||
def __prepare__(cls, _, bases, **kwargs):
|
||||
if bases:
|
||||
return kwargs | bases[0].__dict__
|
||||
return kwargs
|
||||
|
||||
def __new__(cls, _, bases, kwds, **kwargs):
|
||||
cells = {}
|
||||
_base = kwds.get("BASE_URI", "")
|
||||
|
||||
def set_key(v):
|
||||
def wrap(self, *args, **_kwds):
|
||||
return v(self, *args, base_uri=_base, **_kwds)
|
||||
|
||||
return wrap
|
||||
|
||||
if any(
|
||||
map(
|
||||
lambda x: ENDPOINT_METHOD.match(x) is not None,
|
||||
kwds.keys(),
|
||||
)
|
||||
):
|
||||
funcs = list(
|
||||
filter(
|
||||
lambda k: (
|
||||
ENDPOINT_METHOD.match(k[0]) is not None and callable(k[1])
|
||||
),
|
||||
kwds.items(),
|
||||
)
|
||||
)
|
||||
for k, v in funcs:
|
||||
if hasattr(v, "__closure__"):
|
||||
# val = v.__closure__[0].cell_contents
|
||||
cells[k] = set_key(v)
|
||||
|
||||
kwds["handlers"] = cells
|
||||
|
||||
for base in bases:
|
||||
if hasattr(base, "handlers"):
|
||||
kwds["handlers"] |= base.handlers
|
||||
|
||||
kwds.update(**kwds["handlers"])
|
||||
|
||||
return super().__new__(cls, cls.__name__, bases, kwds, **kwargs)
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
|
||||
class SubRouter(metaclass=RouterMeta):
|
||||
BASE_URI: str
|
||||
_api_path: str
|
||||
_client: AsyncClient
|
||||
_bridge_ip: str
|
||||
|
||||
def __new__(cls, hue_api_key: str):
|
||||
if not hasattr(cls, "handlers"):
|
||||
cls.handlers: dict[str, type] = {}
|
||||
|
||||
for base in cls.__bases__:
|
||||
if hasattr(base, "handlers"):
|
||||
cls.handlers |= getattr(base, "handlers")
|
||||
|
||||
return super().__new__(cls)
|
||||
|
||||
def __init__(self, hue_api_key: str):
|
||||
self._hue_api_key = hue_api_key
|
||||
self._headers = {
|
||||
"User-Agent": "Python/HueClient",
|
||||
"hue-application-key": self._hue_api_key,
|
||||
}
|
||||
|
||||
def __init_subclass__(cls, *_, **kwargs) -> None:
|
||||
super().__init_subclass__()
|
||||
if kwargs.get("root"):
|
||||
cls._api_path = f'{kwargs.get("root")}{cls.BASE_URI}'
|
||||
|
||||
def __getattribute__(self, key) -> Any:
|
||||
return object.__getattribute__(self, key)
|
||||
1148
phlyght/http.py
1148
phlyght/http.py
File diff suppressed because it is too large
Load Diff
|
|
@ -1,3 +1,4 @@
|
|||
from types import new_class
|
||||
from typing import (
|
||||
Any,
|
||||
Literal,
|
||||
|
|
@ -7,12 +8,15 @@ from typing import (
|
|||
TypeVar,
|
||||
)
|
||||
from uuid import UUID as _UUID, uuid4
|
||||
from typing import TypeVar, Generic
|
||||
from enum import Enum, auto
|
||||
|
||||
# from dataclasses import dataclass
|
||||
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
|
||||
|
||||
try:
|
||||
from ujson import dumps, loads
|
||||
|
|
@ -23,6 +27,8 @@ except ImportError:
|
|||
from json import dumps, loads
|
||||
|
||||
_type = type
|
||||
_T = TypeVar("_T")
|
||||
_D = TypeVar("_D", bound=dict)
|
||||
|
||||
__all__ = (
|
||||
"Entity",
|
||||
|
|
@ -38,16 +44,31 @@ __all__ = (
|
|||
Ent = TypeVar("Ent", bound="Entity")
|
||||
|
||||
|
||||
def default_uuid():
|
||||
return UUID(str(uuid4()))
|
||||
|
||||
|
||||
def config_dumps(obj: Any) -> str:
|
||||
return ret if isinstance(ret := dumps(obj), str) else ret.decode()
|
||||
|
||||
|
||||
class Config(BaseConfig):
|
||||
json_loads = loads
|
||||
json_dumps = lambda *args, **kwargs: (
|
||||
d if isinstance(d := dumps(*args, **kwargs), str) else d.decode()
|
||||
)
|
||||
smart_union = True
|
||||
allow_mutations = True
|
||||
|
||||
|
||||
class HueConfig(BaseConfig):
|
||||
underscore_attrs_are_private = True
|
||||
allow_population_by_field_name = True
|
||||
json_loads = loads
|
||||
json_dumps = config_dumps
|
||||
json_dumps = lambda *args, **kwargs: (
|
||||
d if isinstance(d := dumps(*args, **kwargs), str) else d.decode()
|
||||
)
|
||||
smart_union = True
|
||||
allow_mutation = True
|
||||
|
||||
|
||||
class UUID(_UUID):
|
||||
|
|
@ -55,6 +76,10 @@ class UUID(_UUID):
|
|||
return self.__str__()
|
||||
|
||||
|
||||
def validate(*args, **kwargs):
|
||||
return kwargs
|
||||
|
||||
|
||||
class Entity(BaseModel):
|
||||
__module__ = "phlyght"
|
||||
__cache__: ClassVar[dict[str, Type]] = {}
|
||||
|
|
@ -62,20 +87,49 @@ class Entity(BaseModel):
|
|||
default_factory=lambda: UUID("00000000-0000-0000-0000-000000000000")
|
||||
)
|
||||
type: ClassVar[str] = "unknown"
|
||||
Config = HueConfig
|
||||
__config__ = HueConfig
|
||||
|
||||
class Config:
|
||||
__root__: "Entity"
|
||||
json_loads = loads
|
||||
json_dumps = dumps
|
||||
smart_union = True
|
||||
@classmethod
|
||||
def cache_client(cls, client):
|
||||
Entity.client = client
|
||||
|
||||
@classmethod
|
||||
def get_entities(cls) -> dict[str, Type]:
|
||||
return cls.__cache__
|
||||
|
||||
@classmethod
|
||||
def __prepare__(cls, name, bases, **kwds):
|
||||
return super().__prepare__(name, bases, **kwds)
|
||||
def get_plural(cls, name):
|
||||
if name.endswith("y"):
|
||||
return name[:-1] + "ies"
|
||||
return name + "s"
|
||||
|
||||
def __new__(cls, client=None, **kwargs):
|
||||
clz = type(cls.__name__, (BaseModel,), {}).__new__(cls)
|
||||
return clz
|
||||
|
||||
async def get(self):
|
||||
return await getattr(self.client, f"get_{self.type}")(self.id)
|
||||
|
||||
async def create(self, **kwargs):
|
||||
if not hasattr(self.client, f"create_{self.type}"):
|
||||
return
|
||||
|
||||
for k, v in kwargs.items():
|
||||
if hasattr(self, k) and getattr(self, k) != v:
|
||||
setattr(self, k, v)
|
||||
|
||||
await getattr(self.client, f"create_{self.type}")(self)
|
||||
|
||||
async def update(self, **kwargs):
|
||||
for k, v in kwargs.items():
|
||||
if hasattr(self, k) and getattr(self, k) != v:
|
||||
setattr(self, k, v)
|
||||
await getattr(self.client, f"set_{self.type}")(self.id, self)
|
||||
|
||||
async def delete(self):
|
||||
if _fn := getattr(self.client, f"delete_{self.type}", None):
|
||||
await _fn(self.id, self)
|
||||
|
||||
def __init_subclass__(cls, **_):
|
||||
super().__init_subclass__()
|
||||
|
|
@ -87,6 +141,14 @@ class Entity(BaseModel):
|
|||
return hash(self.id)
|
||||
|
||||
|
||||
class BaseAttribute(BaseModel):
|
||||
Config = HueConfig
|
||||
__config__ = HueConfig
|
||||
|
||||
def __init_subclass__(cls, *args, **kwargs):
|
||||
cls.Config = HueConfig
|
||||
|
||||
|
||||
class RoomType(Enum):
|
||||
@staticmethod
|
||||
def _generate_next_value_(name, *_):
|
||||
|
|
@ -194,8 +256,20 @@ class _XY:
|
|||
x: float
|
||||
y: float
|
||||
|
||||
def __post_init__(self):
|
||||
if self.x < 0.01:
|
||||
self.x = 0.01
|
||||
if self.y < 0.01:
|
||||
self.y = 0.01
|
||||
if self.x > 0.99:
|
||||
self.x = 0.99
|
||||
if self.y > 0.99:
|
||||
self.y = 0.99
|
||||
|
||||
def __json__(self):
|
||||
return '{"y":' + f'{self.x}, "x": {self.y}' + "}"
|
||||
return (
|
||||
'{"y":' + f'{int(10000*self.x)/10000}, "x": {int(10000*self.y)/10000}' + "}"
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
@ -210,7 +284,7 @@ XY = _XY | tuple[float, float]
|
|||
|
||||
|
||||
class Attributes:
|
||||
class Action(BaseModel):
|
||||
class Action(BaseAttribute):
|
||||
on: "Attributes.On" = Field(default_factory=lambda: Attributes.On(on=True))
|
||||
dimming: "Attributes.Dimming" = Field(
|
||||
default_factory=lambda: Attributes.Dimming(
|
||||
|
|
@ -233,7 +307,7 @@ class Attributes:
|
|||
default_factory=lambda: Attributes.SceneDynamics(duration=1)
|
||||
)
|
||||
|
||||
class Actions(BaseModel):
|
||||
class Actions(BaseAttribute):
|
||||
target: "Attributes.Identifier" = Field(...)
|
||||
action: "Attributes.Action" = Field(default_factory=lambda: Attributes.Action())
|
||||
dimming: "Attributes.Dimming" = Field(
|
||||
|
|
@ -245,10 +319,10 @@ class Attributes:
|
|||
default_factory=lambda: Attributes.ColorPoint()
|
||||
)
|
||||
|
||||
class Alert(BaseModel):
|
||||
class Alert(BaseAttribute):
|
||||
action: Literal["breathe", "unknown"] = "unknown"
|
||||
|
||||
class Button(BaseModel):
|
||||
class Button(BaseAttribute):
|
||||
last_event: Literal[
|
||||
"initial_press",
|
||||
"repeat",
|
||||
|
|
@ -258,107 +332,94 @@ class Attributes:
|
|||
"long_press",
|
||||
] = "initial_press"
|
||||
|
||||
class Color(BaseModel):
|
||||
class Color(BaseAttribute):
|
||||
xy: "Attributes.XY" = Field(default_factory=lambda: Attributes.XY(x=0.0, y=0.0))
|
||||
gamut: "Attributes.Gamut" = Field(default_factory=lambda: Attributes.Gamut())
|
||||
gamut_type: Literal["A", "B", "C"] = "A"
|
||||
|
||||
class ColorPointColor(BaseModel):
|
||||
class ColorPointColor(BaseAttribute):
|
||||
color: "Attributes.ColorPoint" = Field(
|
||||
default_factory=lambda: Attributes.ColorPoint()
|
||||
)
|
||||
|
||||
class ColorPoint(BaseModel):
|
||||
class ColorPoint(BaseAttribute):
|
||||
xy: "Attributes.XY" = Field(default_factory=lambda: Attributes.XY(x=0.0, y=0.0))
|
||||
|
||||
class ColorMirekSchema(BaseModel):
|
||||
class ColorMirekSchema(BaseAttribute):
|
||||
mirek_minimum: int = Field(default=153, ge=153, le=500)
|
||||
mirek_maximum: int = Field(default=500, ge=153, le=500)
|
||||
|
||||
class ColorTemp(BaseModel):
|
||||
class ColorTemp(BaseAttribute):
|
||||
mirek: Optional[int] = Field(default=0, ge=153, le=500)
|
||||
mirek_valid: Optional[bool] = True
|
||||
mirek_schema: Optional["Attributes.ColorMirekSchema"] = Field(
|
||||
default_factory=lambda: Attributes.ColorMirekSchema()
|
||||
)
|
||||
|
||||
class Dependee(BaseModel):
|
||||
class Dependee(BaseAttribute):
|
||||
type: str = "unknown"
|
||||
target: "Attributes.Identifier" = Field(...)
|
||||
level: str = "unknown"
|
||||
|
||||
class Dimming(BaseModel):
|
||||
class Config:
|
||||
frozen = True
|
||||
allow_mutation = False
|
||||
validate_assignment = True
|
||||
|
||||
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)
|
||||
|
||||
class DimmingDelta(BaseModel):
|
||||
class DimmingDelta(BaseAttribute):
|
||||
action: Optional[Literal["up", "down", "stop"]] = "stop"
|
||||
brightness_delta: Optional[float] = Field(default=0, ge=0, le=100)
|
||||
|
||||
class Dynamics(BaseModel):
|
||||
class Dynamics(BaseAttribute):
|
||||
status: str = "unknown"
|
||||
status_values: Optional[list[str]] = Field(default_factory=list)
|
||||
speed: Optional[float] = Field(default=0.0, ge=0, le=100)
|
||||
speed_valid: Optional[bool] = True
|
||||
|
||||
class Effects(BaseModel):
|
||||
class Effects(BaseAttribute):
|
||||
effect: Optional[list[str]] = Field(default_factory=list)
|
||||
status_values: Optional[list[str]] = Field(default_factory=list)
|
||||
status: str = "unknown"
|
||||
effect_values: Optional[list[str]] = Field(default_factory=list)
|
||||
|
||||
class EntChannel(BaseModel):
|
||||
class EntChannel(BaseAttribute):
|
||||
channel_id: int = Field(ge=0, le=255)
|
||||
position: Optional["Attributes.XYZ"] = Field(
|
||||
default_factory=lambda: Attributes.XYZ(x=0.0, y=0.0, z=0.0)
|
||||
)
|
||||
members: list["Attributes.SegmentRef"] = Field(default_factory=list)
|
||||
|
||||
class EntLocation(BaseModel):
|
||||
class EntLocation(BaseAttribute):
|
||||
service_location: list["Attributes.ServiceLocation"] = Field(
|
||||
default_factory=list
|
||||
)
|
||||
|
||||
class Gamut(BaseModel):
|
||||
class Gamut(BaseAttribute):
|
||||
red: XY = Field(default_factory=lambda: Attributes.XY(x=0.0, y=0.0))
|
||||
green: XY = Field(default_factory=lambda: Attributes.XY(x=0.0, y=0.0))
|
||||
blue: XY = Field(default_factory=lambda: Attributes.XY(x=0.0, y=0.0))
|
||||
|
||||
class Gradient(BaseModel):
|
||||
class Gradient(BaseAttribute):
|
||||
points: Optional[list["Attributes.ColorPointColor"]] = Field(
|
||||
default_factory=list
|
||||
)
|
||||
points_capable: Optional[int] = Field(default=1, ge=0, le=255)
|
||||
|
||||
class Identifier(BaseModel):
|
||||
class Config:
|
||||
frozen = True
|
||||
allow_mutation = False
|
||||
validate_assignment = True
|
||||
class Identifier(BaseAttribute):
|
||||
|
||||
rid: UUID = Field(default_factory=uuid4)
|
||||
rid: UUID = Field(default_factory=default_uuid)
|
||||
rtype: str = "unknown"
|
||||
|
||||
class LightColor(BaseModel):
|
||||
class LightColor(BaseAttribute):
|
||||
xy: Optional[XY] = Field(default_factory=lambda: _XY(x=0.0, y=0.0))
|
||||
|
||||
class LightLevelValue(BaseModel):
|
||||
class LightLevelValue(BaseAttribute):
|
||||
light_level: Optional[int] = Field(default=0, ge=0, le=100000)
|
||||
light_level_valid: Optional[bool] = True
|
||||
|
||||
class LightEffect(BaseModel):
|
||||
class LightEffect(BaseAttribute):
|
||||
effect: Optional[Literal["fire", "candle", "no_effect"]] = "no_effect"
|
||||
|
||||
class Metadata(BaseModel):
|
||||
class Config:
|
||||
frozen = True
|
||||
allow_mutation = False
|
||||
validate_assignment = True
|
||||
class Metadata(BaseAttribute):
|
||||
|
||||
name: str = "unknown"
|
||||
archetype: Archetype | RoomType = Archetype.UNKNOWN_ARCHETYPE
|
||||
|
|
@ -366,36 +427,30 @@ class Attributes:
|
|||
default_factory=lambda: Attributes.Identifier(), repr=False
|
||||
)
|
||||
|
||||
class Motion(BaseModel):
|
||||
class Motion(BaseAttribute):
|
||||
motion: Optional[bool] = False
|
||||
motion_valid: Optional[bool] = True
|
||||
|
||||
class On(BaseModel):
|
||||
class Config:
|
||||
frozen = True
|
||||
allow_mutation = False
|
||||
validate_assignment = True
|
||||
class On(BaseAttribute):
|
||||
|
||||
on: Optional[bool] = Field(default=True, alias="on")
|
||||
|
||||
class Palette(BaseModel):
|
||||
class Palette(BaseAttribute):
|
||||
color: Optional[list["Attributes.PaletteColor"]] = Field(default_factory=list)
|
||||
dimming: Optional[list["Attributes.Dimming"]] = Field(default_factory=list)
|
||||
color_temperature: list["Attributes.PaletteTemperature"] = Field(
|
||||
default_factory=list
|
||||
)
|
||||
|
||||
class PaletteColor(BaseModel):
|
||||
class PaletteColor(BaseAttribute):
|
||||
color: Optional["Attributes.ColorPoint"] = Field(
|
||||
default_factory=lambda: Attributes.ColorPoint()
|
||||
)
|
||||
dimming: Optional["Attributes.Dimming"] = Field(
|
||||
default_factory=lambda: Attributes.Dimming(
|
||||
brightness=100.0, min_dim_level=100.0
|
||||
)
|
||||
default_factory=lambda: Attributes.Dimming(brightness=100.0)
|
||||
)
|
||||
|
||||
class PaletteTemperature(BaseModel):
|
||||
class PaletteTemperature(BaseAttribute):
|
||||
color_temperature: Optional["Attributes.ScenePaletteColorTemp"] = Field(
|
||||
default_factory=lambda: Attributes.ScenePaletteColorTemp(mirek=500)
|
||||
)
|
||||
|
|
@ -405,25 +460,27 @@ class Attributes:
|
|||
)
|
||||
)
|
||||
|
||||
class PowerState(BaseModel):
|
||||
class PowerState(BaseAttribute):
|
||||
battery_state: Optional[Literal["normal", "low", "critical"]] = "normal"
|
||||
battery_level: Optional[float] = Field(default=100.0, le=100.0, ge=0.0)
|
||||
|
||||
class ProductData(BaseModel):
|
||||
class ProductData(BaseAttribute):
|
||||
model_id: Optional[str] = "unknown"
|
||||
manufacturer_name: Optional[str] = "unknown"
|
||||
product_name: Optional[str] = "unknown"
|
||||
product_archetype: Optional[Archetype] = Archetype.UNKNOWN_ARCHETYPE
|
||||
certified: Optional[bool] = False
|
||||
software_version: Optional[str] = Field(default="0.0.0", regex=r"\d+\.\d+\.\d+")
|
||||
software_version: Optional[str] = Field(
|
||||
default="0.0.0", regex=r"\d+\.(?:\d+\.)*\d+"
|
||||
)
|
||||
hardware_platform_type: Optional[str] = "unknown"
|
||||
|
||||
class RelativeRotary(BaseModel):
|
||||
class RelativeRotary(BaseAttribute):
|
||||
last_event: Optional["Attributes.RotaryEvent"] = Field(
|
||||
default_factory=lambda: Attributes.RotaryEvent()
|
||||
)
|
||||
|
||||
class RotaryEvent(BaseModel):
|
||||
class RotaryEvent(BaseAttribute):
|
||||
action: Optional[Literal["start", "repeat", "unknown"]] = "unknown"
|
||||
rotation: Optional["Attributes.RotaryRotation"] = Field(
|
||||
default_factory=lambda: Attributes.RotaryRotation(
|
||||
|
|
@ -431,36 +488,36 @@ class Attributes:
|
|||
)
|
||||
)
|
||||
|
||||
class RotaryRotation(BaseModel):
|
||||
class RotaryRotation(BaseAttribute):
|
||||
direction: Literal["clock_wise", "counter_clock_wise"] = "clock_wise"
|
||||
duration: Optional[float] = Field(0.0, ge=0.0)
|
||||
steps: Optional[int] = Field(0, ge=0)
|
||||
|
||||
class SceneDynamics(BaseModel):
|
||||
class SceneDynamics(BaseAttribute):
|
||||
duration: Optional[int] = Field(0, ge=0)
|
||||
|
||||
class SceneEffects(BaseModel):
|
||||
class SceneEffects(BaseAttribute):
|
||||
effect: Optional[Literal["fire", "candle", "no_effect"]] = "no_effect"
|
||||
|
||||
class ScenePaletteColorTemp(BaseModel):
|
||||
class ScenePaletteColorTemp(BaseAttribute):
|
||||
mirek: int = Field(153, ge=153, le=500)
|
||||
|
||||
class Segment(BaseModel):
|
||||
class Segment(BaseAttribute):
|
||||
start: int = Field(0, ge=0)
|
||||
length: int = Field(1, ge=1)
|
||||
|
||||
class SegmentManager(BaseModel):
|
||||
class SegmentManager(BaseAttribute):
|
||||
configurable: Optional[bool] = True
|
||||
max_segments: Optional[int] = Field(default=1, ge=1)
|
||||
segments: Optional[list["Attributes.Segment"]] = Field(default_factory=list)
|
||||
|
||||
class SegmentRef(BaseModel):
|
||||
class SegmentRef(BaseAttribute):
|
||||
service: "Attributes.Identifier" = Field(
|
||||
default_factory=lambda: Attributes.Identifier()
|
||||
)
|
||||
index: int = 0
|
||||
|
||||
class ServiceLocation(BaseModel):
|
||||
class ServiceLocation(BaseAttribute):
|
||||
service: "Attributes.Identifier" = Field(
|
||||
default_factory=lambda: Attributes.Identifier()
|
||||
)
|
||||
|
|
@ -469,33 +526,28 @@ class Attributes:
|
|||
)
|
||||
positions: list[Type["Attributes.XYZ"]] = Field(max_items=2, min_items=1)
|
||||
|
||||
class StreamProxy(BaseModel):
|
||||
class StreamProxy(BaseAttribute):
|
||||
mode: Literal["auto", "manual"] = "manual"
|
||||
node: "Attributes.Identifier" = Field(
|
||||
default_factory=lambda: Attributes.Identifier()
|
||||
)
|
||||
|
||||
class Temp(BaseModel):
|
||||
class Temp(BaseAttribute):
|
||||
temperature: float = Field(default=0.0, lt=100.0, gt=-100.0)
|
||||
temperature_valid: bool = True
|
||||
|
||||
class TimedEffects(BaseModel):
|
||||
class TimedEffects(BaseAttribute):
|
||||
effect: Optional[str] = "none"
|
||||
duration: Optional[float] = Field(default=0.0, ge=0.0)
|
||||
status_values: Optional[list[str]] = Field(default_factory=list)
|
||||
status: Optional[str] = "unknown"
|
||||
effect_values: Optional[list[str]] = Field(default_factory=list)
|
||||
|
||||
class XY(BaseModel):
|
||||
class Config:
|
||||
frozen = True
|
||||
allow_mutation = False
|
||||
validate_assignment = True
|
||||
|
||||
class XY(BaseAttribute):
|
||||
x: float = Field(0.0, ge=0.0, le=1.0)
|
||||
y: float = Field(0.0, ge=0.0, le=1.0)
|
||||
|
||||
class XYZ(BaseModel):
|
||||
class XYZ(BaseAttribute):
|
||||
x: float = Field(ge=-1.0, le=1.0)
|
||||
y: float = Field(ge=-1.0, le=1.0)
|
||||
z: float = Field(ge=-1.0, le=1.0)
|
||||
|
|
@ -545,11 +597,13 @@ class HueEntsV2:
|
|||
id: UUID
|
||||
id_v1: str = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
|
||||
description: str = ""
|
||||
configuration_schema: dict[str, Any] = Field(default_factory=dict)
|
||||
trigger_schema: dict[str, Any] = Field(default_factory=dict)
|
||||
state_schema: dict[str, Any] = Field(default_factory=dict)
|
||||
configuration_schema: dict[Any, Any] = Field(default_factory=dict)
|
||||
trigger_schema: dict[Any, Any] = Field(default_factory=dict)
|
||||
state_schema: dict[Any, Any] = Field(default_factory=dict)
|
||||
version: str = Field("0.0.1", regex=r"^[0-9]+\.[0-9]+\.[0-9]+$")
|
||||
metadata: dict[str, str] = Field(default_factory=dict)
|
||||
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)
|
||||
|
||||
class Bridge(Entity):
|
||||
type: ClassVar[str] = "bridge"
|
||||
|
|
@ -716,13 +770,17 @@ class HueEntsV2:
|
|||
|
||||
class Room(Entity):
|
||||
type: ClassVar[str] = "room"
|
||||
id: UUID
|
||||
id_v1: str = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
|
||||
services: list[Attributes.Identifier] = Field(
|
||||
id: Optional[UUID]
|
||||
id_v1: Optional[str] = Field(
|
||||
"", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$"
|
||||
)
|
||||
services: Optional[list[Attributes.Identifier]] = Field(
|
||||
default_factory=list, alias="service"
|
||||
)
|
||||
metadata: Attributes.Metadata = Field(default_factory=Attributes.Metadata)
|
||||
children: list[Attributes.Identifier] = Field(default_factory=list)
|
||||
metadata: Optional[Attributes.Metadata] = Field(
|
||||
default_factory=Attributes.Metadata
|
||||
)
|
||||
children: Optional[list[Attributes.Identifier]] = Field(default_factory=list)
|
||||
|
||||
class Scene(Entity):
|
||||
id: UUID
|
||||
|
|
|
|||
|
|
@ -0,0 +1,198 @@
|
|||
from collections import deque
|
||||
from inspect import Parameter, signature
|
||||
from typing import Any
|
||||
from time import time
|
||||
|
||||
from re import compile as re_compile
|
||||
|
||||
from httpx._urls import URL as _URL
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
try:
|
||||
from ujson import dumps, loads, JSONDecodeError
|
||||
except ImportError:
|
||||
try:
|
||||
from orjson import dumps, loads, JSONDecodeError
|
||||
except ImportError:
|
||||
from json import dumps, loads, JSONDecodeError
|
||||
|
||||
try:
|
||||
from yarl import URL as UR
|
||||
except ImportError:
|
||||
...
|
||||
|
||||
__all__ = (
|
||||
"ENDPOINT_METHOD",
|
||||
"STR_FMT_RE",
|
||||
"URL_TYPES",
|
||||
"IP_RE",
|
||||
"MSG_RE_BYTES",
|
||||
"MSG_RE_TEXT",
|
||||
"URL",
|
||||
"LRU",
|
||||
"LRUItem",
|
||||
"get_url_args",
|
||||
"get_data_fields",
|
||||
"ret_cls",
|
||||
)
|
||||
|
||||
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")
|
||||
|
||||
MSG_RE_BYTES = re_compile(
|
||||
rb"(?=((?P<hello>^: hi\n\n$)|^id:\s(?P<id>[0-9]+:\d*?)\ndata:(?P<data>[^$]+)\n\n))\1"
|
||||
)
|
||||
MSG_RE_TEXT = re_compile(
|
||||
r"(?=((?P<hello>^: hi\n\n$)|^id:\s(?P<id>[0-9]+:\d*?)\ndata:(?P<data>[^$]+)\n\n))\1"
|
||||
)
|
||||
|
||||
|
||||
def get_url_args(url):
|
||||
kwds = {}
|
||||
match = STR_FMT_RE.finditer(url)
|
||||
for m in match:
|
||||
name = m.group(2)
|
||||
if len(m.groups()) >= 4:
|
||||
type_ = URL_TYPES[m.group(3)]
|
||||
else:
|
||||
type_ = str
|
||||
kwds[name] = type_
|
||||
return kwds
|
||||
|
||||
|
||||
def get_data_fields(fn, data, args, kwargs) -> dict[str, Any]:
|
||||
for param_name, param in signature(fn).parameters.items():
|
||||
if param_name == "self":
|
||||
continue
|
||||
|
||||
if param.kind == Parameter.POSITIONAL_ONLY:
|
||||
data[param_name] = param.annotation(str(args[0]))
|
||||
if len(args) > 1:
|
||||
args = args[1:]
|
||||
|
||||
else:
|
||||
if param_name in fn.__annotations__:
|
||||
anno = fn.__annotations__[param_name]
|
||||
if isinstance(anno, type):
|
||||
type_ = anno
|
||||
elif anno._name == "Optional":
|
||||
if hasattr(anno.__args__[0], "_name"):
|
||||
type_ = str
|
||||
else:
|
||||
type_ = anno.__args__[0]
|
||||
else:
|
||||
type_ = fn.__annotations__[param_name]
|
||||
else:
|
||||
type_ = str
|
||||
|
||||
if param.default is param.empty:
|
||||
if param_name not in kwargs and param_name not in data:
|
||||
raise TypeError(
|
||||
f"Missing required argument {param_name} for {fn.__name__}"
|
||||
)
|
||||
if param_name in kwargs:
|
||||
data[param_name] = type_(kwargs.pop(param_name))
|
||||
|
||||
else:
|
||||
if v := kwargs.pop(param_name, param.default):
|
||||
data[param_name] = type_(v)
|
||||
return data
|
||||
|
||||
|
||||
def ret_cls(cls):
|
||||
def wrapped(fn):
|
||||
async def sub_wrap(self, *args, **kwargs):
|
||||
try:
|
||||
ret = loads(
|
||||
(await fn(self, *args, **kwargs))
|
||||
.content.decode()
|
||||
.rstrip("\\r\\n")
|
||||
.lstrip(" ")
|
||||
)
|
||||
|
||||
kwargs.pop("base_uri", None)
|
||||
ret = ret.get("data", [])
|
||||
_rets = []
|
||||
|
||||
if isinstance(ret, list):
|
||||
for r in ret:
|
||||
_rets.append(cls(**r))
|
||||
else:
|
||||
return cls(**ret)
|
||||
return _rets
|
||||
except JSONDecodeError:
|
||||
return []
|
||||
|
||||
return sub_wrap
|
||||
|
||||
wrapped.__return_type__ = cls
|
||||
|
||||
return wrapped
|
||||
|
||||
|
||||
class LRUItem(BaseModel):
|
||||
access_time: int = Field(default_factory=lambda: int(time()))
|
||||
value: Any = object()
|
||||
|
||||
def __id__(self):
|
||||
return id(self.value)
|
||||
|
||||
def __hash__(self):
|
||||
return hash(self.value)
|
||||
|
||||
|
||||
class LRU(set):
|
||||
def __init__(self, maxsize, /, *items):
|
||||
super().__init__()
|
||||
self.maxsize = maxsize
|
||||
self.items = deque(maxlen=maxsize)
|
||||
for item in items[:maxsize]:
|
||||
self.add(LRUItem(value=item))
|
||||
|
||||
def add(self, item):
|
||||
if len(self) + 1 > self.maxsize:
|
||||
new = self ^ set(
|
||||
sorted(self, key=lambda x: x.access_time)[::-1][
|
||||
: len(self) + 1 - self.maxsize
|
||||
]
|
||||
)
|
||||
old = self - new
|
||||
self -= old
|
||||
|
||||
super().add(LRUItem(value=item))
|
||||
|
||||
def pop(self):
|
||||
super().pop().value
|
||||
|
||||
def remove(self, item):
|
||||
super().remove(*filter(lambda x: x.value == item, self))
|
||||
|
||||
def extend(self, *items):
|
||||
len_new = len(self) + len(items)
|
||||
if len_new > self.maxsize:
|
||||
new = self ^ set(
|
||||
sorted(self, key=lambda x: x.access_time)[::-1][
|
||||
: len_new - self.maxsize
|
||||
]
|
||||
)
|
||||
old = self - new
|
||||
self -= old
|
||||
self |= set([LRUItem(value=item) for item in items])
|
||||
|
||||
|
||||
class URL(_URL):
|
||||
def __truediv__(self, other):
|
||||
# Why am i doing this? good question.
|
||||
try:
|
||||
return URL(str(UR(f"{self}") / other.lstrip("/")))
|
||||
except NameError:
|
||||
return URL(f"{self}{other.lstrip('/')}")
|
||||
|
||||
def __floordiv__(self, other):
|
||||
try:
|
||||
return URL(str(UR(f"{self}") / other.lstrip("/")))
|
||||
except NameError:
|
||||
return URL(f"{self}{other.lstrip('/')}")
|
||||
|
|
@ -29,8 +29,8 @@ pydantic = ">=1.10.0"
|
|||
yarl = ">=1.8.0"
|
||||
ujson = ">=5.6.0"
|
||||
rich = ">=12.6.0"
|
||||
orjson = "^3.8.3"
|
||||
uvloop = "^0.17.0"
|
||||
aiofiles = ">=22.1.0"
|
||||
pyyaml = "^6.0"
|
||||
|
||||
[tool.poetry.dev-dependencies]
|
||||
black = ">=22.10.0"
|
||||
|
|
@ -40,6 +40,13 @@ pycodestyle = "^2.10.0"
|
|||
pylint = "^2.15.7"
|
||||
mypy = "^0.991"
|
||||
|
||||
[tool.poetry.group.linux.dependencies]
|
||||
uvloop = "^0.17.0"
|
||||
|
||||
[tool.poetry.group.linux]
|
||||
optional = true
|
||||
|
||||
|
||||
[build-system]
|
||||
requires = ["setuptools >= 39.2.0", "wheel", "poetry-core>=1.0.0"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
|
|
|||
Loading…
Reference in New Issue