final

scheduleII.py

# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "aioclock",
#     "diskcache",
#     "httpx",
# ]
# ///

from contextlib import asynccontextmanager
from typing import Any, AsyncGenerator, NamedTuple

import httpx
from aioclock import AioClock, Depends, Every, At, Once
from aioclock.group import Group
from diskcache import Cache


group = Group()


class Location(NamedTuple):
    latitude: float
    longitude: float


def get_location() -> Location:
    return Location(latitude=55.57, longitude=12.26)  # Karlslunde


def get_headers() -> dict[str, str]:
    return {"User-Agent": "Prayson Daniel <praysonpi@gmail.com>"}


def get_cache() -> Cache:
    return Cache("/tmp/weather_cache")


@group.task(trigger=At(tz="Europe/Copenhagen", hour=0, minute=0, second=0))
async def get_location_id(
    location: Location = Depends(get_location),
    headers=Depends(get_headers),
    cache: Cache = Depends(get_cache),
) -> str:
    """Return DMI location ID (cached)."""

    base_url = "https://www.dmi.dk/NinJo2DmiDk/ninjo2dmidk"

    cache_key = f"id:{location.latitude}:{location.longitude}"
    match cache.get(cache_key):
        case cached_id if cached_id is not None:
            return cached_id
        case _:
            params = {
                "cmd": "llj",
                "hrs": 24,
                "lon": location.longitude,
                "lat": location.latitude,
            }

            async with httpx.AsyncClient(headers=headers) as client:
                resp = await client.get(base_url, params=params)
                resp.raise_for_status()
                location_id = resp.json().get("id")

            cache.set(cache_key, location_id, expire=60 * 60 * 24)
            return location_id


@group.task(trigger=Every(minutes=30))
async def get_weather(
    location: Location = Depends(get_location),
    headers: dict[str, str] = Depends(get_headers),
    cache: Cache = Depends(get_cache),
) -> dict[str, Any]:
    """Return weather using ID from cache"""

    base_url = "https://www.dmi.dk/dmidk_byvejrWS/rest/json/id"

    id_key = f"id:{location.latitude}:{location.longitude}"
    location_id = cache.get(id_key)

    if not location_id:
        raise ValueError(
            f"No cached ID found for ({location.latitude}, {location.longitude}). "
            f"aioclock must populate it first."
        )

    url = f"{base_url}/{location_id}"

    async with httpx.AsyncClient(headers=headers) as client:
        resp = await client.get(url)
        resp.raise_for_status()
        weather = resp.json()
    # do something awesome with data ...
    return weather


@asynccontextmanager
async def lifespan(app: AioClock) -> AsyncGenerator[AioClock, None]:
    """Manage application startup and shutdown."""
    # on startup
    print("\n" + "=" * 60)
    print("🚀 WEATHER MONITORING SYSTEM STARTING")
    print("=" * 60)
    print("📍 Location: Karlslunde (55.57, 12.26)")
    print("⏰ Schedule:")
    print(r"   - Location ID update:  Daily at midnight Europe\Copenhagen")
    print("   - Weather updated: Every 30 minutes")
    print("=" * 60 + "\n")

    yield app

    # on shutdown
    print("\n" + "=" * 60)
    print("👋 WEATHER MONITORING SYSTEM SHUTTING DOWN")
    print("=" * 60 + "\n")


app = AioClock(lifespan=lifespan)
app.include_group(group)

if __name__ == "__main__":
    import asyncio

    asyncio.run(app.serve())