ra/phlyght
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

endpoints done

Browse Source
This commit is contained in:
rooba 2022-11-29 15:14:56 -08:00
parent 72a98b8541
commit cf95790544
6 changed files with 413 additions and 323 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -1,3 +1,4 @@
**/.venv
**/poetry.lock **/poetry.lock
**/__pycache__ **/__pycache__
**/testing **/testing

4
phlyght/__init__.py
View File

@ -1,6 +1,6 @@
from .api import Router from .api import Router
from .models import ( from .models import (
Archetype, _Archetype,
Room, Room,
Light, Light,
Scene, Scene,
@ -25,7 +25,7 @@ from .models import (
__all__ = ( __all__ = (
"Router", "Router",
"Archetype", "_Archetype",
"Room", "Room",
"Light", "Light",
"Scene", "Scene",

78
phlyght/api.py
View File

@ -1,5 +1,6 @@
__all__ = ("Router", "route", "RouterMeta", "SubRouter", "HughApi") __all__ = ("Router", "route", "RouterMeta", "SubRouter", "HughApi")
from asyncio import get_running_loop, sleep
from inspect import signature from inspect import signature
from re import compile from re import compile
from typing import Any, Literal, Optional from typing import Any, Literal, Optional
@ -7,6 +8,11 @@ from uuid import UUID
from httpx import AsyncClient from httpx import AsyncClient
from httpx._urls import URL as _URL from httpx._urls import URL as _URL
from httpx._exceptions import ConnectTimeout
from httpcore._exceptions import ReadTimeout
from json import loads
from pydantic import BaseModel
try: try:
from yarl import URL as UR from yarl import URL as UR
@ -24,6 +30,21 @@ from . import models
STR_FMT_RE = compile(r"""(?=(\{([^:]+)(?::([^}]+))?\}))\1""") STR_FMT_RE = compile(r"""(?=(\{([^:]+)(?::([^}]+))?\}))\1""")
URL_TYPES = {"str": str, "int": int} URL_TYPES = {"str": str, "int": int}
MSG_RE = compile(
b"""(?=((?P<hello>^hi\\n\\n$)|^id:\\s(?P<id>[0-9]+:\\d*?)\\ndata:(?P<data>[^$]+)\\n\\n))\\1"""
)
TYPE_CACHE = {}
for k in models.__all__:
if (
k == "Literal"
or k.startswith("_")
or not issubclass(getattr(models, k), BaseModel)
):
continue
# print(getattr(models, k).__dict__)
TYPE_CACHE[getattr(models, k).__fields__["type"].default] = getattr(models, k)
def get_url_args(url): def get_url_args(url):
kwds = {} kwds = {}
@ -62,7 +83,6 @@ def ret_cls(cls):
if isinstance(ret, list): if isinstance(ret, list):
for r in ret: for r in ret:
print(r)
_rets.append(cls(**r)) _rets.append(cls(**r))
else: else:
return cls(**ret) return cls(**ret)
@ -85,6 +105,10 @@ def route(method, endpoint) -> Any:
): ):
params = params or {} params = params or {}
data = data or {} data = data or {}
if "headers" in kwargs:
headers = kwargs.pop("headers")
else:
headers = None
for param_name, param in signature(fn).parameters.items(): for param_name, param in signature(fn).parameters.items():
if param_name == "self": if param_name == "self":
continue continue
@ -137,12 +161,23 @@ def route(method, endpoint) -> Any:
else: else:
new_endpoint = URL(f"{self._api_path}") / endpoint new_endpoint = URL(f"{self._api_path}") / endpoint
if headers and headers.get("Accept", "") == "text/event-stream":
return self._client.stream(
method,
new_endpoint,
content=content,
data=data,
params=params,
headers=headers,
)
else:
return await self._client.request( return await self._client.request(
method, method,
new_endpoint, new_endpoint,
content=content, content=content,
data=data, data=data,
params=params, params=params,
headers=headers,
) )
return sub_wrap return sub_wrap
@ -652,8 +687,49 @@ class HughApi(SubRouter):
async def get_resources(self): async def get_resources(self):
... ...
@route("GET", "/../../eventstream/clip/v2")
async def listen_events(self):
...
class Router(HughApi, root="https://192.168.69.104"): class Router(HughApi, root="https://192.168.69.104"):
def __init__(self, hue_api_key: str): def __init__(self, hue_api_key: str):
super().__init__(hue_api_key) super().__init__(hue_api_key)
self._client = AsyncClient(headers=self._headers, verify=False) self._client = AsyncClient(headers=self._headers, verify=False)
self._subscription = None
def subscribe(self, *args, **kwargs):
if not self._subscription or self._subscription.done():
self._subscription = get_running_loop().create_task(self._subscribe())
async def _subscribe(self, *args, **kwargs):
stream = await self.listen_events(
headers={"Accept": "text/event-stream"} | self._headers
)
while get_running_loop().is_running():
resp = await stream.gen.__anext__()
_bound = resp.stream._stream._httpcore_stream
try:
async for msg in _bound:
payload = []
_match = MSG_RE.search(msg)
id_ = ""
if _match:
if _match.groupdict().get("hello", None):
# Handshake / Heartbeat
...
else:
payload = loads(_match.group("data"))
id_ = _match.group("id").decode()
objs = []
for event in payload:
for ob in event["data"]:
objs.append(TYPE_CACHE[ob["type"]](**ob))
print(objs)
except ReadTimeout:
stream = await self.listen_events(
headers={"Accept": "text/event-stream"} | self._headers
)

593
phlyght/models.py
View File

@ -1,13 +1,20 @@
from typing import Any, Literal, Optional, Generic, TypeAlias, TypeVar from typing import (
from uuid import UUID Any as _Any,
from dataclasses import dataclass Literal,
from enum import Enum, auto Optional as _Optional,
Generic as _Generic,
TypeAlias as _TypeAlias,
TypeVar as _TypeVar,
)
from uuid import UUID as _UUID
from dataclasses import dataclass as _dataclass
from enum import Enum as _Enum, auto as _auto
from pydantic import BaseModel, Field from pydantic import BaseModel as _BaseModel, Field as _Field
__all__ = ( __all__ = (
"Archetype", "_Archetype",
"RoomType", "_RoomType",
"Room", "Room",
"Light", "Light",
"Scene", "Scene",
@ -33,138 +40,138 @@ __all__ = (
"Homekit", "Homekit",
) )
_T_M: TypeAlias = "Archetype | Room | Light | Scene | Zone | BridgeHome | GroupedLight | Device | Bridge | DevicePower | ZigbeeConnectivity | ZGPConnectivity | Motion | Temperature | LightLevel | Button | BehaviorScript | BehaviorInstance | GeofenceClient | Geolocation | EntertainmentConfiguration | Entertainment | Resource | Homekit" _T_M: _TypeAlias = "_RoomType | _Archetype | Room | Light | Scene | Zone | BridgeHome | GroupedLight | Device | Bridge | DevicePower | ZigbeeConnectivity | ZGPConnectivity | Motion | Temperature | LightLevel | Button | BehaviorScript | BehaviorInstance | GeofenceClient | Geolocation | EntertainmentConfiguration | Entertainment | Resource | Homekit"
_T = TypeVar("_T") _T = _TypeVar("_T")
class RoomType(Enum): class _RoomType(_Enum):
@staticmethod @staticmethod
def _generate_next_value_(name, start, count, last_values): def _generate_next_value_(name, start, count, last_values):
return name.lower() return name.lower()
LIVING_ROOM = auto() LIVING_ROOM = _auto()
KITCHEN = auto() KITCHEN = _auto()
DINING = auto() DINING = _auto()
BEDROOM = auto() BEDROOM = _auto()
KIDS_BEDROOM = auto() KIDS_BEDROOM = _auto()
BATHROOM = auto() BATHROOM = _auto()
NURSERY = auto() NURSERY = _auto()
RECREATION = auto() RECREATION = _auto()
OFFICE = auto() OFFICE = _auto()
GYM = auto() GYM = _auto()
HALLWAY = auto() HALLWAY = _auto()
TOILET = auto() TOILET = _auto()
FRONT_DOOR = auto() FRONT_DOOR = _auto()
GARAGE = auto() GARAGE = _auto()
TERRACE = auto() TERRACE = _auto()
GARDEN = auto() GARDEN = _auto()
DRIVEWAY = auto() DRIVEWAY = _auto()
CARPORT = auto() CARPORT = _auto()
HOME = auto() HOME = _auto()
DOWNSTAIRS = auto() DOWNSTAIRS = _auto()
UPSTAIRS = auto() UPSTAIRS = _auto()
TOP_FLOOR = auto() TOP_FLOOR = _auto()
ATTIC = auto() ATTIC = _auto()
GUEST_ROOM = auto() GUEST_ROOM = _auto()
STAIRCASE = auto() STAIRCASE = _auto()
LOUNGE = auto() LOUNGE = _auto()
MAN_CAVE = auto() MAN_CAVE = _auto()
COMPUTER = auto() COMPUTER = _auto()
STUDIO = auto() STUDIO = _auto()
MUSIC = auto() MUSIC = _auto()
TV = auto() TV = _auto()
READING = auto() READING = _auto()
CLOSET = auto() CLOSET = _auto()
STORAGE = auto() STORAGE = _auto()
LAUNDRY_ROOM = auto() LAUNDRY_ROOM = _auto()
BALCONY = auto() BALCONY = _auto()
PORCH = auto() PORCH = _auto()
BARBECUE = auto() BARBECUE = _auto()
POOL = auto() POOL = _auto()
OTHER = auto() OTHER = _auto()
class Archetype(Enum): class _Archetype(_Enum):
@staticmethod @staticmethod
def _generate_next_value_(name, start, count, last_values): def _generate_next_value_(name, start, count, last_values):
return name.lower() return name.lower()
BRIDGE_V2 = auto() BRIDGE_V2 = _auto()
UNKNOWN_ARCHETYPE = auto() UNKNOWN_ARCHETYPE = _auto()
CLASSIC_BULB = auto() CLASSIC_BULB = _auto()
SULTAN_BULB = auto() SULTAN_BULB = _auto()
FLOOD_BULB = auto() FLOOD_BULB = _auto()
SPOT_BULB = auto() SPOT_BULB = _auto()
CANDLE_BULB = auto() CANDLE_BULB = _auto()
LUSTER_BULB = auto() LUSTER_BULB = _auto()
PENDANT_ROUND = auto() PENDANT_ROUND = _auto()
PENDANT_LONG = auto() PENDANT_LONG = _auto()
CEILING_ROUND = auto() CEILING_ROUND = _auto()
CEILING_SQUARE = auto() CEILING_SQUARE = _auto()
FLOOR_SHADE = auto() FLOOR_SHADE = _auto()
FLOOR_LANTERN = auto() FLOOR_LANTERN = _auto()
TABLE_SHADE = auto() TABLE_SHADE = _auto()
RECESSED_CEILING = auto() RECESSED_CEILING = _auto()
RECESSED_FLOOR = auto() RECESSED_FLOOR = _auto()
SINGLE_SPOT = auto() SINGLE_SPOT = _auto()
DOUBLE_SPOT = auto() DOUBLE_SPOT = _auto()
TABLE_WASH = auto() TABLE_WASH = _auto()
WALL_LANTERN = auto() WALL_LANTERN = _auto()
WALL_SHADE = auto() WALL_SHADE = _auto()
FLEXIBLE_LAMP = auto() FLEXIBLE_LAMP = _auto()
GROUND_SPOT = auto() GROUND_SPOT = _auto()
WALL_SPOT = auto() WALL_SPOT = _auto()
PLUG = auto() PLUG = _auto()
HUE_GO = auto() HUE_GO = _auto()
HUE_LIGHTSTRIP = auto() HUE_LIGHTSTRIP = _auto()
HUE_IRIS = auto() HUE_IRIS = _auto()
HUE_BLOOM = auto() HUE_BLOOM = _auto()
BOLLARD = auto() BOLLARD = _auto()
WALL_WASHER = auto() WALL_WASHER = _auto()
HUE_PLAY = auto() HUE_PLAY = _auto()
VINTAGE_BULB = auto() VINTAGE_BULB = _auto()
CHRISTMAS_TREE = auto() CHRISTMAS_TREE = _auto()
HUE_CENTRIS = auto() HUE_CENTRIS = _auto()
HUE_LIGHTSTRIP_TV = auto() HUE_LIGHTSTRIP_TV = _auto()
HUE_TUBE = auto() HUE_TUBE = _auto()
HUE_SIGNE = auto() HUE_SIGNE = _auto()
@dataclass @_dataclass
class _Dimming: class _Dimming:
brightness: float brightness: float
min_dim_level: Optional[float] = Field(0, repr=False) min_dim_level: _Optional[float] = _Field(0, repr=False)
@dataclass @_dataclass
class _XY: class _XY:
x: float x: float
y: float y: float
@dataclass @_dataclass
class _On: class _On:
on: bool = Field(..., alias="on") on: bool = _Field(..., alias="on")
@dataclass @_dataclass
class _ColorPoint: class _ColorPoint:
xy: _XY xy: _XY
@dataclass(frozen=True) @_dataclass(frozen=True)
class _Identifier: class _Identifier:
rid: str rid: str
rtype: str rtype: str
@dataclass(frozen=True) @_dataclass(frozen=True)
class _Metadata: class _Metadata:
name: str name: str
archetype: Optional[Archetype | RoomType] = Archetype.UNKNOWN_ARCHETYPE archetype: _Optional[_Archetype | _RoomType] = _Archetype.UNKNOWN_ARCHETYPE
image: Optional[_Identifier] = Field(None, repr=False) image: _Optional[_Identifier] = _Field(None, repr=False)
class _HueGroupedMeta(type): class _HueGroupedMeta(type):
@ -179,101 +186,101 @@ class _HueGrouped(metaclass=_HueGroupedMeta):
class _Lights(_HueGrouped): class _Lights(_HueGrouped):
class ColorTemperature(BaseModel): class ColorTemperature(_BaseModel):
mirek: Optional[int] mirek: _Optional[int]
mirek_valid: bool mirek_valid: bool
mirek_schema: dict[str, float] mirek_schema: dict[str, float]
class Gamut(BaseModel): class Gamut(_BaseModel):
red: _XY red: _XY
green: _XY green: _XY
blue: _XY blue: _XY
class Color(BaseModel): class Color(_BaseModel):
xy: _XY xy: _XY = _Field(None)
gamut: "_Lights.Gamut" gamut: "_Lights.Gamut" = _Field(None)
gamut_type: Literal["A"] | Literal["B"] | Literal["C"] gamut_type: Literal["A", "B", "C"] = _Field(None)
class Dynamics(BaseModel): class Dynamics(_BaseModel):
status: str status: str = _Field(None)
status_values: list[str] status_values: list[str] = _Field(None)
speed: float speed: float = _Field(None)
speed_valid: bool speed_valid: bool = _Field(None)
class Gradient(BaseModel): class Gradient(_BaseModel):
points: list[_ColorPoint] points: list[_ColorPoint] = _Field([])
points_capable: int points_capable: int = _Field(0)
class Effects(BaseModel): class Effects(_BaseModel):
effect: Optional[list[str]] = Field(repr=False) effect: _Optional[list[str]] = _Field(repr=False)
status_values: list[str] = Field(repr=False) status_values: list[str] = _Field(repr=False)
status: str status: str = _Field("")
effect_values: list[str] = Field(repr=False) effect_values: list[str] = _Field(repr=False)
class TimedEffects(BaseModel): class TimedEffects(_BaseModel):
effect: str effect: str = _Field("")
duration: int duration: int = _Field(0)
status_values: list[str] = Field(repr=False) status_values: list[str] = _Field(repr=False)
status: str status: str = _Field("")
effect_values: list[str] = Field(repr=False) effect_values: list[str] = _Field(repr=False)
class Light(Generic[_T], BaseModel): class Light(_Generic[_T], _BaseModel):
id: UUID id: _UUID
id_v1: Optional[str] = Field( id_v1: _Optional[str] = _Field(
..., regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$" ..., regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$"
) )
owner: _Identifier owner: _Optional[_Identifier]
metadata: _Metadata metadata: _Optional[_Metadata]
on: _On = Field(repr=False) on: _Optional[_On] = _Field(repr=False)
dimming: _Dimming dimming: _Optional[_Dimming]
dimming_delta: dict dimming_delta: _Optional[dict]
color_temperature: Optional["_Lights.ColorTemperature"] color_temperature: _Optional["_Lights.ColorTemperature"]
color_temperature_delta: Optional[dict] color_temperature_delta: _Optional[dict]
color: Optional["_Lights.Color"] color: _Optional["_Lights.Color"]
gradient: Optional["_Lights.Gradient"] gradient: _Optional["_Lights.Gradient"]
dynamics: "_Lights.Dynamics" dynamics: _Optional["_Lights.Dynamics"]
alert: dict[str, list[str]] alert: _Optional[dict[str, list[str]]]
signaling: dict signaling: _Optional[dict]
mode: str mode: _Optional[str]
effects: "_Lights.Effects" effects: _Optional["_Lights.Effects"]
type: Literal["light"] type: Literal["light"] = "light"
_Lights.update() _Lights.update()
class _Scenes(_HueGrouped): class _Scenes(_HueGrouped):
class Action(BaseModel): class Action(_BaseModel):
on: Optional[_On] on: _Optional[_On]
dimming: Optional[_Dimming] dimming: _Optional[_Dimming]
color: Optional[_ColorPoint] color: _Optional[_ColorPoint]
color_temperature: Optional[dict[str, float]] color_temperature: _Optional[dict[str, float]]
gradient: Optional[dict[str, list[_ColorPoint]]] gradient: _Optional[dict[str, list[_ColorPoint]]]
effects: Optional[dict[str, str]] effects: _Optional[dict[str, str]]
dynamics: Optional[dict[str, float]] dynamics: _Optional[dict[str, float]]
class Actions(BaseModel): class Actions(_BaseModel):
target: _Identifier target: _Identifier
action: "_Scenes.Action" = Field(repr=False) action: "_Scenes.Action" = _Field(repr=False)
dimming: Optional[_Dimming] dimming: _Optional[_Dimming]
color: Optional[_ColorPoint] color: _Optional[_ColorPoint]
class PaletteColor(BaseModel): class PaletteColor(_BaseModel):
color: _ColorPoint color: _ColorPoint
dimming: _Dimming dimming: _Dimming
class PaletteTemperature(BaseModel): class PaletteTemperature(_BaseModel):
color_temperature: dict[str, float] color_temperature: dict[str, float]
dimming: _Dimming dimming: _Dimming
class Palette(BaseModel): class Palette(_BaseModel):
color: list["_Scenes.PaletteColor"] color: list["_Scenes.PaletteColor"]
dimming: Optional[list[_Dimming]] dimming: _Optional[list[_Dimming]]
color_temperature: list["_Scenes.PaletteTemperature"] color_temperature: list["_Scenes.PaletteTemperature"]
class Scene(BaseModel): class Scene(_BaseModel):
id: UUID id: _UUID
id_v1: Optional[str] = Field( id_v1: _Optional[str] = _Field(
..., regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$" ..., regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$"
) )
metadata: _Metadata metadata: _Metadata
@ -282,90 +289,90 @@ class _Scenes(_HueGrouped):
palette: "_Scenes.Palette" palette: "_Scenes.Palette"
speed: float speed: float
auto_dynamic: bool auto_dynamic: bool
type: Literal["scene"] type: Literal["scene"] = "scene"
_Scenes.update() _Scenes.update()
class Room(BaseModel): class Room(_BaseModel):
type: Literal["room"] type: Literal["room"] = "room"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
services: list[_Identifier] services: list[_Identifier]
metadata: _Metadata metadata: _Metadata
children: list[_Identifier] children: list[_Identifier]
class Zone(BaseModel): class Zone(_BaseModel):
type: Literal["zone"] type: Literal["zone"] = "zone"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
services: list[_Identifier] services: list[_Identifier]
metadata: _Metadata metadata: _Metadata
children: list[_Identifier] children: list[_Identifier]
class BridgeHome(BaseModel): class BridgeHome(_BaseModel):
type: Literal["bridge_home"] type: Literal["bridge_home"] = "bridge_home"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
services: list[_Identifier] services: list[_Identifier]
children: list[_Identifier] children: list[_Identifier]
class GroupedLight(BaseModel): class GroupedLight(_BaseModel):
type: Literal["grouped_light"] type: Literal["grouped_light"] = "grouped_light"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
on: _On = Field(repr=False) on: _On = _Field(repr=False)
alert: dict[str, list[str]] alert: dict[str, list[str]]
class _ProductData(BaseModel): class _ProductData(_BaseModel):
model_id: str model_id: str
manufacturer_name: str manufacturer_name: str
product_name: str product_name: str
product_archetype: Archetype product_archetype: _Archetype
certified: bool certified: bool
software_version: Optional[str] software_version: _Optional[str]
hardware_platform_type: Optional[str] hardware_platform_type: _Optional[str]
class Device(BaseModel): class Device(_BaseModel):
type: Literal["device"] type: Literal["device"] = "device"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
services: list[_Identifier] services: list[_Identifier]
metadata: _Metadata metadata: _Metadata
product_data: _ProductData product_data: _ProductData
class Bridge(BaseModel): class Bridge(_BaseModel):
type: Literal["bridge"] type: Literal["bridge"] = "bridge"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
bridge_id: str bridge_id: str
time_zone: dict[str, str] time_zone: dict[str, str]
class _PowerState(BaseModel): class _PowerState(_BaseModel):
battery_state: Literal["normal", "low", "critical"] battery_state: Literal["normal", "low", "critical"]
battery_level: float = Field(le=100.0, ge=0.0) battery_level: float = _Field(le=100.0, ge=0.0)
class DevicePower(BaseModel): class DevicePower(_BaseModel):
type: Literal["device_power"] type: Literal["device_power"] = "device_power"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
owner: _Identifier owner: _Identifier
power_state: _PowerState power_state: _PowerState
class ZigbeeConnectivity(BaseModel): class ZigbeeConnectivity(_BaseModel):
type: Literal["zigbee_connectivity"] type: Literal["zigbee_connectivity"] = "zigbee_connectivity"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
owner: _Identifier owner: _Identifier
status: Literal[ status: Literal[
"connected", "disconnected", "connectivity_issue", "unidirectional_incoming" "connected", "disconnected", "connectivity_issue", "unidirectional_incoming"
@ -373,10 +380,10 @@ class ZigbeeConnectivity(BaseModel):
mac_address: str mac_address: str
class ZGPConnectivity(BaseModel): class ZGPConnectivity(_BaseModel):
type: Literal["zgp_connectivity"] type: Literal["zgp_connectivity"] = "zgp_connectivity"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
owner: _Identifier owner: _Identifier
status: Literal[ status: Literal[
"connected", "disconnected", "connectivity_issue", "unidirectional_incoming" "connected", "disconnected", "connectivity_issue", "unidirectional_incoming"
@ -384,47 +391,47 @@ class ZGPConnectivity(BaseModel):
source_id: str source_id: str
class Motion(BaseModel): class Motion(_BaseModel):
type: Literal["motion"] type: Literal["motion"] = "motion"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
owner: _Identifier owner: _Identifier
enabled: bool enabled: bool
motion: dict[str, bool] motion: dict[str, bool]
class _Temp(BaseModel): class _Temp(_BaseModel):
temperature: float = Field(lt=100.0, gt=-100.0) temperature: float = _Field(lt=100.0, gt=-100.0)
temperature_valid: bool temperature_valid: bool
class Temperature(BaseModel): class Temperature(_BaseModel):
type: Literal["temperature"] type: Literal["temperature"] = "temperature"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
owner: _Identifier owner: _Identifier
enabled: bool enabled: _Optional[bool]
temperature: _Temp temperature: _Temp
class _Light(BaseModel): class _Light(_BaseModel):
light_level: int light_level: _Optional[int]
light_level_valid: bool light_level_valid: _Optional[bool]
class LightLevel(BaseModel): class LightLevel(_BaseModel):
type: Literal["light_level"] type: Literal["light_level"] = "light_level"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
owner: _Identifier owner: _Optional[_Identifier]
enabled: bool enabled: _Optional[bool]
light: _Light light: _Optional[_Light]
class Button(BaseModel): class Button(_BaseModel):
type: Literal["button"] type: Literal["button"] = "button"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
owner: _Identifier owner: _Identifier
metadata: dict[Literal["control_id"], int] metadata: dict[Literal["control_id"], int]
button: dict[ button: dict[
@ -439,133 +446,133 @@ class Button(BaseModel):
] ]
class BehaviorScript(BaseModel): class BehaviorScript(_BaseModel):
type: Literal["behavior_script"] type: Literal["behavior_script"] = "behavior_script"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
description: str description: str
configuration_schema: dict[str, Any] configuration_schema: dict[str, _Any]
trigger_schema: dict[str, Any] trigger_schema: dict[str, _Any]
state_schema: dict[str, Any] state_schema: dict[str, _Any]
version: str version: str
metadata: dict[str, str] metadata: dict[str, str]
class _Dependee(BaseModel): class _Dependee(_BaseModel):
type: str type: str
target: _Identifier target: _Identifier
level: str level: str
class BehaviorInstance(BaseModel): class BehaviorInstance(_BaseModel):
type: Literal["behavior_instance"] type: Literal["behavior_instance"] = "behavior_instance"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
script_id: str script_id: str
enabled: bool enabled: bool
state: Optional[dict[str, Any]] state: _Optional[dict[str, _Any]]
configuration: dict[str, Any] configuration: dict[str, _Any]
dependees: list[_Dependee] dependees: list[_Dependee]
status: Literal["initializing", "running", "disabled", "errored"] status: Literal["initializing", "running", "disabled", "errored"]
last_error: str last_error: str
metadata: dict[Literal["name"], str] metadata: dict[Literal["name"], str]
migrated_from: Optional[str] = None migrated_from: _Optional[str] = None
class GeofenceClient(BaseModel): class GeofenceClient(_BaseModel):
type: Literal["geofence_client"] type: Literal["geofence_client"] = "geofence_client"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
name: str name: str
class Geolocation(BaseModel): class Geolocation(_BaseModel):
type: Literal["geolocation"] type: Literal["geolocation"] = "geolocation"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
is_configured: bool = False is_configured: bool = False
class _StreamProxy(BaseModel): class _StreamProxy(_BaseModel):
mode: Literal["auto", "manual"] mode: Literal["auto", "manual"]
node: _Identifier node: _Identifier
class _XYZ(BaseModel): class _XYZ(_BaseModel):
x: float = Field(ge=-1.0, le=1.0) x: float = _Field(ge=-1.0, le=1.0)
y: 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) z: float = _Field(ge=-1.0, le=1.0)
class _SegmentRef(BaseModel): class _SegmentRef(_BaseModel):
service: _Identifier service: _Identifier
index: int index: int
class _EntertainmentChannel(BaseModel): class _EntertainmentChannel(_BaseModel):
channel_id: int = Field(ge=0, le=255) channel_id: int = _Field(ge=0, le=255)
position: _XYZ position: _XYZ
members: list[_SegmentRef] members: list[_SegmentRef]
class _ServiceLocation(BaseModel): class _ServiceLocation(_BaseModel):
service: _Identifier service: _Identifier
position: _XYZ position: _XYZ
positions: list[_XYZ] = Field(max_items=2, min_items=1) positions: list[_XYZ] = _Field(max_items=2, min_items=1)
class _EntertainmentLocation(BaseModel): class _EntertainmentLocation(_BaseModel):
service_location: Optional[list[_ServiceLocation]] = [] service_location: _Optional[list[_ServiceLocation]] = []
class EntertainmentConfiguration(BaseModel): class EntertainmentConfiguration(_BaseModel):
type: Literal["entertainment_configuration"] type: Literal["entertainment_configuration"] = "entertainment_configuration"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
metadata: dict[Literal["name"], str] metadata: dict[Literal["name"], str]
name: Optional[str] = "" name: _Optional[str] = ""
configuration_type: Literal["screen", "monitor", "music", "3dspace", "other"] configuration_type: Literal["screen", "monitor", "music", "3dspace", "other"]
status: Literal["active", "inactive"] status: Literal["active", "inactive"]
active_streamer: Optional[_Identifier] = None active_streamer: _Optional[_Identifier] = None
stream_proxy: _StreamProxy stream_proxy: _StreamProxy
channels: list[_EntertainmentChannel] channels: list[_EntertainmentChannel]
locations: Optional[_EntertainmentLocation] = None locations: _Optional[_EntertainmentLocation] = None
light_services: list[_Identifier] light_services: list[_Identifier]
class _Segment(BaseModel): class _Segment(_BaseModel):
start: int = Field(..., ge=0) start: int = _Field(..., ge=0)
length: int = Field(..., ge=1) length: int = _Field(..., ge=1)
class _SegmentManager(BaseModel): class _SegmentManager(_BaseModel):
configurable: bool configurable: bool
max_segments: int = Field(..., ge=1) max_segments: int = _Field(..., ge=1)
segments: list[_Segment] segments: list[_Segment]
class Entertainment(BaseModel): class Entertainment(_BaseModel):
type: Literal["entertainment"] type: Literal["entertainment"] = "entertainment"
id: UUID id: _UUID
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
owner: _Identifier owner: _Identifier
renderer: bool renderer: bool
proxy: bool proxy: bool
max_streams: Optional[int] = Field(1, ge=1) max_streams: _Optional[int] = _Field(1, ge=1)
segments: Optional[_SegmentManager] = None segments: _Optional[_SegmentManager] = None
class Homekit(BaseModel): class Homekit(_BaseModel):
id: UUID id: _UUID
type: Optional[str] type: _Optional[str] = "resource"
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
status: Literal["paired", "pairing", "unpaired"] status: Literal["paired", "pairing", "unpaired"]
class Resource(BaseModel): class Resource(_BaseModel):
id: UUID id: _UUID
type: Optional[str] type: _Optional[str] = "device"
id_v1: Optional[str] = Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$") id_v1: _Optional[str] = _Field("", regex=r"^(\/[a-z]{4,32}\/[0-9a-zA-Z-]{1,32})?$")
Light = _Lights.Light Light = _Lights.Light

3
pyproject.toml
View File

@ -9,6 +9,9 @@ python = ">=3.11,<4.0.0"
httpx = ">=0.23.1" httpx = ">=0.23.1"
pydantic = ">=1.10.2" pydantic = ">=1.10.2"
yarl = ">=1.8.1" yarl = ">=1.8.1"
ujson = "^5.5.0"
loguru = "^0.6.0"
orjson = "^3.8.2"
[tool.poetry.dev-dependencies] [tool.poetry.dev-dependencies]
black = ">=22.10.0" black = ">=22.10.0"

43
test.py
View File

@ -1,4 +1,4 @@
from asyncio import run from asyncio import run, sleep
from phlyght.api import Router from phlyght.api import Router
@ -9,31 +9,34 @@ except ImportError:
async def main(): async def main():
router = Router("your api key") router = Router("ur key")
print(await router.get_lights()) print(await router.get_lights())
print(await router.get_scenes()) print(await router.get_scenes())
print(await router.get_devices()) print(await router.get_devices())
print(await router.get_rooms()) await router.get_rooms()
print(await router.get_zones()) await router.get_zones()
print(await router.get_bridge_homes()) await router.get_bridge_homes()
print(await router.get_grouped_lights()) await router.get_grouped_lights()
print(await router.get_bridges()) await router.get_bridges()
print(await router.get_device_powers()) await router.get_device_powers()
print(await router.get_zigbee_connectivities()) await router.get_zigbee_connectivities()
print(await router.get_zgb_connectivities()) await router.get_zgb_connectivities()
print(await router.get_motions()) print(await router.get_motions())
print(await router.get_temperatures()) print(await router.get_temperatures())
print(await router.get_light_levels()) await router.get_light_levels()
print(await router.get_buttons()) await router.get_buttons()
print(await router.get_behavior_scripts()) await router.get_behavior_scripts()
print(await router.get_behavior_instances()) await router.get_behavior_instances()
print(await router.get_geofence_clients()) await router.get_geofence_clients()
print(await router.get_geolocations()) await router.get_geolocations()
print(await router.get_entertainment_configurations()) await router.get_entertainment_configurations()
print(await router.get_entertainments()) await router.get_entertainments()
print(await router.get_homekits()) await router.get_homekits()
print(await router.get_resources()) await router.get_resources()
await router._subscribe()
while True:
await sleep(5)
run(main()) run(main())