diff --git a/apps/basic_motion.py b/apps/basic_motion.py index 1e90375..77c55c6 100755 --- a/apps/basic_motion.py +++ b/apps/basic_motion.py @@ -2,6 +2,7 @@ from copy import deepcopy from datetime import time, timedelta from typing import Dict, List, Tuple, Union +import astral from appdaemon.plugins.hass.hassapi import Hass from continuous import Continuous @@ -18,6 +19,7 @@ class MotionLight(Hass): def initialize(self): self.state_change_handle = self.listen_state(self.state_change, self.entity) + self.states = self.parse_states() self.sync_state() self.schedule_transitions() self.run_daily(callback=self.schedule_transitions, start='00:00:00') @@ -60,11 +62,11 @@ class MotionLight(Hass): """ - if self.sleeping_active: - _, duration_str = self.sleep_scene() - else: - _, _, duration_str = self.current_setting() - + duration_str = self.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) @@ -113,58 +115,48 @@ class MotionLight(Hass): def is_stateful(self): return 'scene' in self.args and isinstance(self.args['scene'], (list, dict)) - def settings(self) -> List[Tuple[time, Union[Dict, str], timedelta]]: - """Gets the settings for all the scenes based on time of day + def parse_states(self): + def gen(): + for state in deepcopy(self.args['scene']): + if (time := state.get('time')): + state['time'] = self.parse_time(time) - "Settings" refers to `tuple` groups that consist of - - Scene start time - - Dictionary of states or scene entity name - - Time for motion to be off + elif isinstance((elevation := state.get('elevation')), (int, float)): + assert 'direction' in state, f'State needs a direction if it has an elevation' - Returns: - List[Tuple[time, Union[Dict, str], timedelta]]: Sorted list of settings - """ - assert self.is_stateful - return sorted( - (( - self.parse_time(s['time']), - s['scene'], - s.get('off_duration', self.args.get('off_duration', '00:00:00')) - ) - for s in self.args['scene']), - ) + 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"]}') - def current_setting(self) -> Tuple[time, Dict, timedelta]: - assert self.is_stateful - for dt, scene, off_duration in self.settings()[::-1]: - if dt <= self.time(): - # self.log(f'Active scene: {str(self.time())[:8]} {str(dt)[:8]}, {scene}, {off_duration}') - return dt, scene, off_duration - else: - self.log('Setting last scene') - return self.settings()[-1] + state['time'] = self.AD.sched.location.time_at_elevation( + elevation=elevation, direction=dir + ).time() - def current_scene(self) -> Union[str, Dict]: + else: + raise ValueError(f'Missing time') + + yield state + + states = sorted(gen(), key=lambda s: s['time']) + return states + + def current_state(self, time: time = None): if self.sleeping_active: - return self.sleep_scene()[0] + if (state := self.args.get['sleep_state']): + return state else: - if self.is_stateful: - dt, scene, _ = self.current_setting() - self.log(f'Current scene: {str(dt)[:8]}, {scene}') - return scene + time = time or self.get_now().time() + for state in self.states[::-1]: + if state['time'] <= time: + return state else: - return self.args['scene'] + return self.states[-1] - def sleep_scene(self) -> Tuple[Dict, timedelta]: - if (scene := self.args.get('sleep_scene')): - scene = deepcopy(scene) - if isinstance(scene, dict): - off_duration = scene.pop('off_duration', '00:00:00') - else: - off_duration = '00:00:00' - return scene, off_duration - else: - return None, None + def current_scene(self, time: time = None): + return self.current_state(time=time)['scene'] @property def app_entities(self): @@ -354,9 +346,8 @@ class MotionLight(Hass): self.log(f'Turned off {entity}') def schedule_transitions(self, *args, **kwargs): - # times, scenes, offs = zip(*self.settings()) - for dt, scene, off_duration in self.settings(): - dt = str(dt)[:8] + for state in self.states: + dt = str(state['time'])[:8] self.log(f'Scheduling transition at: {dt}') try: self.run_at(callback=self.activate_any_on, start=dt)