DebianandSisyphus
c1b5f6d86a
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <[email protected]>
930 lines
34 KiB
Python
930 lines
34 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,
|
|
TEAM_COLORS)
|
|
from .cvars import autocomplete, COMMAND_SIGNATURES, get_signature_with_highlight, get_argument_suggestions, COMMAND_ARGUMENTS
|
|
from .formatter import get_team_prefix, 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()
|
|
|
|
|
|
def format_damage_summary(server_info, player_tracker=None):
|
|
"""One-line damage summary (top 3 by damage dealt) from LIVESTATS counters,
|
|
tagged with the tracked team of each player (data-driven, so tags work
|
|
during warmup too); None when no damage data exists yet. Dealt/taken are
|
|
stored as strings."""
|
|
entries = []
|
|
for player_name, player_data in server_info.players.items():
|
|
try:
|
|
dealt = int(player_data.get('dealt', 0))
|
|
taken = int(player_data.get('taken', 0))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if dealt <= 0 and taken <= 0:
|
|
continue
|
|
entries.append((strip_color_codes(player_name), dealt, taken))
|
|
if not entries:
|
|
return None
|
|
entries.sort(key=lambda x: x[1], reverse=True)
|
|
parts = []
|
|
for name, dealt, taken in entries[:3]:
|
|
prefix = ''
|
|
if player_tracker is not None:
|
|
prefix = TEAM_COLORS.get(player_tracker.get_team(name), '')
|
|
parts.append(f"{prefix}{name} {dealt}/{taken}")
|
|
return f"^8^3DMG^7 {' '.join(parts)}\n"
|
|
|
|
|
|
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 self.too_small:
|
|
# Warning screen: ignore typing and scrolling, only resizes act
|
|
continue
|
|
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()
|
|
self._flush()
|
|
|
|
# --- 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 = []
|
|
continued_delta = 0
|
|
for piece in pieces:
|
|
if self.line_open:
|
|
if not self.too_small and self.scroll_offset > 0:
|
|
# The continued line can gain wrapped rows; count them so a
|
|
# scrolled-back view stays still
|
|
_, width = self.output_window.getmaxyx()
|
|
rows_before = len(wrap_colored(self.output_lines[-1], width))
|
|
self.output_lines[-1] += piece
|
|
continued_delta += len(wrap_colored(self.output_lines[-1], width)) - rows_before
|
|
else:
|
|
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.scroll_offset += continued_delta
|
|
|
|
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)
|
|
|
|
# Damage summary from LIVESTATS counters (top 3 by damage dealt)
|
|
damage_line = format_damage_summary(server_info, game_state.player_tracker)
|
|
if damage_line:
|
|
print_colored(self.info_window, damage_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()
|