104 lines
2.3 KiB
Python
104 lines
2.3 KiB
Python
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 | None = None
|
|
pop: float
|
|
rain: Rain | None = None
|
|
sys: Sys
|
|
dt_txt: str
|
|
|
|
|
|
class Forecast(Model):
|
|
cod: str
|
|
message: int
|
|
cnt: int
|
|
list: list[ForecastItem]
|
|
|
|
|
|
class Location(Model):
|
|
name: str
|
|
lat: float
|
|
lon: float
|
|
country: str
|
|
state: str | None = None
|
|
local_names: dict[str, str] | None = None
|
|
|
|
|
|
class OpenWeather(ApiSource):
|
|
BASE_URL = "https://api.openweathermap.org"
|
|
|
|
def __init__(self, api_key: str):
|
|
self._api_key = api_key
|
|
super().__init__(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.request(endpoint)
|
|
response_data = json.loads(response)
|
|
return Forecast.model_validate(response_data)
|
|
|
|
async def find_locations(self, query: str, limit: int = 5) -> list[Location]:
|
|
endpoint = f"geo/1.0/direct?q={query}&limit={limit}&appid={self._api_key}"
|
|
response = await self.request(endpoint)
|
|
response_data = json.loads(response)
|
|
return [Location.model_validate(item) for item in response_data]
|
|
|
|
async def get_location(self, lat: float, lon: float) -> Location:
|
|
limit = 1
|
|
endpoint = f"geo/1.0/reverse?lat={lat}&lon={lon}&limit={limit}&appid={self._api_key}"
|
|
response = await self.request(endpoint)
|
|
response_data = json.loads(response)
|
|
return Location.model_validate(response_data[0])
|