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    /// The source's decimal places for this channel (0 to 10). None declares nothing.
51    #[serde(skip_serializing_if = "Option::is_none")]
52    pub decimal_places: Option<i16>,
53}
54
55#[derive(Debug, Clone, Serialize)]
56pub struct IngestReading {
57    pub time: chrono::DateTime<chrono::Utc>,
58    pub raw_value: f64,
59    #[serde(skip_serializing_if = "is_zero")]
60    pub replicate_index: i16,
61    #[serde(skip_serializing_if = "Option::is_none")]
62    pub sensor_id: Option<Uuid>,
63    #[serde(skip_serializing_if = "Option::is_none")]
64    pub calibration_id: Option<Uuid>,
65    #[serde(skip_serializing_if = "Option::is_none")]
66    pub deployment_id: Option<Uuid>,
67    /// Per-reading override ('continuous' | 'spot' | 'derived'). None resolves server-side from
68    /// the stream default, then the owning sensor's data_frequency.
69    #[serde(skip_serializing_if = "Option::is_none")]
70    pub measurement_type: Option<String>,
71    /// Standard curve the source applied to this reading.
72    #[serde(skip_serializing_if = "Option::is_none")]
73    pub standard_curve_id: Option<Uuid>,
74}
75
76impl IngestReading {
77    /// A reading at replicate 0 with no sensor attribution; the server resolves the rest.
78    pub fn new(time: chrono::DateTime<chrono::Utc>, raw_value: f64) -> Self {
79        Self {
80            time,
81            raw_value,
82            replicate_index: 0,
83            sensor_id: None,
84            calibration_id: None,
85            deployment_id: None,
86            measurement_type: None,
87            standard_curve_id: None,
88        }
89    }
90}
91
92fn is_zero(v: &i16) -> bool {
93    *v == 0
94}
95
96#[derive(Debug, Serialize)]
97pub struct IngestStatusEvent {
98    pub time: chrono::DateTime<chrono::Utc>,
99    pub value: String,
100}
101
102#[cfg(test)]
103mod tests {
104    use super::*;
105
106    #[test]
107    fn test_ingest_reading_serialization() {
108        let r = IngestReading::new(chrono::Utc::now(), 42.5);
109        let json = serde_json::to_value(&r).unwrap();
110        assert_eq!(json["raw_value"], 42.5);
111        assert!(json.get("replicate_index").is_none());
112        assert!(json.get("sensor_id").is_none());
113        assert!(json.get("measurement_type").is_none());
114    }
115
116    #[test]
117    fn test_register_stream_request() {
118        let req = RegisterStreamRequest {
119            source_system: "test_system".to_string(),
120            source_key: "source_1".to_string(),
121            source_name: Some("stream_a".to_string()),
122            source_path: None,
123            metadata: serde_json::json!({"device": "dev_001"}),
124            measurement_type: None,
125            sensor_id: None,
126            replicates: None,
127            decimal_places: None,
128        };
129        let json = serde_json::to_value(&req).unwrap();
130        assert_eq!(json["source_system"], "test_system");
131        assert_eq!(json["metadata"]["device"], "dev_001");
132        assert!(json.get("sensor_id").is_none());
133        assert!(json.get("replicates").is_none());
134        assert!(json.get("decimal_places").is_none());
135    }
136
137    #[test]
138    fn test_register_stream_request_declares_decimal_places() {
139        let req = RegisterStreamRequest {
140            source_system: "cnet".to_string(),
141            source_key: "VAD:DOC_rep_1".to_string(),
142            source_name: None,
143            source_path: None,
144            metadata: serde_json::json!({}),
145            measurement_type: Some("spot".to_string()),
146            sensor_id: None,
147            replicates: None,
148            decimal_places: Some(2),
149        };
150        let json = serde_json::to_value(&req).unwrap();
151        assert_eq!(json["decimal_places"], 2);
152    }
153
154    #[test]
155    fn test_register_stream_request_with_replicates() {
156        let req = RegisterStreamRequest {
157            source_system: "cnet".to_string(),
158            source_key: "VAD:DOC_avg_ppb:reps".to_string(),
159            source_name: None,
160            source_path: None,
161            metadata: serde_json::json!({}),
162            measurement_type: Some("spot".to_string()),
163            sensor_id: Some(Uuid::nil()),
164            replicates: Some(crate::models::replicates::ReplicateSpec {
165                source_columns: vec!["DOC_rep_1".into(), "DOC_rep_2".into(), "DOC_rep_3".into()],
166                portal_mean_column: Some("DOC_avg_ppb".into()),
167                portal_sd_column: Some("DOC_sd_ppb".into()),
168                curve_ref_column: Some("doc_std_curve_id".into()),
169                calc: Some("calcDOCavg".into()),
170                sd_estimator: None,
171            }),
172            decimal_places: Some(2),
173        };
174        let json = serde_json::to_value(&req).unwrap();
175        assert_eq!(json["measurement_type"], "spot");
176        assert_eq!(json["replicates"]["source_columns"][2], "DOC_rep_3");
177        assert_eq!(json["replicates"]["curve_ref_column"], "doc_std_curve_id");
178    }
179
180    #[test]
181    fn test_data_stream_deserialization() {
182        let json = serde_json::json!({
183            "id": "550e8400-e29b-41d4-a716-446655440000",
184            "source_system": "test_system",
185            "source_key": "source_1",
186            "source_name": "stream_a",
187            "source_path": null,
188            "metadata": {},
189            "site_parameter_id": null,
190            "is_active": true,
191            "last_data_time": null
192        });
193        let stream: DataStream = serde_json::from_value(json).unwrap();
194        assert_eq!(stream.source_system, "test_system");
195        assert!(stream.is_active);
196        assert!(stream.site_parameter_id.is_none());
197        assert!(stream.replicates.is_none());
198    }
199
200    #[test]
201    fn register_response_replicates_parse() {
202        let json = serde_json::json!({
203            "id": "550e8400-e29b-41d4-a716-446655440000",
204            "source_system": "cnet",
205            "source_key": "VAD:DOC_avg_ppb:reps",
206            "source_name": null,
207            "source_path": null,
208            "metadata": {},
209            "site_parameter_id": null,
210            "is_active": true,
211            "last_data_time": null,
212            "replicates": [
213                {"column": "DOC_rep_1", "index": 0},
214                {"column": "DOC_rep_2", "index": 5, "retired": true},
215            ]
216        });
217        let stream: DataStream = serde_json::from_value(json).unwrap();
218        let assignments = stream.replicates.unwrap();
219        assert_eq!(assignments.len(), 2);
220        assert_eq!(assignments[0].index, 0);
221        assert!(!assignments[0].retired);
222        assert_eq!(assignments[1].column, "DOC_rep_2");
223        assert_eq!(assignments[1].index, 5);
224        assert!(assignments[1].retired);
225    }
226}
river_data_core/models/streams.rs

river_data_core/models/
streams.rs
