use anyhow::Result; use rmon_common::signal::{SignalId, SignalInfo, Sample, SourceType}; use tokio::sync::mpsc; use async_trait::async_trait; use crate::sources::datasource::DataSource; use crate::config::SourceConfig; use std::time::Duration; use tokio::time::sleep; use std::fs::File; use std::io::{BufRead, BufReader, Seek, SeekFrom}; use chrono::{DateTime, Utc, NaiveDateTime}; pub struct CsvSource { name: String, signals: Vec, config: SourceConfig, } impl CsvSource { pub fn new(config: SourceConfig) -> Self { let signals = config.signals.iter().map(|s| { SignalInfo { id: format!("csv://{}/{}", config.name, s.id), label: s.label.clone(), unit: s.unit.clone(), scale: s.scale, offset: s.offset, sampling_period: None, source_type: SourceType::Csv, path: s.path.clone(), } }).collect(); Self { name: config.name.clone(), signals, config, } } fn parse_timestamp(&self, value: &str) -> i64 { let value = value.trim(); match self.config.timestamp.format.as_str() { "nanoseconds_since_epoch" => value.parse::().unwrap_or(0), "milliseconds_since_epoch" => value.parse::().unwrap_or(0) * 1_000_000, "seconds_since_epoch" => value.parse::().map(|v| (v * 1e9) as i64).unwrap_or(0), "none" => Utc::now().timestamp_nanos_opt().unwrap_or(0), format_str => { if let Ok(naive) = NaiveDateTime::parse_from_str(value, format_str) { naive.and_utc().timestamp_nanos_opt().unwrap_or(0) } else if let Ok(dt) = DateTime::parse_from_rfc3339(value) { dt.timestamp_nanos_opt().unwrap_or(0) } else { Utc::now().timestamp_nanos_opt().unwrap_or(0) } } } } } #[async_trait] impl DataSource for CsvSource { fn name(&self) -> &str { &self.name } fn signals(&self) -> Vec { self.signals.clone() } async fn run(self: Box, tx: mpsc::Sender<(SignalId, Sample)>) -> Result<()> { let path = self.config.path.clone().ok_or_else(|| anyhow::anyhow!("CSV path missing"))?; let poll_interval = if let Some(ref interval_str) = self.config.poll_interval { if interval_str.ends_with("ms") { Duration::from_millis(interval_str.trim_end_matches("ms").parse().unwrap_or(100)) } else { Duration::from_secs(interval_str.trim_end_matches("s").parse().unwrap_or(1)) } } else { Duration::from_millis(500) }; let mut pos = 0u64; loop { if let Ok(mut file) = File::open(&path) { let len = file.metadata()?.len(); if len < pos { tracing::info!("CSV file truncated or rotated, resetting position"); pos = 0; } if len > pos { file.seek(SeekFrom::Start(pos))?; let mut reader = BufReader::new(file); let mut line = String::new(); while reader.read_line(&mut line)? > 0 { let bytes_read = line.len() as u64; let trimmed = line.trim(); if !trimmed.is_empty() { // If we are at start of file, skip the header if pos == 0 { pos += bytes_read; line.clear(); continue; } let parts: Vec<&str> = trimmed.split(',').collect(); // 1. Get timestamp let timestamp = if let Some(col_idx) = self.config.timestamp.column.as_ref().and_then(|c| c.parse::().ok()) { if let Some(val) = parts.get(col_idx) { self.parse_timestamp(val) } else { Utc::now().timestamp_nanos_opt().unwrap_or(0) } } else { Utc::now().timestamp_nanos_opt().unwrap_or(0) }; // 2. Map signals for signal_config in &self.config.signals { if let Some(col_idx) = signal_config.column.as_ref().and_then(|c| c.parse::().ok()) { if let Some(raw_str) = parts.get(col_idx) { if let Ok(raw_val) = raw_str.trim().parse::() { let value = raw_val * signal_config.scale + signal_config.offset; let id = format!("csv://{}/{}", self.name, signal_config.id); let _ = tx.send((id, Sample { timestamp, value })).await; } } } } } pos += bytes_read; line.clear(); } } } sleep(poll_interval).await; } } } #[cfg(test)] mod tests { use super::*; use crate::config::{SourceConfig, TimestampConfig, SignalSourceConfig}; use std::io::Write; use tempfile::tempdir; use std::path::PathBuf; #[test] fn test_csv_metadata() { let config = SourceConfig { source_type: "csv".to_string(), name: "tc".to_string(), bind: None, path: Some(PathBuf::from("test.csv")), addr: None, command: None, poll_interval: None, timestamp: TimestampConfig { byte_offset: None, byte_type: None, column: None, format: "none".to_string(), regex: None }, signals: vec![SignalSourceConfig { id: "s1".to_string(), label: "L1".to_string(), unit: "V".to_string(), scale: 1.0, offset: 0.0, byte_offset: None, byte_type: None, column: None, pv: None, regex: None, path: vec![], }], }; let source = CsvSource::new(config); assert_eq!(source.name(), "tc"); assert_eq!(source.signals()[0].id, "csv://tc/s1"); } #[tokio::test] async fn test_csv_batch_parsing() { let dir = tempdir().unwrap(); let file_path = dir.path().join("test_batch.csv"); let mut file = std::fs::File::create(&file_path).unwrap(); // Write header and two lines immediately writeln!(file, "time,val1").unwrap(); writeln!(file, "100,10.0").unwrap(); writeln!(file, "200,20.0").unwrap(); file.sync_all().unwrap(); let config = SourceConfig { source_type: "csv".to_string(), name: "test_batch".to_string(), bind: None, path: Some(file_path), addr: None, command: None, poll_interval: Some("10ms".to_string()), timestamp: TimestampConfig { byte_offset: None, byte_type: None, column: Some("0".to_string()), format: "nanoseconds_since_epoch".to_string(), regex: None }, signals: vec![ SignalSourceConfig { id: "s1".to_string(), label: "L1".to_string(), unit: "V".to_string(), scale: 1.0, offset: 0.0, byte_offset: None, byte_type: None, column: Some("1".to_string()), pv: None, regex: None, path: vec![], } ], }; let source = Box::new(CsvSource::new(config)); let (tx, mut rx) = mpsc::channel(10); tokio::spawn(async move { let _ = source.run(tx).await; }); // Should receive BOTH lines from initial load let msg1 = tokio::time::timeout(Duration::from_secs(1), rx.recv()).await.unwrap().unwrap(); assert_eq!(msg1.1.value, 10.0); let msg2 = tokio::time::timeout(Duration::from_secs(1), rx.recv()).await.unwrap().unwrap(); assert_eq!(msg2.1.value, 20.0); } }