changed state determination

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,10 +62,10 @@ 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(':'))
@@ -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)