/
weather.py
200 lines (161 loc) · 5.8 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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
"""Support for Buienradar.nl weather service."""
import logging
import voluptuous as vol
from homeassistant.components.weather import (
ATTR_FORECAST_CONDITION,
ATTR_FORECAST_TEMP,
ATTR_FORECAST_TEMP_LOW,
ATTR_FORECAST_TIME,
PLATFORM_SCHEMA,
WeatherEntity,
ATTR_FORECAST_PRECIPITATION,
ATTR_FORECAST_WIND_BEARING,
ATTR_FORECAST_WIND_SPEED,
)
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME, TEMP_CELSIUS
from homeassistant.helpers import config_validation as cv
# Reuse data and API logic from the sensor implementation
from .sensor import BrData
_LOGGER = logging.getLogger(__name__)
DATA_CONDITION = "buienradar_condition"
DEFAULT_TIMEFRAME = 60
CONF_FORECAST = "forecast"
CONDITION_CLASSES = {
"cloudy": ["c", "p"],
"fog": ["d", "n"],
"hail": [],
"lightning": ["g"],
"lightning-rainy": ["s"],
"partlycloudy": ["b", "j", "o", "r"],
"pouring": ["l", "q"],
"rainy": ["f", "h", "k", "m"],
"snowy": ["u", "i", "v", "t"],
"snowy-rainy": ["w"],
"sunny": ["a"],
"windy": [],
"windy-variant": [],
"exceptional": [],
}
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Optional(CONF_NAME): cv.string,
vol.Optional(CONF_LATITUDE): cv.latitude,
vol.Optional(CONF_LONGITUDE): cv.longitude,
vol.Optional(CONF_FORECAST, default=True): cv.boolean,
}
)
async def async_setup_platform(hass, config, async_add_entities, discovery_info=None):
"""Set up the buienradar platform."""
latitude = config.get(CONF_LATITUDE, hass.config.latitude)
longitude = config.get(CONF_LONGITUDE, hass.config.longitude)
if None in (latitude, longitude):
_LOGGER.error("Latitude or longitude not set in Home Assistant config")
return False
coordinates = {CONF_LATITUDE: float(latitude), CONF_LONGITUDE: float(longitude)}
# create weather data:
data = BrData(hass, coordinates, DEFAULT_TIMEFRAME, None)
# create weather device:
_LOGGER.debug("Initializing buienradar weather: coordinates %s", coordinates)
# create condition helper
if DATA_CONDITION not in hass.data:
cond_keys = [str(chr(x)) for x in range(97, 123)]
hass.data[DATA_CONDITION] = dict.fromkeys(cond_keys)
for cond, condlst in CONDITION_CLASSES.items():
for condi in condlst:
hass.data[DATA_CONDITION][condi] = cond
async_add_entities([BrWeather(data, config)])
# schedule the first update in 1 minute from now:
await data.schedule_update(1)
class BrWeather(WeatherEntity):
"""Representation of a weather condition."""
def __init__(self, data, config):
"""Initialise the platform with a data instance and station name."""
self._stationname = config.get(CONF_NAME, None)
self._forecast = config.get(CONF_FORECAST)
self._data = data
@property
def attribution(self):
"""Return the attribution."""
return self._data.attribution
@property
def name(self):
"""Return the name of the sensor."""
return self._stationname or "BR {}".format(
self._data.stationname or "(unknown station)"
)
@property
def condition(self):
"""Return the current condition."""
from buienradar.constants import CONDCODE
if self._data and self._data.condition:
ccode = self._data.condition.get(CONDCODE)
if ccode:
conditions = self.hass.data.get(DATA_CONDITION)
if conditions:
return conditions.get(ccode)
@property
def temperature(self):
"""Return the current temperature."""
return self._data.temperature
@property
def pressure(self):
"""Return the current pressure."""
return self._data.pressure
@property
def humidity(self):
"""Return the name of the sensor."""
return self._data.humidity
@property
def visibility(self):
"""Return the current visibility in km."""
if self._data.visibility is None:
return None
return round(self._data.visibility / 1000, 1)
@property
def wind_speed(self):
"""Return the current windspeed in km/h."""
if self._data.wind_speed is None:
return None
return round(self._data.wind_speed * 3.6, 1)
@property
def wind_bearing(self):
"""Return the current wind bearing (degrees)."""
return self._data.wind_bearing
@property
def temperature_unit(self):
"""Return the unit of measurement."""
return TEMP_CELSIUS
@property
def forecast(self):
"""Return the forecast array."""
from buienradar.constants import (
CONDITION,
CONDCODE,
RAIN,
DATETIME,
MIN_TEMP,
MAX_TEMP,
WINDAZIMUTH,
WINDSPEED,
)
if not self._forecast:
return None
fcdata_out = []
cond = self.hass.data[DATA_CONDITION]
if not self._data.forecast:
return None
for data_in in self._data.forecast:
# remap keys from external library to
# keys understood by the weather component:
condcode = data_in.get(CONDITION, []).get(CONDCODE)
data_out = {
ATTR_FORECAST_TIME: data_in.get(DATETIME),
ATTR_FORECAST_CONDITION: cond[condcode],
ATTR_FORECAST_TEMP_LOW: data_in.get(MIN_TEMP),
ATTR_FORECAST_TEMP: data_in.get(MAX_TEMP),
ATTR_FORECAST_PRECIPITATION: data_in.get(RAIN),
ATTR_FORECAST_WIND_BEARING: data_in.get(WINDAZIMUTH),
ATTR_FORECAST_WIND_SPEED: round(data_in.get(WINDSPEED) * 3.6, 1),
}
fcdata_out.append(data_out)
return fcdata_out