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)]
12pub struct ReplicateSpec {
13    pub source_columns: Vec<String>,
14    #[serde(default, skip_serializing_if = "Option::is_none")]
15    pub portal_mean_column: Option<String>,
16    #[serde(default, skip_serializing_if = "Option::is_none")]
17    pub portal_sd_column: Option<String>,
18    #[serde(default, skip_serializing_if = "Option::is_none")]
19    pub curve_ref_column: Option<String>,
20    #[serde(default, skip_serializing_if = "Option::is_none")]
21    pub calc: Option<String>,
22    /// The sd divisor the source's own sd column uses ('sample' | 'population'),
23    /// when the source declares one. Never inferred; None leaves the slot's
24    /// declaration (or the audit gate) to decide.
25    #[serde(default, skip_serializing_if = "Option::is_none")]
26    pub sd_estimator: Option<String>,
27}
28
29/// One source column's pinned replicate index, as the API's register response
30/// reports it (and as stream metadata persists it under
31/// `replicates.assignments`). Sync services assign each value's
32/// `replicate_index` by looking its source column up here.
33#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
34pub struct ColumnAssignment {
35    pub column: String,
36    pub index: i16,
37    /// The source no longer sends this column. The index stays reserved and
38    /// remains the column's identity should it reappear.
39    #[serde(default)]
40    pub retired: bool,
41}
42
43impl ColumnAssignment {
44    /// The pinned mapping a stream's metadata carries, when the API has
45    /// authored one. None on metadata written by an API that predates pinning.
46    pub fn from_metadata(metadata: &serde_json::Value) -> Option<Vec<Self>> {
47        let value = metadata.get("replicates")?.get("assignments")?;
48        let assignments: Vec<Self> = serde_json::from_value(value.clone()).ok()?;
49        if assignments.is_empty() {
50            None
51        } else {
52            Some(assignments)
53        }
54    }
55}
56
57/// Portal-precomputed mean/sd for one replicate group, sent alongside the
58/// group's readings so the API can compare server-side.
59#[derive(Debug, Clone, Serialize)]
60pub struct GroupAudit {
61    pub time: DateTime<Utc>,
62    #[serde(skip_serializing_if = "Option::is_none")]
63    pub expected_mean: Option<f64>,
64    #[serde(skip_serializing_if = "Option::is_none")]
65    pub expected_sd: Option<f64>,
66    /// Count of non-null replicate cells the portal row carries for this
67    /// instant; the API re-counts after admission, so a divergence surfaces
68    /// as an n-mismatch hold.
69    #[serde(default, skip_serializing_if = "Option::is_none")]
70    pub expected_n: Option<i64>,
71}
72
73/// One portal standard curve to register. `source_key` identifies the curve
74/// within the source system; registration is idempotent per (source_system,
75/// source_key).
76#[derive(Debug, Clone, Serialize)]
77pub struct StandardCurveUpsert {
78    pub source_key: String,
79    /// The portal curve's parameter label; the API finds-or-creates one lab
80    /// instrument per (source_system, instrument_label).
81    pub instrument_label: String,
82    pub slope: f64,
83    pub intercept: f64,
84    #[serde(skip_serializing_if = "Option::is_none")]
85    pub r_squared: Option<f64>,
86    #[serde(skip_serializing_if = "Option::is_none")]
87    pub name: Option<String>,
88    /// The date the source fitted the curve, which is how the lab identifies one.
89    #[serde(skip_serializing_if = "Option::is_none")]
90    pub fitted_on: Option<chrono::NaiveDate>,
91}
92
93/// One instrument from a source's own register, to introduce into river-data.
94///
95/// For a source whose instruments do not each have a stream: every other instrument is minted as a
96/// side effect of registering the stream that names it, and a portal's instrument register has no
97/// streams to mint from. Registration is idempotent per (source_system, source_key), and a row the
98/// API already holds under that key is never rewritten.
99#[derive(Debug, Clone, Serialize)]
100pub struct SensorUpsert {
101    /// The instrument's identity within the source, e.g. "sensor_inventory:62".
102    pub source_key: String,
103    pub name: String,
104    /// The lab's own serial. The API claims it only when no other instrument holds it, and says
105    /// which one does when it declines.
106    #[serde(skip_serializing_if = "Option::is_none")]
107    pub serial_number: Option<String>,
108    #[serde(skip_serializing_if = "Option::is_none")]
109    pub manufacturer: Option<String>,
110    #[serde(skip_serializing_if = "Option::is_none")]
111    pub model: Option<String>,
112    #[serde(skip_serializing_if = "Option::is_none")]
113    pub notes: Option<String>,
114    /// True for an instrument that corrects a grab in the lab rather than standing in a river.
115    pub is_lab_instrument: bool,
116    /// Whatever the source knows that river-data has no column for.
117    #[serde(skip_serializing_if = "Option::is_none")]
118    pub metadata: Option<serde_json::Value>,
119}
120
121/// The API-side identity a registered instrument resolved to.
122#[derive(Debug, Clone)]
123pub struct SensorMapping {
124    pub source_key: String,
125    pub id: Uuid,
126    /// False when the API already held an instrument under this key.
127    pub created: bool,
128    /// The instrument already holding the offered serial, when the API declined to claim it.
129    pub serial_claimed_by: Option<Uuid>,
130}
131
132/// The API-side identity a registered curve resolved to.
133#[derive(Debug, Clone)]
134pub struct CurveMapping {
135    pub source_key: String,
136    pub id: Uuid,
137    pub sensor_id: Uuid,
138    pub superseded: bool,
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144
145    #[test]
146    fn replicate_spec_skips_absent_fields() {
147        let spec = ReplicateSpec {
148            source_columns: vec!["DIC_A".into(), "DIC_B".into()],
149            portal_mean_column: Some("DIC_avg".into()),
150            portal_sd_column: None,
151            curve_ref_column: None,
152            calc: Some("calcMean".into()),
153            sd_estimator: None,
154        };
155        let json = serde_json::to_value(&spec).unwrap();
156        assert_eq!(json["source_columns"][1], "DIC_B");
157        assert_eq!(json["portal_mean_column"], "DIC_avg");
158        assert!(json.get("portal_sd_column").is_none());
159        assert!(json.get("curve_ref_column").is_none());
160    }
161
162    #[test]
163    fn sensor_upsert_skips_absent_fields() {
164        let up = SensorUpsert {
165            source_key: "sensor_inventory:62".into(),
166            name: "ANU TURB".into(),
167            serial_number: Some("919402".into()),
168            manufacturer: None,
169            model: Some("Cyclops-7".into()),
170            notes: None,
171            is_lab_instrument: false,
172            metadata: None,
173        };
174        let json = serde_json::to_value(&up).unwrap();
175        assert_eq!(json["source_key"], "sensor_inventory:62");
176        assert_eq!(json["serial_number"], "919402");
177        assert_eq!(json["is_lab_instrument"], false);
178        assert!(json.get("manufacturer").is_none());
179        assert!(json.get("notes").is_none());
180        assert!(json.get("metadata").is_none());
181    }
182
183    #[test]
184    fn group_audit_skips_absent_fields() {
185        let audit = GroupAudit {
186            time: Utc::now(),
187            expected_mean: Some(1.5),
188            expected_sd: None,
189            expected_n: None,
190        };
191        let json = serde_json::to_value(&audit).unwrap();
192        assert_eq!(json["expected_mean"], 1.5);
193        assert!(json.get("expected_sd").is_none());
194        assert!(json.get("expected_n").is_none());
195    }
196
197    #[test]
198    fn group_audit_serializes_expected_n() {
199        let audit = GroupAudit {
200            time: Utc::now(),
201            expected_mean: Some(1.5),
202            expected_sd: Some(0.1),
203            expected_n: Some(2),
204        };
205        let json = serde_json::to_value(&audit).unwrap();
206        assert_eq!(json["expected_n"], 2);
207    }
208
209    #[test]
210    fn column_assignments_parse_from_metadata() {
211        let metadata = serde_json::json!({
212            "replicates": {
213                "source_columns": ["DOC_rep_1", "DOC_rep_3"],
214                "assignments": [
215                    {"column": "DOC_rep_1", "index": 0},
216                    {"column": "DOC_rep_2", "index": 1, "retired": true},
217                    {"column": "DOC_rep_3", "index": 2, "retired": false},
218                ],
219            },
220        });
221        let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
222        assert_eq!(assignments.len(), 3);
223        assert_eq!(assignments[0].column, "DOC_rep_1");
224        assert!(!assignments[0].retired);
225        assert_eq!(assignments[1].index, 1);
226        assert!(assignments[1].retired);
227    }
228
229    #[test]
230    fn metadata_without_pinned_assignments_yields_none() {
231        assert!(ColumnAssignment::from_metadata(&serde_json::json!({})).is_none());
232        let unpinned = serde_json::json!({
233            "replicates": {"source_columns": ["a", "b"]},
234        });
235        assert!(ColumnAssignment::from_metadata(&unpinned).is_none());
236        let empty = serde_json::json!({
237            "replicates": {"source_columns": ["a", "b"], "assignments": []},
238        });
239        assert!(ColumnAssignment::from_metadata(&empty).is_none());
240    }
241
242    #[test]
243    fn curve_upsert_serialization() {
244        let up = StandardCurveUpsert {
245            source_key: "standard_curves:3".into(),
246            instrument_label: "DOC corr".into(),
247            slope: 1.0,
248            intercept: 0.0,
249            r_squared: None,
250            name: Some("DOC corr 2021-01-28".into()),
251            fitted_on: chrono::NaiveDate::from_ymd_opt(2021, 1, 28),
252        };
253        let json = serde_json::to_value(&up).unwrap();
254        assert_eq!(json["source_key"], "standard_curves:3");
255        assert_eq!(json["instrument_label"], "DOC corr");
256        assert_eq!(json["fitted_on"], "2021-01-28");
257        assert!(json.get("r_squared").is_none());
258    }
259}
river_data_core/models/replicates.rs

river_data_core/models/
replicates.rs
