feat(weather): add openweather api

gallery/painting/openweather/api.py

@@ -0,0 +1,70 @@
+import datetime
+import logging
+from collections import defaultdict
+
+from aiocache import cached
+
+from gallery.sketch.weather.api import WeatherApi
+from gallery.sketch.weather.catalog import BUNDLE, LocationId
+from gallery.sketch.weather.model import WeatherResponse, WeatherValue
+from gallery.sketch.weather.util import merge_weather_values
+from gallery.util import TimeUnit
+
+from .openweather import Forecast, OpenWeather
+from .parser import FORECAST_ITEM_PARSER
+
+logger = logging.getLogger("openweather")
+
+
+class OpenWeatherApi(WeatherApi):
+    PROVIDER = "openweather"
+    SOURCE = OpenWeather("517a6bccceaa1c48127f6199ec3fb7cf")
+
+    async def get_locations(self) -> list[str]:
+        return [
+            LocationId.OREL,
+            LocationId.ZMIYEVKA,
+        ]
+
+    @cached(
+        key_builder=lambda fun, self, location_id: f"api.weather.{self.provider}.source.{location_id}.forecast",
+        alias="redis",
+        ttl=TimeUnit.DAY,
+    )
+    async def _get_location_forecast(self, location_id: str) -> Forecast:
+        location = BUNDLE.get_item(location_id)
+        return await self.SOURCE.get_forecast(location.lat, location.lon)
+
+    async def get_day(self, location_id: str, date: datetime.date) -> WeatherResponse:
+        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)
+        location = BUNDLE.get_item(location_id)
+        return WeatherResponse(
+            location=location.name,
+            date=date,
+            period="day",
+            values=values,
+        )
+
+    async def get_days(self, location_id: str, days: int) -> WeatherResponse:
+        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()
+        ]
+        location = BUNDLE.get_item(location_id)
+        return WeatherResponse(
+            location=location.name,
+            date=datetime.date.today(),
+            period="days",
+            values=list(sorted(values, key=lambda item: item.date)),
+        )

gallery/painting/openweather/mock/__init__.py

@@ -0,0 +1,5 @@
+from pathlib import Path
+
+from gallery.sketch.mock import MockData
+
+OPENWEATHER_MOCK_DATA = MockData(Path(__file__).parent / "data")

gallery/painting/openweather/openweather.py

@@ -0,0 +1,83 @@
+import json
+
+from pydantic import BaseModel, Field
+
+from gallery.sketch.source import ApiSource
+
+
+class Model(BaseModel):
+    class Config:
+        use_enum_values = True
+
+
+class Main(Model):
+    temp: float
+    feels_like: float
+    temp_min: float
+    temp_max: float
+    pressure: int
+    sea_level: int
+    grnd_level: int
+    humidity: int
+    temp_kf: float
+
+
+class Weather(Model):
+    id: int
+    main: str
+    description: str
+    icon: str
+
+
+class Clouds(Model):
+    all: int
+
+
+class Wind(Model):
+    speed: float
+    deg: int
+    gust: float
+
+
+class Rain(Model):
+    interval_3h: float = Field(..., alias="3h")
+
+
+class Sys(Model):
+    pod: str
+
+
+class ForecastItem(Model):
+    dt: int
+    main: Main
+    weather: list[Weather]
+    clouds: Clouds
+    wind: Wind
+    visibility: int
+    pop: float
+    rain: Rain | None = None
+    sys: Sys
+    dt_txt: str
+
+
+class Forecast(Model):
+    cod: str
+    message: int
+    cnt: int
+    list: list[ForecastItem]
+
+
+class OpenWeather:
+    BASE_URL = "https://api.openweathermap.org"
+
+    def __init__(self, api_key: str):
+        self._api_key = api_key
+        self._source = ApiSource(self.BASE_URL)
+
+    async def get_forecast(self, lat: float, lon: float) -> Forecast:
+        endpoint = (
+            f"data/2.5/forecast?lat={lat}&lon={lon}&appid={self._api_key}&units=metric"
+        )
+        response = await self._source.request(endpoint)
+        response_data = json.loads(response)
+        return Forecast(**response_data)

gallery/painting/openweather/parser.py

@@ -0,0 +1,52 @@
+import datetime
+
+from gallery.sketch.weather.model import Cloudness, Precipitation, WeatherValue
+from gallery.sketch.weather.util import build_weather_value
+
+from .openweather import ForecastItem
+
+
+class ForecastItemParser:
+    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, 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 / 133.3 * 100)]
+        value.humidity = item.main.humidity
+        value.wind_speed = round(item.wind.speed)
+        value.wind_gust = round(item.wind.gust)
+        value.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
+
+
+FORECAST_ITEM_PARSER = ForecastItemParser()