feat: pyte UI con scrollback e navigazione PgUp/PgDn/frecce
- Modalità scroll: PgUp entra in scroll mode - q o ENTER esce da scroll mode - j/k frecce navigano nello scrollback - Resize automatico del terminale
This commit is contained in:
1 parent
20233db282
commit
39b9288773
2 files changed
+581
-123
No files matched your search
+293
@@ -0,0 +1,293 @@
|
||||
"""
|
||||
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")
|
||||
+288
-123
@@ -1,6 +1,6 @@
|
||||
"""
|
||||
Terminal UI con scrollback usando Pyte
|
||||
Versione semplificata e robusta
|
||||
Versione con navigazione scrollback
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
@@ -20,9 +20,9 @@ class TerminalUI:
|
||||
Terminal emulator con scrollback usando pyte.
|
||||
|
||||
Features:
|
||||
- Scrollback buffer completo (10000 righe)
|
||||
- Output colorato con ANSI
|
||||
- Resize del terminale
|
||||
- Scrollback buffer completo
|
||||
- Navigazione con frecce e PgUp/PgDn
|
||||
- Output colorato
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
@@ -30,160 +30,331 @@ class TerminalUI:
|
||||
scrollback: int = 10000,
|
||||
prompt: str = "❯ "
|
||||
):
|
||||
"""
|
||||
Inizializza il terminale emulator.
|
||||
|
||||
Args:
|
||||
scrollback: Numero di righe nello scrollback buffer
|
||||
prompt: Prompt da mostrare
|
||||
"""
|
||||
if not PYTE_AVAILABLE:
|
||||
raise ImportError("pyte non è installato. Esegui: pip install pyte")
|
||||
raise ImportError("pyte non è installato")
|
||||
|
||||
self.scrollback = scrollback
|
||||
self.prompt = prompt
|
||||
|
||||
# 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))
|
||||
self.rows, self.cols = shutil.get_terminal_size()
|
||||
self.rows = max(24, min(self.rows, 50))
|
||||
self.cols = max(80, min(self.cols, 120))
|
||||
|
||||
# Crea lo screen pyte
|
||||
self.screen = pyte.Screen(self.term_cols, self.term_rows)
|
||||
# Crea screen pyte
|
||||
self.screen = pyte.Screen(self.cols, self.rows)
|
||||
self.screen.scroll_history = scrollback
|
||||
|
||||
# Stream per feed input
|
||||
self.stream = pyte.Stream(self.screen)
|
||||
|
||||
# Buffer per l'output completo (per scrollback)
|
||||
self.output_buffer: list = []
|
||||
# Scroll offset (0 = latest)
|
||||
self.scroll_offset = 0
|
||||
|
||||
# Flag per uscire
|
||||
# Modalità input: "input" o "scroll"
|
||||
self.mode = "input"
|
||||
|
||||
# Linea di input corrente
|
||||
self.input_buffer = ""
|
||||
self.cursor_pos = 0
|
||||
|
||||
# Flag uscita
|
||||
self.running = False
|
||||
|
||||
# Callback per l'agent
|
||||
# Callback agent
|
||||
self.agent_callback: Optional[Callable[[str], str]] = None
|
||||
|
||||
# Signal handler
|
||||
self._old_sigint: Optional[signal.Handler] = None
|
||||
|
||||
def _redraw(self):
|
||||
def _draw(self):
|
||||
"""Ridisegna lo schermo"""
|
||||
# Clear schermo
|
||||
# Clear
|
||||
sys.stdout.write("\033[2J\033[H")
|
||||
|
||||
# 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)}╝"
|
||||
# Header con colori
|
||||
header_color = "\033[1;36m"
|
||||
reset = "\033[0m"
|
||||
|
||||
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")
|
||||
header_line = "═" * (self.cols - 4)
|
||||
sys.stdout.write(f"{header_color}╔{header_line}╗{reset}\n")
|
||||
title = " LangCode Agent - pyte UI "
|
||||
title_pos = (self.cols - len(title)) // 2
|
||||
sys.stdout.write(f"{header_color}║{reset}" + " " * title_pos + f"\033[1;33m{title}\033[0m" + " " * (self.cols - title_pos - len(title) - 3) + f"{header_color}║{reset}\n")
|
||||
sys.stdout.write(f"{header_color}╚{header_line}╝{reset}\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)
|
||||
# Contenuto scrollback
|
||||
# Lo screen.display contiene le linee visibili
|
||||
# Ma dobbiamo mostrare anche lo scroll_history
|
||||
|
||||
for line in self.output_buffer[start_idx:]:
|
||||
sys.stdout.write(line)
|
||||
if self.mode == "scroll" or self.scroll_offset > 0:
|
||||
# Mostra modalità scroll
|
||||
visible_content = self._get_visible_content()
|
||||
for i, line in enumerate(visible_content):
|
||||
# Tronca o pad la linea
|
||||
display = line[:self.cols - 2] if len(line) > self.cols - 2 else line
|
||||
sys.stdout.write(display + " " * (self.cols - len(display) - 1) + "\n")
|
||||
else:
|
||||
# Modalità input normale - mostra output recente
|
||||
# Lo screen.display ha le ultime self.rows linee
|
||||
for line in self.screen.display:
|
||||
# Rimuovi escape codes per display puro
|
||||
clean = self._strip_ansi(line.rstrip())
|
||||
if clean: # Solo linee non vuote
|
||||
sys.stdout.write(clean + "\n")
|
||||
|
||||
# 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)
|
||||
status_color = "\033[1;34m"
|
||||
scroll_info = f" Scrollback: {len(self.screen.scroll_history)} linee | Modalità: {self.mode} "
|
||||
if self.mode == "scroll":
|
||||
scroll_info += f" | Offset: {self.scroll_offset} | q=exit scroll | ENTER=exit "
|
||||
else:
|
||||
scroll_info += f" | PgUp=scroll | q=quit "
|
||||
sys.stdout.write(f"\n{status_color}{'─' * (self.cols - len(scroll_info) - 1)}{reset}{scroll_info}{reset}\n")
|
||||
|
||||
# Input line o prompt
|
||||
if self.mode == "input":
|
||||
sys.stdout.write(f"\033[1;35m{self.prompt}\033[0m {self.input_buffer}")
|
||||
# Move cursor back if needed
|
||||
if self.cursor_pos < len(self.input_buffer):
|
||||
move_back = len(self.input_buffer) - self.cursor_pos
|
||||
sys.stdout.write(f"\033[{move_back}D")
|
||||
else:
|
||||
sys.stdout.write(f"\033[1;33m[SCROLL MODE] Premi q o ENTER per uscire dallo scroll\033[0m")
|
||||
|
||||
# Prompt
|
||||
sys.stdout.write(f"\n\033[1;35m{self.prompt}\033[0m ")
|
||||
sys.stdout.flush()
|
||||
|
||||
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 _get_visible_content(self) -> list:
|
||||
"""Ottieni contenuto visibile considerando lo scroll"""
|
||||
all_lines = list(self.screen.scroll_history) + list(self.screen.display)
|
||||
|
||||
if not all_lines:
|
||||
return []
|
||||
|
||||
# Calcola offset dal fondo
|
||||
total = len(all_lines)
|
||||
visible = self.rows - 8 # -header, status, prompt, etc.
|
||||
|
||||
start = max(0, total - self.scroll_offset - visible)
|
||||
end = total - self.scroll_offset
|
||||
|
||||
if end <= start:
|
||||
end = start + visible
|
||||
|
||||
return all_lines[start:end]
|
||||
|
||||
def _strip_ansi(self, text: str) -> str:
|
||||
"""Rimuovi codici ANSI da una stringa"""
|
||||
import re
|
||||
ansi_pattern = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
|
||||
return ansi_pattern.sub('', text)
|
||||
|
||||
def _handle_key(self, char: str):
|
||||
"""Gestisce un singolo carattere input"""
|
||||
if char == '\x1b': # ESC
|
||||
self.mode = "scroll"
|
||||
self.scroll_offset = len(self.screen.scroll_history) + len(self.screen.display) - self.rows + 5
|
||||
return
|
||||
|
||||
if self.mode == "scroll":
|
||||
if char in ['q', 'Q']:
|
||||
self.mode = "input"
|
||||
self.scroll_offset = 0
|
||||
elif char == '\r' or char == '\n':
|
||||
self.mode = "input"
|
||||
self.scroll_offset = 0
|
||||
elif char in ['\x7f', '\x08']: # Backspace in scroll mode
|
||||
self.scroll_offset = max(0, self.scroll_offset - 1)
|
||||
elif char == 'j': # vim-style down
|
||||
self.scroll_offset = max(0, self.scroll_offset - 1)
|
||||
elif char == 'k': # vim-style up
|
||||
self.scroll_offset += 1
|
||||
return
|
||||
|
||||
# Modalità input
|
||||
if char == '\r' or char == '\n':
|
||||
# Processa comando
|
||||
cmd = self.input_buffer
|
||||
self.input_buffer = ""
|
||||
self.cursor_pos = 0
|
||||
self.stream.feed('\r\n')
|
||||
return cmd
|
||||
elif char == '\x7f' or char == '\x08': # Backspace
|
||||
if self.cursor_pos > 0:
|
||||
self.input_buffer = self.input_buffer[:self.cursor_pos-1] + self.input_buffer[self.cursor_pos:]
|
||||
self.cursor_pos -= 1
|
||||
self.stream.feed('\b \b')
|
||||
elif char == '\x03': # Ctrl+C
|
||||
return None
|
||||
elif char == '\x0c': # Ctrl+L (clear)
|
||||
self.screen.reset()
|
||||
self.input_buffer = ""
|
||||
self.cursor_pos = 0
|
||||
else:
|
||||
# Carattere normale
|
||||
if len(char) == 1 and ord(char) >= 32 and ord(char) <= 126:
|
||||
self.input_buffer = self.input_buffer[:self.cursor_pos] + char + self.input_buffer[self.cursor_pos:]
|
||||
self.cursor_pos += 1
|
||||
self.stream.feed(char)
|
||||
|
||||
return None
|
||||
|
||||
def _process_command(self, command: str):
|
||||
"""Processa un comando"""
|
||||
if not command.strip():
|
||||
return
|
||||
|
||||
# Output comando utente
|
||||
prompt_color = "\033[1;36m"
|
||||
reset = "\033[0m"
|
||||
self.stream.feed(f'\r\n{prompt_color}👤 {command}{reset}\r\n\n')
|
||||
|
||||
# Thinking indicator
|
||||
thinking_color = "\033[1;33m"
|
||||
self.stream.feed(f'{thinking_color}🤔 Pensando...{reset}\r\n')
|
||||
|
||||
# Chiamata agent
|
||||
try:
|
||||
response = self.agent_callback(command)
|
||||
|
||||
# Rimuovi thinking e mostra risposta
|
||||
# Scroll back di una linea
|
||||
if len(self.screen.scroll_history) > 0:
|
||||
# Il pensiero è nell'ultima linea visibile, rimuoviamolo
|
||||
pass
|
||||
|
||||
# Output risposta
|
||||
for line in response.split('\n'):
|
||||
self.stream.feed(f'\r\n\033[1;32m🤖 {line}\033[0m')
|
||||
|
||||
self.stream.feed('\r\n')
|
||||
|
||||
except Exception as e:
|
||||
self.stream.feed(f'\r\n\033[1;31m❌ Errore: {str(e)}\033[0m\r\n')
|
||||
|
||||
def start(self, agent_callback: Callable[[str], str]):
|
||||
"""
|
||||
Avvia il terminale interattivo.
|
||||
|
||||
Args:
|
||||
agent_callback: Funzione che riceve l'input e ritorna la risposta
|
||||
"""
|
||||
"""Avvia il terminale interattivo"""
|
||||
self.agent_callback = agent_callback
|
||||
self.running = True
|
||||
|
||||
# Registra signal handler per Ctrl+C
|
||||
# Signal handler
|
||||
def handle_sigint(signum, frame):
|
||||
sys.stdout.write("\n\033[1;33m👋 Arrivederci!\033[0m\n")
|
||||
sys.stdout.flush()
|
||||
self.running = False
|
||||
sys.exit(0)
|
||||
|
||||
old_handler = signal.signal(signal.SIGINT, handle_sigint)
|
||||
self._old_sigint = signal.signal(signal.SIGINT, handle_sigint)
|
||||
|
||||
try:
|
||||
# Mostra schermo iniziale
|
||||
self._redraw()
|
||||
# Abilita raw mode per catturare singoli tasti
|
||||
import tty
|
||||
import termios
|
||||
|
||||
while self.running:
|
||||
try:
|
||||
# Leggi input
|
||||
user_input = input()
|
||||
except EOFError:
|
||||
break
|
||||
except KeyboardInterrupt:
|
||||
continue
|
||||
fd = sys.stdin.fileno()
|
||||
old_settings = termios.tcgetattr(fd)
|
||||
|
||||
try:
|
||||
# Set raw mode
|
||||
tty.setraw(fd)
|
||||
|
||||
if not user_input.strip():
|
||||
continue
|
||||
# Mostra schermo iniziale
|
||||
self._draw()
|
||||
|
||||
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)
|
||||
while self.running:
|
||||
# Leggi un carattere
|
||||
ch = os.read(fd, 1)
|
||||
|
||||
# Rimuovi "thinking" e mostra risposta
|
||||
if self.output_buffer and "🤔 Pensando..." in self.output_buffer[-1]:
|
||||
self.output_buffer.pop()
|
||||
if not ch:
|
||||
continue
|
||||
|
||||
self._add_output(response, "32")
|
||||
except Exception as e:
|
||||
self._add_output(f"❌ Errore: {str(e)}", "31")
|
||||
|
||||
self._redraw()
|
||||
char = ch.decode('utf-8', errors='replace')
|
||||
|
||||
# Gestisci PgUp/PgDn in raw mode
|
||||
if char == '\x1b': # ESC sequence start
|
||||
# Leggi il resto della sequenza
|
||||
import select
|
||||
if select.select([fd], [], [], 0.1)[0]:
|
||||
seq = os.read(fd, 2).decode('utf-8', errors='replace')
|
||||
if seq == '[5~': # PgUp
|
||||
self.mode = "scroll"
|
||||
self.scroll_offset = max(0, len(self.screen.scroll_history) + len(self.screen.display) - self.rows + 5)
|
||||
self._draw()
|
||||
continue
|
||||
elif seq == '[6~': # PgDn
|
||||
if self.mode == "scroll":
|
||||
self.scroll_offset = max(0, self.scroll_offset - (self.rows - 8))
|
||||
if self.scroll_offset <= 0:
|
||||
self.mode = "input"
|
||||
self._draw()
|
||||
continue
|
||||
elif seq == '[A': # Up arrow
|
||||
if self.mode == "scroll":
|
||||
self.scroll_offset += 1
|
||||
self._draw()
|
||||
continue
|
||||
elif seq == '[B': # Down arrow
|
||||
if self.mode == "scroll":
|
||||
self.scroll_offset = max(0, self.scroll_offset - 1)
|
||||
self._draw()
|
||||
continue
|
||||
|
||||
# Processa il carattere
|
||||
result = self._handle_key(char)
|
||||
|
||||
if result is None:
|
||||
self._draw()
|
||||
continue
|
||||
|
||||
if result == "":
|
||||
self._draw()
|
||||
continue
|
||||
|
||||
if result is None and char == '\x03': # Ctrl+C
|
||||
break
|
||||
|
||||
# Esegui comando
|
||||
if result:
|
||||
if result.lower() == '/exit':
|
||||
break
|
||||
elif result.lower() == '/help':
|
||||
self.stream.feed('\r\n')
|
||||
help_lines = [
|
||||
"\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 schermo ║\033[0m",
|
||||
"\033[1;36m║ /exit - Esci ║\033[0m",
|
||||
"\033[1;36m║ PgUp - Entra in modalità scroll ║\033[0m",
|
||||
"\033[1;36m║ q - Exit modalità scroll ║\033[0m",
|
||||
"\033[1;36m╚══════════════════════════════════════════════╝\033[0m",
|
||||
]
|
||||
for line in help_lines:
|
||||
self.stream.feed(line + '\r\n')
|
||||
elif result.lower() == '/clear':
|
||||
self.screen.reset()
|
||||
self.input_buffer = ""
|
||||
self.cursor_pos = 0
|
||||
else:
|
||||
self._process_command(result)
|
||||
|
||||
self._draw()
|
||||
|
||||
finally:
|
||||
# Ripristina terminale
|
||||
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
||||
|
||||
except Exception as e:
|
||||
print(f"Errore: {e}")
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
finally:
|
||||
signal.signal(signal.SIGINT, old_handler)
|
||||
sys.stdout.write("\n")
|
||||
sys.stdout.flush()
|
||||
|
||||
def close(self):
|
||||
"""Chiude il terminale"""
|
||||
"""Chiudi il terminale"""
|
||||
self.running = False
|
||||
|
||||
|
||||
@@ -191,21 +362,15 @@ def create_terminal_ui(
|
||||
scrollback: int = 10000,
|
||||
prompt: str = "❯ "
|
||||
) -> TerminalUI:
|
||||
"""Factory function per creare TerminalUI"""
|
||||
return TerminalUI(
|
||||
scrollback=scrollback,
|
||||
prompt=prompt
|
||||
)
|
||||
"""Factory function"""
|
||||
return TerminalUI(scrollback=scrollback, prompt=prompt)
|
||||
|
||||
|
||||
# Test standalone
|
||||
# Test
|
||||
if __name__ == "__main__":
|
||||
def test_callback(text):
|
||||
return f"Ricevuto: {text}\n\nRisposta multi-linea:\n1. Prima\n2. Seconda\n3. Terza"
|
||||
return f"Ricevuto: {text}\n\nLunga risposta:\n" + "\n".join([f"Linea {i}" for i in range(1, 21)])
|
||||
|
||||
print("Avvio TerminalUI di test...")
|
||||
term = create_terminal_ui()
|
||||
try:
|
||||
term.start(test_callback)
|
||||
except KeyboardInterrupt:
|
||||
print("\nUscito")
|
||||
print("Avvio TerminalUI...")
|
||||
ui = create_terminal_ui()
|
||||
ui.start(test_callback)
|
||||
Reference in new issue
Block a user