mostlyright.weather.cwop

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, the ONLY real-time path being a socket stream (no REST). This module is deliberately STANDALONE: it never feeds research(), the merge layer, or live._sources, and it carries its own schema.cwop.v1 (NOT schema.observation.v1). A hot-rooftop or indoor PWS must never silently corrupt a Kalshi NHIGH/NLOW settlement target, so CWOP stays on its own island with a 6-layer QC pipeline (see weather.qc._cwop).

Public surface (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 lat/lon 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 backtest cache).
• latest() — the most-recent observation for one station.

Persistence + backtest (v0.2):

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

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, lat, lon, raw_aprs, temp_f=None, humidity=None, pressure_mb=None, wind_speed_mph=None, wind_dir_deg=None, wind_gust_mph=None, rain_1h_in=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)
  • lat (float | None)
  • lon (float | None)
  • raw_aprs (str)
  • temp_f (float | None)
  • humidity (int | None)
  • pressure_mb (float | None)
  • wind_speed_mph (float | None)
  • wind_dir_deg (int | None)
  • wind_gust_mph (float | None)
  • rain_1h_in (float | None)
  • luminosity (int | None)
  • source (str)
  • qc_score (float | None)
  • qc_status (str)

humidity : int | None = None

knowledge_time : datetime

lat : float | None

lon : float | None

luminosity : int | None = None

observed_at : datetime

pressure_mb : float | None = None

qc_score : float | None = None

qc_status : str = ‘unknown’

rain_1h_in : float | None = None

raw_aprs : str

source : str = ‘cwop.live’

station_id : str

temp_f : float | None = None

wind_dir_deg : int | None = None

wind_gust_mph : float | None = None

wind_speed_mph : float | None = None

class mostlyright.weather.cwop.CWOPStation(id, lat, lon, distance_km, last_obs, obs_count, last_temp_f, report_cadence_s, 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_s 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)
  • lat (float | None)
  • lon (float | None)
  • distance_km (float)
  • last_obs (datetime)
  • obs_count (int)
  • last_temp_f (float | None)
  • report_cadence_s (float | None)
  • qc_status (str)

distance_km : float

id : str

last_obs : datetime

last_temp_f : float | None

lat : float | None

lon : float | None

obs_count : int

qc_status : str = ‘unknown’

report_cadence_s : 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

default_error_code : str = ‘NO_CWOP_DATA’

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 backtest-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 (str | None) – optional filter — keep only rows with this QC verdict ("clean" / "flagged" / "dropped" / "unknown"). A backtest 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)

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

Blocks until the first packet arrives or timeout_seconds elapses.

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

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 distinct observations written (post-dedup is applied per partition by write_cwop_cache(); this count is the number SUBMITTED).

This is the collection-side seam: feed it the output of snapshot() / stream() to build a persisted CWOP history for backtesting.

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

mostlyright.weather.cwop.scan(lat, lon, radius_km=25.0, , listen_seconds=60, min_reports=1)

Scan APRS-IS for CWOP stations near (lat, lon). Synchronous.

Cache-then-augment: when stations.json already holds at least min_reports individually-fresh (<24h since their own last_obs) 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:
  • lat (float)
  • lon (float)
  • radius_km (float)
  • listen_seconds (int)
  • min_reports (int)

mostlyright.weather.cwop.snapshot(station, , duration_seconds=60, qc=True, persist=False)

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 qc=True.

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 backtesting. 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)
  • qc (bool)
  • persist (bool)

async mostlyright.weather.cwop.stream(station, , qc=True)

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

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

• 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 ])
  • qc (bool)