//! **#4b ENTSO-E generation-per-type** — `web-api.tp.entsoe.eu/api?documentType=A75`.
//! Complements the existing [`entsoe`](super::entsoe) load/price collector: the
//! `GL_MarketDocument` returned for `documentType=A75` (Actual generation per
//! type) carries a `<MktPSRType><psrType>` production-type code per `TimeSeries`
//! (B16 = Solar, B19 = Wind Onshore, B14 = Nuclear, …) that the plain-load
//! parser drops. Reuses the shared zone mapping (`zone_of`/`eic_of`), the
//! `HttpCore` client, and `arrow_util` — same framework, one extra column.
//!
//! Offline contract: [`parse_entsoe_generation_xml`] turns a canned
//! `GL_MarketDocument` into per-`Point` rows (domain, bidding zone, psr_type,
//! psr_label, resolution, period start, position, quantity) → Arrow.
use anyhow::Result;
use arrow_array::RecordBatch;
use arrow_schema::{DataType, Schema};
use quick_xml::events::Event;
use super::arrow_util::{batch, f64_col, i64_col, schema, str_col};
use super::entsoe::{eic_of, zone_of};
use super::{
AuthKind, BatchSink, Bounds, CollectStats, Collector, CollectorCfg, CollectorMeta,
CollectorPath, Direction, GdprClass, SourceEndpoint, SourceKind, SourceTarget,
};
/// ENTSO-E `psrType` production-type code → human label. The subset returned by
/// `documentType=A75`.
pub fn psr_label(code: &str) -> Option<&'static str> {
Some(match code {
"B01" => "Biomass",
"B02" => "Fossil Brown coal/Lignite",
"B03" => "Fossil Coal-derived gas",
"B04" => "Fossil Gas",
"B05" => "Fossil Hard coal",
"B06" => "Fossil Oil",
"B07" => "Fossil Oil shale",
"B08" => "Fossil Peat",
"B09" => "Geothermal",
"B10" => "Hydro Pumped Storage",
"B11" => "Hydro Run-of-river and poundage",
"B12" => "Hydro Water Reservoir",
"B13" => "Marine",
"B14" => "Nuclear",
"B15" => "Other renewable",
"B16" => "Solar",
"B17" => "Waste",
"B18" => "Wind Offshore",
"B19" => "Wind Onshore",
"B20" => "Other",
"B25" => "Energy storage",
_ => return None,
})
}
/// One generation-per-type time-series `Point`.
#[derive(Debug, Clone, PartialEq)]
pub struct EntsoeGenPoint {
pub in_domain: Option<String>,
pub bidding_zone: Option<String>,
pub psr_type: Option<String>,
pub psr_label: Option<String>,
pub resolution: Option<String>,
pub period_start: Option<String>,
pub position: i64,
pub quantity: f64,
}
fn local_name(name: quick_xml::name::QName) -> String {
String::from_utf8_lossy(name.local_name().as_ref()).into_owned()
}
/// Parse an ENTSO-E `GL_MarketDocument` (generation per type) into time-series
/// points. Like [`super::entsoe::parse_entsoe_xml`] but also captures the
/// `<MktPSRType><psrType>` code that precedes each `TimeSeries`' `<Period>`.
pub fn parse_entsoe_generation_xml(xml: &str) -> Result<Vec<EntsoeGenPoint>> {
let mut reader = quick_xml::Reader::from_str(xml);
let mut points = Vec::new();
let mut cur = String::new();
let mut domain: Option<String> = None;
let mut psr_type: Option<String> = None;
let mut resolution: Option<String> = None;
let mut period_start: Option<String> = None;
let mut position: Option<i64> = None;
let mut quantity: Option<f64> = None;
loop {
match reader.read_event()? {
Event::Start(e) => {
cur = local_name(e.name());
match cur.as_str() {
// A new series: reset the per-series domain + production type.
"TimeSeries" => {
domain = None;
psr_type = None;
}
"Point" => {
position = None;
quantity = None;
}
_ => {}
}
}
Event::Text(t) => {
let txt = t.unescape()?.trim().to_string();
if txt.is_empty() {
continue;
}
match cur.as_str() {
"in_Domain.mRID" => domain = Some(txt),
// <MktPSRType><psrType>B16</psrType></MktPSRType>
"psrType" => psr_type = Some(txt),
"resolution" => resolution = Some(txt),
// The Period's <timeInterval><start> sits closest to the
// Points; overwriting on every <start> leaves that one in effect.
"start" => period_start = Some(txt),
"position" => position = txt.parse::<i64>().ok(),
"quantity" => quantity = txt.parse::<f64>().ok(),
_ => {}
}
}
Event::End(e) => {
let name = local_name(e.name());
if name == "Point" {
if let (Some(pos), Some(q)) = (position, quantity) {
let bidding_zone = domain.as_deref().and_then(zone_of).map(str::to_string);
let label = psr_type.as_deref().and_then(psr_label).map(str::to_string);
points.push(EntsoeGenPoint {
in_domain: domain.clone(),
bidding_zone,
psr_type: psr_type.clone(),
psr_label: label,
resolution: resolution.clone(),
period_start: period_start.clone(),
position: pos,
quantity: q,
});
}
}
// Leaf element closed — stop capturing text into the parent.
cur.clear();
}
Event::Eof => break,
_ => {}
}
}
Ok(points)
}
/// ENTSO-E generation points → one Arrow `RecordBatch`.
pub fn gen_points_batch(points: &[EntsoeGenPoint]) -> Result<RecordBatch> {
let cols = vec![
str_col(
&points
.iter()
.map(|p| p.in_domain.clone())
.collect::<Vec<_>>(),
),
str_col(
&points
.iter()
.map(|p| p.bidding_zone.clone())
.collect::<Vec<_>>(),
),
str_col(
&points
.iter()
.map(|p| p.psr_type.clone())
.collect::<Vec<_>>(),
),
str_col(
&points
.iter()
.map(|p| p.psr_label.clone())
.collect::<Vec<_>>(),
),
str_col(
&points
.iter()
.map(|p| p.resolution.clone())
.collect::<Vec<_>>(),
),
str_col(
&points
.iter()
.map(|p| p.period_start.clone())
.collect::<Vec<_>>(),
),
i64_col(&points.iter().map(|p| Some(p.position)).collect::<Vec<_>>()),
f64_col(&points.iter().map(|p| Some(p.quantity)).collect::<Vec<_>>()),
];
batch(&entsoe_generation_schema(), cols)
}
pub fn entsoe_generation_schema() -> Schema {
schema(&[
("in_domain", DataType::Utf8, true),
("bidding_zone", DataType::Utf8, true),
("psr_type", DataType::Utf8, true),
("psr_label", DataType::Utf8, true),
("resolution", DataType::Utf8, true),
("period_start", DataType::Utf8, true),
("position", DataType::Int64, false),
("quantity", DataType::Float64, false),
])
}
/// The ENTSO-E generation-per-type collector.
#[derive(Default)]
pub struct EntsoeGenerationCollector;
#[async_trait::async_trait]
impl Collector for EntsoeGenerationCollector {
/// **ENTSO-E needs a free security token, and it has a NAME.** Without
/// `ENTSOE_TOKEN` the Transparency Platform answers 401 to a request that was
/// otherwise well formed — a failure the UI can and must predict, so this
/// collector reads `NeedsConfig` rather than offering a ▶ Run that cannot work.
fn requirements(&self) -> crate::requires::Requirements {
crate::requires::Requirements::open(self.source_target().source_endpoint).with_secrets(
vec![crate::requires::SecretReq::new(
"ENTSOE_TOKEN",
"the free ENTSO-E Transparency Platform security token (register at \
transparency.entsoe.eu, then Account Settings ▸ Web Api Security Token)",
)],
)
}
/// **The operator-facing card** — the shared ENTSO-E half (authority, the token
/// email, 400/min per token, the Open Data list) from
/// [`entsoe_human_doc`](super::entsoe::entsoe_human_doc), plus what bites in an A75
/// GENERATION document specifically. The A75 quirks are read off the parser above.
fn human_doc(&self) -> Option<crate::human_doc::HumanDoc> {
Some(
super::entsoe::entsoe_human_doc()
.with_quirk(
"`documentType=A75` (actual generation per type) is what makes this \
collector different from `entsoe`: the production type lives in \
`<MktPSRType><psrType>`, which the plain-load parser drops. Point a load \
document at this collector and every row still lands, but with `psr_type` \
and `psr_label` NULL — the generation mix is then unrecoverable.",
)
.with_quirk(
"`psr_label` covers the B01–B20 + B25 subset A75 returns. An unmapped code \
lands with `psr_type` set and `psr_label` NULL, so JOIN AND GROUP ON \
`psr_type` — the label is for humans and will be empty for anything \
ENTSO-E adds after this table was written.",
)
.with_quirk(
"`psrType` is reset per `<TimeSeries>`, and one document normally carries \
one series per production type. A zone's mix is therefore several series \
wide, not one — expect (zones × production types × positions) rows from a \
single request, and size the window accordingly against the 400/min cap.",
),
)
}
fn meta(&self) -> CollectorMeta {
CollectorMeta {
name: "entsoe_generation".into(),
source_kind: SourceKind::Rest,
path: CollectorPath::RustNative,
direction: Direction::Source,
formats: vec!["xml".into()],
license: "CC-BY 4.0 (free token, redistributable)".into(),
gdpr: GdprClass::Open,
auth: AuthKind::Token,
join_keys: vec!["elområde".into(), "psr_type".into(), "period_start".into()],
description: "ENTSO-E Transparency Platform — actual generation per production type \
(documentType A75), SE1–SE4 via EIC domain codes, psrType labelled."
.into(),
}
}
fn source_target(&self) -> SourceTarget {
SourceTarget::warehouse(
SourceEndpoint::EntsoE,
"https://web-api.tp.entsoe.eu/api",
"entsoe_generation",
)
}
/// **One point per (domain, production type, period, position).** See
/// `entsoe`'s key for why the derived one was wrong in two ways at once.
fn business_key(&self) -> crate::history::BusinessKey {
crate::history::BusinessKey::of(
&["in_domain", "psr_type", "period_start", "position"],
"one generation point per (EIC domain, production type, period, position)",
)
}
fn schema(&self, _cfg: &CollectorCfg) -> Result<Schema> {
Ok(entsoe_generation_schema())
}
async fn collect(
&self,
cfg: &CollectorCfg,
_bounds: Bounds,
sink: &BatchSink,
) -> Result<CollectStats> {
let http = cfg.http(6); // ENTSO-E ≤400 req/min
let auth = cfg.auth.resolve_async().await?; // securityToken via QueryKey auth
// Build the request URL from params (documentType=A75/processType/in_Domain/
// periodStart/periodEnd). A bidding-zone shorthand is mapped to its EIC.
let mut query: Vec<(String, String)> = Vec::new();
if let Some(obj) = cfg.params.as_object() {
for (k, v) in obj {
if let Some(s) = v.as_str() {
let val = if k == "in_Domain" || k == "out_Domain" {
eic_of(s)
.map(str::to_string)
.unwrap_or_else(|| s.to_string())
} else {
s.to_string()
};
query.push((k.clone(), val));
}
}
}
let qs = query
.iter()
.map(|(k, v)| format!("{}={}", k, super::http::percent_encode(v)))
.collect::<Vec<_>>()
.join("&");
let url = if qs.is_empty() {
cfg.base_url.clone()
} else {
format!("{}?{}", cfg.base_url, qs)
};
let xml = http.get_text(&url, auth.as_ref()).await?;
let points = parse_entsoe_generation_xml(&xml)?;
if points.is_empty() {
return Ok(CollectStats {
reason: Some(
"ENTSO-E returned no generation TimeSeries points (window empty?)".into(),
),
..Default::default()
});
}
let n = points.len() as u64;
sink.push(gen_points_batch(&points)?)?;
Ok(CollectStats {
rows: n,
batches: 1,
table: cfg.table.clone(),
reason: None,
})
}
}