river_data_core/models/
replicates.rs

1use chrono::{DateTime, Utc};
2use serde::{Deserialize, Serialize};
3use uuid::Uuid;
4
5/// Replicate-family declaration on a stream registration. The API pins each
6/// column's replicate_index server-side (append-only across re-registrations)
7/// and returns the authoritative mapping as [`ColumnAssignment`]s on the
8/// register response; `source_columns` order is provenance, not the index.
9/// The API requires at least two unique columns and `measurement_type: "spot"`
10/// on the request.
11#[derive(Debug, Clone, Serialize, Deserialize)]
12#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
13pub struct ReplicateSpec {
14    pub source_columns: Vec<String>,
15    #[serde(default, skip_serializing_if = "Option::is_none")]
16    pub portal_mean_column: Option<String>,
17    #[serde(default, skip_serializing_if = "Option::is_none")]
18    pub portal_sd_column: Option<String>,
19    #[serde(default, skip_serializing_if = "Option::is_none")]
20    pub curve_ref_column: Option<String>,
21    #[serde(default, skip_serializing_if = "Option::is_none")]
22    pub calc: Option<String>,
23}
24
25/// One source column's pinned replicate index, as the API's register response
26/// reports it (and as stream metadata persists it under
27/// `replicates.assignments`). Sync services assign each value's
28/// `replicate_index` by looking its source column up here.
29#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
30#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
31pub struct ColumnAssignment {
32    pub column: String,
33    pub index: i16,
34    /// The source no longer sends this column. The index stays reserved and
35    /// remains the column's identity should it reappear.
36    #[serde(default)]
37    pub retired: bool,
38}
39
40impl ColumnAssignment {
41    /// The mapping a stream's metadata carries, resolved the way the API
42    /// resolves it. None when the stream declares no replicate family.
43    pub fn from_metadata(metadata: &serde_json::Value) -> Option<Vec<Self>> {
44        let spec = metadata.get("replicates")?;
45        let pinned: Vec<Self> = spec
46            .get("assignments")
47            .and_then(|v| serde_json::from_value(v.clone()).ok())
48            .unwrap_or_default();
49        let source_columns: Vec<String> = spec
50            .get("source_columns")
51            .and_then(|v| serde_json::from_value(v.clone()).ok())
52            .unwrap_or_default();
53        let resolved = Self::resolve(pinned, &source_columns);
54        (!resolved.is_empty()).then_some(resolved)
55    }
56
57    /// The authoritative mapping, ordered by index: the assignments the API
58    /// pinned, or, on a spec stored before pinning, each declared source column
59    /// at its position, which is the index its readings were stored under.
60    /// Both crates read an unpinned spec through here so that neither invents
61    /// an index the other would not.
62    #[must_use]
63    pub fn resolve(pinned: Vec<Self>, source_columns: &[String]) -> Vec<Self> {
64        let mut resolved = if pinned.is_empty() {
65            source_columns
66                .iter()
67                .enumerate()
68                .map(|(i, column)| Self {
69                    column: column.clone(),
70                    index: i16::try_from(i).unwrap_or(i16::MAX),
71                    retired: false,
72                })
73                .collect()
74        } else {
75            pinned
76        };
77        resolved.sort_by_key(|a| a.index);
78        resolved
79    }
80}
81
82/// Portal-precomputed mean/sd for one replicate group, sent alongside the
83/// group's readings so the API can compare server-side.
84#[derive(Debug, Clone, Serialize, Deserialize)]
85#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
86#[serde(deny_unknown_fields)]
87pub struct GroupAudit {
88    pub time: DateTime<Utc>,
89    #[serde(default, skip_serializing_if = "Option::is_none")]
90    pub expected_mean: Option<f64>,
91    #[serde(default, skip_serializing_if = "Option::is_none")]
92    pub expected_sd: Option<f64>,
93    /// Count of non-null replicate cells the portal row carries for this
94    /// instant; the API re-counts after admission, so a divergence surfaces
95    /// as an n-mismatch hold.
96    #[serde(default, skip_serializing_if = "Option::is_none")]
97    pub expected_n: Option<i64>,
98}
99
100/// One portal standard curve to register. `source_key` identifies the curve
101/// within the source system; registration is idempotent per (source_system,
102/// source_key).
103#[derive(Debug, Clone, Serialize, Deserialize)]
104#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
105pub struct StandardCurveUpsert {
106    pub source_key: String,
107    /// The portal curve's parameter label. The portal names no instrument for a
108    /// curve, so this is where a pairing plan suggests the attachment from; the
109    /// API holds the curve until a plan attaches it to one of its instruments.
110    pub instrument_label: String,
111    pub slope: f64,
112    pub intercept: f64,
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub r_squared: Option<f64>,
115    #[serde(default, skip_serializing_if = "Option::is_none")]
116    pub name: Option<String>,
117    /// The date the source fitted the curve, which is how the lab identifies one.
118    #[serde(default, skip_serializing_if = "Option::is_none")]
119    pub fitted_on: Option<chrono::NaiveDate>,
120    /// Whatever the source records about the fit.
121    #[serde(default, skip_serializing_if = "Option::is_none")]
122    pub notes: Option<String>,
123}
124
125/// One instrument from a source's own register, to introduce into river-data.
126///
127/// For a source whose instruments do not each have a stream: every other instrument is minted as a
128/// side effect of registering the stream that names it, and a portal's instrument register has no
129/// streams to mint from. Registration is idempotent per (source_system, source_key), and a row the
130/// API already holds under that key is never rewritten.
131#[derive(Debug, Clone, Serialize, Deserialize)]
132#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
133#[serde(deny_unknown_fields)]
134pub struct SensorUpsert {
135    /// The instrument's identity within the source, e.g. "sensor_inventory:62".
136    pub source_key: String,
137    pub name: String,
138    /// The lab's own serial. The API claims it only when no other instrument holds it, and says
139    /// which one does when it declines.
140    #[serde(default, skip_serializing_if = "Option::is_none")]
141    pub serial_number: Option<String>,
142    #[serde(default, skip_serializing_if = "Option::is_none")]
143    pub manufacturer: Option<String>,
144    #[serde(default, skip_serializing_if = "Option::is_none")]
145    pub model: Option<String>,
146    #[serde(default, skip_serializing_if = "Option::is_none")]
147    pub notes: Option<String>,
148    /// True for an instrument that corrects a grab in the lab rather than standing in a river.
149    pub is_lab_instrument: bool,
150    /// The cadence the instrument logs at ('high' | 'low'), read as a declaration when a stream
151    /// classifies its readings. None leaves the API's default of 'high'.
152    #[serde(default, skip_serializing_if = "Option::is_none")]
153    pub data_frequency: Option<String>,
154    /// Whatever the source knows that river-data has no column for.
155    #[serde(default, skip_serializing_if = "Option::is_none")]
156    pub metadata: Option<serde_json::Value>,
157}
158
159/// The API-side identity a registered curve resolved to.
160#[derive(Debug, Clone)]
161pub struct CurveMapping {
162    pub source_key: String,
163    pub id: Uuid,
164    pub sensor_id: Uuid,
165    pub superseded: bool,
166}
167
168#[cfg(test)]
169mod tests {
170    use super::*;
171
172    /// The API receives the declaration, the pinned mapping, the audit expectations and both
173    /// registers, so each must read back as what the client sent.
174    #[test]
175    fn the_registered_wire_types_round_trip() {
176        let spec = ReplicateSpec {
177            source_columns: vec!["DOC_rep_1".into(), "DOC_rep_2".into()],
178            portal_mean_column: Some("DOC_avg".into()),
179            portal_sd_column: None,
180            curve_ref_column: None,
181            calc: Some("calcMean".into()),
182        };
183        let back: ReplicateSpec =
184            serde_json::from_value(serde_json::to_value(&spec).unwrap()).unwrap();
185        assert_eq!(back.source_columns, spec.source_columns);
186
187        let audit = GroupAudit {
188            time: Utc::now(),
189            expected_mean: Some(1.5),
190            expected_sd: Some(0.1),
191            expected_n: Some(3),
192        };
193        let back: GroupAudit =
194            serde_json::from_value(serde_json::to_value(&audit).unwrap()).unwrap();
195        assert_eq!(back.expected_n, Some(3));
196
197        let curve = StandardCurveUpsert {
198            source_key: "standard_curves:3".into(),
199            instrument_label: "DOC corr".into(),
200            slope: 1.0,
201            intercept: 0.0,
202            r_squared: None,
203            name: Some("DOC corr 2021-01-28".into()),
204            fitted_on: chrono::NaiveDate::from_ymd_opt(2021, 1, 28),
205            notes: None,
206        };
207        let back: StandardCurveUpsert =
208            serde_json::from_value(serde_json::to_value(&curve).unwrap()).unwrap();
209        assert_eq!(back.fitted_on, curve.fitted_on);
210        assert!(back.r_squared.is_none());
211
212        let sensor = SensorUpsert {
213            source_key: "sensor_inventory:62".into(),
214            name: "ANU TURB".into(),
215            serial_number: Some("919402".into()),
216            manufacturer: None,
217            model: Some("Cyclops-7".into()),
218            notes: None,
219            is_lab_instrument: false,
220            data_frequency: None,
221            metadata: None,
222        };
223        let back: SensorUpsert =
224            serde_json::from_value(serde_json::to_value(&sensor).unwrap()).unwrap();
225        assert_eq!(back.serial_number.as_deref(), Some("919402"));
226        assert!(!back.is_lab_instrument);
227
228        let assignment = ColumnAssignment {
229            column: "DOC_rep_2".into(),
230            index: 1,
231            retired: true,
232        };
233        let back: ColumnAssignment =
234            serde_json::from_value(serde_json::to_value(&assignment).unwrap()).unwrap();
235        assert_eq!(back, assignment);
236    }
237
238    #[test]
239    fn test_replicate_spec_drops_sd_estimator() {
240        let json = serde_json::json!({
241            "source_columns": ["DOC_rep_1", "DOC_rep_2"],
242            "sd_estimator": "population",
243        });
244        let spec: ReplicateSpec = serde_json::from_value(json).unwrap();
245        assert!(serde_json::to_value(&spec).unwrap().get("sd_estimator").is_none());
246    }
247
248    #[test]
249    fn replicate_spec_skips_absent_fields() {
250        let spec = ReplicateSpec {
251            source_columns: vec!["DIC_A".into(), "DIC_B".into()],
252            portal_mean_column: Some("DIC_avg".into()),
253            portal_sd_column: None,
254            curve_ref_column: None,
255            calc: Some("calcMean".into()),
256        };
257        let json = serde_json::to_value(&spec).unwrap();
258        assert_eq!(json["source_columns"][1], "DIC_B");
259        assert_eq!(json["portal_mean_column"], "DIC_avg");
260        assert!(json.get("portal_sd_column").is_none());
261        assert!(json.get("curve_ref_column").is_none());
262    }
263
264    #[test]
265    fn sensor_upsert_skips_absent_fields() {
266        let up = SensorUpsert {
267            source_key: "sensor_inventory:62".into(),
268            name: "ANU TURB".into(),
269            serial_number: Some("919402".into()),
270            manufacturer: None,
271            model: Some("Cyclops-7".into()),
272            notes: None,
273            is_lab_instrument: false,
274            data_frequency: None,
275            metadata: None,
276        };
277        let json = serde_json::to_value(&up).unwrap();
278        assert_eq!(json["source_key"], "sensor_inventory:62");
279        assert_eq!(json["serial_number"], "919402");
280        assert_eq!(json["is_lab_instrument"], false);
281        assert!(json.get("manufacturer").is_none());
282        assert!(json.get("notes").is_none());
283        assert!(json.get("metadata").is_none());
284    }
285
286    #[test]
287    fn group_audit_skips_absent_fields() {
288        let audit = GroupAudit {
289            time: Utc::now(),
290            expected_mean: Some(1.5),
291            expected_sd: None,
292            expected_n: None,
293        };
294        let json = serde_json::to_value(&audit).unwrap();
295        assert_eq!(json["expected_mean"], 1.5);
296        assert!(json.get("expected_sd").is_none());
297        assert!(json.get("expected_n").is_none());
298    }
299
300    #[test]
301    fn group_audit_serializes_expected_n() {
302        let audit = GroupAudit {
303            time: Utc::now(),
304            expected_mean: Some(1.5),
305            expected_sd: Some(0.1),
306            expected_n: Some(2),
307        };
308        let json = serde_json::to_value(&audit).unwrap();
309        assert_eq!(json["expected_n"], 2);
310    }
311
312    #[test]
313    fn column_assignments_parse_from_metadata() {
314        let metadata = serde_json::json!({
315            "replicates": {
316                "source_columns": ["DOC_rep_1", "DOC_rep_3"],
317                "assignments": [
318                    {"column": "DOC_rep_1", "index": 0},
319                    {"column": "DOC_rep_2", "index": 1, "retired": true},
320                    {"column": "DOC_rep_3", "index": 2, "retired": false},
321                ],
322            },
323        });
324        let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
325        assert_eq!(assignments.len(), 3);
326        assert_eq!(assignments[0].column, "DOC_rep_1");
327        assert!(!assignments[0].retired);
328        assert_eq!(assignments[1].index, 1);
329        assert!(assignments[1].retired);
330    }
331
332    #[test]
333    fn metadata_without_a_replicate_spec_yields_none() {
334        assert!(ColumnAssignment::from_metadata(&serde_json::json!({})).is_none());
335        let no_columns = serde_json::json!({ "replicates": {"source_columns": []} });
336        assert!(ColumnAssignment::from_metadata(&no_columns).is_none());
337    }
338
339    /// An unpinned spec means the same thing on both sides of the wire: each
340    /// declared column at its position, which is what the readings registered
341    /// before pinning carry. The API resolves the metadata it stores the same
342    /// way, so neither crate can index a value the other would index
343    /// differently.
344    #[test]
345    fn an_unpinned_spec_resolves_to_column_positions() {
346        for spec in [
347            serde_json::json!({"source_columns": ["DOC_rep_1", "DOC_rep_2", "DOC_rep_3"]}),
348            serde_json::json!({
349                "source_columns": ["DOC_rep_1", "DOC_rep_2", "DOC_rep_3"],
350                "assignments": [],
351            }),
352        ] {
353            let metadata = serde_json::json!({ "replicates": spec });
354            let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
355            assert_eq!(
356                assignments,
357                vec![
358                    ColumnAssignment {
359                        column: "DOC_rep_1".into(),
360                        index: 0,
361                        retired: false,
362                    },
363                    ColumnAssignment {
364                        column: "DOC_rep_2".into(),
365                        index: 1,
366                        retired: false,
367                    },
368                    ColumnAssignment {
369                        column: "DOC_rep_3".into(),
370                        index: 2,
371                        retired: false,
372                    },
373                ]
374            );
375        }
376    }
377
378    #[test]
379    fn pinned_assignments_win_over_column_order() {
380        let pinned = vec![
381            ColumnAssignment {
382                column: "DOC_rep_2".into(),
383                index: 1,
384                retired: false,
385            },
386            ColumnAssignment {
387                column: "DOC_rep_1".into(),
388                index: 0,
389                retired: false,
390            },
391        ];
392        let resolved =
393            ColumnAssignment::resolve(pinned, &["DOC_rep_1".to_string(), "DOC_rep_2".to_string()]);
394        assert_eq!(resolved[0].column, "DOC_rep_1");
395        assert_eq!(resolved[1].index, 1);
396    }
397
398    /// A column the source stopped sending keeps its index, and the index stays out of use. The
399    /// API pins that state; a client that dropped it, or that renumbered around the gap, would
400    /// store the surviving columns' readings under indexes the store already gave to others.
401    #[test]
402    fn a_retired_column_keeps_its_index_and_the_gap_it_leaves() {
403        let metadata = serde_json::json!({ "replicates": {
404            "source_columns": ["DOC_rep_A", "DOC_rep_C"],
405            "assignments": [
406                { "column": "DOC_rep_C", "index": 2 },
407                { "column": "DOC_rep_A", "index": 0 },
408                { "column": "DOC_rep_B", "index": 1, "retired": true },
409            ],
410        }});
411        let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
412        assert_eq!(
413            assignments,
414            vec![
415                ColumnAssignment {
416                    column: "DOC_rep_A".into(),
417                    index: 0,
418                    retired: false,
419                },
420                ColumnAssignment {
421                    column: "DOC_rep_B".into(),
422                    index: 1,
423                    retired: true,
424                },
425                ColumnAssignment {
426                    column: "DOC_rep_C".into(),
427                    index: 2,
428                    retired: false,
429                },
430            ],
431            "the retired column is kept, in its place, and nothing is renumbered over its index"
432        );
433    }
434
435    /// The declared columns are not the mapping when the API has pinned one: a spec whose
436    /// `source_columns` disagree with its pinned indexes resolves to the pinned indexes, so the
437    /// two crates cannot land on different positions for the same column.
438    #[test]
439    fn pinning_beats_the_declared_column_list() {
440        let metadata = serde_json::json!({ "replicates": {
441            "source_columns": ["DOC_rep_B", "DOC_rep_A"],
442            "assignments": [
443                { "column": "DOC_rep_A", "index": 0 },
444                { "column": "DOC_rep_B", "index": 1 },
445            ],
446        }});
447        let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
448        assert_eq!(
449            assignments.iter().map(|a| a.index).collect::<Vec<_>>(),
450            vec![0, 1]
451        );
452        assert_eq!(assignments[0].column, "DOC_rep_A");
453    }
454
455    /// A family that declares nothing and has nothing pinned is not a mapping of zero columns; it
456    /// is no mapping, and a caller must not read it as "index every column at 0".
457    #[test]
458    fn an_empty_spec_resolves_to_no_mapping() {
459        let metadata = serde_json::json!({ "replicates": { "source_columns": [] } });
460        assert!(ColumnAssignment::from_metadata(&metadata).is_none());
461        assert!(ColumnAssignment::resolve(Vec::new(), &[]).is_empty());
462    }
463
464    #[test]
465    fn curve_upsert_serialization() {
466        let up = StandardCurveUpsert {
467            source_key: "standard_curves:3".into(),
468            instrument_label: "DOC corr".into(),
469            slope: 1.0,
470            intercept: 0.0,
471            r_squared: None,
472            name: Some("DOC corr 2021-01-28".into()),
473            fitted_on: chrono::NaiveDate::from_ymd_opt(2021, 1, 28),
474            notes: None,
475        };
476        let json = serde_json::to_value(&up).unwrap();
477        assert_eq!(json["source_key"], "standard_curves:3");
478        assert_eq!(json["instrument_label"], "DOC corr");
479        assert_eq!(json["fitted_on"], "2021-01-28");
480        assert!(json.get("r_squared").is_none());
481    }
482}
river_data_core/models/replicates.rs

river_data_core/models/
replicates.rs
