diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0d14c1a..c2e8d32 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -30,5 +30,12 @@ jobs: - name: Clippy run: cargo clippy --workspace --all-targets -- -D warnings + - name: Smoke-check Python tools + run: | + python3 -m py_compile tools/*.py + for f in tools/test_plot.py tools/test_drone.py tools/test_ansi.py tools/stress_test.py; do + python3 "$f" --help >/dev/null + done + - name: Run tests run: cargo test --workspace diff --git a/README.md b/README.md index c7d3d9c..36a943e 100644 --- a/README.md +++ b/README.md @@ -229,7 +229,7 @@ return { } ``` -参考实现：`tests/lua_line_framer.lua`（按 `\n` 分割的行分帧器） +参考实现：`crates/pipeview-client/tests/fixtures/lua_line_framer.lua`（按 `\n` 分割的行分帧器） ### 解码器 API @@ -363,11 +363,11 @@ RUST_LOG=pipeview_gui=debug cargo run -p pipeview-gui # 详细日志 crates/ pipeview-core/ # 传输、分帧、协议 pipeview-client/ # 会话管理、Lua 运行时 + tests/fixtures/ # Lua 测试夹具 pipeview-gui/ # egui 桌面应用 examples/ # Lua 脚本示例 drone_plot.lua # 飞控遥测波形解码器 drone_text.lua # 飞控遥测文本解码器 -tests/ # Lua 测试 fixture tools/ # 开发辅助工具 test_plot.py # 波形测试数据生成器 test_drone.py # 飞控测试数据生成器 diff --git a/README_en.md b/README_en.md index 5c66867..bccfbe7 100644 --- a/README_en.md +++ b/README_en.md @@ -217,7 +217,7 @@ return { } ``` -Reference: `tests/lua_line_framer.lua` (line-based framer splitting on `\n`). +Reference: `crates/pipeview-client/tests/fixtures/lua_line_framer.lua` (line-based framer splitting on `\n`). ### Decoder API @@ -347,11 +347,11 @@ RUST_LOG=pipeview_gui=debug cargo run -p pipeview-gui crates/ pipeview-core/ # Transport, framing, protocol pipeview-client/ # Session management, Lua runtime + tests/fixtures/ # Lua test fixtures pipeview-gui/ # egui desktop application examples/ # Lua script examples drone_plot.lua # Drone telemetry plot decoder drone_text.lua # Drone telemetry text decoder -tests/ # Lua test fixtures tools/ # Development utilities test_plot.py # Waveform test data generator test_drone.py # Drone test data generator diff --git a/tests/lua_line_framer.lua b/crates/pipeview-client/tests/fixtures/lua_line_framer.lua similarity index 100% rename from tests/lua_line_framer.lua rename to crates/pipeview-client/tests/fixtures/lua_line_framer.lua diff --git a/tests/lua_text_decoder.lua b/crates/pipeview-client/tests/fixtures/lua_text_decoder.lua similarity index 100% rename from tests/lua_text_decoder.lua rename to crates/pipeview-client/tests/fixtures/lua_text_decoder.lua diff --git a/crates/pipeview-client/tests/lua_fixtures.rs b/crates/pipeview-client/tests/lua_fixtures.rs new file mode 100644 index 0000000..96b2214 --- /dev/null +++ b/crates/pipeview-client/tests/lua_fixtures.rs @@ -0,0 +1,66 @@ +use std::fs; +use std::path::PathBuf; + +use mlua::{Function, Lua, Table}; + +fn fixture_path(name: &str) -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("tests") + .join("fixtures") + .join(name) +} + +fn load_fixture(name: &str) -> (Lua, Table) { + let lua = Lua::new(); + let src = fs::read_to_string(fixture_path(name)).unwrap(); + let table: Table = lua.load(&src).set_name(name).eval().unwrap(); + (lua, table) +} + +#[test] +fn lua_line_framer_fixture_splits_crlf_and_lf() { + let (lua, table) = load_fixture("lua_line_framer.lua"); + let feed: Function = table.get("feed").unwrap(); + + let input = lua.create_string(b"hello\r\nworld\n").unwrap(); + let frames: Vec = feed.call(input).unwrap(); + assert_eq!(frames, vec!["hello", "world"]); +} + +#[test] +fn lua_line_framer_fixture_handles_split_chunks_and_flush() { + let (lua, table) = load_fixture("lua_line_framer.lua"); + let feed: Function = table.get("feed").unwrap(); + let flush: Function = table.get("flush").unwrap(); + let pending_len: Function = table.get("pending_len").unwrap(); + let reset: Function = table.get("reset").unwrap(); + + let frames: Vec = feed.call(lua.create_string(b"par").unwrap()).unwrap(); + assert!(frames.is_empty()); + assert_eq!(pending_len.call::(()).unwrap(), 3); + + let frames: Vec = feed.call(lua.create_string(b"tial\n").unwrap()).unwrap(); + assert_eq!(frames, vec!["partial"]); + + let flushed: Option = flush.call(()).unwrap(); + assert_eq!(flushed.as_deref(), None); + + reset.call::<()>(()).unwrap(); + assert_eq!(pending_len.call::(()).unwrap(), 0); +} + +#[test] +fn lua_text_decoder_fixture_decodes_text_and_rejects_empty() { + let (lua, table) = load_fixture("lua_text_decoder.lua"); + let decode: Function = table.get("decode").unwrap(); + + let decoded: Option = decode.call(lua.create_string(b"abc").unwrap()).unwrap(); + let decoded = decoded.expect("non-empty frame should decode"); + let kind: String = decoded.get("kind").unwrap(); + let data: String = decoded.get("data").unwrap(); + assert_eq!(kind, "text"); + assert_eq!(data, "abc"); + + let empty: Option
= decode.call(lua.create_string(b"").unwrap()).unwrap(); + assert!(empty.is_none(), "empty frame should be rejected"); +} diff --git a/tools/pv_protocol.py b/tools/pv_protocol.py new file mode 100644 index 0000000..04a16ed --- /dev/null +++ b/tools/pv_protocol.py @@ -0,0 +1,158 @@ +#!/usr/bin/env python3 +"""Shared wire-format helpers for pipeview test tools. + +Keep these functions in sync with: + - crates/pipeview-core/src/frame/cobs.rs (cobs_encode) + - crates/pipeview-core/src/protocol/mixed.rs (XP plot packet header) +""" + +import math +import struct + +PLOT_ESCAPE = 0x1E +PLOT_MARKER = ord("P") + +DISCONNECT_WINERRORS = { + 10053, # Software caused connection abort + 10054, # Connection reset by peer + 10058, # Socket shutdown race on Windows +} + +# XP plot packet header field values (see protocol/mixed.rs) +PLOT_PACKET_MAGIC = b"XP" +PLOT_PACKET_VERSION = 1 +PLOT_FORMAT_IDS = { + "interleaved": 0, + "block": 1, + "xy": 2, +} +SAMPLE_TYPE_F32 = 8 +ENDIAN_LITTLE = 0 + + +def cobs_encode(payload: bytes) -> bytes: + """Consistent Overhead Byte Stuffing encoder (matches Rust cobs_encode).""" + if not payload: + return b"\x01" + + out = bytearray([0]) + code_index = 0 + code = 1 + for byte in payload: + if byte == 0: + out[code_index] = code + code_index = len(out) + out.append(0) + code = 1 + else: + out.append(byte) + code += 1 + if code == 0xFF: + out[code_index] = code + code_index = len(out) + out.append(0) + code = 1 + out[code_index] = code + return bytes(out) + + +def build_plot_payload( + sample_index: int, + channels: int, + plot_format: str, + samples_per_channel: int, + amplitude: float, + frequency_hz: float, + sample_rate_hz: float, +) -> bytes: + """Build a raw little-endian f32 plot payload without any framing.""" + if channels <= 0: + raise ValueError("channels must be >= 1") + if plot_format not in PLOT_FORMAT_IDS: + raise ValueError(f"unsupported plot format: {plot_format}") + if plot_format == "xy" and channels != 2: + raise ValueError("xy format requires exactly 2 channels") + + values = [] + if plot_format == "xy": + for offset in range(samples_per_channel): + t = (sample_index + offset) / sample_rate_hz + values.extend( + [ + amplitude * math.cos(2 * math.pi * frequency_hz * t), + amplitude * math.sin(2 * math.pi * frequency_hz * t), + ] + ) + elif plot_format == "block": + for ch in range(channels): + phase = 2 * math.pi * ch / channels + for offset in range(samples_per_channel): + t = (sample_index + offset) / sample_rate_hz + values.append( + amplitude * math.sin(2 * math.pi * frequency_hz * t + phase) + ) + else: # interleaved + for offset in range(samples_per_channel): + t = (sample_index + offset) / sample_rate_hz + for ch in range(channels): + phase = 2 * math.pi * ch / channels + values.append( + amplitude * math.sin(2 * math.pi * frequency_hz * t + phase) + ) + + return struct.pack(f"<{len(values)}f", *values) + + +def build_plot_packet( + payload: bytes, + channels: int, + samples_per_channel: int, + plot_format: str, +) -> bytes: + """Build an XP plot packet (the payload inside a MixedTextPlot COBS frame).""" + if not 1 <= channels <= 255: + raise ValueError("channels must be in 1..255") + if not 0 <= samples_per_channel <= 0xFFFF: + raise ValueError("samples_per_channel must fit in u16") + if plot_format not in PLOT_FORMAT_IDS: + raise ValueError(f"unsupported plot format: {plot_format}") + + header = bytearray() + header.extend(PLOT_PACKET_MAGIC) + header.append(PLOT_PACKET_VERSION) + header.append(PLOT_FORMAT_IDS[plot_format]) + header.append(SAMPLE_TYPE_F32) + header.append(ENDIAN_LITTLE) + header.append(channels) + header.extend(struct.pack(" bytes: + """Build a complete MixedTextPlot plot frame (escape + marker + COBS packet + 0x00).""" + packet = build_plot_packet( + payload=payload, + channels=channels, + samples_per_channel=samples_per_channel, + plot_format=plot_format, + ) + return bytes([PLOT_ESCAPE, PLOT_MARKER]) + cobs_encode(packet) + b"\x00" + + +def is_disconnect_error(err: OSError) -> bool: + """Return True when the socket error is an expected client-disconnect error.""" + return isinstance( + err, + ( + BrokenPipeError, + ConnectionAbortedError, + ConnectionResetError, + ), + ) or getattr(err, "winerror", None) in DISCONNECT_WINERRORS diff --git a/tools/stress_test.py b/tools/stress_test.py index 7267589..a56259d 100755 --- a/tools/stress_test.py +++ b/tools/stress_test.py @@ -1,8 +1,8 @@ #!/usr/bin/env python3 """ -pipeview-tauri 极限压力测试脚本 +pipeview GUI（egui）极限压力测试脚本 -模拟各种高数据量场景，测试 Tauri 前端的渲染性能和稳定性： +模拟各种高数据量场景，测试 pipeview GUI 的渲染性能和稳定性： - 文本洪水：大文本行 + ANSI 颜色 - 十六进制洪水：大批量 hex dump - 波形洪水：高频 plot 采样点 @@ -20,20 +20,11 @@ import argparse import math import random import socket -import struct -import sys import threading import time from collections import defaultdict -PLOT_ESCAPE = 0x1E -PLOT_MARKER = ord("P") - -DISCONNECT_WINERRORS = { - 10053, # Software caused connection abort - 10054, # Connection reset by peer - 10058, # Socket shutdown race on Windows -} +import pv_protocol # ── ANSI Color Palette ───────────────────────────────────────────── @@ -108,57 +99,32 @@ def build_plot_frame( frequency_hz: float, sample_rate_hz: float, ) -> bytes: - """Build a raw COBS-encoded plot frame (matching test_plot.py format).""" - values = [] - if plot_format == "xy": - for _ in range(samples_per_channel): - t = (sample_index * samples_per_channel + len(values)) / sample_rate_hz - values.append(amplitude * math.sin(2 * math.pi * frequency_hz * t)) - values.append(amplitude * math.cos(2 * math.pi * frequency_hz * t)) - elif plot_format == "block": - for ch in range(channels): - phase = 2 * math.pi * ch / channels - for _ in range(samples_per_channel): - t = (sample_index * samples_per_channel + len(values) // channels) / sample_rate_hz - values.append(amplitude * math.sin(2 * math.pi * frequency_hz * t + phase)) - else: # interleaved - for _ in range(samples_per_channel): - t = (sample_index * samples_per_channel + len(values) // channels) / sample_rate_hz - for ch in range(channels): - phase = 2 * math.pi * ch / channels - values.append(amplitude * math.sin(2 * math.pi * frequency_hz * t + phase)) - - # Pack as f32 little-endian - payload = struct.pack(f"<{len(values)}f", *values) - return cobs_encode(payload) - - -def cobs_encode(data: bytes) -> bytes: - """Consistent Overhead Byte Stuffing encoder.""" - result = bytearray() - block_start = 0 - while block_start < len(data): - end = min(block_start + 254, len(data)) - block = data[block_start:end] - if end < len(data): - result.append(len(block) + 1) - result.extend(block) - else: - result.append(len(block) + 1 if len(block) < 254 else 255) - result.extend(block) - result.append(0) - block_start = end - return bytes(result) + """Build a raw little-endian f32 plot payload (no framing).""" + return pv_protocol.build_plot_payload( + sample_index=sample_index, + channels=channels, + plot_format=plot_format, + samples_per_channel=samples_per_channel, + amplitude=amplitude, + frequency_hz=frequency_hz, + sample_rate_hz=sample_rate_hz, + ) def build_mixed_frame(seq: int, text_rate_per_plot: int, channels: int) -> bytes: - """Build a MixedTextPlot frame: text lines + one COBS plot frame.""" + """Build a MixedTextPlot frame: text lines + one XP plot frame.""" frames = [] for i in range(text_rate_per_plot): frames.append(generate_text_line(seq * text_rate_per_plot + i, ansi=True)) - # Add one plot frame per text_rate_per_plot text lines - plot_data = build_plot_frame(seq, channels, "interleaved", 32, 100.0, 10.0, 1000.0) - frames.append(bytes([PLOT_ESCAPE, PLOT_MARKER]) + plot_data) + plot_payload = build_plot_frame(seq, channels, "interleaved", 32, 100.0, 10.0, 1000.0) + frames.append( + pv_protocol.build_mixed_plot_frame( + payload=plot_payload, + channels=channels, + samples_per_channel=32, + plot_format="interleaved", + ) + ) return b"".join(frames) @@ -273,7 +239,7 @@ class StressServer: last_count_time = time.time() except OSError as e: - if hasattr(e, "winerror") and e.winerror in DISCONNECT_WINERRORS: + if pv_protocol.is_disconnect_error(e): pass elif not self.running: pass @@ -324,7 +290,7 @@ class StressServer: def main(): parser = argparse.ArgumentParser( - description="pipeview-tauri 极限压力测试", + description="pipeview GUI（egui）极限压力测试", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" 示例: diff --git a/tools/test_plot.py b/tools/test_plot.py index 4524670..63b00ec 100755 --- a/tools/test_plot.py +++ b/tools/test_plot.py @@ -4,115 +4,10 @@ import argparse import math import socket -import struct import threading import time -PLOT_ESCAPE = 0x1E -PLOT_MARKER = ord("P") -DISCONNECT_WINERRORS = { - 10053, # Software caused connection abort - 10054, # Connection reset by peer - 10058, # Socket shutdown race on Windows -} - - -def build_frame( - sample_index: int, - channels: int, - plot_format: str, - samples_per_channel: int, - amplitude: float, - frequency_hz: float, - sample_rate_hz: float, -) -> bytes: - values = [] - if plot_format == "xy": - for offset in range(samples_per_channel): - t = (sample_index + offset) / sample_rate_hz - x = amplitude * math.cos(2 * math.pi * frequency_hz * t) - y = amplitude * math.sin(2 * math.pi * frequency_hz * t) - values.extend([x, y]) - return struct.pack(f"<{len(values)}f", *values) - - for offset in range(samples_per_channel): - t = (sample_index + offset) / sample_rate_hz - for ch in range(channels): - phase = 2 * math.pi * ch / max(channels, 1) - value = amplitude * math.sin(2 * math.pi * frequency_hz * t + phase) - values.append(value) - return struct.pack(f"<{len(values)}f", *values) - - -def cobs_encode(payload: bytes) -> bytes: - if not payload: - return b"\x01" - - out = bytearray([0]) - code_index = 0 - code = 1 - for byte in payload: - if byte == 0: - out[code_index] = code - code_index = len(out) - out.append(0) - code = 1 - else: - out.append(byte) - code += 1 - if code == 0xFF: - out[code_index] = code - code_index = len(out) - out.append(0) - code = 1 - out[code_index] = code - return bytes(out) - - -def build_plot_packet( - payload: bytes, - channels: int, - samples_per_channel: int, - plot_format: str, -) -> bytes: - format_id = {"interleaved": 0, "block": 1, "xy": 2}[plot_format] - header = bytearray() - header.extend(b"XP") - header.append(1) # version - header.append(format_id) - header.append(8) # f32 - header.append(0) # little-endian - header.append(channels) - header.extend(struct.pack(" bytes: - packet = build_plot_packet( - payload=payload, - channels=channels, - samples_per_channel=samples_per_channel, - plot_format=plot_format, - ) - return bytes([PLOT_ESCAPE, PLOT_MARKER]) + cobs_encode(packet) + b"\x00" - - -def is_disconnect_error(err: OSError) -> bool: - return isinstance( - err, - ( - BrokenPipeError, - ConnectionAbortedError, - ConnectionResetError, - ), - ) or getattr(err, "winerror", None) in DISCONNECT_WINERRORS +import pv_protocol def main() -> None: @@ -217,8 +112,8 @@ def main() -> None: 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") + print(" lua test = Lua framer crates/pipeview-client/tests/fixtures/lua_line_framer.lua") + print(" Lua decoder crates/pipeview-client/tests/fixtures/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") @@ -268,7 +163,7 @@ def main() -> None: next_text_at += args.text_interval continue - plot_payload = build_frame( + plot_payload = pv_protocol.build_plot_payload( sample_index=sample_index, channels=args.channels, plot_format=args.format, @@ -278,7 +173,7 @@ def main() -> None: sample_rate_hz=args.rate, ) if args.wire_format == "mixed": - frame = build_mixed_plot_frame( + frame = pv_protocol.build_mixed_plot_frame( payload=plot_payload, channels=args.channels, samples_per_channel=samples_per_channel, @@ -308,7 +203,7 @@ def main() -> None: if wait > 0: time.sleep(wait) except OSError as err: - if not is_disconnect_error(err): + if not pv_protocol.is_disconnect_error(err): raise if args.wire_format == "text": print(f"[conn -] {addr} ({text_count} text lines)")