FallenSigh
4e4c9cee65
- 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
318 lines
12 KiB
Python
Executable File
318 lines
12 KiB
Python
Executable File
#!/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()
|