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 curve resolved to.
164#[derive(Debug, Clone)]
165pub struct CurveMapping {
166    pub source_key: String,
167    pub id: Uuid,
168    pub sensor_id: Uuid,
169    pub superseded: bool,
170}
171
172#[cfg(test)]
173mod tests {
174    use super::*;
175
176    /// The API receives the declaration, the pinned mapping, the audit expectations and both
177    /// registers, so each must read back as what the client sent.
178    #[test]
179    fn the_registered_wire_types_round_trip() {
180        let spec = ReplicateSpec {
181            source_columns: vec!["DOC_rep_1".into(), "DOC_rep_2".into()],
182            portal_mean_column: Some("DOC_avg".into()),
183            portal_sd_column: None,
184            curve_ref_column: None,
185            calc: Some("calcMean".into()),
186            sd_estimator: Some("population".into()),
187        };
188        let back: ReplicateSpec =
189            serde_json::from_value(serde_json::to_value(&spec).unwrap()).unwrap();
190        assert_eq!(back.source_columns, spec.source_columns);
191        assert_eq!(back.sd_estimator.as_deref(), Some("population"));
192
193        let audit = GroupAudit {
194            time: Utc::now(),
195            expected_mean: Some(1.5),
196            expected_sd: Some(0.1),
197            expected_n: Some(3),
198        };
199        let back: GroupAudit =
200            serde_json::from_value(serde_json::to_value(&audit).unwrap()).unwrap();
201        assert_eq!(back.expected_n, Some(3));
202
203        let curve = StandardCurveUpsert {
204            source_key: "standard_curves:3".into(),
205            instrument_label: "DOC corr".into(),
206            slope: 1.0,
207            intercept: 0.0,
208            r_squared: None,
209            name: Some("DOC corr 2021-01-28".into()),
210            fitted_on: chrono::NaiveDate::from_ymd_opt(2021, 1, 28),
211            notes: None,
212        };
213        let back: StandardCurveUpsert =
214            serde_json::from_value(serde_json::to_value(&curve).unwrap()).unwrap();
215        assert_eq!(back.fitted_on, curve.fitted_on);
216        assert!(back.r_squared.is_none());
217
218        let sensor = SensorUpsert {
219            source_key: "sensor_inventory:62".into(),
220            name: "ANU TURB".into(),
221            serial_number: Some("919402".into()),
222            manufacturer: None,
223            model: Some("Cyclops-7".into()),
224            notes: None,
225            is_lab_instrument: false,
226            data_frequency: None,
227            metadata: None,
228        };
229        let back: SensorUpsert =
230            serde_json::from_value(serde_json::to_value(&sensor).unwrap()).unwrap();
231        assert_eq!(back.serial_number.as_deref(), Some("919402"));
232        assert!(!back.is_lab_instrument);
233
234        let assignment = ColumnAssignment {
235            column: "DOC_rep_2".into(),
236            index: 1,
237            retired: true,
238        };
239        let back: ColumnAssignment =
240            serde_json::from_value(serde_json::to_value(&assignment).unwrap()).unwrap();
241        assert_eq!(back, assignment);
242    }
243
244    #[test]
245    fn replicate_spec_skips_absent_fields() {
246        let spec = ReplicateSpec {
247            source_columns: vec!["DIC_A".into(), "DIC_B".into()],
248            portal_mean_column: Some("DIC_avg".into()),
249            portal_sd_column: None,
250            curve_ref_column: None,
251            calc: Some("calcMean".into()),
252            sd_estimator: None,
253        };
254        let json = serde_json::to_value(&spec).unwrap();
255        assert_eq!(json["source_columns"][1], "DIC_B");
256        assert_eq!(json["portal_mean_column"], "DIC_avg");
257        assert!(json.get("portal_sd_column").is_none());
258        assert!(json.get("curve_ref_column").is_none());
259    }
260
261    #[test]
262    fn sensor_upsert_skips_absent_fields() {
263        let up = SensorUpsert {
264            source_key: "sensor_inventory:62".into(),
265            name: "ANU TURB".into(),
266            serial_number: Some("919402".into()),
267            manufacturer: None,
268            model: Some("Cyclops-7".into()),
269            notes: None,
270            is_lab_instrument: false,
271            data_frequency: None,
272            metadata: None,
273        };
274        let json = serde_json::to_value(&up).unwrap();
275        assert_eq!(json["source_key"], "sensor_inventory:62");
276        assert_eq!(json["serial_number"], "919402");
277        assert_eq!(json["is_lab_instrument"], false);
278        assert!(json.get("manufacturer").is_none());
279        assert!(json.get("notes").is_none());
280        assert!(json.get("metadata").is_none());
281    }
282
283    #[test]
284    fn group_audit_skips_absent_fields() {
285        let audit = GroupAudit {
286            time: Utc::now(),
287            expected_mean: Some(1.5),
288            expected_sd: None,
289            expected_n: None,
290        };
291        let json = serde_json::to_value(&audit).unwrap();
292        assert_eq!(json["expected_mean"], 1.5);
293        assert!(json.get("expected_sd").is_none());
294        assert!(json.get("expected_n").is_none());
295    }
296
297    #[test]
298    fn group_audit_serializes_expected_n() {
299        let audit = GroupAudit {
300            time: Utc::now(),
301            expected_mean: Some(1.5),
302            expected_sd: Some(0.1),
303            expected_n: Some(2),
304        };
305        let json = serde_json::to_value(&audit).unwrap();
306        assert_eq!(json["expected_n"], 2);
307    }
308
309    #[test]
310    fn column_assignments_parse_from_metadata() {
311        let metadata = serde_json::json!({
312            "replicates": {
313                "source_columns": ["DOC_rep_1", "DOC_rep_3"],
314                "assignments": [
315                    {"column": "DOC_rep_1", "index": 0},
316                    {"column": "DOC_rep_2", "index": 1, "retired": true},
317                    {"column": "DOC_rep_3", "index": 2, "retired": false},
318                ],
319            },
320        });
321        let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
322        assert_eq!(assignments.len(), 3);
323        assert_eq!(assignments[0].column, "DOC_rep_1");
324        assert!(!assignments[0].retired);
325        assert_eq!(assignments[1].index, 1);
326        assert!(assignments[1].retired);
327    }
328
329    #[test]
330    fn metadata_without_a_replicate_spec_yields_none() {
331        assert!(ColumnAssignment::from_metadata(&serde_json::json!({})).is_none());
332        let no_columns = serde_json::json!({ "replicates": {"source_columns": []} });
333        assert!(ColumnAssignment::from_metadata(&no_columns).is_none());
334    }
335
336    /// An unpinned spec means the same thing on both sides of the wire: each
337    /// declared column at its position, which is what the readings registered
338    /// before pinning carry. The API resolves the metadata it stores the same
339    /// way, so neither crate can index a value the other would index
340    /// differently.
341    #[test]
342    fn an_unpinned_spec_resolves_to_column_positions() {
343        for spec in [
344            serde_json::json!({"source_columns": ["DOC_rep_1", "DOC_rep_2", "DOC_rep_3"]}),
345            serde_json::json!({
346                "source_columns": ["DOC_rep_1", "DOC_rep_2", "DOC_rep_3"],
347                "assignments": [],
348            }),
349        ] {
350            let metadata = serde_json::json!({ "replicates": spec });
351            let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
352            assert_eq!(
353                assignments,
354                vec![
355                    ColumnAssignment {
356                        column: "DOC_rep_1".into(),
357                        index: 0,
358                        retired: false,
359                    },
360                    ColumnAssignment {
361                        column: "DOC_rep_2".into(),
362                        index: 1,
363                        retired: false,
364                    },
365                    ColumnAssignment {
366                        column: "DOC_rep_3".into(),
367                        index: 2,
368                        retired: false,
369                    },
370                ]
371            );
372        }
373    }
374
375    #[test]
376    fn pinned_assignments_win_over_column_order() {
377        let pinned = vec![
378            ColumnAssignment {
379                column: "DOC_rep_2".into(),
380                index: 1,
381                retired: false,
382            },
383            ColumnAssignment {
384                column: "DOC_rep_1".into(),
385                index: 0,
386                retired: false,
387            },
388        ];
389        let resolved =
390            ColumnAssignment::resolve(pinned, &["DOC_rep_1".to_string(), "DOC_rep_2".to_string()]);
391        assert_eq!(resolved[0].column, "DOC_rep_1");
392        assert_eq!(resolved[1].index, 1);
393    }
394
395    /// A column the source stopped sending keeps its index, and the index stays out of use. The
396    /// API pins that state; a client that dropped it, or that renumbered around the gap, would
397    /// store the surviving columns' readings under indexes the store already gave to others.
398    #[test]
399    fn a_retired_column_keeps_its_index_and_the_gap_it_leaves() {
400        let metadata = serde_json::json!({ "replicates": {
401            "source_columns": ["DOC_rep_A", "DOC_rep_C"],
402            "assignments": [
403                { "column": "DOC_rep_C", "index": 2 },
404                { "column": "DOC_rep_A", "index": 0 },
405                { "column": "DOC_rep_B", "index": 1, "retired": true },
406            ],
407        }});
408        let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
409        assert_eq!(
410            assignments,
411            vec![
412                ColumnAssignment {
413                    column: "DOC_rep_A".into(),
414                    index: 0,
415                    retired: false,
416                },
417                ColumnAssignment {
418                    column: "DOC_rep_B".into(),
419                    index: 1,
420                    retired: true,
421                },
422                ColumnAssignment {
423                    column: "DOC_rep_C".into(),
424                    index: 2,
425                    retired: false,
426                },
427            ],
428            "the retired column is kept, in its place, and nothing is renumbered over its index"
429        );
430    }
431
432    /// The declared columns are not the mapping when the API has pinned one: a spec whose
433    /// `source_columns` disagree with its pinned indexes resolves to the pinned indexes, so the
434    /// two crates cannot land on different positions for the same column.
435    #[test]
436    fn pinning_beats_the_declared_column_list() {
437        let metadata = serde_json::json!({ "replicates": {
438            "source_columns": ["DOC_rep_B", "DOC_rep_A"],
439            "assignments": [
440                { "column": "DOC_rep_A", "index": 0 },
441                { "column": "DOC_rep_B", "index": 1 },
442            ],
443        }});
444        let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
445        assert_eq!(
446            assignments.iter().map(|a| a.index).collect::<Vec<_>>(),
447            vec![0, 1]
448        );
449        assert_eq!(assignments[0].column, "DOC_rep_A");
450    }
451
452    /// A family that declares nothing and has nothing pinned is not a mapping of zero columns; it
453    /// is no mapping, and a caller must not read it as "index every column at 0".
454    #[test]
455    fn an_empty_spec_resolves_to_no_mapping() {
456        let metadata = serde_json::json!({ "replicates": { "source_columns": [] } });
457        assert!(ColumnAssignment::from_metadata(&metadata).is_none());
458        assert!(ColumnAssignment::resolve(Vec::new(), &[]).is_empty());
459    }
460
461    #[test]
462    fn curve_upsert_serialization() {
463        let up = StandardCurveUpsert {
464            source_key: "standard_curves:3".into(),
465            instrument_label: "DOC corr".into(),
466            slope: 1.0,
467            intercept: 0.0,
468            r_squared: None,
469            name: Some("DOC corr 2021-01-28".into()),
470            fitted_on: chrono::NaiveDate::from_ymd_opt(2021, 1, 28),
471            notes: None,
472        };
473        let json = serde_json::to_value(&up).unwrap();
474        assert_eq!(json["source_key"], "standard_curves:3");
475        assert_eq!(json["instrument_label"], "DOC corr");
476        assert_eq!(json["fitted_on"], "2021-01-28");
477        assert!(json.get("r_squared").is_none());
478    }
479}
river_data_core/models/replicates.rs

river_data_core/models/
replicates.rs
