cobre_io/extensions/fpha_deviation_points.rs
1//! Row type and reader for per-sampled-point computed-FPHA fit deviations.
2//!
3//! [`FphaDeviationPointRow`] describes one `(V, Q)` grid point at spillage = 0
4//! for a `(hydro, stage)` fit: the exact production-function value `fph_exact`,
5//! the fitted min-envelope value `fpha_fitted`, their signed difference
6//! `deviation`, and `relative` (the absolute deviation against the grid peak).
7//! [`parse_fpha_deviation_points`] reads a Parquet file written by
8//! `crate::output::write_fpha_deviation_points` back into a sorted
9//! `Vec<FphaDeviationPointRow>`.
10//!
11//! ## Parquet schema
12//!
13//! | Column | Type | Required | Description |
14//! | ------------- | ------- | -------- | ---------------------------------------------- |
15//! | `hydro_id` | INT32 | Yes | Hydro plant identifier |
16//! | `stage_id` | INT32? | No | Stage (`null` = single fit covering all stages)|
17//! | `v` | DOUBLE | Yes | Volume grid coordinate (hm³) |
18//! | `q` | DOUBLE | Yes | Turbined-flow grid coordinate (m³/s) |
19//! | `fph_exact` | DOUBLE | Yes | Exact production-function value (MW) |
20//! | `fpha_fitted` | DOUBLE | Yes | Fitted min-envelope value (MW) |
21//! | `deviation` | DOUBLE | Yes | Signed residual `fpha_fitted − fph_exact` (MW) |
22//! | `relative` | DOUBLE | Yes | `|deviation|` relative to the grid peak |
23//!
24//! ## Output ordering
25//!
26//! Rows are sorted by `(hydro_id, stage_id)` ascending. Null `stage_id` sorts
27//! before any non-null value. Within a `(hydro_id, stage_id)` block the grid
28//! order is preserved as written (the canonical `(V, Q)` walk).
29
30use arrow::array::Array;
31use cobre_core::EntityId;
32use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
33use std::fs::File;
34use std::path::Path;
35
36use crate::LoadError;
37use crate::parquet_helpers::{
38 extract_optional_int32, extract_required_float64, extract_required_int32,
39};
40
41/// A single per-sampled-point computed-FPHA fit-deviation row at spillage = 0.
42///
43/// # Examples
44///
45/// ```
46/// use cobre_io::extensions::FphaDeviationPointRow;
47/// use cobre_core::EntityId;
48///
49/// let row = FphaDeviationPointRow {
50/// hydro_id: EntityId::from(66),
51/// stage_id: Some(3),
52/// v: 14_500.0,
53/// q: 1_200.0,
54/// fph_exact: 980.5,
55/// fpha_fitted: 985.0,
56/// deviation: 4.5,
57/// relative: 0.0046,
58/// };
59/// assert_eq!(row.hydro_id, EntityId::from(66));
60/// ```
61#[derive(Debug, Clone, PartialEq)]
62pub struct FphaDeviationPointRow {
63 /// Hydro plant this deviation point belongs to.
64 pub hydro_id: EntityId,
65 /// Stage the covering fit applies to. `None` means a single fit for all stages.
66 pub stage_id: Option<i32>,
67 /// Volume grid coordinate (hm³).
68 pub v: f64,
69 /// Turbined-flow grid coordinate (m³/s).
70 pub q: f64,
71 /// Exact production-function value at `(v, q, 0.0)` (MW).
72 pub fph_exact: f64,
73 /// Fitted min-envelope value at `(v, q, 0.0)` (MW).
74 pub fpha_fitted: f64,
75 /// Signed residual `fpha_fitted − fph_exact` (MW).
76 pub deviation: f64,
77 /// `|deviation|` relative to the grid's peak exact generation (dimensionless).
78 pub relative: f64,
79}
80
81/// Parse an FPHA-deviation-points Parquet file into a table sorted by `(hydro_id,
82/// stage_id)` ascending (NULL `stage_id` first; within-block grid order preserved).
83///
84/// # Errors
85///
86/// | Condition | Error variant |
87/// |------------------------------------------|----------------------------|
88/// | File not found or permission denied | [`LoadError::IoError`] |
89/// | Malformed Parquet (corrupt header, etc.) | [`LoadError::ParseError`] |
90/// | Required column missing or wrong type | [`LoadError::SchemaError`] |
91pub fn parse_fpha_deviation_points(path: &Path) -> Result<Vec<FphaDeviationPointRow>, LoadError> {
92 let file = File::open(path).map_err(|e| LoadError::io(path, e))?;
93
94 let builder = ParquetRecordBatchReaderBuilder::try_new(file)
95 .map_err(|e| LoadError::parse(path, e.to_string()))?;
96
97 let reader = builder
98 .build()
99 .map_err(|e| LoadError::parse(path, e.to_string()))?;
100
101 let mut rows: Vec<FphaDeviationPointRow> = Vec::new();
102
103 for batch_result in reader {
104 let batch = batch_result.map_err(|e| LoadError::parse(path, e.to_string()))?;
105
106 let hydro_id_col = extract_required_int32(&batch, "hydro_id", path)?;
107 let v_col = extract_required_float64(&batch, "v", path)?;
108 let q_col = extract_required_float64(&batch, "q", path)?;
109 let fph_exact_col = extract_required_float64(&batch, "fph_exact", path)?;
110 let fpha_fitted_col = extract_required_float64(&batch, "fpha_fitted", path)?;
111 let deviation_col = extract_required_float64(&batch, "deviation", path)?;
112 let relative_col = extract_required_float64(&batch, "relative", path)?;
113
114 let stage_id_col = extract_optional_int32(&batch, "stage_id", path)?;
115
116 let n = batch.num_rows();
117 rows.reserve(n);
118
119 for i in 0..n {
120 let hydro_id = EntityId::from(hydro_id_col.value(i));
121 let v = v_col.value(i);
122 let q = q_col.value(i);
123 let fph_exact = fph_exact_col.value(i);
124 let fpha_fitted = fpha_fitted_col.value(i);
125 let deviation = deviation_col.value(i);
126 let relative = relative_col.value(i);
127
128 let stage_id = stage_id_col
129 .filter(|col| !col.is_null(i))
130 .map(|col| col.value(i));
131
132 rows.push(FphaDeviationPointRow {
133 hydro_id,
134 stage_id,
135 v,
136 q,
137 fph_exact,
138 fpha_fitted,
139 deviation,
140 relative,
141 });
142 }
143 }
144
145 // Stable sort: preserves the within-block canonical grid order.
146 rows.sort_by(|a, b| {
147 a.hydro_id
148 .0
149 .cmp(&b.hydro_id.0)
150 .then_with(|| a.stage_id.cmp(&b.stage_id))
151 });
152
153 Ok(rows)
154}