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))]
58pub struct IngestReading {
59    pub time: chrono::DateTime<chrono::Utc>,
60    pub raw_value: f64,
61    #[serde(default, skip_serializing_if = "is_zero")]
62    pub replicate_index: i16,
63    #[serde(default, skip_serializing_if = "Option::is_none")]
64    pub sensor_id: Option<Uuid>,
65    #[serde(default, skip_serializing_if = "Option::is_none")]
66    pub calibration_id: Option<Uuid>,
67    #[serde(default, skip_serializing_if = "Option::is_none")]
68    pub deployment_id: Option<Uuid>,
69    /// Per-reading override ('continuous' | 'spot' | 'derived'). None resolves server-side from
70    /// the stream default, then the owning sensor's data_frequency.
71    #[serde(default, skip_serializing_if = "Option::is_none")]
72    pub measurement_type: Option<String>,
73    /// Standard curve the source applied to this reading.
74    #[serde(default, skip_serializing_if = "Option::is_none")]
75    pub standard_curve_id: Option<Uuid>,
76}
77
78impl IngestReading {
79    /// A reading at replicate 0 with no sensor attribution; the server resolves the rest.
80    pub fn new(time: chrono::DateTime<chrono::Utc>, raw_value: f64) -> Self {
81        Self {
82            time,
83            raw_value,
84            replicate_index: 0,
85            sensor_id: None,
86            calibration_id: None,
87            deployment_id: None,
88            measurement_type: None,
89            standard_curve_id: None,
90        }
91    }
92}
93
94fn is_zero(v: &i16) -> bool {
95    *v == 0
96}
97
98#[derive(Debug, Serialize, Deserialize)]
99#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
100pub struct IngestStatusEvent {
101    pub time: chrono::DateTime<chrono::Utc>,
102    pub value: String,
103}
104
105#[cfg(test)]
106mod tests {
107    use super::*;
108
109    /// The API is the receiver of these three, so what the client sends must read back unchanged.
110    #[test]
111    fn ingest_reading_round_trips() {
112        let mut r = IngestReading::new(chrono::Utc::now(), 42.5);
113        r.replicate_index = 2;
114        r.standard_curve_id = Some(Uuid::nil());
115        r.measurement_type = Some("spot".into());
116        let back: IngestReading =
117            serde_json::from_value(serde_json::to_value(&r).unwrap()).unwrap();
118        assert_eq!(back.raw_value, 42.5);
119        assert_eq!(back.replicate_index, 2);
120        assert_eq!(back.measurement_type.as_deref(), Some("spot"));
121        assert_eq!(back.standard_curve_id, Some(Uuid::nil()));
122    }
123
124    /// Replicate 0 is omitted on the wire, so the receiver must read an absent index as 0 rather
125    /// than refusing the reading.
126    #[test]
127    fn an_omitted_replicate_index_reads_as_zero() {
128        let r = IngestReading::new(chrono::Utc::now(), 1.0);
129        let json = serde_json::to_value(&r).unwrap();
130        assert!(json.get("replicate_index").is_none());
131        let back: IngestReading = serde_json::from_value(json).unwrap();
132        assert_eq!(back.replicate_index, 0);
133    }
134
135    #[test]
136    fn register_stream_request_round_trips() {
137        let req = RegisterStreamRequest {
138            source_system: "cnet".to_string(),
139            source_key: "FP1:DOC_avg_ppb:reps".to_string(),
140            source_name: Some("DOC".to_string()),
141            source_path: None,
142            metadata: serde_json::json!({"station": "FP1"}),
143            measurement_type: Some("spot".to_string()),
144            sensor_id: Some(Uuid::nil()),
145            replicates: None,
146            decimal_places: Some(2),
147        };
148        let back: RegisterStreamRequest =
149            serde_json::from_value(serde_json::to_value(&req).unwrap()).unwrap();
150        assert_eq!(back.source_key, req.source_key);
151        assert_eq!(back.decimal_places, Some(2));
152        assert_eq!(back.sensor_id, Some(Uuid::nil()));
153    }
154
155    #[test]
156    fn ingest_status_event_round_trips() {
157        let e = IngestStatusEvent {
158            time: chrono::Utc::now(),
159            value: "unreachable".to_string(),
160        };
161        let back: IngestStatusEvent =
162            serde_json::from_value(serde_json::to_value(&e).unwrap()).unwrap();
163        assert_eq!(back.value, "unreachable");
164    }
165
166    #[test]
167    fn test_ingest_reading_serialization() {
168        let r = IngestReading::new(chrono::Utc::now(), 42.5);
169        let json = serde_json::to_value(&r).unwrap();
170        assert_eq!(json["raw_value"], 42.5);
171        assert!(json.get("replicate_index").is_none());
172        assert!(json.get("sensor_id").is_none());
173        assert!(json.get("measurement_type").is_none());
174    }
175
176    #[test]
177    fn test_register_stream_request() {
178        let req = RegisterStreamRequest {
179            source_system: "test_system".to_string(),
180            source_key: "source_1".to_string(),
181            source_name: Some("stream_a".to_string()),
182            source_path: None,
183            metadata: serde_json::json!({"device": "dev_001"}),
184            measurement_type: None,
185            sensor_id: None,
186            replicates: None,
187            decimal_places: None,
188        };
189        let json = serde_json::to_value(&req).unwrap();
190        assert_eq!(json["source_system"], "test_system");
191        assert_eq!(json["metadata"]["device"], "dev_001");
192        assert!(json.get("sensor_id").is_none());
193        assert!(json.get("replicates").is_none());
194        assert!(json.get("decimal_places").is_none());
195    }
196
197    #[test]
198    fn test_register_stream_request_declares_decimal_places() {
199        let req = RegisterStreamRequest {
200            source_system: "cnet".to_string(),
201            source_key: "VAD:DOC_rep_1".to_string(),
202            source_name: None,
203            source_path: None,
204            metadata: serde_json::json!({}),
205            measurement_type: Some("spot".to_string()),
206            sensor_id: None,
207            replicates: None,
208            decimal_places: Some(2),
209        };
210        let json = serde_json::to_value(&req).unwrap();
211        assert_eq!(json["decimal_places"], 2);
212    }
213
214    #[test]
215    fn test_register_stream_request_with_replicates() {
216        let req = RegisterStreamRequest {
217            source_system: "cnet".to_string(),
218            source_key: "VAD:DOC_avg_ppb:reps".to_string(),
219            source_name: None,
220            source_path: None,
221            metadata: serde_json::json!({}),
222            measurement_type: Some("spot".to_string()),
223            sensor_id: Some(Uuid::nil()),
224            replicates: Some(crate::models::replicates::ReplicateSpec {
225                source_columns: vec!["DOC_rep_1".into(), "DOC_rep_2".into(), "DOC_rep_3".into()],
226                portal_mean_column: Some("DOC_avg_ppb".into()),
227                portal_sd_column: Some("DOC_sd_ppb".into()),
228                curve_ref_column: Some("doc_std_curve_id".into()),
229                calc: Some("calcDOCavg".into()),
230                sd_estimator: None,
231            }),
232            decimal_places: Some(2),
233        };
234        let json = serde_json::to_value(&req).unwrap();
235        assert_eq!(json["measurement_type"], "spot");
236        assert_eq!(json["replicates"]["source_columns"][2], "DOC_rep_3");
237        assert_eq!(json["replicates"]["curve_ref_column"], "doc_std_curve_id");
238    }
239
240    #[test]
241    fn test_data_stream_deserialization() {
242        let json = serde_json::json!({
243            "id": "550e8400-e29b-41d4-a716-446655440000",
244            "source_system": "test_system",
245            "source_key": "source_1",
246            "source_name": "stream_a",
247            "source_path": null,
248            "metadata": {},
249            "site_parameter_id": null,
250            "is_active": true,
251            "last_data_time": null
252        });
253        let stream: DataStream = serde_json::from_value(json).unwrap();
254        assert_eq!(stream.source_system, "test_system");
255        assert!(stream.is_active);
256        assert!(stream.site_parameter_id.is_none());
257        assert!(stream.replicates.is_none());
258    }
259
260    #[test]
261    fn register_response_replicates_parse() {
262        let json = serde_json::json!({
263            "id": "550e8400-e29b-41d4-a716-446655440000",
264            "source_system": "cnet",
265            "source_key": "VAD:DOC_avg_ppb:reps",
266            "source_name": null,
267            "source_path": null,
268            "metadata": {},
269            "site_parameter_id": null,
270            "is_active": true,
271            "last_data_time": null,
272            "replicates": [
273                {"column": "DOC_rep_1", "index": 0},
274                {"column": "DOC_rep_2", "index": 5, "retired": true},
275            ]
276        });
277        let stream: DataStream = serde_json::from_value(json).unwrap();
278        let assignments = stream.replicates.unwrap();
279        assert_eq!(assignments.len(), 2);
280        assert_eq!(assignments[0].index, 0);
281        assert!(!assignments[0].retired);
282        assert_eq!(assignments[1].column, "DOC_rep_2");
283        assert_eq!(assignments[1].index, 5);
284        assert!(assignments[1].retired);
285    }
286}
river_data_core/models/streams.rs

river_data_core/models/
streams.rs
