fix: wrap linee e padding per display corretto
- Testo wrap a colonne del terminale - Padding con spazi per linee corte - Rimosso riferimento a pyte non usato
This commit is contained in:
opencode committed
1 parent
e69a289361
commit
aec5736394
1 file changed
+103
-111
+103
-111
@@ -1,6 +1,6 @@
|
||||
"""
|
||||
Terminal UI con scrollback usando Pyte
|
||||
Versione con buffer manuale per scrollback
|
||||
Terminal UI con scrollback semplice
|
||||
Niente dipendenze esterne, solo standard library
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
@@ -8,19 +8,13 @@ import signal
|
||||
import shutil
|
||||
from typing import Optional, Callable
|
||||
|
||||
try:
|
||||
import pyte
|
||||
PYTE_AVAILABLE = True
|
||||
except ImportError:
|
||||
PYTE_AVAILABLE = False
|
||||
|
||||
|
||||
class TerminalUI:
|
||||
"""
|
||||
Terminal emulator con scrollback usando pyte.
|
||||
Terminal UI con scrollback semplice.
|
||||
|
||||
Features:
|
||||
- Scrollback buffer manuale
|
||||
- Scrollback buffer
|
||||
- Navigazione con frecce e PgUp/PgDn
|
||||
- Output colorato
|
||||
"""
|
||||
@@ -30,9 +24,6 @@ class TerminalUI:
|
||||
scrollback: int = 10000,
|
||||
prompt: str = "❯ "
|
||||
):
|
||||
if not PYTE_AVAILABLE:
|
||||
raise ImportError("pyte non è installato")
|
||||
|
||||
self.scrollback = scrollback
|
||||
self.prompt = prompt
|
||||
|
||||
@@ -41,20 +32,16 @@ class TerminalUI:
|
||||
self.rows = max(24, min(self.rows, 50))
|
||||
self.cols = max(80, min(self.cols, 120))
|
||||
|
||||
# Crea screen pyte
|
||||
self.screen = pyte.Screen(self.cols, self.rows)
|
||||
self.stream = pyte.Stream(self.screen)
|
||||
|
||||
# Il nostro buffer per lo scrollback
|
||||
# Buffer scrollback
|
||||
self.history: list = []
|
||||
|
||||
# Scroll offset (0 = latest)
|
||||
self.scroll_offset = 0
|
||||
|
||||
# Modalità input: "input" o "scroll"
|
||||
# Modalità: "input" o "scroll"
|
||||
self.mode = "input"
|
||||
|
||||
# Linea di input corrente
|
||||
# Input buffer
|
||||
self.input_buffer = ""
|
||||
self.cursor_pos = 0
|
||||
|
||||
@@ -64,58 +51,88 @@ class TerminalUI:
|
||||
# Callback agent
|
||||
self.agent_callback: Optional[Callable[[str], str]] = None
|
||||
|
||||
def _add_to_history(self, line: str):
|
||||
"""Aggiunge una linea allo history buffer"""
|
||||
self.history.append(line)
|
||||
if len(self.history) > self.scrollback:
|
||||
def _add_to_history(self, text: str, color: str = "0"):
|
||||
"""Aggiunge linea allo history (wrap a cols)"""
|
||||
import re
|
||||
# Rimuovi codes ANSI per calcolare lunghezza reale
|
||||
ansi_pattern = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
|
||||
clean = ansi_pattern.sub('', text)
|
||||
|
||||
# Wrap del testo se troppo lungo
|
||||
max_width = self.cols - 2
|
||||
lines = []
|
||||
while len(clean) > max_width:
|
||||
lines.append(clean[:max_width])
|
||||
clean = clean[max_width:]
|
||||
if clean:
|
||||
lines.append(clean)
|
||||
|
||||
# Aggiungi tutte le linee
|
||||
for line in lines:
|
||||
self.history.append((line, color))
|
||||
|
||||
# Limita history
|
||||
while len(self.history) > self.scrollback:
|
||||
self.history.pop(0)
|
||||
|
||||
def _draw(self):
|
||||
"""Ridisegna lo schermo"""
|
||||
reset = "\033[0m"
|
||||
|
||||
# Clear
|
||||
# Clear screen
|
||||
sys.stdout.write("\033[2J\033[H")
|
||||
|
||||
# Header
|
||||
header_color = "\033[1;36m"
|
||||
sys.stdout.write(f"{header_color}╔{'═' * (self.cols - 4)}╗{reset}\n")
|
||||
title = " LangCode Agent - pyte UI "
|
||||
yellow = "\033[1;33m"
|
||||
|
||||
line = "═" * (self.cols - 4)
|
||||
sys.stdout.write(f"{header_color}╔{line}╗{reset}\n")
|
||||
|
||||
title = " LangCode Agent "
|
||||
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}╚{'═' * (self.cols - 4)}║{reset}\n")
|
||||
sys.stdout.write(f"{header_color}║{reset}" + " " * title_pos + f"{yellow}{title}{reset}" + " " * (self.cols - title_pos - len(title) - 3) + f"{header_color}║{reset}\n")
|
||||
|
||||
sys.stdout.write(f"{header_color}╚{line}╝{reset}\n")
|
||||
|
||||
# Contenuto
|
||||
visible = self.rows - 7
|
||||
|
||||
if self.mode == "scroll":
|
||||
# Mostra history buffer
|
||||
visible = self.rows - 8
|
||||
# Mostra scrollback
|
||||
total = len(self.history)
|
||||
|
||||
start = max(0, total - self.scroll_offset - visible)
|
||||
end = min(total, start + visible)
|
||||
|
||||
for i in range(start, end):
|
||||
line = self.history[i][:self.cols - 2]
|
||||
sys.stdout.write(line + " " * (self.cols - len(line) - 1) + "\n")
|
||||
text, color = self.history[i]
|
||||
# Pad con spazi se necessario
|
||||
display = text.ljust(self.cols - 1)
|
||||
sys.stdout.write(f"\033[{color}m{display}{reset}\n")
|
||||
else:
|
||||
# Modalità input - mostra screen.display
|
||||
for line in self.screen.display:
|
||||
clean = self._strip_ansi(line.rstrip())
|
||||
if clean:
|
||||
display = clean[:self.cols - 2]
|
||||
sys.stdout.write(display + " " * (self.cols - len(display) - 1) + "\n")
|
||||
# Modalità input - mostra ultime N linee
|
||||
total = len(self.history)
|
||||
start = max(0, total - visible)
|
||||
|
||||
for i in range(start, total):
|
||||
text, color = self.history[i]
|
||||
display = text.ljust(self.cols - 1)
|
||||
sys.stdout.write(f"\033[{color}m{display}{reset}\n")
|
||||
|
||||
# Status bar
|
||||
status_color = "\033[1;34m"
|
||||
blue = "\033[1;34m"
|
||||
dim = "\033[2m"
|
||||
total_lines = len(self.history)
|
||||
scroll_info = f" History: {total_lines} linee | Modalità: {self.mode} "
|
||||
if self.mode == "scroll":
|
||||
scroll_info += f" | Offset: {self.scroll_offset} | q/ENTER=exit scroll "
|
||||
else:
|
||||
scroll_info += f" | PgUp=scroll | q=quit "
|
||||
mode_str = "SCROLL" if self.mode == "scroll" else "INPUT"
|
||||
scroll_info = f" {mode_str} | Lines: {total_lines} "
|
||||
|
||||
status_bar_width = self.cols - len(scroll_info) - 1
|
||||
sys.stdout.write(f"\n{status_color}{'─' * status_bar_width}{reset}{scroll_info}{reset}\n")
|
||||
if self.mode == "scroll":
|
||||
scroll_info += f"| Offset: {self.scroll_offset} | q/ENTER=exit scroll | arrows/jk=scroll "
|
||||
else:
|
||||
scroll_info += "| PgUp=scroll | q=quit "
|
||||
|
||||
status_width = self.cols - len(scroll_info) - 1
|
||||
sys.stdout.write(f"{blue}{'─' * status_width}{reset}{blue}{scroll_info}{reset}\n")
|
||||
|
||||
# Input line
|
||||
if self.mode == "input":
|
||||
@@ -124,19 +141,14 @@ class TerminalUI:
|
||||
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] q o ENTER per uscire\033[0m")
|
||||
sys.stdout.write(f"\033[1;33m[SCROLL MODE] q o ENTER per uscire dallo scroll{reset}")
|
||||
|
||||
sys.stdout.flush()
|
||||
|
||||
def _strip_ansi(self, text: str) -> str:
|
||||
"""Rimuovi codici ANSI"""
|
||||
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"""
|
||||
if char == '\x1b': # ESC
|
||||
"""Gestisce un carattere input"""
|
||||
# ESC sequence
|
||||
if char == '\x1b':
|
||||
self.mode = "scroll"
|
||||
self.scroll_offset = len(self.history)
|
||||
return
|
||||
@@ -148,35 +160,32 @@ class TerminalUI:
|
||||
elif char == '\r' or char == '\n':
|
||||
self.mode = "input"
|
||||
self.scroll_offset = 0
|
||||
elif char in ['j', 'J']:
|
||||
elif char in ['j', 'J', '\'', '\"']: # down
|
||||
self.scroll_offset = min(len(self.history), self.scroll_offset + 1)
|
||||
elif char in ['k', 'K']:
|
||||
elif char in ['k', 'K']: # up
|
||||
self.scroll_offset = max(0, self.scroll_offset - 1)
|
||||
return
|
||||
|
||||
# Modalità input
|
||||
# Input mode
|
||||
if char == '\r' or char == '\n':
|
||||
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
|
||||
self.screen.reset()
|
||||
self.history.clear()
|
||||
self.input_buffer = ""
|
||||
self.cursor_pos = 0
|
||||
else:
|
||||
if len(char) == 1 and 32 <= 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
|
||||
|
||||
@@ -185,32 +194,29 @@ class TerminalUI:
|
||||
if not command.strip():
|
||||
return
|
||||
|
||||
# Aggiungi al nostro history
|
||||
self._add_to_history(f"\033[1;36m👤 {command}\033[0m")
|
||||
|
||||
# Feed a pyte
|
||||
self.stream.feed(f'\r\n\033[1;36m👤 {command}\033[0m\r\n\n')
|
||||
# Aggiungi comando
|
||||
self._add_to_history(f"👤 {command}", "36")
|
||||
|
||||
# Thinking
|
||||
self._add_to_history(f"\033[1;33m🤔 Pensando...\033[0m")
|
||||
self.stream.feed(f'\033[1;33m🤔 Pensando...\033[0m\r\n')
|
||||
self._add_to_history("🤔 Pensando...", "33")
|
||||
self._draw()
|
||||
|
||||
try:
|
||||
response = self.agent_callback(command)
|
||||
|
||||
# Rimuovi thinking e mostra risposta
|
||||
if self.history and "🤔 Pensando..." in self.history[-1]:
|
||||
# Rimuovi thinking
|
||||
if self.history and "🤔" in self.history[-1][0]:
|
||||
self.history.pop()
|
||||
|
||||
# Aggiungi risposta
|
||||
for line in response.split('\n'):
|
||||
self._add_to_history(f'\033[1;32m🤖 {line}\033[0m')
|
||||
self.stream.feed(f'\r\n\033[1;32m🤖 {line}\033[0m')
|
||||
|
||||
self.stream.feed('\r\n')
|
||||
if line.strip():
|
||||
self._add_to_history(f"🤖 {line}", "32")
|
||||
|
||||
except Exception as e:
|
||||
self._add_to_history(f'\033[1;31m❌ Errore: {str(e)}\033[0m')
|
||||
self.stream.feed(f'\r\n\033[1;31m❌ Errore: {str(e)}\033[0m\r\n')
|
||||
if self.history and "🤔" in self.history[-1][0]:
|
||||
self.history.pop()
|
||||
self._add_to_history(f"❌ Errore: {str(e)}", "31")
|
||||
|
||||
def start(self, agent_callback: Callable[[str], str]):
|
||||
"""Avvia il terminale"""
|
||||
@@ -244,7 +250,7 @@ class TerminalUI:
|
||||
|
||||
char = ch.decode('utf-8', errors='replace')
|
||||
|
||||
# Check for ESC sequence (arrow keys, PgUp, PgDn)
|
||||
# ESC sequence
|
||||
if char == '\x1b':
|
||||
import select
|
||||
if select.select([fd], [], [], 0.1)[0]:
|
||||
@@ -273,44 +279,31 @@ class TerminalUI:
|
||||
continue
|
||||
|
||||
result = self._handle_key(char)
|
||||
self._draw()
|
||||
|
||||
if result is None:
|
||||
self._draw()
|
||||
continue
|
||||
|
||||
if result == "":
|
||||
self._draw()
|
||||
continue
|
||||
|
||||
# Process command
|
||||
if result:
|
||||
if result is not None and result != "":
|
||||
if result.lower() == '/exit':
|
||||
break
|
||||
elif result.lower() == '/help':
|
||||
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 - Modalità scroll ║\033[0m",
|
||||
"\033[1;36m║ q - Exit modalità scroll ║\033[0m",
|
||||
"\033[1;36m╚════════════════════════════════════════╝\033[0m",
|
||||
help_text = [
|
||||
("╔══════════════════════════════════════╗", "36"),
|
||||
("║ COMANDI DISPONIBILI ║", "36"),
|
||||
("╠══════════════════════════════════════╣", "36"),
|
||||
("║ /help - Mostra questo messaggio ║", "36"),
|
||||
("║ /clear - Pulisci schermo ║", "36"),
|
||||
("║ /exit - Esci ║", "36"),
|
||||
("║ PgUp - Modalità scroll ║", "36"),
|
||||
("║ q - Exit modalità scroll ║", "36"),
|
||||
("╚══════════════════════════════════════╝", "36"),
|
||||
]
|
||||
for line in help_lines:
|
||||
self._add_to_history(line)
|
||||
self.stream.feed(line + '\r\n')
|
||||
for text, color in help_text:
|
||||
self._add_to_history(text, color)
|
||||
elif result.lower() == '/clear':
|
||||
self.screen.reset()
|
||||
self.history.clear()
|
||||
self.input_buffer = ""
|
||||
self.cursor_pos = 0
|
||||
else:
|
||||
self._process_command(result)
|
||||
|
||||
self._draw()
|
||||
|
||||
self._draw()
|
||||
|
||||
finally:
|
||||
termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
|
||||
|
||||
@@ -323,7 +316,6 @@ class TerminalUI:
|
||||
sys.stdout.flush()
|
||||
|
||||
def close(self):
|
||||
"""Chiudi"""
|
||||
self.running = False
|
||||
|
||||
|
||||
@@ -338,7 +330,7 @@ def create_terminal_ui(
|
||||
# Test
|
||||
if __name__ == "__main__":
|
||||
def test_callback(text):
|
||||
lines = ["Risposta multi-linea:"]
|
||||
lines = [f"Ricevuto: {text}"]
|
||||
lines += [f"Linea {i}" for i in range(1, 21)]
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
Reference in new issue
Block a user