diff --git a/langcode.py b/langcode.py index df68cca..09a7352 100755 --- a/langcode.py +++ b/langcode.py @@ -105,8 +105,31 @@ def cmd_chat(args): except Exception as e: return f"Errore: {str(e)}" - # Default: CLI semplice, EnhancedCLI solo con flag --enhanced + # Default: CLI semplice, EnhancedCLI/TerminalUI con flag use_enhanced = getattr(args, 'enhanced', False) + use_terminal = getattr(args, 'terminal', False) + + 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="❯ " + ) + term.start(agent_callback) + return + except ImportError: + print("⚠️ pyte non installato. Esegui: pip install pyte") + print("📝 Uso CLI semplice...") + except Exception as e: + print(f"⚠️ TerminalUI errore: {e}") + print("📝 Uso CLI semplice...") if use_enhanced: import errno @@ -611,6 +634,7 @@ Esempi: # === CHAT (default) === chat_parser = subparsers.add_parser("chat", help="Avvia l'agent in modalità chat") chat_parser.add_argument("--enhanced", "-e", action="store_true", help="Usa EnhancedCLI (sperimentale)") + chat_parser.add_argument("--terminal", "-t", action="store_true", help="Usa TerminalUI con scrollback (pyte)") chat_parser.set_defaults(func=cmd_chat) # === SERVER === diff --git a/terminal_ui.py b/terminal_ui.py new file mode 100644 index 0000000..2ba73e4 --- /dev/null +++ b/terminal_ui.py @@ -0,0 +1,286 @@ +""" +Terminal UI con scrollback usando Pyte +Emula un terminale completo con scrollback buffer +""" +import os +import sys +import pty +import select +import signal +import time +import fcntl +import struct +import termios +from typing import Optional, Callable + +try: + import pyte + PYTE_AVAILABLE = True +except ImportError: + PYTE_AVAILABLE = False + + +class TerminalUI: + """ + Terminal emulator con scrollback usando pyte. + + Features: + - Scrollback buffer completo + - Input/output come un vero terminale + - Segnali gestiti (Ctrl+C, Ctrl+L) + """ + + def __init__( + self, + rows: int = 24, + columns: int = 80, + scrollback: int = 10000, + prompt: str = "❯ " + ): + """ + 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) + 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 + + # 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() + + # Imposta dimensioni iniziali + winsize = struct.pack('HHHH', self.rows, self.columns, 0, 0) + fcntl.ioctl(self.slave_fd, termios.TIOCSWINSZ, winsize) + + def _restore_terminal(self): + """Ripristina terminale originale""" + pass # Il terminale è ripristinato automaticamente alla chiusura del PTY + + def start(self, agent_callback: Callable[[str], str]): + """ + Avvia il terminale interattivo. + + Args: + agent_callback: Funzione che riceve l'input e ritorna la risposta + """ + 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: + 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 "" + + 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) + + if char: + # Feed a pyte + self.stream.feed(char) + input_buffer += char + + # 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 "" + + # 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" + ) + + 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 + ) + + +# 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" + + print("Avvio TerminalUI di test...") + term = create_terminal_ui() + try: + term.start(test_callback) + except KeyboardInterrupt: + print("\nUscito")