#!/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()