opencode
c76beb764c
- Nuovo file terminal_ui.py con emulatore terminale - Flag --terminal per attivare UI con scrollback - 10000 righe di scrollback buffer - Supporta resize automatico del terminale - Ctrl+C per uscire
287 lines
8.7 KiB
Python
287 lines
8.7 KiB
Python
"""
|
||
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")
|