src/usgs.rs
//! **USGS earthquakes** — the U.S. Geological Survey real-time GeoJSON feed
//! (`https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/*.geojson`). The
//! document is a **GeoJSON `FeatureCollection`**: a `features[]` array whose every
//! element carries a `Point` `geometry` (`coordinates = [lon, lat, depth_km]`) and
//! a `properties` object (`mag` / `place` / `time` / `type` / `url`). This is a
//! DISTINCT parser from the crate's other JSON collectors: PxWeb's JSON-stat2 cube
//! and Eurostat's SDMX-JSON series/observation split are statistical cubes, and the
//! WFS collector parses **GML** XML — none decode the GeoJSON geometry/property
//! split. [`parse_usgs_geojson`] flattens each feature to one earthquake row (id,
//! place, magnitude, depth, lon/lat, epoch-ms time, event type, url) → Arrow.
//!
//! The USGS feeds are **public, no-auth, CC0-style** open data, so the sample seam
//! (`sample::usgs_try_get_real_sample_data`) can attempt a live fetch with no
//! credential; offline (or `KORP_COLLECTORS_OFFLINE=1`) it falls back to an
//! embedded synthetic `FeatureCollection`.
//!
//! Offline contract: [`parse_usgs_geojson`] turns a canned `FeatureCollection`
//! into per-feature quake rows → Arrow; a feature with no locatable `Point`
//! geometry (missing/NaN lon or lat) contributes NO row (an all-unlocatable
//! 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 earthquake: a single GeoJSON `Point` feature from the USGS feed.
#[derive(Debug, Clone, PartialEq)]
pub struct Quake {
/// USGS event id (feature `id`, e.g. `"nc73012345"`) — the stable join key.
pub id: String,
/// Human place description (`properties.place`, e.g. `"10km N of Reno"`).
pub place: Option<String>,
/// Richter/moment magnitude (`properties.mag`) — nullable (some detections
/// carry no computed magnitude yet).
pub magnitude: Option<f64>,
/// Focal depth in km (`geometry.coordinates[2]`) — nullable (2-D geometries).
pub depth_km: Option<f64>,
/// Longitude (`geometry.coordinates[0]`) — the map X. A feature with no
/// numeric longitude is dropped (unlocatable).
pub longitude: f64,
/// Latitude (`geometry.coordinates[1]`) — the map Y.
pub latitude: f64,
/// Event time as **epoch milliseconds** (`properties.time`) — the CDC /
/// partition key. Nullable (rare feeds omit it).
pub time_ms: Option<i64>,
/// Event type (`properties.type`, e.g. `"earthquake"`, `"quarry blast"`).
pub event_type: Option<String>,
/// Canonical event URL (`properties.url`).
pub url: Option<String>,
}
/// Flatten a USGS GeoJSON `FeatureCollection` → per-feature earthquake rows.
///
/// Each `features[]` element must carry a `Point` `geometry` with numeric
/// `coordinates[0]` (lon) and `[1]` (lat); anything without a locatable point is
/// skipped. `coordinates[2]` (depth km) and every `properties.*` field are
/// nullable. A magnitude quoted as a string is coerced (some mirrors stringify).
pub fn parse_usgs_geojson(json: &Value) -> Result<Vec<Quake>> {
let mut out = Vec::new();
let Some(features) = json.get("features").and_then(|v| v.as_array()) else {
return Ok(out);
};
for f in features {
let geom = f.get("geometry");
// Only Point features carry a placeable epicentre; skip lines/polygons.
let is_point = geom
.and_then(|g| g.get("type"))
.and_then(|t| t.as_str())
.map(|t| t.eq_ignore_ascii_case("Point"))
.unwrap_or(false);
let coords = geom
.and_then(|g| g.get("coordinates"))
.and_then(|c| c.as_array());
let (Some(coords), true) = (coords, is_point) else {
continue;
};
let (Some(lon), Some(lat)) = (
coords.first().and_then(as_num),
coords.get(1).and_then(as_num),
) else {
continue; // unlocatable → no row
};
let depth_km = coords.get(2).and_then(as_num);
let props = f.get("properties");
let get = |k: &str| props.and_then(|p| p.get(k));
let id = f
.get("id")
.and_then(|v| v.as_str())
.map(str::to_string)
.unwrap_or_default();
out.push(Quake {
id,
place: get("place").and_then(|v| v.as_str()).map(str::to_string),
magnitude: get("mag").and_then(as_num),
depth_km,
longitude: lon,
latitude: lat,
time_ms: get("time").and_then(as_i64),
event_type: get("type").and_then(|v| v.as_str()).map(str::to_string),
url: get("url").and_then(|v| v.as_str()).map(str::to_string),
});
}
Ok(out)
}
/// A coordinate / magnitude 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()))
}
/// An epoch-ms timestamp 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()))
}
/// Earthquakes → one Arrow `RecordBatch`.
pub fn quakes_batch(quakes: &[Quake]) -> Result<RecordBatch> {
let cols = vec![
str_col(
&quakes
.iter()
.map(|q| Some(q.id.clone()))
.collect::<Vec<_>>(),
),
str_col(&quakes.iter().map(|q| q.place.clone()).collect::<Vec<_>>()),
f64_col(&quakes.iter().map(|q| q.magnitude).collect::<Vec<_>>()),
f64_col(&quakes.iter().map(|q| q.depth_km).collect::<Vec<_>>()),
f64_col(&quakes.iter().map(|q| Some(q.longitude)).collect::<Vec<_>>()),
f64_col(&quakes.iter().map(|q| Some(q.latitude)).collect::<Vec<_>>()),
i64_col(&quakes.iter().map(|q| q.time_ms).collect::<Vec<_>>()),
str_col(
&quakes
.iter()
.map(|q| q.event_type.clone())
.collect::<Vec<_>>(),
),
str_col(&quakes.iter().map(|q| q.url.clone()).collect::<Vec<_>>()),
];
batch(&usgs_schema(), cols)
}
pub fn usgs_schema() -> Schema {
schema(&[
("id", DataType::Utf8, false),
("place", DataType::Utf8, true),
("magnitude", DataType::Float64, true),
("depth_km", DataType::Float64, true),
("longitude", DataType::Float64, false),
("latitude", DataType::Float64, false),
("time", DataType::Int64, true),
("event_type", DataType::Utf8, true),
("url", DataType::Utf8, true),
])
}
/// The USGS real-time earthquake GeoJSON collector.
#[derive(Default)]
pub struct UsgsQuakeCollector {
/// What the last [`Collector::collect`] read — the `Last-Modified` validator the
/// USGS feed served, the bytes it cost, and whether it answered `304`.
last_read: std::sync::Mutex<Option<Checkpoint>>,
}
#[async_trait::async_trait]
impl Collector for UsgsQuakeCollector {
/// **The card for a keyless feed whose real subtlety is FRESHNESS, not access.**
/// Nothing here gates an operator — there is no key and no sign-up — so the quirks
/// carry what actually costs time: the feed is a mutable document rewritten in place,
/// it serves `Last-Modified` and no `ETag`, and a zero-row collect is routinely
/// correct.
///
/// URLs verified reachable on 2026-09-04; the 304 measurement is the one already
/// recorded on [`freshness`](Collector::freshness) (2026-09-03). USGS publishes no
/// numeric rate limit on either the feed page or the FDSN event service page —
/// searched 2026-09-04, hence
/// [`RateLimit::NonePublished`](crate::human_doc::RateLimit::NonePublished) rather than
/// a guess.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit, Registration};
Some(
HumanDoc::new(
"U.S. Geological Survey — the Earthquake Hazards Program, which publishes the \
real-time summary feeds",
"the USGS and the contributing seismic networks whose solutions it \
republishes — each feature's `net` and `sources` properties name the network \
a given solution came from",
"https://www.usgs.gov/information-policies-and-instructions/copyrights-and-credits",
"https://earthquake.usgs.gov/earthquakes/feed/v1.0/geojson.php",
)
.with_contact(Contact::Form("https://earthquake.usgs.gov/contactus/".into()))
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::NonePublished)
.with_quirk(
"The summary feeds are REWRITTEN IN PLACE at one URL and USGS states they are \
updated every minute — polling faster only re-fetches bytes that did not \
change. The collector's `Freshness::MutableDocument` + `If-Modified-Since` is \
the throttle: measured live 2026-09-03, a conditional second request came \
back HTTP/2 304 with no body against 3 853 bytes for the unconditional one.",
)
.with_quirk(
"No `ETag` is served — only `Last-Modified`. A conditional request built \
around `If-None-Match` will therefore never hit, which is exactly why \
`freshness()` declares only the one validator: asking for one the source \
does not offer is a claim, not a mechanism.",
)
.with_quirk(
"A zero-row collect is routinely CORRECT here — a 304 (nothing changed) and a \
genuinely quiet hour both land no rows and both explain themselves in \
`CollectStats.reason`. Never treat an empty USGS collect as RED without \
reading that reason first.",
)
.with_quirk(
"A feature with no numeric longitude/latitude is DROPPED as unlocatable, and \
`magnitude`, `depth_km` and `time_ms` are all nullable — a fresh detection \
can arrive before a magnitude has been computed, and a 2-D geometry carries \
no depth. Filter on nulls, do not assume them absent.",
)
.with_quirk(
"The USGS event `id` is REUSED as an event is revised, which is why it is the \
business key: the same key legitimately comes back later with a different \
magnitude or depth. That is a history update, not a duplicate.",
),
)
}
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "usgs-earthquakes".into(),
source_kind: SourceKind::Rest,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["geojson".into()],
license: "USGS earthquake feeds — public domain (U.S. Government work)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["id".into()],
description: "USGS real-time earthquake GeoJSON FeatureCollection — each Point \
feature flattened to a quake row (id, place, magnitude, depth, \
lon/lat, epoch-ms time) → map-scene position table."
.into(),
}
}
fn source_target(&self) -> SourceTarget {
SourceTarget::map_scene(
SourceEndpoint::Usgs,
"https://earthquake.usgs.gov/earthquakes/feed/v1.0/summary/all_hour.geojson",
"usgs_quakes",
"osm3d",
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(usgs_schema())
}
/// **A mutable document.** The summary feed is rewritten in place at the same URL
/// and serves `Last-Modified` — measured live on 2026-09-03, where a second request
/// carrying `If-Modified-Since` came back `HTTP/2 304` with no body against 3 853
/// bytes for the unconditional one. No `ETag` is served, so only that validator is
/// declared: asking for one the source does not offer is a claim, not a mechanism.
fn freshness(&self) -> crate::freshness::Freshness {
crate::freshness::Freshness::MutableDocument {
validators: vec![crate::freshness::Validator::LastModified],
}
}
/// The USGS event id — a stable identity the feed reuses as a quake is revised,
/// which is exactly the "same key, changed row" case history exists for.
fn business_key(&self) -> crate::history::BusinessKey {
crate::history::BusinessKey::of(&["id"], "the USGS event id, stable across revisions")
}
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 plan = bounds
.plan
.clone()
.unwrap_or_else(|| crate::freshness::Plan::of(&self.freshness(), None));
let (body, ck) =
super::http::fetch_planned(&http, &cfg.base_url, auth.as_ref(), &plan).await?;
*self.last_read.lock().expect("checkpoint mutex") = Some(ck);
// A 304 proved the feed did not change: nothing is re-landed, and the reason
// reads back as an explained empty, never as RED.
let Some(body) = body else {
return Ok(CollectStats {
table: cfg.table.clone(),
reason: Some(crate::freshness::unchanged_reason(
"usgs-earthquakes",
"304 Not Modified",
)),
..Default::default()
});
};
let json: Value = serde_json::from_str(&body)?;
let quakes = parse_usgs_geojson(&json)?;
if quakes.is_empty() {
return Ok(CollectStats {
reason: Some(
"USGS FeatureCollection carried no locatable Point features (quiet window?)"
.into(),
),
..Default::default()
});
}
let n = quakes.len() as u64;
sink.push(quakes_batch(&quakes)?)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
async fn checkpoint(&self) -> Result<Option<Checkpoint>> {
Ok(self.last_read.lock().expect("checkpoint mutex").clone())
}
}
#[cfg(test)]
mod tests {
use super::*;
/// A GeoJSON `FeatureCollection` flattens to one row per locatable Point
/// feature; magnitude/depth/time/place ride each row, and `coordinates` maps to
/// (lon, lat, depth). Offline DATA contract (no network) + a testmatrix cell.
#[test]
fn usgs_featurecollection_flattens_to_quake_rows() {
let json: Value = serde_json::json!({
"type": "FeatureCollection",
"metadata": { "count": 2 },
"features": [
{
"type": "Feature",
"id": "nc73012345",
"properties": {
"mag": 4.2, "place": "10km N of Reno, Nevada",
"time": 1_752_400_000_000i64, "type": "earthquake",
"url": "https://earthquake.usgs.gov/earthquakes/eventpage/nc73012345"
},
"geometry": { "type": "Point", "coordinates": [-119.81, 39.53, 8.4] }
},
{
"type": "Feature",
"id": "us7000abcd",
"properties": {
"mag": 5.9, "place": "South of Fiji Islands",
"time": 1_752_400_500_000i64, "type": "earthquake",
"url": "https://earthquake.usgs.gov/earthquakes/eventpage/us7000abcd"
},
"geometry": { "type": "Point", "coordinates": [178.12, -22.34, 560.0] }
}
]
});
let quakes = parse_usgs_geojson(&json).expect("parse usgs");
assert_eq!(quakes.len(), 2, "two locatable Point features → two rows");
let reno = quakes
.iter()
.find(|q| q.id == "nc73012345")
.expect("Reno event");
assert_eq!(reno.magnitude, Some(4.2));
assert!(
(reno.longitude - -119.81).abs() < 1e-9,
"lon = coordinates[0]"
);
assert!((reno.latitude - 39.53).abs() < 1e-9, "lat = coordinates[1]");
assert_eq!(reno.depth_km, Some(8.4), "depth = coordinates[2]");
assert_eq!(reno.time_ms, Some(1_752_400_000_000));
assert_eq!(reno.event_type.as_deref(), Some("earthquake"));
let rb = quakes_batch(&quakes).expect("quakes 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 [
"id",
"place",
"magnitude",
"depth_km",
"longitude",
"latitude",
"time",
"event_type",
"url",
] {
assert!(
cols.contains(&want),
"quake table carries `{want}`: {cols:?}"
);
}
// The `time` column is Int64 (epoch ms), `magnitude`/`longitude` Float64.
let mag = rb
.column_by_name("magnitude")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Float64Array>())
.expect("magnitude is f64");
let time = rb
.column_by_name("time")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Int64Array>())
.expect("time is i64");
assert!((mag.value(0) - 4.2).abs() < 1e-9);
assert_eq!(time.value(0), 1_752_400_000_000);
crate::functional_status(
"korp-collectors/usgs-earthquakes",
"geojson_flatten",
true,
&format!(
"{} quake rows from a GeoJSON FeatureCollection",
rb.num_rows()
),
);
}
/// Nullable properties (absent `mag`, absent depth, quoted numeric strings) are
/// tolerated, and a non-Point / unlocatable feature contributes NO row.
#[test]
fn usgs_nullable_props_and_string_coercion_and_unlocatable_dropped() {
let json: Value = serde_json::json!({
"type": "FeatureCollection",
"features": [
// valid, but null mag + only 2-D coords (no depth), time as string
{
"type": "Feature", "id": "ci111",
"properties": { "mag": null, "place": "offshore", "time": "1752400000000",
"type": "earthquake" },
"geometry": { "type": "Point", "coordinates": [-120.0, 34.0] }
},
// magnitude quoted as a string → coerced
{
"type": "Feature", "id": "ci222",
"properties": { "mag": "2.5", "place": "inland", "time": 1752400001000i64 },
"geometry": { "type": "Point", "coordinates": ["-118.5", "35.5", "3.2"] }
},
// a LineString (not a Point) → dropped
{
"type": "Feature", "id": "line1", "properties": { "mag": 9.9 },
"geometry": { "type": "LineString", "coordinates": [[0,0],[1,1]] }
},
// a Point with a non-numeric longitude → unlocatable, dropped
{
"type": "Feature", "id": "bad1", "properties": { "mag": 3.0 },
"geometry": { "type": "Point", "coordinates": ["not-a-number", 10.0] }
}
]
});
let quakes = parse_usgs_geojson(&json).expect("parse");
assert_eq!(
quakes.len(),
2,
"2 locatable Points; LineString + bad-coord dropped"
);
let a = quakes.iter().find(|q| q.id == "ci111").expect("ci111");
assert_eq!(a.magnitude, None, "null mag → None (nullable column)");
assert_eq!(a.depth_km, None, "2-D coords → no depth");
assert_eq!(a.time_ms, Some(1_752_400_000_000), "string epoch coerced");
let b = quakes.iter().find(|q| q.id == "ci222").expect("ci222");
assert_eq!(b.magnitude, Some(2.5), "string magnitude coerced");
assert!(
(b.longitude - -118.5).abs() < 1e-9,
"string longitude coerced"
);
assert_eq!(b.depth_km, Some(3.2), "string depth coerced");
// The batch still builds with the null magnitude/depth cells.
let rb = quakes_batch(&quakes).expect("batch");
assert_eq!(rb.num_rows(), 2);
let mag = rb
.column_by_name("magnitude")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Float64Array>())
.expect("magnitude f64");
assert!(
arrow_array::Array::is_null(&mag, 0),
"ci111's magnitude cell is null"
);
}
/// A document with no locatable features yields zero rows — the collector maps
/// that to a RAGNARÖK reason (the RED-when-broken empty guard).
#[test]
fn usgs_empty_or_unlocatable_document_yields_no_rows() {
let empty: Value = serde_json::json!({ "type": "FeatureCollection", "features": [] });
assert!(parse_usgs_geojson(&empty).expect("parse").is_empty());
// A document with only non-Point geometry contributes nothing.
let only_lines: Value = serde_json::json!({
"type": "FeatureCollection",
"features": [
{ "type": "Feature", "id": "l", "properties": {},
"geometry": { "type": "LineString", "coordinates": [[0,0],[1,1]] } }
]
});
assert!(parse_usgs_geojson(&only_lines).expect("parse").is_empty());
// A missing `features` key is an empty (not a panic).
assert!(
parse_usgs_geojson(&serde_json::json!({}))
.expect("parse")
.is_empty()
);
}
}