Guides

Live streaming

mostlyright.live is the ticker surface — a live feed of fresh METARs from a single source. Use it for dashboards, alerting, or anything that polls and reacts in real time. Both SDKs ship mirrored APIs.

When to use live vs research

live.stream() and live.latest() are intentionally not the same surface as research(). One picks for the other.

	`research()`	`live.stream()` / `live.latest()`
Role	BATCH (training pairs, settlement)	STREAM (real-time monitoring)
Sources	AWC + IEM + GHCNh + CLI (fused)	ONE of `awc` \| `iem`
Cache writes	Yes (parquet, year-aligned)	No
QC	Yes	No
Semantics	Point-in-time	Async generator
Use case	Backtest, train a model, settle	Watch live, page on threshold

If you reach for live.stream(), you almost certainly do not want multi-source fusion: a single feed lets you reason about the latency of “what AWC said at 12:01” without IEM’s ~10-min delay smearing the picture.

1
import asyncio
2
import mostlyright
3

4
async def main() -> None:
5
    async for row in mostlyright.live.stream("KNYC"):
6
        print(row["observed_at"], row["temp_f"])
7

8
asyncio.run(main())

1
import { stream } from "@mostlyrightmd/weather";
2

3
for await (const row of stream("KNYC")) {
4
  console.log(row.observed_at, row.temp_f);
5
}

Signature:

Python: mostlyright.live.stream(station, *, source=None, poll_seconds=None) -> AsyncIterator[dict]
TypeScript: stream(station: string, opts?: { source?, pollSeconds?, signal? }): AsyncGenerator<LiveObservation>

Per-row contract:

Same shape as the canonical Observation schema — station, observed_at (ISO-8601 UTC), observation_type (METAR/SPECI), plus weather fields.
source carries the live-identity tag: "awc.live" or "iem.live" — distinguishes from research()-channel tags ("awc"/"iem").
Duplicate emission is suppressed: the generator yields a row only when observed_at differs from the previous one. Both AWC and IEM serve “the latest METAR” repeatedly between observation cycles; deduping collapses that noise.

latest() — one-shot

Python
TypeScript

1
row = await mostlyright.live.latest("KNYC")
2
print(row["temp_f"])

1
const row = await latest("KNYC");
2
console.log(row.temp_f);

Signature:

Python: mostlyright.live.latest(station, *, source=None) -> dict
TypeScript: latest(station: string, opts?: { source? }): Promise<LiveObservation>

latest() shares the per-tick fetch path with stream() but returns exactly once. Use it from cron-style schedulers, or when you want a single fresh observation without managing a generator’s lifecycle.

Unlike stream(), latest() raises NoLiveDataError when the upstream returns no observations. The error’s toDict() payload includes the resolved station and the source identity tag so caller logs can branch.

Sources

Source	Default?	Endpoint	Typical latency	Use when
`"awc"`	yes	`aviationweather.gov/api/data/metar`	`<1 minute`	Default — lowest latency
`"iem"`	no	`mesonet.agron.iastate.edu/cgi-bin/request/asos.py`	~5–10 minutes	AWC is down, or you want an independent cross-check

Each row carries its source identity on the source field:

AWC rows → source == "awc.live"
IEM rows → source == "iem.live"

The .live suffix distinguishes these from the archive-channel tags ("awc" / "iem") emitted by research() and the historical fetchers.

Polite floors

mostlyright.live will not let you poll an upstream faster than its polite floor:

Source	Floor
`awc`	30s
`iem`	60s

Calls like stream(..., poll_seconds=10) for an AWC stream raise ValueError("poll_seconds=10 below polite floor 30s for source='awc'") before the first poll. Omitting poll_seconds= uses the floor.

Why the floor:

Politeness. Both endpoints are public goods. AWC has no documented rate limit; IEM is a university server.
Useful signal. METARs are issued at roughly hourly cadence (SPECIs are intra-hour). Polling faster than ~30s does not surface new data on most stations — it just costs bandwidth.

Async patterns

Python

1
# Standard `async for` consume
2
async for row in mostlyright.live.stream("KNYC"):
3
    process(row)
4

5
# Break cleanly — the polite-floor sleep is `asyncio.sleep`, which
6
# propagates `CancelledError`, so `break` exits without zombie tasks.
7
async for row in mostlyright.live.stream("KNYC"):
8
    process(row)
9
    if row["temp_f"] > 95:
10
        break
11

12
# Run multiple stations in parallel — one task per station, polite
13
# floors apply per-task (NOT pooled).
14
async def watch(station: str) -> None:
15
    async for row in mostlyright.live.stream(station):
16
        process(station, row)
17

18
await asyncio.gather(watch("KNYC"), watch("KLAX"), watch("KORD"))

TypeScript

1
// Standard `for await` consume
2
for await (const row of stream("KNYC")) {
3
  process(row);
4
}
5

6
// Abort via AbortSignal — interrupts the polite-floor sleep promptly.
7
const controller = new AbortController();
8
setTimeout(() => controller.abort(), 60_000);
9

10
for await (const row of stream("KNYC", { signal: controller.signal })) {
11
  process(row);
12
}
13

14
// Multi-station fan-out
15
await Promise.all(
16
  ["KNYC", "KLAX", "KORD"].map(async (station) => {
17
    for await (const row of stream(station)) {
18
      process(station, row);
19
    }
20
  }),
21
);

FAQ

Can I write live rows to the parquet cache? No. live.stream() never writes to disk. If you want a durable back-of-tape, call research() separately.

Can I fuse AWC and IEM? Not in this surface — it’s intentionally single-source. Use research() for fused output. If you really want both channels live, run two stream() tasks in parallel and join them yourself.

What if poll_seconds is below the polite floor? ValueError is raised before the first poll with the floor and source named.

Can I run this in a browser? Yes — IEM has CORS open, AWC needs a proxy. See Browser integration for the supported pattern.