import logging from contextlib import suppress from dataclasses import InitVar, dataclass, field from datetime import date, datetime, time, timedelta, tzinfo from typing import Dict, Iterable import astral import matplotlib.dates as mdates import matplotlib.pyplot as plt import numpy as np 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 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 parse_periods(observer: Observer, periods: Dict, date: date): for period in periods: if 'time' in period: try: time = datetime.strptime(period['time'], '%I:%M:%S%p').time() except: sun_dict = sun(observer=observer, date=date, tzinfo=HOME_TZ) dt = sun_dict[period['time']] else: dt = datetime.combine(date, time, tzinfo=HOME_TZ) elif 'elevation' in period: if period['direction'] == 'rising': dir = SunDirection.RISING elif period['direction'] == 'setting': dir = SunDirection.SETTING assert isinstance(period['elevation'], (int, float)) dt = time_at_elevation( observer=observer, elevation=period['elevation'], date=date, direction=dir, tzinfo=HOME_TZ, ) # res = {'time': dt.replace(tzinfo=None)} # res = {'time': dt.replace(tzinfo=HOME_TZ)} res = {'time': dt} res.update({k: period[k] for k in ['brightness', 'color_temp'] if k in period}) yield res def elevation_series(observer: Observer, date, **kwargs): times = pd.date_range(start=date, end=(date + timedelta(days=1)), **kwargs) elevations = pd.Series( [elevation(observer, timestamp) for timestamp in times], index=times, name='elevation') return elevations @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__) self.period_df: pd.DataFrame = pd.DataFrame([ p for date in get_today_series() for p in parse_periods(self.observer, periods, date) ]).set_index('time').interpolate(method='index').round(0).astype(int) self.df = self.period_df.join(pd.concat([ elevation_series(self.observer, date, periods=1000, tz=HOME_TZ) for date in get_today_series() ]), how='outer') self.df = ( self.df .sort_index() .interpolate(method='index') .bfill().ffill() # .round(0).astype(int) # .drop_duplicates() ) self.df.index = self.df.index.to_series().dt.tz_localize(None) @property def observer(self) -> astral.Observer: return astral.Observer(self.latitude, self.longitude) @property def current_settings(self) -> Dict: now = datetime.now(HOME_TZ) self.period_df.loc[now] = np.nan return self.period_df.interpolate(method='index').round(0).astype(int).loc[now].to_dict() def elevation_fig(self): fig, ax = plt.subplots(figsize=(10, 7)) elevation = self.df['elevation'] elevation.index = elevation.index.to_series().dt.tz_localize(None) handles = ax.plot(elevation) ax.set_ylabel('Elevation') ax.set_ylim(-100, 100) format_x_axis(fig) ax.set_xlim(elevation.index[0], elevation.index[-1]) print(elevation) # ax.xaxis_date(HOME_TZ) ax2 = ax.twinx() handles.extend(ax2.plot(normalize(self.df['brightness'], 1, 255), 'tab:orange')) handles.extend(ax2.plot(normalize(self.df['color_temp'], 150, 650), 'tab:green')) ax2.set_ylabel('Brightness') ax2.set_ylim(0, 100) handles.append(ax.axvline(datetime.now().astimezone(), 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