river_data_core/models/
streams.rs

1use serde::{Deserialize, Serialize};
2use uuid::Uuid;
3
4use crate::models::replicates::{ColumnAssignment, ReplicateSpec};
5
6#[derive(Debug, Clone, Serialize, Deserialize)]
7pub struct DataStream {
8    pub id: Uuid,
9    pub source_system: String,
10    pub source_key: String,
11    pub source_name: Option<String>,
12    pub source_path: Option<String>,
13    pub metadata: serde_json::Value,
14    pub site_parameter_id: Option<Uuid>,
15    /// Stream-level default for readings.measurement_type ('continuous' | 'spot' | 'derived').
16    /// None defers to the API's sensor-frequency resolution.
17    #[serde(default, skip_serializing_if = "Option::is_none")]
18    pub measurement_type: Option<String>,
19    pub is_active: bool,
20    pub last_data_time: Option<chrono::DateTime<chrono::Utc>>,
21    /// Content digest of the last cleanly-applied windowed pass, as claimed by the sync client.
22    /// Absent on APIs that predate the handshake; the client then sends full windows.
23    #[serde(default, skip_serializing_if = "Option::is_none")]
24    pub last_window_digest: Option<String>,
25    /// The authoritative replicate column-to-index mapping, present on the
26    /// register and list responses for a stream declaring a replicate family.
27    /// Absent on an API that predates it; the same list is then read out of
28    /// `metadata.replicates`.
29    #[serde(default, skip_serializing_if = "Option::is_none")]
30    pub replicates: Option<Vec<ColumnAssignment>>,
31}
32
33#[derive(Debug, Serialize, Deserialize)]
34#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
35pub struct RegisterStreamRequest {
36    pub source_system: String,
37    pub source_key: String,
38    pub source_name: Option<String>,
39    pub source_path: Option<String>,
40    pub metadata: serde_json::Value,
41    /// Stream-level classification declared at discovery. None never clears an operator-set value.
42    #[serde(default, skip_serializing_if = "Option::is_none")]
43    pub measurement_type: Option<String>,
44    /// Owning sensor. Required for curve-carrying streams: the API admits a
45    /// reading's curve claim only when reading-sensor == curve-sensor.
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub sensor_id: Option<Uuid>,
48    /// Replicate-family declaration; requires `measurement_type: "spot"`.
49    #[serde(default, skip_serializing_if = "Option::is_none")]
50    pub replicates: Option<ReplicateSpec>,
51    /// The source's decimal places for this channel (0 to 10). None declares nothing.
52    #[serde(default, skip_serializing_if = "Option::is_none")]
53    pub decimal_places: Option<i16>,
54}
55
56#[derive(Debug, Clone, Serialize, Deserialize)]
57#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
58#[serde(deny_unknown_fields)]
59pub struct IngestReading {
60    pub time: chrono::DateTime<chrono::Utc>,
61    pub raw_value: f64,
62    #[serde(default, skip_serializing_if = "is_zero")]
63    pub replicate_index: i16,
64    #[serde(default, skip_serializing_if = "Option::is_none")]
65    pub sensor_id: Option<Uuid>,
66    #[serde(default, skip_serializing_if = "Option::is_none")]
67    pub calibration_id: Option<Uuid>,
68    #[serde(default, skip_serializing_if = "Option::is_none")]
69    pub deployment_id: Option<Uuid>,
70    /// Per-reading override ('continuous' | 'spot' | 'derived'). None resolves server-side from
71    /// the stream default, then the owning sensor's data_frequency.
72    #[serde(default, skip_serializing_if = "Option::is_none")]
73    pub measurement_type: Option<String>,
74    /// Standard curve the source applied to this reading.
75    #[serde(default, skip_serializing_if = "Option::is_none")]
76    pub standard_curve_id: Option<Uuid>,
77}
78
79impl IngestReading {
80    /// A reading at replicate 0 with no sensor attribution; the server resolves the rest.
81    pub fn new(time: chrono::DateTime<chrono::Utc>, raw_value: f64) -> Self {
82        Self {
83            time,
84            raw_value,
85            replicate_index: 0,
86            sensor_id: None,
87            calibration_id: None,
88            deployment_id: None,
89            measurement_type: None,
90            standard_curve_id: None,
91        }
92    }
93}
94
95fn is_zero(v: &i16) -> bool {
96    *v == 0
97}
98
99#[derive(Debug, Serialize, Deserialize)]
100#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
101#[serde(deny_unknown_fields)]
102pub struct IngestStatusEvent {
103    pub time: chrono::DateTime<chrono::Utc>,
104    pub value: String,
105    /// The instrument the status describes, when the source knows it. None leaves the event
106    /// attributed to the stream alone.
107    #[serde(default, skip_serializing_if = "Option::is_none")]
108    pub sensor_id: Option<Uuid>,
109}
110
111#[cfg(test)]
112mod tests {
113    use super::*;
114
115    /// The API is the receiver of these three, so what the client sends must read back unchanged.
116    #[test]
117    fn ingest_reading_round_trips() {
118        let mut r = IngestReading::new(chrono::Utc::now(), 42.5);
119        r.replicate_index = 2;
120        r.standard_curve_id = Some(Uuid::nil());
121        r.measurement_type = Some("spot".into());
122        let back: IngestReading =
123            serde_json::from_value(serde_json::to_value(&r).unwrap()).unwrap();
124        assert_eq!(back.raw_value, 42.5);
125        assert_eq!(back.replicate_index, 2);
126        assert_eq!(back.measurement_type.as_deref(), Some("spot"));
127        assert_eq!(back.standard_curve_id, Some(Uuid::nil()));
128    }
129
130    /// Replicate 0 is omitted on the wire, so the receiver must read an absent index as 0 rather
131    /// than refusing the reading.
132    #[test]
133    fn an_omitted_replicate_index_reads_as_zero() {
134        let r = IngestReading::new(chrono::Utc::now(), 1.0);
135        let json = serde_json::to_value(&r).unwrap();
136        assert!(json.get("replicate_index").is_none());
137        let back: IngestReading = serde_json::from_value(json).unwrap();
138        assert_eq!(back.replicate_index, 0);
139    }
140
141    #[test]
142    fn register_stream_request_round_trips() {
143        let req = RegisterStreamRequest {
144            source_system: "cnet".to_string(),
145            source_key: "FP1:DOC_avg_ppb:reps".to_string(),
146            source_name: Some("DOC".to_string()),
147            source_path: None,
148            metadata: serde_json::json!({"station": "FP1"}),
149            measurement_type: Some("spot".to_string()),
150            sensor_id: Some(Uuid::nil()),
151            replicates: None,
152            decimal_places: Some(2),
153        };
154        let back: RegisterStreamRequest =
155            serde_json::from_value(serde_json::to_value(&req).unwrap()).unwrap();
156        assert_eq!(back.source_key, req.source_key);
157        assert_eq!(back.decimal_places, Some(2));
158        assert_eq!(back.sensor_id, Some(Uuid::nil()));
159    }
160
161    #[test]
162    fn ingest_status_event_round_trips() {
163        let e = IngestStatusEvent {
164            time: chrono::Utc::now(),
165            value: "unreachable".to_string(),
166            sensor_id: None,
167        };
168        let back: IngestStatusEvent =
169            serde_json::from_value(serde_json::to_value(&e).unwrap()).unwrap();
170        assert_eq!(back.value, "unreachable");
171    }
172
173    #[test]
174    fn test_ingest_reading_serialization() {
175        let r = IngestReading::new(chrono::Utc::now(), 42.5);
176        let json = serde_json::to_value(&r).unwrap();
177        assert_eq!(json["raw_value"], 42.5);
178        assert!(json.get("replicate_index").is_none());
179        assert!(json.get("sensor_id").is_none());
180        assert!(json.get("measurement_type").is_none());
181    }
182
183    #[test]
184    fn test_register_stream_request() {
185        let req = RegisterStreamRequest {
186            source_system: "test_system".to_string(),
187            source_key: "source_1".to_string(),
188            source_name: Some("stream_a".to_string()),
189            source_path: None,
190            metadata: serde_json::json!({"device": "dev_001"}),
191            measurement_type: None,
192            sensor_id: None,
193            replicates: None,
194            decimal_places: None,
195        };
196        let json = serde_json::to_value(&req).unwrap();
197        assert_eq!(json["source_system"], "test_system");
198        assert_eq!(json["metadata"]["device"], "dev_001");
199        assert!(json.get("sensor_id").is_none());
200        assert!(json.get("replicates").is_none());
201        assert!(json.get("decimal_places").is_none());
202    }
203
204    #[test]
205    fn test_register_stream_request_declares_decimal_places() {
206        let req = RegisterStreamRequest {
207            source_system: "cnet".to_string(),
208            source_key: "VAD:DOC_rep_1".to_string(),
209            source_name: None,
210            source_path: None,
211            metadata: serde_json::json!({}),
212            measurement_type: Some("spot".to_string()),
213            sensor_id: None,
214            replicates: None,
215            decimal_places: Some(2),
216        };
217        let json = serde_json::to_value(&req).unwrap();
218        assert_eq!(json["decimal_places"], 2);
219    }
220
221    #[test]
222    fn test_register_stream_request_with_replicates() {
223        let req = RegisterStreamRequest {
224            source_system: "cnet".to_string(),
225            source_key: "VAD:DOC_avg_ppb:reps".to_string(),
226            source_name: None,
227            source_path: None,
228            metadata: serde_json::json!({}),
229            measurement_type: Some("spot".to_string()),
230            sensor_id: Some(Uuid::nil()),
231            replicates: Some(crate::models::replicates::ReplicateSpec {
232                source_columns: vec!["DOC_rep_1".into(), "DOC_rep_2".into(), "DOC_rep_3".into()],
233                portal_mean_column: Some("DOC_avg_ppb".into()),
234                portal_sd_column: Some("DOC_sd_ppb".into()),
235                curve_ref_column: Some("doc_std_curve_id".into()),
236                calc: Some("calcDOCavg".into()),
237                sd_estimator: None,
238            }),
239            decimal_places: Some(2),
240        };
241        let json = serde_json::to_value(&req).unwrap();
242        assert_eq!(json["measurement_type"], "spot");
243        assert_eq!(json["replicates"]["source_columns"][2], "DOC_rep_3");
244        assert_eq!(json["replicates"]["curve_ref_column"], "doc_std_curve_id");
245    }
246
247    #[test]
248    fn test_data_stream_deserialization() {
249        let json = serde_json::json!({
250            "id": "550e8400-e29b-41d4-a716-446655440000",
251            "source_system": "test_system",
252            "source_key": "source_1",
253            "source_name": "stream_a",
254            "source_path": null,
255            "metadata": {},
256            "site_parameter_id": null,
257            "is_active": true,
258            "last_data_time": null
259        });
260        let stream: DataStream = serde_json::from_value(json).unwrap();
261        assert_eq!(stream.source_system, "test_system");
262        assert!(stream.is_active);
263        assert!(stream.site_parameter_id.is_none());
264        assert!(stream.replicates.is_none());
265    }
266
267    #[test]
268    fn register_response_replicates_parse() {
269        let json = serde_json::json!({
270            "id": "550e8400-e29b-41d4-a716-446655440000",
271            "source_system": "cnet",
272            "source_key": "VAD:DOC_avg_ppb:reps",
273            "source_name": null,
274            "source_path": null,
275            "metadata": {},
276            "site_parameter_id": null,
277            "is_active": true,
278            "last_data_time": null,
279            "replicates": [
280                {"column": "DOC_rep_1", "index": 0},
281                {"column": "DOC_rep_2", "index": 5, "retired": true},
282            ]
283        });
284        let stream: DataStream = serde_json::from_value(json).unwrap();
285        let assignments = stream.replicates.unwrap();
286        assert_eq!(assignments.len(), 2);
287        assert_eq!(assignments[0].index, 0);
288        assert!(!assignments[0].retired);
289        assert_eq!(assignments[1].column, "DOC_rep_2");
290        assert_eq!(assignments[1].index, 5);
291        assert!(assignments[1].retired);
292    }
293}
river_data_core/models/streams.rs

river_data_core/models/
streams.rs
