85 lines
3.0 KiB
Python
85 lines
3.0 KiB
Python
import datetime
|
|
|
|
from gallery.sketch.parse.core import Parser
|
|
from gallery.sketch.weather.model import (
|
|
Cloudness,
|
|
Location,
|
|
Precipitation,
|
|
WeatherValue,
|
|
)
|
|
from gallery.sketch.weather.parse.wind import WIND_DIRECTION_PARSER
|
|
from gallery.sketch.weather.util import convert_hPa_to_mmHg
|
|
|
|
from . import PROVIDER
|
|
from .openweather import Forecast, ForecastItem
|
|
from .openweather import Location as OpenWeatherLocation
|
|
|
|
|
|
class ForecastItemParser(Parser[ForecastItem, list[WeatherValue]]):
|
|
CLOUDNESS_MAP: dict[str, Cloudness] = {
|
|
"clear sky": Cloudness.CLEAR,
|
|
"few clouds": Cloudness.PARTLY_CLOUDY,
|
|
"scattered clouds": Cloudness.PARTLY_CLOUDY,
|
|
"broken clouds": Cloudness.CLOUDY,
|
|
"overcast clouds": Cloudness.MAINLY_CLOUDY,
|
|
"light rain": Cloudness.CLOUDY,
|
|
}
|
|
|
|
PRECIPITATION_MAP: dict[str, Precipitation] = {
|
|
"light rain": Precipitation.SMALL_RAIN,
|
|
"rain": Precipitation.RAIN,
|
|
"heavy rain": Precipitation.SHOWER,
|
|
}
|
|
|
|
def parse(self, data: ForecastItem) -> WeatherValue:
|
|
date = (
|
|
datetime.datetime.fromtimestamp(data.dt, datetime.UTC)
|
|
.replace(tzinfo=datetime.timezone.utc)
|
|
.astimezone(tz=None)
|
|
.replace(tzinfo=None)
|
|
)
|
|
return WeatherValue.model_validate(
|
|
{
|
|
"date": date,
|
|
"sky": {
|
|
"cloudness": self.CLOUDNESS_MAP.get(data.weather[0].description, Cloudness.CLEAR),
|
|
"precipitation": self.PRECIPITATION_MAP.get(data.weather[0].description, Precipitation.NO),
|
|
},
|
|
"wind": {
|
|
"speed": round(data.wind.speed),
|
|
"gust": round(data.wind.gust),
|
|
"direction": WIND_DIRECTION_PARSER.parse(data.wind.deg),
|
|
},
|
|
"temperature": [round(data.main.temp)], # [round(data.main.temp_max), round(data.main.temp_min)]
|
|
"precipitation": round(data.rain.interval_3h, 1) if data.rain else 0,
|
|
"pressure": [convert_hPa_to_mmHg(data.main.pressure)],
|
|
"humidity": data.main.humidity,
|
|
}
|
|
)
|
|
|
|
|
|
class OpenWeatherForecastParser(Parser[Forecast, list[WeatherValue]]):
|
|
item_parser = ForecastItemParser()
|
|
|
|
def parse(self, data: Forecast) -> list[WeatherValue]:
|
|
return [self.item_parser.parse(item) for item in data.list]
|
|
|
|
|
|
class OpenWeatherLocationParser(Parser[list[OpenWeatherLocation], list[Location]]):
|
|
|
|
def parse(self, data: list[OpenWeatherLocation]) -> list[Location]:
|
|
return [
|
|
Location(
|
|
id=f"{item.lat}:{item.lon}",
|
|
name=item.name,
|
|
provider=PROVIDER,
|
|
lat=item.lat,
|
|
lon=item.lon,
|
|
country=item.country,
|
|
country_code=item.country.lower(),
|
|
district=item.state or "",
|
|
subdistrict="",
|
|
)
|
|
for item in data
|
|
]
|