diff --git a/langcode.py b/langcode.py index 09a7352..864c84d 100755 --- a/langcode.py +++ b/langcode.py @@ -111,14 +111,8 @@ def cmd_chat(args): if use_terminal: try: - import shutil - rows = min(40, shutil.get_terminal_size().lines) - columns = min(120, shutil.get_terminal_size().columns) - from terminal_ui import create_terminal_ui term = create_terminal_ui( - rows=rows, - columns=columns, scrollback=10000, prompt="❯ " ) diff --git a/terminal_ui.py b/terminal_ui.py index 2ba73e4..52ac3cb 100644 --- a/terminal_ui.py +++ b/terminal_ui.py @@ -1,16 +1,11 @@ """ Terminal UI con scrollback usando Pyte -Emula un terminale completo con scrollback buffer +Versione semplificata e robusta """ import os import sys -import pty -import select import signal -import time -import fcntl -import struct -import termios +import shutil from typing import Optional, Callable try: @@ -25,15 +20,13 @@ class TerminalUI: Terminal emulator con scrollback usando pyte. Features: - - Scrollback buffer completo - - Input/output come un vero terminale - - Segnali gestiti (Ctrl+C, Ctrl+L) + - Scrollback buffer completo (10000 righe) + - Output colorato con ANSI + - Resize del terminale """ def __init__( self, - rows: int = 24, - columns: int = 80, scrollback: int = 10000, prompt: str = "❯ " ): @@ -41,50 +34,70 @@ class TerminalUI: Inizializza il terminale emulator. Args: - rows: Righe del terminale - columns: Colonne del terminale scrollback: Numero di righe nello scrollback buffer prompt: Prompt da mostrare """ if not PYTE_AVAILABLE: raise ImportError("pyte non è installato. Esegui: pip install pyte") - self.rows = rows - self.columns = columns self.scrollback = scrollback self.prompt = prompt - # Crea lo screen pyte con scrollback - self.screen = pyte.Screen(columns, rows) + # Dimensioni terminale + self.term_rows, self.term_cols = shutil.get_terminal_size() + self.term_rows = max(24, min(self.term_rows, 50)) + self.term_cols = max(80, min(self.term_cols, 120)) + + # Crea lo screen pyte + self.screen = pyte.Screen(self.term_cols, self.term_rows) self.screen.scroll_history = scrollback self.stream = pyte.Stream(self.screen) - # Master/Slave PTY - self.master_fd: Optional[int] = None - self.slave_fd: Optional[int] = None - self.pid: Optional[int] = None + # Buffer per l'output completo (per scrollback) + self.output_buffer: list = [] # Flag per uscire self.running = False # Callback per l'agent self.agent_callback: Optional[Callable[[str], str]] = None - - # Signal handler - self._old_sigwinch: Optional[signal.Handler] = None - self._old_sigint: Optional[signal.Handler] = None - def _setup_pty(self): - """Crea pseudo-terminal""" - self.master_fd, self.slave_fd = pty.openpty() + def _redraw(self): + """Ridisegna lo schermo""" + # Clear schermo + sys.stdout.write("\033[2J\033[H") - # Imposta dimensioni iniziali - winsize = struct.pack('HHHH', self.rows, self.columns, 0, 0) - fcntl.ioctl(self.slave_fd, termios.TIOCSWINSZ, winsize) + # Header + header = f"╔{'═' * (self.term_cols - 4)}╗" + title = f" LangCode Agent - Terminal con Scrollback ({self.scrollback} righe) " + header2 = f"║{title:^{self.term_cols - 4}}║" + header3 = f"╚{'═' * (self.term_cols - 4)}╝" + + sys.stdout.write(f"\033[1;36m{header}\033[0m\n") + sys.stdout.write(f"\033[1;36m{header2}\033[0m\n") + sys.stdout.write(f"\033[1;36m{header3}\033[0m\n\n") + + # Mostra output buffer (solo le ultime righe che ci stanno) + visible_lines = self.term_rows - 10 # Lascia spazio per prompt e status + start_idx = max(0, len(self.output_buffer) - visible_lines) + + for line in self.output_buffer[start_idx:]: + sys.stdout.write(line) + + # Status bar + sys.stdout.write("\n") + status_bar = f"\033[1;34m─" * (self.term_cols - 30) + f" Scrollback: {len(self.output_buffer)} righe\033[0m" + sys.stdout.write(status_bar) + + # Prompt + sys.stdout.write(f"\n\033[1;35m{self.prompt}\033[0m ") + sys.stdout.flush() - def _restore_terminal(self): - """Ripristina terminale originale""" - pass # Il terminale è ripristinato automaticamente alla chiusura del PTY + def _add_output(self, text: str, color: str = "32"): + """Aggiunge output al buffer""" + lines = text.split('\n') + for line in lines: + self.output_buffer.append(f"\033[1;{color}m{line}\033[0m\n") def start(self, agent_callback: Callable[[str], str]): """ @@ -96,178 +109,90 @@ class TerminalUI: self.agent_callback = agent_callback self.running = True - # Registra signal handlers - self._old_sigwinch = signal.signal(signal.SIGWINCH, self._handle_resize) - self._old_sigint = signal.signal(signal.SIGINT, self._handle_sigint) - - try: - self._run_loop() - finally: + # Registra signal handler per Ctrl+C + def handle_sigint(signum, frame): + sys.stdout.write("\n\033[1;33m👋 Arrivederci!\033[0m\n") + sys.stdout.flush() self.running = False - # Ripristina signal handlers - if self._old_sigwinch: - signal.signal(signal.SIGWINCH, self._old_sigwinch) - if self._old_sigint: - signal.signal(signal.SIGINT, self._old_sigint) - - def _handle_resize(self, signum, frame): - """Gestisci resize del terminale""" - try: - winsize = struct.pack('HHHH', self.rows, self.columns, 0, 0) - if self.slave_fd is not None: - fcntl.ioctl(self.slave_fd, termios.TIOCSWINSZ, winsize) - except: - pass - - def _handle_sigint(self, signum, frame): - """Gestisci Ctrl+C""" - self._write_output("\n👋 Arrivederci!\n") - self.running = False - sys.exit(0) - - def _write_output(self, text: str): - """Scrive output al terminale""" - if self.master_fd is not None: - os.write(self.master_fd, text.encode('utf-8')) - - def _read_input(self, timeout: float = 0.1) -> str: - """Legge input disponibile""" - if self.master_fd is None: - return "" + sys.exit(0) + + old_handler = signal.signal(signal.SIGINT, handle_sigint) try: - ready, _, _ = select.select([self.master_fd], [], [], timeout) - if ready: - data = os.read(self.master_fd, 1024) - return data.decode('utf-8', errors='replace') - except: - pass - return "" - - def _run_loop(self): - """Loop principale""" - self._setup_pty() - - # Mostra prompt iniziale - self._draw_screen() - - input_buffer = "" - - while self.running: - # Leggi input - char = self._read_input(timeout=0.05) + # Mostra schermo iniziale + self._redraw() - if char: - # Feed a pyte - self.stream.feed(char) - input_buffer += char + while self.running: + try: + # Leggi input + user_input = input() + except EOFError: + break + except KeyboardInterrupt: + continue - # Quando l'utente preme Enter, processa comando - if '\r' in input_buffer or '\n' in input_buffer: - lines = input_buffer.replace('\r\n', '\n').replace('\r', '\n').split('\n') - for line in lines[:-1]: - if line.strip(): - self._process_command(line.strip()) - input_buffer = lines[-1] if lines[-1] else "" + if not user_input.strip(): + continue - # Ridisegna - self._draw_screen() - - # Piccolo sleep per non saturare CPU - time.sleep(0.01) - - def _process_command(self, command: str): - """Processa un comando""" - if not command.strip(): - return - - # Mostra il comando dell'utente - self._write_output(f"\n\033[1;36m👤 {command}\033[0m\n\n") - - # Mostra "thinking" - self._write_output("\033[1;33m🤔 Pensando...\033[0m\n\n") - - # Chiamata all'agent - try: - response = self.agent_callback(command) - - # Pulisci "thinking" - self._write_output("\033[2A\033[K") # Move up 2 lines and clear - - # Formatta e mostra risposta - lines = response.split('\n') - for line in lines: - self._write_output(f"\033[1;32m🤖\033[0m {line}\n") - self._write_output("\n") - - except Exception as e: - self._write_output(f"\033[1;31m❌ Errore: {e}\033[0m\n\n") - - def _draw_screen(self): - """Ridisegna lo schermo""" - # Clear e ricrea contenuto - output = [] - - # Draw top - output.append(f"\r\033[H\033[J") # Home e clear - output.append(self._get_header()) - output.append("\n") - - # Draw screen content - for i, line in enumerate(self.screen.display): - if i == self.rows - 1: - # Last line is prompt - output.append(f"\033[1;35m{self.prompt}\033[0m ") - else: - # Rimuovi trailing spaces ma mantieni contenuto - output.append(line.rstrip() + " " * (self.columns - len(line.rstrip()))) - - # Draw status bar - output.append("\r\033[{0}B".format(self.rows - len(self.screen.display) - 1)) # Move to bottom - output.append(self._get_status_bar()) - - # Write to terminal - sys.stdout.write(''.join(output)) - sys.stdout.flush() - - def _get_header(self) -> str: - """Header del terminale""" - width = self.columns - border = "═" * (width - 4) - title = " LangCode Agent - Terminal UI con Scrollback " - - return ( - f"\r\033[1;36m╔{border}╗\033[0m\n" - f"\r\033[1;36m║\033[0m\033[1;33m{title:^{width-4}}\033[0m\033[1;36m║\033[0m\n" - f"\r\033[1;36m╚{border}╝\033[0m" - ) - - def _get_status_bar(self) -> str: - """Status bar in fondo""" - return ( - f"\r\033[1;30m{'─' * (self.columns - 30)}" - f"\033[0m \033[1;34mScrollback: {len(self.screen.scroll_history)} righe\033[0m" - ) + if user_input.strip().lower() in ['/exit', '/quit']: + sys.stdout.write("\033[1;33m👋 Arrivederci!\033[0m\n") + break + + if user_input.strip() == '/clear': + self.output_buffer.clear() + self._redraw() + continue + + if user_input.strip() == '/help': + help_text = """ +\033[1;36m╔══════════════════════════════════════════════════════╗\033[0m +\033[1;36m║ COMANDI DISPONIBILI ║\033[0m +\033[1;36m╠══════════════════════════════════════════════════════╣\033[0m +\033[1;36m║ /help - Mostra questo messaggio ║\033[0m +\033[1;36m║ /clear - Pulisci lo schermo ║\033[0m +\033[1;36m║ /exit - Esci ║\033[0m +\033[1;36m╚══════════════════════════════════════════════════════╝\033[0m +""" + self._add_output(help_text, "36") + self._redraw() + continue + + # Mostra comando utente + self._add_output(f"👤 {user_input}", "36") + self._redraw() + + # Mostra "thinking" + self._add_output("🤔 Pensando...", "33") + self._redraw() + + # Chiamata all'agent + try: + response = agent_callback(user_input) + + # Rimuovi "thinking" e mostra risposta + if self.output_buffer and "🤔 Pensando..." in self.output_buffer[-1]: + self.output_buffer.pop() + + self._add_output(response, "32") + except Exception as e: + self._add_output(f"❌ Errore: {str(e)}", "31") + + self._redraw() + + finally: + signal.signal(signal.SIGINT, old_handler) def close(self): """Chiude il terminale""" self.running = False - if self.master_fd: - os.close(self.master_fd) - if self.slave_fd: - os.close(self.slave_fd) def create_terminal_ui( - rows: int = 24, - columns: int = 80, scrollback: int = 10000, prompt: str = "❯ " ) -> TerminalUI: """Factory function per creare TerminalUI""" return TerminalUI( - rows=rows, - columns=columns, scrollback=scrollback, prompt=prompt ) @@ -276,7 +201,7 @@ def create_terminal_ui( # Test standalone if __name__ == "__main__": def test_callback(text): - return f"Ricevuto: {text}\n\nEcco una risposta multi-linea:\n1. Prima linea\n2. Seconda linea\n3. Terza linea" + return f"Ricevuto: {text}\n\nRisposta multi-linea:\n1. Prima\n2. Seconda\n3. Terza" print("Avvio TerminalUI di test...") term = create_terminal_ui()