From a5c80de68e419b0dbd153dd9f54160b47fbe7820 Mon Sep 17 00:00:00 2001 From: "Leandro G. Almeida" Date: Sat, 25 Apr 2026 11:32:50 -0700 Subject: [PATCH 1/6] =?UTF-8?q?=E2=9C=A8=20ui:=20prototype=20terminal=20st?= =?UTF-8?q?artup=20animation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/particle_logo_standalone.py | 471 ++++++++++++++++ scripts/tests.py | 394 ++++++++++++++ .../cli/commands/compliance/__init__.py | 15 +- src/repoman/cli/main_cli.py | 18 +- .../utils/theme/terminal_colors.py | 14 - .../tests/test_utils/test_theme.py.jinja | 22 - src/repoman/utils/task_tree_display.py | 197 +++++++ src/repoman/utils/ui/display_tasks.py | 509 ++++++++++++++++++ src/repoman/utils/ui/logo_display.py | 373 +++++++++++++ 9 files changed, 1969 insertions(+), 44 deletions(-) create mode 100644 scripts/particle_logo_standalone.py create mode 100644 scripts/tests.py delete mode 100644 src/repoman/main_template/src/{{python_package_import_name}}/utils/theme/terminal_colors.py create mode 100644 src/repoman/utils/task_tree_display.py create mode 100644 src/repoman/utils/ui/display_tasks.py create mode 100644 src/repoman/utils/ui/logo_display.py diff --git a/scripts/particle_logo_standalone.py b/scripts/particle_logo_standalone.py new file mode 100644 index 0000000..29fbba5 --- /dev/null +++ b/scripts/particle_logo_standalone.py @@ -0,0 +1,471 @@ +#!/usr/bin/env python3 +""" +Standalone particle coalesce logo for terminal use. + +Renders a two-line logo: + HUGGING FACE + ML INTERN + +No project-local imports required. +Dependency: rich +""" + +from __future__ import annotations + +import argparse +import math +import random +import shutil +import time +from dataclasses import dataclass + +from rich.align import Align +from rich.console import Console +from rich.live import Live +from rich.text import Text + +# ---------------------------- +# Small 5x7 bitmap font +# ---------------------------- + +FONT_5X7: dict[str, list[str]] = { + "A": [ + "01110", + "10001", + "10001", + "11111", + "10001", + "10001", + "10001", + ], + "C": [ + "01111", + "10000", + "10000", + "10000", + "10000", + "10000", + "01111", + ], + "E": [ + "11111", + "10000", + "10000", + "11110", + "10000", + "10000", + "11111", + ], + "F": [ + "11111", + "10000", + "10000", + "11110", + "10000", + "10000", + "10000", + ], + "G": [ + "01111", + "10000", + "10000", + "10011", + "10001", + "10001", + "01110", + ], + "H": [ + "10001", + "10001", + "10001", + "11111", + "10001", + "10001", + "10001", + ], + "I": [ + "11111", + "00100", + "00100", + "00100", + "00100", + "00100", + "11111", + ], + "L": [ + "10000", + "10000", + "10000", + "10000", + "10000", + "10000", + "11111", + ], + "M": [ + "10001", + "11011", + "10101", + "10101", + "10001", + "10001", + "10001", + ], + "N": [ + "10001", + "11001", + "10101", + "10011", + "10001", + "10001", + "10001", + ], + "R": [ + "11110", + "10001", + "10001", + "11110", + "10100", + "10010", + "10001", + ], + "T": [ + "11111", + "00100", + "00100", + "00100", + "00100", + "00100", + "00100", + ], + "U": [ + "10001", + "10001", + "10001", + "10001", + "10001", + "10001", + "01110", + ], + " ": [ + "000", + "000", + "000", + "000", + "000", + "000", + "000", + ], +} + + +def text_to_pixels( + text: str, scale: int = 2, letter_spacing: int = 1 +) -> list[tuple[int, int]]: + pixels: list[tuple[int, int]] = [] + x_cursor = 0 + for ch in text.upper(): + glyph = FONT_5X7.get(ch) + if glyph is None: + raise ValueError(f"Unsupported character in font: {ch!r}") + glyph_w = len(glyph[0]) + for gy, row in enumerate(glyph): + for gx, bit in enumerate(row): + if bit == "1": + for sy in range(scale): + for sx in range(scale): + pixels.append((x_cursor + gx * scale + sx, gy * scale + sy)) + x_cursor += glyph_w * scale + letter_spacing * scale + return pixels + + +# ---------------------------- +# Braille canvas +# ---------------------------- + +BRAILLE_BLANK = chr(0x2800) +BRAILLE_OFFSETS = { + (0, 0): 0x01, # dot 1 + (0, 1): 0x02, # dot 2 + (0, 2): 0x04, # dot 3 + (1, 0): 0x08, # dot 4 + (1, 1): 0x10, # dot 5 + (1, 2): 0x20, # dot 6 + (0, 3): 0x40, # dot 7 + (1, 3): 0x80, # dot 8 +} + + +class BrailleCanvas: + def __init__(self, width_chars: int, height_chars: int): + self.width_chars = max(1, width_chars) + self.height_chars = max(1, height_chars) + self.pixel_width = self.width_chars * 2 + self.pixel_height = self.height_chars * 4 + self._pixels: set[tuple[int, int]] = set() + + def clear(self) -> None: + self._pixels.clear() + + def set_pixel(self, x: int, y: int) -> None: + if 0 <= x < self.pixel_width and 0 <= y < self.pixel_height: + self._pixels.add((x, y)) + + def render(self) -> list[str]: + lines: list[str] = [] + for cy in range(self.height_chars): + chars = [] + for cx in range(self.width_chars): + bits = 0 + base_x = cx * 2 + base_y = cy * 4 + for dx in range(2): + for dy in range(4): + if (base_x + dx, base_y + dy) in self._pixels: + bits |= BRAILLE_OFFSETS[(dx, dy)] + chars.append(chr(0x2800 + bits) if bits else BRAILLE_BLANK) + lines.append("".join(chars)) + return lines + + +# ---------------------------- +# Timing / color +# ---------------------------- + + +def settle_curve(progress: float) -> float: + """A mild overshoot/settle profile in [0,1].""" + p = max(0.0, min(1.0, progress)) + # Damped oscillation, normalized to a small positive value. + return math.exp(-4.0 * p) * abs(math.sin(8.0 * math.pi * p)) + + +def warm_gold_from_white(progress: float) -> tuple[int, int, int]: + """Blend from near-white to warm gold.""" + p = max(0.0, min(1.0, progress)) + start = (255, 250, 235) + end = (255, 200, 80) + r = int(start[0] + (end[0] - start[0]) * p) + g = int(start[1] + (end[1] - start[1]) * p) + b = int(start[2] + (end[2] - start[2]) * p) + return r, g, b + + +# ---------------------------- +# Particles +# ---------------------------- + + +@dataclass +class Particle: + x: float + y: float + target_x: float + target_y: float + vx: float = 0.0 + vy: float = 0.0 + phase: float = 0.0 + delay: float = 0.0 + + def update_converge( + self, t: float, strength: float = 0.08, damping: float = 0.92 + ) -> None: + if t < self.delay: + self.x += self.vx + self.y += self.vy + self.vx *= 0.99 + self.vy *= 0.99 + angle = self.phase + t * 2.0 + self.vx += math.cos(angle) * 0.3 + self.vy += math.sin(angle) * 0.3 + return + + dx = self.target_x - self.x + dy = self.target_y - self.y + self.vx += dx * strength + self.vy += dy * strength + self.vx *= damping + self.vy *= damping + self.x += self.vx + self.y += self.vy + + +def get_bounds(pixels: list[tuple[int, int]]) -> tuple[int, int, int, int]: + if not pixels: + return 0, 0, 0, 0 + xs = [p[0] for p in pixels] + ys = [p[1] for p in pixels] + return min(xs), max(xs), min(ys), max(ys) + + +def build_targets(canvas: BrailleCanvas, scale: int = 2) -> list[tuple[int, int]]: + line1 = text_to_pixels("HUGGING FACE", scale=scale) + line2 = text_to_pixels("ML INTERN", scale=scale) + + min_x1, max_x1, min_y1, max_y1 = get_bounds(line1) + min_x2, max_x2, min_y2, max_y2 = get_bounds(line2) + + w1, h1 = max_x1 - min_x1 + 1, max_y1 - min_y1 + 1 + w2, h2 = max_x2 - min_x2 + 1, max_y2 - min_y2 + 1 + + gap = 6 + total_h = h1 + gap + h2 + start_y = (canvas.pixel_height - total_h) // 2 + + offset_x1 = (canvas.pixel_width - w1) // 2 - min_x1 + offset_y1 = start_y - min_y1 + offset_x2 = (canvas.pixel_width - w2) // 2 - min_x2 + offset_y2 = start_y + h1 + gap - min_y2 + + targets_1 = [(x + offset_x1, y + offset_y1) for x, y in line1] + targets_2 = [(x + offset_x2, y + offset_y2) for x, y in line2] + return targets_1 + targets_2 + + +def render_colored(canvas: BrailleCanvas, rgb: tuple[int, int, int]) -> Align: + r, g, b = rgb + result = Text() + for line in canvas.render(): + for ch in line: + if ch == BRAILLE_BLANK: + result.append(ch) + else: + result.append(ch, style=f"rgb({r},{g},{b})") + result.append("\n") + return Align.center(result) + + +def run_particle_logo( + console: Console, + hold_seconds: float = 1.5, + fps: int = 24, + max_particles: int = 1500, + ambient_count: int = 200, + scale: int = 2, +) -> None: + term = shutil.get_terminal_size(fallback=(100, 30)) + term_width = min(term.columns, 120) + term_height = min(max(10, term.lines - 4), 35) + canvas = BrailleCanvas(term_width, term_height) + + all_targets = build_targets(canvas, scale=scale) + step = max(1, len(all_targets) // max_particles) + sampled_targets = all_targets[::step] + + rng = random.Random(42) + particles: list[Particle] = [] + pw, ph = canvas.pixel_width, canvas.pixel_height + + for tx, ty in sampled_targets: + side = rng.choice(["top", "bottom", "left", "right"]) + if side == "top": + sx, sy = rng.uniform(0, pw), rng.uniform(-20, -5) + elif side == "bottom": + sx, sy = rng.uniform(0, pw), rng.uniform(ph + 5, ph + 20) + elif side == "left": + sx, sy = rng.uniform(-20, -5), rng.uniform(0, ph) + else: + sx, sy = rng.uniform(pw + 5, pw + 20), rng.uniform(0, ph) + + delay = rng.uniform(0, 0.4) + phase = random.uniform(0, math.pi * 2) + p = Particle(sx, sy, tx, ty, phase=phase, delay=delay) + + angle = math.atan2(ph / 2 - sy, pw / 2 - sx) + rng.gauss(0, 0.8) + speed = rng.uniform(1.0, 2.5) + p.vx = math.cos(angle) * speed + p.vy = math.sin(angle) * speed + particles.append(p) + + ambient: list[Particle] = [] + for _ in range(ambient_count): + ax = rng.uniform(0, pw) + ay = rng.uniform(0, ph) + ap = Particle(ax, ay, ax, ay, phase=random.uniform(0, math.pi * 2)) + ap.vx = rng.gauss(0, 1) + ap.vy = rng.gauss(0, 1) + ambient.append(ap) + + converge_frames = max(1, int(fps * 0.9)) + hold_frames = max(1, int(fps * hold_seconds)) + total_frames = converge_frames + hold_frames + + with Live(console=console, refresh_per_second=fps, transient=True) as live: + for frame in range(total_frames): + canvas.clear() + t = frame * (1.0 / fps) + + for ap in ambient: + ap.x += ap.vx + math.sin(t + ap.phase) * 0.5 + ap.y += ap.vy + math.cos(t + ap.phase * 1.3) * 0.5 + ap.x %= pw + ap.y %= ph + + if frame < converge_frames: + alpha = 0.3 + 0.2 * math.sin(t * 2 + ap.phase) + else: + fade = (frame - converge_frames) / hold_frames + alpha = (0.3 + 0.2 * math.sin(t * 2 + ap.phase)) * (1 - fade) + if alpha > 0.25: + canvas.set_pixel(int(ap.x), int(ap.y)) + + if frame < converge_frames: + progress = frame / converge_frames + noise = settle_curve(progress) + for p in particles: + p.update_converge(t, strength=0.06, damping=0.90) + canvas.set_pixel(int(p.x), int(p.y)) + trail_scale = 0.2 + 0.5 * noise + trail_x = int(p.x - p.vx * trail_scale) + trail_y = int(p.y - p.vy * trail_scale) + canvas.set_pixel(trail_x, trail_y) + rgb = warm_gold_from_white(progress) + else: + settle_t = (frame - converge_frames) / hold_frames + for p in particles: + jitter = (1 - settle_t) * 0.7 + jx = p.target_x + math.sin(t * 3 + p.phase) * jitter + jy = p.target_y + math.cos(t * 3 + p.phase * 1.5) * jitter + canvas.set_pixel(int(jx), int(jy)) + canvas.set_pixel(int(p.target_x), int(p.target_y)) + rgb = (255, 200, 80) + + live.update(render_colored(canvas, rgb)) + time.sleep(1.0 / fps) + + canvas.clear() + for p in particles: + canvas.set_pixel(int(p.target_x), int(p.target_y)) + console.print(render_colored(canvas, (255, 200, 80))) + + +def main() -> int: + parser = argparse.ArgumentParser(description="Standalone terminal particle logo.") + parser.add_argument("--hold-seconds", type=float, default=1.5) + parser.add_argument("--fps", type=int, default=24) + parser.add_argument("--max-particles", type=int, default=1500) + parser.add_argument("--ambient-count", type=int, default=200) + parser.add_argument("--scale", type=int, default=2) + parser.add_argument( + "--no-color", action="store_true", help="Disable color if desired." + ) + args = parser.parse_args() + + console = Console(color_system=None if args.no_color else "auto") + + try: + run_particle_logo( + console=console, + hold_seconds=args.hold_seconds, + fps=args.fps, + max_particles=args.max_particles, + ambient_count=args.ambient_count, + scale=args.scale, + ) + except KeyboardInterrupt: + console.print("\n[dim]interrupted[/dim]") + return 130 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/tests.py b/scripts/tests.py new file mode 100644 index 0000000..00e26e8 --- /dev/null +++ b/scripts/tests.py @@ -0,0 +1,394 @@ +"""Particle coalesce effect for the HUGGING FACE ML INTERN logo. + +Random particles swirl in from the edges, converge to form the text +"HUGGING FACE / ML INTERN", hold briefly, then the final frame is printed. +Rendered with braille characters for high detail. + +Based on Leandro's particle_coalesce.py demo. +""" + +import math +import random +import time + +from rich.align import Align +from rich.console import Console +from rich.live import Live +from rich.text import Text + +"""Braille-character canvas for high-resolution terminal graphics. + +Each terminal cell maps to a 2x4 dot grid using Unicode braille characters +(U+2800–U+28FF), giving 2× horizontal and 4× vertical resolution. +""" + + +def settle_curve(progress: float, sharpness: float = 3.0) -> float: + """Return noise amount in range 1..0 for normalized progress 0..1.""" + t = max(0.0, min(1.0, progress)) + return math.exp(-sharpness * t) + + +def warm_gold_from_white(progress: float) -> tuple[int, int, int]: + """Interpolate from white to warm gold for progress 0..1.""" + t = max(0.0, min(1.0, progress)) + return 255, int(255 - 55 * t), int(255 - 175 * t) + + +# Braille dot positions: (0,0) (1,0) dots 1,4 +# (0,1) (1,1) dots 2,5 +# (0,2) (1,2) dots 3,6 +# (0,3) (1,3) dots 7,8 +_DOT_MAP = ( + (0x01, 0x08), + (0x02, 0x10), + (0x04, 0x20), + (0x40, 0x80), +) + + +class BrailleCanvas: + """A pixel canvas that renders to braille characters.""" + + def __init__(self, term_width: int, term_height: int): + self.term_width = term_width + self.term_height = term_height + self.pixel_width = term_width * 2 + self.pixel_height = term_height * 4 + self._buf = bytearray(term_width * term_height) + + def clear(self) -> None: + for i in range(len(self._buf)): + self._buf[i] = 0 + + def set_pixel(self, x: int, y: int) -> None: + if 0 <= x < self.pixel_width and 0 <= y < self.pixel_height: + cx, rx = divmod(x, 2) + cy, ry = divmod(y, 4) + self._buf[cy * self.term_width + cx] |= _DOT_MAP[ry][rx] + + def render(self) -> list[str]: + lines = [] + for row in range(self.term_height): + offset = row * self.term_width + line = "".join( + chr(0x2800 + self._buf[offset + col]) for col in range(self.term_width) + ) + lines.append(line) + return lines + + +# ── Bitmap font (5×7 uppercase + digits) ────────────────────────────── + +_FONT: dict[str, list[str]] = {} + + +def _define_font() -> None: + """Define a simple 5×7 bitmap font for uppercase ASCII.""" + glyphs = { + "A": [" ## ", "# #", "# #", "####", "# #", "# #", "# #"], + "B": ["### ", "# #", "# #", "### ", "# #", "# #", "### "], + "C": [" ## ", "# #", "# ", "# ", "# ", "# #", " ## "], + "D": ["### ", "# #", "# #", "# #", "# #", "# #", "### "], + "E": ["####", "# ", "# ", "### ", "# ", "# ", "####"], + "F": ["####", "# ", "# ", "### ", "# ", "# ", "# "], + "G": [" ## ", "# #", "# ", "# ##", "# #", "# #", " ###"], + "H": ["# #", "# #", "# #", "####", "# #", "# #", "# #"], + "I": ["###", " # ", " # ", " # ", " # ", " # ", "###"], + "J": [" ##", " # ", " # ", " # ", " # ", "# # ", " # "], + "K": ["# #", "# # ", "## ", "## ", "# # ", "# #", "# #"], + "L": ["# ", "# ", "# ", "# ", "# ", "# ", "####"], + "M": ["# #", "## ##", "# # #", "# # #", "# #", "# #", "# #"], + "N": ["# #", "## #", "## #", "# ##", "# ##", "# #", "# #"], + "O": [" ## ", "# #", "# #", "# #", "# #", "# #", " ## "], + "P": ["### ", "# #", "# #", "### ", "# ", "# ", "# "], + "Q": [" ## ", "# #", "# #", "# #", "# ##", "# #", " ## "], + "R": ["### ", "# #", "# #", "### ", "# # ", "# #", "# #"], + "S": [" ## ", "# #", "# ", " ## ", " #", "# #", " ## "], + "T": ["#####", " # ", " # ", " # ", " # ", " # ", " # "], + "U": ["# #", "# #", "# #", "# #", "# #", "# #", " ## "], + "V": ["# #", "# #", "# #", " # # ", " # # ", " # ", " # "], + "W": ["# #", "# #", "# #", "# # #", "# # #", "## ##", "# #"], + "X": ["# #", "# #", " ## ", " ## ", " ## ", "# #", "# #"], + "Y": ["# #", "# #", " # # ", " # ", " # ", " # ", " # "], + "Z": ["####", " #", " # ", " # ", "# ", "# ", "####"], + " ": [" ", " ", " ", " ", " ", " ", " "], + "0": [" ## ", "# #", "# #", "# #", "# #", "# #", " ## "], + "1": [" # ", "## ", " # ", " # ", " # ", " # ", "###"], + "2": [" ## ", "# #", " #", " # ", " # ", "# ", "####"], + "3": [" ## ", "# #", " #", " ## ", " #", "# #", " ## "], + "4": ["# #", "# #", "# #", "####", " #", " #", " #"], + "5": ["####", "# ", "### ", " #", " #", "# #", " ## "], + "6": [" ## ", "# ", "### ", "# #", "# #", "# #", " ## "], + "7": ["####", " #", " # ", " # ", " # ", " # ", " # "], + "8": [" ## ", "# #", "# #", " ## ", "# #", "# #", " ## "], + "9": [" ## ", "# #", "# #", " ###", " #", " #", " ## "], + } + _FONT.update(glyphs) + + +_define_font() + + +def text_to_pixels(text: str, scale: int = 1) -> list[tuple[int, int]]: + """Convert text string to a list of (x, y) pixel positions using bitmap font.""" + pixels = [] + cursor_x = 0 + for ch in text.upper(): + glyph = _FONT.get(ch) + if glyph is None: + cursor_x += 4 * scale + continue + for row_idx, row in enumerate(glyph): + for col_idx, cell in enumerate(row): + if cell == "#": + for sy in range(scale): + for sx in range(scale): + pixels.append( + (cursor_x + col_idx * scale + sx, row_idx * scale + sy) + ) + glyph_width = max(len(r) for r in glyph) + cursor_x += (glyph_width + 1) * scale + return pixels + + +class Particle: + __slots__ = ("x", "y", "target_x", "target_y", "vx", "vy", "phase", "delay") + + def __init__( + self, x: float, y: float, target_x: float, target_y: float, delay: float = 0 + ): + self.x = x + self.y = y + self.target_x = target_x + self.target_y = target_y + self.vx = 0.0 + self.vy = 0.0 + self.phase = random.uniform(0, math.pi * 2) + self.delay = delay + + def update_converge(self, t: float, strength: float = 0.08, damping: float = 0.92): + """Move toward target with spring-like physics.""" + if t < self.delay: + # Still in swirl phase + self.x += self.vx + self.y += self.vy + self.vx *= 0.99 + self.vy *= 0.99 + # Gentle spiral + angle = self.phase + t * 2 + self.vx += math.cos(angle) * 0.3 + self.vy += math.sin(angle) * 0.3 + return + + # Spring toward target + dx = self.target_x - self.x + dy = self.target_y - self.y + self.vx += dx * strength + self.vy += dy * strength + self.vx *= damping + self.vy *= damping + self.x += self.vx + self.y += self.vy + + @property + def at_target(self) -> bool: + return abs(self.x - self.target_x) < 1.5 and abs(self.y - self.target_y) < 1.5 + + +def run_particle_logo(console: Console, hold_seconds: float = 1.5) -> None: + """Run the particle coalesce effect.""" + term_width = min(console.width, 120) + term_height = min(console.height - 4, 35) + + canvas = BrailleCanvas(term_width, term_height) + + # Get target positions from text + text_pixels_line1 = text_to_pixels("HUGGING FACE", scale=2) + text_pixels_line2 = text_to_pixels("ML INTERN", scale=2) + + # Calculate dimensions for centering + def get_bounds(pixels): + if not pixels: + return 0, 0, 0, 0 + xs = [p[0] for p in pixels] + ys = [p[1] for p in pixels] + return min(xs), max(xs), min(ys), max(ys) + + min_x1, max_x1, min_y1, max_y1 = get_bounds(text_pixels_line1) + min_x2, max_x2, min_y2, max_y2 = get_bounds(text_pixels_line2) + + w1, h1 = max_x1 - min_x1 + 1, max_y1 - min_y1 + 1 + w2, h2 = max_x2 - min_x2 + 1, max_y2 - min_y2 + 1 + + total_h = h1 + 6 + h2 # gap between lines + start_y = (canvas.pixel_height - total_h) // 2 + + # Center line 1 + offset_x1 = (canvas.pixel_width - w1) // 2 - min_x1 + offset_y1 = start_y - min_y1 + targets_1 = [(p[0] + offset_x1, p[1] + offset_y1) for p in text_pixels_line1] + + # Center line 2 + offset_x2 = (canvas.pixel_width - w2) // 2 - min_x2 + offset_y2 = start_y + h1 + 6 - min_y2 + targets_2 = [(p[0] + offset_x2, p[1] + offset_y2) for p in text_pixels_line2] + + all_targets = targets_1 + targets_2 + + # Subsample for performance — take every Nth pixel + step = max(1, len(all_targets) // 1500) + sampled_targets = all_targets[::step] + + # Create particles at random edge positions + rng = random.Random(42) + particles = [] + pw, ph = canvas.pixel_width, canvas.pixel_height + + for i, (tx, ty) in enumerate(sampled_targets): + # Spawn from random edge + side = rng.choice(["top", "bottom", "left", "right"]) + if side == "top": + sx, sy = rng.uniform(0, pw), rng.uniform(-20, -5) + elif side == "bottom": + sx, sy = rng.uniform(0, pw), rng.uniform(ph + 5, ph + 20) + elif side == "left": + sx, sy = rng.uniform(-20, -5), rng.uniform(0, ph) + else: + sx, sy = rng.uniform(pw + 5, pw + 20), rng.uniform(0, ph) + + delay = rng.uniform(0, 0.4) # staggered start + p = Particle(sx, sy, tx, ty, delay=delay) + # Initial velocity — gentle swirl + angle = math.atan2(ph / 2 - sy, pw / 2 - sx) + rng.gauss(0, 0.8) + speed = rng.uniform(1.0, 2.5) + p.vx = math.cos(angle) * speed + p.vy = math.sin(angle) * speed + particles.append(p) + + # Also add some extra ambient particles that never converge + ambient = [] + for _ in range(200): + ax = rng.uniform(0, pw) + ay = rng.uniform(0, ph) + ap = Particle(ax, ay, ax, ay) + ap.vx = rng.gauss(0, 1) + ap.vy = rng.gauss(0, 1) + ambient.append(ap) + + # Timing: 1s converge + 2s hold = 3s total + fps = 24 + converge_frames = int(fps * 0.9) + hold_frames = int(fps * hold_seconds) + total_frames = converge_frames + hold_frames + + with Live(console=console, refresh_per_second=fps, transient=True) as live: + for frame in range(total_frames): + canvas.clear() + t = frame * 0.03 + + # Update ambient particles (always drifting) + for ap in ambient: + ap.x += ap.vx + math.sin(t + ap.phase) * 0.5 + ap.y += ap.vy + math.cos(t + ap.phase * 1.3) * 0.5 + # Wrap around + ap.x = ap.x % pw + ap.y = ap.y % ph + + # Fade out ambient during hold phase + if frame < converge_frames: + alpha = 0.3 + 0.2 * math.sin(t * 2 + ap.phase) + else: + fade = (frame - converge_frames) / hold_frames + alpha = (0.3 + 0.2 * math.sin(t * 2 + ap.phase)) * (1 - fade) + if alpha > 0.25: + canvas.set_pixel(int(ap.x), int(ap.y)) + + if frame < converge_frames: + # Converge phase + progress = frame / converge_frames + noise = settle_curve(progress) + for p in particles: + p.update_converge(t, strength=0.06, damping=0.90) + canvas.set_pixel(int(p.x), int(p.y)) + + # Trail effect + trail_scale = 0.2 + 0.5 * noise + trail_x = int(p.x - p.vx * trail_scale) + trail_y = int(p.y - p.vy * trail_scale) + canvas.set_pixel(trail_x, trail_y) + + # Color transitions from white to warm gold + r, g, b = warm_gold_from_white(progress) + else: + # Hold phase — settle into solid logo + settle_t = (frame - converge_frames) / hold_frames + for p in particles: + # Jitter decays to zero + jitter = (1 - settle_t) * 0.7 + jx = p.target_x + math.sin(t * 3 + p.phase) * jitter + jy = p.target_y + math.cos(t * 3 + p.phase * 1.5) * jitter + canvas.set_pixel(int(jx), int(jy)) + canvas.set_pixel(int(p.target_x), int(p.target_y)) + + r, g, b = 255, 200, 80 + + # Render with color + lines = canvas.render() + result = Text() + for line in lines: + for ch in line: + if ch == chr(0x2800): + result.append(ch) + else: + result.append(ch, style=f"rgb({r},{g},{b})") + result.append("\n") + + live.update(Align.center(result)) + time.sleep(1.0 / fps) + + # Print final settled frame + canvas.clear() + for p in particles: + canvas.set_pixel(int(p.target_x), int(p.target_y)) + final = Text() + for line in canvas.render(): + for ch in line: + if ch == chr(0x2800): + final.append(ch) + else: + final.append(ch, style="rgb(255,200,80)") + final.append("\n") + console.print(Align.center(final)) + + +import argparse + + +def main() -> int: + parser = argparse.ArgumentParser(description="Standalone terminal particle logo.") + parser.add_argument("--hold-seconds", type=float, default=1.5) + parser.add_argument("--fps", type=int, default=24) + parser.add_argument("--max-particles", type=int, default=1500) + parser.add_argument("--ambient-count", type=int, default=200) + parser.add_argument("--scale", type=int, default=2) + parser.add_argument( + "--no-color", action="store_true", help="Disable color if desired." + ) + args = parser.parse_args() + + console = Console(color_system=None if args.no_color else "auto") + + try: + run_particle_logo( + console=console, + hold_seconds=args.hold_seconds, + ) + except KeyboardInterrupt: + console.print("\n[dim]interrupted[/dim]") + return 130 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/repoman/cli/commands/compliance/__init__.py b/src/repoman/cli/commands/compliance/__init__.py index 951ff38..b480ee7 100644 --- a/src/repoman/cli/commands/compliance/__init__.py +++ b/src/repoman/cli/commands/compliance/__init__.py @@ -30,6 +30,7 @@ app = Typer( add_completion=True, + no_args_is_help=True, help="""Analyze a git repository against built-in compliance profiles. The v0 feature is a readiness check, not a certification claim. @@ -59,7 +60,12 @@ def check( ] = "table", output: Annotated[ Path | None, - Option("--output", "-o", path_type=Path, help="Output file or directory for non-table formats"), + Option( + "--output", + "-o", + path_type=Path, + help="Output file or directory for non-table formats", + ), ] = None, compliance_file: Annotated[ Path | None, @@ -134,7 +140,12 @@ def check( def init_( path: Annotated[ Path, - Option("--path", "-p", path_type=Path, help="Repository path where the starter file should be written"), + Option( + "--path", + "-p", + path_type=Path, + help="Repository path where the starter file should be written", + ), ] = Path("."), profile: Annotated[ list[str] | None, diff --git a/src/repoman/cli/main_cli.py b/src/repoman/cli/main_cli.py index da2ab76..6a70aec 100644 --- a/src/repoman/cli/main_cli.py +++ b/src/repoman/cli/main_cli.py @@ -31,7 +31,6 @@ from pathlib import Path from pydantic import ValidationError -from rich.console import Console from rich.text import Text from typer import Context, Exit, Option, Typer @@ -39,7 +38,8 @@ from repoman.cli.register_commands import _register_commands from repoman.config import Config from repoman.utils.logging import get_logger_console -from repoman.utils.theme.theme import set_theme +from repoman.utils.ui import get_console +from repoman.utils.ui.logo_display import run_particle_logo cli_app = Typer(add_completion=True, invoke_without_command=True, no_args_is_help=True) @@ -57,7 +57,7 @@ def _version_callback(value: bool) -> None: Whether to print version information """ if value: - console = Console(theme=set_theme("dark")) + console = get_console() console.print( version_info(), ) @@ -73,7 +73,7 @@ def _debug_info_callback(value: bool) -> None: Whether to print debug information """ if value: - console = Console(theme=set_theme("dark")) + console = get_console() debug_info(console) raise Exit(0) @@ -118,8 +118,14 @@ def main( repoman extensions sync ./my-app repoman generator add my-command --project-dir ./my-app """ - logger, _console = get_logger_console() - + logger, console = get_logger_console() + run_particle_logo( + console=console, + hold_seconds=1.5, + ) + # Clear screen for CRT boot — starts from top + console.file.write("\033[2J\033[H") + console.file.flush() config: Config | None = None try: config = Config.load(custom_path=Path(config_path) if config_path else None) diff --git a/src/repoman/main_template/src/{{python_package_import_name}}/utils/theme/terminal_colors.py b/src/repoman/main_template/src/{{python_package_import_name}}/utils/theme/terminal_colors.py deleted file mode 100644 index d48f380..0000000 --- a/src/repoman/main_template/src/{{python_package_import_name}}/utils/theme/terminal_colors.py +++ /dev/null @@ -1,14 +0,0 @@ -"""Terminal color utilities for Rich visualization.""" - - -def get_rich_color(label: str) -> str: - """Map class labels to Catppuccin colors for Rich visualization.""" - color_map = { - "TP": "green", - "FP": "maroon", - "FN": "red", - "TN": "peach", - "header": "subtext1", - "border": "overlay1", - } - return color_map.get(label, "text") diff --git a/src/repoman/main_template/tests/test_utils/test_theme.py.jinja b/src/repoman/main_template/tests/test_utils/test_theme.py.jinja index cc42234..de1d32a 100644 --- a/src/repoman/main_template/tests/test_utils/test_theme.py.jinja +++ b/src/repoman/main_template/tests/test_utils/test_theme.py.jinja @@ -3,7 +3,6 @@ import pytest from rich.theme import Theme -from {{ python_package_import_name }}.utils.theme.terminal_colors import get_rich_color from {{ python_package_import_name }}.utils.theme.theme import _create_theme, set_theme @@ -87,24 +86,3 @@ class TestSetTheme: assert theme_dark.styles != theme_light.styles -class TestTerminalColors: - """Test terminal color mapping functionality.""" - - def test_get_rich_color_known_labels(self) -> None: - """Test that known labels return correct colors.""" - assert get_rich_color("TP") == "green" - assert get_rich_color("FP") == "maroon" - assert get_rich_color("FN") == "red" - assert get_rich_color("TN") == "peach" - assert get_rich_color("header") == "subtext1" - assert get_rich_color("border") == "overlay1" - - def test_get_rich_color_unknown_label(self) -> None: - """Test that unknown labels return default 'text' color.""" - assert get_rich_color("unknown") == "text" - assert get_rich_color("") == "text" - - def test_get_rich_color_case_sensitive(self) -> None: - """Test that color mapping is case sensitive.""" - assert get_rich_color("tp") == "text" - assert get_rich_color("TP") == "green" diff --git a/src/repoman/utils/task_tree_display.py b/src/repoman/utils/task_tree_display.py new file mode 100644 index 0000000..393f6fb --- /dev/null +++ b/src/repoman/utils/task_tree_display.py @@ -0,0 +1,197 @@ +from collections.abc import Callable +from datetime import datetime +from typing import Any + +from rich.text import Text +from rich.tree import Tree + + +class TaskTree: + """A ``rich.tree.Tree`` wrapper that displays a command with its stdout and stderr. + + Renders a task as a hidden-root tree with the command label on top and optional + ``stdout`` and ``stderr`` branches beneath it. Output branches are lazily created + the first time content is set and are ordered so that ``stdout`` always appears + before ``stderr`` regardless of which stream produced output first. + + Attributes: + tree: The underlying ``rich.tree.Tree`` instance to be rendered. + cmd: The command string displayed as the first (non-root) node. + """ + + _stdout_text: Text | None + _stdout_str: list[str] + _stderr_text: Text | None + _stderr_str: list[str] + + def __init__(self, cmd: str) -> None: + """Initialize the tree with a command label and no output branches. + + Args: + cmd: The command string to display at the top of the tree. + """ + self.tree = Tree("", hide_root=True) + self.cmd = cmd + self.tree.add(cmd) + self._stdout_text = None + self._stderr_text = None + self._stdout_str = [] + self._stderr_str = [] + + def set_stdout(self, stdout: list[str]): + """Set or replace the lines shown under the ``stdout`` branch. + + Creates the ``stdout`` branch on first non-empty call and reorders the tree + so that ``stdout`` is rendered above ``stderr`` if ``stderr`` was added first. + + Args: + stdout: Lines of standard output to display. No-op if empty. + """ + if not stdout: + return + + if not self._stdout_text: + self._stdout_text = Text("", style="dim") + stdout_branch = self.tree.add("stdout") + stdout_branch.add(self._stdout_text) + + if self._stderr_text: + # Swap the children so that stdout is always first + # In this case, there will be 3 things in the tree. + # 0 - the cmd from init + # 1 - the stderr branch since it was apparently added first + # 2 - the stdout branch we just added + self.tree.children = [self.tree.children[2], self.tree.children[1]] + + self._stdout_str = stdout + self._stdout_text.plain = "\n".join(stdout) + + def set_stderr(self, stderr: list[str]): + """Set or replace the lines shown under the ``stderr`` branch. + + Creates the ``stderr`` branch on first non-empty call, styled in red. + + Args: + stderr: Lines of standard error to display. No-op if empty. + """ + if not stderr: + return + + if not self._stderr_text: + self._stderr_text = Text("", style="red") + stderr_branch = self.tree.add("stderr") + stderr_branch.add(self._stderr_text) + self._stderr_str = stderr + + self._stderr_str = stderr + self._stderr_text.plain = "\n".join(stderr) + + def reset(self): + """Clear stdout/stderr branches and restore the tree to its initial state. + + After this call the tree contains only the original command label, and + subsequent ``set_stdout``/``set_stderr`` calls will recreate the branches. + """ + self._stdout_text = None + self._stderr_text = None + self._stdout_str = [] + self._stderr_str = [] + self.tree.children = [] + self.tree.add(self.cmd) + + def height(self) -> int: + """Return the number of rendered lines the tree currently occupies. + + Accounts for the command line plus one header line per populated output + branch and one line per output line. + + Returns: + The total rendered height in lines. + """ + stdout_height = len(self._stdout_str) + 1 if self._stdout_str else 0 + stderr_height = len(self._stderr_str) + 1 if self._stderr_str else 0 + cmd_height = 1 + + return cmd_height + stdout_height + stderr_height + + +class Padder: + """Tracks the tallest ``TaskTree`` seen and reports bottom padding for others. + + Useful when rendering multiple ``TaskTree`` instances in a table row/column where + each cell should be vertically padded to match the tallest tree so the layout + does not collapse as trees grow. + """ + + def __init__(self) -> None: + """Initialize the padder with a max observed height of zero.""" + self._max_padding = 0 + + def get_padding(self, tree: TaskTree) -> int: + """Update the max observed height and return padding for ``tree``. + + Args: + tree: The tree whose height should be compared against the running max. + + Returns: + The number of blank lines to append to ``tree`` so that its rendered + height matches the tallest tree observed so far. + """ + height = tree.height() + self._max_padding = max(self._max_padding, height) + return self._max_padding - height + + +class UpdateTracker: + """ + Class to enable dynamic updates on the UI tables. By default, rich allows you to set a refresh rate or trigger manual + updates. This makes manual updates more performant by doing quick 'dirty' checks to determine if updating ins required. Updating + is technically always required because the table's 'elapsed time' column always changes, but we don't want to update the table just + because of that. + + This class tracks some state and has a min/max ms time config to keep the table looking responsive without updating too often. The + driver for this was my CPU usage while the installer was running, paired with the size of the asciinema files that were generated + because of frequent updates. Obviously, the more frequent the update the better the table looks, but that's the trade off. + + """ + + def __init__(self, max_update_timeout_ms: float = 5000, min_update_ms: float = 200) -> None: + """ + Args: + max_update_timeout_ms: The maximum amount of time in milliseconds that can pass before an update is forced. This is useful + because the table usually contains an 'elapsed time' column that should update fairly frequently regardless of everything else. + min_update_ms: The minimum amount of time in milliseconds that must pass before an update is allowed. This prevents updates + from getting too frequent. + """ + self._timeout_ms = max_update_timeout_ms + self._min_update_ms = min_update_ms + self._last_update_timestamp_ms = datetime.now().timestamp() * 1000 + self._last_update_state: list[Any] = [] + + def max_update_time_passed(self, now: float) -> bool: + if now - self._last_update_timestamp_ms > self._timeout_ms: + return True + return False + + def min_update_time_passed(self, now: float) -> bool: + if now - self._last_update_timestamp_ms > self._min_update_ms: + return True + return False + + def _update( + self, + now: float, + update_fn: Callable[[], None], + state: list[str | list[str]], + ): + update_fn() + self._last_update_timestamp_ms = now + self._last_update_state = state + + def update(self, update_fn: Callable[[], None], state: list[str | list[str]]): + now = datetime.now().timestamp() * 1000 + + if self.min_update_time_passed(now) and state != self._last_update_state: + self._update(now, update_fn, state) + elif self.max_update_time_passed(now): + self._update(now, update_fn, state) diff --git a/src/repoman/utils/ui/display_tasks.py b/src/repoman/utils/ui/display_tasks.py new file mode 100644 index 0000000..b07c655 --- /dev/null +++ b/src/repoman/utils/ui/display_tasks.py @@ -0,0 +1,509 @@ +""" +Terminal display utilities — rich-powered CLI formatting. +""" + +import re + +from rich.console import Console +from rich.markdown import Heading, Markdown +from rich.panel import Panel + + +class _LeftHeading(Heading): + """Rich's default Markdown renders h1/h2 centered via Align.center. + Yield the styled text directly so headings stay left-aligned.""" + + def __rich_console__(self, console, options): + self.text.justify = "left" + yield self.text + + +Markdown.elements["heading_open"] = _LeftHeading + + +_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]") + + +def _clip_to_width(s: str, width: int) -> str: + """Truncate a string to `width` visible columns, preserving ANSI styles. + + Needed for the sub-agent live redraw: cursor-up-and-erase assumes one + logical line == one terminal row. If a line wraps, cursor-up undershoots + and the next redraw corrupts the display. Truncating prevents wrap. + """ + if width <= 0: + return s + out: list[str] = [] + visible = 0 + i = 0 + # Reserve 1 char for the trailing ellipsis + limit = width - 1 + truncated = False + while i < len(s): + m = _ANSI_RE.match(s, i) + if m: + out.append(m.group()) + i = m.end() + continue + if visible >= limit: + truncated = True + break + out.append(s[i]) + visible += 1 + i += 1 + if truncated: + # Strip styles (so ellipsis isn't left hanging inside a style run) + out.append("\033[0m…") + return "".join(out) + + +# Indent prefix for all agent output (aligns under the `>` prompt) +_I = " " + +# ── Banner ───────────────────────────────────────────────────────────── + + +def print_banner(model: str | None = None, hf_user: str | None = None) -> None: + """Print particle logo then CRT boot sequence with system info.""" + from agent.utils.crt_boot import run_boot_sequence + from agent.utils.particle_logo import run_particle_logo + + # Particle coalesce logo — 1.5s converge, 2s hold + run_particle_logo(_console, hold_seconds=2.0) + + # Clear screen for CRT boot — starts from top + _console.file.write("\033[2J\033[H") + _console.file.flush() + + model_label = model or "bedrock/us.anthropic.claude-opus-4-6-v1" + user_label = hf_user or "not logged in" + + # Warm gold palette matching the shimmer highlight (255, 200, 80) + gold = "rgb(255,200,80)" + dim_gold = "rgb(180,140,40)" + + boot_lines = [ + (f"{_I}Initializing agent runtime...", gold), + (f"{_I} User: {user_label}", dim_gold), + (f"{_I} Model: {model_label}", dim_gold), + (f"{_I} Tools: loading...", dim_gold), + ("", ""), + (f"{_I}/help for commands · /model to switch · /quit to exit", gold), + ] + + run_boot_sequence(_console, boot_lines) + + +# ── Init progress ────────────────────────────────────────────────────── + + +def print_init_done(tool_count: int = 0) -> None: + import time + + f = _console.file + # Overwrite the "Tools: loading..." line with actual count + f.write("\033[A\033[A\033[A\033[K") # Move up 3 lines (blank + help + blank) then up to tools line + f.write("\033[A\033[K") + gold = "\033[38;2;180;140;40m" + reset = "\033[0m" + tool_text = f"{_I} Tools: {tool_count} loaded" + for ch in tool_text: + f.write(f"{gold}{ch}{reset}") + f.flush() + time.sleep(0.012) + f.write("\n\n") + # Reprint the help line + f.write(f"{_I}\033[38;2;255;200;80m/help for commands · /model to switch · /quit to exit{reset}\n\n") + # Ready message — minimal padding + f.write(f"{_I}\033[38;2;255;200;80mReady. Let's build something impressive.{reset}\n") + f.flush() + + +# ── Tool calls ───────────────────────────────────────────────────────── + + +def print_tool_call(tool_name: str, args_preview: str) -> None: + import time + + f = _console.file + # CRT-style: type out tool name in HF yellow + gold = "\033[38;2;255;200;80m" + reset = "\033[0m" + f.write(f"{_I}{gold}▸ ") + for ch in tool_name: + f.write(ch) + f.flush() + time.sleep(0.015) + f.write(f"{reset} \033[2m{args_preview}{reset}\n") + f.flush() + + +def print_tool_output(output: str, success: bool, truncate: bool = True) -> None: + if truncate: + output = _truncate(output, max_lines=10) + style = "tool.ok" if success else "tool.fail" + # Indent each line of tool output + indented = "\n".join(f"{_I} {line}" for line in output.split("\n")) + _console.print(f"[{style}]{indented}[/{style}]") + + +class SubAgentDisplayManager: + """Manages multiple concurrent sub-agent displays. + + Each agent gets its own stats and rolling tool-call log. + All agents are rendered together so terminal escape-code + erase/redraw stays consistent. + """ + + _MAX_VISIBLE = 4 # tool-call lines shown per agent + + def __init__(self): + self._agents: dict[str, dict] = {} # agent_id -> state dict + self._lines_on_screen = 0 + self._ticker_task = None + + def start(self, agent_id: str, label: str = "research") -> None: + import asyncio + import time + + self._agents[agent_id] = { + "label": label, + "calls": [], + "tool_count": 0, + "token_count": 0, + "start_time": time.monotonic(), + } + if not self._ticker_task: + self._ticker_task = asyncio.ensure_future(self._tick()) + self._redraw() + + def set_tokens(self, agent_id: str, tokens: int) -> None: + if agent_id in self._agents: + self._agents[agent_id]["token_count"] = tokens + + def set_tool_count(self, agent_id: str, count: int) -> None: + if agent_id in self._agents: + self._agents[agent_id]["tool_count"] = count + + def add_call(self, agent_id: str, tool_desc: str) -> None: + if agent_id in self._agents: + self._agents[agent_id]["calls"].append(tool_desc) + self._redraw() + + def clear(self, agent_id: str) -> None: + # On completion: erase the live region, freeze a single-line summary + # for this agent ("✓ research: … (stats)") above the live region so + # the user sees each sub-agent finish cleanly without the tool-call + # noise, then redraw remaining live agents. + agent = self._agents.pop(agent_id, None) + self._erase() + if agent is not None: + width = max(10, _console.width) + line = _clip_to_width(self._render_completion_line(agent), width) + _console.file.write(line + "\n") + _console.file.flush() + self._lines_on_screen = 0 + if not self._agents: + if self._ticker_task: + self._ticker_task.cancel() + self._ticker_task = None + else: + self._redraw() + + @staticmethod + def _render_completion_line(agent: dict) -> str: + stats = SubAgentDisplayManager._format_stats(agent) + label = agent["label"] + # dim green check + dim label; stats in parens + line = f"{_I}\033[38;2;120;200;140m✓\033[0m \033[2m{label}\033[0m" + if stats: + line += f" \033[2m({stats})\033[0m" + return line + + async def _tick(self) -> None: + import asyncio + + try: + while True: + await asyncio.sleep(1.0) + if self._agents: + self._redraw() + except asyncio.CancelledError: + pass + + @staticmethod + def _format_stats(agent: dict) -> str: + import time + + start = agent["start_time"] + if start is None: + return "" + elapsed = time.monotonic() - start + if elapsed < 60: + time_str = f"{elapsed:.0f}s" + else: + time_str = f"{elapsed / 60:.0f}m {elapsed % 60:.0f}s" + tok = agent["token_count"] + tok_str = f"{tok / 1000:.1f}k" if tok >= 1000 else str(tok) + return f"{agent['tool_count']} tool uses · {tok_str} tokens · {time_str}" + + def _erase(self) -> None: + if self._lines_on_screen > 0: + f = _console.file + for _ in range(self._lines_on_screen): + f.write("\033[A\033[K") + f.flush() + + def _render_agent_lines(self, agent: dict, compact: bool = False) -> list[str]: + """Render one agent's block. + + compact=True → single line (label + stats + most-recent tool name); + compact=False → header + up to _MAX_VISIBLE rolling tool-call lines. + We use compact mode when multiple agents are live so the total live + region stays small enough to fit on one screen. Otherwise cursor-up + can't reach lines that have scrolled into scrollback, and every + redraw pollutes history with a stale copy. + """ + stats = self._format_stats(agent) + label = agent["label"] + header = f"{_I}\033[38;2;255;200;80m▸ {label}\033[0m" + if stats: + header += f" \033[2m({stats})\033[0m" + if compact: + latest = agent["calls"][-1] if agent["calls"] else "" + if latest: + # Strip long json tails for the inline view + short = latest.split(" ")[0] if " " in latest else latest + header += f" \033[2m·\033[0m \033[2m{short}\033[0m" + return [header] + lines = [header] + visible = agent["calls"][-self._MAX_VISIBLE :] + for desc in visible: + lines.append(f"{_I} \033[2m{desc}\033[0m") + return lines + + def _redraw(self) -> None: + f = _console.file + self._erase() + compact = len(self._agents) > 1 + width = max(10, _console.width) + lines: list[str] = [] + for agent in self._agents.values(): + for ln in self._render_agent_lines(agent, compact=compact): + lines.append(_clip_to_width(ln, width)) + for line in lines: + f.write(line + "\n") + f.flush() + self._lines_on_screen = len(lines) + + +_subagent_display = SubAgentDisplayManager() + + +def print_tool_log(tool: str, log: str, agent_id: str = "", label: str = "") -> None: + """Handle tool log events — sub-agent calls get the rolling display.""" + if tool == "research": + aid = agent_id or "research" + if log == "Starting research sub-agent...": + _subagent_display.start(aid, label or "research") + elif log == "Research complete.": + _subagent_display.clear(aid) + elif log.startswith("tokens:"): + _subagent_display.set_tokens(aid, int(log[7:])) + elif log.startswith("tools:"): + _subagent_display.set_tool_count(aid, int(log[6:])) + else: + _subagent_display.add_call(aid, log) + else: + _console.print(f"{_I}[dim]{tool}: {log}[/dim]") + + +# ── Messages ─────────────────────────────────────────────────────────── + + +async def print_markdown( + text: str, + cancel_event: "asyncio.Event | None" = None, + instant: bool = False, +) -> None: + import asyncio + import io + import random + + from rich.padding import Padding + + _console.print() + + # Render markdown to a string buffer so we can type it out + buf = io.StringIO() + # Important: StringIO is not a TTY, so Rich would normally strip styles. + # Force terminal rendering so ANSI style codes are preserved for typewriter output. + buf_console = Console( + file=buf, + width=_console.width, + highlight=False, + theme=_THEME, + force_terminal=True, + color_system=_console.color_system or "truecolor", + ) + buf_console.print(Padding(Markdown(text), (0, 0, 0, 2))) + rendered = buf.getvalue() + + # Strip trailing whitespace from each line so we don't type across the full width + lines = rendered.split("\n") + rendered = "\n".join(line.rstrip() for line in lines) + + f = _console.file + + # Headless / non-interactive: dump the rendered markdown in one write. + if instant: + f.write(rendered) + f.write("\n") + f.flush() + return + + # CRT typewriter effect — async so the event loop can service signal + # handlers (Ctrl+C during streaming) between characters. If cancelled + # mid-type, stop cleanly: write an ANSI reset so half-open color state + # doesn't bleed onto the "interrupted" line, and return. + rng = random.Random(42) + cancelled = False + for ch in rendered: + if cancel_event is not None and cancel_event.is_set(): + cancelled = True + break + f.write(ch) + f.flush() + if ch == "\n": + await asyncio.sleep(0.002) + elif ch == " ": + await asyncio.sleep(0.002) + elif rng.random() < 0.03: + await asyncio.sleep(0.015) + else: + await asyncio.sleep(0.004) + f.write("\033[0m\n" if cancelled else "\n") + f.flush() + + +def print_error(message: str) -> None: + _console.print(f"\n{_I}[bold red]Error:[/bold red] {message}") + + +def print_turn_complete() -> None: + pass # no separator — clean output + + +def print_interrupted() -> None: + _console.print(f"\n{_I}[dim italic]interrupted[/dim italic]") + + +def print_compacted(old_tokens: int, new_tokens: int) -> None: + _console.print(f"{_I}[dim]context compacted: {old_tokens:,} → {new_tokens:,} tokens[/dim]") + + +# ── Approval ─────────────────────────────────────────────────────────── + + +def print_approval_header(count: int) -> None: + label = f"Approval required — {count} item{'s' if count != 1 else ''}" + _console.print() + _console.print( + f"{_I}", + Panel(f"[bold yellow]{label}[/bold yellow]", border_style="yellow", expand=False), + ) + + +def print_approval_item(index: int, total: int, tool_name: str, operation: str) -> None: + _console.print(f"\n{_I}[bold]\\[{index}/{total}][/bold] [tool.name]{tool_name}[/tool.name] {operation}") + + +def print_yolo_approve(count: int) -> None: + _console.print(f"{_I}[bold yellow]yolo →[/bold yellow] auto-approved {count} item(s)") + + +# ── Help ─────────────────────────────────────────────────────────────── + +HELP_TEXT = f"""\ +{_I}[bold]Commands[/bold] +{_I} [cyan]/help[/cyan] Show this help +{_I} [cyan]/undo[/cyan] Undo last turn +{_I} [cyan]/compact[/cyan] Compact context window +{_I} [cyan]/model[/cyan] [id] Show available models or switch +{_I} [cyan]/effort[/cyan] [level] Reasoning effort (minimal|low|medium|high|xhigh|max|off) +{_I} [cyan]/yolo[/cyan] Toggle auto-approve mode +{_I} [cyan]/status[/cyan] Current model & turn count +{_I} [cyan]/quit[/cyan] Exit""" + + +def print_help() -> None: + _console.print() + _console.print(HELP_TEXT) + _console.print() + + +# ── Plan display ─────────────────────────────────────────────────────── + + +def format_plan_display() -> str: + """Format the current plan for display.""" + from agent.tools.plan_tool import get_current_plan + + plan = get_current_plan() + if not plan: + return "" + + completed = [t for t in plan if t["status"] == "completed"] + in_progress = [t for t in plan if t["status"] == "in_progress"] + pending = [t for t in plan if t["status"] == "pending"] + + lines = [] + for t in completed: + lines.append(f"{_I}[green]✓[/green] [dim]{t['content']}[/dim]") + for t in in_progress: + lines.append(f"{_I}[yellow]▸[/yellow] {t['content']}") + for t in pending: + lines.append(f"{_I}[dim]○ {t['content']}[/dim]") + + summary = f"[dim]{len(completed)}/{len(plan)} done[/dim]" + lines.append(f"{_I}{summary}") + return "\n".join(lines) + + +def print_plan() -> None: + plan_str = format_plan_display() + if plan_str: + _console.print(plan_str) + + +# ── Formatting for plan_tool output (used by plan_tool handler) ──────── + + +def format_plan_tool_output(todos: list) -> str: + if not todos: + return "Plan is empty." + + lines = ["Plan updated:", ""] + completed = [t for t in todos if t["status"] == "completed"] + in_progress = [t for t in todos if t["status"] == "in_progress"] + pending = [t for t in todos if t["status"] == "pending"] + + for t in completed: + lines.append(f" [x] {t['id']}. {t['content']}") + for t in in_progress: + lines.append(f" [~] {t['id']}. {t['content']}") + for t in pending: + lines.append(f" [ ] {t['id']}. {t['content']}") + + lines.append(f"\n{len(completed)}/{len(todos)} done") + return "\n".join(lines) + + +# ── Internal helpers ─────────────────────────────────────────────────── + + +def _truncate(text: str, max_lines: int = 6) -> str: + lines = text.split("\n") + if len(lines) <= max_lines: + return text + return "\n".join(lines[:max_lines]) + f"\n... ({len(lines) - max_lines} more lines)" diff --git a/src/repoman/utils/ui/logo_display.py b/src/repoman/utils/ui/logo_display.py new file mode 100644 index 0000000..f86ee7d --- /dev/null +++ b/src/repoman/utils/ui/logo_display.py @@ -0,0 +1,373 @@ +"""Particle coalesce effect for the HUGGING FACE ML INTERN logo. + +Random particles swirl in from the edges, converge to form the text +"HUGGING FACE / ML INTERN", hold briefly, then the final frame is printed. +Rendered with braille characters for high detail. + +Based on Leandro's particle_coalesce.py demo. +""" + +import math +import random +import time + +from rich.align import Align +from rich.console import Console +from rich.live import Live +from rich.text import Text + +from repoman.utils.ui.theme.theme import cool_ramp, set_theme + +"""Braille-character canvas for high-resolution terminal graphics. + +Each terminal cell maps to a 2x4 dot grid using Unicode braille characters +(U+2800–U+28FF), giving 2× horizontal and 4× vertical resolution. +""" + + +def settle_curve(progress: float, sharpness: float = 4.0) -> float: + """Return noise amount in range 1..0 for normalized progress 0..1.""" + t = max(0.0, min(1.0, progress)) + return math.exp(-sharpness * t) + + +# Braille dot positions: (0,0) (1,0) dots 1,4 +# (0,1) (1,1) dots 2,5 +# (0,2) (1,2) dots 3,6 +# (0,3) (1,3) dots 7,8 +_DOT_MAP = ( + (0x01, 0x08), + (0x02, 0x10), + (0x04, 0x20), + (0x40, 0x80), +) + + +class BrailleCanvas: + """A pixel canvas that renders to braille characters.""" + + def __init__(self, term_width: int, term_height: int): + self.term_width = term_width + self.term_height = term_height + self.pixel_width = term_width * 2 + self.pixel_height = term_height * 4 + self._buf = bytearray(term_width * term_height) + + def clear(self) -> None: + for i in range(len(self._buf)): + self._buf[i] = 0 + + def set_pixel(self, x: int, y: int) -> None: + if 0 <= x < self.pixel_width and 0 <= y < self.pixel_height: + cx, rx = divmod(x, 2) + cy, ry = divmod(y, 4) + self._buf[cy * self.term_width + cx] |= _DOT_MAP[ry][rx] + + def render(self) -> list[str]: + lines = [] + for row in range(self.term_height): + offset = row * self.term_width + line = "".join(chr(0x2800 + self._buf[offset + col]) for col in range(self.term_width)) + lines.append(line) + return lines + + +# ── Bitmap font (5×7 uppercase + digits) ────────────────────────────── + +_FONT: dict[str, list[str]] = {} + + +def _define_font() -> None: + """Define a simple 5×7 bitmap font for uppercase ASCII.""" + glyphs = { + "A": [" ## ", "# #", "# #", "####", "# #", "# #", "# #"], + "B": ["### ", "# #", "# #", "### ", "# #", "# #", "### "], + "C": [" ## ", "# #", "# ", "# ", "# ", "# #", " ## "], + "D": ["### ", "# #", "# #", "# #", "# #", "# #", "### "], + "E": ["####", "# ", "# ", "### ", "# ", "# ", "####"], + "F": ["####", "# ", "# ", "### ", "# ", "# ", "# "], + "G": [" ## ", "# #", "# ", "# ##", "# #", "# #", " ###"], + "H": ["# #", "# #", "# #", "####", "# #", "# #", "# #"], + "I": ["###", " # ", " # ", " # ", " # ", " # ", "###"], + "J": [" ##", " # ", " # ", " # ", " # ", "# # ", " # "], + "K": ["# #", "# # ", "## ", "## ", "# # ", "# #", "# #"], + "L": ["# ", "# ", "# ", "# ", "# ", "# ", "####"], + "M": ["# #", "## ##", "# # #", "# # #", "# #", "# #", "# #"], + "N": ["# #", "## #", "## #", "# ##", "# ##", "# #", "# #"], + "O": [" ## ", "# #", "# #", "# #", "# #", "# #", " ## "], + "P": ["### ", "# #", "# #", "### ", "# ", "# ", "# "], + "Q": [" ## ", "# #", "# #", "# #", "# ##", "# #", " ## "], + "R": ["### ", "# #", "# #", "### ", "# # ", "# #", "# #"], + "S": [" ## ", "# #", "# ", " ## ", " #", "# #", " ## "], + "T": ["#####", " # ", " # ", " # ", " # ", " # ", " # "], + "U": ["# #", "# #", "# #", "# #", "# #", "# #", " ## "], + "V": ["# #", "# #", "# #", " # # ", " # # ", " # ", " # "], + "W": ["# #", "# #", "# #", "# # #", "# # #", "## ##", "# #"], + "X": ["# #", "# #", " ## ", " ## ", " ## ", "# #", "# #"], + "Y": ["# #", "# #", " # # ", " # ", " # ", " # ", " # "], + "Z": ["####", " #", " # ", " # ", "# ", "# ", "####"], + " ": [" ", " ", " ", " ", " ", " ", " "], + "0": [" ## ", "# #", "# #", "# #", "# #", "# #", " ## "], + "1": [" # ", "## ", " # ", " # ", " # ", " # ", "###"], + "2": [" ## ", "# #", " #", " # ", " # ", "# ", "####"], + "3": [" ## ", "# #", " #", " ## ", " #", "# #", " ## "], + "4": ["# #", "# #", "# #", "####", " #", " #", " #"], + "5": ["####", "# ", "### ", " #", " #", "# #", " ## "], + "6": [" ## ", "# ", "### ", "# #", "# #", "# #", " ## "], + "7": ["####", " #", " # ", " # ", " # ", " # ", " # "], + "8": [" ## ", "# #", "# #", " ## ", "# #", "# #", " ## "], + "9": [" ## ", "# #", "# #", " ###", " #", " #", " ## "], + } + _FONT.update(glyphs) + + +_define_font() + + +def text_to_pixels(text: str, scale: int = 1) -> list[tuple[int, int]]: + """Convert text string to a list of (x, y) pixel positions using bitmap font.""" + pixels = [] + cursor_x = 0 + for ch in text.upper(): + glyph = _FONT.get(ch) + if glyph is None: + cursor_x += 4 * scale + continue + for row_idx, row in enumerate(glyph): + for col_idx, cell in enumerate(row): + if cell == "#": + for sy in range(scale): + for sx in range(scale): + pixels.append((cursor_x + col_idx * scale + sx, row_idx * scale + sy)) + glyph_width = max(len(r) for r in glyph) + cursor_x += (glyph_width + 1) * scale + return pixels + + +class Particle: + __slots__ = ("x", "y", "target_x", "target_y", "vx", "vy", "phase", "delay") + + def __init__(self, x: float, y: float, target_x: float, target_y: float, delay: float = 0): + self.x = x + self.y = y + self.target_x = target_x + self.target_y = target_y + self.vx = 0.0 + self.vy = 0.0 + self.phase = random.uniform(0, math.pi * 2) + self.delay = delay + + def update_converge(self, t: float, strength: float = 0.08, damping: float = 0.92): + """Move toward target with spring-like physics.""" + if t < self.delay: + # Still in swirl phase + self.x += self.vx + self.y += self.vy + self.vx *= 0.99 + self.vy *= 0.99 + # Gentle spiral + angle = self.phase + t * 2 + self.vx += math.cos(angle) * 0.3 + self.vy += math.sin(angle) * 0.3 + return + + # Spring toward target + dx = self.target_x - self.x + dy = self.target_y - self.y + self.vx += dx * strength + self.vy += dy * strength + self.vx *= damping + self.vy *= damping + self.x += self.vx + self.y += self.vy + + @property + def at_target(self) -> bool: + return abs(self.x - self.target_x) < 1.5 and abs(self.y - self.target_y) < 1.5 + + +def run_particle_logo(console: Console, hold_seconds: float = 1.5) -> None: + """Run the particle coalesce effect.""" + term_width = min(console.width, 120) + term_height = min(console.height - 4, 35) + + canvas = BrailleCanvas(term_width, term_height) + + theme = set_theme("dark") + style_final = theme.styles["lavender"] + # Get target positions from text + text_pixels_line1 = text_to_pixels("REPOsitory", scale=2) + text_pixels_line2 = text_to_pixels("MANager", scale=2) + + # Calculate dimensions for centering + def get_bounds(pixels): + if not pixels: + return 0, 0, 0, 0 + xs = [p[0] for p in pixels] + ys = [p[1] for p in pixels] + return min(xs), max(xs), min(ys), max(ys) + + min_x1, max_x1, min_y1, max_y1 = get_bounds(text_pixels_line1) + min_x2, max_x2, min_y2, max_y2 = get_bounds(text_pixels_line2) + + w1, h1 = max_x1 - min_x1 + 1, max_y1 - min_y1 + 1 + w2, h2 = max_x2 - min_x2 + 1, max_y2 - min_y2 + 1 + + total_h = h1 + 6 + h2 # gap between lines + start_y = (canvas.pixel_height - total_h) // 2 + + # Center line 1 + offset_x1 = (canvas.pixel_width - w1) // 2 - min_x1 + offset_y1 = start_y - min_y1 + targets_1 = [(p[0] + offset_x1, p[1] + offset_y1) for p in text_pixels_line1] + + # Center line 2 + offset_x2 = (canvas.pixel_width - w2) // 2 - min_x2 + offset_y2 = start_y + h1 + 6 - min_y2 + targets_2 = [(p[0] + offset_x2, p[1] + offset_y2) for p in text_pixels_line2] + + all_targets = targets_1 + targets_2 + + # Subsample for performance — take every Nth pixel + step = max(1, len(all_targets) // 1500) + sampled_targets = all_targets[::step] + + # Create particles at random edge positions + rng = random.Random(42) + particles = [] + pw, ph = canvas.pixel_width, canvas.pixel_height + + for i, (tx, ty) in enumerate(sampled_targets): + # Spawn from random edge + side = rng.choice(["top", "bottom", "left", "right"]) + if side == "top": + sx, sy = rng.uniform(0, pw), rng.uniform(-20, -5) + elif side == "bottom": + sx, sy = rng.uniform(0, pw), rng.uniform(ph + 5, ph + 20) + elif side == "left": + sx, sy = rng.uniform(-20, -5), rng.uniform(0, ph) + else: + sx, sy = rng.uniform(pw + 5, pw + 20), rng.uniform(0, ph) + + delay = rng.uniform(0, 0.4) # staggered start + p = Particle(sx, sy, tx, ty, delay=delay) + # Initial velocity — gentle swirl + angle = math.atan2(ph / 2 - sy, pw / 2 - sx) + rng.gauss(0, 0.8) + speed = rng.uniform(1.0, 2.5) + p.vx = math.cos(angle) * speed + p.vy = math.sin(angle) * speed + particles.append(p) + + # Also add some extra ambient particles that never converge + ambient = [] + for _ in range(200): + ax = rng.uniform(0, pw) + ay = rng.uniform(0, ph) + ap = Particle(ax, ay, ax, ay) + ap.vx = rng.gauss(0, 1) + ap.vy = rng.gauss(0, 1) + ambient.append(ap) + + # Timing: 1s converge + 2s hold = 3s total + fps = 24 + converge_frames = int(fps * 0.9) + hold_frames = int(fps * hold_seconds) + total_frames = converge_frames + hold_frames + + with Live(console=console, refresh_per_second=fps, transient=True) as live: + for frame in range(total_frames): + canvas.clear() + t = frame * 0.03 + + # Update ambient particles (always drifting) + for ap in ambient: + ap.x += ap.vx + math.sin(t + ap.phase) * 0.5 + ap.y += ap.vy + math.cos(t + ap.phase * 1.3) * 0.5 + # Wrap around + ap.x = ap.x % pw + ap.y = ap.y % ph + + # Fade out ambient during hold phase + if frame < converge_frames: + alpha = 0.3 + 0.2 * math.sin(t * 2 + ap.phase) + else: + fade = (frame - converge_frames) / hold_frames + alpha = (0.3 + 0.2 * math.sin(t * 2 + ap.phase)) * (1 - fade) + if alpha > 0.25: + canvas.set_pixel(int(ap.x), int(ap.y)) + + if frame < converge_frames: + # Converge phase + progress = frame / converge_frames + noise = settle_curve(progress) + for p in particles: + p.update_converge(t, strength=0.06, damping=0.90) + canvas.set_pixel(int(p.x), int(p.y)) + + # Trail effect + trail_scale = 0.2 + 0.5 * noise + trail_x = int(p.x - p.vx * trail_scale) + trail_y = int(p.y - p.vy * trail_scale) + canvas.set_pixel(trail_x, trail_y) + + # Color transitions from white to warm gold + r, g, b = cool_ramp(progress) + else: + # Hold phase — settle into solid logo + settle_t = (frame - converge_frames) / hold_frames + for p in particles: + # Jitter decays to zero + jitter = (1 - settle_t) * 0.7 + jx = p.target_x + math.sin(t * 3 + p.phase) * jitter + jy = p.target_y + math.cos(t * 3 + p.phase * 1.5) * jitter + canvas.set_pixel(int(jx), int(jy)) + canvas.set_pixel(int(p.target_x), int(p.target_y)) + + r, g, b = style_final.color.triplet + + # Render with color + lines = canvas.render() + result = Text() + for line in lines: + for ch in line: + if ch == chr(0x2800): + result.append(ch) + else: + result.append(ch, style=f"rgb({r},{g},{b})") + result.append("\n") + + live.update(Align.center(result)) + time.sleep(1.0 / fps) + + # Print final settled frame + canvas.clear() + for p in particles: + canvas.set_pixel(int(p.target_x), int(p.target_y)) + final = Text() + + for line in canvas.render(): + for ch in line: + if ch == chr(0x2800): + final.append(ch) + else: + final.append(ch, style=style_final) + final.append("\n") + console.print(Align.center(final)) + + +def main() -> int: + console = Console(color_system="auto") + + try: + run_particle_logo( + console=console, + hold_seconds=1.5, + ) + except KeyboardInterrupt: + console.print("\n[dim]interrupted[/dim]") + return 130 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) From 17c750f960e7ae7281534b9fecf979b5de0a79b4 Mon Sep 17 00:00:00 2001 From: "Leandro G. Almeida" Date: Sat, 25 Apr 2026 14:24:29 -0700 Subject: [PATCH 2/6] =?UTF-8?q?=E2=9C=A8=20refactor(ui):=20central=20conso?= =?UTF-8?q?le=20and=20move/theme=20imports=20to=20ui?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/repoman/utils/task_tree_display.py | 18 +++++++----------- src/repoman/utils/ui/display_tasks.py | 10 ++++------ src/repoman/utils/ui/logo_display.py | 2 +- 3 files changed, 12 insertions(+), 18 deletions(-) diff --git a/src/repoman/utils/task_tree_display.py b/src/repoman/utils/task_tree_display.py index 393f6fb..4d00f58 100644 --- a/src/repoman/utils/task_tree_display.py +++ b/src/repoman/utils/task_tree_display.py @@ -143,8 +143,7 @@ def get_padding(self, tree: TaskTree) -> int: class UpdateTracker: - """ - Class to enable dynamic updates on the UI tables. By default, rich allows you to set a refresh rate or trigger manual + """Class to enable dynamic updates on the UI tables. By default, rich allows you to set a refresh rate or trigger manual updates. This makes manual updates more performant by doing quick 'dirty' checks to determine if updating ins required. Updating is technically always required because the table's 'elapsed time' column always changes, but we don't want to update the table just because of that. @@ -156,12 +155,11 @@ class UpdateTracker: """ def __init__(self, max_update_timeout_ms: float = 5000, min_update_ms: float = 200) -> None: - """ - Args: - max_update_timeout_ms: The maximum amount of time in milliseconds that can pass before an update is forced. This is useful - because the table usually contains an 'elapsed time' column that should update fairly frequently regardless of everything else. - min_update_ms: The minimum amount of time in milliseconds that must pass before an update is allowed. This prevents updates - from getting too frequent. + """Args: + max_update_timeout_ms: The maximum amount of time in milliseconds that can pass before an update is forced. This is useful + because the table usually contains an 'elapsed time' column that should update fairly frequently regardless of everything else. + min_update_ms: The minimum amount of time in milliseconds that must pass before an update is allowed. This prevents updates + from getting too frequent. """ self._timeout_ms = max_update_timeout_ms self._min_update_ms = min_update_ms @@ -191,7 +189,5 @@ def _update( def update(self, update_fn: Callable[[], None], state: list[str | list[str]]): now = datetime.now().timestamp() * 1000 - if self.min_update_time_passed(now) and state != self._last_update_state: - self._update(now, update_fn, state) - elif self.max_update_time_passed(now): + if (self.min_update_time_passed(now) and state != self._last_update_state) or self.max_update_time_passed(now): self._update(now, update_fn, state) diff --git a/src/repoman/utils/ui/display_tasks.py b/src/repoman/utils/ui/display_tasks.py index b07c655..46c3658 100644 --- a/src/repoman/utils/ui/display_tasks.py +++ b/src/repoman/utils/ui/display_tasks.py @@ -1,5 +1,4 @@ -""" -Terminal display utilities — rich-powered CLI formatting. +"""Terminal display utilities — rich-powered CLI formatting. """ import re @@ -11,7 +10,8 @@ class _LeftHeading(Heading): """Rich's default Markdown renders h1/h2 centered via Align.center. - Yield the styled text directly so headings stay left-aligned.""" + Yield the styled text directly so headings stay left-aligned. + """ def __rich_console__(self, console, options): self.text.justify = "left" @@ -374,9 +374,7 @@ async def print_markdown( break f.write(ch) f.flush() - if ch == "\n": - await asyncio.sleep(0.002) - elif ch == " ": + if ch == "\n" or ch == " ": await asyncio.sleep(0.002) elif rng.random() < 0.03: await asyncio.sleep(0.015) diff --git a/src/repoman/utils/ui/logo_display.py b/src/repoman/utils/ui/logo_display.py index f86ee7d..305bec1 100644 --- a/src/repoman/utils/ui/logo_display.py +++ b/src/repoman/utils/ui/logo_display.py @@ -145,7 +145,7 @@ def text_to_pixels(text: str, scale: int = 1) -> list[tuple[int, int]]: class Particle: - __slots__ = ("x", "y", "target_x", "target_y", "vx", "vy", "phase", "delay") + __slots__ = ("delay", "phase", "target_x", "target_y", "vx", "vy", "x", "y") def __init__(self, x: float, y: float, target_x: float, target_y: float, delay: float = 0): self.x = x From 5aeb11ed469dfa5a466b81f17e1e70defb719a77 Mon Sep 17 00:00:00 2001 From: "Leandro G. Almeida" Date: Sat, 25 Apr 2026 16:18:58 -0700 Subject: [PATCH 3/6] =?UTF-8?q?=E2=9C=A8=20cli(update):=20reformat=20argum?= =?UTF-8?q?ents=20and=20wrapped=20long=20lines=20for=20readability?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/repoman/_version.py | 12 ++- src/repoman/cli/commands/update/__init__.py | 113 +++++++++++++++----- 2 files changed, 97 insertions(+), 28 deletions(-) diff --git a/src/repoman/_version.py b/src/repoman/_version.py index bf15195..8e10d1a 100644 --- a/src/repoman/_version.py +++ b/src/repoman/_version.py @@ -162,7 +162,9 @@ def _make_debug_layout(env: Environment) -> Layout: Text("Interpreter Path", style="rosewater"), Text(env.interpreter_path, style="bold"), ) - header_table.add_row(Text("Platform", style="rosewater"), Text(env.platform, style="bold")) + header_table.add_row( + Text("Platform", style="rosewater"), Text(env.platform, style="bold") + ) header = Panel( header_table, title="Debug Information", @@ -181,7 +183,9 @@ def _make_debug_layout(env: Environment) -> Layout: for pkg in env.packages: packages_table.add_row(pkg.name, pkg.version) - env_table = Table(highlight=True, box=None, show_header=True, title="Environment Variables") + env_table = Table( + highlight=True, box=None, show_header=True, title="Environment Variables" + ) env_table.add_column("Variable", style="rosewater") env_table.add_column("Value", style="bold") for var in env.variables: @@ -193,7 +197,9 @@ def _make_debug_layout(env: Environment) -> Layout: Layout(name="main", ratio=1), ) layout["main"].split_row( - Layout(Panel(packages_table, border_style="bright_blue"), name="packages", ratio=1), + Layout( + Panel(packages_table, border_style="bright_blue"), name="packages", ratio=1 + ), Layout( Panel(env_table, border_style="bright_blue"), name="vars", diff --git a/src/repoman/cli/commands/update/__init__.py b/src/repoman/cli/commands/update/__init__.py index 46b79d1..f6c0a5b 100644 --- a/src/repoman/cli/commands/update/__init__.py +++ b/src/repoman/cli/commands/update/__init__.py @@ -38,7 +38,9 @@ from repoman.inspection import InspectionError, InspectionReport, inspect_repository from repoman.utils.logging import get_logger_console -app = Typer(add_completion=True) +app = Typer( + add_completion=True, +) @app.callback(invoke_without_command=True) @@ -64,12 +66,24 @@ def update( help="Path to .copier-answers.yml file (defaults to .copier-answers.yml in project_dir)", ), force: bool = Option(False, "--force", "-f", help="Force overwrite without asking"), - dry_run: bool = Option(False, "--dry-run", help="Show what would be updated without making changes"), - plan: bool = Option(False, "--plan", help="Show an update plan without writing files"), - repair: bool = Option(False, "--repair", help="Repair Copier metadata in the answers file"), - commit: str | None = Option(None, "--commit", help="Copier commit to write when using --repair"), - conflict: str = Option("inline", "--conflict", help="Conflict resolution mode: 'inline' or 'rej'"), - skip_extensions: bool = Option(False, "--skip-extensions", help="Skip syncing Copier-managed extensions"), + dry_run: bool = Option( + False, "--dry-run", help="Show what would be updated without making changes" + ), + plan: bool = Option( + False, "--plan", help="Show an update plan without writing files" + ), + repair: bool = Option( + False, "--repair", help="Repair Copier metadata in the answers file" + ), + commit: str | None = Option( + None, "--commit", help="Copier commit to write when using --repair" + ), + conflict: str = Option( + "inline", "--conflict", help="Conflict resolution mode: 'inline' or 'rej'" + ), + skip_extensions: bool = Option( + False, "--skip-extensions", help="Skip syncing Copier-managed extensions" + ), ) -> None: """Update an existing Python project using the repoman template. @@ -131,27 +145,43 @@ def update( return if not answers_file_path.exists(): - console.print(error_panel(copier_answers_not_found_for_update(answers_file_path), console=console)) + console.print( + error_panel( + copier_answers_not_found_for_update(answers_file_path), console=console + ) + ) raise Exit(1) from None if report.answers_file.within_project is False: - console.print(error_panel(answers_file_must_live_in_project(answers_file_path), console=console)) + console.print( + error_panel( + answers_file_must_live_in_project(answers_file_path), console=console + ) + ) raise Exit(1) from None if report.template.src_path and not report.template.commit: console.print( error_panel( - copier_update_missing_commit_remediation(answers_basename=answers_file_path.name), + copier_update_missing_commit_remediation( + answers_basename=answers_file_path.name + ), console=console, ) ) raise Exit(1) from None if report.fatal: - console.print(error_panel("\n".join(report.update_readiness.blockers), console=console)) + console.print( + error_panel("\n".join(report.update_readiness.blockers), console=console) + ) raise Exit(1) from None if report.update_readiness.blockers: - console.print(error_panel("\n".join(report.update_readiness.blockers), console=console)) + console.print( + error_panel("\n".join(report.update_readiness.blockers), console=console) + ) raise Exit(1) from None - answers_file_for_worker = str(answers_file_path.resolve().relative_to(project_dir_obj.resolve())) + answers_file_for_worker = str( + answers_file_path.resolve().relative_to(project_dir_obj.resolve()) + ) copier_options = _build_copier_options( project_dir=project_dir_obj, answers_file_for_worker=answers_file_for_worker, @@ -186,7 +216,9 @@ def update( console.print(error_panel(str(exc), console=console)) raise Exit(1) from exc - copier_options_serializable: dict[str, object] = {k: v for k, v in copier_options.items() if v is not None} + copier_options_serializable: dict[str, object] = { + k: v for k, v in copier_options.items() if v is not None + } if extension_dry_run_options is not None: copier_options_serializable["extensions"] = extension_dry_run_options console.print( @@ -222,7 +254,9 @@ def update( progress.update(task, description="Project updated successfully!") - copier_options_serializable = {k: v for k, v in copier_options.items() if v is not None} + copier_options_serializable = { + k: v for k, v in copier_options.items() if v is not None + } console.print( project_updated( project_dir_obj, @@ -255,7 +289,9 @@ def _resolve_project_dir(project_dir: Path, console: Any) -> Path: console.print(error_panel(project_dir_not_found(project_dir), console=console)) raise Exit(1) from None if not project_dir.is_dir(): - console.print(error_panel(project_path_not_directory(project_dir), console=console)) + console.print( + error_panel(project_path_not_directory(project_dir), console=console) + ) raise Exit(1) from None return project_dir @@ -266,13 +302,19 @@ def _resolve_answers_path(project_dir: Path, answers_file: str | None) -> Path: return Path(answers_file).resolve() -def _resolve_template_path(template_path: str | None, logger: Any, console: Any) -> Path | None: +def _resolve_template_path( + template_path: str | None, logger: Any, console: Any +) -> Path | None: if template_path is None: logger.info("Template path will be read from .copier-answers.yml") return None template_path_obj = Path(template_path).resolve() if not template_path_obj.exists(): - console.print(error_panel(template_path_does_not_exist(template_path_obj), console=console)) + console.print( + error_panel( + template_path_does_not_exist(template_path_obj), console=console + ) + ) raise Exit(1) from None logger.info(f"Using custom template at {template_path_obj}") return template_path_obj @@ -318,11 +360,15 @@ def _print_update_plan( summary.add_row("Answers file", report.answers_file.path) summary.add_row( "Template source", - str(template_path) if template_path is not None else (report.template.src_path or "-"), + str(template_path) + if template_path is not None + else (report.template.src_path or "-"), ) summary.add_row("Template commit", report.template.commit or "-") summary.add_row("Template ref", vcs_ref or report.template.vcs_ref or "-") - summary.add_row("Extensions", "skipped" if skip_extensions else _extension_plan_text(report)) + summary.add_row( + "Extensions", "skipped" if skip_extensions else _extension_plan_text(report) + ) summary.add_row("Update ready", _bool_text(report.update_readiness.ready)) next_steps = [ @@ -330,7 +376,10 @@ def _print_update_plan( "Use --dry-run for Copier options preview", "Review and test the project after the update", ] - body: list[RenderableType] = [Text("Plan for repoman update", style="blue"), summary] + body: list[RenderableType] = [ + Text("Plan for repoman update", style="blue"), + summary, + ] if report.update_readiness.blockers: body.append( Panel( @@ -374,10 +423,18 @@ def _run_repair( report: InspectionReport, ) -> None: if not answers_file_path.exists(): - console.print(error_panel(copier_answers_not_found_for_update(answers_file_path), console=console)) + console.print( + error_panel( + copier_answers_not_found_for_update(answers_file_path), console=console + ) + ) raise Exit(1) from None if report.answers_file.within_project is False: - console.print(error_panel(answers_file_must_live_in_project(answers_file_path), console=console)) + console.print( + error_panel( + answers_file_must_live_in_project(answers_file_path), console=console + ) + ) raise Exit(1) from None if commit is None or not commit.strip(): console.print(error_panel(repair_requires_commit(), console=console)) @@ -389,7 +446,11 @@ def _run_repair( console.print(error_panel(invalid_yaml(exc), console=console)) raise Exit(1) from exc - selected_template_source = str(template_path_obj) if template_path_obj is not None else report.template.src_path + selected_template_source = ( + str(template_path_obj) + if template_path_obj is not None + else report.template.src_path + ) if not selected_template_source: console.print(error_panel(repair_requires_template_source(), console=console)) raise Exit(1) from None @@ -412,7 +473,9 @@ def _run_repair( if not changes: changes_text = "No Copier metadata changes were necessary." else: - changes_text = "Updated metadata:\n" + "\n".join(f" - {change}" for change in changes) + changes_text = "Updated metadata:\n" + "\n".join( + f" - {change}" for change in changes + ) console.print( Panel( Text( From a23291fe1eb43fb6173ebd3ed02e777a49401e95 Mon Sep 17 00:00:00 2001 From: "Leandro G. Almeida" Date: Sat, 25 Apr 2026 16:44:11 -0700 Subject: [PATCH 4/6] =?UTF-8?q?=E2=9C=A8=20ui:=20Improveille=20canvas=20ty?= =?UTF-8?q?pes,=20docs=20and=20particle=20logic?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/repoman/_version.py | 12 +- src/repoman/cli/commands/update/__init__.py | 104 +--- src/repoman/utils/task_tree_display.py | 44 +- src/repoman/utils/ui/display_tasks.py | 507 -------------------- src/repoman/utils/ui/logo_display.py | 52 +- 5 files changed, 84 insertions(+), 635 deletions(-) delete mode 100644 src/repoman/utils/ui/display_tasks.py diff --git a/src/repoman/_version.py b/src/repoman/_version.py index 8e10d1a..bf15195 100644 --- a/src/repoman/_version.py +++ b/src/repoman/_version.py @@ -162,9 +162,7 @@ def _make_debug_layout(env: Environment) -> Layout: Text("Interpreter Path", style="rosewater"), Text(env.interpreter_path, style="bold"), ) - header_table.add_row( - Text("Platform", style="rosewater"), Text(env.platform, style="bold") - ) + header_table.add_row(Text("Platform", style="rosewater"), Text(env.platform, style="bold")) header = Panel( header_table, title="Debug Information", @@ -183,9 +181,7 @@ def _make_debug_layout(env: Environment) -> Layout: for pkg in env.packages: packages_table.add_row(pkg.name, pkg.version) - env_table = Table( - highlight=True, box=None, show_header=True, title="Environment Variables" - ) + env_table = Table(highlight=True, box=None, show_header=True, title="Environment Variables") env_table.add_column("Variable", style="rosewater") env_table.add_column("Value", style="bold") for var in env.variables: @@ -197,9 +193,7 @@ def _make_debug_layout(env: Environment) -> Layout: Layout(name="main", ratio=1), ) layout["main"].split_row( - Layout( - Panel(packages_table, border_style="bright_blue"), name="packages", ratio=1 - ), + Layout(Panel(packages_table, border_style="bright_blue"), name="packages", ratio=1), Layout( Panel(env_table, border_style="bright_blue"), name="vars", diff --git a/src/repoman/cli/commands/update/__init__.py b/src/repoman/cli/commands/update/__init__.py index f6c0a5b..28d0a04 100644 --- a/src/repoman/cli/commands/update/__init__.py +++ b/src/repoman/cli/commands/update/__init__.py @@ -66,24 +66,12 @@ def update( help="Path to .copier-answers.yml file (defaults to .copier-answers.yml in project_dir)", ), force: bool = Option(False, "--force", "-f", help="Force overwrite without asking"), - dry_run: bool = Option( - False, "--dry-run", help="Show what would be updated without making changes" - ), - plan: bool = Option( - False, "--plan", help="Show an update plan without writing files" - ), - repair: bool = Option( - False, "--repair", help="Repair Copier metadata in the answers file" - ), - commit: str | None = Option( - None, "--commit", help="Copier commit to write when using --repair" - ), - conflict: str = Option( - "inline", "--conflict", help="Conflict resolution mode: 'inline' or 'rej'" - ), - skip_extensions: bool = Option( - False, "--skip-extensions", help="Skip syncing Copier-managed extensions" - ), + dry_run: bool = Option(False, "--dry-run", help="Show what would be updated without making changes"), + plan: bool = Option(False, "--plan", help="Show an update plan without writing files"), + repair: bool = Option(False, "--repair", help="Repair Copier metadata in the answers file"), + commit: str | None = Option(None, "--commit", help="Copier commit to write when using --repair"), + conflict: str = Option("inline", "--conflict", help="Conflict resolution mode: 'inline' or 'rej'"), + skip_extensions: bool = Option(False, "--skip-extensions", help="Skip syncing Copier-managed extensions"), ) -> None: """Update an existing Python project using the repoman template. @@ -145,43 +133,27 @@ def update( return if not answers_file_path.exists(): - console.print( - error_panel( - copier_answers_not_found_for_update(answers_file_path), console=console - ) - ) + console.print(error_panel(copier_answers_not_found_for_update(answers_file_path), console=console)) raise Exit(1) from None if report.answers_file.within_project is False: - console.print( - error_panel( - answers_file_must_live_in_project(answers_file_path), console=console - ) - ) + console.print(error_panel(answers_file_must_live_in_project(answers_file_path), console=console)) raise Exit(1) from None if report.template.src_path and not report.template.commit: console.print( error_panel( - copier_update_missing_commit_remediation( - answers_basename=answers_file_path.name - ), + copier_update_missing_commit_remediation(answers_basename=answers_file_path.name), console=console, ) ) raise Exit(1) from None if report.fatal: - console.print( - error_panel("\n".join(report.update_readiness.blockers), console=console) - ) + console.print(error_panel("\n".join(report.update_readiness.blockers), console=console)) raise Exit(1) from None if report.update_readiness.blockers: - console.print( - error_panel("\n".join(report.update_readiness.blockers), console=console) - ) + console.print(error_panel("\n".join(report.update_readiness.blockers), console=console)) raise Exit(1) from None - answers_file_for_worker = str( - answers_file_path.resolve().relative_to(project_dir_obj.resolve()) - ) + answers_file_for_worker = str(answers_file_path.resolve().relative_to(project_dir_obj.resolve())) copier_options = _build_copier_options( project_dir=project_dir_obj, answers_file_for_worker=answers_file_for_worker, @@ -216,9 +188,7 @@ def update( console.print(error_panel(str(exc), console=console)) raise Exit(1) from exc - copier_options_serializable: dict[str, object] = { - k: v for k, v in copier_options.items() if v is not None - } + copier_options_serializable: dict[str, object] = {k: v for k, v in copier_options.items() if v is not None} if extension_dry_run_options is not None: copier_options_serializable["extensions"] = extension_dry_run_options console.print( @@ -254,9 +224,7 @@ def update( progress.update(task, description="Project updated successfully!") - copier_options_serializable = { - k: v for k, v in copier_options.items() if v is not None - } + copier_options_serializable = {k: v for k, v in copier_options.items() if v is not None} console.print( project_updated( project_dir_obj, @@ -289,9 +257,7 @@ def _resolve_project_dir(project_dir: Path, console: Any) -> Path: console.print(error_panel(project_dir_not_found(project_dir), console=console)) raise Exit(1) from None if not project_dir.is_dir(): - console.print( - error_panel(project_path_not_directory(project_dir), console=console) - ) + console.print(error_panel(project_path_not_directory(project_dir), console=console)) raise Exit(1) from None return project_dir @@ -302,19 +268,13 @@ def _resolve_answers_path(project_dir: Path, answers_file: str | None) -> Path: return Path(answers_file).resolve() -def _resolve_template_path( - template_path: str | None, logger: Any, console: Any -) -> Path | None: +def _resolve_template_path(template_path: str | None, logger: Any, console: Any) -> Path | None: if template_path is None: logger.info("Template path will be read from .copier-answers.yml") return None template_path_obj = Path(template_path).resolve() if not template_path_obj.exists(): - console.print( - error_panel( - template_path_does_not_exist(template_path_obj), console=console - ) - ) + console.print(error_panel(template_path_does_not_exist(template_path_obj), console=console)) raise Exit(1) from None logger.info(f"Using custom template at {template_path_obj}") return template_path_obj @@ -360,15 +320,11 @@ def _print_update_plan( summary.add_row("Answers file", report.answers_file.path) summary.add_row( "Template source", - str(template_path) - if template_path is not None - else (report.template.src_path or "-"), + str(template_path) if template_path is not None else (report.template.src_path or "-"), ) summary.add_row("Template commit", report.template.commit or "-") summary.add_row("Template ref", vcs_ref or report.template.vcs_ref or "-") - summary.add_row( - "Extensions", "skipped" if skip_extensions else _extension_plan_text(report) - ) + summary.add_row("Extensions", "skipped" if skip_extensions else _extension_plan_text(report)) summary.add_row("Update ready", _bool_text(report.update_readiness.ready)) next_steps = [ @@ -423,18 +379,10 @@ def _run_repair( report: InspectionReport, ) -> None: if not answers_file_path.exists(): - console.print( - error_panel( - copier_answers_not_found_for_update(answers_file_path), console=console - ) - ) + console.print(error_panel(copier_answers_not_found_for_update(answers_file_path), console=console)) raise Exit(1) from None if report.answers_file.within_project is False: - console.print( - error_panel( - answers_file_must_live_in_project(answers_file_path), console=console - ) - ) + console.print(error_panel(answers_file_must_live_in_project(answers_file_path), console=console)) raise Exit(1) from None if commit is None or not commit.strip(): console.print(error_panel(repair_requires_commit(), console=console)) @@ -446,11 +394,7 @@ def _run_repair( console.print(error_panel(invalid_yaml(exc), console=console)) raise Exit(1) from exc - selected_template_source = ( - str(template_path_obj) - if template_path_obj is not None - else report.template.src_path - ) + selected_template_source = str(template_path_obj) if template_path_obj is not None else report.template.src_path if not selected_template_source: console.print(error_panel(repair_requires_template_source(), console=console)) raise Exit(1) from None @@ -473,9 +417,7 @@ def _run_repair( if not changes: changes_text = "No Copier metadata changes were necessary." else: - changes_text = "Updated metadata:\n" + "\n".join( - f" - {change}" for change in changes - ) + changes_text = "Updated metadata:\n" + "\n".join(f" - {change}" for change in changes) console.print( Panel( Text( diff --git a/src/repoman/utils/task_tree_display.py b/src/repoman/utils/task_tree_display.py index 4d00f58..d428317 100644 --- a/src/repoman/utils/task_tree_display.py +++ b/src/repoman/utils/task_tree_display.py @@ -1,5 +1,7 @@ +"""Rich tree helpers for displaying running task output.""" + from collections.abc import Callable -from datetime import datetime +from datetime import UTC, datetime from typing import Any from rich.text import Text @@ -38,7 +40,7 @@ def __init__(self, cmd: str) -> None: self._stdout_str = [] self._stderr_str = [] - def set_stdout(self, stdout: list[str]): + def set_stdout(self, stdout: list[str]) -> None: """Set or replace the lines shown under the ``stdout`` branch. Creates the ``stdout`` branch on first non-empty call and reorders the tree @@ -66,7 +68,7 @@ def set_stdout(self, stdout: list[str]): self._stdout_str = stdout self._stdout_text.plain = "\n".join(stdout) - def set_stderr(self, stderr: list[str]): + def set_stderr(self, stderr: list[str]) -> None: """Set or replace the lines shown under the ``stderr`` branch. Creates the ``stderr`` branch on first non-empty call, styled in red. @@ -86,7 +88,7 @@ def set_stderr(self, stderr: list[str]): self._stderr_str = stderr self._stderr_text.plain = "\n".join(stderr) - def reset(self): + def reset(self) -> None: """Clear stdout/stderr branches and restore the tree to its initial state. After this call the tree contains only the original command label, and @@ -143,10 +145,7 @@ def get_padding(self, tree: TaskTree) -> int: class UpdateTracker: - """Class to enable dynamic updates on the UI tables. By default, rich allows you to set a refresh rate or trigger manual - updates. This makes manual updates more performant by doing quick 'dirty' checks to determine if updating ins required. Updating - is technically always required because the table's 'elapsed time' column always changes, but we don't want to update the table just - because of that. + """Track whether dynamic UI tables should be refreshed. This class tracks some state and has a min/max ms time config to keep the table looking responsive without updating too often. The driver for this was my CPU usage while the installer was running, paired with the size of the asciinema files that were generated @@ -155,39 +154,38 @@ class UpdateTracker: """ def __init__(self, max_update_timeout_ms: float = 5000, min_update_ms: float = 200) -> None: - """Args: - max_update_timeout_ms: The maximum amount of time in milliseconds that can pass before an update is forced. This is useful - because the table usually contains an 'elapsed time' column that should update fairly frequently regardless of everything else. - min_update_ms: The minimum amount of time in milliseconds that must pass before an update is allowed. This prevents updates - from getting too frequent. + """Initialize the tracker. + + Args: + max_update_timeout_ms: Maximum milliseconds before an update is forced. + min_update_ms: Minimum milliseconds between allowed updates. """ self._timeout_ms = max_update_timeout_ms self._min_update_ms = min_update_ms - self._last_update_timestamp_ms = datetime.now().timestamp() * 1000 + self._last_update_timestamp_ms = datetime.now(UTC).timestamp() * 1000 self._last_update_state: list[Any] = [] def max_update_time_passed(self, now: float) -> bool: - if now - self._last_update_timestamp_ms > self._timeout_ms: - return True - return False + """Return whether the forced update timeout has elapsed.""" + return now - self._last_update_timestamp_ms > self._timeout_ms def min_update_time_passed(self, now: float) -> bool: - if now - self._last_update_timestamp_ms > self._min_update_ms: - return True - return False + """Return whether the minimum interval between updates has elapsed.""" + return now - self._last_update_timestamp_ms > self._min_update_ms def _update( self, now: float, update_fn: Callable[[], None], state: list[str | list[str]], - ): + ) -> None: update_fn() self._last_update_timestamp_ms = now self._last_update_state = state - def update(self, update_fn: Callable[[], None], state: list[str | list[str]]): - now = datetime.now().timestamp() * 1000 + def update(self, update_fn: Callable[[], None], state: list[str | list[str]]) -> None: + """Run ``update_fn`` when elapsed time and state changes require it.""" + now = datetime.now(UTC).timestamp() * 1000 if (self.min_update_time_passed(now) and state != self._last_update_state) or self.max_update_time_passed(now): self._update(now, update_fn, state) diff --git a/src/repoman/utils/ui/display_tasks.py b/src/repoman/utils/ui/display_tasks.py deleted file mode 100644 index 46c3658..0000000 --- a/src/repoman/utils/ui/display_tasks.py +++ /dev/null @@ -1,507 +0,0 @@ -"""Terminal display utilities — rich-powered CLI formatting. -""" - -import re - -from rich.console import Console -from rich.markdown import Heading, Markdown -from rich.panel import Panel - - -class _LeftHeading(Heading): - """Rich's default Markdown renders h1/h2 centered via Align.center. - Yield the styled text directly so headings stay left-aligned. - """ - - def __rich_console__(self, console, options): - self.text.justify = "left" - yield self.text - - -Markdown.elements["heading_open"] = _LeftHeading - - -_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]") - - -def _clip_to_width(s: str, width: int) -> str: - """Truncate a string to `width` visible columns, preserving ANSI styles. - - Needed for the sub-agent live redraw: cursor-up-and-erase assumes one - logical line == one terminal row. If a line wraps, cursor-up undershoots - and the next redraw corrupts the display. Truncating prevents wrap. - """ - if width <= 0: - return s - out: list[str] = [] - visible = 0 - i = 0 - # Reserve 1 char for the trailing ellipsis - limit = width - 1 - truncated = False - while i < len(s): - m = _ANSI_RE.match(s, i) - if m: - out.append(m.group()) - i = m.end() - continue - if visible >= limit: - truncated = True - break - out.append(s[i]) - visible += 1 - i += 1 - if truncated: - # Strip styles (so ellipsis isn't left hanging inside a style run) - out.append("\033[0m…") - return "".join(out) - - -# Indent prefix for all agent output (aligns under the `>` prompt) -_I = " " - -# ── Banner ───────────────────────────────────────────────────────────── - - -def print_banner(model: str | None = None, hf_user: str | None = None) -> None: - """Print particle logo then CRT boot sequence with system info.""" - from agent.utils.crt_boot import run_boot_sequence - from agent.utils.particle_logo import run_particle_logo - - # Particle coalesce logo — 1.5s converge, 2s hold - run_particle_logo(_console, hold_seconds=2.0) - - # Clear screen for CRT boot — starts from top - _console.file.write("\033[2J\033[H") - _console.file.flush() - - model_label = model or "bedrock/us.anthropic.claude-opus-4-6-v1" - user_label = hf_user or "not logged in" - - # Warm gold palette matching the shimmer highlight (255, 200, 80) - gold = "rgb(255,200,80)" - dim_gold = "rgb(180,140,40)" - - boot_lines = [ - (f"{_I}Initializing agent runtime...", gold), - (f"{_I} User: {user_label}", dim_gold), - (f"{_I} Model: {model_label}", dim_gold), - (f"{_I} Tools: loading...", dim_gold), - ("", ""), - (f"{_I}/help for commands · /model to switch · /quit to exit", gold), - ] - - run_boot_sequence(_console, boot_lines) - - -# ── Init progress ────────────────────────────────────────────────────── - - -def print_init_done(tool_count: int = 0) -> None: - import time - - f = _console.file - # Overwrite the "Tools: loading..." line with actual count - f.write("\033[A\033[A\033[A\033[K") # Move up 3 lines (blank + help + blank) then up to tools line - f.write("\033[A\033[K") - gold = "\033[38;2;180;140;40m" - reset = "\033[0m" - tool_text = f"{_I} Tools: {tool_count} loaded" - for ch in tool_text: - f.write(f"{gold}{ch}{reset}") - f.flush() - time.sleep(0.012) - f.write("\n\n") - # Reprint the help line - f.write(f"{_I}\033[38;2;255;200;80m/help for commands · /model to switch · /quit to exit{reset}\n\n") - # Ready message — minimal padding - f.write(f"{_I}\033[38;2;255;200;80mReady. Let's build something impressive.{reset}\n") - f.flush() - - -# ── Tool calls ───────────────────────────────────────────────────────── - - -def print_tool_call(tool_name: str, args_preview: str) -> None: - import time - - f = _console.file - # CRT-style: type out tool name in HF yellow - gold = "\033[38;2;255;200;80m" - reset = "\033[0m" - f.write(f"{_I}{gold}▸ ") - for ch in tool_name: - f.write(ch) - f.flush() - time.sleep(0.015) - f.write(f"{reset} \033[2m{args_preview}{reset}\n") - f.flush() - - -def print_tool_output(output: str, success: bool, truncate: bool = True) -> None: - if truncate: - output = _truncate(output, max_lines=10) - style = "tool.ok" if success else "tool.fail" - # Indent each line of tool output - indented = "\n".join(f"{_I} {line}" for line in output.split("\n")) - _console.print(f"[{style}]{indented}[/{style}]") - - -class SubAgentDisplayManager: - """Manages multiple concurrent sub-agent displays. - - Each agent gets its own stats and rolling tool-call log. - All agents are rendered together so terminal escape-code - erase/redraw stays consistent. - """ - - _MAX_VISIBLE = 4 # tool-call lines shown per agent - - def __init__(self): - self._agents: dict[str, dict] = {} # agent_id -> state dict - self._lines_on_screen = 0 - self._ticker_task = None - - def start(self, agent_id: str, label: str = "research") -> None: - import asyncio - import time - - self._agents[agent_id] = { - "label": label, - "calls": [], - "tool_count": 0, - "token_count": 0, - "start_time": time.monotonic(), - } - if not self._ticker_task: - self._ticker_task = asyncio.ensure_future(self._tick()) - self._redraw() - - def set_tokens(self, agent_id: str, tokens: int) -> None: - if agent_id in self._agents: - self._agents[agent_id]["token_count"] = tokens - - def set_tool_count(self, agent_id: str, count: int) -> None: - if agent_id in self._agents: - self._agents[agent_id]["tool_count"] = count - - def add_call(self, agent_id: str, tool_desc: str) -> None: - if agent_id in self._agents: - self._agents[agent_id]["calls"].append(tool_desc) - self._redraw() - - def clear(self, agent_id: str) -> None: - # On completion: erase the live region, freeze a single-line summary - # for this agent ("✓ research: … (stats)") above the live region so - # the user sees each sub-agent finish cleanly without the tool-call - # noise, then redraw remaining live agents. - agent = self._agents.pop(agent_id, None) - self._erase() - if agent is not None: - width = max(10, _console.width) - line = _clip_to_width(self._render_completion_line(agent), width) - _console.file.write(line + "\n") - _console.file.flush() - self._lines_on_screen = 0 - if not self._agents: - if self._ticker_task: - self._ticker_task.cancel() - self._ticker_task = None - else: - self._redraw() - - @staticmethod - def _render_completion_line(agent: dict) -> str: - stats = SubAgentDisplayManager._format_stats(agent) - label = agent["label"] - # dim green check + dim label; stats in parens - line = f"{_I}\033[38;2;120;200;140m✓\033[0m \033[2m{label}\033[0m" - if stats: - line += f" \033[2m({stats})\033[0m" - return line - - async def _tick(self) -> None: - import asyncio - - try: - while True: - await asyncio.sleep(1.0) - if self._agents: - self._redraw() - except asyncio.CancelledError: - pass - - @staticmethod - def _format_stats(agent: dict) -> str: - import time - - start = agent["start_time"] - if start is None: - return "" - elapsed = time.monotonic() - start - if elapsed < 60: - time_str = f"{elapsed:.0f}s" - else: - time_str = f"{elapsed / 60:.0f}m {elapsed % 60:.0f}s" - tok = agent["token_count"] - tok_str = f"{tok / 1000:.1f}k" if tok >= 1000 else str(tok) - return f"{agent['tool_count']} tool uses · {tok_str} tokens · {time_str}" - - def _erase(self) -> None: - if self._lines_on_screen > 0: - f = _console.file - for _ in range(self._lines_on_screen): - f.write("\033[A\033[K") - f.flush() - - def _render_agent_lines(self, agent: dict, compact: bool = False) -> list[str]: - """Render one agent's block. - - compact=True → single line (label + stats + most-recent tool name); - compact=False → header + up to _MAX_VISIBLE rolling tool-call lines. - We use compact mode when multiple agents are live so the total live - region stays small enough to fit on one screen. Otherwise cursor-up - can't reach lines that have scrolled into scrollback, and every - redraw pollutes history with a stale copy. - """ - stats = self._format_stats(agent) - label = agent["label"] - header = f"{_I}\033[38;2;255;200;80m▸ {label}\033[0m" - if stats: - header += f" \033[2m({stats})\033[0m" - if compact: - latest = agent["calls"][-1] if agent["calls"] else "" - if latest: - # Strip long json tails for the inline view - short = latest.split(" ")[0] if " " in latest else latest - header += f" \033[2m·\033[0m \033[2m{short}\033[0m" - return [header] - lines = [header] - visible = agent["calls"][-self._MAX_VISIBLE :] - for desc in visible: - lines.append(f"{_I} \033[2m{desc}\033[0m") - return lines - - def _redraw(self) -> None: - f = _console.file - self._erase() - compact = len(self._agents) > 1 - width = max(10, _console.width) - lines: list[str] = [] - for agent in self._agents.values(): - for ln in self._render_agent_lines(agent, compact=compact): - lines.append(_clip_to_width(ln, width)) - for line in lines: - f.write(line + "\n") - f.flush() - self._lines_on_screen = len(lines) - - -_subagent_display = SubAgentDisplayManager() - - -def print_tool_log(tool: str, log: str, agent_id: str = "", label: str = "") -> None: - """Handle tool log events — sub-agent calls get the rolling display.""" - if tool == "research": - aid = agent_id or "research" - if log == "Starting research sub-agent...": - _subagent_display.start(aid, label or "research") - elif log == "Research complete.": - _subagent_display.clear(aid) - elif log.startswith("tokens:"): - _subagent_display.set_tokens(aid, int(log[7:])) - elif log.startswith("tools:"): - _subagent_display.set_tool_count(aid, int(log[6:])) - else: - _subagent_display.add_call(aid, log) - else: - _console.print(f"{_I}[dim]{tool}: {log}[/dim]") - - -# ── Messages ─────────────────────────────────────────────────────────── - - -async def print_markdown( - text: str, - cancel_event: "asyncio.Event | None" = None, - instant: bool = False, -) -> None: - import asyncio - import io - import random - - from rich.padding import Padding - - _console.print() - - # Render markdown to a string buffer so we can type it out - buf = io.StringIO() - # Important: StringIO is not a TTY, so Rich would normally strip styles. - # Force terminal rendering so ANSI style codes are preserved for typewriter output. - buf_console = Console( - file=buf, - width=_console.width, - highlight=False, - theme=_THEME, - force_terminal=True, - color_system=_console.color_system or "truecolor", - ) - buf_console.print(Padding(Markdown(text), (0, 0, 0, 2))) - rendered = buf.getvalue() - - # Strip trailing whitespace from each line so we don't type across the full width - lines = rendered.split("\n") - rendered = "\n".join(line.rstrip() for line in lines) - - f = _console.file - - # Headless / non-interactive: dump the rendered markdown in one write. - if instant: - f.write(rendered) - f.write("\n") - f.flush() - return - - # CRT typewriter effect — async so the event loop can service signal - # handlers (Ctrl+C during streaming) between characters. If cancelled - # mid-type, stop cleanly: write an ANSI reset so half-open color state - # doesn't bleed onto the "interrupted" line, and return. - rng = random.Random(42) - cancelled = False - for ch in rendered: - if cancel_event is not None and cancel_event.is_set(): - cancelled = True - break - f.write(ch) - f.flush() - if ch == "\n" or ch == " ": - await asyncio.sleep(0.002) - elif rng.random() < 0.03: - await asyncio.sleep(0.015) - else: - await asyncio.sleep(0.004) - f.write("\033[0m\n" if cancelled else "\n") - f.flush() - - -def print_error(message: str) -> None: - _console.print(f"\n{_I}[bold red]Error:[/bold red] {message}") - - -def print_turn_complete() -> None: - pass # no separator — clean output - - -def print_interrupted() -> None: - _console.print(f"\n{_I}[dim italic]interrupted[/dim italic]") - - -def print_compacted(old_tokens: int, new_tokens: int) -> None: - _console.print(f"{_I}[dim]context compacted: {old_tokens:,} → {new_tokens:,} tokens[/dim]") - - -# ── Approval ─────────────────────────────────────────────────────────── - - -def print_approval_header(count: int) -> None: - label = f"Approval required — {count} item{'s' if count != 1 else ''}" - _console.print() - _console.print( - f"{_I}", - Panel(f"[bold yellow]{label}[/bold yellow]", border_style="yellow", expand=False), - ) - - -def print_approval_item(index: int, total: int, tool_name: str, operation: str) -> None: - _console.print(f"\n{_I}[bold]\\[{index}/{total}][/bold] [tool.name]{tool_name}[/tool.name] {operation}") - - -def print_yolo_approve(count: int) -> None: - _console.print(f"{_I}[bold yellow]yolo →[/bold yellow] auto-approved {count} item(s)") - - -# ── Help ─────────────────────────────────────────────────────────────── - -HELP_TEXT = f"""\ -{_I}[bold]Commands[/bold] -{_I} [cyan]/help[/cyan] Show this help -{_I} [cyan]/undo[/cyan] Undo last turn -{_I} [cyan]/compact[/cyan] Compact context window -{_I} [cyan]/model[/cyan] [id] Show available models or switch -{_I} [cyan]/effort[/cyan] [level] Reasoning effort (minimal|low|medium|high|xhigh|max|off) -{_I} [cyan]/yolo[/cyan] Toggle auto-approve mode -{_I} [cyan]/status[/cyan] Current model & turn count -{_I} [cyan]/quit[/cyan] Exit""" - - -def print_help() -> None: - _console.print() - _console.print(HELP_TEXT) - _console.print() - - -# ── Plan display ─────────────────────────────────────────────────────── - - -def format_plan_display() -> str: - """Format the current plan for display.""" - from agent.tools.plan_tool import get_current_plan - - plan = get_current_plan() - if not plan: - return "" - - completed = [t for t in plan if t["status"] == "completed"] - in_progress = [t for t in plan if t["status"] == "in_progress"] - pending = [t for t in plan if t["status"] == "pending"] - - lines = [] - for t in completed: - lines.append(f"{_I}[green]✓[/green] [dim]{t['content']}[/dim]") - for t in in_progress: - lines.append(f"{_I}[yellow]▸[/yellow] {t['content']}") - for t in pending: - lines.append(f"{_I}[dim]○ {t['content']}[/dim]") - - summary = f"[dim]{len(completed)}/{len(plan)} done[/dim]" - lines.append(f"{_I}{summary}") - return "\n".join(lines) - - -def print_plan() -> None: - plan_str = format_plan_display() - if plan_str: - _console.print(plan_str) - - -# ── Formatting for plan_tool output (used by plan_tool handler) ──────── - - -def format_plan_tool_output(todos: list) -> str: - if not todos: - return "Plan is empty." - - lines = ["Plan updated:", ""] - completed = [t for t in todos if t["status"] == "completed"] - in_progress = [t for t in todos if t["status"] == "in_progress"] - pending = [t for t in todos if t["status"] == "pending"] - - for t in completed: - lines.append(f" [x] {t['id']}. {t['content']}") - for t in in_progress: - lines.append(f" [~] {t['id']}. {t['content']}") - for t in pending: - lines.append(f" [ ] {t['id']}. {t['content']}") - - lines.append(f"\n{len(completed)}/{len(todos)} done") - return "\n".join(lines) - - -# ── Internal helpers ─────────────────────────────────────────────────── - - -def _truncate(text: str, max_lines: int = 6) -> str: - lines = text.split("\n") - if len(lines) <= max_lines: - return text - return "\n".join(lines[:max_lines]) + f"\n... ({len(lines) - max_lines} more lines)" diff --git a/src/repoman/utils/ui/logo_display.py b/src/repoman/utils/ui/logo_display.py index 305bec1..375ce7c 100644 --- a/src/repoman/utils/ui/logo_display.py +++ b/src/repoman/utils/ui/logo_display.py @@ -21,9 +21,14 @@ """Braille-character canvas for high-resolution terminal graphics. Each terminal cell maps to a 2x4 dot grid using Unicode braille characters -(U+2800–U+28FF), giving 2× horizontal and 4× vertical resolution. +(U+2800-U+28FF), giving 2x horizontal and 4x vertical resolution. """ +_TARGET_EPSILON = 1.5 +_AMBIENT_ALPHA_THRESHOLD = 0.25 + +_Bounds = tuple[int, int, int, int] + def settle_curve(progress: float, sharpness: float = 4.0) -> float: """Return noise amount in range 1..0 for normalized progress 0..1.""" @@ -46,7 +51,8 @@ def settle_curve(progress: float, sharpness: float = 4.0) -> float: class BrailleCanvas: """A pixel canvas that renders to braille characters.""" - def __init__(self, term_width: int, term_height: int): + def __init__(self, term_width: int, term_height: int) -> None: + """Initialize a terminal-sized braille canvas.""" self.term_width = term_width self.term_height = term_height self.pixel_width = term_width * 2 @@ -54,16 +60,19 @@ def __init__(self, term_width: int, term_height: int): self._buf = bytearray(term_width * term_height) def clear(self) -> None: + """Clear all pixels from the canvas.""" for i in range(len(self._buf)): self._buf[i] = 0 def set_pixel(self, x: int, y: int) -> None: + """Set one high-resolution pixel if it is within bounds.""" if 0 <= x < self.pixel_width and 0 <= y < self.pixel_height: cx, rx = divmod(x, 2) cy, ry = divmod(y, 4) self._buf[cy * self.term_width + cx] |= _DOT_MAP[ry][rx] def render(self) -> list[str]: + """Render the canvas as terminal lines.""" lines = [] for row in range(self.term_height): offset = row * self.term_width @@ -72,13 +81,13 @@ def render(self) -> list[str]: return lines -# ── Bitmap font (5×7 uppercase + digits) ────────────────────────────── +# ── Bitmap font (5x7 uppercase + digits) ────────────────────────────── _FONT: dict[str, list[str]] = {} def _define_font() -> None: - """Define a simple 5×7 bitmap font for uppercase ASCII.""" + """Define a simple 5x7 bitmap font for uppercase ASCII.""" glyphs = { "A": [" ## ", "# #", "# #", "####", "# #", "# #", "# #"], "B": ["### ", "# #", "# #", "### ", "# #", "# #", "### "], @@ -145,19 +154,30 @@ def text_to_pixels(text: str, scale: int = 1) -> list[tuple[int, int]]: class Particle: + """A moving particle that converges toward a target pixel.""" + __slots__ = ("delay", "phase", "target_x", "target_y", "vx", "vy", "x", "y") - def __init__(self, x: float, y: float, target_x: float, target_y: float, delay: float = 0): + def __init__( + self, + x: float, + y: float, + target_x: float, + target_y: float, + delay: float = 0, + phase: float = 0, + ) -> None: + """Initialize a particle and its convergence target.""" self.x = x self.y = y self.target_x = target_x self.target_y = target_y self.vx = 0.0 self.vy = 0.0 - self.phase = random.uniform(0, math.pi * 2) + self.phase = phase self.delay = delay - def update_converge(self, t: float, strength: float = 0.08, damping: float = 0.92): + def update_converge(self, t: float, strength: float = 0.08, damping: float = 0.92) -> None: """Move toward target with spring-like physics.""" if t < self.delay: # Still in swirl phase @@ -183,7 +203,8 @@ def update_converge(self, t: float, strength: float = 0.08, damping: float = 0.9 @property def at_target(self) -> bool: - return abs(self.x - self.target_x) < 1.5 and abs(self.y - self.target_y) < 1.5 + """Return whether the particle is visually close to its target.""" + return abs(self.x - self.target_x) < _TARGET_EPSILON and abs(self.y - self.target_y) < _TARGET_EPSILON def run_particle_logo(console: Console, hold_seconds: float = 1.5) -> None: @@ -200,7 +221,7 @@ def run_particle_logo(console: Console, hold_seconds: float = 1.5) -> None: text_pixels_line2 = text_to_pixels("MANager", scale=2) # Calculate dimensions for centering - def get_bounds(pixels): + def get_bounds(pixels: list[tuple[int, int]]) -> _Bounds: if not pixels: return 0, 0, 0, 0 xs = [p[0] for p in pixels] @@ -233,11 +254,11 @@ def get_bounds(pixels): sampled_targets = all_targets[::step] # Create particles at random edge positions - rng = random.Random(42) + rng = random.Random(42) # noqa: S311 - deterministic animation jitter, not security-sensitive. particles = [] pw, ph = canvas.pixel_width, canvas.pixel_height - for i, (tx, ty) in enumerate(sampled_targets): + for tx, ty in sampled_targets: # Spawn from random edge side = rng.choice(["top", "bottom", "left", "right"]) if side == "top": @@ -250,7 +271,7 @@ def get_bounds(pixels): sx, sy = rng.uniform(pw + 5, pw + 20), rng.uniform(0, ph) delay = rng.uniform(0, 0.4) # staggered start - p = Particle(sx, sy, tx, ty, delay=delay) + p = Particle(sx, sy, tx, ty, delay=delay, phase=rng.uniform(0, math.pi * 2)) # Initial velocity — gentle swirl angle = math.atan2(ph / 2 - sy, pw / 2 - sx) + rng.gauss(0, 0.8) speed = rng.uniform(1.0, 2.5) @@ -263,7 +284,7 @@ def get_bounds(pixels): for _ in range(200): ax = rng.uniform(0, pw) ay = rng.uniform(0, ph) - ap = Particle(ax, ay, ax, ay) + ap = Particle(ax, ay, ax, ay, phase=rng.uniform(0, math.pi * 2)) ap.vx = rng.gauss(0, 1) ap.vy = rng.gauss(0, 1) ambient.append(ap) @@ -293,7 +314,7 @@ def get_bounds(pixels): else: fade = (frame - converge_frames) / hold_frames alpha = (0.3 + 0.2 * math.sin(t * 2 + ap.phase)) * (1 - fade) - if alpha > 0.25: + if alpha > _AMBIENT_ALPHA_THRESHOLD: canvas.set_pixel(int(ap.x), int(ap.y)) if frame < converge_frames: @@ -323,7 +344,7 @@ def get_bounds(pixels): canvas.set_pixel(int(jx), int(jy)) canvas.set_pixel(int(p.target_x), int(p.target_y)) - r, g, b = style_final.color.triplet + r, g, b = cool_ramp(1.0) # Render with color lines = canvas.render() @@ -356,6 +377,7 @@ def get_bounds(pixels): def main() -> int: + """Run the logo animation from the command line.""" console = Console(color_system="auto") try: From 0cad4b0f9016bef4f89ff1380731f68fdab6b189 Mon Sep 17 00:00:00 2001 From: "Leandro G. Almeida" Date: Thu, 28 May 2026 16:31:36 -0700 Subject: [PATCH 5/6] =?UTF-8?q?=F0=9F=90=9B=20cli:=20skip=20startup=20art?= =?UTF-8?q?=20for=20redirected=20output?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/repoman/cli/main_cli.py | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/src/repoman/cli/main_cli.py b/src/repoman/cli/main_cli.py index 6a70aec..0785462 100644 --- a/src/repoman/cli/main_cli.py +++ b/src/repoman/cli/main_cli.py @@ -31,6 +31,7 @@ from pathlib import Path from pydantic import ValidationError +from rich.console import Console from rich.text import Text from typer import Context, Exit, Option, Typer @@ -78,6 +79,12 @@ def _debug_info_callback(value: bool) -> None: raise Exit(0) +def _should_run_startup_logo(console: Console) -> bool: + stream = getattr(console, "file", None) + is_tty = bool(getattr(stream, "isatty", lambda: False)()) + return bool(console.is_terminal and is_tty) + + @cli_app.callback(invoke_without_command=True, no_args_is_help=True) def main( ctx: Context, @@ -119,13 +126,14 @@ def main( repoman generator add my-command --project-dir ./my-app """ logger, console = get_logger_console() - run_particle_logo( - console=console, - hold_seconds=1.5, - ) - # Clear screen for CRT boot — starts from top - console.file.write("\033[2J\033[H") - console.file.flush() + if _should_run_startup_logo(console): + run_particle_logo( + console=console, + hold_seconds=1.5, + ) + # Clear screen for CRT boot — starts from top + console.file.write("\033[2J\033[H") + console.file.flush() config: Config | None = None try: config = Config.load(custom_path=Path(config_path) if config_path else None) From a51064689da477ec3e254cce32653c1e124d52da Mon Sep 17 00:00:00 2001 From: "Leandro G. Almeida" Date: Fri, 29 May 2026 15:47:22 -0700 Subject: [PATCH 6/6] =?UTF-8?q?=E2=9C=A8=20tests(utils):=20Add=20task=20tr?= =?UTF-8?q?ee,=20pad,=20tracker=20and=20logging=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../utils/logging.py.jinja | 13 ++- src/repoman/utils/logging.py | 6 +- src/repoman/utils/ui/console.py | 13 ++- tests/template_testing.py | 10 +- tests/test_utils/test_logging.py | 56 +++++++++++ tests/test_utils/test_logo_display.py | 98 +++++++++++++++++++ tests/test_utils/test_task_tree_display.py | 92 +++++++++++++++++ tests/test_utils/test_template_testing.py | 26 +++++ 8 files changed, 306 insertions(+), 8 deletions(-) create mode 100644 tests/test_utils/test_logo_display.py create mode 100644 tests/test_utils/test_task_tree_display.py diff --git a/src/repoman/main_template/src/{{python_package_import_name}}/utils/logging.py.jinja b/src/repoman/main_template/src/{{python_package_import_name}}/utils/logging.py.jinja index 2c865f4..fc8e966 100644 --- a/src/repoman/main_template/src/{{python_package_import_name}}/utils/logging.py.jinja +++ b/src/repoman/main_template/src/{{python_package_import_name}}/utils/logging.py.jinja @@ -30,6 +30,15 @@ def _is_running_in_pytest() -> bool: ) +def _refresh_console_file(console: Console) -> Console: + """Keep a cached Rich console bound to a writable output stream.""" + stream = getattr(console, "file", None) + stream_is_closed = bool(getattr(stream, "closed", False)) + if _is_running_in_pytest() or stream_is_closed: + console.file = sys.stdout + return console + + def _set_up_logger( name: str = "{{python_package_import_name}}", console: Console | None = None, @@ -56,6 +65,8 @@ def _set_up_logger( if len(module_logger.handlers) > 0: for handler in module_logger.handlers: if handler.get_name() == "rich": + rich_handler = cast("RichHandler", handler) + _refresh_console_file(rich_handler.console) # Set the log level even for existing loggers module_logger.setLevel(level=log_level) return module_logger @@ -160,7 +171,7 @@ def get_logger_console( if handler.get_name() == "rich": rich_handler: RichHandler = cast("RichHandler", handler) # use console from handler - console = rich_handler.console + console = _refresh_console_file(rich_handler.console) return logger, console # If no console was found and none was provided, create a new one diff --git a/src/repoman/utils/logging.py b/src/repoman/utils/logging.py index 8c524b5..043cf17 100644 --- a/src/repoman/utils/logging.py +++ b/src/repoman/utils/logging.py @@ -11,7 +11,7 @@ from rich.logging import RichHandler from repoman.config import Config -from repoman.utils.ui.console import get_console +from repoman.utils.ui.console import get_console, refresh_console_file def _set_up_logger( @@ -40,6 +40,8 @@ def _set_up_logger( if len(module_logger.handlers) > 0: for handler in module_logger.handlers: if handler.get_name() == "rich": + rich_handler = cast("RichHandler", handler) + refresh_console_file(rich_handler.console) # Set the log level even for existing loggers module_logger.setLevel(level=log_level) return module_logger @@ -134,7 +136,7 @@ def get_logger_console( if handler.get_name() == "rich": rich_handler: RichHandler = cast("RichHandler", handler) # use console from handler - console = rich_handler.console + console = refresh_console_file(rich_handler.console) return logger, console if console is None: diff --git a/src/repoman/utils/ui/console.py b/src/repoman/utils/ui/console.py index 2b4cfa0..311d214 100644 --- a/src/repoman/utils/ui/console.py +++ b/src/repoman/utils/ui/console.py @@ -30,9 +30,7 @@ def get_console() -> Console: console = _console_cache[0] if console is not None: - if _is_running_in_pytest(): - console.file = sys.stdout - return console + return refresh_console_file(console) if _is_running_in_pytest(): # Use a console that outputs plain text (no colors/formatting) @@ -47,4 +45,13 @@ def get_console() -> Console: console = Console(theme=set_theme("dark")) _console_cache[0] = console + return refresh_console_file(console) + + +def refresh_console_file(console: Console) -> Console: + """Keep a cached Rich console bound to a writable output stream.""" + stream = getattr(console, "file", None) + stream_is_closed = bool(getattr(stream, "closed", False)) + if _is_running_in_pytest() or stream_is_closed: + console.file = sys.stdout return console diff --git a/tests/template_testing.py b/tests/template_testing.py index b7105f6..62754ef 100644 --- a/tests/template_testing.py +++ b/tests/template_testing.py @@ -15,10 +15,16 @@ from copier.errors import CopierError from repoman.utils.logging import get_logger_console +from repoman.utils.ui.console import refresh_console_file logger, console = get_logger_console(__name__) +def _print_status(message: str) -> None: + """Print through a console refreshed against pytest's current capture stream.""" + refresh_console_file(console).print(message) + + def _slugify(value: str, separator: str = "-") -> str: """Slugify a string (convert to URL-friendly format). @@ -170,13 +176,13 @@ def instantiate_template( logger.info(f"Instantiating template from {template_path} to {project_dir}") run_copy(**copier_options) logger.info(f"Template instantiated successfully at {project_dir}") - console.print(f"[green]✓[/green] Template instantiated at {project_dir}") + _print_status(f"[green]✓[/green] Template instantiated at {project_dir}") # Format generated Python so instantiated project passes make format-check _run_ruff_format(project_dir) except CopierError as e: error_msg = f"Failed to instantiate template: {e}" logger.exception(error_msg) - console.print(f"[red]✗[/red] {error_msg}") + _print_status(f"[red]✗[/red] {error_msg}") raise else: return project_dir diff --git a/tests/test_utils/test_logging.py b/tests/test_utils/test_logging.py index 36b314a..189188d 100644 --- a/tests/test_utils/test_logging.py +++ b/tests/test_utils/test_logging.py @@ -1,7 +1,9 @@ """Tests for logging utilities.""" +import io import logging import os +import sys import tempfile from logging import DEBUG, INFO, Logger from pathlib import Path @@ -10,6 +12,7 @@ import pytest from rich.console import Console +from rich.logging import RichHandler from repoman.utils.logging import ( _attach_rotating_file_handler, @@ -128,6 +131,27 @@ def test_set_up_logger_existing_rich_handler(self) -> None: # (pytest adds additional handlers for test capture) assert any(h.get_name() == "rich" for h in logger1.handlers) + def test_set_up_logger_refreshes_closed_rich_console(self) -> None: + """Existing Rich handlers should not keep a closed capture stream.""" + stale_stream = io.StringIO() + stale_console = Console(file=stale_stream, force_terminal=False, no_color=True) + logger = _set_up_logger("test_logger_closed_console", console=stale_console) + + try: + stale_stream.close() + + refreshed_logger = _set_up_logger("test_logger_closed_console") + rich_handler = next(handler for handler in refreshed_logger.handlers if handler.get_name() == "rich") + assert isinstance(rich_handler, RichHandler) + refreshed_console = rich_handler.console + + assert refreshed_console.file is sys.stdout + refreshed_console.print("console still writable") + finally: + for handler in logger.handlers[:]: + handler.close() + logger.removeHandler(handler) + def test_set_up_logger_environment_log_level(self) -> None: """Test that environment variable affects log level.""" original_level = os.environ.get("_REPOMAN_LOG_LEVEL") @@ -321,6 +345,38 @@ def test_get_logger_console_rich_handler_console(self) -> None: handler = rich_handlers[0] assert hasattr(handler, "console") + def test_get_logger_console_refreshes_closed_handler_console(self) -> None: + """Child loggers should receive a writable console from the root handler.""" + root_logger = logging.getLogger("repoman") + original_handlers = root_logger.handlers[:] + original_level = root_logger.level + + stale_stream = io.StringIO() + stale_console = Console(file=stale_stream, force_terminal=False, no_color=True) + stale_handler = RichHandler(rich_tracebacks=True, console=stale_console) + stale_handler.set_name("rich") + + try: + for handler in root_logger.handlers[:]: + root_logger.removeHandler(handler) + root_logger.addHandler(stale_handler) + root_logger.setLevel(INFO) + stale_stream.close() + + logger, console = get_logger_console("test_logger_closed_child") + + assert logger.name == "test_logger_closed_child" + assert console.file is sys.stdout + console.print("console still writable") + finally: + for handler in root_logger.handlers[:]: + root_logger.removeHandler(handler) + if handler not in original_handlers: + handler.close() + for handler in original_handlers: + root_logger.addHandler(handler) + root_logger.setLevel(original_level) + def test_get_logger_console_fallback_return(self) -> None: """Test that get_logger_console falls back to default return when no rich handler.""" # Create a logger without rich handler to test fallback diff --git a/tests/test_utils/test_logo_display.py b/tests/test_utils/test_logo_display.py new file mode 100644 index 0000000..630f2d9 --- /dev/null +++ b/tests/test_utils/test_logo_display.py @@ -0,0 +1,98 @@ +"""Tests for the terminal logo renderer.""" + +# ruff: noqa: D102,D103,D107 + +import io +from typing import Any, Self + +from rich.console import Console + +from repoman.utils.ui import logo_display +from repoman.utils.ui.logo_display import BrailleCanvas, Particle, settle_curve, text_to_pixels + + +class DummyLive: + """Small stand-in for ``rich.live.Live`` that records frame updates.""" + + updates: list[Any] + + def __init__(self, *_args: Any, **_kwargs: Any) -> None: + self.updates = [] + + def __enter__(self) -> Self: + return self + + def __exit__(self, *_args: object) -> None: + return None + + def update(self, renderable: Any) -> None: + self.updates.append(renderable) + + +def test_settle_curve_clamps_progress() -> None: + assert settle_curve(-1) == settle_curve(0) + assert settle_curve(2) == settle_curve(1) + assert settle_curve(0) > settle_curve(0.5) > settle_curve(1) + + +def test_braille_canvas_sets_pixels_and_clears() -> None: + canvas = BrailleCanvas(term_width=2, term_height=1) + + canvas.set_pixel(0, 0) + canvas.set_pixel(1, 3) + canvas.set_pixel(-1, 0) + canvas.set_pixel(4, 0) + + rendered = canvas.render() + assert rendered[0][0] != chr(0x2800) + + canvas.clear() + + assert canvas.render() == [chr(0x2800) * 2] + + +def test_text_to_pixels_scales_known_glyphs_and_skips_unknowns() -> None: + pixels = text_to_pixels("A?", scale=2) + + assert (2, 0) in pixels + assert max(x for x, _y in pixels) > 6 + assert max(y for _x, y in pixels) == 13 + + +def test_particle_swirl_then_converges() -> None: + particle = Particle(0, 0, target_x=10, target_y=0, delay=1, phase=0) + + particle.update_converge(0.5) + assert particle.vx > 0 + assert not particle.at_target + + for _ in range(30): + particle.update_converge(2.0, strength=0.2, damping=0.6) + + assert particle.at_target + + +def test_run_particle_logo_renders_final_frame(monkeypatch: Any) -> None: + live_instances: list[DummyLive] = [] + + def make_live(*args: Any, **kwargs: Any) -> DummyLive: + live = DummyLive(*args, **kwargs) + live_instances.append(live) + return live + + monkeypatch.setattr(logo_display, "Live", make_live) + monkeypatch.setattr(logo_display.time, "sleep", lambda _seconds: None) + monkeypatch.setattr( + logo_display, + "text_to_pixels", + lambda _text, scale=1: [(0, 0), (1 * scale, 0), (0, 1 * scale)], + ) + + output = io.StringIO() + console = Console(file=output, force_terminal=False, width=30, height=12, record=True) + + logo_display.run_particle_logo(console, hold_seconds=0.1) + + assert live_instances + assert live_instances[0].updates + assert output.getvalue().strip() diff --git a/tests/test_utils/test_task_tree_display.py b/tests/test_utils/test_task_tree_display.py new file mode 100644 index 0000000..f769ea5 --- /dev/null +++ b/tests/test_utils/test_task_tree_display.py @@ -0,0 +1,92 @@ +"""Tests for task tree display helpers.""" + +# ruff: noqa: D103 + +from rich.text import Text + +from repoman.utils.task_tree_display import Padder, TaskTree, UpdateTracker + + +def test_task_tree_orders_stdout_before_stderr_when_stderr_arrives_first() -> None: + tree = TaskTree("make test") + + tree.set_stderr(["warning"]) + tree.set_stdout(["ok", "done"]) + + assert tree.height() == 6 + assert [str(child.label) for child in tree.tree.children] == ["stdout", "stderr"] + stdout_text = tree.tree.children[0].children[0].label + stderr_text = tree.tree.children[1].children[0].label + + assert isinstance(stdout_text, Text) + assert isinstance(stderr_text, Text) + assert stdout_text.plain == "ok\ndone" + assert stderr_text.plain == "warning" + + +def test_task_tree_reset_and_empty_updates() -> None: + tree = TaskTree("ruff check") + + tree.set_stdout([]) + tree.set_stderr([]) + assert tree.height() == 1 + + tree.set_stdout(["clean"]) + tree.reset() + + assert tree.height() == 1 + assert [str(child.label) for child in tree.tree.children] == ["ruff check"] + + +def test_padder_tracks_tallest_tree() -> None: + short = TaskTree("short") + tall = TaskTree("tall") + tall.set_stdout(["a", "b", "c"]) + padder = Padder() + + assert padder.get_padding(tall) == 0 + assert padder.get_padding(short) == tall.height() - short.height() + + +def test_update_tracker_refreshes_for_state_change_and_timeout() -> None: + calls = 0 + + def mark_updated() -> None: + nonlocal calls + calls += 1 + + tracker = UpdateTracker(max_update_timeout_ms=100, min_update_ms=10) + tracker._last_update_timestamp_ms = 0 + + tracker.update(mark_updated, ["stdout"]) + tracker._last_update_timestamp_ms = 0 + tracker.update(mark_updated, ["stdout"]) + + assert calls == 2 + + +def test_update_tracker_skips_when_state_is_unchanged_and_too_recent() -> None: + calls: list[str] = [] + tracker = UpdateTracker(max_update_timeout_ms=10_000, min_update_ms=10_000) + tracker._last_update_state = ["same"] + + def update_fn() -> None: + calls.append("updated") + + tracker.update(update_fn, ["same"]) + + assert calls == [] + + +def test_update_tracker_private_update_sets_state() -> None: + tracker = UpdateTracker() + calls: list[str] = [] + + def update_fn() -> None: + calls.append("updated") + + tracker._update(123.0, update_fn, ["new"]) + + assert calls == ["updated"] + assert tracker._last_update_timestamp_ms == 123.0 + assert tracker._last_update_state == ["new"] diff --git a/tests/test_utils/test_template_testing.py b/tests/test_utils/test_template_testing.py index b266419..a94b3a8 100644 --- a/tests/test_utils/test_template_testing.py +++ b/tests/test_utils/test_template_testing.py @@ -1,11 +1,15 @@ """Unit tests for template_testing utility functions.""" +import io +import sys from pathlib import Path from typing import Any import pytest from copier.errors import CopierError +from rich.console import Console +from tests import template_testing from tests.template_testing import cleanup_project_artifacts, instantiate_template @@ -111,6 +115,28 @@ def test_instantiate_template_custom_data(tmp_path: Path, mock_template_structur assert result.exists() +def test_instantiate_template_refreshes_closed_console( + tmp_path: Path, + mock_template_structure: Any, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Template instantiation should not fail when pytest closes a cached console stream.""" + stale_stream = io.StringIO() + stale_console = Console(file=stale_stream, force_terminal=False, no_color=True) + monkeypatch.setattr(template_testing, "console", stale_console) + stale_stream.close() + + result = instantiate_template( + output_dir=tmp_path, + template_path=mock_template_structure, + project_name="test-project", + force=True, + ) + + assert result.exists() + assert stale_console.file is sys.stdout + + def test_instantiate_template_invalid_path(tmp_path: Path) -> None: """Test that instantiate_template raises error for invalid template path.""" invalid_path = tmp_path / "nonexistent-template"