Skip to content
Mostly Right

Use case

Monitor live weather for alerting

You run weather-sensitive operations across many stations. When the realized temp at any of them crosses a threshold, you want a page. live.stream() is the right tool: async iteration, dedupe by observed_at, polite-floor compliant out of the box.

What we’re building

Section titled “What we’re building”

A daemon that:

  • Watches N stations in parallel
  • Triggers when temp_f crosses a per-station threshold
  • Posts to Slack (sub process() for any other sink)
  • Deduplicates so the same observation never alerts twice

Python — asyncio

Section titled “Python — asyncio”
import asyncio
import os
import httpx
import mostlyright


SLACK_WEBHOOK = os.environ["SLACK_WEBHOOK"]


ALERTS = {
    "KNYC": {"above_f": 95, "below_f": 10},
    "KORD": {"above_f": 95, "below_f": -5},
    "KLAX": {"above_f": 105, "below_f": 35},
}


async def post_slack(client: httpx.AsyncClient, text: str) -> None:
    await client.post(SLACK_WEBHOOK, json={"text": text}, timeout=10)


async def watch(station: str, thresh: dict, client: httpx.AsyncClient) -> None:
    last_alerted_at = None
    async for row in mostlyright.live.stream(station):
        temp = row["temp_f"]
        if temp is None:
            continue
        crossed = (
            (thresh.get("above_f") and temp >= thresh["above_f"]) or
            (thresh.get("below_f") and temp <= thresh["below_f"])
        )
        if crossed and row["observed_at"] != last_alerted_at:
            await post_slack(
                client,
                f"🚨 {station}: {temp}°F at {row['observed_at']} (source={row['source']})",
            )
            last_alerted_at = row["observed_at"]


async def main() -> None:
    async with httpx.AsyncClient() as client:
        await asyncio.gather(*[
            watch(station, thresh, client)
            for station, thresh in ALERTS.items()
        ])


asyncio.run(main())

Run it as a long-lived process (systemd, k8s deployment, Cloudflare durable object).

TypeScript — Node / Bun / Worker

Section titled “TypeScript — Node / Bun / Worker”
import { stream } from "@mostlyrightmd/weather";


const SLACK_WEBHOOK = process.env.SLACK_WEBHOOK!;


const ALERTS: Record<string, { aboveF?: number; belowF?: number }> = {
  KNYC: { aboveF: 95, belowF: 10 },
  KORD: { aboveF: 95, belowF: -5 },
  KLAX: { aboveF: 105, belowF: 35 },
};


async function postSlack(text: string): Promise<void> {
  await fetch(SLACK_WEBHOOK, {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({ text }),
  });
}


async function watch(station: string, thresh: { aboveF?: number; belowF?: number }) {
  let lastAlertedAt: string | null = null;
  for await (const row of stream(station)) {
    if (row.temp_f == null) continue;
    const crossed =
      (thresh.aboveF != null && row.temp_f >= thresh.aboveF) ||
      (thresh.belowF != null && row.temp_f <= thresh.belowF);
    if (crossed && row.observed_at !== lastAlertedAt) {
      await postSlack(
        `🚨 ${station}: ${row.temp_f}°F at ${row.observed_at} (source=${row.source})`,
      );
      lastAlertedAt = row.observed_at;
    }
  }
}


await Promise.all(
  Object.entries(ALERTS).map(([station, thresh]) => watch(station, thresh)),
);

Graceful shutdown

Section titled “Graceful shutdown”

The Python version handles SIGTERM cleanly via asyncio.CancelledError:

import signal


async def main() -> None:
    stop = asyncio.Event()
    for sig in (signal.SIGTERM, signal.SIGINT):
        asyncio.get_running_loop().add_signal_handler(sig, stop.set)


    tasks = [asyncio.create_task(watch(s, t, client)) for s, t in ALERTS.items()]
    await stop.wait()
    for t in tasks:
        t.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)

TypeScript: wrap the AbortController around the stream(..., { signal }) opt.

Polite floors apply automatically

Section titled “Polite floors apply automatically”

You won’t hammer AWC/IEM — live.stream() enforces the 30s / 60s polite floor per source. A passing observation cycle is roughly hourly per station, so a 10-station watcher costs about 10 fetches per hour per source.

Cross-source cross-check

Section titled “Cross-source cross-check”

If AWC has been silent for >2 cycles for a station, you might want to fall back to IEM. Add a parallel IEM stream and union the alerts:

async def watch_both(station: str, thresh: dict, client: httpx.AsyncClient) -> None:
    async def watch_source(source: str) -> None:
        async for row in mostlyright.live.stream(station, source=source):
            # …same threshold check; dedupe by (observed_at, source)
            ...
    await asyncio.gather(watch_source("awc"), watch_source("iem"))

The polite floors apply independently per (station, source), so a station has two concurrent feeders without violating either floor.

What this recipe doesn’t do

Section titled “What this recipe doesn’t do”
  • Cache live observations. live.stream() is the ticker; nothing writes to parquet. If you also want backtest tape, run a parallel research() cron.
  • Treat live rows as final. Live observations can be revised. For a stable record, use research() after the window closes.
  • Handle station outages gracefully. A station that goes dark won’t yield. Run a separate liveness watchdog if you need detection.

See also

Section titled “See also”