From edec8113fb2b273316d03ad4c16572dd93f7957c Mon Sep 17 00:00:00 2001 From: FallenSigh Date: Thu, 11 Jun 2026 15:07:02 +0800 Subject: [PATCH] Add Lua pipeline support --- crates/xserial-client/src/config.rs | 59 +++- crates/xserial-client/src/lua/codec.rs | 404 +++++++++++++++++++++++ crates/xserial-client/src/lua/mod.rs | 1 + crates/xserial-client/src/session.rs | 37 ++- crates/xserial-client/tests/lua_tests.rs | 107 ++++++ crates/xserial-gui/src/panels/config.rs | 66 +++- tests/lua_line_framer.lua | 43 +++ tests/lua_text_decoder.lua | 12 + tools/test_plot.py | 78 +++-- 9 files changed, 769 insertions(+), 38 deletions(-) create mode 100644 crates/xserial-client/src/lua/codec.rs create mode 100644 tests/lua_line_framer.lua create mode 100644 tests/lua_text_decoder.lua diff --git a/crates/xserial-client/src/config.rs b/crates/xserial-client/src/config.rs index 3872156..25ca50f 100644 --- a/crates/xserial-client/src/config.rs +++ b/crates/xserial-client/src/config.rs @@ -16,6 +16,9 @@ use xserial_core::protocol::{ }; use xserial_core::transport::TransportConfig; +use crate::error::Result; +use crate::lua::codec::{LuaDecoder, LuaFramer}; + #[derive(Debug, Clone, Serialize, Deserialize)] pub enum FramerConfig { Line { @@ -49,6 +52,9 @@ pub enum FramerConfig { #[serde(default = "default_max_frame")] max_plot_frame: usize, }, + Lua { + script_path: String, + }, } fn default_true() -> bool { @@ -68,15 +74,15 @@ fn default_max_frame() -> usize { } impl FramerConfig { - pub fn build(self) -> Box { - match self { + pub fn build(self) -> Result> { + Ok(match self { FramerConfig::Line { strip_cr, max_line_len, } => Box::new(LineFramer::new(LineConfig { strip_cr, max_line_len, - })), + })) as Box, FramerConfig::Fixed { frame_len } => Box::new(FixedLengthFramer::new(frame_len)), FramerConfig::Length { len_bytes, @@ -99,7 +105,10 @@ impl FramerConfig { max_line_len, max_plot_frame, })), - } + FramerConfig::Lua { script_path } => { + Box::new(LuaFramer::from_script_path(script_path)?) + } + }) } } @@ -132,6 +141,9 @@ pub enum DecoderConfig { #[serde(default)] encoding: TextEncoding, }, + Lua { + script_path: String, + }, } fn default_sep() -> String { @@ -142,9 +154,11 @@ fn default_one() -> usize { } impl DecoderConfig { - pub fn build(self) -> Box { - match self { - DecoderConfig::Text { encoding } => Box::new(TextDecoder::new(encoding)), + pub fn build(self) -> Result> { + Ok(match self { + DecoderConfig::Text { encoding } => { + Box::new(TextDecoder::new(encoding)) as Box + } DecoderConfig::Hex { uppercase, separator, @@ -170,7 +184,10 @@ impl DecoderConfig { DecoderConfig::MixedTextPlot { encoding } => { Box::new(MixedTextPlotDecoder::new(MixedDecoderConfig { encoding })) } - } + DecoderConfig::Lua { script_path } => { + Box::new(LuaDecoder::from_script_path(script_path)?) + } + }) } } @@ -285,6 +302,16 @@ mod tests { } } + #[test] + fn framer_lua_serde() { + let cfg: FramerConfig = + serde_json::from_str(r#"{"Lua":{"script_path":"/tmp/framer.lua"}}"#).unwrap(); + match cfg { + FramerConfig::Lua { script_path } => assert_eq!(script_path, "/tmp/framer.lua"), + _ => panic!("expected Lua"), + } + } + #[test] fn decoder_text_serde() { let cfg: DecoderConfig = serde_json::from_str(r#"{"Text":{"encoding":"Latin1"}}"#).unwrap(); @@ -345,6 +372,16 @@ mod tests { } } + #[test] + fn decoder_lua_serde() { + let cfg: DecoderConfig = + serde_json::from_str(r#"{"Lua":{"script_path":"/tmp/decoder.lua"}}"#).unwrap(); + match cfg { + DecoderConfig::Lua { script_path } => assert_eq!(script_path, "/tmp/decoder.lua"), + _ => panic!("expected Lua"), + } + } + #[test] fn session_config_full_roundtrip() { let json = r#"{ @@ -388,7 +425,7 @@ mod tests { max_plot_frame: 1024, }, ] { - let f = cfg.build(); + let f = cfg.build().unwrap(); assert_eq!(f.pending_len(), 0); } } @@ -400,6 +437,7 @@ mod tests { encoding: TextEncoding::Utf8 } .build() + .unwrap() .name(), "Text" ); @@ -411,6 +449,7 @@ mod tests { endian: Endian::Big } .build() + .unwrap() .name(), "Hex" ); @@ -422,6 +461,7 @@ mod tests { format: PlotFormat::XY } .build() + .unwrap() .name(), "Plot" ); @@ -430,6 +470,7 @@ mod tests { encoding: TextEncoding::Utf8 } .build() + .unwrap() .name(), "MixedTextPlot" ); diff --git a/crates/xserial-client/src/lua/codec.rs b/crates/xserial-client/src/lua/codec.rs new file mode 100644 index 0000000..dc7e1c4 --- /dev/null +++ b/crates/xserial-client/src/lua/codec.rs @@ -0,0 +1,404 @@ +use std::fs; +use std::sync::Mutex; + +use mlua::{Function, Lua, RegistryKey, Table, Value}; +use tracing::warn; +use xserial_core::frame::Framer; +use xserial_core::protocol::plot::{PlotFormat, PlotFrame, SampleType}; +use xserial_core::protocol::{DecodedData, ProtocolDecoder}; + +use crate::error::{Error, Result}; + +pub struct LuaFramer { + lua: Lua, + feed: RegistryKey, + flush: RegistryKey, + reset: RegistryKey, + pending_len: RegistryKey, + script_path: String, +} + +impl LuaFramer { + pub fn from_script_path(script_path: impl Into) -> Result { + let script_path = script_path.into(); + let lua = Lua::new(); + let table = load_script_table(&lua, &script_path)?; + let feed = registry_function(&lua, &table, "feed", &script_path)?; + let flush = registry_function(&lua, &table, "flush", &script_path)?; + let reset = registry_function(&lua, &table, "reset", &script_path)?; + let pending_len = registry_function(&lua, &table, "pending_len", &script_path)?; + + Ok(Self { + lua, + feed, + flush, + reset, + pending_len, + script_path, + }) + } +} + +impl Framer for LuaFramer { + fn feed(&mut self, data: &[u8]) -> Vec> { + let result = (|| -> mlua::Result>> { + let function: Function = self.lua.registry_value(&self.feed)?; + let bytes = self.lua.create_string(data)?; + frames_from_value(function.call::(bytes)?) + })(); + + match result { + Ok(frames) => frames, + Err(err) => { + warn!(script = %self.script_path, error = %err, "lua framer feed failed"); + Vec::new() + } + } + } + + fn flush(&mut self) -> Option> { + let result = (|| -> mlua::Result>> { + let function: Function = self.lua.registry_value(&self.flush)?; + optional_bytes_from_value(function.call::(())?) + })(); + + match result { + Ok(frame) => frame, + Err(err) => { + warn!(script = %self.script_path, error = %err, "lua framer flush failed"); + None + } + } + } + + fn reset(&mut self) { + let result = (|| -> mlua::Result<()> { + let function: Function = self.lua.registry_value(&self.reset)?; + function.call::<()>(()) + })(); + + if let Err(err) = result { + warn!(script = %self.script_path, error = %err, "lua framer reset failed"); + } + } + + fn pending_len(&self) -> usize { + let result = (|| -> mlua::Result { + let function: Function = self.lua.registry_value(&self.pending_len)?; + function.call::(()) + })(); + + match result { + Ok(len) => len, + Err(err) => { + warn!(script = %self.script_path, error = %err, "lua framer pending_len failed"); + 0 + } + } + } +} + +pub struct LuaDecoder { + lua: Lua, + decode: RegistryKey, + script_path: String, + call_lock: Mutex<()>, +} + +impl LuaDecoder { + pub fn from_script_path(script_path: impl Into) -> Result { + let script_path = script_path.into(); + let lua = Lua::new(); + let table = load_script_table(&lua, &script_path)?; + let decode = registry_function(&lua, &table, "decode", &script_path)?; + + Ok(Self { + lua, + decode, + script_path, + call_lock: Mutex::new(()), + }) + } +} + +impl ProtocolDecoder for LuaDecoder { + fn name(&self) -> &str { + "Lua" + } + + fn decode(&self, frame: &[u8]) -> Option { + let _guard = self + .call_lock + .lock() + .expect("lua decoder call mutex poisoned"); + let result = (|| -> mlua::Result> { + let function: Function = self.lua.registry_value(&self.decode)?; + let bytes = self.lua.create_string(frame)?; + decoded_from_value(function.call::(bytes)?, frame) + })(); + + match result { + Ok(decoded) => decoded, + Err(err) => { + warn!(script = %self.script_path, error = %err, "lua decoder decode failed"); + None + } + } + } +} + +fn load_script_table(lua: &Lua, script_path: &str) -> Result { + let script = fs::read_to_string(script_path)?; + match lua.load(&script).eval::()? { + Value::Table(table) => Ok(table), + other => Err(Error::Other(format!( + "lua script '{script_path}' must return a table, got {}", + other.type_name() + ))), + } +} + +fn registry_function( + lua: &Lua, + table: &Table, + name: &str, + script_path: &str, +) -> Result { + let value: Value = table.get(name)?; + match value { + Value::Function(function) => Ok(lua.create_registry_value(function)?), + other => Err(Error::Other(format!( + "lua script '{script_path}' field '{name}' must be a function, got {}", + other.type_name() + ))), + } +} + +fn frames_from_value(value: Value) -> mlua::Result>> { + match value { + Value::Nil => Ok(Vec::new()), + Value::String(bytes) => Ok(vec![bytes.as_bytes().to_vec()]), + Value::Table(table) => { + let mut frames = Vec::new(); + for value in table.sequence_values::() { + frames.push(value?.as_bytes().to_vec()); + } + Ok(frames) + } + other => Err(mlua::Error::RuntimeError(format!( + "framer feed must return nil, string, or table of strings, got {}", + other.type_name() + ))), + } +} + +fn optional_bytes_from_value(value: Value) -> mlua::Result>> { + match value { + Value::Nil => Ok(None), + Value::String(bytes) => Ok(Some(bytes.as_bytes().to_vec())), + other => Err(mlua::Error::RuntimeError(format!( + "framer flush must return nil or string, got {}", + other.type_name() + ))), + } +} + +fn decoded_from_value(value: Value, raw: &[u8]) -> mlua::Result> { + match value { + Value::Nil => Ok(None), + Value::String(text) => Ok(Some(DecodedData::Text(text.to_str()?.to_string()))), + Value::Table(table) => decoded_from_table(table, raw).map(Some), + other => Err(mlua::Error::RuntimeError(format!( + "decoder decode must return nil, string, or table, got {}", + other.type_name() + ))), + } +} + +fn decoded_from_table(table: Table, raw: &[u8]) -> mlua::Result { + let kind: String = table.get("kind")?; + match kind.as_str() { + "text" => Ok(DecodedData::Text(table.get("data")?)), + "hex" => Ok(DecodedData::Hex(table.get("data")?)), + "binary" => { + let bytes: mlua::String = table.get("data")?; + Ok(DecodedData::Binary(bytes.as_bytes().to_vec())) + } + "plot" => Ok(DecodedData::Plot(plot_from_table(table, raw)?)), + other => Err(mlua::Error::RuntimeError(format!( + "unknown lua decoder kind '{other}'" + ))), + } +} + +fn plot_from_table(table: Table, raw: &[u8]) -> mlua::Result { + let channels_table: Table = table.get("channels")?; + let mut channels = Vec::new(); + for channel in channels_table.sequence_values::
() { + let channel = channel?; + let mut samples = Vec::new(); + for sample in channel.sequence_values::() { + samples.push(sample?); + } + channels.push(samples); + } + + Ok(PlotFrame { + channels, + raw: raw.to_vec(), + sample_type: match table.get::>("sample_type")?.as_deref() { + Some(value) => parse_sample_type(value)?, + None => SampleType::F64, + }, + format: match table.get::>("format")?.as_deref() { + Some(value) => parse_plot_format(value)?, + None => PlotFormat::Interleaved, + }, + }) +} + +fn parse_sample_type(value: &str) -> mlua::Result { + match value { + "i8" | "I8" => Ok(SampleType::I8), + "u8" | "U8" => Ok(SampleType::U8), + "i16" | "I16" => Ok(SampleType::I16), + "u16" | "U16" => Ok(SampleType::U16), + "i32" | "I32" => Ok(SampleType::I32), + "u32" | "U32" => Ok(SampleType::U32), + "i64" | "I64" => Ok(SampleType::I64), + "u64" | "U64" => Ok(SampleType::U64), + "f32" | "F32" => Ok(SampleType::F32), + "f64" | "F64" => Ok(SampleType::F64), + other => Err(mlua::Error::RuntimeError(format!( + "unknown plot sample_type '{other}'" + ))), + } +} + +fn parse_plot_format(value: &str) -> mlua::Result { + match value { + "Interleaved" | "interleaved" => Ok(PlotFormat::Interleaved), + "Block" | "block" => Ok(PlotFormat::Block), + "XY" | "xy" => Ok(PlotFormat::XY), + other => Err(mlua::Error::RuntimeError(format!( + "unknown plot format '{other}'" + ))), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::path::PathBuf; + use std::time::{SystemTime, UNIX_EPOCH}; + + fn write_script(name: &str, script: &str) -> PathBuf { + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let path = std::env::temp_dir().join(format!("xserial_{name}_{nonce}.lua")); + fs::write(&path, script).unwrap(); + path + } + + #[test] + fn lua_framer_splits_lines_across_chunks() { + let path = write_script( + "framer", + r#" + local buffer = "" + return { + feed = function(bytes) + buffer = buffer .. bytes + local frames = {} + while true do + local i = buffer:find("\n", 1, true) + if not i then break end + frames[#frames + 1] = buffer:sub(1, i - 1) + buffer = buffer:sub(i + 1) + end + return frames + end, + flush = function() + if #buffer == 0 then return nil end + local frame = buffer + buffer = "" + return frame + end, + reset = function() + buffer = "" + end, + pending_len = function() + return #buffer + end, + } + "#, + ); + + let mut framer = LuaFramer::from_script_path(path.to_string_lossy()).unwrap(); + assert_eq!(framer.feed(b"one\nt"), vec![b"one".to_vec()]); + assert_eq!(framer.pending_len(), 1); + assert_eq!(framer.feed(b"wo\nthree"), vec![b"two".to_vec()]); + assert_eq!(framer.flush(), Some(b"three".to_vec())); + assert_eq!(framer.pending_len(), 0); + + fs::remove_file(path).unwrap(); + } + + #[test] + fn lua_decoder_maps_text_binary_and_plot() { + let path = write_script( + "decoder", + r#" + return { + decode = function(frame) + if frame == "skip" then + return nil + elseif frame == "bin" then + return { kind = "binary", data = "\1\2" } + elseif frame == "plot" then + return { + kind = "plot", + channels = { { 1.0, 2.0 }, { 3.5, 4.5 } }, + sample_type = "F64", + format = "Block", + } + end + return { kind = "text", data = "decoded:" .. frame } + end, + } + "#, + ); + + let decoder = LuaDecoder::from_script_path(path.to_string_lossy()).unwrap(); + assert!(decoder.decode(b"skip").is_none()); + assert!(matches!( + decoder.decode(b"abc"), + Some(DecodedData::Text(text)) if text == "decoded:abc" + )); + assert!(matches!( + decoder.decode(b"bin"), + Some(DecodedData::Binary(bytes)) if bytes == vec![1, 2] + )); + match decoder.decode(b"plot").unwrap() { + DecodedData::Plot(frame) => { + assert_eq!(frame.channels, vec![vec![1.0, 2.0], vec![3.5, 4.5]]); + assert_eq!(frame.sample_type, SampleType::F64); + assert_eq!(frame.format, PlotFormat::Block); + assert_eq!(frame.raw, b"plot"); + } + other => panic!("expected plot, got {other:?}"), + } + + fs::remove_file(path).unwrap(); + } + + #[test] + fn lua_framer_requires_expected_functions() { + let path = write_script("bad_framer", "return { feed = function() return {} end }"); + let result = LuaFramer::from_script_path(path.to_string_lossy()); + assert!(result.is_err()); + fs::remove_file(path).unwrap(); + } +} diff --git a/crates/xserial-client/src/lua/mod.rs b/crates/xserial-client/src/lua/mod.rs index 76b0d94..3be1979 100644 --- a/crates/xserial-client/src/lua/mod.rs +++ b/crates/xserial-client/src/lua/mod.rs @@ -6,6 +6,7 @@ use mlua::{Lua, LuaSerdeExt, Result as LuaResult, Table, Value}; use crate::config::SessionConfig; use crate::manager::SessionManager; +pub(crate) mod codec; mod session_api; #[derive(Clone)] diff --git a/crates/xserial-client/src/session.rs b/crates/xserial-client/src/session.rs index 6ce4ad8..1805766 100644 --- a/crates/xserial-client/src/session.rs +++ b/crates/xserial-client/src/session.rs @@ -206,6 +206,7 @@ pub struct Session { config: SessionConfig, conn: Option, pipeline: MultiPipeline, + pipeline_build_error: Option, history: RingBuffer, cmd_rx: mpsc::Receiver, event_tx: broadcast::Sender, @@ -217,9 +218,17 @@ impl Session { pub fn spawn(id: SessionId, config: SessionConfig) -> SessionHandle { let (cmd_tx, cmd_rx) = mpsc::channel(32); let (event_tx, _) = broadcast::channel(256); + let (pipeline, pipeline_build_error) = match Self::build_pipeline(&config) { + Ok(pipeline) => (pipeline, None), + Err(err) => ( + MultiPipeline::new(), + Some(format!("pipeline build failed: {err}")), + ), + }; let mut session = Self { id, - pipeline: Self::build_pipeline(&config), + pipeline, + pipeline_build_error, history: RingBuffer::new(config.history_limit), config, conn: None, @@ -233,22 +242,29 @@ impl Session { SessionHandle::new(inner) } - fn build_pipeline(config: &SessionConfig) -> MultiPipeline { + fn build_pipeline(config: &SessionConfig) -> crate::Result { let mut pipelines = MultiPipeline::new(); for pipeline in &config.pipelines { pipelines.add(Pipeline::new( pipeline.name.clone(), - pipeline.framer.clone().build(), - pipeline.decoder.clone().build(), + pipeline.framer.clone().build()?, + pipeline.decoder.clone().build()?, )); } - pipelines + Ok(pipelines) } async fn run(&mut self) { info!(session_id = self.id, "session started"); - if let Err(err) = self.connect().await { + if let Some(message) = self.pipeline_build_error.take() { + self.emit_error(message); + self.desired_connected = false; + } + + if self.desired_connected + && let Err(err) = self.connect().await + { warn!(session_id = self.id, error = %err, "initial connect failed"); self.emit_error(format!("connect failed: {err}")); } @@ -346,9 +362,16 @@ impl Session { async fn handle_reconfigure(&mut self, config: SessionConfig) { info!(session_id = self.id, "reconfiguring session"); + let pipeline = match Self::build_pipeline(&config) { + Ok(pipeline) => pipeline, + Err(err) => { + self.emit_error(format!("pipeline build failed: {err}")); + return; + } + }; self.disconnect(true).await; self.config = config; - self.pipeline = Self::build_pipeline(&self.config); + self.pipeline = pipeline; self.history = RingBuffer::new(self.config.history_limit); if let Err(err) = self.connect().await { diff --git a/crates/xserial-client/tests/lua_tests.rs b/crates/xserial-client/tests/lua_tests.rs index a854c50..73cb349 100644 --- a/crates/xserial-client/tests/lua_tests.rs +++ b/crates/xserial-client/tests/lua_tests.rs @@ -1,5 +1,6 @@ use std::sync::Once; use std::time::Duration; +use std::{fs, path::PathBuf}; use mlua::Lua; use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -16,6 +17,16 @@ fn init_tracing() { }); } +fn write_temp_lua(name: &str, script: &str) -> PathBuf { + let nonce = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + let path = std::env::temp_dir().join(format!("xserial_{name}_{nonce}.lua")); + fs::write(&path, script).unwrap(); + path +} + fn text_pipeline_lua() -> &'static str { r#" { @@ -190,6 +201,102 @@ async fn lua_session_hex_decoder() { server.await.unwrap(); } +// ── xserial.open with Lua framer + Lua decoder ─────────────────── + +#[tokio::test] +async fn lua_session_custom_lua_pipeline() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap().to_string(); + + let framer_path = write_temp_lua( + "custom_framer", + r#" + local buffer = "" + return { + feed = function(bytes) + buffer = buffer .. bytes + local frames = {} + while true do + local i = buffer:find("|", 1, true) + if not i then break end + frames[#frames + 1] = buffer:sub(1, i - 1) + buffer = buffer:sub(i + 1) + end + return frames + end, + flush = function() + if #buffer == 0 then return nil end + local frame = buffer + buffer = "" + return frame + end, + reset = function() + buffer = "" + end, + pending_len = function() + return #buffer + end, + } + "#, + ); + let decoder_path = write_temp_lua( + "custom_decoder", + r#" + return { + decode = function(frame) + return { kind = "text", data = string.upper(frame) } + end, + } + "#, + ); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + stream.write_all(b"alpha|").await.unwrap(); + tokio::time::sleep(Duration::from_millis(50)).await; + stream.write_all(b"beta|").await.unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; + }); + + let lua = Lua::new(); + xserial_client::lua::register(&lua).unwrap(); + + let script = format!( + r#" + local sess = xserial.open({{ + transport = {{ Tcp = {{ addr = "{}" }} }}, + pipelines = {{ + {{ + name = "custom", + framer = {{ Lua = {{ script_path = [[{}]] }} }}, + decoder = {{ Lua = {{ script_path = [[{}]] }} }} + }} + }} + }}) + + local r1 = sess:read(5000) + assert(r1 ~= nil, "expected first custom frame") + assert(r1.pipeline == "custom", "expected custom pipeline, got " .. tostring(r1.pipeline)) + assert(r1.kind == "text", "expected text kind, got " .. tostring(r1.kind)) + assert(r1.data == "ALPHA", "expected ALPHA, got " .. tostring(r1.data)) + + local r2 = sess:read(5000) + assert(r2 ~= nil, "expected second custom frame") + assert(r2.data == "BETA", "expected BETA, got " .. tostring(r2.data)) + + sess:close() + "#, + addr, + framer_path.display(), + decoder_path.display() + ); + + lua.load(&script).exec_async().await.unwrap(); + server.await.unwrap(); + fs::remove_file(framer_path).unwrap(); + fs::remove_file(decoder_path).unwrap(); +} + // ── session:on_data callback ───────────────────────────────────── #[tokio::test] diff --git a/crates/xserial-gui/src/panels/config.rs b/crates/xserial-gui/src/panels/config.rs index fd1e949..78de1e5 100644 --- a/crates/xserial-gui/src/panels/config.rs +++ b/crates/xserial-gui/src/panels/config.rs @@ -1,12 +1,12 @@ use egui::{ComboBox, DragValue, ScrollArea, TextEdit, Ui}; use xserial_client::config::{DecoderConfig, FramerConfig, PipelineConfig, SessionConfig}; -use xserial_core::protocol::Endian; use xserial_core::protocol::plot::{PlotFormat, SampleType}; use xserial_core::protocol::text::TextEncoding; -use xserial_core::transport::TransportConfig; +use xserial_core::protocol::Endian; use xserial_core::transport::serial::{ SerialDataBits, SerialFlowControl, SerialParity, SerialStopBits, SerialTransport, }; +use xserial_core::transport::TransportConfig; #[derive(Clone)] pub struct ConfigForm { @@ -44,6 +44,7 @@ pub struct PipelineForm { pub mixed_strip_cr: bool, pub mixed_max_line_len: usize, pub mixed_max_plot_frame: usize, + pub lua_framer_script_path: String, pub hex_uppercase: bool, pub hex_separator: String, pub hex_bytes_per_group: usize, @@ -52,6 +53,7 @@ pub struct PipelineForm { pub plot_channels: usize, pub plot_endian: Endian, pub plot_format: PlotFormat, + pub lua_decoder_script_path: String, } #[derive(Clone, Copy, PartialEq, Eq)] @@ -68,6 +70,7 @@ pub enum FramerChoice { Length, Cobs, MixedTextPlot, + Lua, } #[derive(Clone, Copy, PartialEq, Eq)] @@ -76,6 +79,7 @@ pub enum DecoderChoice { Hex, Plot, MixedTextPlot, + Lua, } impl PipelineForm { @@ -96,6 +100,7 @@ impl PipelineForm { mixed_strip_cr: true, mixed_max_line_len: 1_048_576, mixed_max_plot_frame: 1_048_576, + lua_framer_script_path: String::new(), hex_uppercase: false, hex_separator: String::from(" "), hex_bytes_per_group: 1, @@ -104,6 +109,7 @@ impl PipelineForm { plot_channels: 1, plot_endian: Endian::Little, plot_format: PlotFormat::Interleaved, + lua_decoder_script_path: String::new(), } } } @@ -271,6 +277,10 @@ impl PipelineForm { form.mixed_max_line_len = *max_line_len; form.mixed_max_plot_frame = *max_plot_frame; } + FramerConfig::Lua { script_path } => { + form.framer = FramerChoice::Lua; + form.lua_framer_script_path = script_path.clone(); + } } match &config.decoder { @@ -306,6 +316,10 @@ impl PipelineForm { form.decoder = DecoderChoice::MixedTextPlot; form.text_encoding = *encoding; } + DecoderConfig::Lua { script_path } => { + form.decoder = DecoderChoice::Lua; + form.lua_decoder_script_path = script_path.clone(); + } } form @@ -336,6 +350,9 @@ impl PipelineForm { max_line_len: self.mixed_max_line_len, max_plot_frame: self.mixed_max_plot_frame, }, + FramerChoice::Lua => FramerConfig::Lua { + script_path: self.lua_framer_script_path.clone(), + }, }, decoder: match self.decoder { DecoderChoice::Text => DecoderConfig::Text { @@ -356,6 +373,9 @@ impl PipelineForm { DecoderChoice::MixedTextPlot => DecoderConfig::MixedTextPlot { encoding: self.text_encoding, }, + DecoderChoice::Lua => DecoderConfig::Lua { + script_path: self.lua_decoder_script_path.clone(), + }, }, } } @@ -592,12 +612,14 @@ fn render_pipeline_header( FramerChoice::Length => "Length", FramerChoice::Cobs => "Cobs", FramerChoice::MixedTextPlot => "MixedTextPlot", + FramerChoice::Lua => "Lua", }) .show_ui(ui, |ui| { ui.selectable_value(&mut pipeline.framer, FramerChoice::Line, "Line"); ui.selectable_value(&mut pipeline.framer, FramerChoice::Fixed, "Fixed"); ui.selectable_value(&mut pipeline.framer, FramerChoice::Length, "Length"); ui.selectable_value(&mut pipeline.framer, FramerChoice::Cobs, "Cobs"); + ui.selectable_value(&mut pipeline.framer, FramerChoice::Lua, "Lua"); if ui .selectable_label( matches!(pipeline.framer, FramerChoice::MixedTextPlot), @@ -616,10 +638,12 @@ fn render_pipeline_header( DecoderChoice::Hex => "Hex", DecoderChoice::Plot => "Plot", DecoderChoice::MixedTextPlot => "MixedTextPlot", + DecoderChoice::Lua => "Lua", }) .show_ui(ui, |ui| { ui.selectable_value(&mut pipeline.decoder, DecoderChoice::Text, "Text"); ui.selectable_value(&mut pipeline.decoder, DecoderChoice::Hex, "Hex"); + ui.selectable_value(&mut pipeline.decoder, DecoderChoice::Lua, "Lua"); if ui .selectable_label(matches!(pipeline.decoder, DecoderChoice::Plot), "Plot") .clicked() @@ -710,6 +734,14 @@ fn render_framer_settings(ui: &mut Ui, pipeline: &mut PipelineForm, index: usize ui.add(DragValue::new(&mut pipeline.mixed_max_plot_frame).range(1..=16_777_216)); }); } + FramerChoice::Lua => { + ui.horizontal(|ui| { + ui.label("Script:"); + ui.add( + TextEdit::singleline(&mut pipeline.lua_framer_script_path).desired_width(260.0), + ); + }); + } } } @@ -789,6 +821,15 @@ fn render_decoder_settings(ui: &mut Ui, pipeline: &mut PipelineForm, index: usiz "Plot sample type, endian, format, and channels come from the mixed packet header.", ); } + DecoderChoice::Lua => { + ui.horizontal(|ui| { + ui.label("Script:"); + ui.add( + TextEdit::singleline(&mut pipeline.lua_decoder_script_path) + .desired_width(260.0), + ); + }); + } } } @@ -850,6 +891,18 @@ fn validate_pipeline(ui: &mut Ui, pipeline: &PipelineForm, errors: &mut Vec bool: def main() -> None: parser = argparse.ArgumentParser(description="xserial plot test server") parser.add_argument("--host", default="127.0.0.1") - parser.add_argument("--port", type=int, default=8091) + parser.add_argument("--port", type=int, default=8080) parser.add_argument("--channels", type=int, default=1) parser.add_argument( "--format", @@ -148,9 +148,9 @@ def main() -> None: ) parser.add_argument( "--wire-format", - choices=("mixed", "raw"), + choices=("mixed", "raw", "text"), default="mixed", - help="mixed sends text+plot on one connection; raw sends plot only", + help="mixed sends text+plot; raw sends plot only; text sends text only", ) args = parser.parse_args() @@ -193,19 +193,38 @@ def main() -> None: values.append(args.amp * math.sin(2 * math.pi * args.freq * t + phase)) return values + def build_text_line(started_at: float, now: float, sample_index: int) -> str: + values = sample_values(sample_index) + joined = ", ".join( + f"ch{index + 1}={value:8.3f}" for index, value in enumerate(values) + ) + return ( + f"测试 time={now - started_at:7.3f}s " + f"sample={sample_index:7d} {joined}\n" + ) + print(f"Listening {args.host}:{args.port} - Ctrl+C to stop") print("GUI:") print(f" transport = TCP {args.host}:{args.port}") if args.wire_format == "mixed": print(" framer = MixedTextPlot") print(" decoder = MixedTextPlot\n") - else: + elif args.wire_format == "raw": print(f" framer = Fixed({frame_bytes})") print( f" decoder = Plot(f32, little-endian, {args.channels} channel, {args.format})\n" ) - print(f"sending {samples_per_channel} samples/frame at {args.rate:.1f} samples/sec") - print(f"effective frame rate: {frame_rate:.2f} fps\n") + else: + print(" framer = Line") + print(" decoder = Text") + print(" lua test = Lua framer tests/lua_line_framer.lua") + print(" Lua decoder tests/lua_text_decoder.lua\n") + if args.wire_format == "text": + print(f"sending text only at {args.text_interval:.3f}s intervals") + print(f"sample clock: {args.rate:.1f} samples/sec\n") + else: + print(f"sending {samples_per_channel} samples/frame at {args.rate:.1f} samples/sec") + print(f"effective frame rate: {frame_rate:.2f} fps\n") stop_event = threading.Event() @@ -230,6 +249,25 @@ def main() -> None: next_text_at = started_at try: while not stop_event.is_set(): + if args.wire_format == "text": + now = time.perf_counter() + wait = next_text_at - now + if wait > 0: + time.sleep(min(wait, 0.1)) + continue + + sample_index = int((now - started_at) * args.rate) + conn.sendall( + build_text_line( + started_at, + now, + sample_index, + ).encode("utf-8") + ) + text_count += 1 + next_text_at += args.text_interval + continue + plot_payload = build_frame( sample_index=sample_index, channels=args.channels, @@ -252,16 +290,13 @@ def main() -> None: now = time.perf_counter() if args.wire_format == "mixed" and now >= next_text_at: - values = sample_values(sample_index) - joined = ", ".join( - f"ch{index + 1}={value:8.3f}" - for index, value in enumerate(values) + conn.sendall( + build_text_line( + started_at, + now, + sample_index, + ).encode("utf-8") ) - line = ( - f"测试 time={now - started_at:7.3f}s " - f"sample={sample_index:7d} {joined}\n" - ) - conn.sendall(line.encode("utf-8")) text_count += 1 next_text_at += args.text_interval @@ -275,10 +310,13 @@ def main() -> None: except OSError as err: if not is_disconnect_error(err): raise - print( - f"[conn -] {addr} " - f"({frame_count} plot frames, {text_count} text lines)" - ) + if args.wire_format == "text": + print(f"[conn -] {addr} ({text_count} text lines)") + else: + print( + f"[conn -] {addr} " + f"({frame_count} plot frames, {text_count} text lines)" + ) finally: conn.close() finally: