xbl/qlpycon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
c9487b954a81d09c8a06680d9d4a59e58995d62e
qlpycon/main.py
T

913 lines
37 KiB
Python
Raw Blame History

#!/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, STATUS_OUTPUT_WINDOW, STATUS_SEED_WINDOW
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
from lib.namelist import NameCapture
# 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+(.+?)$')
STATUS_HEAD_PATTERN = re.compile(r'^\s*\d+\s+-?\d+\s+(bot|\d+)\s+\S', re.MULTILINE)
STATUS_MAP_PATTERN = re.compile(r'^map:\s+\S+\s*$', re.MULTILINE)
STEAMID_PATTERN = re.compile(r'\b\d{16,}\b')
LIVESTATS_PATTERN = re.compile(r'^LIVESTATS\s+(.+)\s+(-?\d+)\s+(-?\d+)\s+(-?\d+)\s+(-?\d+)\s*$')
STATUS_ROW_PATTERN = re.compile(
r'^\s*(\d+)\s+(-?\d+)\s+(\d+)\s+(.+?)\s+(\d+)\s+(\S+)\s+(\d+)\s+(\d+)(?:\s+(\d{15,}))?\s*$')
ROSTER_CHAT_GUARD_PATTERN = re.compile(r'^.{1,40}:\s')
VOTE_CAST_PATTERN = re.compile(r'^(.+?) voted for (.+)\.$')
CHAT_SHAPE_PATTERN = re.compile(r'^[\w(][^:]*:\s')
GAME_END_PATTERN = re.compile(
r'((?:timelimit|fraglimit|capturelimit|roundlimit)\s+hit'
r'|hit\s+the\s+(?:timelimit|fraglimit|capturelimit|roundlimit)'
r'|game\s+has\s+been\s+forfeited'
r'|wins\s+the\s+round)', re.IGNORECASE)
ENGINE_OUTPUT_PREFIXES = (
'warning:', 'server:', 'error:', 'fatal:', 'livestats',
'gamename:', 'gamedate:', 'protocol:', 'cheats:', 'sv_tags:', 'client ',
)
# 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
"""
# 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 is_command_echo(message):
"""True for the server's echo lines announcing that an RCON client sent a command"""
return message.startswith('zmq RCON command')
def is_command_output(message):
"""
Detect console output produced by RCON commands: cvar dumps and status
tables. These are hidden; the 'zmq RCON command' echo lines still show
which client ran what. Shape checks run on color-stripped text because
the server sprinkles Quake color codes into status rows. Rows also arrive
broadcast-wrapped or split across several messages, so the checks are
containment-based and a row fragment (head without steamid, tail without
row prefix) is caught too. JSON stats events are exempt: they contain
16+ digit STEAM_IDs and are handled by the event parser.
"""
clean = strip_color_codes(message)
stripped = clean.lstrip()
if stripped[:1] in ('{', '['):
return False
if CVAR_RESPONSE_PATTERN.search(clean):
return True
if 'num score ping' in clean or '--- -----' in clean:
return True
if STEAMID_PATTERN.search(clean) or STATUS_HEAD_PATTERN.search(clean):
return True
if STATUS_MAP_PATTERN.search(clean):
return True
return False
def is_engine_output(message):
"""Engine/banner prints share the chat shape but are never chat
(they would surface as fake chat from players named WARNING or Server)"""
clean = strip_color_codes(message).lstrip()
return clean.lower().startswith(ENGINE_OUTPUT_PREFIXES)
def handle_livestats(message, game_state, ui, verbose=0):
"""
Consume every LIVESTATS heartbeat line (engine damage counters emitted
every 10s by the livedamage plugin, plus its summary variants). Lines
matching the full player shape update live score/damage/ping; names may
contain spaces, so the counters anchor from the end. The raw lines
display only at -v or higher verbosity; they are never treated as chat.
"""
clean = strip_color_codes(message).strip()
if not clean.lower().startswith('livestats'):
return False
match = LIVESTATS_PATTERN.match(clean)
if match:
name, score, dealt, taken, ping = match.groups()
game_state.player_tracker.update_livestats(name, int(score), int(dealt), int(taken), int(ping))
ui.update_server_info(game_state)
if verbose > 0:
timestamp = time.strftime('%H:%M:%S')
ui.print_message(f"^3[^7{timestamp}^3]^7 {message.rstrip()}\n")
return True
def handle_vote_broadcast(message, game_state, ui):
"""
Track vote lifecycle broadcasts and display them verbatim (color codes
intact), annotating the outcome with the cast subject when every cast
agreed. Returns True when the message is a vote line and has been
handled; chat-shaped lines are never vote events.
"""
clean = strip_color_codes(message).strip()
if not clean or ROSTER_CHAT_GUARD_PATTERN.match(clean):
return False
tracker = game_state.vote_tracker
timestamp = time.strftime('%H:%M:%S')
if clean.endswith('called a vote.'):
tracker.on_call()
ui.print_message(f"^3[^7{timestamp}^3]^7 {message.rstrip()}\n")
return True
cast = VOTE_CAST_PATTERN.match(clean)
if cast:
tracker.on_cast(cast.group(2))
ui.print_message(f"^3[^7{timestamp}^3]^7 {message.rstrip()}\n")
return True
if clean in ('Vote passed.', 'Vote failed.'):
line = message.rstrip()
subject = tracker.result_annotation()
if subject:
line = f"{line} — {subject}"
ui.print_message(f"^3[^7{timestamp}^3]^7 {line}\n")
tracker.on_call()
return True
return False
def is_real_content(message):
"""
True for messages that are definitely not status table debris: JSON,
chat-shaped lines and game initialization. Used to end the status
suppression window early; everything else inside the window is treated
as table debris, because the server splits rows into arbitrary chunks
(names, flag columns, '0 16384', bare steamids) that no shape check
can distinguish from real content. The status table's own 'map:' line
is chat-shaped and explicitly excluded, or it would end the window
on the first chunk.
"""
clean = strip_color_codes(message).strip()
if not clean:
return False
if clean[:1] in ('{', '['):
return True
if clean.startswith('zmq'):
return False
if clean.startswith('map:'):
return False
if clean.startswith('broadcast:'):
clean = clean[10:].lstrip()
if clean.startswith('print "'):
clean = clean[7:]
if clean.endswith('"'):
clean = clean[:-1]
if 'Game Initialization' in clean:
return True
if GAME_END_PATTERN.search(clean):
return True
if CHAT_SHAPE_PATTERN.match(clean):
return True
return False
class StatusSeeder:
"""
Seeds the player roster from a 'status' command dump, sent on connect and
after map changes. The table arrives chunked across many messages (rows
split at arbitrary column boundaries, wrapped or raw), so raw text is
buffered and split on newlines; the tail is flushed when the window
closes. Rows carry no team column, so untracked names seed as SPECTATOR
(immediately visible; active players self-correct on their first stats
event) and known players only get their authoritative score/ping.
All seeding output stays suppressed.
"""
def __init__(self, game_state):
self.game_state = game_state
self.buffer = ''
self.active_until = 0.0
def start(self):
self.buffer = ''
self.active_until = time.time() + STATUS_SEED_WINDOW
def is_active(self):
if self.active_until and time.time() >= self.active_until:
self.finish()
return bool(self.active_until) and time.time() < self.active_until
def feed(self, message):
"""Buffer one RCON message and parse every completed line"""
text = strip_color_codes(message)
if text.startswith('broadcast:'):
text = text[10:].lstrip()
if text.startswith('print "'):
text = text[7:]
if text.endswith('"'):
text = text[:-1]
self.buffer += text.replace('\\n', '\n')
while '\n' in self.buffer:
line, self.buffer = self.buffer.split('\n', 1)
self._parse_line(line)
def finish(self):
if self.buffer.strip():
self._parse_line(self.buffer)
self.buffer = ''
def _parse_line(self, line):
match = STATUS_ROW_PATTERN.match(line.strip())
if not match:
return
num, score, ping, name, lastmsg, address, qport, rate, steamid = match.groups()
tracker = self.game_state.player_tracker
clean_name = strip_color_codes(name).strip()
if not clean_name:
return
player_data = self.game_state.server_info.players.get(clean_name)
if player_data is not None:
player_data['score'] = str(score)
player_data['ping'] = str(ping)
return
tracker.add_player(clean_name, score=str(score), ping=str(ping))
if tracker.get_team(clean_name) is None:
tracker.update_team(clean_name, 'SPECTATOR')
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('Got 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 (parse_event returns a display-ready string)
parsed = event_parser.parse_event(stats_msg)
if parsed:
ui.print_message(parsed)
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)
# Roster events are only roster events when the line is not chat-shaped:
# "Foo: I just connected my router" must not spawn a phantom join for
# a player named "Foo: I just"
if ROSTER_CHAT_GUARD_PATTERN.match(clean_msg):
return False
# 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()
name_capture = NameCapture() # cvar/command names learned from the server for autocomplete
status_seeder = StatusSeeder(game_state) # seeds the roster from a status dump on connect
ui = UIManager(screen, args.title, args.max_history,
player_names_provider=lambda: game_state.player_tracker.get_player_names(),
learned_names_provider=lambda: name_capture.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', 'namelist'):
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
password_display = '***' if args.password else '(none)'
timestamp = time.strftime('%H:%M:%S')
ui.print_message(f"^3[^7{timestamp}^3]^7 Connecting with host={args.host} password={password_display}\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
status_suppress_until = 0.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:
timestamp = time.strftime('%H:%M:%S')
ui.print_message(f"^3[^7{timestamp}^3]^7 Connected to server ({monitor_event[2]}) "
f"as rcon client {rcon.identity}\n")
rcon.send_command(b'register')
logger.info('Registration message sent')
timestamp = time.strftime('%H:%M:%S')
ui.print_message(f"^3[^7{timestamp}^3]^7 Requesting connection info...\n")
rcon.send_command(b'zmq_stats_password')
rcon.send_command(b'net_port')
rcon.send_command(b'status')
status_seeder.start()
if args.learn_names:
name_capture.request(rcon)
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 ({monitor_event[2]}) "
f"as rcon client {rcon.identity} - 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
# LIVESTATS prints feed live score/damage/ping; the raw
# line only displays at -v or higher verbosity
if handle_livestats(message, game_state, ui, args.verbose):
continue
# Vote lifecycle broadcasts are tracked and shown verbatim
if handle_vote_broadcast(message, game_state, ui):
continue
# Seed the roster from the status dump after connect/map change;
# table chunks stay suppressed, real content flows normally
if status_seeder.is_active():
if not is_real_content(message):
status_seeder.feed(message)
continue
# Command echoes only show at -v or higher verbosity; a
# status echo opens a window that hides its chunked output
if is_command_echo(message):
if message.rstrip().endswith(': status'):
status_suppress_until = time.time() + STATUS_OUTPUT_WINDOW
if args.verbose == 0:
logger.debug('Suppressed RCON command echo')
continue
# Suppress everything after a 'status' echo until real
# content arrives (chat, JSON, game init) or the window closes
if status_suppress_until:
if time.time() >= status_suppress_until:
status_suppress_until = 0.0
elif is_real_content(message):
status_suppress_until = 0.0
else:
logger.debug('Suppressed status output chunk')
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()
rcon.send_command(b'status')
status_seeder.start()
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:
timestamp = time.strftime('%H:%M:%S')
ui.print_message(f"^3[^7{timestamp}^3]^7 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)
timestamp = time.strftime('%H:%M:%S')
ui.print_message(f"^3[^7{timestamp}^3]^7 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}')
# Hide output of RCON commands (cvar dumps, status tables)
if is_command_output(message):
logger.debug('Suppressed RCON command output')
continue
# Hide the cvarlist/cmdlist output while learning names from it.
# Runs after the cvar handling above so the stats credentials and
# server info responses are never swallowed.
if name_capture.handle(message):
continue
# 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; engine output shares the shape
if (':' in message and not message.startswith(('print', 'broadcast', 'zmq'))
and not is_engine_output(message)):
message = format_chat_message(message, game_state.player_tracker)
# Format and display message
formatted_msg, _ = 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)
args.learn_names = config.get_bool('behavior', 'learn_names', False)
# 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)
Reference in New Issue View Git Blame Copy Permalink