river_data_core/models/
streams.rs

1use serde::{Deserialize, Serialize};
2use uuid::Uuid;
3
4#[derive(Debug, Clone, Serialize, Deserialize)]
5pub struct DataStream {
6    pub id: Uuid,
7    pub source_system: String,
8    pub source_key: String,
9    pub source_name: Option<String>,
10    pub source_path: Option<String>,
11    pub metadata: serde_json::Value,
12    pub site_parameter_id: Option<Uuid>,
13    /// Stream-level default for readings.measurement_type ('continuous' | 'spot' | 'derived').
14    /// None defers to the API's sensor-frequency resolution.
15    #[serde(default, skip_serializing_if = "Option::is_none")]
16    pub measurement_type: Option<String>,
17    pub is_active: bool,
18    pub last_data_time: Option<chrono::DateTime<chrono::Utc>>,
19}
20
21#[derive(Debug, Serialize)]
22pub struct RegisterStreamRequest {
23    pub source_system: String,
24    pub source_key: String,
25    pub source_name: Option<String>,
26    pub source_path: Option<String>,
27    pub metadata: serde_json::Value,
28    /// Stream-level classification declared at discovery. None never clears an operator-set value.
29    #[serde(skip_serializing_if = "Option::is_none")]
30    pub measurement_type: Option<String>,
31}
32
33#[derive(Debug, Clone, Serialize)]
34pub struct IngestReading {
35    pub time: chrono::DateTime<chrono::Utc>,
36    pub raw_value: f64,
37    #[serde(skip_serializing_if = "is_zero")]
38    pub replicate_index: i16,
39    #[serde(skip_serializing_if = "Option::is_none")]
40    pub sensor_id: Option<Uuid>,
41    #[serde(skip_serializing_if = "Option::is_none")]
42    pub calibration_id: Option<Uuid>,
43    #[serde(skip_serializing_if = "Option::is_none")]
44    pub deployment_id: Option<Uuid>,
45    /// Per-reading override ('continuous' | 'spot' | 'derived'). None resolves server-side from
46    /// the stream default, then the owning sensor's data_frequency.
47    #[serde(skip_serializing_if = "Option::is_none")]
48    pub measurement_type: Option<String>,
49}
50
51impl IngestReading {
52    /// A reading at replicate 0 with no sensor attribution; the server resolves the rest.
53    pub fn new(time: chrono::DateTime<chrono::Utc>, raw_value: f64) -> Self {
54        Self {
55            time,
56            raw_value,
57            replicate_index: 0,
58            sensor_id: None,
59            calibration_id: None,
60            deployment_id: None,
61            measurement_type: None,
62        }
63    }
64}
65
66fn is_zero(v: &i16) -> bool {
67    *v == 0
68}
69
70#[derive(Debug, Serialize)]
71pub struct IngestStatusEvent {
72    pub time: chrono::DateTime<chrono::Utc>,
73    pub value: String,
74}
75
76#[cfg(test)]
77mod tests {
78    use super::*;
79
80    #[test]
81    fn test_ingest_reading_serialization() {
82        let r = IngestReading::new(chrono::Utc::now(), 42.5);
83        let json = serde_json::to_value(&r).unwrap();
84        assert_eq!(json["raw_value"], 42.5);
85        assert!(json.get("replicate_index").is_none());
86        assert!(json.get("sensor_id").is_none());
87        assert!(json.get("measurement_type").is_none());
88    }
89
90    #[test]
91    fn test_register_stream_request() {
92        let req = RegisterStreamRequest {
93            source_system: "test_system".to_string(),
94            source_key: "source_1".to_string(),
95            source_name: Some("stream_a".to_string()),
96            source_path: None,
97            metadata: serde_json::json!({"device": "dev_001"}),
98            measurement_type: None,
99        };
100        let json = serde_json::to_value(&req).unwrap();
101        assert_eq!(json["source_system"], "test_system");
102        assert_eq!(json["metadata"]["device"], "dev_001");
103    }
104
105    #[test]
106    fn test_data_stream_deserialization() {
107        let json = serde_json::json!({
108            "id": "550e8400-e29b-41d4-a716-446655440000",
109            "source_system": "test_system",
110            "source_key": "source_1",
111            "source_name": "stream_a",
112            "source_path": null,
113            "metadata": {},
114            "site_parameter_id": null,
115            "is_active": true,
116            "last_data_time": null
117        });
118        let stream: DataStream = serde_json::from_value(json).unwrap();
119        assert_eq!(stream.source_system, "test_system");
120        assert!(stream.is_active);
121        assert!(stream.site_parameter_id.is_none());
122    }
123}
river_data_core/models/streams.rs

river_data_core/models/
streams.rs
