trmnl-report/main.py

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_date(dt):
    dt = dt.astimezone(eastern)
    return dt.strftime("%a %d").lower()


def format_time(dt):
    dt = dt.astimezone(eastern)
    return dt.strftime("%I:%M%p").lower()


def format_datetime(dt):
    dt = dt.astimezone(eastern)
    return f"{format_date(dt)} {format_time(dt)}"


def pollen_desc(index):
    if index < 2.5:
        return f"{index} l"
    elif index < 4.9:
        return f"{index} l-m"
    elif index < 7.3:
        return f"{index} m"
    elif index < 9.7:
        return f"{index} m-h"
    else:
        return f"{index} h"


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": pollen_desc(d["Index"]),
                        "period": relative_day_to_date(d["Type"]),
                    }
                    for d in data["Location"]["periods"]
                    if d["Type"].lower().strip() in ["today", "tomorrow"]
                ],
            }
        ]


@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"][:2]
        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"]),
    )

    pollen_periods = {p["period"]: p for p in pollen[0]["periods"]} if pollen and pollen[0]["periods"] else {}
    weather_periods = {p["period"]: p for p in weather[0]["periods"]} if weather and weather[0]["periods"] else {}
    
    # Add current weather data as a "current" period
    if weather and weather[0]:
        data = weather[0]
        weather_periods["current"] = {
            "period": "current",
            "temp": data["current_temp"],
            "feels_like": data["current_feels_like"],
            "humidity": data["current_humidity"],
            "pressure": data["current_pressure"],
            "desc": data["current_desc"],
            "sunrise": data["sunrise"],
            "sunset": data["sunset"]
        }
    
    all_periods = set(pollen_periods.keys()) | set(weather_periods.keys())
    merged_periods = []
    
    sorted_periods = sorted(all_periods, key=lambda x: (x != "current", x))
    
    for period in sorted_periods:
        merged_period = {"period": period}
        
        if period in pollen_periods:
            merged_period["pollen"] = pollen_periods[period]["index"]
        
        if period in weather_periods:
            weather_data = weather_periods[period]
            if period == "current":
                merged_period.update({
                    "temp": weather_data["temp"],
                    "feels_like": weather_data["feels_like"],
                    "desc": weather_data["desc"],
                    "humidity": weather_data["humidity"],
                    "pressure": weather_data["pressure"]
                })
            else:
                merged_period.update({
                    "low": weather_data["low"],
                    "high": weather_data["high"],
                    "desc": weather_data["desc"],
                    "humidity": weather_data["humidity"],
                    "sunrise": weather_data["sunrise"],
                    "sunset": weather_data["sunset"],
                    "pressure": weather_data["pressure"]
                })
        
        merged_periods.append(merged_period)
    
    return {
        "fetched_at": format_datetime(datetime.now()),
        "data": merged_periods,
    }