diff --git a/button.py b/button.py index bba89dd..fcf513d 100644 --- a/button.py +++ b/button.py @@ -1,23 +1,27 @@ import json from dataclasses import dataclass +from typing import List from appdaemon.plugins.mqtt.mqttapi import Mqtt from console import setup_component_logging +from model import ButtonConfig from room_control import RoomController @dataclass(init=False) class Button(Mqtt): - button: str + button: str | List[str] rich: bool = False + config: ButtonConfig async def initialize(self): + self.config = ButtonConfig(**self.args) setup_component_logging(self) self.app: RoomController = await self.get_app(self.args['app']) self.log(f'Connected to AD app [room]{self.app.name}[/]', level='DEBUG') - self.button = self.args['button'] + self.button = self.config.button self.setup_buttons(self.button) def setup_buttons(self, buttons): diff --git a/model.py b/model.py index 370770d..98f5d29 100644 --- a/model.py +++ b/model.py @@ -1,10 +1,13 @@ -from datetime import datetime, timedelta +from datetime import datetime, timedelta, time from pathlib import Path from typing import Annotated, Dict, List, Self import yaml from astral import SunDirection -from pydantic import BaseModel, BeforeValidator, ValidationError, conint, root_validator +from pydantic import BaseModel, BeforeValidator, conint, root_validator +from pydantic_core import PydanticCustomError +from rich.console import Console, ConsoleOptions, RenderResult +from rich.table import Column, Table def str_to_timedelta(input_str: str) -> timedelta: @@ -13,7 +16,7 @@ def str_to_timedelta(input_str: str) -> timedelta: return timedelta(hours=hours, minutes=minutes, seconds=seconds) except Exception: return timedelta() - + def str_to_direction(input_str: str) -> SunDirection: if input_str.lower() == 'setting': @@ -21,7 +24,7 @@ def str_to_direction(input_str: str) -> SunDirection: elif input_str == 'rising': return SunDirection.RISING else: - raise ValidationError(f'Invalid sun direction: {input_str}') + raise PydanticCustomError('invalid_dir', 'Invalid sun direction: {dir}', dict(dir=input_str)) OffDuration = Annotated[timedelta, BeforeValidator(str_to_timedelta)] @@ -51,18 +54,18 @@ class ControllerStateConfig(BaseModel): def check_args(cls, values): time, elevation = values.get('time'), values.get('elevation') if time is None and elevation is None: - raise ValueError('Either time or elevation must be set.') + raise PydanticCustomError('bad_time_spec', 'Either time or elevation must be set.') elif time is not None and elevation is not None: - raise ValueError('Only one of time or elevation can be set.') - elif elevation is not None: - assert 'direction' in values + 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') return values - def to_apply_kwargs(self, transition: int = 0): - return ApplyKwargs(entities=self.scene, transition=transition).model_dump(exclude_none=True) + def to_apply_kwargs(self, **kwargs): + return ApplyKwargs(entities=self.scene, **kwargs).model_dump(exclude_none=True) -class RoomConfig(BaseModel): +class RoomControllerConfig(BaseModel): states: List[ControllerStateConfig] off_duration: OffDuration = None @@ -75,3 +78,59 @@ class RoomConfig(BaseModel): break print(app_cfg) return cls(**app_cfg) + + def __rich_console__(self, console: Console, options: ConsoleOptions) -> RenderResult: + table = Table( + Column('Time', width=15), + Column('Scene'), + highlight=True, + padding=1, + collapse_padding=True, + ) + for state in self.states: + scene_json = state.to_apply_kwargs() + lines = [ + f'{name:20}{state["state"]} Brightness: {state["brightness"]:<4} Temp: {state["color_temp"]}' + for name, state in scene_json['entities'].items() + ] + table.add_row(state.time.strftime('%I:%M:%S %p'), '\n'.join(lines)) + yield table + + def sort_states(self): + """Should only be called after all the times have been resolved""" + assert all( + isinstance(state.time, time) for state in self.states + ), 'Times have not all been resolved yet' + 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: + state = self.current_state(now) + return state.scene + + def current_off_duration(self, now: time) -> timedelta: + state = self.current_state(now) + if state.off_duration is None: + if self.off_duration is None: + raise ValueError('Need an off duration') + else: + return self.off_duration + else: + return state.off_duration + + +class ButtonConfig(BaseModel): + app: str + button: str | List[str] + ref_entity: str diff --git a/room_control.py b/room_control.py index 209a217..ab6df81 100755 --- a/room_control.py +++ b/room_control.py @@ -1,174 +1,17 @@ import datetime import logging from copy import deepcopy -from dataclasses import dataclass, field -from pathlib import Path from typing import Dict, List -import yaml from appdaemon.entity import Entity from appdaemon.plugins.hass.hassapi import Hass from appdaemon.plugins.mqtt.mqttapi import Mqtt -from astral import SunDirection from console import console, setup_handler -from model import ApplyKwargs -from rich.console import Console, ConsoleOptions, RenderResult -from rich.table import Column, Table - - -def str_to_timedelta(input_str: str) -> datetime.timedelta: - try: - hours, minutes, seconds = map(int, input_str.split(':')) - return datetime.timedelta(hours=hours, minutes=minutes, seconds=seconds) - except Exception: - return datetime.timedelta() - +from model import ControllerStateConfig, RoomControllerConfig logger = logging.getLogger(__name__) -@dataclass -class RoomState: - scene: Dict[str, Dict[str, str | int]] - off_duration: datetime.timedelta = None - time: datetime.time = None - time_fmt: List[str] = field(default_factory=lambda: ['%H:%M:%S', '%I:%M:%S %p'], repr=False) - elevation: int | float = None - direction: SunDirection = None - - def __post_init__(self): - if isinstance(self.time, str): - for fmt in self.time_fmt: - try: - self.time = datetime.datetime.strptime(self.time, fmt).time() - except Exception: - continue - else: - break - - if self.elevation is not None: - assert self.direction is not None, 'Elevation setting requires a direction' - if self.direction.lower() == 'setting': - self.direction = SunDirection.SETTING - elif self.direction.lower() == 'rising': - self.direction = SunDirection.RISING - else: - raise ValueError(f'Invalid sun direction: {self.direction}') - - if isinstance(self.elevation, str): - self.elevation = float(self.elevation) - - if isinstance(self.off_duration, str): - self.off_duration = str_to_timedelta(self.off_duration) - - @classmethod - def from_json(cls, json_input): - return cls(**json_input) - - def __rich_console__(self, console: Console, options: ConsoleOptions) -> RenderResult: - table = Table( - Column('Entity ID', width=15), - Column('State'), - highlight=True, - padding=1, - collapse_padding=True, - ) - for name, state in self.scene.items(): - table.add_row(name, ApplyKwargs(entites=self.scene).model_dump(exclude_none=True)) - yield table - - def scene_model(self) -> ApplyKwargs: - return ApplyKwargs( - entities=self.scene, - transition=0 - ).model_dump(exclude_none=True) - - -@dataclass -class RoomConfig: - states: List[RoomState] - off_duration: datetime.timedelta = None - - def __post_init__(self): - if isinstance(self.off_duration, str): - self.off_duration = str_to_timedelta(self.off_duration) - - @classmethod - def from_app_config(cls, app_cfg: Dict[str, Dict]): - if 'off_duration' in app_cfg: - kwargs = {'off_duration': app_cfg['off_duration']} - else: - kwargs = {} - - self = cls(states=[RoomState.from_json(s) for s in app_cfg['states']], **kwargs) - - return self - - @classmethod - def from_yaml(cls, yaml_path: Path, app_name: str): - with yaml_path.open('r') as f: - cfg: Dict = yaml.load(f, Loader=yaml.SafeLoader)[app_name] - return cls.from_app_config(cfg) - - def __rich_console__(self, console: Console, options: ConsoleOptions) -> RenderResult: - table = Table( - Column('Time', width=15), - Column('Scene'), - highlight=True, - padding=1, - collapse_padding=True, - ) - for state in self.states: - scene_json = ApplyKwargs(entities=state.scene).model_dump(exclude_none=True) - lines = [ - f'{name:20}{state["state"]} Brightness: {state["brightness"]:<4} Temp: {state["color_temp"]}' - for name, state in scene_json['entities'].items() - ] - table.add_row(state.time.strftime('%I:%M:%S %p'), '\n'.join(lines)) - yield table - - def rich_table(self, app_name: str) -> Table: - table = Table(title=app_name, expand=True, highlight=True, padding=1, collapse_padding=True) - table.add_column('Time') - table.add_column('Scene') - for state in self.states: - table.add_row(state.time.strftime('%I:%M:%S %p'), str(state.scene)) - return table - - def sort_states(self): - """Should only be called after all the times have been resolved""" - assert all( - isinstance(state.time, datetime.time) for state in self.states - ), 'Times have not all been resolved yet' - self.states = sorted(self.states, key=lambda s: s.time, reverse=True) - - def current_state(self, now: datetime.time) -> RoomState: - # 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: datetime.time) -> Dict: - state = self.current_state(now) - return state.scene - - def current_off_duration(self, now: datetime.time) -> datetime.timedelta: - state = self.current_state(now) - if state.off_duration is None: - if self.off_duration is None: - raise ValueError('Need an off duration') - else: - return self.off_duration - else: - return state.off_duration - - class RoomController(Hass, Mqtt): """Class for linking room's lights with a motion sensor. @@ -180,12 +23,12 @@ class RoomController(Hass, Mqtt): """ @property - def states(self) -> List[RoomState]: + def states(self) -> List[ControllerStateConfig]: return self._room_config.states @states.setter - def states(self, new: List[RoomState]): - assert all(isinstance(s, RoomState) for s in new), f'Invalid: {new}' + def states(self, new: List[ControllerStateConfig]): + assert all(isinstance(s, ControllerStateConfig) for s in new), f'Invalid: {new}' self._room_config.states = new def initialize(self): @@ -233,8 +76,8 @@ class RoomController(Hass, Mqtt): Parsed states have an absolute time for the current day. """ # re-parse the state strings into times for the current day - self._room_config = RoomConfig.from_app_config(self.args) - self.log(f'{len(self._room_config.states)} states in the RoomConfig', level='DEBUG') + self._room_config = RoomControllerConfig(**self.args) + self.log(f'{len(self._room_config.states)} states in the app configuration', level='DEBUG') for state in self._room_config.states: if state.time is None and state.elevation is not None: @@ -264,35 +107,21 @@ class RoomController(Hass, Mqtt): except Exception as e: self.log(f'Failed with {type(e)}: {e}') - def current_state(self, now: datetime.time = None) -> RoomState: + def current_state(self, now: datetime.time = None) -> ControllerStateConfig: if self.sleep_bool(): self.log('sleep: active') if state := self.args.get('sleep_state'): - return RoomState.from_json(state) + return ControllerStateConfig(**state) else: - return RoomState(scene={}) + return ControllerStateConfig(scene={}) else: - now = now or self.get_now().time() - self.log(f'Getting state for {now}', level='DEBUG') + now = now or self.get_now().time().replace(microsecond=0) + self.log(f'Getting state for {now.strftime("%I:%M:%S %p")}', level='DEBUG') state = self._room_config.current_state(now) - - if self.logger.isEnabledFor(logging.DEBUG): - self.log('Current state', level='DEBUG') - console.print(state) - + self.log(f'Current state: {state.time}', level='DEBUG') return state - def current_scene(self, now: datetime.time = None) -> Dict[str, Dict[str, str | int]]: - state = self.current_state(now) - # needed this way instead of isinstance(...) for the reloading to work - assert type(state).__name__ == 'RoomState' - - if self.logger.isEnabledFor(logging.DEBUG): - self.log('Current scene:', level='DEBUG') - console.print(state) - return state.scene - def app_entity_states(self) -> Dict[str, str]: states = {entity: self.get_state(entity) for entity in self.app_entities} return states @@ -356,26 +185,24 @@ class RoomController(Hass, Mqtt): cause = 'unknown' self.log(f'Activating: {cause}') - scene = self.current_scene() + scene_kwargs = self.current_state().to_apply_kwargs(transition=0) - if isinstance(scene, str): - self.turn_on(scene) - self.log(f'Turned on scene: {scene}') + if isinstance(scene_kwargs, str): + self.turn_on(scene_kwargs) + self.log(f'Turned on scene: {scene_kwargs}') - elif isinstance(scene, dict): - kwargs = ApplyKwargs(entities=scene, transition=0).model_dump(exclude_none=True) - self.log('Validated scene JSON', level='DEBUG') - self.call_service('scene/apply', **kwargs) + elif isinstance(scene_kwargs, dict): + self.call_service('scene/apply', **scene_kwargs) if self.logger.isEnabledFor(logging.INFO): self.log('Applied scene:') - console.print(scene) + console.print(scene_kwargs['entities']) - elif scene is None: + elif scene_kwargs is None: self.log('No scene, ignoring...') # Need to act as if the light had just turned off to reset the motion (and maybe other things?) # self.callback_light_off() else: - self.log(f'ERROR: unknown scene: {scene}') + self.log(f'ERROR: unknown scene: {scene_kwargs}') def activate_all_off(self, *args, **kwargs): """Activate if all of the entities are off. Args and kwargs are passed directly to self.activate()"""