pfl
3d1dc19d9b
Send cvarlist and cmdlist on connect and merge the answer into the autocomplete candidates. Number the server menu from 0, keep suggesting sibling commands while the command name is typed, mask the password in the startup line, ignore venv/, fix README claims.
883 lines
32 KiB
Python
883 lines
32 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Curses-based UI for QLPyCon
|
|
Handles terminal display, windows, and color rendering
|
|
|
|
Design rules that keep the UI stable:
|
|
- Every curses call happens on the main thread (curses is not thread-safe).
|
|
- Nothing lives only inside a curses window. Output lines, the input line
|
|
and the last game state are kept in Python objects, so a resize simply
|
|
recreates the windows and redraws them from that state.
|
|
"""
|
|
|
|
import curses
|
|
import itertools
|
|
import logging
|
|
import re
|
|
import time
|
|
from collections import deque
|
|
|
|
from .constants import (INFO_WINDOW_HEIGHT, INFO_WINDOW_Y, OUTPUT_WINDOW_Y, INPUT_WINDOW_HEIGHT,
|
|
MIN_ROWS, MIN_COLS, OUTPUT_SCROLLBACK_LINES, TEAM_MODES, MAX_COMMAND_HISTORY)
|
|
from .cvars import autocomplete, COMMAND_SIGNATURES, get_signature_with_highlight, get_argument_suggestions, COMMAND_ARGUMENTS
|
|
from .formatter import strip_color_codes
|
|
|
|
logger = logging.getLogger('ui')
|
|
|
|
# Splits a message into tokens: a color code (^0-^9) or a single character.
|
|
# Example: '^1ab' -> ['^1', 'a', 'b']
|
|
COLOR_TOKEN_PATTERN = re.compile(r'\^[0-9]|.', re.DOTALL)
|
|
|
|
# Number of suggestions shown in the autocomplete hint line
|
|
MAX_HINT_SUGGESTIONS = 10
|
|
|
|
# Output rows scrolled per mouse wheel notch
|
|
MOUSE_WHEEL_ROWS = 3
|
|
|
|
|
|
class CursesHandler(logging.Handler):
|
|
"""Logging handler that writes log records into the output window"""
|
|
|
|
def __init__(self, manager):
|
|
logging.Handler.__init__(self)
|
|
self.manager = manager
|
|
|
|
def emit(self, record):
|
|
try:
|
|
self.manager.print_message(self.format(record) + "\n")
|
|
except (KeyboardInterrupt, SystemExit):
|
|
raise
|
|
except Exception:
|
|
self.handleError(record)
|
|
|
|
|
|
class ColorState:
|
|
"""
|
|
The Quake color codes (^N) currently in effect while printing a message.
|
|
^1-^6 = color, ^7 = default color, ^8 = bold, ^9 = underline, ^0 = bold and underline off
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.color = 0
|
|
self.bold = False
|
|
self.underline = False
|
|
|
|
def apply(self, digit):
|
|
"""Apply one color code digit ('0'-'9')"""
|
|
if digit == '8':
|
|
self.bold = True
|
|
elif digit == '9':
|
|
self.underline = True
|
|
elif digit == '0':
|
|
self.bold = False
|
|
self.underline = False
|
|
elif digit == '7':
|
|
self.color = 0
|
|
else:
|
|
self.color = int(digit)
|
|
|
|
def attributes(self, extra=0):
|
|
"""Curses attributes for this state, combined with extra attributes"""
|
|
attributes = curses.color_pair(self.color) | extra
|
|
if self.bold:
|
|
attributes |= curses.A_BOLD
|
|
if self.underline:
|
|
attributes |= curses.A_UNDERLINE
|
|
return attributes
|
|
|
|
def codes(self):
|
|
"""Color codes that recreate this state at the start of a wrapped row"""
|
|
codes = ''
|
|
if self.color:
|
|
codes += f'^{self.color}'
|
|
if self.bold:
|
|
codes += '^8'
|
|
if self.underline:
|
|
codes += '^9'
|
|
return codes
|
|
|
|
|
|
def print_colored(window, message, attributes=0):
|
|
"""Print a message at the window cursor, interpreting Quake color codes (^N)"""
|
|
state = ColorState()
|
|
for token in COLOR_TOKEN_PATTERN.findall(message):
|
|
if len(token) == 2:
|
|
state.apply(token[1])
|
|
continue
|
|
try:
|
|
window.addch(token, state.attributes(attributes))
|
|
except curses.error:
|
|
# The window is full (bottom-right corner reached)
|
|
return
|
|
|
|
|
|
def wrap_colored(line, width):
|
|
"""
|
|
Split one line into rows of at most `width` visible characters.
|
|
Color codes take no space and the active codes are repeated at the start
|
|
of each continuation row, so colors survive the wrap.
|
|
Example: wrap_colored('^1abcdef', 4) -> ['^1abcd', '^1ef']
|
|
"""
|
|
rows = []
|
|
state = ColorState()
|
|
row = ''
|
|
visible_chars = 0
|
|
for token in COLOR_TOKEN_PATTERN.findall(line):
|
|
if len(token) == 2:
|
|
state.apply(token[1])
|
|
row += token
|
|
continue
|
|
if visible_chars == width:
|
|
rows.append(row)
|
|
row = state.codes()
|
|
visible_chars = 0
|
|
row += token
|
|
visible_chars += 1
|
|
rows.append(row)
|
|
return rows
|
|
|
|
|
|
def safe_addstr(window, y, x, text, attributes=0):
|
|
"""Write text at (y, x), truncated to the window; never raises"""
|
|
height, width = window.getmaxyx()
|
|
if y >= height or x >= width:
|
|
return
|
|
try:
|
|
window.addstr(y, x, text[:width - x - 1], attributes)
|
|
except curses.error:
|
|
pass
|
|
|
|
|
|
def signature_hint(command, arg_position):
|
|
"""Hint parts for a command signature with the current argument highlighted"""
|
|
parts = []
|
|
for arg_text, is_current in get_signature_with_highlight(command, arg_position):
|
|
parts.append((arg_text, curses.A_REVERSE if is_current else curses.A_DIM))
|
|
parts.append((' ', 0))
|
|
return parts
|
|
|
|
|
|
def suggestion_list_hint(suggestions, arg_type=None):
|
|
"""One hint line listing the first suggestions, e.g. '<map>: campgrounds bloodrun (+3 more)'"""
|
|
shown = ' '.join(suggestions[:MAX_HINT_SUGGESTIONS])
|
|
hidden = len(suggestions) - MAX_HINT_SUGGESTIONS
|
|
more = f' (+{hidden} more)' if hidden > 0 else ''
|
|
if arg_type:
|
|
return f'<{arg_type}>: {shown}{more}'
|
|
return f'{shown}{more}'
|
|
|
|
|
|
def compute_autocomplete(words, ends_with_space, player_names_provider, learned_names_provider=None):
|
|
"""
|
|
Work out what to suggest for the current input.
|
|
player_names_provider / learned_names_provider: callables returning the
|
|
current player names and the cvar/command names learned from the server.
|
|
Returns (suggestions, original_word, hint_parts):
|
|
- suggestions: values cycled with Tab
|
|
- original_word: the word Tab replaces ('' means Tab appends a new word)
|
|
- hint_parts: [(text, attributes)] shown below the input line
|
|
"""
|
|
first_word = words[0].lower()
|
|
typing_command_name = len(words) == 1 and not ends_with_space
|
|
|
|
if first_word in COMMAND_ARGUMENTS and not typing_command_name:
|
|
if ends_with_space:
|
|
arg_position = len(words) - 1
|
|
current_value = ''
|
|
else:
|
|
arg_position = len(words) - 2
|
|
current_value = words[-1]
|
|
|
|
player_names = player_names_provider() if player_names_provider else None
|
|
arg_suggestions = get_argument_suggestions(first_word, arg_position, current_value, player_list=player_names)
|
|
if arg_suggestions:
|
|
arg_type = COMMAND_ARGUMENTS[first_word][arg_position]['type']
|
|
hint = suggestion_list_hint(arg_suggestions, arg_type)
|
|
return arg_suggestions, current_value, [(hint, curses.A_DIM)]
|
|
|
|
# No value suggestions (free text, no players known): show the signature
|
|
return [], '', signature_hint(first_word, arg_position)
|
|
|
|
if COMMAND_SIGNATURES.get(first_word) and not typing_command_name:
|
|
# Command with a signature but no argument definitions
|
|
return [], '', signature_hint(first_word, 0)
|
|
|
|
# Still typing the command name, or an unknown command: suggest names matching
|
|
# the current word. A complete name like `map` stays here on purpose because it
|
|
# can be the prefix of others (map_restart, maplist); its signature is shown too.
|
|
current_word = words[-1]
|
|
if len(current_word) < 2:
|
|
return [], '', []
|
|
learned_names = learned_names_provider() if learned_names_provider else ()
|
|
suggestions = autocomplete(current_word, learned_names, max_results=5)
|
|
hint_parts = []
|
|
if suggestions:
|
|
hint_parts.append((' '.join(suggestions), curses.A_DIM))
|
|
if typing_command_name and COMMAND_SIGNATURES.get(first_word):
|
|
hint_parts.append((' ', 0))
|
|
hint_parts.extend(signature_hint(first_word, 0))
|
|
return suggestions, current_word, hint_parts
|
|
|
|
|
|
class InputLine:
|
|
"""The command input line: text, cursor, history and autocomplete state"""
|
|
|
|
def __init__(self, max_history, player_names_provider=None, learned_names_provider=None):
|
|
self.max_history = max_history
|
|
self.player_names_provider = player_names_provider
|
|
self.learned_names_provider = learned_names_provider
|
|
self.text = ''
|
|
self.cursor = 0
|
|
self.history = []
|
|
self.history_index = -1 # -1 = not browsing history
|
|
self.saved_text = '' # what was typed before browsing history
|
|
self.suggestions = []
|
|
self.suggestion_index = -1
|
|
self.original_word = ''
|
|
self.hint_parts = [] # [(text, attributes)] drawn below the input
|
|
|
|
def handle_key(self, key):
|
|
"""Apply one key press. Returns the submitted command on Enter, else None."""
|
|
if key in (curses.KEY_ENTER, 10, 13):
|
|
return self._submit()
|
|
|
|
if key == 9: # Tab
|
|
self._cycle_suggestion()
|
|
elif key == curses.KEY_UP:
|
|
self._history_previous()
|
|
elif key == curses.KEY_DOWN:
|
|
self._history_next()
|
|
elif key == curses.KEY_LEFT:
|
|
self.cursor = max(0, self.cursor - 1)
|
|
elif key == curses.KEY_RIGHT:
|
|
self.cursor = min(len(self.text), self.cursor + 1)
|
|
elif key == curses.KEY_HOME:
|
|
self.cursor = 0
|
|
elif key == curses.KEY_END:
|
|
self.cursor = len(self.text)
|
|
elif key in (curses.KEY_BACKSPACE, 127, 8):
|
|
if self.cursor > 0:
|
|
self.text = self.text[:self.cursor - 1] + self.text[self.cursor:]
|
|
self.cursor -= 1
|
|
self._text_changed()
|
|
elif key == curses.KEY_DC:
|
|
if self.cursor < len(self.text):
|
|
self.text = self.text[:self.cursor] + self.text[self.cursor + 1:]
|
|
self._text_changed()
|
|
elif 32 <= key <= 126:
|
|
self.text = self.text[:self.cursor] + chr(key) + self.text[self.cursor:]
|
|
self.cursor += 1
|
|
self._text_changed()
|
|
return None
|
|
|
|
def render(self, window):
|
|
"""Draw text, autocomplete hint and cursor into the input window"""
|
|
window.erase()
|
|
height, width = window.getmaxyx()
|
|
usable = width - 1
|
|
|
|
# Show the part of the text around the cursor when it is wider than the window
|
|
start = max(0, self.cursor - usable)
|
|
safe_addstr(window, 0, 0, self.text[start:start + usable])
|
|
|
|
x = 0
|
|
for text, attributes in self.hint_parts:
|
|
if x >= usable:
|
|
break
|
|
safe_addstr(window, 1, x, text, attributes)
|
|
x += len(text)
|
|
|
|
window.move(0, self.cursor - start)
|
|
|
|
def _submit(self):
|
|
command = self.text
|
|
if not command:
|
|
return None
|
|
self.history.append(command)
|
|
if len(self.history) > self.max_history:
|
|
self.history.pop(0)
|
|
self.text = ''
|
|
self.cursor = 0
|
|
self.history_index = -1
|
|
self.saved_text = ''
|
|
self._clear_suggestions()
|
|
return command
|
|
|
|
def _clear_suggestions(self):
|
|
self.suggestions = []
|
|
self.suggestion_index = -1
|
|
self.original_word = ''
|
|
self.hint_parts = []
|
|
|
|
def _text_changed(self):
|
|
"""Recompute autocomplete after an edit; editing leaves history browsing"""
|
|
self.history_index = -1
|
|
words = self.text.split()
|
|
if not words:
|
|
self._clear_suggestions()
|
|
return
|
|
self.suggestions, self.original_word, self.hint_parts = compute_autocomplete(
|
|
words, self.text.endswith(' '), self.player_names_provider, self.learned_names_provider)
|
|
self.suggestion_index = -1
|
|
|
|
def _history_previous(self):
|
|
if not self.history:
|
|
return
|
|
if self.history_index == -1:
|
|
self.saved_text = self.text
|
|
self.history_index = len(self.history)
|
|
if self.history_index > 0:
|
|
self.history_index -= 1
|
|
self._show_history_entry(self.history[self.history_index])
|
|
|
|
def _history_next(self):
|
|
if self.history_index == -1:
|
|
return
|
|
self.history_index += 1
|
|
if self.history_index >= len(self.history):
|
|
self._show_history_entry(self.saved_text)
|
|
self.history_index = -1
|
|
self.saved_text = ''
|
|
else:
|
|
self._show_history_entry(self.history[self.history_index])
|
|
|
|
def _show_history_entry(self, text):
|
|
self.text = text
|
|
self.cursor = len(text)
|
|
self._clear_suggestions()
|
|
|
|
def _cycle_suggestion(self):
|
|
"""Tab: put the next suggestion into the input line"""
|
|
if not self.suggestions:
|
|
return
|
|
words = self.text.split()
|
|
if not words:
|
|
return
|
|
|
|
self.suggestion_index = (self.suggestion_index + 1) % len(self.suggestions)
|
|
selected = self.suggestions[self.suggestion_index]
|
|
|
|
if self.original_word == '':
|
|
# The input ended with a space: append the suggestion as a new word
|
|
words.append(selected)
|
|
# From now on Tab replaces this word instead of appending another
|
|
self.original_word = selected
|
|
else:
|
|
words[-1] = selected
|
|
|
|
self.text = ' '.join(words)
|
|
self.cursor = len(self.text)
|
|
self.hint_parts = [(self._cycle_hint(words, selected), curses.A_DIM)]
|
|
|
|
def _cycle_hint(self, words, selected):
|
|
"""Hint line while cycling: argument values with their type, or commands with signature"""
|
|
first_word = words[0].lower()
|
|
if first_word in COMMAND_ARGUMENTS and len(words) > 1:
|
|
# After ' '.join the text never ends with a space, so the last word is the argument
|
|
arg_position = len(words) - 2
|
|
if arg_position < len(COMMAND_ARGUMENTS[first_word]):
|
|
arg_type = COMMAND_ARGUMENTS[first_word][arg_position]['type']
|
|
return suggestion_list_hint(self.suggestions, arg_type)
|
|
return suggestion_list_hint(self.suggestions)
|
|
|
|
hint = suggestion_list_hint(self.suggestions)
|
|
signature = COMMAND_SIGNATURES.get(selected)
|
|
if signature:
|
|
return f'{hint} → {signature}'
|
|
return hint
|
|
|
|
|
|
def select_server(screen, servers):
|
|
"""
|
|
Full-screen menu listing the configured servers.
|
|
servers: dict name -> host. Returns the chosen name, or None when the user quits.
|
|
Keys: Up/Down or j/k move, Enter connects, 0-9 connect directly, q or Esc quit.
|
|
"""
|
|
names = list(servers)
|
|
selected = 0
|
|
screen.keypad(True)
|
|
screen.nodelay(False)
|
|
curses.curs_set(0)
|
|
|
|
while True:
|
|
_draw_server_menu(screen, servers, names, selected)
|
|
try:
|
|
key = screen.getch()
|
|
except KeyboardInterrupt:
|
|
return None
|
|
|
|
if key in (curses.KEY_UP, ord('k')):
|
|
selected = (selected - 1) % len(names)
|
|
elif key in (curses.KEY_DOWN, ord('j')):
|
|
selected = (selected + 1) % len(names)
|
|
elif key in (curses.KEY_ENTER, 10, 13):
|
|
return names[selected]
|
|
elif ord('0') <= key <= ord('9') and key - ord('0') < len(names):
|
|
return names[key - ord('0')]
|
|
elif key in (ord('q'), 27): # 27 = Escape
|
|
return None
|
|
# Any other key (including KEY_RESIZE) just redraws
|
|
|
|
|
|
def _draw_server_menu(screen, servers, names, selected):
|
|
screen.erase()
|
|
safe_addstr(screen, 0, 2, 'Quake Live PyCon - select a server', curses.A_BOLD)
|
|
for index, name in enumerate(names):
|
|
line = f'{index:>2}. {name:<12} {servers[name]}'
|
|
attributes = curses.A_REVERSE if index == selected else 0
|
|
safe_addstr(screen, 2 + index, 2, line, attributes)
|
|
safe_addstr(screen, 3 + len(names), 2,
|
|
'Up/Down or j/k: move Enter or number: connect q: quit', curses.A_DIM)
|
|
screen.refresh()
|
|
|
|
|
|
class UIManager:
|
|
"""Manages curses windows and display"""
|
|
|
|
def __init__(self, screen, title, max_history=MAX_COMMAND_HISTORY,
|
|
player_names_provider=None, learned_names_provider=None):
|
|
self.screen = screen
|
|
self.title = title
|
|
self.info_window = None
|
|
self.output_window = None
|
|
self.divider_window = None
|
|
self.input_window = None
|
|
self.too_small = False
|
|
|
|
# Output scrollback: one entry per line, color codes included
|
|
self.output_lines = deque(maxlen=OUTPUT_SCROLLBACK_LINES)
|
|
# True when the last message had no trailing newline: the next message continues that line
|
|
# (the server sends e.g. each column of a status row as its own message)
|
|
self.line_open = False
|
|
# Visual rows scrolled up from the newest output; 0 = follow new output
|
|
self.scroll_offset = 0
|
|
|
|
self.input_line = InputLine(max_history, player_names_provider, learned_names_provider)
|
|
self.game_state = None # last state shown in the info window
|
|
|
|
self._init_curses()
|
|
self._layout()
|
|
self.redraw()
|
|
|
|
# --- setup and layout ---
|
|
|
|
def _init_curses(self):
|
|
self.screen.nodelay(True)
|
|
self.screen.keypad(True)
|
|
curses.start_color()
|
|
curses.use_default_colors()
|
|
curses.curs_set(1) # Show cursor in input window
|
|
|
|
# Ask the terminal to report the mouse wheel so it scrolls the output
|
|
# (the terminal then needs Shift held for text selection)
|
|
curses.mousemask(curses.BUTTON4_PRESSED | curses.BUTTON5_PRESSED)
|
|
|
|
# Color pairs 1-6 on the terminal's default background
|
|
for i in range(1, 7):
|
|
curses.init_pair(i, i, -1)
|
|
|
|
# Swap cyan and magenta (5 and 6)
|
|
curses.init_pair(5, 6, -1)
|
|
curses.init_pair(6, 5, -1)
|
|
|
|
def _layout(self):
|
|
"""Create the windows for the current terminal size, or show the too-small warning"""
|
|
rows, cols = self.screen.getmaxyx()
|
|
self.screen.erase()
|
|
|
|
if rows < MIN_ROWS or cols < MIN_COLS:
|
|
self.too_small = True
|
|
self.info_window = None
|
|
self.output_window = None
|
|
self.divider_window = None
|
|
self.input_window = None
|
|
warning = f'Terminal too small - resize to at least {MIN_ROWS}x{MIN_COLS}'
|
|
safe_addstr(self.screen, rows // 2, max(0, (cols - len(warning)) // 2), warning)
|
|
self.screen.noutrefresh()
|
|
curses.doupdate()
|
|
return
|
|
|
|
self.too_small = False
|
|
safe_addstr(self.screen, 0, 0, self.title)
|
|
safe_addstr(self.screen, rows - 2, 2, '$ ')
|
|
self.screen.noutrefresh()
|
|
|
|
panel_width = cols - 4
|
|
output_height = rows - OUTPUT_WINDOW_Y - 3 # 3 = divider + input window
|
|
|
|
self.info_window = curses.newwin(INFO_WINDOW_HEIGHT, panel_width, INFO_WINDOW_Y, 2)
|
|
self.info_window.leaveok(True)
|
|
|
|
self.output_window = curses.newwin(output_height, panel_width, OUTPUT_WINDOW_Y, 2)
|
|
self.output_window.leaveok(True)
|
|
|
|
self.divider_window = curses.newwin(1, panel_width, rows - 3, 2)
|
|
self.divider_window.leaveok(True)
|
|
|
|
self.input_window = curses.newwin(INPUT_WINDOW_HEIGHT, cols - 6, rows - 2, 4)
|
|
|
|
def handle_resize(self):
|
|
curses.update_lines_cols()
|
|
self._layout()
|
|
self.redraw()
|
|
|
|
# --- drawing ---
|
|
|
|
def redraw(self):
|
|
"""Redraw every window from state"""
|
|
if self.too_small:
|
|
return
|
|
self._draw_output()
|
|
self._draw_divider()
|
|
if self.game_state is not None:
|
|
self._draw_info()
|
|
self._flush()
|
|
|
|
def _flush(self):
|
|
"""Push pending drawing to the terminal with the cursor in the input line"""
|
|
if self.too_small:
|
|
return
|
|
self.input_line.render(self.input_window)
|
|
self.input_window.noutrefresh()
|
|
curses.doupdate()
|
|
|
|
def _rows_newest_first(self, width):
|
|
"""Yield the wrapped visual rows of the output, newest row first"""
|
|
for line in reversed(self.output_lines):
|
|
for row in reversed(wrap_colored(line, width)):
|
|
yield row
|
|
|
|
def _draw_output(self):
|
|
height, width = self.output_window.getmaxyx()
|
|
wanted = height + self.scroll_offset
|
|
rows = list(itertools.islice(self._rows_newest_first(width), wanted))
|
|
if len(rows) < wanted:
|
|
# Scrolled past the oldest line: clamp to the top
|
|
self.scroll_offset = max(0, len(rows) - height)
|
|
|
|
visible = rows[self.scroll_offset:self.scroll_offset + height]
|
|
visible.reverse()
|
|
|
|
self.output_window.erase()
|
|
for y, row in enumerate(visible):
|
|
self.output_window.move(y, 0)
|
|
print_colored(self.output_window, row)
|
|
self.output_window.noutrefresh()
|
|
|
|
def _draw_divider(self):
|
|
_, width = self.divider_window.getmaxyx()
|
|
self.divider_window.erase()
|
|
self.divider_window.hline(0, 0, curses.ACS_HLINE, width)
|
|
if self.scroll_offset > 0:
|
|
marker = f' scrolled back {self.scroll_offset} rows - PgDn/End for latest '
|
|
safe_addstr(self.divider_window, 0, max(0, width - len(marker) - 2), marker, curses.A_REVERSE)
|
|
self.divider_window.noutrefresh()
|
|
|
|
# --- input ---
|
|
|
|
def process_input(self):
|
|
"""Read all pending keys from the terminal. Returns the commands the user submitted."""
|
|
commands = []
|
|
key_seen = False
|
|
while True:
|
|
key = self.screen.getch()
|
|
if key == -1:
|
|
break
|
|
key_seen = True
|
|
if key == curses.KEY_RESIZE:
|
|
self.handle_resize()
|
|
elif key == curses.KEY_PPAGE:
|
|
self.scroll_output(self._page_rows())
|
|
elif key == curses.KEY_NPAGE:
|
|
self.scroll_output(-self._page_rows())
|
|
elif key == curses.KEY_MOUSE:
|
|
self._handle_mouse()
|
|
else:
|
|
command = self.input_line.handle_key(key)
|
|
if command:
|
|
commands.append(command)
|
|
if key_seen:
|
|
self._flush()
|
|
return commands
|
|
|
|
def _handle_mouse(self):
|
|
"""Mouse wheel up scrolls the output back, wheel down scrolls forward"""
|
|
try:
|
|
_, _, _, _, button_state = curses.getmouse()
|
|
except curses.error:
|
|
# Event already consumed or not a mouse event
|
|
return
|
|
if button_state & curses.BUTTON4_PRESSED:
|
|
self.scroll_output(MOUSE_WHEEL_ROWS)
|
|
elif button_state & curses.BUTTON5_PRESSED:
|
|
self.scroll_output(-MOUSE_WHEEL_ROWS)
|
|
|
|
def _page_rows(self):
|
|
if self.output_window is None:
|
|
return 0
|
|
height, _ = self.output_window.getmaxyx()
|
|
return max(1, height - 1)
|
|
|
|
def scroll_output(self, rows):
|
|
"""Scroll the output view up (positive) or down (negative) by visual rows"""
|
|
self.scroll_offset = max(0, self.scroll_offset + rows)
|
|
if self.too_small:
|
|
return
|
|
self._draw_output()
|
|
self._draw_divider()
|
|
|
|
# --- output ---
|
|
|
|
def setup_logging(self):
|
|
"""Create the logging handler that writes into the output window"""
|
|
handler = CursesHandler(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):
|
|
"""
|
|
Append a message to the output.
|
|
Example: 'abc' then 'def\\n' gives one line 'abcdef'; 'x\\ny\\n' gives two lines.
|
|
"""
|
|
pieces = message.split('\n')
|
|
ends_with_newline = pieces[-1] == ''
|
|
if ends_with_newline:
|
|
pieces.pop()
|
|
|
|
new_lines = []
|
|
for piece in pieces:
|
|
if self.line_open:
|
|
self.output_lines[-1] += piece
|
|
self.line_open = False
|
|
else:
|
|
self.output_lines.append(piece)
|
|
new_lines.append(piece)
|
|
self.line_open = not ends_with_newline
|
|
|
|
if self.too_small:
|
|
return
|
|
|
|
if self.scroll_offset > 0:
|
|
# Keep the view still while the user reads old output
|
|
_, width = self.output_window.getmaxyx()
|
|
for line in new_lines:
|
|
self.scroll_offset += len(wrap_colored(line, width))
|
|
|
|
self._draw_output()
|
|
self._draw_divider()
|
|
self._flush()
|
|
|
|
def update_server_info(self, game_state):
|
|
"""Update server info window"""
|
|
self.game_state = game_state
|
|
if self.too_small:
|
|
return
|
|
self._draw_info()
|
|
self._flush()
|
|
|
|
def _draw_info(self):
|
|
game_state = self.game_state
|
|
self.info_window.erase()
|
|
|
|
max_y, max_x = self.info_window.getmaxyx()
|
|
server_info = game_state.server_info
|
|
|
|
# Line 1: Hostname with Timer and Warmup Indicator
|
|
hostname = server_info.hostname
|
|
|
|
timer_display = ""
|
|
if server_info.match_time > 0 and not server_info.warmup:
|
|
# Calculate live time: add elapsed seconds since last server update
|
|
current_time = server_info.match_time
|
|
if server_info.match_time_last_sync > 0:
|
|
elapsed = int(time.time() - server_info.match_time_last_sync)
|
|
current_time += elapsed
|
|
|
|
mins = current_time // 60
|
|
secs = current_time % 60
|
|
timer_display = f"^3^0Time:^8^7 {mins}:{secs:02d}^0"
|
|
else:
|
|
timer_display = "^3^0Time:^8^7 0:00^0"
|
|
|
|
warmup_display = "^3^0Warmup:^8 ^2YES^0" if server_info.warmup else "^3^0Warmup: ^8^1NO^0"
|
|
|
|
print_colored(self.info_window, f"^3Name:^8 {hostname} {warmup_display} {timer_display}\n", 0)
|
|
|
|
# Line 2: Game info
|
|
gametype = server_info.gametype
|
|
mapname = server_info.map
|
|
timelimit = server_info.timelimit
|
|
fraglimit = server_info.fraglimit
|
|
roundlimit = server_info.roundlimit
|
|
caplimit = server_info.capturelimit
|
|
curclients = len(server_info.players)
|
|
maxclients = server_info.maxclients
|
|
|
|
# Context-sensitive limit display based on gametype
|
|
if gametype == 'Capture the Flag':
|
|
limit_display = f"^3^0| Capturelimit:^7^8 {caplimit}"
|
|
elif gametype == 'Clan Arena':
|
|
limit_display = f"^3^0| Roundlimit:^7^8 {roundlimit}"
|
|
elif gametype == 'Duel':
|
|
limit_display = f"^3^0| Timelimit:^7^8 {timelimit}"
|
|
elif gametype == 'Race':
|
|
limit_display = f"^3^0| Timelimit:^7^8 {timelimit}"
|
|
else:
|
|
limit_display = f"^3^0| Timelimit:^7^8 {timelimit} ^0^3| Fraglimit:^7^8 {fraglimit}"
|
|
|
|
print_colored(self.info_window,
|
|
f"^3^0Type:^7^8 {gametype} ^0^3| Map:^7^8 {mapname} ^0^3| Players:^7^8 {curclients}/{maxclients} "
|
|
f"{limit_display}^0\n", 0)
|
|
|
|
# Blank lines to fill
|
|
try:
|
|
self.info_window.addstr("\n")
|
|
except curses.error:
|
|
pass
|
|
|
|
# Line 3: Team headers and player lists
|
|
teams = game_state.player_tracker.get_players_by_team()
|
|
|
|
if server_info.gametype in TEAM_MODES:
|
|
if server_info.gametype == 'Clan Arena':
|
|
red_score = f"{server_info.red_rounds:>3} "
|
|
blue_score = f"{server_info.blue_rounds:>3} "
|
|
|
|
else:
|
|
red_total = 0
|
|
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
|
|
elif team == 'BLUE':
|
|
blue_total += score
|
|
|
|
red_score = f"{red_total:>3} "
|
|
blue_score = f"{blue_total:>3} "
|
|
|
|
print_colored(self.info_window, f"^8^7{red_score}^9^1RED TEAM^0 ^7^8{blue_score}^9^4BLUE TEAM\n", 0)
|
|
|
|
# Sort players by score within each team
|
|
red_players_with_scores = []
|
|
blue_players_with_scores = []
|
|
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
|
|
red_players_with_scores.sort(key=lambda x: x[1], reverse=True)
|
|
blue_players_with_scores.sort(key=lambda x: x[1], reverse=True)
|
|
|
|
red_players = [name for name, score in red_players_with_scores[:4]]
|
|
blue_players = [name for name, score in blue_players_with_scores[:4]]
|
|
spec_players = teams['SPECTATOR'][:4]
|
|
|
|
for i in range(4):
|
|
red_name = red_players[i] if i < len(red_players) else ''
|
|
blue_name = blue_players[i] if i < len(blue_players) else ''
|
|
|
|
# Get scores for team players
|
|
red_score = server_info.players.get(red_name, {}).get('score', '0') if red_name else ''
|
|
blue_score = server_info.players.get(blue_name, {}).get('score', '0') if blue_name else ''
|
|
|
|
# Check if players are dead
|
|
red_dead = red_name in server_info.dead_players
|
|
blue_dead = blue_name in server_info.dead_players
|
|
|
|
# Format with strikethrough for dead players (using dim text)
|
|
red = f"{red_score:>3} {'^8^1X^7^0 ' if red_dead else ''}{red_name}" if red_name else ''
|
|
blue = f"{blue_score:>3} {'^8^1X^7^0 ' if blue_dead else ''}{blue_name}" if blue_name else ''
|
|
|
|
red_clean = strip_color_codes(red)
|
|
blue_clean = strip_color_codes(blue)
|
|
|
|
red_pad = 24 - len(red_clean)
|
|
|
|
line = f"^8{red}^0{' ' * red_pad}^8{blue}^0\n"
|
|
print_colored(self.info_window, line, 0)
|
|
else:
|
|
print_colored(self.info_window, f" ^8^9^5FREE\n", 0)
|
|
# Sort FREE players by score (highest first)
|
|
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
|
|
free_players_with_scores.sort(key=lambda x: x[1], reverse=True)
|
|
sorted_free_players = [name for name, score in free_players_with_scores]
|
|
|
|
spec_players = teams['SPECTATOR'][:4]
|
|
free_col1 = sorted_free_players[:4]
|
|
free_col2 = sorted_free_players[4:8]
|
|
|
|
for i in range(4):
|
|
col1_name = free_col1[i] if i < len(free_col1) else ''
|
|
col2_name = free_col2[i] if i < len(free_col2) else ''
|
|
|
|
# Get scores for FREE players
|
|
col1_score = server_info.players.get(col1_name, {}).get('score', '0') if col1_name else ''
|
|
col2_score = server_info.players.get(col2_name, {}).get('score', '0') if col2_name else ''
|
|
|
|
# Check if players are dead
|
|
col1_dead = col1_name in server_info.dead_players
|
|
col2_dead = col2_name in server_info.dead_players
|
|
|
|
# Format: " 9 PlayerName" with right-aligned score and dead marker
|
|
col1 = f"{col1_score:>3} {'^8^1X^7^0 ' if col1_dead else ''}{col1_name}" if col1_name else ''
|
|
col2 = f"{col2_score:>3} {'^8^1X^7^0 ' if col2_dead else ''}{col2_name}" if col2_name else ''
|
|
|
|
col1_clean = strip_color_codes(col1)
|
|
col2_clean = strip_color_codes(col2)
|
|
|
|
col1_pad = 24 - len(col1_clean)
|
|
|
|
line = f"^8{col1}^0{' ' * col1_pad}^8{col2}^0\n"
|
|
print_colored(self.info_window, line, 0)
|
|
|
|
# Blank lines to fill
|
|
try:
|
|
self.info_window.addstr("\n")
|
|
except curses.error:
|
|
pass
|
|
|
|
# List spectators on one line
|
|
spec_list = " ".join(spec_players)
|
|
line = f"^8^3Spectators:^7 {spec_list}\n"
|
|
print_colored(self.info_window, line, 0)
|
|
|
|
# Blank lines to fill
|
|
try:
|
|
self.info_window.addstr("\n")
|
|
except curses.error:
|
|
pass
|
|
|
|
# Separator
|
|
separator = "^7" + "=" * (max_x - 1) + "^7"
|
|
print_colored(self.info_window, separator, 0)
|
|
|
|
self.info_window.noutrefresh()
|