Working on experimental features

Tools/gui_client/src/models.rs

@@ -0,0 +1,226 @@
+use serde::{Deserialize, Serialize};
+use std::collections::VecDeque;
+use crossbeam_channel::Sender;
+use eframe::egui;
+use egui_plot::LineStyle;
+
+#[allow(dead_code)]
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct Signal {
+    #[serde(alias = "Name")]
+    pub name: String,
+    #[serde(alias = "ID")]
+    pub id: u32,
+    #[serde(rename = "type", alias = "Type")]
+    pub sig_type: String,
+    #[serde(default, alias = "Dimensions")]
+    pub dimensions: u8,
+    #[serde(default = "default_elements", alias = "Elements")]
+    pub elements: u32,
+    #[serde(default, alias = "IsState")]
+    pub is_state: bool,
+    #[serde(default)]
+    pub config: Option<serde_json::Value>,
+}
+
+pub fn default_elements() -> u32 { 1 }
+
+#[derive(Deserialize)]
+pub struct DiscoverResponse {
+    #[serde(rename = "Signals")]
+    pub signals: Vec<Signal>,
+}
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+#[serde(rename_all = "PascalCase")]
+pub struct TreeItem {
+    pub name: String,
+    pub class: String,
+    pub children: Option<Vec<TreeItem>>,
+    #[serde(rename = "Type")]
+    pub sig_type: Option<String>,
+    pub dimensions: Option<u8>,
+    pub elements: Option<u32>,
+    pub is_traceable: Option<bool>,
+    pub is_forcable: Option<bool>,
+}
+
+#[derive(Clone)]
+pub struct LogEntry {
+    pub time: String,
+    pub level: String,
+    pub message: String,
+}
+
+#[allow(dead_code)]
+pub struct TraceData {
+    pub values: VecDeque<[f64; 2]>,
+    pub last_value: f64,
+    pub recording_tx: Option<Sender<[f64; 2]>>,
+    pub recording_path: Option<String>,
+    pub is_monitored: bool,
+}
+
+#[allow(dead_code)]
+pub struct SignalMetadata {
+    pub names: Vec<String>,
+    pub sig_type: String,
+    pub dimensions: u8,
+    pub elements: u32,
+    pub is_state: bool,
+}
+
+#[derive(Clone)]
+pub struct ConnectionConfig {
+    pub ip: String,
+    pub tcp_port: String,
+    pub udp_port: String,
+    pub log_port: String,
+    pub version: u64,
+}
+
+#[allow(dead_code)]
+#[derive(Clone, Copy, PartialEq, Debug)]
+pub enum PlotType {
+    Normal,
+    LogicAnalyzer,
+}
+
+#[derive(Clone, Copy, PartialEq, Debug)]
+pub enum AcquisitionMode {
+    FreeRun,
+    Triggered,
+}
+
+#[allow(dead_code)]
+#[derive(Clone, Copy, PartialEq, Debug)]
+pub enum TriggerEdge {
+    Rising,
+    Falling,
+    Both,
+}
+
+#[derive(Clone, Copy, PartialEq, Debug)]
+pub enum TriggerType {
+    Single,
+    Continuous,
+}
+
+#[allow(dead_code)]
+#[derive(Clone, Copy, PartialEq, Debug)]
+pub enum MarkerType {
+    None,
+    Circle,
+    Square,
+}
+
+#[allow(dead_code)]
+impl MarkerType {
+    pub fn to_shape(&self) -> Option<egui_plot::MarkerShape> {
+        match self {
+            MarkerType::None => None,
+            MarkerType::Circle => Some(egui_plot::MarkerShape::Circle),
+            MarkerType::Square => Some(egui_plot::MarkerShape::Square),
+        }
+    }
+}
+
+#[allow(dead_code)]
+#[derive(Clone)]
+pub struct SignalPlotConfig {
+    pub source_name: String,
+    pub label: String,
+    pub unit: String,
+    pub color: egui::Color32,
+    pub line_style: LineStyle,
+    pub marker_type: MarkerType,
+    pub gain: f64,
+    pub offset: f64,
+}
+
+#[allow(dead_code)]
+pub struct PlotInstance {
+    pub id: String,
+    pub plot_type: PlotType,
+    pub signals: Vec<SignalPlotConfig>,
+    pub auto_bounds: bool,
+}
+
+#[allow(dead_code)]
+pub enum InternalEvent {
+    Log(LogEntry),
+    Discovery(Vec<Signal>),
+    Tree(TreeItem),
+    CommandResponse(String),
+    NodeInfo(String),
+    Connected,
+    Disconnected,
+    InternalLog(String),
+    TraceRequested(String, bool), // Name, IsMonitored
+    ClearTrace(String),
+    UdpStats(u64),
+    UdpDropped(u32),
+    RecordPathChosen(String, String), // SignalName, FilePath
+    RecordingError(String, String),   // SignalName, ErrorMessage
+    TelemMatched(u32),                // Signal ID
+    ServiceConfig { udp_port: String, log_port: String },
+    StateUpdate(String, String), // Path, StateName
+    Config(serde_json::Value),
+}
+
+#[allow(dead_code)]
+pub struct ForcingDialog {
+    pub signal_path: String,
+    pub value: String,
+}
+
+#[allow(dead_code)]
+pub struct MonitorDialog {
+    pub signal_path: String,
+    pub period_ms: String,
+}
+
+pub struct MessageDialog {
+    pub destination: String,
+    pub function: String,
+    pub payload: String,
+    pub expect_reply: bool,
+}
+
+#[allow(dead_code)]
+pub struct LogFilters {
+    pub show_debug: bool,
+    pub show_info: bool,
+    pub show_warning: bool,
+    pub show_error: bool,
+    pub paused: bool,
+    pub content_regex: String,
+}
+
+#[derive(Clone, Copy, PartialEq)]
+pub enum BottomTab {
+    Logs,
+    State,
+}
+
+#[derive(Clone, Copy, PartialEq)]
+pub enum MainTab {
+    Plots,
+    Config,
+}
+
+#[allow(dead_code)]
+pub struct ScopeSettings {
+    pub enabled: bool,
+    pub window_ms: f64,
+    pub mode: AcquisitionMode,
+    pub paused: bool,
+    pub trigger_type: TriggerType,
+    pub trigger_source: String,
+    pub trigger_edge: TriggerEdge,
+    pub trigger_threshold: f64,
+    pub pre_trigger_percent: f64,
+    pub trigger_active: bool,
+    pub last_trigger_time: f64,
+    pub is_armed: bool,
+}

Tools/gui_client/src/workers.rs

@@ -0,0 +1,498 @@
+use crate::models::*;
+use arrow::array::Float64Array;
+use arrow::datatypes::{DataType, Field, Schema};
+use arrow::record_batch::RecordBatch;
+use chrono::Local;
+use crossbeam_channel::{Receiver, Sender};
+use once_cell::sync::Lazy;
+use parquet::arrow::arrow_writer::ArrowWriter;
+use parquet::file::properties::WriterProperties;
+use socket2::{Domain, Protocol, Socket, Type};
+use std::collections::HashMap;
+use std::fs::File;
+use std::io::{BufRead, BufReader, Write};
+use std::net::{TcpStream, UdpSocket};
+use std::sync::{Arc, Mutex};
+use std::thread;
+
+pub static BASE_TELEM_TS: Lazy<Mutex<Option<u64>>> = Lazy::new(|| Mutex::new(None));
+
+pub fn tcp_command_worker(
+    shared_config: Arc<Mutex<ConnectionConfig>>,
+    rx_cmd: Receiver<String>,
+    tx_events: Sender<InternalEvent>,
+) {
+    let mut current_version = 0;
+    let mut current_addr = String::new();
+    loop {
+        {
+            let config = shared_config.lock().unwrap();
+            if config.version != current_version {
+                current_version = config.version;
+                current_addr = format!("{}:{}", config.ip, config.tcp_port);
+            }
+        }
+        if current_addr.is_empty() || current_addr.starts_with(":") {
+            thread::sleep(std::time::Duration::from_secs(1));
+            continue;
+        }
+        if let Ok(mut stream) = TcpStream::connect(&current_addr) {
+            let _ = stream.set_nodelay(true);
+            let mut reader = BufReader::new(stream.try_clone().unwrap());
+            let _ = tx_events.send(InternalEvent::Connected);
+            let stop_flag = Arc::new(Mutex::new(false));
+            let stop_flag_reader = stop_flag.clone();
+            let tx_events_inner = tx_events.clone();
+            thread::spawn(move || {
+                let mut line = String::new();
+                let mut json_acc = String::new();
+                let mut in_json = false;
+                while reader.read_line(&mut line).is_ok() {
+                    if *stop_flag_reader.lock().unwrap() {
+                        break;
+                    }
+                    let trimmed = line.trim();
+                    if trimmed.is_empty() {
+                        line.clear();
+                        continue;
+                    }
+
+                    if !in_json && trimmed.starts_with("{") {
+                        in_json = true;
+                        json_acc.clear();
+                    }
+
+                    if in_json {
+                        json_acc.push_str(trimmed);
+                        if trimmed.contains("OK DISCOVER") {
+                            in_json = false;
+                            let json_clean =
+                                json_acc.split("OK DISCOVER").next().unwrap_or("").trim();
+                            match serde_json::from_str::<DiscoverResponse>(json_clean) {
+                                Ok(resp) => {
+                                    let _ = tx_events_inner
+                                        .send(InternalEvent::Discovery(resp.signals));
+                                }
+                                Err(e) => {
+                                    let _ =
+                                        tx_events_inner.send(InternalEvent::InternalLog(format!(
+                                            "Discovery JSON Error: {} | Payload: {}",
+                                            e, json_clean
+                                        )));
+                                }
+                            }
+                            json_acc.clear();
+                        } else if trimmed.contains("OK TREE") {
+                            in_json = false;
+                            let json_clean = json_acc.split("OK TREE").next().unwrap_or("").trim();
+                            match serde_json::from_str::<TreeItem>(json_clean) {
+                                Ok(resp) => {
+                                    let _ = tx_events_inner.send(InternalEvent::Tree(resp));
+                                }
+                                Err(e) => {
+                                    let _ = tx_events_inner.send(InternalEvent::InternalLog(
+                                        format!("Tree JSON Error: {}", e),
+                                    ));
+                                }
+                            }
+                            json_acc.clear();
+                        } else if trimmed.contains("OK INFO") {
+                            in_json = false;
+                            let json_clean = json_acc.split("OK INFO").next().unwrap_or("").trim();
+                            let _ = tx_events_inner
+                                .send(InternalEvent::NodeInfo(json_clean.to_string()));
+                            json_acc.clear();
+                        } else if trimmed.contains("OK CONFIG") {
+                            in_json = false;
+                            let json_clean = json_acc.split("OK CONFIG").next().unwrap_or("").trim();
+                            match serde_json::from_str::<serde_json::Value>(json_clean) {
+                                Ok(val) => {
+                                    let _ = tx_events_inner.send(InternalEvent::Config(val));
+                                }
+                                Err(e) => {
+                                    let _ = tx_events_inner.send(InternalEvent::InternalLog(
+                                        format!("Config JSON Error: {} | Payload sample: {}", e, &json_clean[..json_clean.len().min(100)]),
+                                    ));
+                                }
+                            }
+                            json_acc.clear();
+                        }
+                    } else {
+                        if trimmed.starts_with("OK SERVICE_INFO") {
+                            let parts: Vec<&str> = trimmed.split_whitespace().collect();
+                            let mut udp = String::new();
+                            let mut log = String::new();
+                            for p in parts {
+                                if p.starts_with("UDP_STREAM:") {
+                                    udp = p.split(':').nth(1).unwrap_or("").to_string();
+                                }
+                                if p.starts_with("TCP_LOG:") {
+                                    log = p.split(':').nth(1).unwrap_or("").to_string();
+                                }
+                            }
+                            if !udp.is_empty() || !log.is_empty() {
+                                let _ = tx_events_inner.send(InternalEvent::ServiceConfig {
+                                    udp_port: udp,
+                                    log_port: log,
+                                });
+                            }
+                        }
+                        let _ = tx_events_inner
+                            .send(InternalEvent::CommandResponse(trimmed.to_string()));
+                    }
+                    line.clear();
+                }
+            });
+            while let Ok(cmd) = rx_cmd.recv() {
+                {
+                    let config = shared_config.lock().unwrap();
+                    if config.version != current_version {
+                        *stop_flag.lock().unwrap() = true;
+                        let _ = tx_events.send(InternalEvent::Disconnected);
+                        break;
+                    }
+                }
+                if stream.write_all(format!("{}
+", cmd).as_bytes()).is_err() {
+                    let _ = tx_events.send(InternalEvent::Disconnected);
+                    break;
+                }
+                thread::sleep(std::time::Duration::from_millis(100));
+            }
+            let _ = tx_events.send(InternalEvent::Disconnected);
+        }
+        thread::sleep(std::time::Duration::from_secs(2));
+    }
+}
+
+pub fn tcp_log_worker(shared_config: Arc<Mutex<ConnectionConfig>>, tx_events: Sender<InternalEvent>) {
+    let mut current_version = 0;
+    let mut current_addr = String::new();
+    loop {
+        {
+            let config = shared_config.lock().unwrap();
+            if config.version != current_version {
+                current_version = config.version;
+                current_addr = format!("{}:{}", config.ip, config.log_port);
+            }
+        }
+        if current_addr.is_empty() || current_addr.starts_with(":") {
+            thread::sleep(std::time::Duration::from_secs(1));
+            continue;
+        }
+        if let Ok(stream) = TcpStream::connect(&current_addr) {
+            let mut reader = BufReader::new(stream);
+            let mut line = String::new();
+            while reader.read_line(&mut line).is_ok() {
+                if shared_config.lock().unwrap().version != current_version {
+                    break;
+                }
+                let trimmed = line.trim();
+                if trimmed.starts_with("LOG ") {
+                    let parts: Vec<&str> = trimmed[4..].splitn(2, ' ').collect();
+                    if parts.len() == 2 {
+                        let _ = tx_events.send(InternalEvent::Log(LogEntry {
+                            time: Local::now().format("%H:%M:%S%.3f").to_string(),
+                            level: parts[0].to_string(),
+                            message: parts[1].to_string(),
+                        }));
+                    }
+                }
+                line.clear();
+            }
+        }
+        thread::sleep(std::time::Duration::from_secs(2));
+    }
+}
+
+#[allow(dead_code)]
+pub fn recording_worker(
+    rx: Receiver<[f64; 2]>,
+    path: String,
+    signal_name: String,
+    tx_events: Sender<InternalEvent>,
+) {
+    let file = match File::create(&path) {
+        Ok(f) => f,
+        Err(e) => {
+            let _ = tx_events.send(InternalEvent::RecordingError(
+                signal_name,
+                format!("File Error: {}", e),
+            ));
+            return;
+        }
+    };
+    let schema = Arc::new(Schema::new(vec![
+        Field::new("timestamp", DataType::Float64, false),
+        Field::new("value", DataType::Float64, false),
+    ]));
+    let mut writer = match ArrowWriter::try_new(
+        file,
+        schema.clone(),
+        Some(WriterProperties::builder().build()),
+    ) {
+        Ok(w) => w,
+        Err(e) => {
+            let _ = tx_events.send(InternalEvent::RecordingError(
+                signal_name,
+                format!("Parquet Error: {}", e),
+            ));
+            return;
+        }
+    };
+    let (mut t_acc, mut v_acc) = (Vec::with_capacity(1000), Vec::with_capacity(1000));
+    while let Ok([t, v]) = rx.recv() {
+        t_acc.push(t);
+        v_acc.push(v);
+        if t_acc.len() >= 1000 {
+            let batch = RecordBatch::try_new(
+                schema.clone(),
+                vec![
+                    Arc::new(Float64Array::from(t_acc.clone())),
+                    Arc::new(Float64Array::from(v_acc.clone())),
+                ],
+            )
+            .unwrap();
+            let _ = writer.write(&batch);
+            t_acc.clear();
+            v_acc.clear();
+        }
+    }
+    if !t_acc.is_empty() {
+        let batch = RecordBatch::try_new(
+            schema.clone(),
+            vec![
+                Arc::new(Float64Array::from(t_acc)),
+                Arc::new(Float64Array::from(v_acc)),
+            ],
+        )
+        .unwrap();
+        let _ = writer.write(&batch);
+    }
+    let _ = writer.close();
+}
+
+pub fn udp_worker(
+    shared_config: Arc<Mutex<ConnectionConfig>>,
+    id_to_meta: Arc<Mutex<HashMap<u32, SignalMetadata>>>,
+    traced_data: Arc<Mutex<HashMap<String, TraceData>>>,
+    tx_events: Sender<InternalEvent>,
+) {
+    let mut current_version = 0;
+    let mut socket: Option<UdpSocket> = None;
+    let mut last_seq: Option<u32> = None;
+    let mut last_warning_time = std::time::Instant::now();
+
+    loop {
+        let (ver, port) = {
+            let config = shared_config.lock().unwrap();
+            (config.version, config.udp_port.clone())
+        };
+        if ver != current_version || socket.is_none() {
+            current_version = ver;
+            {
+                let mut base = BASE_TELEM_TS.lock().unwrap();
+                *base = None;
+            }
+            if port.is_empty() {
+                socket = None;
+                continue;
+            }
+            let port_num: u16 = port.parse().unwrap_or(8081);
+            let s = Socket::new(Domain::IPV4, Type::DGRAM, Some(Protocol::UDP)).ok();
+            let mut bound = false;
+            if let Some(sock) = s {
+                let _ = sock.set_reuse_address(true);
+                #[cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos")))]
+                let _ = sock.set_reuse_port(true);
+                let _ = sock.set_recv_buffer_size(10 * 1024 * 1024);
+                let addr = format!("0.0.0.0:{}", port_num)
+                    .parse::<std::net::SocketAddr>()
+                    .unwrap();
+                if sock.bind(&addr.into()).is_ok() {
+                    socket = Some(sock.into());
+                    bound = true;
+                }
+            }
+            if !bound {
+                thread::sleep(std::time::Duration::from_secs(5));
+                continue;
+            }
+            let _ = socket
+                .as_ref()
+                .unwrap()
+                .set_read_timeout(Some(std::time::Duration::from_millis(500)));
+            last_seq = None;
+        }
+        let s = if let Some(sock) = socket.as_ref() {
+            sock
+        } else {
+            thread::sleep(std::time::Duration::from_secs(1));
+            continue;
+        };
+        let mut buf = [0u8; 4096];
+        let mut total_packets = 0u64;
+        loop {
+            if shared_config.lock().unwrap().version != current_version {
+                break;
+            }
+            if let Ok(n) = s.recv(&mut buf) {
+                total_packets += 1;
+                if (total_packets % 500) == 0 {
+                    let _ = tx_events.send(InternalEvent::UdpStats(total_packets));
+                }
+                if n < 20 {
+                    continue;
+                }
+                if u32::from_le_bytes(buf[0..4].try_into().unwrap()) != 0xDA7A57AD {
+                    continue;
+                }
+                let seq = u32::from_le_bytes(buf[4..8].try_into().unwrap());
+                if let Some(last) = last_seq {
+                    if seq != last + 1 && seq > last {
+                        let _ = tx_events.send(InternalEvent::UdpDropped(seq - last - 1));
+                    }
+                }
+                last_seq = Some(seq);
+                let count = u32::from_le_bytes(buf[16..20].try_into().unwrap());
+
+                let mut offset = 20;
+                let mut local_updates: HashMap<String, Vec<[f64; 2]>> = HashMap::new();
+                let mut last_values: HashMap<String, f64> = HashMap::new();
+                let metas = id_to_meta.lock().unwrap();
+
+                if metas.is_empty() && count > 0 && last_warning_time.elapsed().as_secs() > 5 {
+                    let _ = tx_events.send(InternalEvent::InternalLog(
+                        "UDP received but Metadata empty. Still discovering?".to_string(),
+                    ));
+                    last_warning_time = std::time::Instant::now();
+                }
+
+                for _ in 0..count {
+                    if offset + 16 > n {
+                        break;
+                    }
+                    let id = u32::from_le_bytes(buf[offset..offset + 4].try_into().unwrap());
+                    let ts_raw =
+                        u64::from_le_bytes(buf[offset + 4..offset + 12].try_into().unwrap());
+                    let size =
+                        u32::from_le_bytes(buf[offset + 12..offset + 16].try_into().unwrap());
+                    offset += 16;
+
+                    if offset + size as usize > n {
+                        break;
+                    }
+                    let data_slice = &buf[offset..offset + size as usize];
+
+                    let mut base_ts_guard = BASE_TELEM_TS.lock().unwrap();
+                    if base_ts_guard.is_none() {
+                        *base_ts_guard = Some(ts_raw);
+                    }
+
+                    let base = base_ts_guard.unwrap();
+                    let ts_s = if ts_raw >= base {
+                        (ts_raw - base) as f64 / 1000000.0
+                    } else {
+                        0.0
+                    };
+                    drop(base_ts_guard);
+
+                    if let Some(meta) = metas.get(&id) {
+                        let _ = tx_events.send(InternalEvent::TelemMatched(id));
+
+                        if meta.is_state {
+                            let state_name = String::from_utf8_lossy(data_slice)
+                                .trim_matches(char::from(0))
+                                .to_string();
+                            for name in &meta.names {
+                                let _ = tx_events.send(InternalEvent::StateUpdate(name.clone(), state_name.clone()));
+                            }
+                        } else {
+                            let t = meta.sig_type.as_str();
+                            let type_size = if meta.elements > 0 { size / meta.elements } else { size };
+                            
+                            for i in 0..meta.elements {
+                                let elem_offset = (i * type_size) as usize;
+                                if elem_offset + type_size as usize > data_slice.len() { break; }
+                                let elem_data = &data_slice[elem_offset..elem_offset + type_size as usize];
+
+                                let val = match type_size {
+                                    1 => {
+                                        if t.contains('u') {
+                                            elem_data[0] as f64
+                                        } else {
+                                            (elem_data[0] as i8) as f64
+                                        }
+                                    }
+                                    2 => {
+                                        let b = elem_data[0..2].try_into().unwrap();
+                                        if t.contains('u') {
+                                            u16::from_le_bytes(b) as f64
+                                        } else {
+                                            i16::from_le_bytes(b) as f64
+                                        }
+                                    }
+                                    4 => {
+                                        let b = elem_data[0..4].try_into().unwrap();
+                                        if t.contains("float") {
+                                            f32::from_le_bytes(b) as f64
+                                        } else if t.contains('u') {
+                                            u32::from_le_bytes(b) as f64
+                                        } else {
+                                            i32::from_le_bytes(b) as f64
+                                        }
+                                    }
+                                    8 => {
+                                        let b = elem_data[0..8].try_into().unwrap();
+                                        if t.contains("float") {
+                                            f64::from_le_bytes(b)
+                                        } else if t.contains('u') {
+                                            u64::from_le_bytes(b) as f64
+                                        } else {
+                                            i64::from_le_bytes(b) as f64
+                                        }
+                                    }
+                                    _ => 0.0,
+                                };
+
+                                for name in &meta.names {
+                                    let target_name = if meta.elements > 1 {
+                                        format!("{}[{}]", name, i)
+                                    } else {
+                                        name.clone()
+                                    };
+                                    local_updates
+                                        .entry(target_name.clone())
+                                        .or_default()
+                                        .push([ts_s, val]);
+                                    last_values.insert(target_name, val);
+                                }
+                            }
+                        }
+                    }
+                    offset += size as usize;
+                }
+                drop(metas);
+                if !local_updates.is_empty() {
+                    let mut data_map = traced_data.lock().unwrap();
+                    for (name, new_points) in local_updates {
+                        if let Some(entry) = data_map.get_mut(&name) {
+                            for point in new_points {
+                                entry.values.push_back(point);
+                                if let Some(tx) = &entry.recording_tx {
+                                    let _ = tx.send(point);
+                                }
+                            }
+                            if let Some(lv) = last_values.get(&name) {
+                                entry.last_value = *lv;
+                            }
+                            while entry.values.len() > 100000 {
+                                entry.values.pop_front();
+                            }
+                        }
+                    }
+                }
+            }
+        }
+    }
+}