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 a
26    /// register response for a stream declaring a replicate family. Absent on
27    /// list responses and on APIs that predate pinning; the same list persists
28    /// under `metadata.replicates.assignments`.
29    #[serde(default, skip_serializing_if = "Option::is_none")]
30    pub replicates: Option<Vec<ColumnAssignment>>,
31}
32
33#[derive(Debug, Serialize)]
34pub struct RegisterStreamRequest {
35    pub source_system: String,
36    pub source_key: String,
37    pub source_name: Option<String>,
38    pub source_path: Option<String>,
39    pub metadata: serde_json::Value,
40    /// Stream-level classification declared at discovery. None never clears an operator-set value.
41    #[serde(skip_serializing_if = "Option::is_none")]
42    pub measurement_type: Option<String>,
43    /// Owning sensor. Required for curve-carrying streams: the API admits a
44    /// reading's curve claim only when reading-sensor == curve-sensor.
45    #[serde(skip_serializing_if = "Option::is_none")]
46    pub sensor_id: Option<Uuid>,
47    /// Replicate-family declaration; requires `measurement_type: "spot"`.
48    #[serde(skip_serializing_if = "Option::is_none")]
49    pub replicates: Option<ReplicateSpec>,
50}
51
52#[derive(Debug, Clone, Serialize)]
53pub struct IngestReading {
54    pub time: chrono::DateTime<chrono::Utc>,
55    pub raw_value: f64,
56    #[serde(skip_serializing_if = "is_zero")]
57    pub replicate_index: i16,
58    #[serde(skip_serializing_if = "Option::is_none")]
59    pub sensor_id: Option<Uuid>,
60    #[serde(skip_serializing_if = "Option::is_none")]
61    pub calibration_id: Option<Uuid>,
62    #[serde(skip_serializing_if = "Option::is_none")]
63    pub deployment_id: Option<Uuid>,
64    /// Per-reading override ('continuous' | 'spot' | 'derived'). None resolves server-side from
65    /// the stream default, then the owning sensor's data_frequency.
66    #[serde(skip_serializing_if = "Option::is_none")]
67    pub measurement_type: Option<String>,
68    /// Standard curve the source applied to this reading.
69    #[serde(skip_serializing_if = "Option::is_none")]
70    pub standard_curve_id: Option<Uuid>,
71}
72
73impl IngestReading {
74    /// A reading at replicate 0 with no sensor attribution; the server resolves the rest.
75    pub fn new(time: chrono::DateTime<chrono::Utc>, raw_value: f64) -> Self {
76        Self {
77            time,
78            raw_value,
79            replicate_index: 0,
80            sensor_id: None,
81            calibration_id: None,
82            deployment_id: None,
83            measurement_type: None,
84            standard_curve_id: None,
85        }
86    }
87}
88
89fn is_zero(v: &i16) -> bool {
90    *v == 0
91}
92
93#[derive(Debug, Serialize)]
94pub struct IngestStatusEvent {
95    pub time: chrono::DateTime<chrono::Utc>,
96    pub value: String,
97}
98
99#[cfg(test)]
100mod tests {
101    use super::*;
102
103    #[test]
104    fn test_ingest_reading_serialization() {
105        let r = IngestReading::new(chrono::Utc::now(), 42.5);
106        let json = serde_json::to_value(&r).unwrap();
107        assert_eq!(json["raw_value"], 42.5);
108        assert!(json.get("replicate_index").is_none());
109        assert!(json.get("sensor_id").is_none());
110        assert!(json.get("measurement_type").is_none());
111    }
112
113    #[test]
114    fn test_register_stream_request() {
115        let req = RegisterStreamRequest {
116            source_system: "test_system".to_string(),
117            source_key: "source_1".to_string(),
118            source_name: Some("stream_a".to_string()),
119            source_path: None,
120            metadata: serde_json::json!({"device": "dev_001"}),
121            measurement_type: None,
122            sensor_id: None,
123            replicates: None,
124        };
125        let json = serde_json::to_value(&req).unwrap();
126        assert_eq!(json["source_system"], "test_system");
127        assert_eq!(json["metadata"]["device"], "dev_001");
128        assert!(json.get("sensor_id").is_none());
129        assert!(json.get("replicates").is_none());
130    }
131
132    #[test]
133    fn test_register_stream_request_with_replicates() {
134        let req = RegisterStreamRequest {
135            source_system: "cnet".to_string(),
136            source_key: "VAD:DOC_avg_ppb:reps".to_string(),
137            source_name: None,
138            source_path: None,
139            metadata: serde_json::json!({}),
140            measurement_type: Some("spot".to_string()),
141            sensor_id: Some(Uuid::nil()),
142            replicates: Some(crate::models::replicates::ReplicateSpec {
143                source_columns: vec!["DOC_rep_1".into(), "DOC_rep_2".into(), "DOC_rep_3".into()],
144                portal_mean_column: Some("DOC_avg_ppb".into()),
145                portal_sd_column: Some("DOC_sd_ppb".into()),
146                curve_ref_column: Some("doc_std_curve_id".into()),
147                calc: Some("calcDOCavg".into()),
148                sd_estimator: None,
149            }),
150        };
151        let json = serde_json::to_value(&req).unwrap();
152        assert_eq!(json["measurement_type"], "spot");
153        assert_eq!(json["replicates"]["source_columns"][2], "DOC_rep_3");
154        assert_eq!(json["replicates"]["curve_ref_column"], "doc_std_curve_id");
155    }
156
157    #[test]
158    fn test_data_stream_deserialization() {
159        let json = serde_json::json!({
160            "id": "550e8400-e29b-41d4-a716-446655440000",
161            "source_system": "test_system",
162            "source_key": "source_1",
163            "source_name": "stream_a",
164            "source_path": null,
165            "metadata": {},
166            "site_parameter_id": null,
167            "is_active": true,
168            "last_data_time": null
169        });
170        let stream: DataStream = serde_json::from_value(json).unwrap();
171        assert_eq!(stream.source_system, "test_system");
172        assert!(stream.is_active);
173        assert!(stream.site_parameter_id.is_none());
174        assert!(stream.replicates.is_none());
175    }
176
177    #[test]
178    fn register_response_replicates_parse() {
179        let json = serde_json::json!({
180            "id": "550e8400-e29b-41d4-a716-446655440000",
181            "source_system": "cnet",
182            "source_key": "VAD:DOC_avg_ppb:reps",
183            "source_name": null,
184            "source_path": null,
185            "metadata": {},
186            "site_parameter_id": null,
187            "is_active": true,
188            "last_data_time": null,
189            "replicates": [
190                {"column": "DOC_rep_1", "index": 0},
191                {"column": "DOC_rep_2", "index": 5, "retired": true},
192            ]
193        });
194        let stream: DataStream = serde_json::from_value(json).unwrap();
195        let assignments = stream.replicates.unwrap();
196        assert_eq!(assignments.len(), 2);
197        assert_eq!(assignments[0].index, 0);
198        assert!(!assignments[0].retired);
199        assert_eq!(assignments[1].column, "DOC_rep_2");
200        assert_eq!(assignments[1].index, 5);
201        assert!(assignments[1].retired);
202    }
203}
river_data_core/models/streams.rs

river_data_core/models/
streams.rs
