Handle undersized terminals and wire autocomplete/history options
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <[email protected]>
This commit is contained in:
@@ -10,7 +10,7 @@ import threading
|
||||
import queue
|
||||
import logging
|
||||
import time
|
||||
from .constants import COLOR_PAIRS, INFO_WINDOW_HEIGHT, INFO_WINDOW_Y, OUTPUT_WINDOW_Y, INPUT_WINDOW_HEIGHT, TEAM_MODES, MAX_COMMAND_HISTORY
|
||||
from .constants import INFO_WINDOW_HEIGHT, INFO_WINDOW_Y, OUTPUT_WINDOW_Y, INPUT_WINDOW_HEIGHT, TEAM_MODES, MAX_COMMAND_HISTORY
|
||||
from .cvars import autocomplete, COMMAND_SIGNATURES, get_signature_with_highlight, get_argument_suggestions, COMMAND_ARGUMENTS
|
||||
|
||||
logger = logging.getLogger('ui')
|
||||
@@ -19,12 +19,15 @@ logger = logging.getLogger('ui')
|
||||
class CursesHandler(logging.Handler):
|
||||
"""Logging handler that outputs to curses window"""
|
||||
|
||||
def __init__(self, window):
|
||||
def __init__(self, window, manager=None):
|
||||
logging.Handler.__init__(self)
|
||||
self.window = window
|
||||
self.manager = manager
|
||||
|
||||
def emit(self, record):
|
||||
try:
|
||||
if self.manager is not None and self.manager._too_small:
|
||||
return
|
||||
msg = self.format(record)
|
||||
fs = "%s\n"
|
||||
try:
|
||||
@@ -86,7 +89,7 @@ def print_colored(window, message, attributes=0):
|
||||
except curses.error:
|
||||
return
|
||||
|
||||
def update_autocomplete_display(window, current_input, first_word, words, ends_with_space):
|
||||
def update_autocomplete_display(window, first_word, words, ends_with_space, player_names_provider=None):
|
||||
"""
|
||||
Update autocomplete display based on current input state.
|
||||
Returns (suggestions, suggestion_index, original_word) tuple for Tab cycling.
|
||||
@@ -129,11 +132,12 @@ def update_autocomplete_display(window, current_input, first_word, words, ends_w
|
||||
current_value = words[-1]
|
||||
|
||||
# Get argument suggestions
|
||||
player_names = player_names_provider() if player_names_provider else None
|
||||
arg_suggestions = get_argument_suggestions(
|
||||
first_word,
|
||||
arg_position,
|
||||
current_value,
|
||||
player_list=None # TODO: pass player list from game_state
|
||||
player_list=player_names
|
||||
)
|
||||
|
||||
if arg_suggestions:
|
||||
@@ -199,7 +203,7 @@ def update_autocomplete_display(window, current_input, first_word, words, ends_w
|
||||
class UIManager:
|
||||
"""Manages curses windows and display"""
|
||||
|
||||
def __init__(self, screen, host):
|
||||
def __init__(self, screen, host, max_history=MAX_COMMAND_HISTORY):
|
||||
self.screen = screen
|
||||
self.host = host
|
||||
self.info_window = None
|
||||
@@ -210,14 +214,15 @@ class UIManager:
|
||||
self.command_history = []
|
||||
self.history_index = -1
|
||||
self.cursor_pos = 0 # Track cursor position in input
|
||||
self.max_history = max_history
|
||||
self._too_small = False
|
||||
|
||||
self._init_curses()
|
||||
self._create_windows()
|
||||
if not self._create_windows():
|
||||
raise SystemExit('Terminal too small: minimum 20 rows x 80 columns required')
|
||||
|
||||
def _init_curses(self):
|
||||
"""Initialize curses settings"""
|
||||
curses.endwin()
|
||||
curses.initscr()
|
||||
self.screen.nodelay(1)
|
||||
curses.start_color()
|
||||
curses.use_default_colors()
|
||||
@@ -308,8 +313,20 @@ class UIManager:
|
||||
|
||||
# Minimum size check
|
||||
if maxy < 20 or maxx < 80:
|
||||
if not self._too_small:
|
||||
self._too_small = True
|
||||
self._delete_windows()
|
||||
self._draw_too_small_warning()
|
||||
return False
|
||||
|
||||
# Rebuild windows when returning from a too-small terminal
|
||||
if self._too_small:
|
||||
self._too_small = False
|
||||
curses.update_lines_cols()
|
||||
self.screen.clear()
|
||||
self.screen.addstr(0, 0, f"Quake Live PyCon: {self.host}")
|
||||
return self._create_windows()
|
||||
|
||||
# Update screen
|
||||
curses.update_lines_cols()
|
||||
self.screen.clear()
|
||||
@@ -346,8 +363,38 @@ class UIManager:
|
||||
except curses.error:
|
||||
return False
|
||||
|
||||
def setup_input_queue(self):
|
||||
def _delete_windows(self):
|
||||
"""Delete all windows (terminal too small for the UI)"""
|
||||
for window in (self.info_window, self.output_window, self.divider_window, self.input_window):
|
||||
if window:
|
||||
try:
|
||||
window.erase()
|
||||
window.noutrefresh()
|
||||
except curses.error:
|
||||
pass
|
||||
curses.doupdate()
|
||||
self.info_window = None
|
||||
self.output_window = None
|
||||
self.divider_window = None
|
||||
self.input_window = None
|
||||
|
||||
def _draw_too_small_warning(self):
|
||||
"""Draw centered warning on the bare screen"""
|
||||
try:
|
||||
maxy, maxx = self.screen.getmaxyx()
|
||||
warning = 'Terminal too small - resize to at least 20x80'
|
||||
self.screen.erase()
|
||||
self.screen.addstr(maxy // 2, max(0, (maxx - len(warning)) // 2), warning)
|
||||
self.screen.noutrefresh()
|
||||
curses.doupdate()
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
def setup_input_queue(self, player_names_provider=None, max_history=None):
|
||||
"""Setup threaded input queue with command history and autocomplete"""
|
||||
if max_history is None:
|
||||
max_history = self.max_history
|
||||
|
||||
def wait_stdin(q, window, manager):
|
||||
current_input = ""
|
||||
cursor_pos = 0
|
||||
@@ -362,8 +409,13 @@ class UIManager:
|
||||
original_word = "" # Store original word before cycling
|
||||
|
||||
while True:
|
||||
try:
|
||||
try:
|
||||
key = window.getch()
|
||||
except curses.error:
|
||||
# Window may be deleted during a too-small resize
|
||||
time.sleep(0.1)
|
||||
continue
|
||||
|
||||
if key == -1: # No input
|
||||
continue
|
||||
@@ -371,6 +423,8 @@ class UIManager:
|
||||
# Handle terminal resize
|
||||
if key == curses.KEY_RESIZE:
|
||||
manager.handle_resize()
|
||||
window = manager.input_window or window
|
||||
if not manager._too_small:
|
||||
# Redraw input
|
||||
window.erase()
|
||||
window.addstr(0, 0, current_input)
|
||||
@@ -455,7 +509,7 @@ class UIManager:
|
||||
if len(current_input) > 0:
|
||||
# Add to history
|
||||
manager.command_history.append(current_input)
|
||||
if len(manager.command_history) > MAX_COMMAND_HISTORY:
|
||||
if len(manager.command_history) > max_history:
|
||||
manager.command_history.pop(0)
|
||||
|
||||
q.put(current_input)
|
||||
@@ -542,10 +596,10 @@ class UIManager:
|
||||
words = current_input.split()
|
||||
ends_with_space = current_input.endswith(' ')
|
||||
|
||||
if words:
|
||||
if words and not manager._too_small:
|
||||
first_word = words[0].lower()
|
||||
suggestions, suggestion_index, original_word = update_autocomplete_display(
|
||||
window, current_input, first_word, words, ends_with_space
|
||||
window, first_word, words, ends_with_space, player_names_provider
|
||||
)
|
||||
|
||||
window.move(0, cursor_pos)
|
||||
@@ -567,10 +621,10 @@ class UIManager:
|
||||
words = current_input.split()
|
||||
ends_with_space = current_input.endswith(' ')
|
||||
|
||||
if words:
|
||||
if words and not manager._too_small:
|
||||
first_word = words[0].lower()
|
||||
suggestions, suggestion_index, original_word = update_autocomplete_display(
|
||||
window, current_input, first_word, words, ends_with_space
|
||||
window, first_word, words, ends_with_space, player_names_provider
|
||||
)
|
||||
|
||||
window.move(0, cursor_pos)
|
||||
@@ -593,13 +647,15 @@ class UIManager:
|
||||
|
||||
def setup_logging(self):
|
||||
"""Setup logging handler for output window"""
|
||||
handler = CursesHandler(self.output_window)
|
||||
handler = CursesHandler(self.output_window, self)
|
||||
formatter = logging.Formatter('%(asctime)-8s|%(name)-12s|%(levelname)-6s|%(message)-s', '%H:%M:%S')
|
||||
handler.setFormatter(formatter)
|
||||
return handler
|
||||
|
||||
def print_message(self, message, attributes=0):
|
||||
"""Print formatted message to output window"""
|
||||
if self._too_small:
|
||||
return
|
||||
print_colored(self.output_window, message, attributes)
|
||||
self.output_window.noutrefresh()
|
||||
# Restore cursor to input window at current position
|
||||
@@ -609,6 +665,8 @@ class UIManager:
|
||||
|
||||
def update_server_info(self, game_state):
|
||||
"""Update server info window"""
|
||||
if self._too_small:
|
||||
return
|
||||
self.info_window.erase()
|
||||
|
||||
max_y, max_x = self.info_window.getmaxyx()
|
||||
@@ -680,7 +738,10 @@ class UIManager:
|
||||
blue_total = 0
|
||||
for player_name, player_data in server_info.players.items():
|
||||
team = game_state.player_tracker.get_team(player_name)
|
||||
try:
|
||||
score = int(player_data.get('score', 0))
|
||||
except ValueError:
|
||||
score = 0
|
||||
|
||||
if team == 'RED':
|
||||
red_total += score
|
||||
@@ -698,11 +759,17 @@ class UIManager:
|
||||
spec_players = []
|
||||
|
||||
for player_name in teams['RED']:
|
||||
try:
|
||||
score = int(server_info.players.get(player_name, {}).get('score', 0))
|
||||
except ValueError:
|
||||
score = 0
|
||||
red_players_with_scores.append((player_name, score))
|
||||
|
||||
for player_name in teams['BLUE']:
|
||||
try:
|
||||
score = int(server_info.players.get(player_name, {}).get('score', 0))
|
||||
except ValueError:
|
||||
score = 0
|
||||
blue_players_with_scores.append((player_name, score))
|
||||
|
||||
# Sort by score descending
|
||||
@@ -743,7 +810,10 @@ class UIManager:
|
||||
free_players = teams['FREE']
|
||||
free_players_with_scores = []
|
||||
for player_name in free_players:
|
||||
try:
|
||||
score = int(server_info.players.get(player_name, {}).get('score', 0))
|
||||
except ValueError:
|
||||
score = 0
|
||||
free_players_with_scores.append((player_name, score))
|
||||
|
||||
# Sort by score descending
|
||||
@@ -797,7 +867,7 @@ class UIManager:
|
||||
pass
|
||||
|
||||
# Separator
|
||||
separator = "^7" + "═" * (max_x - 1) + "^7"
|
||||
separator = "^7" + "=" * (max_x - 1) + "^7"
|
||||
print_colored(self.info_window, separator, 0)
|
||||
|
||||
self.info_window.noutrefresh()
|
||||
|
||||
Reference in New Issue
Block a user