fix: riscrivi TerminalUI piu semplice e robusto
- Usa output_buffer per scrollback manuale - Non usa pty/select complessi - Piu stabile su terminali diversi
This commit is contained in:
opencode committed
1 parent
c76beb764c
commit
20233db282
2 files changed
+115
-196
No files matched your search
+115
-190
@@ -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()
|
||||
|
||||
Reference in new issue
Block a user