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            }),
255            decimal_places: Some(2),
256            instrument_granularity: None,
257        };
258        let json = serde_json::to_value(&req).unwrap();
259        assert_eq!(json["measurement_type"], "spot");
260        assert_eq!(json["replicates"]["source_columns"][2], "DOC_rep_3");
261        assert_eq!(json["replicates"]["curve_ref_column"], "doc_std_curve_id");
262    }
263
264    #[test]
265    fn test_register_stream_request_declares_instrument_granularity() {
266        let req = RegisterStreamRequest {
267            source_system: "vaisala".to_string(),
268            source_key: "1270".to_string(),
269            source_name: None,
270            source_path: None,
271            metadata: serde_json::json!({}),
272            measurement_type: None,
273            sensor_id: None,
274            replicates: None,
275            decimal_places: None,
276            instrument_granularity: Some(InstrumentGranularity::PerSiteParameter),
277        };
278        let json = serde_json::to_value(&req).unwrap();
279        assert_eq!(json["instrument_granularity"], "per_site_parameter");
280        let back: RegisterStreamRequest = serde_json::from_value(json).unwrap();
281        assert_eq!(
282            back.instrument_granularity,
283            Some(InstrumentGranularity::PerSiteParameter)
284        );
285    }
286
287    /// A connector built against a core without the field declares nothing, which the API reads as
288    /// its own inference.
289    #[test]
290    fn test_register_stream_request_without_instrument_granularity() {
291        let req: RegisterStreamRequest = serde_json::from_value(serde_json::json!({
292            "source_system": "cnet",
293            "source_key": "VAD:DOC_rep_1",
294            "metadata": {}
295        }))
296        .unwrap();
297        assert_eq!(req.instrument_granularity, None);
298        assert!(
299            serde_json::to_value(&req)
300                .unwrap()
301                .get("instrument_granularity")
302                .is_none()
303        );
304    }
305
306    #[test]
307    fn test_data_stream_deserialization() {
308        let json = serde_json::json!({
309            "id": "550e8400-e29b-41d4-a716-446655440000",
310            "source_system": "test_system",
311            "source_key": "source_1",
312            "source_name": "stream_a",
313            "source_path": null,
314            "metadata": {},
315            "site_parameter_id": null,
316            "is_active": true,
317            "last_data_time": null
318        });
319        let stream: DataStream = serde_json::from_value(json).unwrap();
320        assert_eq!(stream.source_system, "test_system");
321        assert!(stream.is_active);
322        assert!(stream.site_parameter_id.is_none());
323        assert!(stream.replicates.is_none());
324    }
325
326    #[test]
327    fn register_response_replicates_parse() {
328        let json = serde_json::json!({
329            "id": "550e8400-e29b-41d4-a716-446655440000",
330            "source_system": "cnet",
331            "source_key": "VAD:DOC_avg_ppb:reps",
332            "source_name": null,
333            "source_path": null,
334            "metadata": {},
335            "site_parameter_id": null,
336            "is_active": true,
337            "last_data_time": null,
338            "replicates": [
339                {"column": "DOC_rep_1", "index": 0},
340                {"column": "DOC_rep_2", "index": 5, "retired": true},
341            ]
342        });
343        let stream: DataStream = serde_json::from_value(json).unwrap();
344        let assignments = stream.replicates.unwrap();
345        assert_eq!(assignments.len(), 2);
346        assert_eq!(assignments[0].index, 0);
347        assert!(!assignments[0].retired);
348        assert_eq!(assignments[1].column, "DOC_rep_2");
349        assert_eq!(assignments[1].index, 5);
350        assert!(assignments[1].retired);
351    }
352}
river_data_core/models/streams.rs

river_data_core/models/
streams.rs
