InfernalGames/gallery
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(openweather): rework openweather parser

Browse Source
This commit is contained in:
shmyga
2026-09-17 22:03:24 +03:00
parent 2e9d04c10e
commit 6178c694f9
4 changed files with 95 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gallery/painting/openweather/api.py
+18 -33
View File
@@ -1,6 +1,6 @@
import datetime
import logging
from collections import defaultdict
from itertools import groupby
from os import environ
from aiocache import cached
@@ -10,10 +10,9 @@ from gallery.sketch.weather.model import Location, WeatherResponse, WeatherValue
from gallery.sketch.weather.util import merge_weather_values
from gallery.util import TimeUnit
from .openweather import Forecast
from .openweather import Location as OpenWeatherLocation
from .openweather import OpenWeather
from .parser import FORECAST_ITEM_PARSER
from .parser import OpenWeatherForecastParser, OpenWeatherLocationParser
logger = logging.getLogger("openweather")
@@ -21,6 +20,9 @@ logger = logging.getLogger("openweather")
class OpenWeatherApi(WeatherApi[OpenWeather]):
PROVIDER = "openweather"
forecast_parser = OpenWeatherForecastParser()
location_parser = OpenWeatherLocationParser()
def _build_source(self) -> OpenWeather:
return OpenWeather(environ["OPENWEATHER_KEY"])
@@ -41,34 +43,17 @@ class OpenWeatherApi(WeatherApi[OpenWeather]):
alias="redis",
ttl=TimeUnit.HOUR,
)
async def _get_location_forecast(self, location_id: str) -> Forecast:
return await self.source.get_forecast(*self._parse_location(location_id))
async def _get_location_forecast(self, location_id: str) -> list[WeatherValue]:
content = await self.source.get_forecast(*self._parse_location(location_id))
return self.forecast_parser.parse(content)
async def find_locations(self, query: str) -> list[Location]:
result = await self.source.find_locations(query)
return [
Location(
id=f"{item.lat}:{item.lon}",
name=item.name,
provider=self.provider,
lat=item.lat,
lon=item.lon,
country=item.country,
country_code=item.country.lower(),
district=item.state or "",
subdistrict="",
)
for item in result
]
content = await self.source.find_locations(query)
return self.location_parser.parse(content)
async def get_day(self, location_id: str, date: datetime.date) -> WeatherResponse:
location: OpenWeatherLocation = await self._get_location(location_id)
data: Forecast = await self._get_location_forecast(location_id)
values = []
for item in data.list:
value = FORECAST_ITEM_PARSER.parse(item)
if value.date.date() == date:
values.append(value)
values = [value for value in await self._get_location_forecast(location_id) if value.date.date() == date]
return WeatherResponse(
location=location.name,
date=date,
@@ -78,13 +63,13 @@ class OpenWeatherApi(WeatherApi[OpenWeather]):
async def get_days(self, location_id: str, days: int) -> WeatherResponse:
location: OpenWeatherLocation = await self._get_location(location_id)
data: Forecast = await self._get_location_forecast(location_id)
values_by_date: dict[datetime.datetime, list[WeatherValue]] = defaultdict(list)
for item in data.list:
value = FORECAST_ITEM_PARSER.parse(item)
item_date = value.date.replace(hour=0, minute=0)
values_by_date[item_date].append(value)
values = [merge_weather_values(date, values) for date, values in values_by_date.items()]
values = [
merge_weather_values(datetime.datetime.combine(k, datetime.datetime.min.time()), list(g))
for k, g in groupby(
sorted(await self._get_location_forecast(location_id), key=lambda item: item.date.date()),
lambda item: item.date.date(),
)
]
return WeatherResponse(
location=location.name,
date=datetime.date.today(),
gallery/painting/openweather/parser.py
+61 -23
View File
@@ -1,13 +1,20 @@
import datetime
from gallery.sketch.weather.model import Cloudness, Precipitation, WeatherValue
from gallery.sketch.weather.util import build_weather_value
from gallery.sketch.parse.core import Parser
from gallery.sketch.weather.model import (
Cloudness,
Location,
Precipitation,
WeatherValue,
)
from gallery.sketch.weather.parser import parse_wind_direction
from gallery.sketch.weather.util import convert_hPa_to_mmHg
from ...sketch.weather.parser import parse_wind_direction
from .openweather import ForecastItem
from .openweather import Forecast, ForecastItem
from .openweather import Location as OpenWeatherLocation
class ForecastItemParser:
class ForecastItemParser(Parser[ForecastItem, list[WeatherValue]]):
CLOUDNESS_MAP: dict[str, Cloudness] = {
"clear sky": Cloudness.CLEAR,
"few clouds": Cloudness.PARTLY_CLOUDY,
@@ -23,23 +30,54 @@ class ForecastItemParser:
"heavy rain": Precipitation.SHOWER,
}
def parse(self, item: ForecastItem) -> WeatherValue:
item_date = datetime.datetime.fromtimestamp(item.dt, datetime.UTC)
item_date = item_date.replace(tzinfo=datetime.timezone.utc).astimezone(tz=None).replace(tzinfo=None)
value = build_weather_value(item_date)
# TODO parse temperature interval flag
value.temperature = [round(item.main.temp)]
# value.temperature = [round(item.main.temp_max), round(item.main.temp_min)]
value.pressure = [round(item.main.pressure * 0.75006)]
value.humidity = item.main.humidity
value.wind.speed = round(item.wind.speed)
value.wind.gust = round(item.wind.gust)
value.wind.direction = parse_wind_direction(item.wind.deg)
value.sky.cloudness = self.CLOUDNESS_MAP.get(item.weather[0].description, Cloudness.CLEAR)
value.sky.precipitation = self.PRECIPITATION_MAP.get(item.weather[0].description, Precipitation.NO)
if item.rain:
value.precipitation = round(item.rain.interval_3h, 1)
return value
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": parse_wind_direction(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,
}
)
FORECAST_ITEM_PARSER = ForecastItemParser()
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="openweather",
lat=item.lat,
lon=item.lon,
country=item.country,
country_code=item.country.lower(),
district=item.state or "",
subdistrict="",
)
for item in data
]
gallery/sketch/weather/model.py
+6 -6
View File
@@ -1,6 +1,6 @@
import datetime
from enum import StrEnum, auto
from typing import Self
from typing import Literal, Self
from pydantic import BaseModel, Field
@@ -76,16 +76,16 @@ class Wind(Model):
class WeatherValue(Model):
date: datetime.datetime
sky: Sky
wind: Wind
temperature: list[int]
sky: Sky = Field(default_factory=Sky)
wind: Wind = Field(default_factory=Wind)
temperature: list[int] = Field(default_factory=list)
precipitation: float = 0
pressure: list[int] = Field(default=[])
pressure: list[int] = Field(default_factory=list)
humidity: int = -1
class WeatherResponse(Model):
location: str
date: datetime.date
period: str
period: Literal["day", "days"] = "day"
values: list[WeatherValue]
gallery/sketch/weather/util.py
+10 -23
View File
@@ -2,32 +2,11 @@ import datetime
import statistics
from typing import Counter, Literal
from .model import Cloudness, Precipitation, Sky, WeatherValue, Wind
from .model import Cloudness, Precipitation, WeatherValue
MergeMode = Literal["interval", "average"]
def build_weather_value(date: datetime.datetime | None = None) -> WeatherValue:
return WeatherValue(
date=date or datetime.datetime.now(),
sky=Sky(
cloudness=Cloudness.CLEAR,
precipitation=Precipitation.NO,
thunder=False,
fog=False,
),
wind=Wind(
speed=0,
gust=0,
direction=None,
),
temperature=[],
precipitation=0,
pressure=[],
humidity=-1,
)
def _merge_values(values: list[int], mode: MergeMode) -> list[int]:
if mode == "interval":
if (max_value := max(values)) == (min_value := min(values)):
@@ -43,7 +22,7 @@ def merge_weather_values(
values: list[WeatherValue],
mode: MergeMode = "interval",
) -> WeatherValue:
result = build_weather_value(date)
result = WeatherValue(date=date)
temperatures = []
pressures = []
humidities = []
@@ -70,10 +49,18 @@ def merge_weather_values(
result.wind.gust = round(statistics.mean(wind_gusts))
result.wind.direction = Counter(wind_directions).most_common(1)[0][0]
for item in cloudnesses:
# TODO: Instance of 'FieldInfo' has no 'cloudness' member (no-member)
# pylint:disable=no-member
if Cloudness.values.index(item) > Cloudness.values.index(result.sky.cloudness):
result.sky.cloudness = item
for item in precipitations:
# TODO: Instance of 'FieldInfo' has no 'precipitation' member (no-member)
# pylint:disable=no-member
if Precipitation.values.index(item) > Precipitation.values.index(result.sky.precipitation):
result.sky.precipitation = item
result.precipitation = precipitation
return result
def convert_hPa_to_mmHg(value: int) -> int:
return round(value / 1.333)