//! **Energy-Charts** — the Fraunhofer ISE *Energy-Charts* public-power API
//! (`https://api.energy-charts.info/public_power?country=de`). It reports the
//! live electricity **generation mix** of a European country (or the `eu`
//! aggregate) as a set of PARALLEL, index-aligned arrays: a single
//! `unix_seconds[]` time axis plus a `production_types[]` list where every entry
//! is `{ name, data[] }` and `data[i]` is that source's power (MW) at
//! `unix_seconds[i]`.
//!
//! This is a DISTINCT parser shape from every other collector in the crate:
//! Nord Pool iterates an array of period OBJECTS each carrying an `entryPerArea`
//! map; ENTSO-E walks nested XML `TimeSeries/Period/Point`; PxWeb/SDMX are
//! statistical CUBES with an explicit index; USGS/OpenAQ are per-feature object
//! rows. None of them decode a **columnar / transposed** document where the
//! measurement value must be recovered by ZIPPING two independent parallel
//! arrays by position. [`parse_energy_charts_json`] transposes the
//! `time × production_type` matrix into one flat `(time, production_type, power)`
//! row per cell — with an index-alignment guard (a `data[]` shorter than the
//! time axis contributes only its own length) and null/non-numeric cells
//! skipped.
//!
//! Energy-Charts data is **public, no-auth, CC-BY-4.0** (Fraunhofer ISE /
//! Bundesnetzagentur SMARD / ENTSO-E derived), so — like USGS — the sample seam
//! can attempt a live fetch with no credential; offline (or
//! `KORP_COLLECTORS_OFFLINE=1`) it falls back to an embedded synthetic document.
//!
//! Offline contract: [`parse_energy_charts_json`] turns a canned
//! `{unix_seconds, production_types}` document into per-`(time, type)` power rows
//! → Arrow; a document whose arrays zip to no numeric cell contributes NO row
//! (an all-empty 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 generation-mix reading: a single production type's power at a
/// single instant of the shared time axis.
#[derive(Debug, Clone, PartialEq)]
pub struct PowerReading {
/// Instant as **epoch seconds** (`unix_seconds[i]`) — the CDC / partition key.
pub time_s: i64,
/// Production type / source name (`production_types[].name`, e.g. `"Solar"`,
/// `"Wind onshore"`, `"Nuclear"`, `"Load"`).
pub production_type: String,
/// Power for this `(time, type)` cell in **MW** (`production_types[].data[i]`).
pub power_mw: f64,
}
/// Transpose an Energy-Charts `{unix_seconds, production_types}` document → one
/// `(time, production_type, power)` row per numeric cell.
///
/// The value of each cell is recovered by ZIPPING the shared `unix_seconds[]`
/// time axis against each production type's parallel `data[]` array by index.
/// Robustness guards: a `data[]` array shorter (or longer) than the time axis is
/// index-aligned to the shorter of the two (no panic, no phantom cells); a cell
/// whose time or power is null / non-numeric is skipped (some feeds null-pad the
/// most recent, not-yet-reported quarter-hour). A power quoted as a string is
/// coerced (a few mirrors stringify).
pub fn parse_energy_charts_json(json: &Value) -> Result<Vec<PowerReading>> {
let mut out = Vec::new();
let Some(times) = json.get("unix_seconds").and_then(|v| v.as_array()) else {
return Ok(out);
};
let Some(types) = json.get("production_types").and_then(|v| v.as_array()) else {
return Ok(out);
};
// Decode the time axis once (parallel arrays share it).
let times: Vec<Option<i64>> = times.iter().map(as_i64).collect();
for pt in types {
let Some(name) = pt.get("name").and_then(|v| v.as_str()) else {
continue; // an unnamed series carries no join key → skip
};
let Some(data) = pt.get("data").and_then(|v| v.as_array()) else {
continue;
};
// Index-alignment guard: zip only the overlap of the two parallel arrays.
let n = times.len().min(data.len());
for i in 0..n {
let (Some(t), Some(power)) = (times[i], as_num(&data[i])) else {
continue; // null-padded / non-numeric cell → no row
};
out.push(PowerReading {
time_s: t,
production_type: name.to_string(),
power_mw: power,
});
}
}
Ok(out)
}
/// A power value 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-seconds timestamp may be a JSON integer, a float, 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()))
}
/// Generation-mix readings → one Arrow `RecordBatch`.
pub fn power_batch(readings: &[PowerReading]) -> Result<RecordBatch> {
let cols = vec![
i64_col(&readings.iter().map(|r| Some(r.time_s)).collect::<Vec<_>>()),
str_col(
&readings
.iter()
.map(|r| Some(r.production_type.clone()))
.collect::<Vec<_>>(),
),
f64_col(
&readings
.iter()
.map(|r| Some(r.power_mw))
.collect::<Vec<_>>(),
),
];
batch(&energy_charts_schema(), cols)
}
pub fn energy_charts_schema() -> Schema {
schema(&[
("time", DataType::Int64, false),
("production_type", DataType::Utf8, false),
("power_mw", DataType::Float64, false),
])
}
/// The Energy-Charts public-power (generation-mix) collector.
#[derive(Default)]
pub struct EnergyChartsCollector;
#[async_trait::async_trait]
impl Collector for EnergyChartsCollector {
/// **The card for the tightest budget in the crate.** Energy-Charts is keyless and
/// CC-BY-4.0, so the thing an operator actually needs from this card is the rate
/// limit: **2 requests per minute**, per IP *and* per endpoint. Everything below was
/// read off the API's own OpenAPI document (`https://api.energy-charts.info/openapi.json`,
/// `info.version` 2.0) on 2026-09-04 — including the contact, licence and terms URLs,
/// which the document itself publishes as `info.contact.url`, `info.license.url` and
/// `info.termsOfService`.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
use crate::human_doc::{Contact, HumanDoc, RateLimit, Registration};
Some(
HumanDoc::new(
"Fraunhofer Institute for Solar Energy Systems ISE — the Energy-Charts team, \
which builds and serves the API",
"the TSOs and regulators whose publications Energy-Charts aggregates \
(Bundesnetzagentur SMARD, ENTSO-E and national sources); Fraunhofer ISE \
republishes them, and the CC-BY-4.0 attribution is owed to Energy-Charts.info",
"https://energy-charts.info/publishing-notes.html",
"https://api.energy-charts.info/",
)
.with_contact(Contact::Form("https://energy-charts.info/contact.html".into()))
.with_registration(Registration::NotNeeded)
.with_rate_limit(RateLimit::Documented(
"2 requests/minute per client IP AND per endpoint (token bucket, burst 4); \
/signal 20/min (burst 40), /price 2/min (burst 2); every /v2 endpoint \
inherits its v1 counterpart's limit, and the limits are lowered DYNAMICALLY \
under server load"
.into(),
))
.with_quirk(
"2 requests/minute is the tightest budget of any collector here, and it is \
charged per IP AND per endpoint. Over it the API returns HTTP 429 with a \
`Retry-After` header; the published guidance is to honour that header rather \
than assume a fixed number, because the effective limit drops at peak times. \
Cache on your side — this is not a feed to poll.",
)
.with_quirk(
"Every timestamp the API returns marks the BEGINNING of the period its value \
refers to. (The single exception is `/installed_power`, where the stamp \
labels the reporting year or month and the value is the installed power at \
its END.) The collector lands `unix_seconds[i]` unchanged, so anything \
downstream must read the column as a period start, not a midpoint.",
)
.with_quirk(
"Responses carry a `deprecated` flag, and Energy-Charts announces an \
endpoint's removal at least six months ahead by flipping it to `true`. The \
parser ignores that field — if this feed matters, alert on it yourself, \
because the six months of warning are the only warning there is.",
)
.with_quirk(
"The mix arrives as PARALLEL arrays, and a feed may null-pad the most recent, \
not-yet-reported quarter-hour. The parser index-aligns each `data[]` to the \
shorter of it and `unix_seconds[]` and skips null cells, so a legitimate \
window can land fewer rows than `types × instants` — a short count is not \
evidence of a broken parse.",
)
.with_quirk(
"Higher (commercial) limits are arranged by email with the Energy-Charts team \
at Fraunhofer ISE, not self-service. The free tier needs no key at all — do \
not go looking for one.",
),
)
}
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "energy-charts-power".into(),
source_kind: SourceKind::Rest,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["json".into()],
license: "Energy-Charts (Fraunhofer ISE) — CC-BY-4.0 open data".into(),
gdpr: GdprClass::Open,
auth: AuthKind::None,
join_keys: vec!["time".into(), "production_type".into()],
description: "Energy-Charts public-power generation mix — the parallel \
unix_seconds[] time axis zipped against each production_types[].data[] \
into per-(time, type) power rows (Europe-wide via ?country=)."
.into(),
}
}
fn source_target(&self) -> SourceTarget {
SourceTarget::warehouse(
SourceEndpoint::EnergyCharts,
"https://api.energy-charts.info/public_power?country=de",
"energy_charts_power",
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(energy_charts_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 readings = parse_energy_charts_json(&json)?;
if readings.is_empty() {
return Ok(CollectStats {
reason: Some(
"Energy-Charts document carried no numeric (time, type) power cells (feed gap?)"
.into(),
),
..Default::default()
});
}
let n = readings.len() as u64;
sink.push(power_batch(&readings)?)?;
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::*;
/// A `{unix_seconds, production_types}` document TRANSPOSES to one row per
/// numeric `(time, type)` cell: the two parallel arrays are zipped by index.
/// Offline DATA contract (no network) + a testmatrix cell.
#[test]
fn energy_charts_parallel_arrays_transpose_to_time_type_rows() {
let json: Value = serde_json::json!({
"unix_seconds": [1_751_328_000i64, 1_751_328_900i64],
"production_types": [
{ "name": "Solar", "data": [1200.5, 1810.0] },
{ "name": "Wind onshore", "data": [ 640.0, 655.5] },
{ "name": "Nuclear", "data": [4000.0, 4000.0] }
]
});
let readings = parse_energy_charts_json(&json).expect("parse energy-charts");
assert_eq!(
readings.len(),
6,
"3 types × 2 instants = 6 transposed rows"
);
// A specific (type, time) → power lookup proves the index zip is correct.
let solar_t1 = readings
.iter()
.find(|r| r.production_type == "Solar" && r.time_s == 1_751_328_900)
.expect("Solar at t1 present");
assert!(
(solar_t1.power_mw - 1810.0).abs() < 1e-9,
"Solar t1 power = data[1]"
);
let wind_t0 = readings
.iter()
.find(|r| r.production_type == "Wind onshore" && r.time_s == 1_751_328_000)
.expect("Wind at t0 present");
assert!(
(wind_t0.power_mw - 640.0).abs() < 1e-9,
"Wind t0 power = data[0]"
);
let rb = power_batch(&readings).expect("power batch");
assert_eq!(rb.num_rows(), 6);
let schema = rb.schema();
let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
for want in ["time", "production_type", "power_mw"] {
assert!(
cols.contains(&want),
"power table carries `{want}`: {cols:?}"
);
}
// `time` is Int64 (epoch seconds), `power_mw` Float64.
let time = rb
.column_by_name("time")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Int64Array>())
.expect("time is i64");
let power = rb
.column_by_name("power_mw")
.and_then(|a| a.as_any().downcast_ref::<arrow_array::Float64Array>())
.expect("power is f64");
// Row order is type-major then time: first row is Solar @ t0.
assert_eq!(time.value(0), 1_751_328_000);
assert!((power.value(0) - 1200.5).abs() < 1e-9);
crate::functional_status(
"korp-collectors/energy-charts-power",
"parallel_array_transpose",
true,
&format!(
"{} (time, type) power rows from zipped parallel arrays",
rb.num_rows()
),
);
}
/// The index-alignment guard: a `data[]` shorter than the time axis contributes
/// only its own length (no panic, no phantom cells), null/non-numeric cells are
/// skipped, and a string-quoted power is coerced.
#[test]
fn energy_charts_ragged_arrays_nulls_and_string_coercion() {
let json: Value = serde_json::json!({
"unix_seconds": [100i64, 200i64, 300i64],
"production_types": [
// 3 instants, but a null-padded most-recent cell → 2 rows
{ "name": "Solar", "data": [50.0, 60.0, null] },
// SHORT data (only 2 of 3 instants) → index-aligned to 2 rows
{ "name": "Wind onshore", "data": [10.0, 12.0] },
// one string-quoted power (coerced), one non-numeric (skipped)
{ "name": "Hydro", "data": ["33.5", "n/a", 40.0] }
]
});
let readings = parse_energy_charts_json(&json).expect("parse ragged");
// Solar 2 + Wind 2 + Hydro 2 = 6
assert_eq!(
readings.len(),
6,
"ragged/null/non-numeric cells guarded: {readings:#?}"
);
// Wind's short array never read a phantom third instant.
assert!(
!readings
.iter()
.any(|r| r.production_type == "Wind onshore" && r.time_s == 300),
"short data[] must not fabricate a t=300 wind cell"
);
// Solar's null-padded cell at t=300 is absent.
assert!(
!readings
.iter()
.any(|r| r.production_type == "Solar" && r.time_s == 300)
);
// Hydro's string "33.5" coerced; "n/a" dropped; 40.0 kept.
let hydro: Vec<&PowerReading> = readings
.iter()
.filter(|r| r.production_type == "Hydro")
.collect();
assert_eq!(
hydro.len(),
2,
"Hydro: string coerced + numeric kept, n/a dropped"
);
assert!(
hydro.iter().any(|r| (r.power_mw - 33.5).abs() < 1e-9),
"string power coerced"
);
assert!(
hydro.iter().all(|r| r.time_s != 200),
"the n/a cell (t=200) contributes no row"
);
let rb = power_batch(&readings).expect("batch");
assert_eq!(rb.num_rows(), 6);
crate::functional_status(
"korp-collectors/energy-charts-power",
"ragged_null_coercion_guard",
true,
"short data[] index-aligned, null/non-numeric cells skipped, string power coerced",
);
}
/// A document with no zippable numeric cells yields zero rows — the collector
/// maps that to a RAGNARÖK reason (the RED-when-broken empty guard).
#[test]
fn energy_charts_empty_or_axis_only_document_yields_no_rows() {
// No production_types at all.
let axis_only: Value = serde_json::json!({ "unix_seconds": [1, 2, 3] });
assert!(
parse_energy_charts_json(&axis_only)
.expect("parse")
.is_empty()
);
// Types present but every data cell is null → nothing zips.
let all_null: Value = serde_json::json!({
"unix_seconds": [1, 2],
"production_types": [ { "name": "Solar", "data": [null, null] } ]
});
assert!(
parse_energy_charts_json(&all_null)
.expect("parse")
.is_empty()
);
// A missing `unix_seconds` axis is an empty (not a panic).
let no_axis: Value = serde_json::json!({
"production_types": [ { "name": "Solar", "data": [1.0] } ]
});
assert!(
parse_energy_charts_json(&no_axis)
.expect("parse")
.is_empty()
);
// A fully empty object.
assert!(
parse_energy_charts_json(&serde_json::json!({}))
.expect("parse")
.is_empty()
);
}
}