94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
import datetime
|
|
import logging
|
|
from collections import defaultdict
|
|
from os import environ
|
|
|
|
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
|
|
from .openweather import Location as OpenWeatherLocation
|
|
from .openweather import OpenWeather
|
|
from .parser import FORECAST_ITEM_PARSER
|
|
|
|
logger = logging.getLogger("openweather")
|
|
|
|
|
|
class OpenWeatherApi(WeatherApi[OpenWeather]):
|
|
PROVIDER = "openweather"
|
|
|
|
def _build_source(self) -> OpenWeather:
|
|
return OpenWeather(environ["OPENWEATHER_KEY"])
|
|
|
|
@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}.location",
|
|
alias="redis",
|
|
ttl=TimeUnit.DAY,
|
|
)
|
|
async def _get_location(self, location_id: str) -> OpenWeatherLocation:
|
|
return await self.source.get_location(*self._parse_location(location_id))
|
|
|
|
@cached(
|
|
key_builder=lambda fun, self, location_id: f"api.weather.{self.provider}.source.{location_id}.forecast",
|
|
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 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
|
|
]
|
|
|
|
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)
|
|
return WeatherResponse(
|
|
location=location.name,
|
|
date=date,
|
|
period="day",
|
|
values=values,
|
|
)
|
|
|
|
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()]
|
|
return WeatherResponse(
|
|
location=location.name,
|
|
date=datetime.date.today(),
|
|
period="days",
|
|
values=list(sorted(values, key=lambda item: item.date)),
|
|
)
|