starting point
This commit is contained in:
154
apps/daylight_adjuster.py
Executable file
154
apps/daylight_adjuster.py
Executable file
@@ -0,0 +1,154 @@
|
||||
import logging
|
||||
from contextlib import suppress
|
||||
from dataclasses import InitVar, dataclass, field
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Dict, Iterable
|
||||
|
||||
import astral
|
||||
import matplotlib.dates as mdates
|
||||
import matplotlib.pyplot as plt
|
||||
import pandas as pd
|
||||
from astral import Observer, SunDirection
|
||||
from astral.sun import elevation, sun, time_at_elevation
|
||||
from IPython.display import display
|
||||
|
||||
HOME_TZ = datetime.now().astimezone().tzinfo
|
||||
|
||||
def format_x_axis(fig):
|
||||
ax: plt.Axes = fig.axes[0]
|
||||
# ax.xaxis.set_major_locator(mdates.HourLocator(byhour=range(0, 24, 2)))
|
||||
ax.xaxis.set_major_formatter(mdates.DateFormatter('%I%p'))
|
||||
ax.grid(True)
|
||||
fig.autofmt_xdate()
|
||||
|
||||
|
||||
def normalize(s: pd.Series, min=None, max=None):
|
||||
min = min or s.min()
|
||||
max = max or s.max()
|
||||
rng = max - min
|
||||
return ((s - min) / rng) * 100
|
||||
|
||||
|
||||
def parse_periods(observer: Observer, periods: Dict):
|
||||
for period in periods:
|
||||
if 'elevation' in period:
|
||||
if period['direction'] == 'rising':
|
||||
dir = SunDirection.RISING
|
||||
elif period['direction'] == 'setting':
|
||||
dir = SunDirection.SETTING
|
||||
|
||||
if isinstance(period['elevation'], int):
|
||||
time = time_at_elevation(
|
||||
observer,
|
||||
elevation=period['elevation'],
|
||||
direction=dir,
|
||||
tzinfo=HOME_TZ,
|
||||
)
|
||||
elif period['elevation'] == 'min':
|
||||
pass
|
||||
elif period['elevation'] == 'max':
|
||||
pass
|
||||
elif 'time' in period:
|
||||
try:
|
||||
time = sun(observer, datetime.today().date())[period['time']].astimezone()
|
||||
except:
|
||||
time = datetime.combine(
|
||||
datetime.today().date(),
|
||||
datetime.strptime(period['time'], '%I:%M:%S%p').time()
|
||||
).astimezone()
|
||||
|
||||
# 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
|
||||
|
||||
|
||||
|
||||
@dataclass
|
||||
class DaylightAdjuster:
|
||||
latitude: float
|
||||
longitude: float
|
||||
periods: InitVar[Dict]
|
||||
datetime: datetime = field(default_factory=datetime.now)
|
||||
resolution: InitVar[int] = field(default=200)
|
||||
|
||||
def __post_init__(self, periods: Dict, resolution: int):
|
||||
self.logger: logging.Logger = logging.getLogger(type(self).__name__)
|
||||
today = self.datetime.date()
|
||||
self.times = pd.date_range(
|
||||
today, today + timedelta(days=1),
|
||||
periods=resolution,
|
||||
tz=HOME_TZ
|
||||
)
|
||||
# self.logger.info(
|
||||
# f'{type(self.times).__name__}:\n' +
|
||||
# '\n'.join(f' {dt}' for dt in self.times[:5]) +
|
||||
# '\n ...\n' +
|
||||
# '\n'.join(f' {dt}' for dt in self.times[-5:])
|
||||
# )
|
||||
|
||||
with suppress(UserWarning):
|
||||
self.elevation = pd.Series(
|
||||
(elevation(self.observer, dt.to_pydatetime()) for dt in self.times),
|
||||
self.times,
|
||||
name='elevation'
|
||||
)
|
||||
self.elevation.index = self.elevation.index.tz_convert(HOME_TZ)
|
||||
|
||||
self.df = pd.concat([
|
||||
pd.DataFrame(parse_periods(self.observer, periods)).set_index('time'),
|
||||
self.elevation
|
||||
], axis=1).sort_index().interpolate().bfill().ffill()
|
||||
|
||||
# self.df.index = self.df.index.to_series().dt.tz_localize(None)
|
||||
|
||||
# print('Done making daylight_adjuster')
|
||||
|
||||
@property
|
||||
def observer(self) -> astral.Observer:
|
||||
return astral.Observer(self.latitude, self.longitude)
|
||||
|
||||
@property
|
||||
def current_settings(self) -> pd.Series:
|
||||
return self.df[:datetime.now().astimezone()].iloc[0].drop('elevation').astype(int).to_dict()
|
||||
|
||||
def elevation_fig(self):
|
||||
fig, ax = plt.subplots(figsize=(10, 7))
|
||||
handles = ax.plot(self.elevation)
|
||||
|
||||
ax.set_ylabel('Elevation')
|
||||
ax.set_ylim(-100, 100)
|
||||
format_x_axis(fig)
|
||||
ax.set_xlim(self.elevation.index[0], self.elevation.index[-1])
|
||||
|
||||
ax2 = ax.twinx()
|
||||
# handles.extend(ax2.plot(normalize(self.df['brightness'], 1, 255), 'tab:orange'))
|
||||
# handles.extend(ax2.plot(normalize(self.df['color_temp'], 1, 255), 'tab:green'))
|
||||
ax2.set_ylabel('Brightness')
|
||||
ax2.set_ylim(0, 100)
|
||||
|
||||
# handles.append(ax.axvline(datetime.now(),
|
||||
# linestyle='--',
|
||||
# color='g'))
|
||||
|
||||
# handles.append(ax2.axhline(self.get_brightness(),
|
||||
# linestyle='--',
|
||||
# color='r'))
|
||||
|
||||
# handles.append(ax.axhline(self.get_elevation(),
|
||||
# linestyle='--',
|
||||
# color=handles[0].get_color()))
|
||||
|
||||
ax.legend(handles=handles, loc='lower center', labels=[
|
||||
'Sun Elevation Angle',
|
||||
'Brightness Setting',
|
||||
'Color Temp Setting',
|
||||
'Current Time',
|
||||
# 'Current Brightness',
|
||||
# 'Current Elevation'
|
||||
])
|
||||
|
||||
fig.tight_layout()
|
||||
plt.close(fig)
|
||||
return fig
|
||||
Reference in New Issue
Block a user