mostlyright.weather.cwop

mostlyright.weather.cwop — standalone CWOP (Citizen Weather Observer Program) live adapter.

CWOP is Personal Weather Station (PWS) data delivered over the APRS-IS TCP network — ~7K active North American stations; CWOP publishes no official REST API, so the real-time path is the APRS-IS socket stream. This module is standalone: it never feeds research(), the merge code, or live._sources, and it carries its own schema.cwop.v1 rather than schema.observation.v1. A hot-rooftop or indoor PWS must never silently corrupt a Kalshi NHIGH/NLOW settlement target, so CWOP stays isolated behind a 6-layer QC pipeline (see weather.cwop._qc).

Public functions (all source-tagged "cwop.live"):

• nearby() — the “start here” entry point: resolve an ICAO/NWS station to coordinates and scan around it.
• scan() — discover stations near a latitude/longitude by listening on APRS-IS.
• stream() — async generator of fresh observations.
• snapshot() — collect a fixed window into a schema.cwop.v1 DataFrame (pass persist=True to also write it to the research cache).
• latest() — the most-recent observation for one station.

Persistence and research replay:

• history() — replay persisted CWOP observations for a date range as a schema.cwop.v1 DataFrame (source="cwop.cache"). This is the parity-safe access path for ML strategies: CWOP is never wired into research() / merge / live._sources, so a strategy joins this frame itself.
• persist_observations() — write a list of observations to the monthly parquet cache. Feed it snapshot() / stream() output to collect a history.

Data classes: CWOPStation, CWOPObservation. Errors: NoCWOPDataError.

Importing this module registers schema.cwop.v1 with the validator (lazy — a base install that never touches CWOP pays nothing).

class mostlyright.weather.cwop.CWOPObservation(station_id, observed_at, knowledge_time, latitude, longitude, raw_aprs, temp_f=None, humidity=None, pressure_mb=None, wind_speed_mph=None, wind_direction_degrees=None, wind_gust_mph=None, precipitation_in_1h=None, luminosity=None, source=‘cwop.live’, qc_score=None, qc_status=‘unknown’)

Bases: object

One CWOP weather observation (the row the public surface yields).

observed_at is the APRS packet timestamp (event-time); knowledge_time is when our socket received the packet (the event_time + lag convention — for CWOP the lag is the network/queue delay). qc_score/qc_status are populated by the QC pipeline when qc=True; otherwise they stay None/"unknown".

• Parameters:
  • station_id (str)
  • observed_at (datetime)
  • knowledge_time (datetime)
  • latitude (float | None)
  • longitude (float | None)
  • raw_aprs (str)
  • temp_f (float | None)
  • humidity (int | None)
  • pressure_mb (float | None)
  • wind_speed_mph (float | None)
  • wind_direction_degrees (int | None)
  • wind_gust_mph (float | None)
  • precipitation_in_1h (float | None)
  • luminosity (int | None)
  • source (str)
  • qc_score (float | None)
  • qc_status (str)

humidity: int | None

knowledge_time: datetime

latitude: float | None

longitude: float | None

luminosity: int | None

observed_at: datetime

precipitation_in_1h: float | None

pressure_mb: float | None

qc_score: float | None

qc_status: str

raw_aprs: str

source: str

station_id: str

temp_f: float | None

wind_direction_degrees: int | None

wind_gust_mph: float | None

wind_speed_mph: float | None

class mostlyright.weather.cwop.CWOPStation(id, latitude, longitude, distance_km, last_observation_at, observation_count, last_temp_f, report_cadence_seconds, qc_status=‘unknown’)

Bases: object

A CWOP station discovered on the APRS-IS stream.

distance_km is the great-circle distance from the scan centre. report_cadence_seconds is the median seconds between this station’s reports (None until it has reported twice). qc_status is a forward-looking trust signal, "unknown" until a QC pass annotates the station.

• Parameters:
  • id (str)
  • latitude (float | None)
  • longitude (float | None)
  • distance_km (float)
  • last_observation_at (datetime)
  • observation_count (int)
  • last_temp_f (float | None)
  • report_cadence_seconds (float | None)
  • qc_status (str)

distance_km: float

id: str

last_observation_at: datetime

last_temp_f: float | None

latitude: float | None

longitude: float | None

observation_count: int

qc_status: str

report_cadence_seconds: float | None

exception mostlyright.weather.cwop.NoCWOPDataError(station, reason, , request_id=None, error_code=None)

Bases: NoLiveDataError

No CWOP (APRS-IS) data available for the requested station/area.

Raised by the standalone mostlyright.weather.cwop surface (scan/nearby/stream/snapshot/latest) instead of returning []/None, so callers get an actionable signal — CWOP is a live-only TCP stream and “nothing arrived” is the common failure mode (the station never reported, the area is empty, or the listen window was too short). Carries the resolved station/area identifier, the fixed source="cwop.live" tag, and a human-readable reason.

• Parameters:
  • station (str)
  • reason (str)
  • request_id (str | None)
  • error_code (str)
• Return type: None

cause: str | None

default_error_code: str

Subclass override — the stable string enum surfaced via error_code.

error_code: str

message: str

reason: str

request_id: str | None

source: str | None

station: str

mostlyright.weather.cwop.history(station, from_date, to_date, , qc_status=None)

Return persisted CWOP observations for station in [from_date, to_date].

The research-replay entry point: reads the monthly parquet cache and returns a schema.cwop.v1 DataFrame (source="cwop.cache", validated before return) that an ML strategy can join against its official-obs features. The date range is inclusive on both ends; bare date arguments cover the whole calendar day in UTC.

• Parameters:
  • station (str) – CWOP station id (e.g. "CW0875").
  • from_date (date | datetime) – window start (inclusive). date or datetime.
  • to_date (date | datetime) – window end (inclusive). date or datetime.
  • qc_status (Literal['clean', 'flagged', 'dropped', 'unknown'] | None) – optional filter — keep only rows with this QC verdict ("clean" / "flagged" / "dropped" / "unknown"). Research typically passes "clean" to drop QC-rejected PWS rows.
• Return type: DataFrame
• Returns: A schema.cwop.v1 DataFrame sorted by observed_at.
• Raises:
  • ValueError – from_date is after to_date, or qc_status is not a recognized QC verdict.
  • NoCWOPDataError – no persisted CWOP data in range (after the optional qc_status filter) — never returns an empty frame.

mostlyright.weather.cwop.latest(station, , timeout_seconds=120, buddy_station=None)

Return the most-recent observation for a CWOP station. Synchronous.

Blocks until the first packet arrives or timeout_seconds elapses. QC scores against the station’s persisted history (there is no in-session history for a single observation); buddy_station is the opt-in official ICAO reference for the buddy layer.

• Raises: NoCWOPDataError – nothing received within the timeout.
• Return type: CWOPObservation
• Parameters:
  • station (str)
  • timeout_seconds (int)
  • buddy_station (str | None)

mostlyright.weather.cwop.nearby(station, , radius_km=25.0, listen_seconds=60)

Resolve an ICAO/NWS station to coordinates, then scan() around it.

The “start here” entry point for most users.

• Raises: NoCWOPDataError – the station code is unknown, or no CWOP stations reported nearby.
• Return type: list[CWOPStation]
• Parameters:
  • station (str)
  • radius_km (float)
  • listen_seconds (int)

mostlyright.weather.cwop.persist_observations(observations)

Persist live CWOPObservation rows to the monthly parquet cache.

Groups observations by (station_id, year, month) of their UTC observed_at (so a window straddling a month boundary lands in the right partitions) and merge-writes each partition. Returns the number of observations submitted; write_cwop_cache() applies the dedup per partition after that count is taken.

Feed it the output of snapshot() / stream() to build a persisted CWOP history for research.

• Return type: int
• Parameters: observations (list [CWOPObservation ])

mostlyright.weather.cwop.scan(latitude, longitude, radius_km=25.0, , listen_seconds=60, min_reports=1)

Scan APRS-IS for CWOP stations near (latitude, longitude). Synchronous.

Cache-then-augment: when stations.json already holds at least min_reports individually-fresh (<24h since their own last_observation_at) stations inside radius_km, they are returned immediately; a stale or sparse cache triggers a live listen of up to listen_seconds (early-exit once min_reports stations are seen). Live results are merged back into the registry.

Returns stations sorted nearest-first.

• Raises: NoCWOPDataError – no stations found after the listen window.
• Return type: list[CWOPStation]
• Parameters:
  • latitude (float)
  • longitude (float)
  • radius_km (float)
  • listen_seconds (int)
  • min_reports (int)

mostlyright.weather.cwop.snapshot(station, , duration_seconds=60, quality_control=True, persist=False, buddy_station=None)

Collect CWOP observations for duration_seconds into a DataFrame.

Synchronous. Returns a schema.cwop.v1 frame (validated before return), with QC scores/status populated when quality_control=True.

QC scores against the station’s PERSISTED history (read-only, from the parquet cache) as well as the in-session rolling one, so the indoor, solar-bias and reliability layers activate once a station has accumulated a day / week / month of collected data. buddy_station is the opt-in official reference (an ICAO, e.g. "KNYC") for the buddy-check layer: set it and the nearest official reading is fetched once per call, best-effort. Leave it unset and that layer keeps passing.

When persist=True the collected observations are also written to the monthly parquet cache ($HOME/.mostlyright/cache/cwop/...) so they can be replayed later via mostlyright.weather.cwop.history() for research. Persistence is a side effect — the returned (live) frame is unchanged and still tagged source="cwop.live"; only the on-disk rows carry the "cwop.cache" provenance tag.

• Raises: NoCWOPDataError – no observations arrived during the window.
• Return type: DataFrame
• Parameters:
  • station (str | list [str ])
  • duration_seconds (int)
  • quality_control (bool)
  • persist (bool)
  • buddy_station (str | None)

async mostlyright.weather.cwop.stream(station, , quality_control=True, buddy_station=None)

Yield fresh CWOP observations for station (one or many). Async.

Dedups by (station_id, observed_at); applies the QC pipeline when quality_control=True (default). Source tag is "cwop.live".

QC scores against each station’s persisted history as well as the in-session one, so the indoor / solar-bias / reliability layers activate once enough collected data exists. buddy_station is the opt-in official ICAO reference for the buddy layer (refreshed periodically over a long stream).

• Raises: NoCWOPDataError – the station never reported within FIRST_OBS_TIMEOUT_S (raised before the first yield only).
• Return type: AsyncIterator[CWOPObservation]
• Parameters:
  • station (str | list [str ])
  • quality_control (bool)
  • buddy_station (str | None)