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 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309
|
"""AEMET OpenData Station."""
from datetime import datetime
from typing import Any
from zoneinfo import ZoneInfo
from .const import (
AEMET_ATTR_IDEMA,
AEMET_ATTR_STATION_ALTITUDE,
AEMET_ATTR_STATION_DATE,
AEMET_ATTR_STATION_DEWPOINT,
AEMET_ATTR_STATION_HUMIDITY,
AEMET_ATTR_STATION_LATITUDE,
AEMET_ATTR_STATION_LOCATION,
AEMET_ATTR_STATION_LONGITUDE,
AEMET_ATTR_STATION_PRECIPITATION,
AEMET_ATTR_STATION_PRESSURE,
AEMET_ATTR_STATION_PRESSURE_SEA,
AEMET_ATTR_STATION_TEMPERATURE,
AEMET_ATTR_STATION_TEMPERATURE_MAX,
AEMET_ATTR_STATION_TEMPERATURE_MIN,
AEMET_ATTR_STATION_WIND_DIRECTION,
AEMET_ATTR_STATION_WIND_SPEED,
AEMET_ATTR_STATION_WIND_SPEED_MAX,
AOD_ALTITUDE,
AOD_COORDS,
AOD_DATETIME,
AOD_DEW_POINT,
AOD_DISTANCE,
AOD_HUMIDITY,
AOD_ID,
AOD_NAME,
AOD_OUTDATED,
AOD_PRECIPITATION,
AOD_PRESSURE,
AOD_TEMP,
AOD_TEMP_MAX,
AOD_TEMP_MIN,
AOD_TIMESTAMP_UTC,
AOD_TIMEZONE,
AOD_WIND_DIRECTION,
AOD_WIND_SPEED,
AOD_WIND_SPEED_MAX,
ATTR_DATA,
ATTR_DISTANCE,
STATION_MAX_DELTA,
)
from .helpers import get_current_datetime, parse_api_timestamp, timezone_from_coords
class Station:
"""AEMET OpenData Station."""
altitude: float
coords: tuple[float, float]
_datetime: datetime
distance: float
dew_point: float | None = None
humidity: float | None = None
id: str
name: str
precipitation: float | None = None
pressure: float | None = None
temp: float | None = None
temp_max: float | None = None
temp_min: float | None = None
wind_direction: float | None = None
wind_speed: float | None = None
wind_speed_max: float | None = None
zoneinfo: ZoneInfo
def __init__(self, data: dict[str, Any]) -> None:
"""Init AEMET OpenData Station."""
self.altitude = float(data[AEMET_ATTR_STATION_ALTITUDE])
self.coords = (
float(data[AEMET_ATTR_STATION_LATITUDE]),
float(data[AEMET_ATTR_STATION_LONGITUDE]),
)
self.distance = float(data[ATTR_DISTANCE])
self.id = str(data[AEMET_ATTR_IDEMA])
self.name = str(data[AEMET_ATTR_STATION_LOCATION])
self.zoneinfo = timezone_from_coords(self.coords)
self.update_sample(data)
def get_altitude(self) -> float:
"""Return Station altitude."""
return self.altitude
def get_coords(self) -> tuple[float, float]:
"""Return Station coordinates."""
return self.coords
def get_datetime(self) -> datetime:
"""Return Station datetime of data."""
return self._datetime
def get_distance(self) -> float:
"""Return Station distance from selected coordinates."""
return round(self.distance, 3)
def get_dew_point(self) -> float | None:
"""Return Station dew point."""
return self.dew_point
def get_humidity(self) -> float | None:
"""Return Station humidity."""
return self.humidity
def get_id(self) -> str:
"""Return Station ID."""
return self.id
def get_name(self) -> str | None:
"""Return Station name."""
return self.name
def get_outdated(self) -> bool:
"""Return Station data outdated."""
cur_dt = get_current_datetime()
return cur_dt > self.get_datetime() + STATION_MAX_DELTA
def get_precipitation(self) -> float | None:
"""Return Station precipitation."""
return self.precipitation
def get_pressure(self) -> float | None:
"""Return Station pressure."""
return self.pressure
def get_temp(self) -> float | None:
"""Return Station temperature."""
return self.temp
def get_temp_max(self) -> float | None:
"""Return Station maximum temperature."""
return self.temp_max
def get_temp_min(self) -> float | None:
"""Return Station minimum temperature."""
return self.temp_min
def get_timestamp_utc(self) -> str:
"""Return Station UTC timestamp."""
return self._datetime.isoformat()
def get_timezone(self) -> ZoneInfo:
"""Return Station timezone."""
return self.zoneinfo
def get_wind_direction(self) -> float | None:
"""Return Station wind direction."""
return self.wind_direction
def get_wind_speed(self) -> float | None:
"""Return Station wind speed."""
return self.wind_speed
def get_wind_speed_max(self) -> float | None:
"""Return Station maximum wind speed."""
return self.wind_speed_max
def update_sample(self, data: dict[str, Any]) -> None:
"""Update Station data from sample."""
station_dt = parse_api_timestamp(data[AEMET_ATTR_STATION_DATE])
self._datetime = station_dt.astimezone(self.get_timezone())
if AEMET_ATTR_STATION_DEWPOINT in data:
self.dew_point = float(data[AEMET_ATTR_STATION_DEWPOINT])
if AEMET_ATTR_STATION_HUMIDITY in data:
self.humidity = float(data[AEMET_ATTR_STATION_HUMIDITY])
if AEMET_ATTR_STATION_PRECIPITATION in data:
self.precipitation = float(data[AEMET_ATTR_STATION_PRECIPITATION])
if AEMET_ATTR_STATION_PRESSURE_SEA in data:
self.pressure = float(data[AEMET_ATTR_STATION_PRESSURE_SEA])
elif AEMET_ATTR_STATION_PRESSURE in data:
self.pressure = float(data[AEMET_ATTR_STATION_PRESSURE])
if AEMET_ATTR_STATION_TEMPERATURE in data:
self.temp = float(data[AEMET_ATTR_STATION_TEMPERATURE])
if AEMET_ATTR_STATION_TEMPERATURE_MAX in data:
self.temp_max = float(data[AEMET_ATTR_STATION_TEMPERATURE_MAX])
if AEMET_ATTR_STATION_TEMPERATURE_MIN in data:
self.temp_min = float(data[AEMET_ATTR_STATION_TEMPERATURE_MIN])
if AEMET_ATTR_STATION_WIND_DIRECTION in data:
self.wind_direction = float(data[AEMET_ATTR_STATION_WIND_DIRECTION])
if AEMET_ATTR_STATION_WIND_SPEED in data:
self.wind_speed = float(data[AEMET_ATTR_STATION_WIND_SPEED])
if AEMET_ATTR_STATION_WIND_SPEED_MAX in data:
self.wind_speed_max = float(data[AEMET_ATTR_STATION_WIND_SPEED_MAX])
def update_samples(self, samples: dict[str, Any]) -> None:
"""Update Station data from samples."""
latest: dict[str, Any]
latest_dt: datetime | None = None
for sample in samples[ATTR_DATA]:
dt = parse_api_timestamp(sample[AEMET_ATTR_STATION_DATE])
if latest_dt is None or dt > latest_dt:
latest = sample
latest_dt = dt
if (
latest_dt is not None
and self.get_datetime() < latest_dt <= get_current_datetime()
):
self.update_sample(latest)
def data(self) -> dict[str, Any]:
"""Return station data."""
data: dict[str, Any] = {
AOD_ALTITUDE: self.get_altitude(),
AOD_COORDS: self.get_coords(),
AOD_DATETIME: self.get_datetime(),
AOD_DISTANCE: self.get_distance(),
AOD_ID: self.get_id(),
AOD_NAME: self.get_name(),
AOD_OUTDATED: self.get_outdated(),
AOD_TIMESTAMP_UTC: self.get_timestamp_utc(),
AOD_TIMEZONE: self.get_timezone(),
}
dew_point = self.get_dew_point()
if dew_point is not None:
data[AOD_DEW_POINT] = dew_point
humidity = self.get_humidity()
if humidity is not None:
data[AOD_HUMIDITY] = humidity
precipitation = self.get_precipitation()
if precipitation is not None:
data[AOD_PRECIPITATION] = precipitation
pressure = self.get_pressure()
if pressure is not None:
data[AOD_PRESSURE] = pressure
temp = self.get_temp()
if temp is not None:
data[AOD_TEMP] = temp
temp_max = self.get_temp_max()
if temp_max is not None:
data[AOD_TEMP_MAX] = temp_max
temp_min = self.get_temp_min()
if temp_min is not None:
data[AOD_TEMP_MIN] = temp_min
wind_direction = self.get_wind_direction()
if wind_direction is not None:
data[AOD_WIND_DIRECTION] = wind_direction
wind_speed = self.get_wind_speed()
if wind_speed is not None:
data[AOD_WIND_SPEED] = wind_speed
wind_speed_max = self.get_wind_speed_max()
if wind_speed_max is not None:
data[AOD_WIND_SPEED_MAX] = wind_speed_max
return data
def weather(self) -> dict[str, Any]:
"""Return Station weather data."""
weather: dict[str, Any] = {}
dew_point = self.get_dew_point()
if dew_point is not None:
weather[AOD_DEW_POINT] = dew_point
humidity = self.get_humidity()
if humidity is not None:
weather[AOD_HUMIDITY] = humidity
precipitation = self.get_precipitation()
if precipitation is not None:
weather[AOD_PRECIPITATION] = precipitation
pressure = self.get_pressure()
if pressure is not None:
weather[AOD_PRESSURE] = pressure
temp = self.get_temp()
if temp is not None:
weather[AOD_TEMP] = temp
wind_direction = self.get_wind_direction()
if wind_direction is not None:
weather[AOD_WIND_DIRECTION] = wind_direction
wind_speed = self.get_wind_speed()
if wind_speed is not None:
weather[AOD_WIND_SPEED] = wind_speed
wind_speed_max = self.get_wind_speed_max()
if wind_speed_max is not None:
weather[AOD_WIND_SPEED_MAX] = wind_speed_max
return weather
|