//! **ERDDAP** collector — a uniform reader over the ERDDAP scientific-data
//! server family (EMODnet ocean data + the global NOAA/CoastWatch and academic
//! ERDDAP deployments). ERDDAP fronts **any** gridded (`griddap`) or tabular
//! (`tabledap`) dataset behind ONE query grammar and a fixed set of response
//! formats; the `.json` form is **self-describing** — a single `table` object
//! carrying `columnNames`, `columnTypes` and a `rows` matrix — so ONE parser
//! decodes every dataset on every server with no per-dataset code.
//!
//! [`parse_erddap_json`] maps the ERDDAP `columnTypes` to Arrow types
//! (`String`→Utf8, `float`/`double`→Float64, `byte`/`short`/`int`/`long`→Int64)
//! and unpacks the `rows` matrix column-wise → a typed `RecordBatch`. Pure-Rust
//! `serde_json`, no new deps. The `.csv` response variant (a header row + a units
//! row + data) is reachable through the generic [`crate::csv`] collector; this
//! module owns the richer, self-typed `.json` path.
//!
//! The concrete server + dataset id ride `cfg.base_url` at collect time (the full
//! `…/tabledap/<id>.json?<query>` request URL), so the same collector serves any
//! ERDDAP endpoint. Presets ([`ErddapCollector::emodnet_physics`],
//! [`ErddapCollector::noaa_coastwatch`]) carry a representative server for
//! discovery / the sample seam.
use anyhow::{Result, anyhow};
use arrow_array::{ArrayRef, 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, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
/// The Arrow type an ERDDAP `columnTypes[i]` maps to. ERDDAP column types are a
/// small fixed vocabulary (the CF / netCDF numeric kinds plus `String`/`char`).
fn erddap_arrow_type(erddap_type: &str) -> DataType {
match erddap_type {
"double" | "float" => DataType::Float64,
"long" | "int" | "short" | "byte" | "ubyte" | "ushort" | "uint" | "ulong" => {
DataType::Int64
}
// "String", "char" and anything unrecognised land as Utf8 (lossless).
_ => DataType::Utf8,
}
}
fn cell_as_string(v: &Value) -> Option<String> {
match v {
Value::Null => None,
Value::String(s) => Some(s.clone()),
other => Some(other.to_string()),
}
}
fn cell_as_i64(v: &Value) -> Option<i64> {
match v {
Value::Null => None,
_ => 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())),
}
}
fn cell_as_f64(v: &Value) -> Option<f64> {
match v {
Value::Null => None,
_ => v.as_f64().or_else(|| {
v.as_str().and_then(|s| {
// ERDDAP writes missing numeric values as the string "NaN".
let t = s.trim();
if t.eq_ignore_ascii_case("nan") {
None
} else {
t.parse().ok()
}
})
}),
}
}
/// Parse an ERDDAP `.json` `table` document → a typed Arrow `RecordBatch`. The
/// **offline DATA contract**: `columnNames` + `columnTypes` name+type each
/// column, `rows` is the row-major value matrix. A row shorter than the header
/// (ERDDAP never emits ragged rows, but be defensive) yields nulls for the
/// missing trailing cells.
pub fn parse_erddap_json(json: &Value) -> Result<RecordBatch> {
let table = json
.get("table")
.ok_or_else(|| anyhow!("ERDDAP response has no `table` object"))?;
let names = table
.get("columnNames")
.and_then(|v| v.as_array())
.ok_or_else(|| anyhow!("ERDDAP `table.columnNames` missing or not an array"))?;
let types = table
.get("columnTypes")
.and_then(|v| v.as_array())
.ok_or_else(|| anyhow!("ERDDAP `table.columnTypes` missing or not an array"))?;
let rows = table
.get("rows")
.and_then(|v| v.as_array())
.ok_or_else(|| anyhow!("ERDDAP `table.rows` missing or not an array"))?;
if names.len() != types.len() {
return Err(anyhow!(
"ERDDAP columnNames ({}) / columnTypes ({}) length mismatch",
names.len(),
types.len()
));
}
let col_names: Vec<String> = names
.iter()
.enumerate()
.map(|(i, v)| {
v.as_str()
.map(str::to_string)
.unwrap_or_else(|| format!("col_{i}"))
})
.collect();
let col_types: Vec<DataType> = types
.iter()
.map(|v| erddap_arrow_type(v.as_str().unwrap_or("String")))
.collect();
// Column-wise unpack of the row-major `rows` matrix.
let mut columns: Vec<ArrayRef> = Vec::with_capacity(col_names.len());
for (c, col_type) in col_types.iter().enumerate() {
let cells = rows.iter().map(|row| row.as_array().and_then(|r| r.get(c)));
let col: ArrayRef = match col_type {
DataType::Float64 => {
f64_col(&cells.map(|c| c.and_then(cell_as_f64)).collect::<Vec<_>>())
}
DataType::Int64 => i64_col(&cells.map(|c| c.and_then(cell_as_i64)).collect::<Vec<_>>()),
_ => str_col(
&cells
.map(|c| c.and_then(cell_as_string))
.collect::<Vec<_>>(),
),
};
columns.push(col);
}
let fdefs: Vec<(&str, DataType, bool)> = col_names
.iter()
.zip(&col_types)
.map(|(n, t)| (n.as_str(), t.clone(), true))
.collect();
batch(&schema(&fdefs), columns)
}
/// **The quirks every ERDDAP deployment shares**, applied on top of a per-server card.
/// ERDDAP is ONE piece of software behind both presets, so these belong to the software
/// rather than to EMODnet or NOAA — written once and appended to each card rather than
/// twinned (LAW 5).
///
/// The three MEASURED lines were taken against `coastwatch.pfeg.noaa.gov` on 2026-09-04
/// from this machine; the licence line is what ERDDAP's own Legal Notices page states.
fn with_erddap_software_quirks(doc: crate::human_doc::HumanDoc) -> crate::human_doc::HumanDoc {
doc.with_quirk(
"**A query that matches nothing is an HTTP ERROR, not an empty table.** Measured \
2026-09-04: an out-of-range time window answered HTTP 404 with the body \
`Error { code=404; message=\"Not Found: Your query produced no matching results. \
…\" }`. So this collector's \"returned an empty table\" RAGNARÖK reason can only \
fire for a table ERDDAP actually served — an empty WINDOW surfaces as a fetch \
failure instead, and reads like a broken endpoint when it is a quiet one.",
)
.with_quirk(
"An unknown or mis-cased variable name is HTTP 400 `Unrecognized variable=\"…\"` \
(measured 2026-09-04). ERDDAP variable names are CASE-SENSITIVE — `WTMP`, not \
`wtmp`. Check them against `…/info/<datasetID>/index.json`, which lists every \
variable with its metadata.",
)
.with_quirk(
"The preset's `sample_url()` carries NO query string, and ERDDAP reads that as \
*every variable, every row*. Always supply variables and constraints through \
`cfg.base_url`; measured 2026-09-04, the unconstrained CoastWatch request was \
refused with a bare HTTP 429 `Too Many Requests` HTML page — not the ERDDAP JSON \
error envelope, so the parser sees garbage rather than a diagnosis.",
)
.with_quirk(
"Licences are PER DATASET, not per server: ERDDAP's own Legal Notices warn that \
some datasets carry usage limitations or restrictions, and point at the dataset's \
own licence metadata on its Data Access Form (`.html`), Make-A-Graph (`.graph`) \
and Dataset Attribute Structure (`.das`) pages. Read it before redistributing — \
the server-level licence on `CollectorMeta` is the default, not a promise.",
)
.with_quirk(
"`columnUnits` rides the `.json` table alongside `columnNames`/`columnTypes`, and \
this parser DROPS it — a Float64 column lands dimensionless in Arrow. If units \
matter, read them from `…/info/<datasetID>/index.json`. Missing numeric cells \
arrive as JSON `null` or the string `\"NaN\"`; both become Arrow nulls, never 0.0.",
)
}
/// The uniform ERDDAP collector. `cfg.base_url` is the full ERDDAP request URL
/// (`…/tabledap/<dataset>.json?<query>` or the `griddap` equivalent) supplied at
/// collect time; the preset carries a representative server for discovery.
pub struct ErddapCollector {
name: String,
description: String,
license: String,
/// The ERDDAP server base (`…/erddap`) — the representative discovery URL.
server: String,
/// The access protocol this preset represents (`tabledap` / `griddap`).
protocol: String,
/// A representative dataset id on that server.
dataset_id: String,
}
impl ErddapCollector {
/// **EMODnet Physics** ERDDAP — the European Marine Observation and Data
/// Network physics node (in-situ ocean physics: waves, currents, sea level,
/// temperature). `tabledap` over the moored/drifting-platform datasets.
pub fn emodnet_physics() -> Self {
Self {
name: "erddap-emodnet-physics".into(),
description:
"EMODnet Physics ERDDAP (in-situ ocean physics: sea level, waves, currents, \
temperature) → typed Arrow via the self-describing .json table."
.into(),
license: "CC-BY 4.0 (EMODnet)".into(),
server: "https://erddap.emodnet-physics.eu/erddap".into(),
protocol: "tabledap".into(),
dataset_id: "EP_ERD_INT_TZTM_AL_TS_NRT".into(),
}
}
/// **NOAA CoastWatch / ERD** ERDDAP — the canonical global reference server
/// (satellite SST, chlorophyll, winds; both `griddap` grids and `tabledap`
/// tables). Stands in for the wider global ocean/atmosphere ERDDAP fleet.
pub fn noaa_coastwatch() -> Self {
Self {
name: "erddap-noaa-coastwatch".into(),
description:
"NOAA CoastWatch/ERD ERDDAP (global ocean/atmosphere: SST, chlorophyll, winds) \
→ typed Arrow via the self-describing .json table."
.into(),
license: "public domain (US Government / NOAA)".into(),
server: "https://coastwatch.pfeg.noaa.gov/erddap".into(),
protocol: "tabledap".into(),
dataset_id: "cwwcNDBCMet".into(),
}
}
/// The representative ERDDAP server base (`…/erddap`).
pub fn server(&self) -> &str {
&self.server
}
/// The access protocol (`tabledap` / `griddap`).
pub fn protocol(&self) -> &str {
&self.protocol
}
/// A representative dataset id on the server.
pub fn dataset_id(&self) -> &str {
&self.dataset_id
}
/// The representative `.json` request URL (server + protocol + dataset) used
/// by discovery and the sample seam. A real collect overrides it (with the
/// dataset variables + constraints) through `cfg.base_url`.
pub fn sample_url(&self) -> String {
format!("{}/{}/{}.json", self.server, self.protocol, self.dataset_id)
}
}
#[async_trait::async_trait]
impl Collector for ErddapCollector {
/// **Two presets, two authorities, two cards.** They share a struct and the ERDDAP
/// software's quirks (see `with_erddap_software_quirks` in this module), but EMODnet
/// Physics and
/// NOAA CoastWatch are different institutions with different contacts, different terms
/// and — as measured below — very different health. An unknown preset gets `None`
/// rather than somebody else's card.
///
/// Verified 2026-09-04: both servers' Legal Notices, and live probes of both.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit, Registration};
let doc = match self.name.as_str() {
"erddap-emodnet-physics" => HumanDoc::new(
"EMODnet Physics — the in-situ physics thematic node of the European Marine \
Observation and Data Network, run for the European Commission",
"the originating observing programmes and national data providers whose \
platforms EMODnet aggregates; each dataset's own ERDDAP metadata names its \
source and its licence",
"https://emodnet.ec.europa.eu/en/terms-use-emodnet-online-services-data-and-data-products",
"https://erddap.emodnet-physics.eu/erddap/tabledap/documentation.html",
)
.with_contact(Contact::Form(
"https://emodnet.ec.europa.eu/en/contact-us".into(),
))
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::Unknown)
.with_quirk(
"**The preset's representative dataset id is STALE.** Measured 2026-09-04: \
`…/tabledap/EP_ERD_INT_TZTM_AL_TS_NRT.json` answers HTTP 404 `Currently \
unknown datasetID`, while the server was serving 679 active tabledap \
datasets. `sample_url()` is a DISCOVERY hint, not a working URL — pick a \
live id from `…/tabledap/allDatasets.json?datasetID,title` before you \
configure a collect.",
)
.with_quirk(
"This server publishes no working contact of its own: its ERDDAP Legal \
Notices page still carries the unedited ERDDAP placeholder address \
(`set-me at domain dot com`, seen 2026-09-04). Questions go through the \
EMODnet contact form above, not to the server admin.",
),
"erddap-noaa-coastwatch" => HumanDoc::new(
"NOAA — NMFS SWFSC Environmental Research Division (ERD), the CoastWatch West \
Coast Node, which operates this ERDDAP (\"Brought to you by NOAA NMFS SWFSC \
ERD\", per its own Legal Notices page)",
"NOAA and the originating providers ERD republishes — the preset's own \
reference dataset names `contributor_name = NOAA NDBC` in its metadata, and \
every dataset names its own source the same way",
"https://coastwatch.pfeg.noaa.gov/erddap/legal.html",
"https://coastwatch.pfeg.noaa.gov/erddap/tabledap/documentation.html",
)
.with_contact(Contact::Email("erd.data@noaa.gov".into()))
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::Unknown)
.with_quirk(
"The preset dataset `cwwcNDBCMet` (NDBC buoy meteorology) is LIVE — measured \
2026-09-04, a two-hour single-station window returned a 200 `.json` table \
with 10-minute rows and a JSON `null` for a missing `WTMP`. Use it as the \
shape reference when wiring a new dataset.",
),
// A preset nobody has written a card for gets no card, rather than the wrong
// institution's contact details.
_ => return None,
};
Some(with_erddap_software_quirks(doc))
}
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: self.name.clone(),
direction: Direction::Source,
source_kind: SourceKind::Rest,
path: CollectorPath::RustNative,
formats: vec!["json".into(), "csv".into()],
license: self.license.clone(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["time".into(), "coordinate".into()],
description: self.description.clone(),
}
}
fn source_target(&self) -> SourceTarget {
SourceTarget::warehouse(
SourceEndpoint::Erddap,
self.sample_url(),
self.name.replace('-', "_"),
)
}
/// **Append-only, and it says why.**
///
/// An ERDDAP table's columns come from the dataset (`columnNames`/`columnTypes` in
/// the `.json` response), so they are not knowable at declaration time and are not
/// even knowable from the cfg — only from the response. The derived key named
/// `coordinate`, which no ERDDAP dataset emits under that name (it is
/// `latitude`/`longitude`, or a station id, or a cast number, per dataset).
///
/// Rather than guess, this declares append-only: an ERDDAP series re-read over an
/// overlapping time range genuinely re-lands its rows, and saying so is the honest
/// answer until a `params.business_key` lets an operator name the dataset's own
/// identity columns.
fn business_key(&self) -> crate::history::BusinessKey {
crate::history::BusinessKey::append_only(
"an ERDDAP table's columns come from the DATASET (`columnNames` in the \
`.json` response), so its identity is not knowable before the response \
arrives — `coordinate` is a join-spine word, not a column any ERDDAP \
dataset emits",
)
}
fn schema(&self, cfg: &CollectorCfg) -> Result<Schema> {
// ERDDAP schemas are dataset-specific and only knowable from the `.json`
// `columnTypes`, which needs the (remote) response — so the schema is
// derived in `collect()`. A caller that needs it up front may pass an
// ERDDAP `table` document under `params.schema_sample`.
if let Some(sample) = cfg.params.get("schema_sample") {
return Ok(parse_erddap_json(sample)?.schema().as_ref().clone());
}
Err(anyhow!(
"ERDDAP schema is derived in collect() from the .json table (columnNames/columnTypes); \
pass a `table` document under params.schema_sample to resolve it up front"
))
}
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 rb = parse_erddap_json(&json)?;
let n = rb.num_rows() as u64;
if n == 0 {
return Ok(CollectStats {
reason: Some(format!("ERDDAP `{}` returned an empty table", cfg.base_url)),
..Default::default()
});
}
sink.push(rb)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use arrow_array::Array;
/// An ERDDAP tabledap `.json` `table` document decodes to a typed batch: the
/// `columnTypes` drive Arrow types (String→Utf8, double/float→Float64,
/// int→Int64) and the `rows` matrix unpacks column-wise. RED if the type map
/// or the column-wise unpack breaks. The offline DATA contract (no network).
#[test]
fn erddap_table_decodes_to_typed_columns() {
// Two NDBC-buoy rows: a time string, lat/lon doubles, an int station id,
// an SST float — one carrying a null (missing) SST.
let json: Value = serde_json::json!({
"table": {
"columnNames": ["time", "latitude", "longitude", "wmo_id", "sea_surface_temperature"],
"columnTypes": ["String", "double", "double", "int", "float"],
"rows": [
["2026-07-18T00:00:00Z", 55.02, 10.51, 6201, 17.8],
["2026-07-18T01:00:00Z", 55.02, 10.51, 6201, null]
]
}
});
let rb = parse_erddap_json(&json).expect("parse erddap table");
assert_eq!(rb.num_rows(), 2, "two table rows decoded");
assert_eq!(rb.num_columns(), 5, "five columns");
// Types came from columnTypes.
let s = rb.schema();
let by = |n: &str| s.field_with_name(n).unwrap().data_type().clone();
assert_eq!(by("time"), DataType::Utf8, "String → Utf8");
assert_eq!(by("latitude"), DataType::Float64, "double → Float64");
assert_eq!(by("wmo_id"), DataType::Int64, "int → Int64");
assert_eq!(
by("sea_surface_temperature"),
DataType::Float64,
"float → Float64"
);
let lat = rb
.column_by_name("latitude")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Float64Array>())
.expect("lat f64");
assert!((lat.value(0) - 55.02).abs() < 1e-9, "latitude decoded");
let wmo = rb
.column_by_name("wmo_id")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Int64Array>())
.expect("wmo i64");
assert_eq!(wmo.value(0), 6201, "int station id decoded");
let sst = rb
.column_by_name("sea_surface_temperature")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Float64Array>())
.expect("sst f64");
assert!((sst.value(0) - 17.8).abs() < 1e-9, "sst decoded");
assert!(sst.is_null(1), "missing SST → null (not 0.0)");
// Typed source endpoint + a warehouse target.
let c = ErddapCollector::emodnet_physics();
let st = c.source_target();
assert_eq!(st.source, SourceEndpoint::Erddap, "typed ERDDAP endpoint");
assert!(
st.source_endpoint.contains("erddap.emodnet-physics.eu"),
"EMODnet server"
);
assert!(
c.sample_url().ends_with(".json"),
"sample URL is the .json form"
);
crate::functional_status(
"korp-collectors/erddap",
"table_decode",
true,
&format!("{} ERDDAP rows → typed Arrow", rb.num_rows()),
);
}
/// `columnNames` / `columnTypes` length mismatch is a hard error (a corrupt
/// ERDDAP response, not silently truncated).
#[test]
fn erddap_ragged_header_is_rejected() {
let json: Value = serde_json::json!({
"table": { "columnNames": ["a", "b"], "columnTypes": ["int"], "rows": [] }
});
assert!(
parse_erddap_json(&json).is_err(),
"names/types mismatch rejected"
);
}
}