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}
23
24/// One source column's pinned replicate index, as the API's register response
25/// reports it (and as stream metadata persists it under
26/// `replicates.assignments`). Sync services assign each value's
27/// `replicate_index` by looking its source column up here.
28#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
29pub struct ColumnAssignment {
30    pub column: String,
31    pub index: i16,
32    /// The source no longer sends this column. The index stays reserved and
33    /// remains the column's identity should it reappear.
34    #[serde(default)]
35    pub retired: bool,
36}
37
38impl ColumnAssignment {
39    /// The pinned mapping a stream's metadata carries, when the API has
40    /// authored one. None on metadata written by an API that predates pinning.
41    pub fn from_metadata(metadata: &serde_json::Value) -> Option<Vec<Self>> {
42        let value = metadata.get("replicates")?.get("assignments")?;
43        let assignments: Vec<Self> = serde_json::from_value(value.clone()).ok()?;
44        if assignments.is_empty() {
45            None
46        } else {
47            Some(assignments)
48        }
49    }
50}
51
52/// Portal-precomputed mean/sd for one replicate group, sent alongside the
53/// group's readings so the API can compare server-side.
54#[derive(Debug, Clone, Serialize)]
55pub struct GroupAudit {
56    pub time: DateTime<Utc>,
57    #[serde(skip_serializing_if = "Option::is_none")]
58    pub expected_mean: Option<f64>,
59    #[serde(skip_serializing_if = "Option::is_none")]
60    pub expected_sd: Option<f64>,
61    /// Count of non-null replicate cells the portal row carries for this
62    /// instant; the API re-counts after admission, so a divergence surfaces
63    /// as an n-mismatch hold.
64    #[serde(default, skip_serializing_if = "Option::is_none")]
65    pub expected_n: Option<i64>,
66}
67
68/// One portal standard curve to register. `source_key` identifies the curve
69/// within the source system; registration is idempotent per (source_system,
70/// source_key).
71#[derive(Debug, Clone, Serialize)]
72pub struct StandardCurveUpsert {
73    pub source_key: String,
74    /// The portal curve's parameter label; the API finds-or-creates one lab
75    /// instrument per (source_system, instrument_label).
76    pub instrument_label: String,
77    pub slope: f64,
78    pub intercept: f64,
79    #[serde(skip_serializing_if = "Option::is_none")]
80    pub r_squared: Option<f64>,
81    #[serde(skip_serializing_if = "Option::is_none")]
82    pub name: Option<String>,
83}
84
85/// The API-side identity a registered curve resolved to.
86#[derive(Debug, Clone)]
87pub struct CurveMapping {
88    pub source_key: String,
89    pub id: Uuid,
90    pub sensor_id: Uuid,
91    pub superseded: bool,
92}
93
94#[cfg(test)]
95mod tests {
96    use super::*;
97
98    #[test]
99    fn replicate_spec_skips_absent_fields() {
100        let spec = ReplicateSpec {
101            source_columns: vec!["DIC_A".into(), "DIC_B".into()],
102            portal_mean_column: Some("DIC_avg".into()),
103            portal_sd_column: None,
104            curve_ref_column: None,
105            calc: Some("calcMean".into()),
106        };
107        let json = serde_json::to_value(&spec).unwrap();
108        assert_eq!(json["source_columns"][1], "DIC_B");
109        assert_eq!(json["portal_mean_column"], "DIC_avg");
110        assert!(json.get("portal_sd_column").is_none());
111        assert!(json.get("curve_ref_column").is_none());
112    }
113
114    #[test]
115    fn group_audit_skips_absent_fields() {
116        let audit = GroupAudit {
117            time: Utc::now(),
118            expected_mean: Some(1.5),
119            expected_sd: None,
120            expected_n: None,
121        };
122        let json = serde_json::to_value(&audit).unwrap();
123        assert_eq!(json["expected_mean"], 1.5);
124        assert!(json.get("expected_sd").is_none());
125        assert!(json.get("expected_n").is_none());
126    }
127
128    #[test]
129    fn group_audit_serializes_expected_n() {
130        let audit = GroupAudit {
131            time: Utc::now(),
132            expected_mean: Some(1.5),
133            expected_sd: Some(0.1),
134            expected_n: Some(2),
135        };
136        let json = serde_json::to_value(&audit).unwrap();
137        assert_eq!(json["expected_n"], 2);
138    }
139
140    #[test]
141    fn column_assignments_parse_from_metadata() {
142        let metadata = serde_json::json!({
143            "replicates": {
144                "source_columns": ["DOC_rep_1", "DOC_rep_3"],
145                "assignments": [
146                    {"column": "DOC_rep_1", "index": 0},
147                    {"column": "DOC_rep_2", "index": 1, "retired": true},
148                    {"column": "DOC_rep_3", "index": 2, "retired": false},
149                ],
150            },
151        });
152        let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
153        assert_eq!(assignments.len(), 3);
154        assert_eq!(assignments[0].column, "DOC_rep_1");
155        assert!(!assignments[0].retired);
156        assert_eq!(assignments[1].index, 1);
157        assert!(assignments[1].retired);
158    }
159
160    #[test]
161    fn metadata_without_pinned_assignments_yields_none() {
162        assert!(ColumnAssignment::from_metadata(&serde_json::json!({})).is_none());
163        let unpinned = serde_json::json!({
164            "replicates": {"source_columns": ["a", "b"]},
165        });
166        assert!(ColumnAssignment::from_metadata(&unpinned).is_none());
167        let empty = serde_json::json!({
168            "replicates": {"source_columns": ["a", "b"], "assignments": []},
169        });
170        assert!(ColumnAssignment::from_metadata(&empty).is_none());
171    }
172
173    #[test]
174    fn curve_upsert_serialization() {
175        let up = StandardCurveUpsert {
176            source_key: "standard_curves:3".into(),
177            instrument_label: "DOC corr".into(),
178            slope: 1.0,
179            intercept: 0.0,
180            r_squared: None,
181            name: Some("DOC corr 2021-01-28".into()),
182        };
183        let json = serde_json::to_value(&up).unwrap();
184        assert_eq!(json["source_key"], "standard_curves:3");
185        assert_eq!(json["instrument_label"], "DOC corr");
186        assert!(json.get("r_squared").is_none());
187    }
188}
river_data_core/models/replicates.rs

river_data_core/models/
replicates.rs
