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

river_data_core/models/
replicates.rs
