WIP

apps/daylight_adjuster.py

@@ -1,7 +1,7 @@
 import logging
 from contextlib import suppress
 from dataclasses import InitVar, dataclass, field
-from datetime import datetime, timedelta, tzinfo
+from datetime import datetime, timedelta, tzinfo, date, time
 from typing import Dict, Iterable
 
 import astral
@@ -30,6 +30,32 @@ def normalize(s: pd.Series, min=None, max=None):
     return ((s - min) / rng) * 100
 
 
+def get_today_series():
+    days = pd.date_range(start=datetime.today() - timedelta(days=1), periods=3, freq='1D')
+    days = days.to_series().dt.date
+    return days.values
+
+
+def times_at_elevation(observer: Observer, elevation, direction, days = None):
+    kwargs = dict(
+        observer=observer,
+        elevation=elevation,
+        direction=direction,
+        tzinfo=HOME_TZ
+    )
+
+    days = days if days is not None else get_today_series()
+
+    df = pd.DataFrame(pd.Series(
+        data=[time_at_elevation(date=day, **kwargs) for day in days],
+        index=days,
+        name='time_at_elevation'
+    ))
+    df['elevation'] = elevation
+    df['direction'] = direction
+    return df
+
+
 def get_next_time_at_elevation(*args, **kwargs):
     time = time_at_elevation(*args, **kwargs)
     if time < (now := datetime.now(HOME_TZ)):
@@ -65,34 +91,37 @@ def get_next_sun_time(named_time: str, observer: Observer):
     return time
 
 
-def parse_periods(observer: Observer, periods: Dict):
+def parse_periods(observer: Observer, periods: Dict, date: date):
     now = datetime.now(HOME_TZ)
 
     for period in periods:
-        if 'elevation' in period:
+        if 'time' in period:
+            try:
+                time = datetime.strptime(period['time'], '%I:%M:%S%p')
+            except:
+                sun_dict = sun(observer=observer, date=date, tzinfo=HOME_TZ)
+                dt = sun_dict[period['time']]
+            else:
+                dt = datetime.combine(date, time)
+
+        elif 'elevation' in period:
             if period['direction'] == 'rising':
                 dir = SunDirection.RISING
             elif period['direction'] == 'setting':
                 dir = SunDirection.SETTING
 
-            if isinstance(period['elevation'], int):
+            assert isinstance(period['elevation'], (int, float))
-                time = get_next_time_at_elevation(
+            dt = time_at_elevation(
-                    observer,
+                observer=observer,
-                    elevation=period['elevation'],
+                elevation=period['elevation'],
-                    direction=dir,
+                date=date,
-                    tzinfo=HOME_TZ,
+                direction=dir,
-                )
+                tzinfo=HOME_TZ,
+            )
 
-            elif period['elevation'] == 'min':
+        # res = {'time': dt.replace(tzinfo=None)}
-                pass
+        # res = {'time': dt.replace(tzinfo=HOME_TZ)}
-            elif period['elevation'] == 'max':
+        res = {'time': dt}
-                pass
-        elif 'time' in period:
-            time = get_next_sun_time(period['time'], observer)
-
-        # res = {'time': time.replace(tzinfo=None)}
-        # res = {'time': time.replace(tzinfo=HOME_TZ)}
-        res = {'time': time}
         res.update({k: period[k] for k in ['brightness', 'color_temp'] if k in period})
         yield res