pfl
db123993c0
- All curses drawing on the main thread; stdin polled together with the ZMQ sockets, Ctrl-C only sets a flag - Output kept in a 2000-line buffer and redrawn from it, so resizes keep the history; PgUp/PgDn and mouse wheel scroll it - Messages without trailing newline continue the current line (status table columns arrive as separate messages) - Too-small terminal shows a warning instead of exiting - Start without arguments shows a server selection menu - Colors on the terminal default background; version 0.9.0
639 lines
25 KiB
Python
639 lines
25 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
QLPyCon - Quake Live Python Console
|
|
Main entry point
|
|
"""
|
|
|
|
import argparse
|
|
import uuid
|
|
import logging
|
|
import time
|
|
import re
|
|
import curses
|
|
import zmq
|
|
import signal
|
|
import sys
|
|
import os
|
|
|
|
from lib.constants import VERSION, DEFAULT_HOST, POLL_TIMEOUT, QUIT_CONFIRM_TIMEOUT, RESPAWN_DELAY, MAX_COMMAND_HISTORY
|
|
from lib.state import GameState
|
|
from lib.network import RconConnection, StatsConnection
|
|
from lib.parser import EventParser
|
|
from lib.formatter import format_message, format_chat_message, format_powerup_message, strip_color_codes
|
|
from lib.ui import UIManager, select_server
|
|
from lib.settings import ConfigLoader
|
|
|
|
# Pre-compiled regex patterns
|
|
CVAR_RESPONSE_PATTERN = re.compile(r'"([^"]+)"\s+is:"([^"]*)"')
|
|
PORT_PATTERN = re.compile(r'(\d+)')
|
|
BROADCAST_PATTERN = re.compile(r'^broadcast:\s*print\s*"(.+?)(?:\\n)?"\s*$')
|
|
TIMESTAMP_PATTERN = re.compile(r'\^\d\[\^\d[0-9:]+\^\d\]\^\d\s*|\[[0-9:]+\]\s*')
|
|
CONNECT_PATTERN = re.compile(r'^(.+?)\s+connected')
|
|
DISCONNECT_PATTERN = re.compile(r'^(.+?)\s+disconnected')
|
|
KICK_PATTERN = re.compile(r'^(.+?)\s+was kicked')
|
|
INACTIVITY_PATTERN = re.compile(r'^(.+?)\s+Dropped due to inactivity')
|
|
RENAME_PATTERN = re.compile(r'^(.+?)\s+renamed to\s+(.+?)$')
|
|
|
|
# Configure logging
|
|
logger = logging.getLogger('main')
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
all_json_logger = logging.getLogger('all_json')
|
|
all_json_logger.setLevel(logging.DEBUG)
|
|
|
|
unknown_json_logger = logging.getLogger('unknown_json')
|
|
unknown_json_logger.setLevel(logging.DEBUG)
|
|
|
|
# Ctrl-C state: the signal handler only records the press, the main loop
|
|
# reacts to it, so no screen drawing happens inside a signal handler.
|
|
ctrl_c_pressed = False
|
|
quit_confirm_time = None
|
|
|
|
# Configurable timeouts (set from config in __main__)
|
|
_QUIT_CONFIRM_TIMEOUT = QUIT_CONFIRM_TIMEOUT
|
|
_RESPAWN_DELAY = RESPAWN_DELAY
|
|
|
|
# Global shutdown flag (set by handle_ctrl_c, checked by main_loop)
|
|
shutdown_requested = False
|
|
|
|
def signal_handler(sig, frame):
|
|
"""Record Ctrl+C; force exit if the graceful shutdown already started"""
|
|
global ctrl_c_pressed
|
|
if shutdown_requested:
|
|
# Third Ctrl-C: last-resort forced exit
|
|
curses.endwin()
|
|
os._exit(0)
|
|
ctrl_c_pressed = True
|
|
|
|
|
|
def handle_ctrl_c():
|
|
"""Ask for confirmation on the first Ctrl-C, quit on the second one within the timeout"""
|
|
global ctrl_c_pressed, quit_confirm_time, shutdown_requested
|
|
if not ctrl_c_pressed:
|
|
return
|
|
ctrl_c_pressed = False
|
|
|
|
current_time = time.time()
|
|
if quit_confirm_time is None or (current_time - quit_confirm_time) > _QUIT_CONFIRM_TIMEOUT:
|
|
logger.warning(f"^1^8Press Ctrl-C again within {_QUIT_CONFIRM_TIMEOUT:.0f} seconds to quit^0")
|
|
quit_confirm_time = current_time
|
|
else:
|
|
logger.warning("^1^8Quittin'^0")
|
|
shutdown_requested = True
|
|
|
|
def parse_cvar_response(message, game_state, ui):
|
|
"""
|
|
Parse server cvar responses and update state
|
|
Returns True if message should be suppressed from display
|
|
"""
|
|
# Suppress command echo lines
|
|
if message.startswith('zmq RCON command') and 'from' in message:
|
|
suppress_cmds = ['status', 'roundlimit', 'qlx_serverBrandName', 'g_factoryTitle',
|
|
'mapname', 'timelimit', 'fraglimit', 'capturelimit', 'sv_maxclients']
|
|
if any(f': {cmd}' in message for cmd in suppress_cmds):
|
|
return True
|
|
|
|
# Parse cvar responses (format: "cvar_name" is:"value" default:...)
|
|
cvar_match = CVAR_RESPONSE_PATTERN.search(message)
|
|
if cvar_match:
|
|
cvar_name = cvar_match.group(1)
|
|
value = cvar_match.group(2)
|
|
|
|
if game_state.server_info.update_from_cvar(cvar_name, value):
|
|
ui.update_server_info(game_state)
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def handle_stats_connection(message):
|
|
"""
|
|
Handle stats connection info extraction
|
|
Returns (stats_port, stats_password) or (None, None)
|
|
"""
|
|
stats_port = None
|
|
stats_password = None
|
|
|
|
# Extract stats port
|
|
if 'net_port' in message and ' is:' in message and '"net_port"' in message:
|
|
match = CVAR_RESPONSE_PATTERN.search(message)
|
|
if match:
|
|
port_str = match.group(2).strip()
|
|
digit_match = PORT_PATTERN.search(port_str)
|
|
if digit_match:
|
|
stats_port = digit_match.group(1)
|
|
logger.info(f'Got stats port: {stats_port}')
|
|
|
|
# Extract stats password
|
|
if 'zmq_stats_password' in message and ' is:' in message and '"zmq_stats_password"' in message:
|
|
match = CVAR_RESPONSE_PATTERN.search(message)
|
|
if match:
|
|
password_str = match.group(2)
|
|
password_str = strip_color_codes(password_str)
|
|
stats_password = password_str.strip()
|
|
logger.info(f'Got stats password: {stats_password}')
|
|
|
|
return stats_port, stats_password
|
|
|
|
def handle_user_input(ui, rcon):
|
|
"""Read pending keys and send the commands the user submitted"""
|
|
for command in ui.process_input():
|
|
logger.info(f'Sending command: {repr(command.strip())}')
|
|
|
|
# Display command with timestamp
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^5[^7{timestamp}^5] >>> {command.strip()}^7\n")
|
|
|
|
rcon.send_command(command)
|
|
|
|
|
|
def handle_stats_stream(stats_conn, stats_check_counter, event_parser, ui, game_state):
|
|
"""Poll and process stats stream events"""
|
|
if not stats_conn or not stats_conn.connected:
|
|
return stats_check_counter
|
|
|
|
stats_check_counter += 1
|
|
|
|
if stats_check_counter % 100 == 0:
|
|
logger.debug(f'Stats polling active (check #{stats_check_counter // 100})')
|
|
|
|
stats_msg = stats_conn.recv_message()
|
|
if stats_msg:
|
|
logger.info(f'Stats event received ({len(stats_msg)} bytes)')
|
|
|
|
# Parse game event
|
|
parsed = event_parser.parse_event(stats_msg)
|
|
if parsed:
|
|
# Format with timestamp before displaying
|
|
formatted_msg, attributes = format_message(parsed)
|
|
ui.print_message(formatted_msg)
|
|
ui.update_server_info(game_state)
|
|
|
|
return stats_check_counter
|
|
|
|
|
|
def handle_player_respawns(game_state, ui):
|
|
"""Check and revive dead players after respawn delay"""
|
|
if game_state.server_info.gametype == 'Clan Arena':
|
|
# CA: revive all players after round end
|
|
if game_state.server_info.round_end_time:
|
|
if time.time() - game_state.server_info.round_end_time >= _RESPAWN_DELAY:
|
|
game_state.server_info.dead_players.clear()
|
|
game_state.server_info.round_end_time = None
|
|
ui.update_server_info(game_state)
|
|
else:
|
|
# Other modes: revive individual players after death
|
|
current_time = time.time()
|
|
players_to_revive = [
|
|
name for name, death_time in game_state.server_info.dead_players.items()
|
|
if current_time - death_time >= _RESPAWN_DELAY
|
|
]
|
|
if players_to_revive:
|
|
for name in players_to_revive:
|
|
del game_state.server_info.dead_players[name]
|
|
ui.update_server_info(game_state)
|
|
|
|
|
|
def parse_player_events(message, game_state, ui):
|
|
"""
|
|
Parse connect, disconnect, kick, and rename messages
|
|
Returns True if message should be suppressed
|
|
"""
|
|
try:
|
|
msg = message
|
|
|
|
# Strip broadcast: print "..." wrapper with regex
|
|
broadcast_match = BROADCAST_PATTERN.match(msg)
|
|
if broadcast_match:
|
|
msg = broadcast_match.group(1)
|
|
|
|
# Strip timestamp: [HH:MM:SS] or ^3[^7HH:MM:SS^3]^7
|
|
msg = TIMESTAMP_PATTERN.sub('', msg)
|
|
msg = msg.strip()
|
|
|
|
if not msg:
|
|
return False
|
|
|
|
logger.debug(f'parse_player_events: {repr(msg)}')
|
|
|
|
# Strip color codes for matching
|
|
clean_msg = strip_color_codes(msg)
|
|
|
|
# Match connects: "NAME connected" or "NAME connected with Steam ID"
|
|
connect_match = CONNECT_PATTERN.match(clean_msg)
|
|
if connect_match:
|
|
player_name_match = CONNECT_PATTERN.match(msg)
|
|
player_name = player_name_match.group(1).strip() if player_name_match else connect_match.group(1).strip()
|
|
player_name = strip_color_codes(player_name)
|
|
|
|
logger.info(f'CONNECT: {repr(player_name)}')
|
|
game_state.player_tracker.update_team(player_name, 'SPECTATOR')
|
|
game_state.player_tracker.add_player(player_name)
|
|
ui.update_server_info(game_state)
|
|
|
|
# Only print if this is NOT the Steam ID line
|
|
if 'Steam ID' not in message:
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3]^7 ^8{player_name} ^9^2connected\n")
|
|
|
|
return True
|
|
|
|
# Regular disconnect
|
|
disconnect_match = DISCONNECT_PATTERN.match(clean_msg)
|
|
if disconnect_match:
|
|
original_match = DISCONNECT_PATTERN.match(msg)
|
|
player_name = original_match.group(1).strip() if original_match else disconnect_match.group(1).strip()
|
|
player_name = strip_color_codes(player_name)
|
|
|
|
logger.info(f'DISCONNECT: {repr(player_name)}')
|
|
game_state.player_tracker.remove_player(player_name)
|
|
ui.update_server_info(game_state)
|
|
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3]^7 ^8{player_name} ^9^1disconnected\n")
|
|
return True
|
|
|
|
# Kick
|
|
kick_match = KICK_PATTERN.match(clean_msg)
|
|
if kick_match:
|
|
original_match = KICK_PATTERN.match(msg)
|
|
player_name = original_match.group(1).strip() if original_match else kick_match.group(1).strip()
|
|
player_name = strip_color_codes(player_name)
|
|
|
|
logger.info(f'KICK: {repr(player_name)}')
|
|
game_state.player_tracker.remove_player(player_name)
|
|
ui.update_server_info(game_state)
|
|
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3]^7 ^8{player_name} ^1was kicked\n")
|
|
return True
|
|
|
|
# Inactivity
|
|
inactivity_match = INACTIVITY_PATTERN.match(clean_msg)
|
|
if inactivity_match:
|
|
original_match = INACTIVITY_PATTERN.match(msg)
|
|
player_name = original_match.group(1).strip() if original_match else inactivity_match.group(1).strip()
|
|
player_name = strip_color_codes(player_name)
|
|
|
|
logger.info(f'INACTIVITY DROP: {repr(player_name)}')
|
|
game_state.player_tracker.remove_player(player_name)
|
|
ui.update_server_info(game_state)
|
|
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3]^7 ^8{player_name}^0 ^3dropped due to inactivity^7\n")
|
|
return True
|
|
|
|
# Match renames: "OldName renamed to NewName"
|
|
rename_match = RENAME_PATTERN.match(clean_msg)
|
|
if rename_match:
|
|
# Extract from original message
|
|
original_match = RENAME_PATTERN.match(msg)
|
|
if original_match:
|
|
old_name = original_match.group(1).strip()
|
|
new_name = original_match.group(2).strip()
|
|
else:
|
|
old_name = rename_match.group(1).strip()
|
|
new_name = rename_match.group(2).strip()
|
|
|
|
# Remove color codes from both names
|
|
old_name = strip_color_codes(old_name)
|
|
new_name = strip_color_codes(new_name)
|
|
old_name = old_name.rstrip('\n\r') # Remove trailing newline
|
|
new_name = new_name.rstrip('\n\r') # Remove trailing newline
|
|
|
|
logger.info(f'RENAME: {repr(old_name)} -> {repr(new_name)}')
|
|
game_state.player_tracker.rename_player(old_name, new_name)
|
|
ui.update_server_info(game_state)
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3]^7 ^8{old_name} ^6renamed to^7 ^8{new_name}\n")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f'Error in parse_player_events: {e}')
|
|
|
|
return False
|
|
|
|
def main_loop(screen, args):
|
|
"""
|
|
Main application loop.
|
|
Runs single-threaded: ZMQ sockets and the terminal (stdin) are polled together,
|
|
so all curses drawing happens here and never from another thread.
|
|
"""
|
|
|
|
# Setup signal handler for Ctrl+C with confirmation
|
|
signal.signal(signal.SIGINT, signal_handler)
|
|
|
|
# Set logging level
|
|
if args.verbose == 0:
|
|
logger.setLevel(args.config_log_level if args.config_log_level is not None else logging.WARNING)
|
|
elif args.verbose == 1:
|
|
logger.setLevel(logging.INFO)
|
|
else:
|
|
logger.setLevel(logging.DEBUG)
|
|
|
|
# Setup file logging for unknown events
|
|
unknown_handler = logging.FileHandler(args.unknown_log, mode='a')
|
|
unknown_formatter = logging.Formatter('%(asctime)s - %(message)s', '%Y-%m-%d %H:%M:%S')
|
|
unknown_handler.setFormatter(unknown_formatter)
|
|
unknown_json_logger.addHandler(unknown_handler)
|
|
unknown_json_logger.propagate = False
|
|
|
|
# Initialize components
|
|
game_state = GameState()
|
|
ui = UIManager(screen, args.title, args.max_history,
|
|
player_names_provider=lambda: game_state.player_tracker.get_player_names())
|
|
|
|
# Setup logging to output window
|
|
log_handler = ui.setup_logging()
|
|
logger.addHandler(log_handler)
|
|
|
|
# Attach UI log handler to lib module loggers
|
|
for lib_name in ('network', 'parser', 'state'):
|
|
lib_logger = logging.getLogger(lib_name)
|
|
if log_handler not in lib_logger.handlers:
|
|
lib_logger.addHandler(log_handler)
|
|
lib_logger.setLevel(logger.level)
|
|
|
|
# Display startup messages
|
|
ui.print_message(f"*** QL pyCon Version {VERSION} starting ***\n")
|
|
ui.print_message(f"zmq python bindings {zmq.__version__}, libzmq version {zmq.zmq_version()}\n")
|
|
|
|
# Initialize network connections
|
|
ui.print_message(f"Connecting with host={args.host} password={args.password}\n")
|
|
rcon = RconConnection(args.host, args.password, args.identity)
|
|
rcon.connect()
|
|
|
|
# One poller wakes the loop on RCON data, connection events or a key press
|
|
poller = zmq.Poller()
|
|
poller.register(rcon.socket, zmq.POLLIN)
|
|
poller.register(rcon.monitor, zmq.POLLIN)
|
|
poller.register(sys.stdin, zmq.POLLIN)
|
|
|
|
stats_conn = None
|
|
stats_port = None
|
|
stats_password = None
|
|
stats_check_counter = 0
|
|
|
|
# Timer refresh tracking (update UI once per second)
|
|
last_ui_update = 0
|
|
|
|
# Setup JSON logging if requested
|
|
json_logger = None
|
|
if args.json_log:
|
|
json_handler = logging.FileHandler(args.json_log, mode='a')
|
|
json_formatter = logging.Formatter('%(asctime)s - %(message)s', '%Y-%m-%d %H:%M:%S')
|
|
json_handler.setFormatter(json_formatter)
|
|
all_json_logger.addHandler(json_handler)
|
|
all_json_logger.propagate = False
|
|
json_logger = all_json_logger
|
|
|
|
# Create event parser
|
|
event_parser = EventParser(game_state, json_logger, unknown_json_logger)
|
|
|
|
# Main event loop with resource cleanup
|
|
try:
|
|
while not shutdown_requested:
|
|
# Wait for RCON data, a connection event, a stats event or a key press
|
|
ready = dict(poller.poll(POLL_TIMEOUT))
|
|
|
|
handle_ctrl_c()
|
|
|
|
# Check monitor for connection events
|
|
monitor_event = rcon.check_monitor()
|
|
if monitor_event and monitor_event[0] == zmq.EVENT_CONNECTED:
|
|
ui.print_message("Connected to server\n")
|
|
rcon.send_command(b'register')
|
|
logger.info('Registration message sent')
|
|
|
|
ui.print_message("Requesting connection info...\n")
|
|
rcon.send_command(b'zmq_stats_password')
|
|
rcon.send_command(b'net_port')
|
|
elif monitor_event and monitor_event[0] == zmq.EVENT_DISCONNECTED:
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3] ^8^1Disconnected from server - waiting for reconnect...^0^7\n")
|
|
if stats_conn is not None:
|
|
poller.unregister(stats_conn.socket)
|
|
stats_conn.close()
|
|
stats_conn = None
|
|
stats_port = None
|
|
stats_password = None
|
|
|
|
# Handle user input
|
|
handle_user_input(ui, rcon)
|
|
|
|
# Poll stats stream if connected
|
|
stats_check_counter = handle_stats_stream(stats_conn, stats_check_counter, event_parser, ui, game_state)
|
|
|
|
# Check if we need to revive players
|
|
handle_player_respawns(game_state, ui)
|
|
|
|
# Process RCON messages
|
|
if rcon.socket in ready:
|
|
logger.debug('Socket has data available')
|
|
msg_count = 0
|
|
|
|
while True:
|
|
message = rcon.recv_message()
|
|
if message is None:
|
|
if msg_count > 0:
|
|
logger.debug(f'Read {msg_count} message(s)')
|
|
break
|
|
|
|
msg_count += 1
|
|
|
|
if len(message) == 0:
|
|
logger.debug('Received empty message (keepalive)')
|
|
continue
|
|
|
|
logger.debug(f'Received message ({len(message)} bytes): {repr(message[:100])}')
|
|
|
|
# Check for player connect/disconnect/rename events
|
|
if parse_player_events(message, game_state, ui):
|
|
continue
|
|
|
|
if 'Game Initialization' in message:
|
|
logger.info('Game initialization detected - refreshing server info')
|
|
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^3[^7{timestamp}^3] ^8^3Game initialized - Refreshing server info^0^7\n")
|
|
|
|
rcon.send_command(b'qlx_serverBrandName')
|
|
rcon.send_command(b'g_factoryTitle')
|
|
rcon.send_command(b'mapname')
|
|
rcon.send_command(b'timelimit')
|
|
rcon.send_command(b'fraglimit')
|
|
rcon.send_command(b'roundlimit')
|
|
rcon.send_command(b'capturelimit')
|
|
rcon.send_command(b'sv_maxclients')
|
|
|
|
# Clear player dict since map changed
|
|
game_state.server_info.players = {}
|
|
game_state.player_tracker.player_teams = {}
|
|
game_state.server_info.reset_round_scores()
|
|
game_state.server_info.dead_players.clear()
|
|
ui.update_server_info(game_state)
|
|
|
|
# Try to parse as cvar response
|
|
if parse_cvar_response(message, game_state, ui):
|
|
logger.debug('Suppressed cvar response')
|
|
continue
|
|
|
|
# Check for stats connection info
|
|
port, password = handle_stats_connection(message)
|
|
if port:
|
|
stats_port = port
|
|
if password:
|
|
stats_password = password
|
|
|
|
# Connect to stats if we have both credentials
|
|
if stats_port and stats_password and stats_conn is None:
|
|
try:
|
|
ui.print_message("Connecting to stats stream...\n")
|
|
host_ip = args.host.split('//')[1].split(':')[0]
|
|
|
|
stats_conn = StatsConnection(host_ip, stats_port, stats_password)
|
|
stats_conn.connect()
|
|
poller.register(stats_conn.socket, zmq.POLLIN)
|
|
|
|
ui.print_message("Stats stream connected - ready for game events\n")
|
|
|
|
# Request initial server info
|
|
logger.info('Sending initial server info queries')
|
|
rcon.send_command(b'qlx_serverBrandName')
|
|
rcon.send_command(b'g_factoryTitle')
|
|
rcon.send_command(b'mapname')
|
|
rcon.send_command(b'timelimit')
|
|
rcon.send_command(b'fraglimit')
|
|
rcon.send_command(b'roundlimit')
|
|
rcon.send_command(b'capturelimit')
|
|
rcon.send_command(b'sv_maxclients')
|
|
|
|
if args.json_log:
|
|
ui.print_message(f"*** JSON capture enabled: {args.json_log} ***\n")
|
|
|
|
except Exception as e:
|
|
timestamp = time.strftime('%H:%M:%S')
|
|
ui.print_message(f"^1[^7{timestamp}^1] Error: Stats connection failed: {e}^7\n")
|
|
logger.error(f'Stats connection failed: {e}')
|
|
|
|
# Try to parse as game event
|
|
parsed_event = event_parser.parse_event(message)
|
|
if parsed_event:
|
|
ui.print_message(parsed_event)
|
|
continue
|
|
|
|
# Check if it looks like JSON but wasn't parsed
|
|
stripped = message.strip()
|
|
if stripped and stripped[0] in ('{', '['):
|
|
logger.debug('Unparsed JSON event')
|
|
continue
|
|
|
|
# Try powerup message formatting
|
|
powerup_msg = format_powerup_message(message, game_state.player_tracker)
|
|
if powerup_msg:
|
|
ui.print_message(powerup_msg)
|
|
continue
|
|
|
|
# Filter bot debug messages in default mode
|
|
is_bot_debug = (' entered ' in message and
|
|
any(x in message for x in [' seek ', ' battle ', ' chase', ' fight']))
|
|
if is_bot_debug and args.verbose == 0:
|
|
logger.debug(f'Filtered bot debug: {message[:50]}')
|
|
continue
|
|
|
|
# Check if it's a chat message
|
|
if ':' in message and not message.startswith(('print', 'broadcast', 'zmq')):
|
|
message = format_chat_message(message, game_state.player_tracker)
|
|
|
|
# Format and display message
|
|
formatted_msg, attributes = format_message(message)
|
|
ui.print_message(formatted_msg)
|
|
|
|
# Update server info panel every second (for live timer display)
|
|
current_time = time.time()
|
|
if current_time - last_ui_update >= 1.0:
|
|
ui.update_server_info(game_state)
|
|
last_ui_update = current_time
|
|
|
|
finally:
|
|
# Clean up resources
|
|
logger.info("Shutting down...")
|
|
if rcon:
|
|
logger.debug("Closing RCON connection")
|
|
rcon.close()
|
|
if stats_conn:
|
|
logger.debug("Closing stats connection")
|
|
stats_conn.close()
|
|
logger.info("Shutdown complete")
|
|
|
|
if __name__ == '__main__':
|
|
# Load config
|
|
config = ConfigLoader()
|
|
config.load()
|
|
|
|
# Read settings from config ([logging]/[ui]/[behavior])
|
|
_QUIT_CONFIRM_TIMEOUT = config.get_quit_timeout(QUIT_CONFIRM_TIMEOUT)
|
|
_RESPAWN_DELAY = config.get_respawn_delay(RESPAWN_DELAY)
|
|
|
|
# Parse arguments
|
|
parser = argparse.ArgumentParser(description='Quake Live Python Console')
|
|
parser.add_argument('server', nargs='?', default=None,
|
|
help='Named server from qlpycon.conf (e.g. ffa, duel)')
|
|
parser.add_argument('--host', default=None,
|
|
help='ZMQ URI to connect to (e.g. tcp://1.2.3.4:28960)')
|
|
parser.add_argument('--password', default=None,
|
|
help='RCON password')
|
|
parser.add_argument('--list', action='store_true',
|
|
help='List configured servers and exit')
|
|
parser.add_argument('--identity', default=uuid.uuid1().hex,
|
|
help='Socket identity (random UUID by default)')
|
|
parser.add_argument('-v', '--verbose', action='count', default=0,
|
|
help='Increase verbosity (-v INFO, -vv DEBUG)')
|
|
parser.add_argument('--unknown-log', default='unknown_events.log',
|
|
help='File to log unknown JSON events')
|
|
parser.add_argument('-j', '--json', dest='json_log', default=None,
|
|
help='File to log all JSON events')
|
|
args = parser.parse_args()
|
|
|
|
# Apply config settings to args
|
|
args.config_log_level = config.get_log_level()
|
|
args.max_history = config.get_max_history(MAX_COMMAND_HISTORY)
|
|
|
|
# Handle --list
|
|
if args.list:
|
|
servers = config.get_servers()
|
|
if not servers:
|
|
print('No servers configured in qlpycon.conf')
|
|
else:
|
|
print('Configured servers:')
|
|
for name, host in servers.items():
|
|
print(f' {name:<12} {host}')
|
|
sys.exit(0)
|
|
|
|
# No server name and no --host: let the user pick from the configured servers
|
|
if args.server is None and args.host is None:
|
|
servers = config.get_servers()
|
|
if servers:
|
|
args.server = curses.wrapper(select_server, servers)
|
|
if args.server is None:
|
|
sys.exit(0)
|
|
|
|
# Resolve host and password
|
|
if args.server:
|
|
host, password = config.get_server(args.server)
|
|
if host is None:
|
|
print(f"Error: server '{args.server}' not found in qlpycon.conf")
|
|
print("Use 'qlpycon --list' to see configured servers.")
|
|
sys.exit(1)
|
|
args.host = args.host or host
|
|
args.password = args.password or password
|
|
args.title = f'Quake Live PyCon: {args.server} ({args.host})'
|
|
else:
|
|
args.host = args.host or config.get_host() or DEFAULT_HOST
|
|
if not args.host.startswith('tcp://'):
|
|
args.host = f'tcp://{args.host}'
|
|
args.password = args.password or config.get_password()
|
|
args.title = f'Quake Live PyCon: {args.host}'
|
|
|
|
curses.wrapper(main_loop, args)
|