#!/usr/bin/env python3 """ CLI Feedback System - Non-blocking Status Display and Progress Tracking Provides responsive feedback to the user while tasks run in the background: - Immediate job confirmation with job_id - Live progress indicators - Status polling without blocking - Pretty-printed status displays - Multi-task tracking """ import json import sys from typing import Dict, Optional, List from datetime import datetime from pathlib import Path class Colors: """ANSI color codes for terminal output""" GREEN = "\033[92m" YELLOW = "\033[93m" RED = "\033[91m" BLUE = "\033[94m" CYAN = "\033[96m" GRAY = "\033[90m" BOLD = "\033[1m" RESET = "\033[0m" @staticmethod def status_color(status: str) -> str: """Get color for status""" colors = { "dispatched": Colors.CYAN, "starting": Colors.BLUE, "running": Colors.YELLOW, "completed": Colors.GREEN, "failed": Colors.RED, "killed": Colors.RED, "stalled": Colors.YELLOW, } return colors.get(status, Colors.GRAY) class ProgressBar: """ASCII progress bar renderer""" @staticmethod def render(progress: int, width: int = 20) -> str: """Render progress bar""" filled = int(width * progress / 100) bar = "█" * filled + "░" * (width - filled) return f"[{bar}] {progress}%" class CLIFeedback: """Non-blocking feedback system for task dispatch""" @staticmethod def job_dispatched(job_id: str, project: str, task: str, show_details: bool = False) -> None: """Show immediate feedback when job is dispatched""" print(f"\n{Colors.GREEN}{Colors.BOLD}✓ Dispatched{Colors.RESET}") print(f" {Colors.BOLD}Job ID:{Colors.RESET} {job_id}") print(f" {Colors.BOLD}Project:{Colors.RESET} {project}") if show_details and len(task) <= 60: print(f" {Colors.BOLD}Task:{Colors.RESET} {task}") elif show_details and len(task) > 60: print(f" {Colors.BOLD}Task:{Colors.RESET} {task[:57]}...") print(f"\n {Colors.GRAY}Use: {Colors.CYAN}luzia jobs{Colors.GRAY} to view status") print(f" {Colors.CYAN}luzia jobs {job_id}{Colors.GRAY} for details{Colors.RESET}\n") @staticmethod def show_status(status: Dict, show_full: bool = False) -> None: """Pretty-print job status""" job_id = status.get("id", "unknown") job_status = status.get("status", "unknown") progress = status.get("progress", 0) message = status.get("message", "") project = status.get("project", "") status_color = Colors.status_color(job_status) status_text = job_status.upper() # Single line summary bar = ProgressBar.render(progress) print(f" {status_color}{status_text:12}{Colors.RESET} {bar} {message}") if show_full: print(f"\n {Colors.BOLD}Details:{Colors.RESET}") print(f" Job ID: {job_id}") print(f" Project: {project}") print(f" Status: {job_status}") print(f" Progress: {progress}%") print(f" Message: {message}") # Show timestamps created = status.get("dispatched_at") updated = status.get("updated_at") if created: print(f" Created: {created}") if updated: print(f" Updated: {updated}") # Show exit code if completed if "exit_code" in status: print(f" Exit Code: {status['exit_code']}") @staticmethod def show_status_line(status: Dict) -> str: """Format status as single line for list views""" job_id = status.get("id", "unknown") job_status = status.get("status", "unknown") progress = status.get("progress", 0) message = status.get("message", "") project = status.get("project", "") status_color = Colors.status_color(job_status) status_text = f"{status_color}{job_status:10}{Colors.RESET}" progress_text = f"{progress:3d}%" project_text = f"{project:12}" # Truncate message if len(message) > 40: message = message[:37] + "..." return f" {job_id:13} {status_text} {progress_text} {project_text} {message}" @staticmethod def show_jobs_list(jobs: List[Dict]) -> None: """Pretty-print list of jobs""" if not jobs: print(f" {Colors.GRAY}No jobs found{Colors.RESET}") return print(f"\n {Colors.BOLD}Recent Jobs:{Colors.RESET}\n") print(f" {'Job ID':13} {'Status':10} {'Prog'} {'Project':12} Message") print(f" {'-' * 100}") for job in jobs[:20]: # Show last 20 print(CLIFeedback.show_status_line(job)) print() @staticmethod def show_concurrent_jobs(jobs: List[Dict], max_shown: int = 5) -> None: """Show summary of concurrent jobs""" if not jobs: return running = [j for j in jobs if j.get("status") == "running"] pending = [j for j in jobs if j.get("status") == "dispatched"] completed = [j for j in jobs if j.get("status") == "completed"] failed = [j for j in jobs if j.get("status") == "failed"] print(f"\n{Colors.BOLD}Task Summary:{Colors.RESET}") print(f" {Colors.YELLOW}Running:{Colors.RESET} {len(running)}") print(f" {Colors.CYAN}Pending:{Colors.RESET} {len(pending)}") print(f" {Colors.GREEN}Completed:{Colors.RESET} {len(completed)}") print(f" {Colors.RED}Failed:{Colors.RESET} {len(failed)}") if running: print(f"\n{Colors.BOLD}Currently Running:{Colors.RESET}") for job in running[:max_shown]: CLIFeedback.show_status(job) @staticmethod def spinner(status_func, interval: float = 0.1): """Show spinning indicator while waiting""" import itertools spinner = itertools.cycle(["|", "/", "-", "\\"]) while True: char = next(spinner) print(f"\r {char} ", end="", flush=True) result = status_func() if result: print(f"\r ✓ ", end="") return result sys.stdout.flush() class ResponseiveOutput: """Context manager for responsive output during long operations""" def __init__(self, message: str = "Processing"): self.message = message self.status = "running" def __enter__(self): print(f"{Colors.CYAN}{self.message}...{Colors.RESET}", end="", flush=True) return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is None: print(f"\r{Colors.GREEN}✓ {self.message}{Colors.RESET}") else: print(f"\r{Colors.RED}✗ {self.message} ({exc_type.__name__}){Colors.RESET}") return False def update(self, message: str): """Update the message""" self.message = message print(f"\r{Colors.CYAN}{self.message}...{Colors.RESET}", end="", flush=True) def format_duration(seconds: float) -> str: """Format duration in human-readable format""" if seconds < 60: return f"{int(seconds)}s" elif seconds < 3600: return f"{int(seconds // 60)}m {int(seconds % 60)}s" else: return f"{int(seconds // 3600)}h {int((seconds % 3600) // 60)}m"