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}
89
90/// The API-side identity a registered curve resolved to.
91#[derive(Debug, Clone)]
92pub struct CurveMapping {
93    pub source_key: String,
94    pub id: Uuid,
95    pub sensor_id: Uuid,
96    pub superseded: bool,
97}
98
99#[cfg(test)]
100mod tests {
101    use super::*;
102
103    #[test]
104    fn replicate_spec_skips_absent_fields() {
105        let spec = ReplicateSpec {
106            source_columns: vec!["DIC_A".into(), "DIC_B".into()],
107            portal_mean_column: Some("DIC_avg".into()),
108            portal_sd_column: None,
109            curve_ref_column: None,
110            calc: Some("calcMean".into()),
111            sd_estimator: None,
112        };
113        let json = serde_json::to_value(&spec).unwrap();
114        assert_eq!(json["source_columns"][1], "DIC_B");
115        assert_eq!(json["portal_mean_column"], "DIC_avg");
116        assert!(json.get("portal_sd_column").is_none());
117        assert!(json.get("curve_ref_column").is_none());
118    }
119
120    #[test]
121    fn group_audit_skips_absent_fields() {
122        let audit = GroupAudit {
123            time: Utc::now(),
124            expected_mean: Some(1.5),
125            expected_sd: None,
126            expected_n: None,
127        };
128        let json = serde_json::to_value(&audit).unwrap();
129        assert_eq!(json["expected_mean"], 1.5);
130        assert!(json.get("expected_sd").is_none());
131        assert!(json.get("expected_n").is_none());
132    }
133
134    #[test]
135    fn group_audit_serializes_expected_n() {
136        let audit = GroupAudit {
137            time: Utc::now(),
138            expected_mean: Some(1.5),
139            expected_sd: Some(0.1),
140            expected_n: Some(2),
141        };
142        let json = serde_json::to_value(&audit).unwrap();
143        assert_eq!(json["expected_n"], 2);
144    }
145
146    #[test]
147    fn column_assignments_parse_from_metadata() {
148        let metadata = serde_json::json!({
149            "replicates": {
150                "source_columns": ["DOC_rep_1", "DOC_rep_3"],
151                "assignments": [
152                    {"column": "DOC_rep_1", "index": 0},
153                    {"column": "DOC_rep_2", "index": 1, "retired": true},
154                    {"column": "DOC_rep_3", "index": 2, "retired": false},
155                ],
156            },
157        });
158        let assignments = ColumnAssignment::from_metadata(&metadata).unwrap();
159        assert_eq!(assignments.len(), 3);
160        assert_eq!(assignments[0].column, "DOC_rep_1");
161        assert!(!assignments[0].retired);
162        assert_eq!(assignments[1].index, 1);
163        assert!(assignments[1].retired);
164    }
165
166    #[test]
167    fn metadata_without_pinned_assignments_yields_none() {
168        assert!(ColumnAssignment::from_metadata(&serde_json::json!({})).is_none());
169        let unpinned = serde_json::json!({
170            "replicates": {"source_columns": ["a", "b"]},
171        });
172        assert!(ColumnAssignment::from_metadata(&unpinned).is_none());
173        let empty = serde_json::json!({
174            "replicates": {"source_columns": ["a", "b"], "assignments": []},
175        });
176        assert!(ColumnAssignment::from_metadata(&empty).is_none());
177    }
178
179    #[test]
180    fn curve_upsert_serialization() {
181        let up = StandardCurveUpsert {
182            source_key: "standard_curves:3".into(),
183            instrument_label: "DOC corr".into(),
184            slope: 1.0,
185            intercept: 0.0,
186            r_squared: None,
187            name: Some("DOC corr 2021-01-28".into()),
188        };
189        let json = serde_json::to_value(&up).unwrap();
190        assert_eq!(json["source_key"], "standard_curves:3");
191        assert_eq!(json["instrument_label"], "DOC corr");
192        assert!(json.get("r_squared").is_none());
193    }
194}
river_data_core/models/replicates.rs

river_data_core/models/
replicates.rs
