68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
import datetime
|
|
import logging
|
|
from collections import defaultdict
|
|
|
|
from aiocache import cached
|
|
|
|
from gallery.sketch.weather.api import WeatherApi
|
|
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, OpenWeather
|
|
from .parser import FORECAST_ITEM_PARSER
|
|
|
|
logger = logging.getLogger("openweather")
|
|
|
|
|
|
class OpenWeatherApi(WeatherApi):
|
|
PROVIDER = "openweather"
|
|
SOURCE = OpenWeather("517a6bccceaa1c48127f6199ec3fb7cf")
|
|
|
|
@classmethod
|
|
def _parse_location(cls, location_id: str) -> tuple[float, float]:
|
|
return tuple(map(float, location_id.split(":", maxsplit=2)))
|
|
|
|
@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:
|
|
return await self.SOURCE.get_forecast(*self._parse_location(location_id))
|
|
|
|
async def find_locations(self, query: str) -> list[Location]:
|
|
raise NotImplementedError
|
|
|
|
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)
|
|
return WeatherResponse(
|
|
location=location_id,
|
|
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()
|
|
]
|
|
return WeatherResponse(
|
|
location=location_id,
|
|
date=datetime.date.today(),
|
|
period="days",
|
|
values=list(sorted(values, key=lambda item: item.date)),
|
|
)
|