WIP adjuster
This commit is contained in:
John Lancaster
@@ -1,12 +1,13 @@
|
||||
import logging
|
||||
from contextlib import suppress
|
||||
from dataclasses import InitVar, dataclass, field
|
||||
from datetime import datetime, timedelta, tzinfo, date, time
|
||||
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
|
||||
@@ -56,53 +57,16 @@ def times_at_elevation(observer: Observer, elevation, direction, days = None):
|
||||
return df
|
||||
|
||||
|
||||
def get_next_time_at_elevation(*args, **kwargs):
|
||||
time = time_at_elevation(*args, **kwargs)
|
||||
if time < (now := datetime.now(HOME_TZ)):
|
||||
time = time_at_elevation(
|
||||
*args,
|
||||
date=(now + timedelta(days=1)).date(),
|
||||
**kwargs
|
||||
)
|
||||
return time
|
||||
|
||||
|
||||
def get_next_sun_time(named_time: str, observer: Observer):
|
||||
sun_times_dict = sun(observer, datetime.today().date())
|
||||
try:
|
||||
time = sun_times_dict[named_time].astimezone()
|
||||
except KeyError:
|
||||
time = datetime.combine(
|
||||
datetime.today().date(),
|
||||
datetime.strptime(named_time, '%I:%M:%S%p').time()
|
||||
).astimezone()
|
||||
|
||||
if time < (now := datetime.now(HOME_TZ)):
|
||||
tomorrow = (now + timedelta(days=1)).date()
|
||||
sun_times_dict = sun(observer, tomorrow)
|
||||
try:
|
||||
time = sun_times_dict[named_time].astimezone()
|
||||
except KeyError:
|
||||
time = datetime.combine(
|
||||
tomorrow,
|
||||
datetime.strptime(named_time, '%I:%M:%S%p').time()
|
||||
).astimezone()
|
||||
|
||||
return time
|
||||
|
||||
|
||||
def parse_periods(observer: Observer, periods: Dict, date: date):
|
||||
now = datetime.now(HOME_TZ)
|
||||
|
||||
for period in periods:
|
||||
if 'time' in period:
|
||||
try:
|
||||
time = datetime.strptime(period['time'], '%I:%M:%S%p')
|
||||
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)
|
||||
dt = datetime.combine(date, time, tzinfo=HOME_TZ)
|
||||
|
||||
elif 'elevation' in period:
|
||||
if period['direction'] == 'rising':
|
||||
@@ -126,6 +90,14 @@ def parse_periods(observer: Observer, periods: Dict, date: date):
|
||||
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
|
||||
@@ -136,40 +108,37 @@ class DaylightAdjuster:
|
||||
|
||||
def __post_init__(self, periods: Dict, resolution: int):
|
||||
self.logger: logging.Logger = logging.getLogger(type(self).__name__)
|
||||
now = datetime.now().astimezone()
|
||||
times = pd.date_range(
|
||||
start=now, end=now + timedelta(days=1),
|
||||
periods=resolution,
|
||||
tz=HOME_TZ
|
||||
|
||||
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.logger.debug(times)
|
||||
|
||||
pytimes = [dt.to_pydatetime() for dt in times]
|
||||
|
||||
el = pd.Series(
|
||||
(elevation(self.observer, dt) for dt in pytimes),
|
||||
index=times, name='elevation'
|
||||
)
|
||||
self.logger.debug(el)
|
||||
|
||||
# el.index = el.index.tz_convert(HOME_TZ)
|
||||
# el.index = el.index.tz_convert(None)
|
||||
|
||||
self.periods = parse_periods(self.observer, periods)
|
||||
|
||||
# self.df = pd.DataFrame(el)
|
||||
self.df = pd.concat([pd.DataFrame(self.periods).set_index('time'), el], axis=1)
|
||||
self.df = self.df.sort_index().interpolate().bfill().ffill()
|
||||
|
||||
# self.df.index = self.df.index.to_series().dt.tz_localize(None)
|
||||
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) -> pd.Series:
|
||||
return self.df[:datetime.now().astimezone()].iloc[-1].drop('elevation').astype(int).to_dict()
|
||||
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))
|
||||
@@ -191,9 +160,9 @@ class DaylightAdjuster:
|
||||
ax2.set_ylabel('Brightness')
|
||||
ax2.set_ylim(0, 100)
|
||||
|
||||
# handles.append(ax.axvline(datetime.now().astimezone(),
|
||||
# linestyle='--',
|
||||
# color='g'))
|
||||
handles.append(ax.axvline(datetime.now().astimezone(),
|
||||
linestyle='--',
|
||||
color='g'))
|
||||
|
||||
# handles.append(ax2.axhline(self.get_brightness(),
|
||||
# linestyle='--',
|
||||
|
||||
Reference in New Issue
Block a user