src/openaq.rs
//! **OpenAQ air quality** — the OpenAQ open air-quality API
//! (`https://api.openaq.org/v3/measurements`). The document is a JSON envelope
//! whose `results[]` array carries one **pollutant measurement** per element: a
//! nested `coordinates` object (`{latitude, longitude}`), a nested `date` object
//! (`{utc, local}`), a `parameter` code (`pm25` / `no2` / `o3` / …), a numeric
//! `value` + `unit`, and station identity (`locationId` / `location` / `country`
//! / `city`). This is a DISTINCT parser from the crate's other JSON collectors:
//! it flattens the air-quality `results[]` shape's TWO nested objects
//! (coordinates + date) into one measurement row → Arrow, where USGS decodes a
//! GeoJSON `FeatureCollection` geometry, PxWeb/Eurostat decode statistical cubes,
//! and the generic REST field-map carries no typed air-quality row.
//! [`parse_openaq_json`] flattens each result to one measurement row (station,
//! parameter, value, unit, lon/lat, UTC time) → Arrow.
//!
//! OpenAQ data is **CC BY 4.0** open data; v3 needs a free self-serve API key,
//! carried as the `X-API-Key` header via the collector's pluggable [`AuthKind`]
//! at collect time. The sample seam (`sample::openaq_try_get_real_sample_data`)
//! attempts a live fetch only when a URL is wired (endpoint/key are
//! deployment-specific); offline (or `KORP_COLLECTORS_OFFLINE=1`) it falls back
//! to an embedded synthetic `results` envelope.
//!
//! Offline contract: [`parse_openaq_json`] turns a canned `results` envelope into
//! per-measurement rows → Arrow; a result with no numeric `value` or no locatable
//! `coordinates` (missing/NaN lon or lat) contributes NO row (an all-unusable
//! document becomes a RAGNARÖK `reason`, never a silent empty).
use anyhow::Result;
use arrow_array::RecordBatch;
use arrow_schema::{DataType, Schema};
use serde_json::Value;
use super::arrow_util::{batch, f64_col, i64_col, schema, str_col};
use super::{
AuthKind, BatchSink, Bounds, Checkpoint, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
/// One flattened air-quality measurement: a single OpenAQ `results[]` element.
#[derive(Debug, Clone, PartialEq)]
pub struct AirQualityMeasurement {
/// OpenAQ location/station id (`locationId`) — nullable (some feeds omit it).
pub location_id: Option<i64>,
/// Human station name (`location`, e.g. `"Stockholm Torkel Knutssonsgatan"`).
pub location: Option<String>,
/// Pollutant parameter code (`parameter`, e.g. `"pm25"`, `"no2"`, `"o3"`) —
/// a measurement with no parameter is not a usable air-quality reading and
/// contributes no row.
pub parameter: String,
/// Measured concentration (`value`) — the reading. A result with no numeric
/// value is dropped (nothing to land).
pub value: f64,
/// Unit of the reading (`unit`, e.g. `"µg/m³"`, `"ppm"`).
pub unit: Option<String>,
/// Longitude (`coordinates.longitude`) — the map X. A result with no numeric
/// longitude is dropped (unlocatable).
pub longitude: f64,
/// Latitude (`coordinates.latitude`) — the map Y.
pub latitude: f64,
/// Country ISO code (`country`, e.g. `"SE"`).
pub country: Option<String>,
/// City / municipality (`city`).
pub city: Option<String>,
/// Measurement time as an ISO-8601 UTC string (`date.utc`) — the CDC /
/// partition key. Nullable (rare feeds omit it).
pub date_utc: Option<String>,
}
/// Flatten an OpenAQ `results` envelope → per-measurement air-quality rows.
///
/// Each `results[]` element must carry a numeric `value`, a `parameter` code, and
/// a `coordinates` object with numeric `latitude`/`longitude`; anything without a
/// value, a parameter, or a locatable point is skipped. `unit`, `locationId`,
/// `location`, `country`, `city` and `date.utc` are nullable. A `value` or a
/// coordinate quoted as a string is coerced (some mirrors stringify).
pub fn parse_openaq_json(json: &Value) -> Result<Vec<AirQualityMeasurement>> {
let mut out = Vec::new();
let Some(results) = json.get("results").and_then(|v| v.as_array()) else {
return Ok(out);
};
for r in results {
// A usable reading needs a numeric value and a parameter code.
let Some(value) = r.get("value").and_then(as_num) else {
continue;
};
let Some(parameter) = r
.get("parameter")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
else {
continue;
};
// The coordinates object holds the placeable station point.
let coords = r.get("coordinates");
let (Some(lon), Some(lat)) = (
coords.and_then(|c| c.get("longitude")).and_then(as_num),
coords.and_then(|c| c.get("latitude")).and_then(as_num),
) else {
continue; // unlocatable → no row
};
out.push(AirQualityMeasurement {
location_id: r.get("locationId").and_then(as_i64),
location: r
.get("location")
.and_then(|v| v.as_str())
.map(str::to_string),
parameter: parameter.to_string(),
value,
unit: r.get("unit").and_then(|v| v.as_str()).map(str::to_string),
longitude: lon,
latitude: lat,
country: r
.get("country")
.and_then(|v| v.as_str())
.map(str::to_string),
city: r.get("city").and_then(|v| v.as_str()).map(str::to_string),
date_utc: r
.get("date")
.and_then(|d| d.get("utc"))
.and_then(|v| v.as_str())
.map(str::to_string),
});
}
Ok(out)
}
/// A value / coordinate may arrive as a JSON number or a (rare) quoted string.
fn as_num(v: &Value) -> Option<f64> {
v.as_f64()
.or_else(|| v.as_str().and_then(|s| s.trim().parse().ok()))
}
/// A `locationId` may be a JSON integer or a numeric string.
fn as_i64(v: &Value) -> Option<i64> {
v.as_i64()
.or_else(|| v.as_f64().map(|f| f as i64))
.or_else(|| v.as_str().and_then(|s| s.trim().parse().ok()))
}
/// Measurements → one Arrow `RecordBatch`.
pub fn measurements_batch(rows: &[AirQualityMeasurement]) -> Result<RecordBatch> {
let cols = vec![
i64_col(&rows.iter().map(|m| m.location_id).collect::<Vec<_>>()),
str_col(&rows.iter().map(|m| m.location.clone()).collect::<Vec<_>>()),
str_col(
&rows
.iter()
.map(|m| Some(m.parameter.clone()))
.collect::<Vec<_>>(),
),
f64_col(&rows.iter().map(|m| Some(m.value)).collect::<Vec<_>>()),
str_col(&rows.iter().map(|m| m.unit.clone()).collect::<Vec<_>>()),
f64_col(&rows.iter().map(|m| Some(m.longitude)).collect::<Vec<_>>()),
f64_col(&rows.iter().map(|m| Some(m.latitude)).collect::<Vec<_>>()),
str_col(&rows.iter().map(|m| m.country.clone()).collect::<Vec<_>>()),
str_col(&rows.iter().map(|m| m.city.clone()).collect::<Vec<_>>()),
str_col(&rows.iter().map(|m| m.date_utc.clone()).collect::<Vec<_>>()),
];
batch(&openaq_schema(), cols)
}
pub fn openaq_schema() -> Schema {
schema(&[
("location_id", DataType::Int64, true),
("location", DataType::Utf8, true),
("parameter", DataType::Utf8, false),
("value", DataType::Float64, false),
("unit", DataType::Utf8, true),
("longitude", DataType::Float64, false),
("latitude", DataType::Float64, false),
("country", DataType::Utf8, true),
("city", DataType::Utf8, true),
("date_utc", DataType::Utf8, true),
])
}
/// The OpenAQ air-quality measurements collector.
#[derive(Default)]
pub struct OpenAqCollector;
#[async_trait::async_trait]
impl Collector for OpenAqCollector {
/// **OpenAQ v3 requires a free self-serve API key** (`X-API-Key`); v2 did not, and
/// the endpoint this collector declares is v3. Named, so the UI can say which key.
fn requirements(&self) -> crate::requires::Requirements {
crate::requires::Requirements::open(self.source_target().source_endpoint).with_secrets(
vec![crate::requires::SecretReq::new(
"OPENAQ_API_KEY",
"a free OpenAQ v3 API key (register at explore.openaq.org ▸ Account)",
)],
)
}
/// **The card whose first quirk is a URL that does not exist.** Measured against
/// OpenAQ's own live OpenAPI document on 2026-09-04: v3 publishes
/// `/v3/sensors/{sensors_id}/measurements` and there is NO bare `/v3/measurements`
/// path — which is exactly the endpoint `source_target()` declares. An operator would
/// otherwise spend the afternoon on the KEY, because every v3 path answers 401 without
/// one, a wrong path included.
///
/// Rate limit, key handling and terms read off `docs.openaq.org` the same day.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit, Registration};
Some(
HumanDoc::new(
"OpenAQ — the non-profit that runs the open air-quality platform",
"the national and local monitoring agencies whose stations OpenAQ \
aggregates; OpenAQ only republishes, and the API names each measurement's \
provider and owner through its own `/v3/providers` and `/v3/owners` \
resources — attribution is owed to the source, not to OpenAQ",
"https://docs.openaq.org/about/terms",
"https://docs.openaq.org/",
)
.with_contact(Contact::Email("platform@openaq.org".into()))
.with_registration(Registration::SelfService {
url: "https://explore.openaq.org/register".into(),
})
.with_rate_limit(RateLimit::Documented(
"60 requests/minute and 2 000/hour on the free tier, counted per API KEY".into(),
))
.with_quirk(
"**The declared base URL has no endpoint behind it.** Measured 2026-09-04 \
against OpenAQ's published OpenAPI document: v3 exposes \
`/v3/sensors/{sensors_id}/measurements` (plus `/hourly` and `/daily`) and no \
bare `/v3/measurements`. Point `cfg.base_url` at a sensor's measurements URL \
— the flattener does not care which one, it only needs the `results[]` shape.",
)
.with_quirk(
"Every v3 path answers HTTP 401 without a key — including a path that does \
not exist. So a 401 does NOT tell you whether the key or the URL is wrong. \
Prove the key against a known-good path (`/v3/locations`) first, then fix \
the URL.",
)
.with_quirk(
"v2 needed no key; v3 does, and this collector's endpoint is v3. The key \
travels in the `X-API-Key` HEADER, never the query string. Treat it like a \
password — if it leaks, rotate it from the OpenAQ Explorer account page.",
)
.with_quirk(
"Over the limit the API returns HTTP 429, and repeatedly exceeding it can \
earn a temporary or permanent ban. Every response carries \
`x-ratelimit-used` / `-remaining` / `-limit` / `-reset` headers — pace on \
those rather than on the published number, and a higher limit is a paid \
custom agreement, not a form.",
)
.with_quirk(
"A result with no numeric `value`, no `parameter`, or no locatable \
`coordinates` contributes NO row. An all-unusable document lands zero rows \
with a RAGNARÖK reason rather than a silent empty — read the reason before \
calling a quiet collect a failure.",
),
)
}
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "openaq-air-quality".into(),
source_kind: SourceKind::Rest,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["json".into()],
license: "OpenAQ air-quality data — CC BY 4.0".into(),
gdpr: GdprClass::Open,
auth: AuthKind::ApiKey,
join_keys: vec!["location_id".into(), "parameter".into()],
description: "OpenAQ air-quality measurements — each `results[]` reading flattened \
to a measurement row (station, parameter, value, unit, lon/lat, UTC \
time) → map-scene position table."
.into(),
}
}
fn source_target(&self) -> SourceTarget {
SourceTarget::map_scene(
SourceEndpoint::OpenAq,
"https://api.openaq.org/v3/measurements",
"openaq_measurements",
"osm3d",
)
}
/// **One measurement per (location, parameter, instant).**
///
/// The derived key stopped at `(location_id, parameter)`, which is a SERIES, not a
/// measurement: every reading a station has ever taken of PM2.5 keys identically, so
/// an Upsert would have kept one row per series and thrown the time axis away.
fn business_key(&self) -> crate::history::BusinessKey {
crate::history::BusinessKey::of(
&["location_id", "parameter", "date_utc"],
"one measurement per (location, parameter, instant) — `(location_id, \
parameter)` alone names a SERIES and would keep only its latest reading",
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(openaq_schema())
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let http = cfg.http(10);
let auth = cfg.auth.resolve_async().await?;
let body = http.get_text(&cfg.base_url, auth.as_ref()).await?;
let json: Value = serde_json::from_str(&body)?;
let rows = parse_openaq_json(&json)?;
if rows.is_empty() {
return Ok(CollectStats {
reason: Some(
"OpenAQ results carried no usable located measurements (quiet window / \
coordinate-less feed?)"
.into(),
),
..Default::default()
});
}
let n = rows.len() as u64;
sink.push(measurements_batch(&rows)?)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
async fn checkpoint(&self) -> Result<Option<Checkpoint>> {
Ok(None)
}
}
#[cfg(test)]
mod tests {
use super::*;
/// An OpenAQ `results` envelope flattens to one row per usable located
/// measurement; parameter/value/unit + the nested coordinates + `date.utc`
/// ride each row. Offline DATA contract (no network) + a testmatrix cell.
#[test]
fn openaq_results_flatten_to_measurement_rows() {
let json: Value = serde_json::json!({
"meta": { "found": 2, "page": 1 },
"results": [
{
"locationId": 8118, "location": "Stockholm Torkel Knutssonsgatan",
"parameter": "pm25", "value": 7.4, "unit": "µg/m³",
"country": "SE", "city": "Stockholm",
"coordinates": { "latitude": 59.3162, "longitude": 18.0575 },
"date": { "utc": "2026-07-14T08:00:00Z", "local": "2026-07-14T10:00:00+02:00" }
},
{
"locationId": 2178, "location": "Göteborg Femman",
"parameter": "no2", "value": 21.9, "unit": "µg/m³",
"country": "SE", "city": "Göteborg",
"coordinates": { "latitude": 57.7075, "longitude": 11.9675 },
"date": { "utc": "2026-07-14T08:00:00Z" }
}
]
});
let rows = parse_openaq_json(&json).expect("parse openaq");
assert_eq!(rows.len(), 2, "two located readings → two rows");
let sthlm = rows
.iter()
.find(|m| m.location_id == Some(8118))
.expect("Stockholm reading");
assert_eq!(sthlm.parameter, "pm25");
assert_eq!(sthlm.value, 7.4);
assert_eq!(sthlm.unit.as_deref(), Some("µg/m³"));
assert!(
(sthlm.longitude - 18.0575).abs() < 1e-9,
"lon = coordinates.longitude"
);
assert!(
(sthlm.latitude - 59.3162).abs() < 1e-9,
"lat = coordinates.latitude"
);
assert_eq!(sthlm.country.as_deref(), Some("SE"));
assert_eq!(
sthlm.date_utc.as_deref(),
Some("2026-07-14T08:00:00Z"),
"date.utc flattened"
);
let rb = measurements_batch(&rows).expect("measurements batch");
assert_eq!(rb.num_rows(), 2);
let schema = rb.schema();
let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
for want in [
"location_id",
"location",
"parameter",
"value",
"unit",
"longitude",
"latitude",
"country",
"city",
"date_utc",
] {
assert!(
cols.contains(&want),
"measurement table carries `{want}`: {cols:?}"
);
}
// `value`/`longitude` are Float64, `location_id` Int64.
let value = rb
.column_by_name("value")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Float64Array>())
.expect("value is f64");
let loc_id = rb
.column_by_name("location_id")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Int64Array>())
.expect("location_id is i64");
assert!((value.value(0) - 7.4).abs() < 1e-9);
assert_eq!(loc_id.value(0), 8118);
crate::functional_status(
"korp-collectors/openaq-air-quality",
"openaq_flatten",
true,
&format!(
"{} air-quality measurement rows from an OpenAQ results envelope",
rb.num_rows()
),
);
}
/// Nullable fields (absent unit/locationId/date, quoted numeric strings) are
/// tolerated, and a valueless / parameterless / unlocatable result contributes
/// NO row.
#[test]
fn openaq_nullable_props_and_string_coercion_and_unusable_dropped() {
let json: Value = serde_json::json!({
"results": [
// valid, but null unit + no locationId + value & coords as strings
{
"parameter": "o3", "value": "48.0", "unit": null,
"coordinates": { "latitude": "63.82", "longitude": "20.26" }
},
// no numeric value → dropped
{
"locationId": 1, "parameter": "pm10", "value": null,
"coordinates": { "latitude": 59.0, "longitude": 18.0 }
},
// no parameter → dropped
{
"locationId": 2, "value": 3.0,
"coordinates": { "latitude": 59.0, "longitude": 18.0 }
},
// no coordinates → unlocatable, dropped
{
"locationId": 3, "parameter": "so2", "value": 1.0
},
// non-numeric longitude → unlocatable, dropped
{
"locationId": 4, "parameter": "co", "value": 0.5,
"coordinates": { "latitude": 59.0, "longitude": "not-a-number" }
}
]
});
let rows = parse_openaq_json(&json).expect("parse");
assert_eq!(
rows.len(),
1,
"only the one usable located reading survives"
);
let a = &rows[0];
assert_eq!(a.parameter, "o3");
assert_eq!(a.value, 48.0, "string value coerced");
assert_eq!(a.unit, None, "null unit → None (nullable column)");
assert_eq!(a.location_id, None, "absent locationId → None");
assert!(
(a.longitude - 20.26).abs() < 1e-9,
"string longitude coerced"
);
assert!((a.latitude - 63.82).abs() < 1e-9, "string latitude coerced");
assert_eq!(a.date_utc, None, "absent date → None");
// The batch still builds with the null unit / location_id cells.
let rb = measurements_batch(&rows).expect("batch");
assert_eq!(rb.num_rows(), 1);
let unit = rb
.column_by_name("unit")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::StringArray>())
.expect("unit utf8");
assert!(
arrow_array::Array::is_null(&unit, 0),
"the null-unit cell is null"
);
}
/// A document with no usable measurements yields zero rows — the collector
/// maps that to a RAGNARÖK reason (the RED-when-broken empty guard).
#[test]
fn openaq_empty_or_unusable_document_yields_no_rows() {
let empty: Value = serde_json::json!({ "meta": {}, "results": [] });
assert!(parse_openaq_json(&empty).expect("parse").is_empty());
// A document whose only result has no coordinates contributes nothing.
let no_coords: Value = serde_json::json!({
"results": [ { "parameter": "pm25", "value": 5.0 } ]
});
assert!(parse_openaq_json(&no_coords).expect("parse").is_empty());
// A missing `results` key is an empty (not a panic).
assert!(
parse_openaq_json(&serde_json::json!({}))
.expect("parse")
.is_empty()
);
}
}