ad-nix/apps/daylight_adjuster.py

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