-
Notifications
You must be signed in to change notification settings - Fork 0
/
weather.py
135 lines (104 loc) · 4.27 KB
/
weather.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
import datetime
import re
from dataclasses import dataclass
import pandas as pd
import requests
import constants
import models
from constants import DARKSKY_API_KEY
from preferences import DefaultPreferences
camel_case_pattern = re.compile(r'(?<!^)(?=[A-Z])')
name_mapper = lambda x: camel_case_pattern.sub('_', x).lower()
class DarkSky:
columns = [
"weather_timestamp",
"recorded_timestamp",
"summary",
"icon",
"precipIntensity",
"precipProbability",
"precipType",
"temperature",
"apparentTemperature",
"dewPoint",
"humidity",
"pressure",
"windSpeed",
"windGust",
"windBearing",
"cloudCover",
"uvIndex",
"visibility",
"ozone",
]
def df_template(self):
""" Returns basic dataframe with DarkSky columns."""
return pd.DataFrame(columns=self.columns)
def post_process(self, df):
""" Post-process df which comes from a DarkSky API call"""
df["weather_timestamp"] = df["time"].apply(lambda x: datetime.datetime.fromtimestamp(x))
df["recorded_timestamp"] = datetime.datetime.now()
df.sort_values(by=["time"], inplace=True)
df = df[self.columns]
df.rename(axis=1, mapper=name_mapper, inplace=True)
return df
@staticmethod
def _filter_for_tomorrow(df: pd.DataFrame):
today = datetime.datetime.now().date()
df['weather_date'] = df['weather_timestamp'].apply(lambda x: x.date())
df = df.loc[df['weather_date'] == today + datetime.timedelta(days=1)]
df.drop('weather_date', axis=1, inplace=True)
return df
# TODO: refactor this to avoid pandas at all, will be faster
def get_forecast_tomorrow(self, longitude: str, latitude: str):
df = self.df_template()
r = requests.get(
f"https://api.darksky.net/forecast/{DARKSKY_API_KEY}/{latitude},{longitude}?exclude=[currently, minutely]",
params={"units": "si"},
)
j = r.json()
for h in j["hourly"]["data"]:
df = df.append(h, ignore_index=True)
df = self.post_process(df)
df = self._filter_for_tomorrow(df)
return df
@dataclass
class Preferences:
temperature_weighting: dict
weightings: dict
day_start: int = DefaultPreferences.day_start
day_end: int = DefaultPreferences.day_end
temperature: str = DefaultPreferences.temperature
def __post_init__(self):
self.weightings['apparent_temperature'] = self.temperature_weighting[self.temperature]
class WeatherWindowFinder:
def __init__(self,
preferences: Preferences):
self.weightings = preferences.weightings
self.possible_window_start = preferences.day_start
self.possible_window_end = preferences.day_end
def filter_for_latest_forecasts(self, df: pd.DataFrame):
latest_forecast = df['recorded_timestamp'].max()
return df.loc[df['recorded_timestamp'] == latest_forecast]
def filter_for_working_day(self, df: pd.DataFrame):
df['hours'] = df['weather_timestamp'].apply(lambda x: x.hour)
return df.loc[(df['hours'] > self.possible_window_start) &
(df['hours'] < self.possible_window_end)]
def get_weather_window_for_forecast(self, df: pd.DataFrame):
df = self.filter_for_latest_forecasts(df)
df = self.filter_for_working_day(df)
for columns_to_score, weight in self.weightings.items():
df[columns_to_score + '_score'] = df[columns_to_score] * weight
df['total_score'] = df.filter(like='_score').sum(axis=1)
return df.loc[df['total_score'].idxmax()]
def convert_db_preferences_to_weather_preferences(preferences: models.Preferences):
return Preferences(temperature_weighting=constants.TEMPERATURE_WEIGHTINGS,
weightings=constants.DEFAULT_WEIGHTINGS,
day_start=preferences.day_start,
day_end=preferences.day_end,
temperature=preferences.temperature)
if __name__ == "__main__":
location = dict(place="Rumney NH", latitude=45.5017, longitude=-73.5673)
forecast = DarkSky()
df = forecast.get_forecast_tomorrow(location['longitude'], location['latitude'])
print(df)