john/room_control
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

pydantic work

Browse Source
This commit is contained in:
John Lancaster
2024-04-02 22:30:23 -05:00
parent c0a22b63f8
commit 7f8d0311ab
2 changed files with 71 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
model.py
View File

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

57
motion.py
View File

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