xbl/qlpycon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

serverwindow

Browse Source
This commit is contained in:
xbl
2025-12-22 01:27:33 +01:00
parent 499525b714
commit 1dcd8941b5
1 changed files with 224 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

257
qlpycon.py
View File

@ -1,5 +1,5 @@
#!/usr/bin/env python3
# Version: 0.6.9
# Version: 0.7.0
import sys
import re
@ -33,13 +33,35 @@ import unittest
import locale
locale.setlocale(locale.LC_ALL, '')
# Server info state
server_info = {
'hostname': 'Unknown',
'map': 'Unknown',
'gametype': 'Unknown',
'timelimit': '0',
'fraglimit': '0',
'roundlimit': '0',
'capturelimit': '0',
'players': [],
'red_score': 0,
'blue_score': 0,
'last_update': 0
}
# Message buffering for status output
status_buffer = []
in_status_output = False
# Pending background commands
pending_background_commands = set()
current_gametype = None
# Track recent events to avoid duplicates
recent_events = []
MAX_RECENT_EVENTS = 10
team_modes = ["TDM", "CA", "CTF", "ONEFLAG", "OVERLOAD", "HARVESTER", "FT" ]
team_modes = ["Team Deathmatch", "Clan Arena", "Capture The Flag", "One Flag CTF", "Overload", "Harvester", "Freeze Tag" ]
player_teams = {}
@ -311,6 +333,153 @@ def format_powerup_message(msg_str):
# Not a powerup message
return None
def parse_server_response(msg_str, info_window):
"""Parse server responses to update server_info, returns True if message should be suppressed"""
global server_info, pending_background_commands, status_buffer, in_status_output, current_gametype
# Suppress only the command echo lines
if msg_str.startswith('zmq RCON command') and 'from' in msg_str:
if any(f': {cmd}' in msg_str for cmd in ['status', 'roundlimit', 'qlx_serverBrandName', 'g_factoryTitle', 'mapname', 'timelimit', 'fraglimit', 'roundlimit', 'capturelimit', 'sv_maxclients']):
return True
# Parse qlx_serverBrandName
if '"qlx_serverbrandname"' in msg_str and ' is:' in msg_str:
match = re.search(r' is:"(.+?)" default:', msg_str)
if match:
brandname = match.group(1).strip()
server_info['hostname'] = brandname
logger.info(f'Got hostname: {server_info["hostname"]}')
UpdateServerInfoWindow(info_window)
# Parse g_factoryTitle
if '"g_factoryTitle"' in msg_str and ' is:' in msg_str:
match = re.search(r' is:"([^"]*)"', msg_str) # Changed [^"]+ to [^"]*
if match:
gametype = match.group(1).strip()
current_gametype = re.sub(r'\^\d', '', gametype) # Strip color codes
server_info['gametype'] = current_gametype
logger.info(f'Got gametype: {current_gametype}')
UpdateServerInfoWindow(info_window)
# Parse Map
if '"mapname"' in msg_str and ' is:' in msg_str:
match = re.search(r' is:"([^"]*)"', msg_str)
if match:
mapname = match.group(1).strip()
server_info['map'] = re.sub(r'\^\d', '', mapname)
logger.info(f'Got mapname: {server_info["map"]}')
UpdateServerInfoWindow(info_window)
# Parse timelimit
if '"timelimit"' in msg_str and ' is:' in msg_str:
match = re.search(r' is:"([^"]*)"', msg_str)
if match:
server_info['timelimit'] = match.group(1).strip()
logger.info(f'Got Timelimit: {server_info["timelimit"]}')
UpdateServerInfoWindow(info_window)
# Parse fraglimit
if '"fraglimit"' in msg_str and ' is:' in msg_str:
match = re.search(r' is:"([^"]*)"', msg_str)
if match:
server_info['fraglimit'] = match.group(1).strip()
logger.info(f'Got Fraglimit: {server_info["fraglimit"]}')
UpdateServerInfoWindow(info_window)
# Parse roundlimit
if '"roundlimit"' in msg_str and ' is:' in msg_str:
match = re.search(r' is:"([^"]*)"', msg_str)
if match:
server_info['roundlimit'] = match.group(1).strip()
logger.info(f'Got Roundlimit: {server_info["roundlimit"]}')
UpdateServerInfoWindow(info_window)
# Parse caplimit
if '"capturelimit"' in msg_str and ' is:' in msg_str:
match = re.search(r' is:"([^"]*)"', msg_str)
if match:
server_info['capturelimit'] = match.group(1).strip()
logger.info(f'Got Caplimit: {server_info["capturelimit"]}')
UpdateServerInfoWindow(info_window)
# Parse Max Clients
if '"sv_maxclients"' in msg_str and ' is:' in msg_str:
match = re.search(r' is:"([^"]*)"', msg_str)
if match:
server_info['maxclients'] = match.group(1).strip()
logger.info(f'Got Max Clients: {server_info["maxclients"]}')
UpdateServerInfoWindow(info_window)
# Handle status command output buffering
if 'status' in pending_background_commands:
# Buffer ALL short messages (likely parts of status output)
if len(msg_str) < 100: # Status fields are short
status_buffer.append(msg_str)
in_status_output = True
return True # Suppress
# When we get an empty message or long message, process the buffer
if msg_str.strip() == '' or len(msg_str) > 100:
if status_buffer:
# Reconstruct the line
full_line = ' '.join(status_buffer)
logger.info(f'Reconstructed status line: {full_line[:100]}')
# Parse if it's a player line
parts = full_line.split()
if len(parts) >= 8 and parts[0].isdigit():
server_info['players'].append({
'name': parts[3],
'score': parts[1],
'ping': parts[2]
})
status_buffer = []
if msg_str.strip() == '':
pending_background_commands.discard('status')
in_status_output = False
return True
return False
def UpdateServerInfoWindow(info_window):
"""Update the server info window with current data"""
info_window.clear()
max_y, max_x = info_window.getmaxyx()
# Line 1: Hostname with decorative border
hostname_display = server_info.get('hostname', 'Unknown Server')
PrintMessageColored(info_window, f"^6═══ {hostname_display} ^6═══^7\n", 0)
# Line 2: Type / Map / Limits / Players
gametype_display = server_info.get('gametype', 'Unknown Type')
mapname_display = server_info.get('map', 'Unknown Map')
timelimit_display = server_info.get('timelimit', '0')
fraglimit_display = server_info.get('fraglimit', '0')
roundlimit_display = server_info.get('roundlimit', '0')
caplimit_display = server_info.get('capturelimit', '0')
curclients_display = server_info.get('curclients', '0')
maxclients_display = server_info.get('maxclients', '0')
PrintMessageColored(info_window, f"^3Type:^7 {gametype_display} ^3Map:^7 {mapname_display} ^3Players:^7 {curclients_display}/{maxclients_display} ^3Limits (T/F/R/C):^7 {timelimit_display}/{fraglimit_display}/{roundlimit_display}/{caplimit_display}\n", 0)
# Line 4: Teams, if any
if current_gametype in team_modes:
PrintMessageColored(info_window, f"^1(RED) ^4(BLUE)\n", 0)
else:
PrintMessageColored(info_window, f"^3(FREE)\n", 0)
# Lines 5-10: Reserved for future use (empty for now)
info_window.addstr("\n\n\n\n\n\n")
# Line 11: Separator
separator = "^6" + "═" * (max_x - 1) + "^7"
PrintMessageColored(info_window, separator, 0)
info_window.refresh()
def ParseGameEvent(message):
"""Parse JSON game events and return formatted message"""
global current_gametype, recent_events
@ -395,6 +564,14 @@ def ParseGameEvent(message):
# Update victim team from event data
if 'TEAM' in victim:
update_player_team(victim_name, victim['TEAM'])
# ADD THIS: Update server_info players list
global server_info
if not any(p['name'] == victim_name for p in server_info['players']):
server_info['players'].append({
'name': victim_name,
'score': '0',
'ping': '0'
})
victim_team_prefix = get_team_color(victim_name)
@ -421,6 +598,12 @@ def ParseGameEvent(message):
# Update killer team from event data
if 'TEAM' in killer:
update_player_team(killer_name, killer['TEAM'])
if not any(p['name'] == killer_name for p in server_info['players']):
server_info['players'].append({
'name': killer_name,
'score': '0',
'ping': '0'
})
killer_team_prefix = get_team_color(killer_name)
@ -557,7 +740,7 @@ def InitWindows(screen, args):
curses.use_default_colors()
curses.cbreak()
curses.setsyx(-1, -1)
screen.addstr("Quake Live rcon: %s" % args.host)
screen.addstr("Quake Live PyCon: %s" % args.host)
screen.refresh()
maxy, maxx = screen.getmaxyx()
@ -567,8 +750,19 @@ def InitWindows(screen, args):
curses.init_pair(5, 6, 0)
curses.init_pair(6, 5, 0)
# Server info window at top
begin_x = 2; width = maxx - 4
begin_y = 2; height = maxy - 5
begin_y = 2; height = 10
info_window = curses.newwin(height, width, begin_y, begin_x)
screen.refresh()
info_window.scrollok(False)
info_window.idlok(False)
info_window.leaveok(True)
info_window.refresh()
# Output window (main chat/events)
begin_x = 2; width = maxx - 4
begin_y = 12; height = maxy - 15
output_window = curses.newwin(height, width, begin_y, begin_x)
screen.refresh()
output_window.scrollok(True)
@ -576,15 +770,17 @@ def InitWindows(screen, args):
output_window.leaveok(True)
output_window.refresh()
# Input window
begin_x = 4; width = maxx - 6
begin_y = maxy - 2; height = 1
input_window = curses.newwin(height, width, begin_y, begin_x)
screen.addstr(begin_y, begin_x - 2, '>')
screen.addstr(begin_y, begin_x - 2, '$ ')
screen.refresh()
input_window.idlok(True)
input_window.leaveok(False)
input_window.refresh()
# Divider window
begin_x = 2; width = maxx - 4
begin_y = maxy - 3; height = 1
divider_window = curses.newwin(height, width, begin_y, begin_x)
@ -599,7 +795,7 @@ def InitWindows(screen, args):
screen.refresh()
return input_window, output_window
return input_window, output_window, info_window
def main(screen):
global current_gametype
@ -631,9 +827,11 @@ def main(screen):
unknown_json_logger.addHandler(file_handler)
unknown_json_logger.propagate = False # Don't send to parent logger
input_window, output_window = InitWindows(screen, args)
input_window, output_window, info_window = InitWindows(screen, args)
PrintMessageFormatted(output_window, "*** QL pyCon Version 0.6.9 starting ***\n")
UpdateServerInfoWindow(info_window)
PrintMessageFormatted(output_window, "*** QL pyCon Version 0.7.0 starting ***\n")
PrintMessageFormatted(output_window, "zmq python bindings {}, libzmq version {}\n".format(repr(zmq.__version__), zmq.zmq_version()))
stats_port = None
@ -668,12 +866,10 @@ def main(screen):
socket.send( b'register' )
logger.info( 'Registration message sent.' )
PrintMessageFormatted(output_window, "Requesting game info...\n")
PrintMessageFormatted(output_window, "Requesting connection info...\n")
socket.send( b'zmq_stats_password' )
socket.send( b'net_port' )
logger.info('Requesting Gametype...')
socket.send(b'cvarlist g_gametype')
while ( not q.empty() ):
l = q.get()
logger.info( 'Sending command: %s' % repr( l.strip() ) )
@ -734,6 +930,10 @@ def main(screen):
y,x = curses.getsyx()
msg_str = msg.decode('utf-8', errors='replace')
is_background_response = parse_server_response(msg_str, info_window)
if is_background_response:
logger.debug('Suppressing background command response')
continue # Skip printing this message
if 'net_port' in msg_str and ' is:' in msg_str and '"net_port"' in msg_str:
match = re.search(r' is:"([^"]+)"', msg_str)
if match:
@ -786,6 +986,17 @@ def main(screen):
stats_connected = True
PrintMessageFormatted(output_window, "Stats stream connected - ready for game events\n")
# Send initial server info queries (once only)
logger.info('Sending initial server info queries')
socket.send(b'qlx_serverBrandName')
socket.send(b'g_factoryTitle')
socket.send(b'mapname')
socket.send(b'timelimit')
socket.send(b'fraglimit')
socket.send(b'roundlimit')
socket.send(b'capturelimit')
socket.send(b'sv_maxclients')
# Set up file handler for all JSON events if --json flag is provided
if args.json_log:
@ -801,26 +1012,6 @@ def main(screen):
logger.error('Stats connection failed: %s' % e)
import traceback
logger.debug(traceback.format_exc())
if 'g_gametype' in msg_str:
match = re.search(r'\bg_gametype\s+"(\d+)"', msg_str)
if match:
gametypes = {
"0": "FFA",
"1": "DUEL",
"2": "RACE",
"3": "TDM",
"4": "CA",
"5": "CTF",
"6": "ONEFLAG",
"7": "OVERLOAD",
"8": "HARVESTER",
"9": "FT"
}
gametypes_num = match.group(1).strip()
current_gametype = gametypes.get(gametypes_num, "UNKNOWN")
logger.info('Got gametype: %s' % current_gametype)
PrintMessageFormatted(output_window, f'Detected gametype: {current_gametype}\n')
# Try to parse as JSON game event
parsed_event = ParseGameEvent(msg_str)
@ -976,4 +1167,4 @@ def main(screen):
if ( __name__ == '__main__' ):
curses.wrapper(main)
# Version: 0.6.9
# Version: 0.7.0