Stahovani pocasi, editace kopilota, nahravani GPX k letu

src/external/weather.py

@@ -0,0 +1,58 @@
+import datetime
+import urllib.parse
+from typing import Tuple, Dict
+import aiohttp
+from aiocache import cached
+
+
+class Weather:
+    FORECAST_URL = "https://api.open-meteo.com/v1/forecast?"
+    ARCHIVE_URL = "https://archive-api.open-meteo.com/v1/archive?"
+    TIMEZONE = "Europe/Prague"
+    METRICS = (
+        "pressure_msl", "temperature_2m", "dewpoint_2m", "rain", "cloudcover_low", "cloudcover", "windspeed_10m",
+        "winddirection_10m"
+    )
+
+    def get_weather_info_url(self, start_date: datetime.date, end_date: datetime.date, gps: Tuple[float, float]) -> str:
+        today = datetime.datetime.now().date()
+        date_diff = today - end_date
+
+        if date_diff.days >= 7:
+            # historical API offers data only older than 5 days
+            url = self.ARCHIVE_URL
+        else:
+            # forecast contains data even 14 days ago
+            url = self.FORECAST_URL
+
+        params = {
+            "latitude": gps[0],
+            "longitude": gps[1],
+            "timezone": self.TIMEZONE,
+            "hourly": ','.join(self.METRICS),
+            "start_date": start_date,
+            "end_date": end_date,
+        }
+
+        query_string = urllib.parse.urlencode(params)
+        return f"{url}{query_string}"
+
+    @cached(ttl=6*3600)
+    async def download_weather_for_day(self, date: datetime.date, gps: Tuple[float, float]):
+        url = self.get_weather_info_url(start_date=date, end_date=date, gps=gps)
+
+        async with aiohttp.ClientSession() as session:
+            async with session.get(url) as resp:
+                resp.raise_for_status()
+                return await resp.json()
+
+    async def get_weather_for_hour(self, date_time: datetime.datetime, gps: Tuple[float, float]) -> Dict[str, float|str]:
+        data = await self.download_weather_for_day(date_time.date(), gps)
+
+        # TODO: kontrola timezone!
+        # TODO: interpolace - udelat vazenyprumer z dvou po sobe jdoucich hodin
+        idx = date_time.hour
+        result_data = {metric: data['hourly'][metric][idx] for metric in self.METRICS}
+        result_data['datetime'] = datetime.datetime.strptime(data['hourly']['time'][idx], "%Y-%m-%dT%H:%M")
+
+        return result_data