opencode
39b9288773
- Modalità scroll: PgUp entra in scroll mode - q o ENTER esce da scroll mode - j/k frecce navigano nello scrollback - Resize automatico del terminale
294 lines
11 KiB
Python
294 lines
11 KiB
Python
"""
|
||
Terminal UI con scrollback usando Curses
|
||
Scrollback vero con frecce e PgUp/PgDn
|
||
"""
|
||
import sys
|
||
import signal
|
||
import curses
|
||
from typing import Optional, Callable
|
||
|
||
try:
|
||
import curses
|
||
CURSES_AVAILABLE = True
|
||
except ImportError:
|
||
CURSES_AVAILABLE = False
|
||
|
||
|
||
class CursesUI:
|
||
"""
|
||
Terminal UI con scrollback usando curses.
|
||
|
||
Features:
|
||
- Scrollback completo con PgUp/PgDn
|
||
- Frecce su/giù per navigare
|
||
- Resize automatico
|
||
- Colors (se supportati)
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
scrollback: int = 10000,
|
||
prompt: str = "❯ "
|
||
):
|
||
"""
|
||
Inizializza il terminale UI con curses.
|
||
|
||
Args:
|
||
scrollback: Numero di righe nello scrollback buffer
|
||
prompt: Prompt da mostrare
|
||
"""
|
||
if not CURSES_AVAILABLE:
|
||
raise ImportError("curses non è disponibile su questo sistema")
|
||
|
||
self.scrollback = scrollback
|
||
self.prompt = prompt
|
||
|
||
# Buffer per scrollback
|
||
self.lines: list = []
|
||
|
||
# Callback per l'agent
|
||
self.agent_callback: Optional[Callable[[str], str]] = None
|
||
|
||
def _init_colors(self, stdscr):
|
||
"""Inizializza i colori"""
|
||
if curses.has_colors():
|
||
curses.start_color()
|
||
curses.use_default_colors()
|
||
# Colori: 1=cyan, 2=green, 3=yellow, 4=red, 5=magenta, 6=blue
|
||
curses.init_pair(1, curses.COLOR_CYAN, -1) # Header
|
||
curses.init_pair(2, curses.COLOR_GREEN, -1) # Agent output
|
||
curses.init_pair(3, curses.COLOR_YELLOW, -1) # Thinking
|
||
curses.init_pair(4, curses.COLOR_RED, -1) # Error
|
||
curses.init_pair(5, curses.COLOR_MAGENTA, -1) # Prompt
|
||
curses.init_pair(6, curses.COLOR_BLUE, -1) # User input
|
||
|
||
def _add_line(self, text: str, color: int = 2):
|
||
"""Aggiunge una linea al buffer"""
|
||
self.lines.append((text, color))
|
||
# Limita la dimensione del buffer
|
||
if len(self.lines) > self.scrollback:
|
||
self.lines.pop(0)
|
||
|
||
def _draw(self, stdscr):
|
||
"""Disegna lo schermo"""
|
||
stdscr.clear()
|
||
|
||
# Ottieni dimensioni
|
||
max_y, max_x = stdscr.getmaxyx()
|
||
|
||
# Header
|
||
header = "═" * (max_x - 4)
|
||
stdscr.addstr(0, 0, f"╔{header}╗", curses.color_pair(1) | curses.A_BOLD)
|
||
title = " LangCode Agent - Curses UI "
|
||
stdscr.addstr(1, (max_x - len(title)) // 2, title, curses.color_pair(1) | curses.A_BOLD)
|
||
stdscr.addstr(2, 0, f"╚{header}╝", curses.color_pair(1) | curses.A_BOLD)
|
||
|
||
# Calcola area visibile
|
||
start_row = 4
|
||
visible_rows = max_y - start_row - 2 # -2 per status e prompt
|
||
|
||
# Mostra linee visibili
|
||
total_lines = len(self.lines)
|
||
|
||
# Scroll offset (0 = showing latest)
|
||
# La scrolling position va da 0 a max(0, total_lines - visible_rows)
|
||
self.scroll_offset = max(0, total_lines - visible_rows)
|
||
|
||
# Disegna le linee visibili
|
||
for i in range(min(visible_rows, total_lines)):
|
||
line_idx = self.scroll_offset + i
|
||
if line_idx < total_lines:
|
||
text, color = self.lines[line_idx]
|
||
# Tronca se troppo lunga
|
||
display_text = text[:max_x - 2] if len(text) > max_x - 2 else text
|
||
try:
|
||
stdscr.addstr(start_row + i, 0, display_text, curses.color_pair(color))
|
||
except curses.error:
|
||
pass
|
||
|
||
# Status bar
|
||
status_y = max_y - 2
|
||
scroll_info = f"Linee: {total_lines} | Scroll: {self.scroll_offset} | PgUp/PgDn=Frecce=Scroll | q=quit | Ctrl+C=exit"
|
||
stdscr.addstr(status_y, 0, "─" * (max_x - len(scroll_info) - 1), curses.A_DIM)
|
||
stdscr.addstr(status_y, max_x - len(scroll_info) - 1, scroll_info, curses.A_DIM)
|
||
|
||
# Prompt
|
||
stdscr.addstr(max_y - 1, 0, f"\033[1;35m{self.prompt}\033[0m ", curses.color_pair(5) | curses.A_BOLD)
|
||
|
||
stdscr.refresh()
|
||
|
||
def _handle_input(self, stdscr):
|
||
"""Gestisce l'input"""
|
||
curses.curs_set(1) # Mostra cursore
|
||
|
||
# Abilita arrow keys e PgUp/PgDn
|
||
stdscr.keypad(True)
|
||
|
||
# Configura timeout per read non-blocking
|
||
stdscr.timeout(100) # 100ms timeout
|
||
|
||
input_buffer = ""
|
||
while True:
|
||
self._draw(stdscr)
|
||
|
||
try:
|
||
key = stdscr.getch()
|
||
except curses.error:
|
||
continue
|
||
|
||
if key == -1:
|
||
# Timeout - continua
|
||
continue
|
||
|
||
elif key in [curses.KEY_ENTER, 10, 13]:
|
||
# Enter - processa comando
|
||
if input_buffer.strip():
|
||
return input_buffer.strip()
|
||
input_buffer = ""
|
||
|
||
elif key in [curses.KEY_UP]:
|
||
# Scroll up nel buffer
|
||
if self.scroll_offset > 0:
|
||
self.scroll_offset -= 1
|
||
|
||
elif key in [curses.KEY_DOWN]:
|
||
# Scroll down nel buffer
|
||
max_y, max_x = stdscr.getmaxyx()
|
||
visible_rows = max_y - 6
|
||
if self.scroll_offset < max(0, len(self.lines) - visible_rows):
|
||
self.scroll_offset += 1
|
||
|
||
elif key in [curses.KEY_PPAGE]:
|
||
# PgUp - scroll su di una pagina
|
||
max_y, _ = stdscr.getmaxyx()
|
||
self.scroll_offset = max(0, self.scroll_offset - (max_y - 6))
|
||
|
||
elif key in [curses.KEY_NPAGE]:
|
||
# PgDn - scroll giu di una pagina
|
||
max_y, max_x = stdscr.getmaxyx()
|
||
visible_rows = max_y - 6
|
||
self.scroll_offset = min(max(0, len(self.lines) - visible_rows),
|
||
self.scroll_offset + visible_rows)
|
||
|
||
elif key == 27: # ESC
|
||
return "/exit"
|
||
|
||
elif key in [3, 26]: # Ctrl+C
|
||
return "/exit"
|
||
|
||
elif key in [curses.KEY_BACKSPACE, 127, 8]:
|
||
# Backspace
|
||
input_buffer = input_buffer[:-1]
|
||
|
||
elif 32 <= key <= 126:
|
||
# Carattere stampabile
|
||
input_buffer += chr(key)
|
||
|
||
# Mostra input corrente da qualche parte se necessario
|
||
|
||
def start(self, agent_callback: Callable[[str], str]):
|
||
"""
|
||
Avvia il terminale interattivo con curses.
|
||
"""
|
||
self.agent_callback = agent_callback
|
||
|
||
def handle_sigint(signum, frame):
|
||
sys.exit(0)
|
||
|
||
old_handler = signal.signal(signal.SIGINT, handle_sigint)
|
||
|
||
try:
|
||
curses.wrapper(self._main_loop)
|
||
finally:
|
||
signal.signal(signal.SIGINT, old_handler)
|
||
|
||
def _main_loop(self, stdscr):
|
||
"""Loop principale curses"""
|
||
self._init_colors(stdscr)
|
||
curses.cbreak()
|
||
curses.noecho()
|
||
|
||
# Aggiungi header
|
||
self._add_line("╔═══════════════════════════════════════════════════════╗", 1)
|
||
self._add_line("║ LangCode Agent - Curses UI con Scrollback ║", 1)
|
||
self._add_line("╚═══════════════════════════════════════════════════════╝", 1)
|
||
self._add_line("", 0)
|
||
self._add_line("Scrivi un messaggio e premi Enter. Usa PgUp/PgDn per scrollare.", 6)
|
||
self._add_line("Comandi: /help, /clear, /exit", 6)
|
||
self._add_line("", 0)
|
||
|
||
while True:
|
||
# Ottieni input
|
||
stdscr.timeout(-1) # Bloccante per l'input
|
||
command = self._handle_input(stdscr)
|
||
|
||
if command.lower() in ['/exit', '/quit']:
|
||
break
|
||
|
||
if not command:
|
||
continue
|
||
|
||
if command == '/help':
|
||
self._add_line("╔═══════════════════════════════════════════════════════╗", 1)
|
||
self._add_line("║ COMANDI ║", 1)
|
||
self._add_line("╠═══════════════════════════════════════════════════════╣", 1)
|
||
self._add_line("║ /help - Mostra questo messaggio ║", 1)
|
||
self._add_line("║ /clear - Pulisci lo schermo ║", 1)
|
||
self._add_line("║ /exit - Esci dall'agente ║", 1)
|
||
self._add_line("║ PgUp - Scroll su ║", 1)
|
||
self._add_line("║ PgDn - Scroll giu ║", 1)
|
||
self._add_line("║ Frecce - Naviga nel scrollback ║", 1)
|
||
self._add_line("╚═══════════════════════════════════════════════════════╝", 1)
|
||
continue
|
||
|
||
if command == '/clear':
|
||
self.lines.clear()
|
||
continue
|
||
|
||
# Mostra comando utente
|
||
self._add_line("", 0)
|
||
self._add_line(f"👤 {command}", 6)
|
||
|
||
# Chiamata all'agent
|
||
self._add_line("", 0)
|
||
self._add_line("🤔 Pensando...", 3)
|
||
|
||
try:
|
||
response = agent_callback(command)
|
||
|
||
# Rimuovi "thinking" e mostra risposta
|
||
if self.lines and "🤔 Pensando..." in self.lines[-1][0]:
|
||
self.lines.pop()
|
||
|
||
# Aggiungi risposta
|
||
for line in response.split('\n'):
|
||
self._add_line(f"🤖 {line}", 2)
|
||
except Exception as e:
|
||
self._add_line(f"❌ Errore: {str(e)}", 4)
|
||
|
||
self._add_line("", 0)
|
||
|
||
|
||
def create_curses_ui(
|
||
scrollback: int = 10000,
|
||
prompt: str = "❯ "
|
||
) -> CursesUI:
|
||
"""Factory function per creare CursesUI"""
|
||
return CursesUI(
|
||
scrollback=scrollback,
|
||
prompt=prompt
|
||
)
|
||
|
||
|
||
# Test standalone
|
||
if __name__ == "__main__":
|
||
def test_callback(text):
|
||
return f"Ricevuto: {text}\n\nRisposta multi-linea:\n1. Prima linea\n2. Seconda\n3. Terza\n4. Quarta\n5. Quinta"
|
||
|
||
print("Avvio CursesUI di test...")
|
||
ui = create_curses_ui()
|
||
try:
|
||
ui.start(test_callback)
|
||
except KeyboardInterrupt:
|
||
print("\nUscito")
|