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))]
91pub struct GroupAudit {
92    pub time: DateTime<Utc>,
93    #[serde(default, skip_serializing_if = "Option::is_none")]
94    pub expected_mean: Option<f64>,
95    #[serde(default, skip_serializing_if = "Option::is_none")]
96    pub expected_sd: Option<f64>,
97    /// Count of non-null replicate cells the portal row carries for this
98    /// instant; the API re-counts after admission, so a divergence surfaces
99    /// as an n-mismatch hold.
100    #[serde(default, skip_serializing_if = "Option::is_none")]
101    pub expected_n: Option<i64>,
102}
103
104/// One portal standard curve to register. `source_key` identifies the curve
105/// within the source system; registration is idempotent per (source_system,
106/// source_key).
107#[derive(Debug, Clone, Serialize, Deserialize)]
108#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
109pub struct StandardCurveUpsert {
110    pub source_key: String,
111    /// The portal curve's parameter label; the API finds-or-creates one lab
112    /// instrument per (source_system, instrument_label).
113    pub instrument_label: String,
114    pub slope: f64,
115    pub intercept: f64,
116    #[serde(default, skip_serializing_if = "Option::is_none")]
117    pub r_squared: Option<f64>,
118    #[serde(default, skip_serializing_if = "Option::is_none")]
119    pub name: Option<String>,
120    /// The date the source fitted the curve, which is how the lab identifies one.
121    #[serde(default, skip_serializing_if = "Option::is_none")]
122    pub fitted_on: Option<chrono::NaiveDate>,
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))]
133pub struct SensorUpsert {
134    /// The instrument's identity within the source, e.g. "sensor_inventory:62".
135    pub source_key: String,
136    pub name: String,
137    /// The lab's own serial. The API claims it only when no other instrument holds it, and says
138    /// which one does when it declines.
139    #[serde(default, skip_serializing_if = "Option::is_none")]
140    pub serial_number: Option<String>,
141    #[serde(default, skip_serializing_if = "Option::is_none")]
142    pub manufacturer: Option<String>,
143    #[serde(default, skip_serializing_if = "Option::is_none")]
144    pub model: Option<String>,
145    #[serde(default, skip_serializing_if = "Option::is_none")]
146    pub notes: Option<String>,
147    /// True for an instrument that corrects a grab in the lab rather than standing in a river.
148    pub is_lab_instrument: bool,
149    /// Whatever the source knows that river-data has no column for.
150    #[serde(default, skip_serializing_if = "Option::is_none")]
151    pub metadata: Option<serde_json::Value>,
152}
153
154/// The API-side identity a registered instrument resolved to.
155#[derive(Debug, Clone)]
156pub struct SensorMapping {
157    pub source_key: String,
158    pub id: Uuid,
159    /// False when the API already held an instrument under this key.
160    pub created: bool,
161    /// The instrument already holding the offered serial, when the API declined to claim it.
162    pub serial_claimed_by: Option<Uuid>,
163}
164
165/// The API-side identity a registered curve resolved to.
166#[derive(Debug, Clone)]
167pub struct CurveMapping {
168    pub source_key: String,
169    pub id: Uuid,
170    pub sensor_id: Uuid,
171    pub superseded: bool,
172}
173
174#[cfg(test)]
175mod tests {
176    use super::*;
177
178    /// The API receives the declaration, the pinned mapping, the audit expectations and both
179    /// registers, so each must read back as what the client sent.
180    #[test]
181    fn the_registered_wire_types_round_trip() {
182        let spec = ReplicateSpec {
183            source_columns: vec!["DOC_rep_1".into(), "DOC_rep_2".into()],
184            portal_mean_column: Some("DOC_avg".into()),
185            portal_sd_column: None,
186            curve_ref_column: None,
187            calc: Some("calcMean".into()),
188            sd_estimator: Some("population".into()),
189        };
190        let back: ReplicateSpec =
191            serde_json::from_value(serde_json::to_value(&spec).unwrap()).unwrap();
192        assert_eq!(back.source_columns, spec.source_columns);
193        assert_eq!(back.sd_estimator.as_deref(), Some("population"));
194
195        let audit = GroupAudit {
196            time: Utc::now(),
197            expected_mean: Some(1.5),
198            expected_sd: Some(0.1),
199            expected_n: Some(3),
200        };
201        let back: GroupAudit =
202            serde_json::from_value(serde_json::to_value(&audit).unwrap()).unwrap();
203        assert_eq!(back.expected_n, Some(3));
204
205        let curve = StandardCurveUpsert {
206            source_key: "standard_curves:3".into(),
207            instrument_label: "DOC corr".into(),
208            slope: 1.0,
209            intercept: 0.0,
210            r_squared: None,
211            name: Some("DOC corr 2021-01-28".into()),
212            fitted_on: chrono::NaiveDate::from_ymd_opt(2021, 1, 28),
213        };
214        let back: StandardCurveUpsert =
215            serde_json::from_value(serde_json::to_value(&curve).unwrap()).unwrap();
216        assert_eq!(back.fitted_on, curve.fitted_on);
217        assert!(back.r_squared.is_none());
218
219        let sensor = SensorUpsert {
220            source_key: "sensor_inventory:62".into(),
221            name: "ANU TURB".into(),
222            serial_number: Some("919402".into()),
223            manufacturer: None,
224            model: Some("Cyclops-7".into()),
225            notes: None,
226            is_lab_instrument: false,
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            metadata: None,
272        };
273        let json = serde_json::to_value(&up).unwrap();
274        assert_eq!(json["source_key"], "sensor_inventory:62");
275        assert_eq!(json["serial_number"], "919402");
276        assert_eq!(json["is_lab_instrument"], false);
277        assert!(json.get("manufacturer").is_none());
278        assert!(json.get("notes").is_none());
279        assert!(json.get("metadata").is_none());
280    }
281
282    #[test]
283    fn group_audit_skips_absent_fields() {
284        let audit = GroupAudit {
285            time: Utc::now(),
286            expected_mean: Some(1.5),
287            expected_sd: None,
288            expected_n: None,
289        };
290        let json = serde_json::to_value(&audit).unwrap();
291        assert_eq!(json["expected_mean"], 1.5);
292        assert!(json.get("expected_sd").is_none());
293        assert!(json.get("expected_n").is_none());
294    }
295
296    #[test]
297    fn group_audit_serializes_expected_n() {
298        let audit = GroupAudit {
299            time: Utc::now(),
300            expected_mean: Some(1.5),
301            expected_sd: Some(0.1),
302            expected_n: Some(2),
303        };
304        let json = serde_json::to_value(&audit).unwrap();
305        assert_eq!(json["expected_n"], 2);
306    }
307
308    #[test]
309    fn column_assignments_parse_from_metadata() {
310        let metadata = serde_json::json!({
311            "replicates": {
312                "source_columns": ["DOC_rep_1", "DOC_rep_3"],
313                "assignments": [
314                    {"column": "DOC_rep_1", "index": 0},
315                    {"column": "DOC_rep_2", "index": 1, "retired": true},
316                    {"column": "DOC_rep_3", "index": 2, "retired": false},
317                ],
318            },
319        });
320        let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
321        assert_eq!(assignments.len(), 3);
322        assert_eq!(assignments[0].column, "DOC_rep_1");
323        assert!(!assignments[0].retired);
324        assert_eq!(assignments[1].index, 1);
325        assert!(assignments[1].retired);
326    }
327
328    #[test]
329    fn metadata_without_a_replicate_spec_yields_none() {
330        assert!(ColumnAssignment::from_metadata(&serde_json::json!({})).is_none());
331        let no_columns = serde_json::json!({ "replicates": {"source_columns": []} });
332        assert!(ColumnAssignment::from_metadata(&no_columns).is_none());
333    }
334
335    /// An unpinned spec means the same thing on both sides of the wire: each
336    /// declared column at its position, which is what the readings registered
337    /// before pinning carry. The API resolves the metadata it stores the same
338    /// way, so neither crate can index a value the other would index
339    /// differently.
340    #[test]
341    fn an_unpinned_spec_resolves_to_column_positions() {
342        for spec in [
343            serde_json::json!({"source_columns": ["DOC_rep_1", "DOC_rep_2", "DOC_rep_3"]}),
344            serde_json::json!({
345                "source_columns": ["DOC_rep_1", "DOC_rep_2", "DOC_rep_3"],
346                "assignments": [],
347            }),
348        ] {
349            let metadata = serde_json::json!({ "replicates": spec });
350            let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
351            assert_eq!(
352                assignments,
353                vec![
354                    ColumnAssignment {
355                        column: "DOC_rep_1".into(),
356                        index: 0,
357                        retired: false,
358                    },
359                    ColumnAssignment {
360                        column: "DOC_rep_2".into(),
361                        index: 1,
362                        retired: false,
363                    },
364                    ColumnAssignment {
365                        column: "DOC_rep_3".into(),
366                        index: 2,
367                        retired: false,
368                    },
369                ]
370            );
371        }
372    }
373
374    #[test]
375    fn pinned_assignments_win_over_column_order() {
376        let pinned = vec![
377            ColumnAssignment {
378                column: "DOC_rep_2".into(),
379                index: 1,
380                retired: false,
381            },
382            ColumnAssignment {
383                column: "DOC_rep_1".into(),
384                index: 0,
385                retired: false,
386            },
387        ];
388        let resolved = ColumnAssignment::resolve(
389            pinned,
390            &["DOC_rep_1".to_string(), "DOC_rep_2".to_string()],
391        );
392        assert_eq!(resolved[0].column, "DOC_rep_1");
393        assert_eq!(resolved[1].index, 1);
394    }
395
396    /// A column the source stopped sending keeps its index, and the index stays out of use. The
397    /// API pins that state; a client that dropped it, or that renumbered around the gap, would
398    /// store the surviving columns' readings under indexes the store already gave to others.
399    #[test]
400    fn a_retired_column_keeps_its_index_and_the_gap_it_leaves() {
401        let metadata = serde_json::json!({ "replicates": {
402            "source_columns": ["DOC_rep_A", "DOC_rep_C"],
403            "assignments": [
404                { "column": "DOC_rep_C", "index": 2 },
405                { "column": "DOC_rep_A", "index": 0 },
406                { "column": "DOC_rep_B", "index": 1, "retired": true },
407            ],
408        }});
409        let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
410        assert_eq!(
411            assignments,
412            vec![
413                ColumnAssignment {
414                    column: "DOC_rep_A".into(),
415                    index: 0,
416                    retired: false,
417                },
418                ColumnAssignment {
419                    column: "DOC_rep_B".into(),
420                    index: 1,
421                    retired: true,
422                },
423                ColumnAssignment {
424                    column: "DOC_rep_C".into(),
425                    index: 2,
426                    retired: false,
427                },
428            ],
429            "the retired column is kept, in its place, and nothing is renumbered over its index"
430        );
431    }
432
433    /// The declared columns are not the mapping when the API has pinned one: a spec whose
434    /// `source_columns` disagree with its pinned indexes resolves to the pinned indexes, so the
435    /// two crates cannot land on different positions for the same column.
436    #[test]
437    fn pinning_beats_the_declared_column_list() {
438        let metadata = serde_json::json!({ "replicates": {
439            "source_columns": ["DOC_rep_B", "DOC_rep_A"],
440            "assignments": [
441                { "column": "DOC_rep_A", "index": 0 },
442                { "column": "DOC_rep_B", "index": 1 },
443            ],
444        }});
445        let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
446        assert_eq!(
447            assignments.iter().map(|a| a.index).collect::<Vec<_>>(),
448            vec![0, 1]
449        );
450        assert_eq!(assignments[0].column, "DOC_rep_A");
451    }
452
453    /// A family that declares nothing and has nothing pinned is not a mapping of zero columns; it
454    /// is no mapping, and a caller must not read it as "index every column at 0".
455    #[test]
456    fn an_empty_spec_resolves_to_no_mapping() {
457        let metadata = serde_json::json!({ "replicates": { "source_columns": [] } });
458        assert!(ColumnAssignment::from_metadata(&metadata).is_none());
459        assert!(ColumnAssignment::resolve(Vec::new(), &[]).is_empty());
460    }
461
462    #[test]
463    fn curve_upsert_serialization() {
464        let up = StandardCurveUpsert {
465            source_key: "standard_curves:3".into(),
466            instrument_label: "DOC corr".into(),
467            slope: 1.0,
468            intercept: 0.0,
469            r_squared: None,
470            name: Some("DOC corr 2021-01-28".into()),
471            fitted_on: chrono::NaiveDate::from_ymd_opt(2021, 1, 28),
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
