From 7f8d0311abd4e669168eee20b0ed73158d724d14 Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Tue, 2 Apr 2024 22:30:23 -0500 Subject: [PATCH] pydantic work --- model.py | 51 ++++++++++++++++++++----------------------------- motion.py | 57 ++++++++++++++++++++++++++++++++++++++++++++++++------- 2 files changed, 71 insertions(+), 37 deletions(-) diff --git a/model.py b/model.py index 98f5d29..c3b0521 100644 --- a/model.py +++ b/model.py @@ -1,10 +1,10 @@ -from datetime import datetime, timedelta, time +from datetime import datetime, time, timedelta from pathlib import Path -from typing import Annotated, Dict, List, Self +from typing import Annotated, Dict, List, Optional, Self import yaml from astral import SunDirection -from pydantic import BaseModel, BeforeValidator, conint, root_validator +from pydantic import BaseModel, BeforeValidator, Field, conint, root_validator from pydantic_core import PydanticCustomError from rich.console import Console, ConsoleOptions, RenderResult from rich.table import Column, Table @@ -19,11 +19,9 @@ def str_to_timedelta(input_str: str) -> timedelta: def str_to_direction(input_str: str) -> SunDirection: - if input_str.lower() == 'setting': - return SunDirection.SETTING - elif input_str == 'rising': - return SunDirection.RISING - else: + try: + return getattr(SunDirection, input_str.upper()) + except AttributeError: raise PydanticCustomError('invalid_dir', 'Invalid sun direction: {dir}', dict(dir=input_str)) @@ -32,30 +30,28 @@ OffDuration = Annotated[timedelta, BeforeValidator(str_to_timedelta)] class State(BaseModel): state: bool = True - brightness: conint(ge=1, le=255) = None - color_temp: conint(ge=200, le=650) = None + brightness: Optional[int] = Field(default=None, ge=1, le=255) + color_temp: Optional[int] = Field(default=None, ge=200, le=650) + rgb_color: Optional[list[int]] = Field(default=None, min_length=3, max_length=3) class ApplyKwargs(BaseModel): """Arguments to call with the 'scene/apply' service""" - entities: Dict[str, State] - transition: int = None + transition: Optional[int] = None class ControllerStateConfig(BaseModel): - time: str | datetime = None - elevation: float = None - direction: Annotated[SunDirection, BeforeValidator(str_to_direction)] = None - off_duration: OffDuration = None - scene: Dict[str, State] + time: Optional[str | datetime] = None + elevation: Optional[float] = None + direction: Optional[Annotated[SunDirection, BeforeValidator(str_to_direction)]] = None + off_duration: Optional[OffDuration] = None + scene: dict[str, State] @root_validator(pre=True) def check_args(cls, values): time, elevation = values.get('time'), values.get('elevation') - if time is None and elevation is None: - raise PydanticCustomError('bad_time_spec', 'Either time or elevation must be set.') - elif time is not None and elevation is not None: + if time is not None and elevation is not None: raise PydanticCustomError('bad_time_spec', 'Only one of time or elevation can be set.') elif elevation is not None and 'direction' not in values: raise PydanticCustomError('no_sun_dir', 'Needs sun direction with elevation') @@ -66,18 +62,17 @@ class ControllerStateConfig(BaseModel): class RoomControllerConfig(BaseModel): - states: List[ControllerStateConfig] - off_duration: OffDuration = None + states: List[ControllerStateConfig] = Field(default_factory=list) + off_duration: Optional[OffDuration] = None + sleep_state: Optional[ControllerStateConfig] = None @classmethod - def from_yaml(cls: Self, yaml_path: Path): + def from_yaml(cls: Self, yaml_path: Path) -> Self: yaml_path = Path(yaml_path) with yaml_path.open('r') as f: for appname, app_cfg in yaml.load(f, Loader=yaml.SafeLoader).items(): if app_cfg['class'] == 'RoomController': - break - print(app_cfg) - return cls(**app_cfg) + return cls.model_validate(app_cfg) def __rich_console__(self, console: Console, options: ConsoleOptions) -> RenderResult: table = Table( @@ -104,15 +99,11 @@ class RoomControllerConfig(BaseModel): self.states = sorted(self.states, key=lambda s: s.time, reverse=True) def current_state(self, now: time) -> ControllerStateConfig: - # time_fmt = '%I:%M:%S %p' - # print(now.strftime(time_fmt)) - self.sort_states() for state in self.states: if state.time <= now: return state else: - # self.log(f'Defaulting to first state') return self.states[0] def current_scene(self, now: time) -> Dict: diff --git a/motion.py b/motion.py index 214031a..27a43e3 100644 --- a/motion.py +++ b/motion.py @@ -1,13 +1,29 @@ import re from datetime import timedelta +from typing import Literal, Optional from appdaemon.entity import Entity from appdaemon.plugins.hass.hassapi import Hass -from console import setup_component_logging +from console import console, setup_component_logging +from pydantic import BaseModel, TypeAdapter from room_control import RoomController +class CallbackEntry(BaseModel): + entity: str + event: Optional[str] = None + type: Literal['state', 'event'] + kwargs: str + function: str + name: str + pin_app: bool + pin_thread: int + + +Callbacks = dict[str, dict[str, CallbackEntry]] + + class Motion(Hass): @property def sensor(self) -> Entity: @@ -39,14 +55,33 @@ class Motion(Hass): ) if self.sensor_state != self.ref_entity_state: - self.log(f'Sensor is {self.sensor_state} ' f'and light is {self.ref_entity_state}', level='WARNING') + self.log( + f'Sensor is {self.sensor_state} ' f'and light is {self.ref_entity_state}', + level='WARNING', + ) if self.sensor_state: self.app.activate(kwargs={'cause': f'Syncing state with {self.sensor.entity_id}'}) # don't need to await these because they'll already get turned into a task by the utils.sync_wrapper decorator - self.listen_state(**base_kwargs, attribute='brightness', callback=self.callback_light_on,) + self.listen_state( + **base_kwargs, + attribute='brightness', + callback=self.callback_light_on, + ) self.listen_state(**base_kwargs, new='off', callback=self.callback_light_off) + if callbacks := self.callbacks(): + for handle, entry in callbacks.items(): + self.log(f'Handle [yellow]{handle[:4]}[/]: {entry.function}') + + def callbacks(self): + data = TypeAdapter(Callbacks).validate_python(self.get_callback_entries()) + name: str = self.name + try: + return data[name] + except KeyError: + return [] + def listen_motion_on(self): """Sets up the motion on callback to activate the room""" self.cancel_motion_callback() @@ -59,7 +94,9 @@ class Motion(Hass): ) self.log(f'Waiting for motion on [friendly_name]{self.sensor.friendly_name}[/]') if self.sensor_state: - self.log(f'[friendly_name]{self.sensor.friendly_name}[/] is already on', level='WARNING') + self.log( + f'[friendly_name]{self.sensor.friendly_name}[/] is already on', level='WARNING' + ) def listen_motion_off(self, duration: timedelta): """Sets up the motion off callback to deactivate the room""" @@ -72,10 +109,14 @@ class Motion(Hass): oneshot=True, cause='motion off', ) - self.log(f'Waiting for motion to stop on [friendly_name]{self.sensor.friendly_name}[/] for {duration}') + self.log( + f'Waiting for [friendly_name]{self.sensor.friendly_name}[/] to be clear for {duration}' + ) if not self.sensor_state: - self.log(f'[friendly_name]{self.sensor.friendly_name}[/] is currently off', level='WARNING') + self.log( + f'[friendly_name]{self.sensor.friendly_name}[/] is currently off', level='WARNING' + ) def callback_light_on(self, entity=None, attribute=None, old=None, new=None, kwargs=None): """Called when the light turns on""" @@ -102,7 +143,9 @@ class Motion(Hass): def get_sensor_callbacks(self): return { - handle: info for handle, info in self.get_app_callbacks().items() if info['entity'] == self.sensor.entity_id + handle: info + for handle, info in self.get_app_callbacks().items() + if info['entity'] == self.sensor.entity_id } def cancel_motion_callback(self):