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/// How many instruments a source's channels stand for, declared by the connector.
34#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
35#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
36#[serde(rename_all = "snake_case")]
37pub enum InstrumentGranularity {
38    /// One instrument per parameter across every site the source reports it at (a field portal).
39    PerParameter,
40    /// One instrument per site and parameter, stationed there (a logger network).
41    PerSiteParameter,
42}
43
44#[derive(Debug, Serialize, Deserialize)]
45#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
46pub struct RegisterStreamRequest {
47    pub source_system: String,
48    pub source_key: String,
49    pub source_name: Option<String>,
50    pub source_path: Option<String>,
51    pub metadata: serde_json::Value,
52    /// Stream-level classification declared at discovery. None never clears an operator-set value.
53    #[serde(default, skip_serializing_if = "Option::is_none")]
54    pub measurement_type: Option<String>,
55    /// Owning sensor. Required for curve-carrying streams: the API admits a
56    /// reading's curve claim only when reading-sensor == curve-sensor.
57    #[serde(default, skip_serializing_if = "Option::is_none")]
58    pub sensor_id: Option<Uuid>,
59    /// Replicate-family declaration; requires `measurement_type: "spot"`.
60    #[serde(default, skip_serializing_if = "Option::is_none")]
61    pub replicates: Option<ReplicateSpec>,
62    /// The source's decimal places for this channel (0 to 10). None declares nothing.
63    #[serde(default, skip_serializing_if = "Option::is_none")]
64    pub decimal_places: Option<i16>,
65    /// The instrument this channel suggests at pairing. None leaves it to the API's inference.
66    #[serde(default, skip_serializing_if = "Option::is_none")]
67    pub instrument_granularity: Option<InstrumentGranularity>,
68}
69
70#[derive(Debug, Clone, Serialize, Deserialize)]
71#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
72#[serde(deny_unknown_fields)]
73pub struct IngestReading {
74    pub time: chrono::DateTime<chrono::Utc>,
75    pub raw_value: f64,
76    #[serde(default, skip_serializing_if = "is_zero")]
77    pub replicate_index: i16,
78    #[serde(default, skip_serializing_if = "Option::is_none")]
79    pub sensor_id: Option<Uuid>,
80    #[serde(default, skip_serializing_if = "Option::is_none")]
81    pub calibration_id: Option<Uuid>,
82    #[serde(default, skip_serializing_if = "Option::is_none")]
83    pub deployment_id: Option<Uuid>,
84    /// Per-reading override ('continuous' | 'spot' | 'derived'). None resolves server-side from
85    /// the stream default, then the owning sensor's data_frequency.
86    #[serde(default, skip_serializing_if = "Option::is_none")]
87    pub measurement_type: Option<String>,
88    /// Standard curve the source applied to this reading.
89    #[serde(default, skip_serializing_if = "Option::is_none")]
90    pub standard_curve_id: Option<Uuid>,
91}
92
93impl IngestReading {
94    /// A reading at replicate 0 with no sensor attribution; the server resolves the rest.
95    pub fn new(time: chrono::DateTime<chrono::Utc>, raw_value: f64) -> Self {
96        Self {
97            time,
98            raw_value,
99            replicate_index: 0,
100            sensor_id: None,
101            calibration_id: None,
102            deployment_id: None,
103            measurement_type: None,
104            standard_curve_id: None,
105        }
106    }
107}
108
109fn is_zero(v: &i16) -> bool {
110    *v == 0
111}
112
113#[derive(Debug, Serialize, Deserialize)]
114#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
115#[serde(deny_unknown_fields)]
116pub struct IngestStatusEvent {
117    pub time: chrono::DateTime<chrono::Utc>,
118    pub value: String,
119    /// The instrument the status describes, when the source knows it. None leaves the event
120    /// attributed to the stream alone.
121    #[serde(default, skip_serializing_if = "Option::is_none")]
122    pub sensor_id: Option<Uuid>,
123}
124
125#[cfg(test)]
126mod tests {
127    use super::*;
128
129    /// The API is the receiver of these three, so what the client sends must read back unchanged.
130    #[test]
131    fn ingest_reading_round_trips() {
132        let mut r = IngestReading::new(chrono::Utc::now(), 42.5);
133        r.replicate_index = 2;
134        r.standard_curve_id = Some(Uuid::nil());
135        r.measurement_type = Some("spot".into());
136        let back: IngestReading =
137            serde_json::from_value(serde_json::to_value(&r).unwrap()).unwrap();
138        assert_eq!(back.raw_value, 42.5);
139        assert_eq!(back.replicate_index, 2);
140        assert_eq!(back.measurement_type.as_deref(), Some("spot"));
141        assert_eq!(back.standard_curve_id, Some(Uuid::nil()));
142    }
143
144    /// Replicate 0 is omitted on the wire, so the receiver must read an absent index as 0 rather
145    /// than refusing the reading.
146    #[test]
147    fn an_omitted_replicate_index_reads_as_zero() {
148        let r = IngestReading::new(chrono::Utc::now(), 1.0);
149        let json = serde_json::to_value(&r).unwrap();
150        assert!(json.get("replicate_index").is_none());
151        let back: IngestReading = serde_json::from_value(json).unwrap();
152        assert_eq!(back.replicate_index, 0);
153    }
154
155    #[test]
156    fn register_stream_request_round_trips() {
157        let req = RegisterStreamRequest {
158            source_system: "cnet".to_string(),
159            source_key: "FP1:DOC_avg_ppb:reps".to_string(),
160            source_name: Some("DOC".to_string()),
161            source_path: None,
162            metadata: serde_json::json!({"station": "FP1"}),
163            measurement_type: Some("spot".to_string()),
164            sensor_id: Some(Uuid::nil()),
165            replicates: None,
166            decimal_places: Some(2),
167            instrument_granularity: None,
168        };
169        let back: RegisterStreamRequest =
170            serde_json::from_value(serde_json::to_value(&req).unwrap()).unwrap();
171        assert_eq!(back.source_key, req.source_key);
172        assert_eq!(back.decimal_places, Some(2));
173        assert_eq!(back.sensor_id, Some(Uuid::nil()));
174    }
175
176    #[test]
177    fn ingest_status_event_round_trips() {
178        let e = IngestStatusEvent {
179            time: chrono::Utc::now(),
180            value: "unreachable".to_string(),
181            sensor_id: None,
182        };
183        let back: IngestStatusEvent =
184            serde_json::from_value(serde_json::to_value(&e).unwrap()).unwrap();
185        assert_eq!(back.value, "unreachable");
186    }
187
188    #[test]
189    fn test_ingest_reading_serialization() {
190        let r = IngestReading::new(chrono::Utc::now(), 42.5);
191        let json = serde_json::to_value(&r).unwrap();
192        assert_eq!(json["raw_value"], 42.5);
193        assert!(json.get("replicate_index").is_none());
194        assert!(json.get("sensor_id").is_none());
195        assert!(json.get("measurement_type").is_none());
196    }
197
198    #[test]
199    fn test_register_stream_request() {
200        let req = RegisterStreamRequest {
201            source_system: "test_system".to_string(),
202            source_key: "source_1".to_string(),
203            source_name: Some("stream_a".to_string()),
204            source_path: None,
205            metadata: serde_json::json!({"device": "dev_001"}),
206            measurement_type: None,
207            sensor_id: None,
208            replicates: None,
209            decimal_places: None,
210            instrument_granularity: None,
211        };
212        let json = serde_json::to_value(&req).unwrap();
213        assert_eq!(json["source_system"], "test_system");
214        assert_eq!(json["metadata"]["device"], "dev_001");
215        assert!(json.get("sensor_id").is_none());
216        assert!(json.get("replicates").is_none());
217        assert!(json.get("decimal_places").is_none());
218    }
219
220    #[test]
221    fn test_register_stream_request_declares_decimal_places() {
222        let req = RegisterStreamRequest {
223            source_system: "cnet".to_string(),
224            source_key: "VAD:DOC_rep_1".to_string(),
225            source_name: None,
226            source_path: None,
227            metadata: serde_json::json!({}),
228            measurement_type: Some("spot".to_string()),
229            sensor_id: None,
230            replicates: None,
231            decimal_places: Some(2),
232            instrument_granularity: None,
233        };
234        let json = serde_json::to_value(&req).unwrap();
235        assert_eq!(json["decimal_places"], 2);
236    }
237
238    #[test]
239    fn test_register_stream_request_with_replicates() {
240        let req = RegisterStreamRequest {
241            source_system: "cnet".to_string(),
242            source_key: "VAD:DOC_avg_ppb:reps".to_string(),
243            source_name: None,
244            source_path: None,
245            metadata: serde_json::json!({}),
246            measurement_type: Some("spot".to_string()),
247            sensor_id: Some(Uuid::nil()),
248            replicates: Some(crate::models::replicates::ReplicateSpec {
249                source_columns: vec!["DOC_rep_1".into(), "DOC_rep_2".into(), "DOC_rep_3".into()],
250                portal_mean_column: Some("DOC_avg_ppb".into()),
251                portal_sd_column: Some("DOC_sd_ppb".into()),
252                curve_ref_column: Some("doc_std_curve_id".into()),
253                calc: Some("calcDOCavg".into()),
254                sd_estimator: None,
255            }),
256            decimal_places: Some(2),
257            instrument_granularity: None,
258        };
259        let json = serde_json::to_value(&req).unwrap();
260        assert_eq!(json["measurement_type"], "spot");
261        assert_eq!(json["replicates"]["source_columns"][2], "DOC_rep_3");
262        assert_eq!(json["replicates"]["curve_ref_column"], "doc_std_curve_id");
263    }
264
265    #[test]
266    fn test_register_stream_request_declares_instrument_granularity() {
267        let req = RegisterStreamRequest {
268            source_system: "vaisala".to_string(),
269            source_key: "1270".to_string(),
270            source_name: None,
271            source_path: None,
272            metadata: serde_json::json!({}),
273            measurement_type: None,
274            sensor_id: None,
275            replicates: None,
276            decimal_places: None,
277            instrument_granularity: Some(InstrumentGranularity::PerSiteParameter),
278        };
279        let json = serde_json::to_value(&req).unwrap();
280        assert_eq!(json["instrument_granularity"], "per_site_parameter");
281        let back: RegisterStreamRequest = serde_json::from_value(json).unwrap();
282        assert_eq!(
283            back.instrument_granularity,
284            Some(InstrumentGranularity::PerSiteParameter)
285        );
286    }
287
288    /// A connector built against a core without the field declares nothing, which the API reads as
289    /// its own inference.
290    #[test]
291    fn test_register_stream_request_without_instrument_granularity() {
292        let req: RegisterStreamRequest = serde_json::from_value(serde_json::json!({
293            "source_system": "cnet",
294            "source_key": "VAD:DOC_rep_1",
295            "metadata": {}
296        }))
297        .unwrap();
298        assert_eq!(req.instrument_granularity, None);
299        assert!(
300            serde_json::to_value(&req)
301                .unwrap()
302                .get("instrument_granularity")
303                .is_none()
304        );
305    }
306
307    #[test]
308    fn test_data_stream_deserialization() {
309        let json = serde_json::json!({
310            "id": "550e8400-e29b-41d4-a716-446655440000",
311            "source_system": "test_system",
312            "source_key": "source_1",
313            "source_name": "stream_a",
314            "source_path": null,
315            "metadata": {},
316            "site_parameter_id": null,
317            "is_active": true,
318            "last_data_time": null
319        });
320        let stream: DataStream = serde_json::from_value(json).unwrap();
321        assert_eq!(stream.source_system, "test_system");
322        assert!(stream.is_active);
323        assert!(stream.site_parameter_id.is_none());
324        assert!(stream.replicates.is_none());
325    }
326
327    #[test]
328    fn register_response_replicates_parse() {
329        let json = serde_json::json!({
330            "id": "550e8400-e29b-41d4-a716-446655440000",
331            "source_system": "cnet",
332            "source_key": "VAD:DOC_avg_ppb:reps",
333            "source_name": null,
334            "source_path": null,
335            "metadata": {},
336            "site_parameter_id": null,
337            "is_active": true,
338            "last_data_time": null,
339            "replicates": [
340                {"column": "DOC_rep_1", "index": 0},
341                {"column": "DOC_rep_2", "index": 5, "retired": true},
342            ]
343        });
344        let stream: DataStream = serde_json::from_value(json).unwrap();
345        let assignments = stream.replicates.unwrap();
346        assert_eq!(assignments.len(), 2);
347        assert_eq!(assignments[0].index, 0);
348        assert!(!assignments[0].retired);
349        assert_eq!(assignments[1].column, "DOC_rep_2");
350        assert_eq!(assignments[1].index, 5);
351        assert!(assignments[1].retired);
352    }
353}
river_data_core/models/streams.rs

river_data_core/models/
streams.rs
