initial pass at pollen and weather data

main.py

@@ -0,0 +1,151 @@
+import asyncio
+from datetime import datetime, timedelta
+import functools
+import logging
+import os
+import pprint
+import zoneinfo
+
+from fastapi import FastAPI, HTTPException
+import httpx
+
+
+pp = pprint.PrettyPrinter()
+logger = logging.getLogger("uvicorn.error")
+eastern = zoneinfo.ZoneInfo("America/New_York")
+
+app = FastAPI()
+
+
+CONFIG = {
+    # salem, ma
+    "zip": "01970",
+    "lat": "42.3554334",
+    "lon": "-71.060511",
+    "weather_api_key": os.environ.get("WEATHER_API_KEY", ""),
+    "auth_token": os.environ["AUTH_TOKEN"],
+}
+
+
+def format_datetime(dt):
+    return dt.astimezone(eastern).isoformat()
+
+
+def format_time(dt):
+    return dt.strftime("%I:%M%p").lower()
+
+
+def format_date(dt):
+    return dt.strftime("%a %d").lower()
+
+
+def relative_day_to_date(rel_dt):
+    dt = datetime.now()
+    day = timedelta(days=1)
+    match rel_dt.lower().strip():
+        case "yesterday":
+            return format_date(dt - day)
+        case "today":
+            return format_date(dt)
+        case "tomorrow":
+            return format_date(dt + day)
+        case passthrough:
+            return passthrough
+
+
+def fallback_handler(func):
+    @functools.wraps(func)
+    async def wrapper(*args, **kwargs):
+        try:
+            result = await func(*args, **kwargs)
+        except Exception as e:
+            logger.exception(e)
+            return []
+        return result
+
+    return wrapper
+
+
+@fallback_handler
+async def fetch_pollen(zipcode):
+    url = f"https://www.pollen.com/api/forecast/current/pollen/{zipcode}"
+    headers = {
+        "User-Agent": (
+            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
+            "AppleWebKit/537.36 (KHTML, like Gecko) "
+            "Chrome/123.0.0.0 Safari/537.36"
+        ),
+        "Accept": "application/json, text/plain, */*",
+        "Referer": url,
+        "Cookie": f"geo={zipcode}",
+    }
+    async with httpx.AsyncClient() as client:
+        response = await client.get(url, headers=headers)
+        response.raise_for_status()
+        data = response.json()
+        return [
+            {
+                "forecast_date": format_datetime(
+                    datetime.fromisoformat(data["ForecastDate"])
+                ),
+                "periods": [
+                    {"index": d["Index"], "period": relative_day_to_date(d["Type"])}
+                    for d in data["Location"]["periods"]
+                ],
+            }
+        ]
+
+
+@fallback_handler
+async def fetch_weather(lat, lon, weather_api_key):
+    url = f"https://api.openweathermap.org/data/3.0/onecall?lat={lat}&lon={lon}&appid={weather_api_key}&units=imperial"
+    async with httpx.AsyncClient() as client:
+        response = await client.get(url)
+        response.raise_for_status()
+        data = response.json()
+        current, periods = data["current"], data["daily"][:3]
+        return [
+            {
+                "forecast_date": format_datetime(datetime.fromtimestamp(current["dt"])),
+                "current_temp": int(round(current["temp"])),
+                "current_feels_like": int(round(current["feels_like"])),
+                "current_humidity": current["humidity"],
+                "sunrise": format_time(datetime.fromtimestamp(current["sunrise"])),
+                "sunset": format_time(datetime.fromtimestamp(current["sunset"])),
+                "current_pressure": current["pressure"],
+                "current_desc": current["weather"][0]["description"],
+                "periods": [
+                    {
+                        "low": int(round(p["temp"]["min"])),
+                        "high": int(round(p["temp"]["max"])),
+                        "desc": p["weather"][0]["description"],
+                        "humidity": p["humidity"],
+                        "sunrise": format_time(datetime.fromtimestamp(p["sunrise"])),
+                        "sunset": format_time(datetime.fromtimestamp(p["sunset"])),
+                        "pressure": p["pressure"],
+                        "period": format_date(datetime.fromtimestamp(p["dt"])),
+                    }
+                    for p in periods
+                ],
+            }
+        ]
+
+
+@app.get("/")
+async def read_root(token: str):
+    if token != CONFIG["auth_token"]:
+        raise HTTPException(status_code=403, detail="unauthorized")
+
+    [
+        pollen,
+        weather,
+    ] = await asyncio.gather(
+        fetch_pollen(CONFIG["zip"]),
+        fetch_weather(CONFIG["lat"], CONFIG["lon"], CONFIG["weather_api_key"]),
+    )
+
+    return {
+        "fetched_at": format_datetime(datetime.now()),
+        "pollen": pollen,
+        "weather": weather,
+    }