From 4e4c9cee65b16843d80829886d1be71bf5a77a25 Mon Sep 17 00:00:00 2001 From: FallenSigh Date: Sun, 14 Jun 2026 04:12:35 +0800 Subject: [PATCH] feat(gui): add ANSI escape sequence renderer with full panel integration - new ansi_render module: SGR, cursor movement, erase, scroll - integrate ANSI rendering into console panel output pipeline - wire app_state, buffers, hex_view, and shortcuts for ANSI-aware display - add test_ansi.py for manual ANSI sequence testing --- Cargo.lock | 21 + crates/pipeview-gui/Cargo.toml | 1 + crates/pipeview-gui/src/ansi_render.rs | 451 +++++++++++++++++++++ crates/pipeview-gui/src/app.rs | 8 +- crates/pipeview-gui/src/app_state.rs | 4 +- crates/pipeview-gui/src/buffers.rs | 4 +- crates/pipeview-gui/src/main.rs | 1 + crates/pipeview-gui/src/panels/console.rs | 49 +-- crates/pipeview-gui/src/panels/hex_view.rs | 13 +- crates/pipeview-gui/src/shortcuts.rs | 5 +- tools/test_ansi.py | 317 +++++++++++++++ 11 files changed, 819 insertions(+), 55 deletions(-) create mode 100644 crates/pipeview-gui/src/ansi_render.rs create mode 100755 tools/test_ansi.py diff --git a/Cargo.lock b/Cargo.lock index 0cf7aea..f28362f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -189,6 +189,16 @@ dependencies = [ "libc", ] +[[package]] +name = "ansitok" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0a8acea8c2f1c60f0a92a8cd26bf96ca97db56f10bbcab238bbe0cceba659ee" +dependencies = [ + "nom 7.1.3", + "vte", +] + [[package]] name = "anstream" version = "1.0.0" @@ -3297,6 +3307,7 @@ dependencies = [ name = "pipeview-gui" version = "0.1.0" dependencies = [ + "ansitok", "eframe", "egui", "egui_plot", @@ -4720,6 +4731,16 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "vte" +version = "0.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "231fdcd7ef3037e8330d8e17e61011a2c244126acc0a982f4040ac3f9f0bc077" +dependencies = [ + "arrayvec", + "memchr", +] + [[package]] name = "vtparse" version = "0.6.2" diff --git a/crates/pipeview-gui/Cargo.toml b/crates/pipeview-gui/Cargo.toml index 1b2307d..3ab5a06 100644 --- a/crates/pipeview-gui/Cargo.toml +++ b/crates/pipeview-gui/Cargo.toml @@ -19,3 +19,4 @@ tracing-subscriber = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } hex = "0.4" +ansitok = "0.3" diff --git a/crates/pipeview-gui/src/ansi_render.rs b/crates/pipeview-gui/src/ansi_render.rs new file mode 100644 index 0000000..206206a --- /dev/null +++ b/crates/pipeview-gui/src/ansi_render.rs @@ -0,0 +1,451 @@ +use ansitok::{AnsiColor, ElementKind, VisualAttribute, parse_ansi, parse_ansi_sgr}; +use egui::Color32; +use egui::text::{LayoutJob, TextFormat}; +use std::ops::Range; + +pub fn ansi_to_layout_job(text: &str, base_format: TextFormat) -> LayoutJob { + let mut job = LayoutJob::default(); + for segment in ansi_segments(text, &base_format) { + job.append(segment.text, 0.0, segment.format); + } + + job +} + +pub fn ansi_to_layout_job_highlighted( + text: &str, + base_format: TextFormat, + query: &str, + case_sensitive: bool, + highlight_format: TextFormat, +) -> LayoutJob { + if query.is_empty() { + return ansi_to_layout_job(text, base_format); + } + + let segments = ansi_segments(text, &base_format); + let mut visible_text = String::with_capacity(segments.iter().map(|s| s.text.len()).sum()); + for segment in &segments { + visible_text.push_str(segment.text); + } + + let ranges = search_ranges(&visible_text, query, case_sensitive); + if ranges.is_empty() { + let mut job = LayoutJob::default(); + for segment in segments { + job.append(segment.text, 0.0, segment.format); + } + return job; + } + + let mut job = LayoutJob::default(); + let mut offset = 0; + let mut range_idx = 0; + for segment in segments { + let len = segment.text.len(); + append_segment_with_highlights( + &mut job, + segment.text, + segment.format, + &highlight_format, + offset, + &ranges, + &mut range_idx, + ); + offset += len; + } + job +} + +struct StyledSegment<'a> { + text: &'a str, + format: TextFormat, +} + +#[derive(Default)] +struct AnsiStyle { + fg: Option, + bg: Option, + bold: bool, + italic: bool, + underline: bool, + strikethrough: bool, +} + +impl AnsiStyle { + fn apply_sgr(&mut self, attr: VisualAttribute) { + match attr { + VisualAttribute::Reset(0) => self.reset_all(), + VisualAttribute::Reset(22) => self.bold = false, + VisualAttribute::Reset(23) => self.italic = false, + VisualAttribute::Reset(24) => self.underline = false, + VisualAttribute::Reset(29) => self.strikethrough = false, + VisualAttribute::Reset(39) => self.fg = None, + VisualAttribute::Reset(49) => self.bg = None, + VisualAttribute::Reset(_) => {} + VisualAttribute::Bold => self.bold = true, + VisualAttribute::Faint => self.bold = false, + VisualAttribute::Italic => self.italic = true, + VisualAttribute::Underline => self.underline = true, + VisualAttribute::Crossedout => self.strikethrough = true, + VisualAttribute::FgColor(c) => self.fg = Some(ansi_color_to_egui(c)), + VisualAttribute::BgColor(c) => self.bg = Some(ansi_color_to_egui(c)), + _ => {} + } + } + + fn reset_all(&mut self) { + *self = Self::default(); + } + + fn format(&self, base_format: &TextFormat) -> TextFormat { + let mut fmt = base_format.clone(); + + if let Some(c) = self.fg { + if self.bold { + fmt.color = brighten(c); + } else { + fmt.color = c; + } + } + if let Some(c) = self.bg { + fmt.background = c; + } + if self.italic { + fmt.italics = true; + } + if self.underline { + fmt.underline = egui::Stroke::new(1.0, fmt.color); + } + if self.strikethrough { + fmt.strikethrough = egui::Stroke::new(1.0, fmt.color); + } + + fmt + } +} + +fn ansi_segments<'a>(text: &'a str, base_format: &TextFormat) -> Vec> { + let mut segments = Vec::new(); + let mut style = AnsiStyle::default(); + + for element in parse_ansi(text) { + match element.kind() { + ElementKind::Text => { + let slice = &text[element.start()..element.end()]; + if !slice.is_empty() { + segments.push(StyledSegment { + text: slice, + format: style.format(base_format), + }); + } + } + ElementKind::Sgr => { + let sgr = &text[element.start()..element.end()]; + apply_sgr_sequence(&mut style, sgr); + } + _ => {} + } + } + + segments +} + +fn apply_sgr_sequence(style: &mut AnsiStyle, sgr: &str) { + let params = sgr + .strip_prefix("\x1b[") + .and_then(|s| s.strip_suffix('m')) + .unwrap_or(sgr); + + if params.is_empty() { + style.reset_all(); + return; + } + + for output in parse_ansi_sgr(sgr) { + if let Some(attr) = output.as_escape() { + style.apply_sgr(attr); + } + } +} + +fn search_ranges(text: &str, query: &str, case_sensitive: bool) -> Vec> { + if query.is_empty() { + return Vec::new(); + } + + if case_sensitive { + return text + .match_indices(query) + .map(|(start, matched)| start..start + matched.len()) + .collect(); + } + + let (search_text, byte_to_original) = lowercase_with_byte_map(text); + let query = query.to_lowercase(); + let mut ranges = Vec::new(); + let mut idx = 0; + while let Some(pos) = search_text[idx..].find(&query) { + let start = idx + pos; + let end = start + query.len(); + let original_start = byte_to_original.get(start).copied().unwrap_or(text.len()); + let original_end = byte_to_original.get(end).copied().unwrap_or(text.len()); + if original_start < original_end { + ranges.push(original_start..original_end); + } + idx = end; + } + ranges +} + +fn lowercase_with_byte_map(text: &str) -> (String, Vec) { + let mut lowered = String::new(); + let mut byte_to_original = Vec::new(); + + for (original_idx, ch) in text.char_indices() { + for lower_ch in ch.to_lowercase() { + lowered.push(lower_ch); + for _ in 0..lower_ch.len_utf8() { + byte_to_original.push(original_idx); + } + } + } + + byte_to_original.push(text.len()); + (lowered, byte_to_original) +} + +fn append_segment_with_highlights( + job: &mut LayoutJob, + text: &str, + format: TextFormat, + highlight_format: &TextFormat, + segment_start: usize, + ranges: &[Range], + range_idx: &mut usize, +) { + let segment_end = segment_start + text.len(); + while *range_idx < ranges.len() && ranges[*range_idx].end <= segment_start { + *range_idx += 1; + } + + let mut idx = *range_idx; + let mut last = 0; + while idx < ranges.len() && ranges[idx].start < segment_end { + let range = &ranges[idx]; + let start = range.start.max(segment_start) - segment_start; + let end = range.end.min(segment_end) - segment_start; + + if last < start { + job.append(&text[last..start], 0.0, format.clone()); + } + if start < end { + job.append(&text[start..end], 0.0, highlight_format.clone()); + } + last = end; + + if range.end <= segment_end { + idx += 1; + } else { + break; + } + } + + *range_idx = idx; + if last < text.len() { + job.append(&text[last..], 0.0, format); + } +} + +fn ansi_color_to_egui(color: AnsiColor) -> Color32 { + match color { + AnsiColor::Bit4(c) => ansi_4bit(c), + AnsiColor::Bit8(c) => ansi_256(c), + AnsiColor::Bit24 { r, g, b } => Color32::from_rgb(r, g, b), + } +} + +fn ansi_4bit(code: u8) -> Color32 { + match code { + 30 => Color32::BLACK, + 31 => Color32::from_rgb(194, 54, 33), + 32 => Color32::from_rgb(37, 188, 36), + 33 => Color32::from_rgb(173, 173, 39), + 34 => Color32::from_rgb(73, 46, 225), + 35 => Color32::from_rgb(211, 56, 211), + 36 => Color32::from_rgb(51, 187, 200), + 37 => Color32::from_rgb(203, 204, 205), + + 90 => Color32::from_rgb(129, 131, 131), + 91 => Color32::from_rgb(252, 57, 31), + 92 => Color32::from_rgb(49, 231, 34), + 93 => Color32::from_rgb(234, 236, 35), + 94 => Color32::from_rgb(88, 51, 255), + 95 => Color32::from_rgb(249, 53, 248), + 96 => Color32::from_rgb(20, 240, 240), + 97 => Color32::from_rgb(233, 235, 235), + + 40 => Color32::BLACK, + 41 => Color32::from_rgb(194, 54, 33), + 42 => Color32::from_rgb(37, 188, 36), + 43 => Color32::from_rgb(173, 173, 39), + 44 => Color32::from_rgb(73, 46, 225), + 45 => Color32::from_rgb(211, 56, 211), + 46 => Color32::from_rgb(51, 187, 200), + 47 => Color32::from_rgb(203, 204, 205), + + 100 => Color32::from_rgb(129, 131, 131), + 101 => Color32::from_rgb(252, 57, 31), + 102 => Color32::from_rgb(49, 231, 34), + 103 => Color32::from_rgb(234, 236, 35), + 104 => Color32::from_rgb(88, 51, 255), + 105 => Color32::from_rgb(249, 53, 248), + 106 => Color32::from_rgb(20, 240, 240), + 107 => Color32::from_rgb(233, 235, 235), + + _ => Color32::WHITE, + } +} + +fn ansi_256(code: u8) -> Color32 { + match code { + 0..=15 => ansi_4bit(if code < 8 { code + 30 } else { code + 82 }), + 16..=231 => { + let idx = code - 16; + let cube = [0, 95, 135, 175, 215, 255]; + let r = cube[(idx / 36) as usize]; + let g = cube[((idx / 6) % 6) as usize]; + let b = cube[(idx % 6) as usize]; + Color32::from_rgb(r, g, b) + } + 232..=255 => { + let v = (code - 232) * 10 + 8; + Color32::from_rgb(v, v, v) + } + } +} + +fn brighten(color: Color32) -> Color32 { + Color32::from_rgb( + (color.r() as u32 * 13 / 10).min(255) as u8, + (color.g() as u32 * 13 / 10).min(255) as u8, + (color.b() as u32 * 13 / 10).min(255) as u8, + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn visible_text(job: &LayoutJob) -> String { + job.sections + .iter() + .map(|s| &job.text[s.byte_range.clone()]) + .collect() + } + + #[test] + fn plain_text_no_ansi() { + let fmt = TextFormat::default(); + let job = ansi_to_layout_job("hello", fmt); + assert_eq!(job.sections.len(), 1); + } + + #[test] + fn fg_red_reset() { + let fmt = TextFormat::default(); + let job = ansi_to_layout_job("\x1b[31mred\x1b[0m plain", fmt); + assert!(job.sections.len() >= 2); + assert!(job.sections[0].format.color != job.sections[1].format.color); + } + + #[test] + fn empty_sgr_resets_all_styles() { + let fmt = TextFormat { + color: Color32::from_rgb(1, 2, 3), + ..Default::default() + }; + let job = ansi_to_layout_job("\x1b[31mred\x1b[m plain", fmt.clone()); + + assert_eq!(visible_text(&job), "red plain"); + assert!(job.sections.len() >= 2); + assert_ne!(job.sections[0].format.color, job.sections[1].format.color); + assert_eq!(job.sections[1].format.color, fmt.color); + } + + #[test] + fn selective_foreground_reset_keeps_background() { + let fmt = TextFormat { + color: Color32::from_rgb(1, 2, 3), + ..Default::default() + }; + let job = ansi_to_layout_job("\x1b[41;37mtext\x1b[39mmore", fmt.clone()); + + assert_eq!(visible_text(&job), "textmore"); + assert!(job.sections.len() >= 2); + assert_ne!(job.sections[0].format.color, job.sections[1].format.color); + assert_eq!(job.sections[1].format.color, fmt.color); + assert_ne!(job.sections[0].format.background, fmt.background); + assert_eq!( + job.sections[0].format.background, + job.sections[1].format.background + ); + } + + #[test] + fn bold_brightens_color() { + let fmt = TextFormat::default(); + let job = ansi_to_layout_job("\x1b[1;31mbold red\x1b[0m", fmt); + assert_eq!(job.sections.len(), 1); + let c = job.sections[0].format.color; + assert!(c.r() > 200 || c.g() > 50 || c.b() > 30); + } + + #[test] + fn strip_ansi_codes_from_output() { + let fmt = TextFormat::default(); + let job = ansi_to_layout_job("\x1b[32mgreen\x1b[0m", fmt); + let text: String = job + .sections + .iter() + .map(|s| &job.text[s.byte_range.clone()]) + .collect(); + assert_eq!(text, "green"); + } + + #[test] + fn highlighted_search_uses_visible_ansi_text() { + let default = TextFormat { + color: Color32::WHITE, + ..Default::default() + }; + let highlight = TextFormat { + color: Color32::BLACK, + background: Color32::from_rgb(255, 255, 0), + ..Default::default() + }; + let job = ansi_to_layout_job_highlighted( + "\x1b[31mred\x1b[0m plain", + default, + "red plain", + true, + highlight.clone(), + ); + + assert_eq!(visible_text(&job), "red plain"); + assert!(job.sections.len() >= 2); + assert!( + job.sections + .iter() + .all(|section| section.format.background == highlight.background) + ); + } + + #[test] + fn xterm_256_color_cube_values() { + assert_eq!(ansi_256(16), Color32::from_rgb(0, 0, 0)); + assert_eq!(ansi_256(21), Color32::from_rgb(0, 0, 255)); + assert_eq!(ansi_256(52), Color32::from_rgb(95, 0, 0)); + assert_eq!(ansi_256(67), Color32::from_rgb(95, 135, 175)); + } +} diff --git a/crates/pipeview-gui/src/app.rs b/crates/pipeview-gui/src/app.rs index b1ae20a..671233a 100644 --- a/crates/pipeview-gui/src/app.rs +++ b/crates/pipeview-gui/src/app.rs @@ -1310,18 +1310,22 @@ fn render_send_panel(ui: &mut egui::Ui, manager: &SessionManager, tab: &mut Sess ui.add_space(3.0); let hint = match tab.send_mode { - SendMode::Text => "Enter text to send", + SendMode::Text => "Enter to send, Ctrl+Enter for newline", SendMode::Hex => "Enter hex bytes, e.g. 48 65 6C 6C 6F", }; let response = ui.add( TextEdit::multiline(&mut tab.send_input) .desired_rows(6) .desired_width(f32::INFINITY) + .return_key(egui::KeyboardShortcut::new( + egui::Modifiers::COMMAND, + egui::Key::Enter, + )) .hint_text(hint), ); let wants_submit = response.has_focus() - && ui.input(|input| input.key_pressed(egui::Key::Enter) && input.modifiers.command_only()); + && ui.input(|input| input.key_pressed(egui::Key::Enter) && input.modifiers.is_none()); let mut send_clicked = false; ui.horizontal(|ui| { diff --git a/crates/pipeview-gui/src/app_state.rs b/crates/pipeview-gui/src/app_state.rs index 2052697..6fe0004 100644 --- a/crates/pipeview-gui/src/app_state.rs +++ b/crates/pipeview-gui/src/app_state.rs @@ -3,9 +3,9 @@ use std::fs; use std::io; use std::path::{Path, PathBuf}; +use pipeview_client::config::SessionConfig; use serde::{Deserialize, Serialize}; use tracing::warn; -use pipeview_client::config::SessionConfig; const GUI_STATE_FILE_NAME: &str = "gui-state.json"; @@ -123,8 +123,8 @@ const fn default_true() -> bool { #[cfg(test)] mod tests { use super::*; - use std::time::{SystemTime, UNIX_EPOCH}; use pipeview_core::transport::TransportConfig; + use std::time::{SystemTime, UNIX_EPOCH}; #[test] fn windows_gui_state_path_uses_appdata() { diff --git a/crates/pipeview-gui/src/buffers.rs b/crates/pipeview-gui/src/buffers.rs index fad79cc..b0fb019 100644 --- a/crates/pipeview-gui/src/buffers.rs +++ b/crates/pipeview-gui/src/buffers.rs @@ -1,10 +1,10 @@ use egui_plot::PlotBounds; -use std::collections::VecDeque; -use std::time::{Duration, Instant}; use pipeview_client::RingBuffer; use pipeview_client::event::DecodedEntry; use pipeview_core::protocol::DecodedData; use pipeview_core::protocol::plot::{PlotFormat, PlotFrame}; +use std::collections::VecDeque; +use std::time::{Duration, Instant}; #[derive(Clone)] pub enum LineDirection { diff --git a/crates/pipeview-gui/src/main.rs b/crates/pipeview-gui/src/main.rs index 2c7eb38..7329806 100644 --- a/crates/pipeview-gui/src/main.rs +++ b/crates/pipeview-gui/src/main.rs @@ -1,5 +1,6 @@ #![cfg_attr(windows, windows_subsystem = "windows")] +mod ansi_render; mod app; mod app_state; mod buffers; diff --git a/crates/pipeview-gui/src/panels/console.rs b/crates/pipeview-gui/src/panels/console.rs index 353c6c9..73f9604 100644 --- a/crates/pipeview-gui/src/panels/console.rs +++ b/crates/pipeview-gui/src/panels/console.rs @@ -1,3 +1,4 @@ +use crate::ansi_render::{ansi_to_layout_job, ansi_to_layout_job_highlighted}; use crate::app::{DisplayOptions, SearchState}; use crate::buffers::{ConsoleLine, LineDirection, TextBuffer}; use egui::{ @@ -31,42 +32,6 @@ pub fn render( line_count } -fn highlight_matches( - text: &str, - query: &str, - case_sensitive: bool, - default_format: TextFormat, - highlight_format: TextFormat, -) -> LayoutJob { - let search_text = if case_sensitive { - text.to_string() - } else { - text.to_lowercase() - }; - let query = if case_sensitive { - query.to_string() - } else { - query.to_lowercase() - }; - let mut job = LayoutJob::default(); - let mut last_end = 0; - let mut idx = 0; - while let Some(pos) = search_text[idx..].find(&query) { - let start = idx + pos; - let end = start + query.len(); - if last_end < start { - job.append(&text[last_end..start], 0.0, default_format.clone()); - } - job.append(&text[start..end], 0.0, highlight_format.clone()); - last_end = end; - idx = end; - } - if last_end < text.len() { - job.append(&text[last_end..], 0.0, default_format); - } - job -} - fn monospace_format(style: &egui::Style) -> TextFormat { TextFormat { font_id: style @@ -121,9 +86,13 @@ pub fn format_console_line( }; match search { - Some(state) if state.active && !state.query.is_empty() => { - highlight_matches(&full_text, &state.query, state.case_sensitive, default, highlight) - } - _ => LayoutJob::single_section(full_text, default), + Some(state) if state.active && !state.query.is_empty() => ansi_to_layout_job_highlighted( + &full_text, + default, + &state.query, + state.case_sensitive, + highlight, + ), + _ => ansi_to_layout_job(&full_text, default), } } diff --git a/crates/pipeview-gui/src/panels/hex_view.rs b/crates/pipeview-gui/src/panels/hex_view.rs index e737f13..9de76c8 100644 --- a/crates/pipeview-gui/src/panels/hex_view.rs +++ b/crates/pipeview-gui/src/panels/hex_view.rs @@ -78,7 +78,7 @@ fn monospace_format(style: &egui::Style) -> TextFormat { .get(&TextStyle::Monospace) .cloned() .unwrap_or_default(), - color: Color32::WHITE, + color: Color32::WHITE, ..Default::default() } } @@ -125,10 +125,13 @@ fn format_hex_line( }; match search { - Some(state) if state.active && !state.query.is_empty() => { - highlight_matches(&full_text, &state.query, state.case_sensitive, default, highlight) - } + Some(state) if state.active && !state.query.is_empty() => highlight_matches( + &full_text, + &state.query, + state.case_sensitive, + default, + highlight, + ), _ => LayoutJob::single_section(full_text, default), } } - diff --git a/crates/pipeview-gui/src/shortcuts.rs b/crates/pipeview-gui/src/shortcuts.rs index 1f27a5c..c47bbba 100644 --- a/crates/pipeview-gui/src/shortcuts.rs +++ b/crates/pipeview-gui/src/shortcuts.rs @@ -56,10 +56,7 @@ pub fn default_bindings() -> Vec<(Action, KeyboardShortcut)> { ), (Search, KeyboardShortcut::new(Modifiers::CTRL, Key::F)), (SearchNext, KeyboardShortcut::new(Modifiers::NONE, Key::F3)), - ( - SearchPrev, - KeyboardShortcut::new(Modifiers::SHIFT, Key::F3), - ), + (SearchPrev, KeyboardShortcut::new(Modifiers::SHIFT, Key::F3)), ] } diff --git a/tools/test_ansi.py b/tools/test_ansi.py new file mode 100755 index 0000000..6907b8b --- /dev/null +++ b/tools/test_ansi.py @@ -0,0 +1,317 @@ +#!/usr/bin/env python3 +"""Generate ANSI-colored terminal output for testing pipeview ANSI rendering.""" + +import argparse +import math +import random +import socket +import sys +import time +import itertools + +# ── helpers ────────────────────────────────────────────────────────────────── + +RESET = "\x1b[0m" + +# Standard 3/4-bit foreground +FG = { + "black": "\x1b[30m", "red": "\x1b[31m", "green": "\x1b[32m", + "yellow": "\x1b[33m", "blue": "\x1b[34m", "magenta":"\x1b[35m", + "cyan": "\x1b[36m", "white": "\x1b[37m", +} +FG_BRIGHT = { + "black": "\x1b[90m", "red": "\x1b[91m", "green": "\x1b[92m", + "yellow": "\x1b[93m", "blue": "\x1b[94m", "magenta":"\x1b[95m", + "cyan": "\x1b[96m", "white": "\x1b[97m", +} + +# Standard 3/4-bit background +BG = { + "black": "\x1b[40m", "red": "\x1b[41m", "green": "\x1b[42m", + "yellow": "\x1b[43m", "blue": "\x1b[44m", "magenta":"\x1b[45m", + "cyan": "\x1b[46m", "white": "\x1b[47m", +} +BG_BRIGHT = { + "black": "\x1b[100m", "red": "\x1b[101m", "green": "\x1b[102m", + "yellow": "\x1b[103m", "blue": "\x1b[104m", "magenta":"\x1b[105m", + "cyan": "\x1b[106m", "white": "\x1b[107m", +} + +BOLD = "\x1b[1m" +FAINT = "\x1b[2m" +ITALIC = "\x1b[3m" +UNDERLINE = "\x1b[4m" +STRIKE = "\x1b[9m" + + +def fg256(n: int) -> str: + return f"\x1b[38;5;{n}m" + +def bg256(n: int) -> str: + return f"\x1b[48;5;{n}m" + +def fg_rgb(r: int, g: int, b: int) -> str: + return f"\x1b[38;2;{r};{g};{b}m" + +def bg_rgb(r: int, g: int, b: int) -> str: + return f"\x1b[48;2;{r};{g};{b}m" + + +# ── test scenes ────────────────────────────────────────────────────────────── + +def scene_16color(delay: float): + """Show all 8 standard + 8 bright foreground colors.""" + lines = [] + lines.append(f"{BOLD}─── Standard 16 Foreground Colors ───{RESET}") + for name in FG: + lines.append(f" {FG[name]}■ {name:>8}{RESET} " + f"{FG_BRIGHT[name]}■ bright {name}{RESET}") + for line in lines: + yield line, delay + + +def scene_bg_colors(delay: float): + """Show foreground colors on colored backgrounds.""" + lines = [] + lines.append(f"{BOLD}─── Background Colors ───{RESET}") + for bg_name in BG: + line = "" + for fg_name in ["white", "black"]: + line += f"{BG[bg_name]}{FG[fg_name]} {fg_name} on {bg_name} {RESET} " + lines.append(line) + for line in lines: + yield line, delay + + +def scene_styles(delay: float): + """Show bold, italic, underline, strikethrough.""" + lines = [] + lines.append(f"{BOLD}─── Text Styles ───{RESET}") + lines.append(f" {BOLD}Bold text{RESET}") + lines.append(f" {ITALIC}Italic text{RESET}") + lines.append(f" {UNDERLINE}Underlined text{RESET}") + lines.append(f" {STRIKE}Strikethrough text{RESET}") + lines.append(f" {BOLD}{ITALIC}Bold + Italic{RESET}") + lines.append(f" {BOLD}{FG['red']}Bold Red{RESET} vs {FG['red']}Normal Red{RESET}") + for line in lines: + yield line, delay + + +def scene_256_ramp(delay: float): + """Show a 256-color ramp.""" + lines = [] + lines.append(f"{BOLD}─── 256-Color Ramp ───{RESET}") + # 16 basic colors + line = "Basic: " + for n in range(16): + line += f"{fg256(n)}██{RESET}" + lines.append(line) + for row in range(3): + line = f"Cube {row+1}: " + for col in range(6): + n = 16 + row * 36 + col * 6 + line += f"{fg256(n)}██{RESET}" + lines.append(line) + line = "Gray: " + for n in range(232, 256): + line += f"{fg256(n)}█{RESET}" + lines.append(line) + for line in lines: + yield line, delay + + +def scene_truecolor(delay: float): + """Show 24-bit truecolor gradient.""" + lines = [] + lines.append(f"{BOLD}─── Truecolor (24-bit) ───{RESET}") + line = "R→G: " + for i in range(32): + r = 255 - i * 8 + g = i * 8 + line += f"{fg_rgb(r, g, 0)}█{RESET}" + lines.append(line) + line = "B→Y: " + for i in range(32): + b = 255 - i * 8 + r = g = i * 8 + line += f"{fg_rgb(r, g, b)}█{RESET}" + lines.append(line) + line = "Rainbow: " + for i in range(48): + hue = i / 48.0 * 6.0 + r, g, b = _hsv_to_rgb(hue, 1.0, 1.0) + line += f"{fg_rgb(r, g, b)}━{RESET}" + lines.append(line) + for line in lines: + yield line, delay + + +def scene_status_log(delay: float): + """Simulate systemd-style status output.""" + lines = [ + f"{FG_BRIGHT['green']}[ OK ]{RESET} Started {BOLD}System Logging Service{RESET}.", + f"{FG_BRIGHT['green']}[ OK ]{RESET} Started {BOLD}Network Manager{RESET}.", + f"{FG_BRIGHT['green']}[ OK ]{RESET} Reached target {ITALIC}multi-user.target{RESET}.", + f"{FG_BRIGHT['yellow']}[ WARN ]{RESET} Failed to load kernel module {UNDERLINE}nvidia{RESET}.", + f"{FG_BRIGHT['red']}[ FAIL ]{RESET} {FG['red']}{BOLD}ssh.service{RESET}{FG['red']} failed to start.{RESET}", + f"{FG_BRIGHT['cyan']}[ INFO ]{RESET} Listening on {fg256(33)}0.0.0.0:8080{RESET}.", + f" {ITALIC}─ subject=CN=example.com{RESET}", + f" {ITALIC}─ fingerprint={FG['yellow']}SHA256:abcd1234{RESET}", + f"{FG_BRIGHT['green']}[ OK ]{RESET} Mounted {BG['blue']}{FG['white']} /var {RESET} filesystem.", + f"{FG_BRIGHT['magenta']}[STATUS]{RESET} CPU: {fg256(46)}32%{RESET} " + f"Mem: {fg256(220)}1.2G{RESET}/{fg256(33)}4.0G{RESET} " + f"Temp: {_temp_color(58)}58°C{RESET}", + ] + for line in lines: + yield line, delay + + +def scene_colored_log_stream(delay: float): + """Continuously generate log lines cycling through themes.""" + themes = [ + ('green', 'INFO ', "Connection accepted from 192.168.1.100"), + ('cyan', 'DEBUG', "Processing frame #{}"), + ('yellow', 'WARN ', "Buffer usage at {:.0f}%"), + ('red', 'ERROR', "CRC mismatch on packet {}"), + ('white', 'TRACE', "Entering function handle_request()"), + ('magenta','AUDIT', "User admin performed action {}"), + ] + counter = itertools.count(1) + while True: + color_name, level, template = random.choice(themes) + n = next(counter) + msg = template.format(n, random.uniform(60, 95), n) + ts = time.strftime("%H:%M:%S") + line = (f"{ITALIC}{ts}{RESET} " + f"{FG_BRIGHT[color_name]}{BOLD}[{level}]{RESET} " + f"{msg}") + yield line, delay + + +def scene_rainbow_wave(delay: float): + """Animated rainbow wave (for live testing).""" + t0 = time.time() + while True: + t = time.time() - t0 + line = "" + for x in range(60): + hue = (x / 60.0 + t * 0.3) % 1.0 + r, g, b = _hsv_to_rgb(hue * 6.0, 1.0, 0.8 + 0.2 * math.sin(t * 2 + x * 0.3)) + line += f"{fg_rgb(r, g, b)}━{RESET}" + yield line, delay + + +def scene_system_boot(delay: float): + """Simulate a system boot sequence with progress.""" + services = [ + ("udev", "Kernel Device Manager", 0.6), + ("systemd-journald","Journal Service", 0.8), + ("NetworkManager", "Network Manager", 0.7), + ("sshd", "OpenSSH Daemon", 0.5), + ("nginx", "HTTP Server", 0.9), + ("postgresql", "PostgreSQL 16", 1.2), + ("docker", "Docker Engine", 1.5), + ("pipeview", "Pipe Data Monitor", 0.3), + ] + t0 = time.time() + for name, desc, startup_time in services: + yield (f"{FG_BRIGHT['cyan']}[ .... ]{RESET} Starting {BOLD}{name}{RESET} - {desc}...", + startup_time * 0.3) + yield (f"{FG_BRIGHT['green']}[ OK ]{RESET} Started {BOLD}{name}{RESET} - {desc}.", + delay) + yield (f"\n{FG_BRIGHT['green']}{BOLD}Boot complete.{RESET} " + f"({time.time() - t0:.1f}s)", delay) + + +# ── helpers ────────────────────────────────────────────────────────────────── + +def _hsv_to_rgb(h: float, s: float, v: float) -> tuple[int, int, int]: + """HSV → RGB, h in [0, 6), s,v in [0, 1].""" + c = v * s + x = c * (1 - abs((h % 2) - 1)) + m = v - c + r, g, b = { + 0: (c, x, 0), 1: (x, c, 0), 2: (0, c, x), + 3: (0, x, c), 4: (x, 0, c), 5: (c, 0, x), + }[int(h) % 6] + return int((r + m) * 255), int((g + m) * 255), int((b + m) * 255) + + +def _temp_color(temp: float) -> str: + if temp < 50: return fg256(46) + elif temp < 70: return fg256(220) + else: return fg256(196) + + +# ── scenes registry ────────────────────────────────────────────────────────── + +STATIC_SCENES = [ + ("16color", scene_16color, "Standard 16 foreground colors"), + ("bg", scene_bg_colors, "Background colors"), + ("styles", scene_styles, "Bold, italic, underline, strikethrough"), + ("256ramp", scene_256_ramp, "256-color ramp"), + ("truecolor", scene_truecolor, "24-bit truecolor gradients"), + ("status", scene_status_log, "systemd-style status log"), + ("boot", scene_system_boot, "System boot sequence"), +] + +LIVE_SCENES = [ + ("logstream", scene_colored_log_stream, "Continuous colored log stream"), + ("rainbow", scene_rainbow_wave, "Animated rainbow wave"), +] + + +# ── main ───────────────────────────────────────────────────────────────────── + +def main(): + parser = argparse.ArgumentParser( + description="ANSI color test data generator for pipeview", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog="Scenes:\n" + "\n".join( + f" {n:<14} {d}" for n, _, d in STATIC_SCENES + LIVE_SCENES + ), + ) + parser.add_argument("--host", default="127.0.0.1", help="TCP bind host") + parser.add_argument("--port", type=int, default=8099, help="TCP port") + parser.add_argument("--scene", choices=[n for n, _, _ in STATIC_SCENES + LIVE_SCENES] + ["all"], + default="all", help="Scene to play") + parser.add_argument("--delay", type=float, default=0.15, help="Inter-line delay (seconds)") + parser.add_argument("--loop", action="store_true", help="Repeat the scene forever") + args = parser.parse_args() + + scenes = STATIC_SCENES + LIVE_SCENES + if args.scene != "all": + scenes = [(n, f, d) for n, f, d in scenes if n == args.scene] + + print(f"ANSI color test → {args.host}:{args.port}") + print(f"Scene: {args.scene}, delay: {args.delay}s, loop: {args.loop}") + + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server: + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server.bind((args.host, args.port)) + server.listen(1) + print(f"Listening on {args.host}:{args.port}, waiting for connections...") + + while True: + sock, addr = server.accept() + print(f"Client connected from {addr}") + try: + _play_scenes(sock, scenes, args.delay, args.loop) + except (BrokenPipeError, ConnectionResetError): + pass + print("Client disconnected. Waiting for new connection...") + + +def _play_scenes(sock: socket.socket, scenes, delay: float, loop_scenes: bool): + first = True + while first or loop_scenes: + first = False + for name, scene_fn, _desc in scenes: + print(f" [{name}]") + for line, line_delay in scene_fn(delay): + sock.sendall((line + "\n").encode()) + time.sleep(line_delay) + + +if __name__ == "__main__": + main()