From 12ac6c4cc4897fae1bf0723f8fd960da6b9df8bc Mon Sep 17 00:00:00 2001 From: John Lancaster <32917998+jsl12@users.noreply.github.com> Date: Tue, 2 Jan 2024 22:56:58 -0600 Subject: [PATCH] added RoomState and RoomConfig dataclasses --- room_control.py | 162 +++++++++++++++++++++++++++++++++--------------- 1 file changed, 111 insertions(+), 51 deletions(-) diff --git a/room_control.py b/room_control.py index 9b288e7..608e6c4 100755 --- a/room_control.py +++ b/room_control.py @@ -1,13 +1,79 @@ -import asyncio from copy import deepcopy +from dataclasses import dataclass, field from datetime import datetime, time, timedelta +from pathlib import Path from typing import Dict, List import appdaemon.utils as utils import astral +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 + + +@dataclass +class RoomState: + scene: Dict[str, Dict[str, str | int]] + off_duration: timedelta = field(default_factory=timedelta) + time: 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.strptime(self.time, fmt).time() + except: + continue + else: + break + + if self.elevation is not None: + assert self.direction is not None, f'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): + try: + hours, minutes, seconds = map(int, self.off_duration.split(':')) + self.off_duration = timedelta(hours=hours, minutes=minutes, seconds=seconds) + except Exception: + self.off_duration = timedelta() + + @classmethod + def from_json(cls, json_input): + return cls(**json_input) + + +@dataclass +class RoomConfig: + states: List[RoomState] + off_duration: timedelta = field(default_factory=timedelta) + + @classmethod + def from_app_config(cls, app_cfg: Dict[str, Dict]): + if 'class' in app_cfg: + app_cfg.pop('class') + if 'module' in app_cfg: + app_cfg.pop('module') + return cls(states=[RoomState.from_json(s) for s in app_cfg['states']]) + + @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) class RoomController(Hass, Mqtt): @@ -20,6 +86,15 @@ class RoomController(Hass, Mqtt): - When the light comes on, check if it's attributes match what they should, given the time. """ + @property + def states(self) -> List[RoomState]: + 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}' + self._room_config.states = new + async def initialize(self): self.app_entities = await self.gather_app_entities() # self.log(f'entities: {self.app_entities}') @@ -58,51 +133,42 @@ class RoomController(Hass, Mqtt): Parsed states have an absolute time for a certain day. """ # re-parse the state strings into times for the current day - self.states = await self.parse_states() + self._room_config = RoomConfig.from_app_config(self.args) + self.log(f'{len(self._room_config.states)} states in the RoomConfig') + + for state in self._room_config.states: + if state.time is None and state.elevation is not None: + state.time = self.AD.sched.location.time_at_elevation( + elevation=state.elevation, + direction=state.direction + ).time() + elif isinstance(state.time, str): + state.time = await self.parse_time(state.time) + + assert isinstance(state.time, time), f'Invalid time: {state.time}' + + for state in self.states: + self.log(f'State: {state.time.strftime("%I:%M:%S %p")} {str(state.scene)[:50]}...') + + self.states = sorted(self.states, key=lambda s: s.time, reverse=True) # schedule the transitions - for state in self.states: - t: time = state['time'] + for state in self.states[::-1]: + # t: time = state['time'] + t: time = state.time try: - await self.run_at(callback=self.activate_any_on, start=t.strftime('%H:%M:%S'), cause='scheduled transition') + await self.run_at( + callback=self.activate_any_on, + start=t.strftime('%H:%M:%S'), + cause='scheduled transition' + ) except ValueError: # happens when the callback time is in the past pass except Exception as e: self.log(f'Failed with {type(e)}: {e}') - else: - self.log(f'Scheduled transition at: {t.strftime("%I:%M:%S %p")}') - async def parse_states(self): - async def gen(): - for state in deepcopy(self.args['states']): - if (time := state.get('time')): - state['time'] = await self.parse_time(time) - - elif isinstance((elevation := state.get('elevation')), (int, float)): - assert 'direction' in state, f'State needs a direction if it has an elevation' - - if state['direction'] == 'rising': - dir = astral.SunDirection.RISING - elif state['direction'] == 'setting': - dir = astral.SunDirection.SETTING - else: - raise ValueError(f'Invalid sun direction: {state["direction"]}') - - state['time'] = self.AD.sched.location.time_at_elevation( - elevation=elevation, direction=dir - ).time() - - else: - raise ValueError(f'Missing time') - - yield state - - states = [s async for s in gen()] - states = sorted(states, key=lambda s: s['time'], reverse=True) - return states - - async def current_state(self, time: time = None): + async def current_state(self, time: time = None) -> RoomState: if (await self.sleep_bool()): self.log(f'sleep: active') if (state := self.args.get('sleep_state')): @@ -113,11 +179,11 @@ class RoomController(Hass, Mqtt): # now: datetime = await self.get_now() # self.log(f'Getting state for datetime: {now.strftime("%I:%M:%S %p")}') time = time or (await self.get_now()).time() - time_fmt = "%I:%M %p" + time_fmt = "%I:%M:%S %p" self.log(f'Getting state before: {time.strftime(time_fmt)}') for state in self.states: - time_str = state["time"].strftime(time_fmt) - if state['time'] <= time: + time_str = state.time.strftime(time_fmt) + if state.time <= time: self.log(f'Selected state from {time_str}') return state else: @@ -126,9 +192,9 @@ class RoomController(Hass, Mqtt): self.log(f'Defaulting to first state') return self.states[0] - async def current_scene(self, time: time = None): + async def current_scene(self, time: time = None) -> Dict[str, Dict[str, str | int]]: if (state := (await self.current_state(time=time))) is not None: - return state['scene'] + return state.scene async def app_entity_states(self) -> Dict[str, str]: states = { @@ -182,15 +248,9 @@ class RoomController(Hass, Mqtt): """ current_state = await self.current_state() - duration_str = current_state.get( - 'off_duration', - self.args.get('off_duration', '00:00:00') - ) - - try: - hours, minutes, seconds = map(int, duration_str.split(':')) - return timedelta(hours=hours, minutes=minutes, seconds=seconds) - except Exception: + if isinstance(current_state.off_duration, timedelta): + return current_state.off_duration + else: return timedelta() @utils.sync_wrapper