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

modified: .gitignore

Browse Source 
modified:   README.md
	new file:   install.sh
	renamed:    config.py -> lib/constants.py
	renamed:    cvars.py -> lib/cvars.py
	renamed:    formatter.py -> lib/formatter.py
	renamed:    network.py -> lib/network.py
	renamed:    parser.py -> lib/parser.py
	renamed:    qlpycon_config.py -> lib/settings.py
	renamed:    state.py -> lib/state.py
	renamed:    ui.py -> lib/ui.py
	modified:   main.py
	modified:   qlpycon.bash
	new file:   qlpycon.conf.example
This commit is contained in:
xbl
2026-06-13 10:21:29 +02:00
parent be6ae8d137
commit 7ca9795a39
14 changed files with 476 additions and 174 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/constants.py
+115
View File
@@ -0,0 +1,115 @@
#!/usr/bin/env python3
"""
Configuration and constants for QLPyCon
"""
import re
VERSION = "0.8.1"
# Pattern matching
COLOR_CODE_PATTERN = re.compile(r'\^\d') # Quake color codes (^0-^9)
SPECIAL_CHAR = chr(25) # ASCII EM (End of Medium) - used in Quake messages
# Network defaults
DEFAULT_HOST = 'tcp://127.0.0.1:27961'
POLL_TIMEOUT = 100
# Timing constants
QUIT_CONFIRM_TIMEOUT = 3.0 # Seconds to confirm quit (Ctrl-C twice)
RESPAWN_DELAY = 3.0 # Seconds before players respawn after death
STATS_CONNECTION_DELAY = 0.5 # Initial stats connection delay
# UI dimensions
INFO_WINDOW_HEIGHT = 12
INFO_WINDOW_Y = 2
OUTPUT_WINDOW_Y = 14
INPUT_WINDOW_HEIGHT = 2
# Event deduplication
MAX_RECENT_EVENTS = 10
# UI settings
MAX_COMMAND_HISTORY = 10 # Number of commands to remember
# Team game modes
TEAM_MODES = [
"Team Deathmatch",
"Clan Arena",
"Capture The Flag",
"One Flag CTF",
"Overload",
"Harvester",
"Freeze Tag"
]
# Team mappings
TEAM_MAP = {
0: 'FREE',
1: 'RED',
2: 'BLUE',
3: 'SPECTATOR'
}
TEAM_COLORS = {
'RED': '^1(RED)^7',
'BLUE': '^4(BLUE)^7',
'FREE': '',
'SPECTATOR': '^3(SPEC)^7'
}
# Weapon names
WEAPON_NAMES = {
'ROCKET': '^8^1Rocket Launcher^7^0',
'LIGHTNING': '^8^3Lightning Gun^7^0',
'RAILGUN': '^8^2Railgun^7^0',
'SHOTGUN': '^8^3Shotgun^7^0',
'GAUNTLET': '^8^1Gauntlet^7^0',
'GRENADE': '^8^2Grenade Launcher^7^0',
'PLASMA': '^8^6Plasma Gun^7^0',
'MACHINEGUN': '^8^3Machine Gun^7^0'
}
# Weapon names for kill messages
WEAPON_KILL_NAMES = {
'ROCKET': 'the ^8^1Rocket Launcher',
'LIGHTNING': 'the ^8^3Lightning Gun',
'RAILGUN': 'the ^8^2Railgun',
'SHOTGUN': 'the ^8^3Shotgun',
'GAUNTLET': 'the ^8^1Gauntlet',
'GRENADE': 'the ^8^2Grenade Launcher',
'PLASMA': 'the ^8^6Plasma Gun',
'MACHINEGUN': 'the ^8^3Machine Gun'
}
# Death messages
DEATH_MESSAGES = {
'FALLING': "%s^8%s^0 ^7cratered.",
'HURT': "%s^8%s^0 ^7was in the wrong place.",
'LAVA': "%s^8%s^0 ^7does a backflip into the lava.",
'WATER': "%s^8%s^0 ^7sank like a rock.",
'SLIME': "%s^8%s^0 ^7melted.",
'CRUSH': "%s^8%s^0 ^7was crushed."
}
# Powerup names and colors
POWERUP_COLORS = {
'Quad Damage': '^8^5Quad Damage^7^0',
'Battle Suit': '^8^3Battle Suit^7^0',
'Regeneration': '^8^1Regeneration^7^0',
'Haste': '^8^3Haste^7^0',
'Invisibility': '^8^5Invisibility^7^0',
'Flight': '^8^5Flight^7^0',
'Medkit': '^8^1Medkit^7^0',
'MegaHealth': '^8^4MegaHealth^7^0'
}
# Curses color pairs
COLOR_PAIRS = {
1: 1, # Red
2: 2, # Green
3: 3, # Yellow
4: 4, # Blue
5: 6, # Cyan (swapped)
6: 5 # Magenta (swapped)
}
lib/cvars.py
+592
View File
@@ -0,0 +1,592 @@
#!/usr/bin/env python3
"""
Quake Live console variables (cvars) database
Common cvars for autocomplete and fuzzy search
"""
# Server configuration
SERVER_CVARS = [
'sv_hostname',
'sv_maxclients',
'sv_privateclients',
'sv_fps',
'sv_timeout',
'sv_minrate',
'sv_maxrate',
'sv_floodprotect',
'sv_pure',
'sv_allowdownload',
]
# Game rules
GAME_CVARS = [
'g_gametype',
'g_factoryTitle',
'g_motd',
'timelimit',
'fraglimit',
'capturelimit',
'roundlimit',
'scorelimit',
'g_allowKill',
'g_allowSpecVote',
'g_friendlyFire',
'g_teamAutoJoin',
'g_teamForceBalance',
'g_warmupDelay',
'g_inactivity',
'g_quadHog',
'g_training',
'g_instagib',
]
# Map and rotation
MAP_CVARS = [
'mapname',
'nextmap',
'g_nextmap',
]
# Voting
VOTE_CVARS = [
'g_voteDelay',
'g_voteLimit',
'g_allowVote',
]
# Items and weapons
ITEM_CVARS = [
'dmflags',
'weapon_reload_rg',
'weapon_reload_sg',
'weapon_reload_gl',
'weapon_reload_rl',
'weapon_reload_lg',
'weapon_reload_pg',
'weapon_reload_hmg',
]
# Network
NETWORK_CVARS = [
'net_port',
'net_ip',
'net_strict',
]
# ZMQ
ZMQ_CVARS = [
'zmq_rcon_enable',
'zmq_rcon_ip',
'zmq_rcon_port',
'zmq_rcon_password',
'zmq_stats_enable',
'zmq_stats_ip',
'zmq_stats_port',
'zmq_stats_password',
]
# QLX (minqlx) specific
QLX_CVARS = [
'qlx_serverBrandName',
'qlx_owner',
'qlx_redditAuth',
]
# Bot cvars
BOT_CVARS = [
'bot_enable',
'bot_nochat',
'bot_minplayers',
]
# Common commands (not cvars but useful for autocomplete)
COMMANDS = [
'status',
'map',
'map_restart',
'kick',
'kickban',
'ban',
'unban',
'tempban',
'tell',
'say',
'callvote',
'vote',
'rcon',
'addbot',
'removebot',
'killserver',
'quit',
'team',
# Match control
'readyall',
'allready',
'abort',
'pause',
'unpause',
'lock',
'unlock',
'timeout',
'timein',
# Player management
'shuffle',
'put',
'mute',
'unmute',
'slap',
'slay',
# Server control
'restart',
'endgame',
'nextmap',
'forcemap',
# QLX commands
'qlx',
'elo',
'balance',
'teams',
'scores',
# Info commands
'serverinfo',
'players',
'maplist',
'configstrings',
]
# Bot names for addbot command (complete list)
BOT_NAMES = [
'Anarki', 'Angel', 'Biker', 'Bitterman', 'Bones', 'Cadaver',
'Crash', 'Daemia', 'Doom', 'Gorre', 'Grunt', 'Hossman',
'Hunter', 'Keel', 'Klesk', 'Lucy', 'Major', 'Mynx', 'Orbb',
'Patriot', 'Phobos', 'Ranger', 'Razor', 'Sarge', 'Slash',
'Sorlag', 'Stripe', 'TankJr', 'Uriel', 'Visor', 'Wrack', 'Xaero'
]
# Skill levels for bots
BOT_SKILL_LEVELS = ['1', '2', '3', '4', '5']
# Team values
TEAM_VALUES = ['red', 'blue', 'free', 'spectator', 'r', 'b', 'f', 's']
# Popular competitive Quake Live maps
MAP_NAMES = [
'aerowalk', 'almostlost', 'arenagate', 'asylum', 'battleforged',
'bloodrun', 'brimstoneabbey', 'campgrounds', 'cannedheat',
'citycrossings', 'cure', 'deepinside', 'dismemberment',
'elder', 'eviscerated', 'falloutbunker', 'finnegans',
'furiousheights', 'gospelcrossings', 'grimdungeons',
'hearth', 'hektik', 'lostworld', 'monsoon', 'reflux',
'repent', 'shiningforces', 'sinister', 'spacectf',
'spidercrossings', 'stonekeep', 'terminus', 'tornado',
'toxicity', 'trinity', 'verticalvengeance', 'warehouses', 'whisper'
]
# Game type names (string values)
GAMETYPE_NAMES = [
'ffa', 'duel', 'race', 'tdm', 'ca', 'ctf', 'oneflag',
'har', 'ft', 'dom', 'ad', 'rr'
]
# Game type numbers (legacy numeric values)
GAMETYPE_NUMBERS = ['0', '1', '2', '3', '4', '5', '6', '7', '8', '9', '10', '11']
# Vote types for callvote command
VOTE_TYPES = [
'map', 'map_restart', 'nextmap', 'gametype', 'kick',
'timelimit', 'fraglimit', 'shuffle', 'teamsize',
'cointoss', 'random', 'loadouts', 'ammo', 'timers'
]
# Vote values
VOTE_VALUES = ['yes', 'no', '1', '2']
# Boolean values (0/1)
BOOLEAN_VALUES = ['0', '1']
# sv_fps valid values
SV_FPS_VALUES = ['20', '30', '40', '60', '125']
# sv_maxclients common values
SV_MAXCLIENTS_VALUES = ['8', '12', '16', '24', '32']
# Common time limits (minutes)
TIMELIMIT_VALUES = ['10', '15', '20', '30']
# Common frag limits
FRAGLIMIT_VALUES = ['30', '50', '75', '100']
# Common capture limits
CAPTURELIMIT_VALUES = ['5', '8', '10']
# Common round limits
ROUNDLIMIT_VALUES = ['3', '5', '7', '10']
# Common teamsize values
TEAMSIZE_VALUES = ['2', '3', '4', '5', '6']
# Factory types for map command
FACTORY_TYPES = ['ffa', 'duel', 'tdm', 'ca', 'ctf', 'iffa', 'ictf', 'ift']
# Command signatures (usage help)
COMMAND_SIGNATURES = {
# Bot commands
'addbot': '<botname:Sarge|Ranger|Visor|etc> [skill 1-5] [team] [msec delay] [altname]',
'removebot': '<botname|altname>',
# Player management
'kick': '<player>',
'kickban': '<player>',
'ban': '<player>',
'unban': '<player>',
'tempban': '<player> <seconds>',
'tell': '<player> <message>',
# Chat & voting
'say': '<message>',
'callvote': '<vote type> [args...]',
'vote': '<yes|no>',
# Map commands
'map': '<mapname>',
'map_restart': '',
# Server
'rcon': '<command>',
'killserver': '',
'quit': '',
'status': '',
# Game cvars with common values
'g_gametype': '<0=FFA 1=Duel 2=TDM 3=CA 4=CTF 5=OCTF 6=Harv 7=FT 8=Dom>',
'timelimit': '<minutes>',
'fraglimit': '<frags>',
'capturelimit': '<captures>',
'roundlimit': '<rounds>',
'scorelimit': '<score>',
# Server settings
'sv_maxclients': '<1-64>',
'sv_hostname': '<name>',
'sv_fps': '<20|30|40|60|125>',
# Network
'net_port': '<port number>',
# ZMQ
'zmq_rcon_enable': '<0|1>',
'zmq_rcon_port': '<port>',
'zmq_rcon_password': '<password>',
'zmq_stats_enable': '<0|1>',
'zmq_stats_port': '<port>',
'zmq_stats_password': '<password>',
}
# Combine all cvars
ALL_CVARS = (
SERVER_CVARS +
GAME_CVARS +
MAP_CVARS +
VOTE_CVARS +
ITEM_CVARS +
NETWORK_CVARS +
ZMQ_CVARS +
QLX_CVARS +
BOT_CVARS +
COMMANDS
)
# Sort for binary search
ALL_CVARS.sort()
# Argument value mappings - maps argument type to list of valid values
ARGUMENT_VALUES = {
'botname': BOT_NAMES,
'skill': BOT_SKILL_LEVELS,
'team': TEAM_VALUES,
'mapname': MAP_NAMES,
'gametype': GAMETYPE_NAMES + GAMETYPE_NUMBERS,
'gametype_name': GAMETYPE_NAMES,
'vote_type': VOTE_TYPES,
'vote_value': VOTE_VALUES,
'boolean': BOOLEAN_VALUES,
'sv_fps': SV_FPS_VALUES,
'sv_maxclients': SV_MAXCLIENTS_VALUES,
'timelimit': TIMELIMIT_VALUES,
'fraglimit': FRAGLIMIT_VALUES,
'capturelimit': CAPTURELIMIT_VALUES,
'roundlimit': ROUNDLIMIT_VALUES,
'teamsize': TEAMSIZE_VALUES,
'factory': FACTORY_TYPES,
}
# Command argument definitions - maps command to list of argument types
# Each argument is a dict with: type (value list key), required (bool)
COMMAND_ARGUMENTS = {
'addbot': [
{'type': 'botname', 'required': True},
{'type': 'skill', 'required': False},
{'type': 'team', 'required': False},
{'type': 'msec delay number', 'required': False}, # msec delay
{'type': 'freetext', 'required': False}, # altname
],
'removebot': [
{'type': 'botname', 'required': True},
],
'kick': [
{'type': 'player', 'required': True},
],
'kickban': [
{'type': 'player', 'required': True},
],
'ban': [
{'type': 'player', 'required': True},
],
'tempban': [
{'type': 'player', 'required': True},
{'type': 'number', 'required': True}, # seconds
],
'tell': [
{'type': 'player', 'required': True},
{'type': 'freetext', 'required': True}, # message
],
'map': [
{'type': 'mapname', 'required': True},
{'type': 'factory', 'required': False},
],
'callvote': [
{'type': 'vote_type', 'required': True},
{'type': 'dynamic', 'required': False}, # Depends on vote type
],
'vote': [
{'type': 'vote_value', 'required': True},
],
'team': [
{'type': 'team', 'required': True},
],
'g_gametype': [
{'type': 'gametype', 'required': True},
],
'timelimit': [
{'type': 'timelimit', 'required': True},
],
'fraglimit': [
{'type': 'fraglimit', 'required': True},
],
'capturelimit': [
{'type': 'capturelimit', 'required': True},
],
'roundlimit': [
{'type': 'roundlimit', 'required': True},
],
'sv_fps': [
{'type': 'sv_fps', 'required': True},
],
'sv_maxclients': [
{'type': 'sv_maxclients', 'required': True},
],
'sv_hostname': [
{'type': 'freetext', 'required': True},
],
'sv_pure': [
{'type': 'boolean', 'required': True},
],
'zmq_rcon_enable': [
{'type': 'boolean', 'required': True},
],
'zmq_rcon_port': [
{'type': 'number', 'required': True},
],
'zmq_rcon_password': [
{'type': 'freetext', 'required': True},
],
'zmq_stats_enable': [
{'type': 'boolean', 'required': True},
],
'zmq_stats_port': [
{'type': 'number', 'required': True},
],
'zmq_stats_password': [
{'type': 'freetext', 'required': True},
],
}
def get_argument_suggestions(command, arg_position, current_value, player_list=None):
"""
Get autocomplete suggestions for a command argument
Args:
command: Command name (e.g., 'addbot')
arg_position: Argument index (0 = first argument after command)
current_value: What user has typed so far for this argument
player_list: List of player names (for dynamic 'player' type)
Returns:
List of suggestion strings matching current_value
"""
# Get command's argument definitions
if command not in COMMAND_ARGUMENTS:
return []
arg_defs = COMMAND_ARGUMENTS[command]
# Check if arg_position is valid
if arg_position >= len(arg_defs):
return []
arg_def = arg_defs[arg_position]
arg_type = arg_def['type']
# Handle special types
if arg_type == 'player':
# Dynamic: get from player_list parameter
if player_list:
return fuzzy_match(current_value, player_list, max_results=5)
return []
elif arg_type == 'dynamic':
# Special case for callvote second argument
# Would need first argument to determine suggestions
# For now, return empty (could be enhanced later)
return []
elif arg_type == 'number':
# Show common numeric values
return ['50', '100', '150', '200', '250']
elif arg_type == 'freetext':
# No suggestions for free text
return []
elif arg_type in ARGUMENT_VALUES:
# Static list from ARGUMENT_VALUES
values = ARGUMENT_VALUES[arg_type]
return fuzzy_match(current_value, values, max_results=5)
return []
def fuzzy_match(query, candidates, max_results=5):
"""
Fuzzy match query against candidates
Returns list of (match, score) tuples sorted by score
Scoring:
- Exact match: 1000
- Prefix match: 500 + remaining chars
- Substring match: 100
- Contains all chars in order: 50
- Levenshtein-like: based on edit distance
"""
if not query:
# Return first max_results candidates when query is empty
return candidates[:max_results]
query_lower = query.lower()
matches = []
for candidate in candidates:
candidate_lower = candidate.lower()
score = 0
# Exact match
if query_lower == candidate_lower:
score = 1000
# Prefix match (best after exact)
elif candidate_lower.startswith(query_lower):
score = 500 + (100 - len(candidate)) # Prefer shorter matches
# Substring match
elif query_lower in candidate_lower:
# Score higher if match is earlier in string
pos = candidate_lower.index(query_lower)
score = 100 - pos
# Contains all characters in order (fuzzy)
else:
query_idx = 0
for char in candidate_lower:
if query_idx < len(query_lower) and char == query_lower[query_idx]:
query_idx += 1
if query_idx == len(query_lower): # All chars found
score = 50
if score > 0:
matches.append((candidate, score))
# Sort by score (highest first), then alphabetically
matches.sort(key=lambda x: (-x[1], x[0]))
return [match for match, score in matches[:max_results]]
def autocomplete(partial, max_results=5):
"""
Autocomplete a partial cvar/command
Returns list of suggestions
"""
return fuzzy_match(partial, ALL_CVARS, max_results)
def parse_signature(signature):
"""
Parse command signature into individual arguments
Returns list of argument strings
Example:
'<botname> [skill 1-5] [team]' -> ['<botname>', '[skill 1-5]', '[team]']
"""
import re
# Match <arg> or [arg] patterns, including content with spaces
pattern = r'(<[^>]+>|\[[^\]]+\])'
args = re.findall(pattern, signature)
return args
def get_signature_with_highlight(command, arg_position):
"""
Get command signature with current argument highlighted
Args:
command: Command name (e.g., 'addbot')
arg_position: Current argument index (0-based, 0 = first arg after command)
Returns:
List of (text, is_highlighted) tuples
"""
if command not in COMMAND_SIGNATURES:
return []
signature = COMMAND_SIGNATURES[command]
if not signature:
return []
args = parse_signature(signature)
if not args:
return [(signature, False)]
# Build list of (arg_text, is_current) tuples
result = []
for i, arg in enumerate(args):
is_current = (i == arg_position)
result.append((arg, is_current))
return result
if __name__ == '__main__':
# Test autocomplete
print("Testing autocomplete:")
print(f"Total cvars/commands: {len(ALL_CVARS)}")
print()
test_queries = ['sv_', 'time', 'zmq', 'qlx', 'map', 'stat', 'g_team']
for query in test_queries:
results = autocomplete(query)
print(f"'{query}' -> {results}")
lib/formatter.py
+211
View File
@@ -0,0 +1,211 @@
#!/usr/bin/env python3
"""
Message formatting and colorization for QLPyCon
Handles Quake color codes and team prefixes
"""
import re
import time
from .constants import TEAM_COLORS, COLOR_CODE_PATTERN, SPECIAL_CHAR
def strip_color_codes(text):
"""Remove Quake color codes (^N) from text"""
return COLOR_CODE_PATTERN.sub('', text)
def get_team_prefix(player_name, player_tracker):
"""Get color-coded team prefix for a player"""
if not player_tracker.server_info.is_team_mode():
return ''
team = player_tracker.get_team(player_name)
if not team:
return ''
return TEAM_COLORS.get(team, '')
def should_add_timestamp(message):
"""Determine if a message should get a timestamp"""
# Skip status command output
skip_keywords = ['map:', 'num score', '---', 'bot', 'status']
if any(kw in message for kw in skip_keywords):
# But allow "zmq RCON" lines (command echoes)
if 'zmq RCON' not in message:
return False
# Skip very short messages or fragments
stripped = message.strip()
if len(stripped) <= 2:
return False
# Skip messages with leading spaces (status fragments)
if message.startswith(' ') and len(stripped) < 50:
return False
# Skip pure numbers
if stripped.isdigit():
return False
# Skip short single words
if len(stripped) < 20 and ' ' not in stripped:
return False
# Skip IP addresses (xxx.xxx.xxx.xxx or xxx.xxx.xxx.xxx:port)
allowed_chars = set('0123456789.:')
if stripped.count('.') == 3 and all(c in allowed_chars for c in stripped):
return False
# Skip messages starting with ***
if message.startswith('***'):
return False
return True
def format_message(message, add_timestamp=True):
"""
Format a message for display
- Strips special characters
- Adds timestamp if appropriate
- Handles broadcast formatting
"""
# Clean up message
message = message.replace("\\n", "")
message = message.replace(SPECIAL_CHAR, "")
# Handle broadcast messages
attributes = 0
if message[:10] == "broadcast:":
message = message[11:]
attributes = 1 # Bold
# Handle print messages
if message[:7] == "print \"":
message = message[7:-2] + "\n"
# Add timestamp if requested and appropriate
if add_timestamp and should_add_timestamp(message):
timestamp = time.strftime('%H:%M:%S')
message = f"^3[^7{timestamp}^3]^7 {message}"
return message, attributes
def format_chat_message(message, player_tracker):
"""
Format chat messages with team prefixes and colors
Handles both regular chat (Name: msg) and team chat ((Name): msg or (Name) (Location): msg)
"""
# Strip special character
clean_msg = message.replace(SPECIAL_CHAR, '')
# Team chat with location: (PlayerName) (Location): message
# Location can have nested parens like (Lower Floor (Near Yellow Armour))
if clean_msg.strip().startswith('(') and ')' in clean_msg:
# Extract player name (first parenthetical)
player_match = re.match(r'^(\([^)]+\))', clean_msg)
if not player_match:
return message
player_part = player_match.group(1)
rest = clean_msg[len(player_part):].lstrip()
# Check for location (another parenthetical)
if rest.startswith('('):
# Count parens to handle nesting
paren_count = 0
location_end = -1
for i, char in enumerate(rest):
if char == '(':
paren_count += 1
elif char == ')':
paren_count -= 1
if paren_count == 0:
location_end = i + 1
break
# Check if location ends with colon
if location_end > 0 and location_end < len(rest) and rest[location_end] == ':':
location_part = rest[:location_end]
message_part = rest[location_end + 1:]
# Get team prefix
name_match = re.match(r'\(([^)]+)\)', player_part)
if name_match:
player_name = strip_color_codes(name_match.group(1).strip())
team_prefix = get_team_prefix(player_name, player_tracker)
location_clean = strip_color_codes(location_part)
return f"^8^5[TEAMSAY]^7^0 {team_prefix}^8{player_part}^0^3{location_clean}^7:^5{message_part}"
# Team chat without location: (PlayerName): message
colon_match = re.match(r'^(\([^)]+\)):(\s*.*)', clean_msg)
if colon_match:
player_part = colon_match.group(1)
message_part = colon_match.group(2)
name_match = re.match(r'\(([^)]+)\)', player_part)
if name_match:
player_name = strip_color_codes(name_match.group(1).strip())
team_prefix = get_team_prefix(player_name, player_tracker)
return f"^8^5[TEAMSAY]^7^0 {team_prefix}^8{player_part}^7^0:^5{message_part}\n"
# Regular chat: PlayerName: message
parts = clean_msg.split(':', 1)
if len(parts) == 2:
player_name = strip_color_codes(parts[0].strip())
team_prefix = get_team_prefix(player_name, player_tracker)
# Preserve original color-coded name
original_parts = message.replace(SPECIAL_CHAR, '').split(':', 1)
if len(original_parts) == 2:
return f"^8^2[SAY]^7^0 {team_prefix}^8{original_parts[0]}^0^7:^2{original_parts[1]}"
return message
def format_powerup_message(message, player_tracker):
"""
Format powerup pickup and carrier kill messages
Returns formatted message or None if not a powerup message
"""
from .constants import POWERUP_COLORS
import time
if message.startswith("broadcast:"):
message = message[11:].strip()
# Strip print " wrapper
if message.startswith('print "'):
message = message[7:]
if message.endswith('"'):
message = message[:-1]
message = message.strip()
# Powerup pickup: "PlayerName got the PowerupName!"
pickup_match = re.match(r'^(.+?)\s+got the\s+(.+?)!', message)
if pickup_match:
player_name = pickup_match.group(1).strip()
powerup_name = pickup_match.group(2).strip()
player_clean = strip_color_codes(player_name)
team_prefix = get_team_prefix(player_clean, player_tracker)
colored_powerup = POWERUP_COLORS.get(powerup_name, f'^6{powerup_name}^7')
timestamp = time.strftime('%H:%M:%S')
return f"^3[^7{timestamp}^3] ^8^5[POWERUP]^7^0 {team_prefix}^8{player_name}^0 ^7got the {colored_powerup}!\n"
# Powerup carrier kill: "PlayerName killed the PowerupName carrier!"
carrier_match = re.match(r'^(.+?)\s+killed the\s+(.+?)\s+carrier!', message)
if carrier_match:
player_name = carrier_match.group(1).strip()
powerup_name = carrier_match.group(2).strip()
player_clean = strip_color_codes(player_name)
team_prefix = get_team_prefix(player_clean, player_tracker)
colored_powerup = POWERUP_COLORS.get(powerup_name, f'^6{powerup_name}^7')
timestamp = time.strftime('%H:%M:%S')
return f"^3[^7{timestamp}^3] ^8^5[POWERUP]^7^0 {team_prefix}^8{player_name}^0 ^7killed the {colored_powerup} ^7carrier!\n"
return None
lib/network.py
+167
View File
@@ -0,0 +1,167 @@
#!/usr/bin/env python3
"""
ZMQ network layer for QLPyCon
Handles RCON and stats stream connections
"""
import zmq
import struct
import logging
import time
logger = logging.getLogger('network')
def read_socket_event(msg):
"""Parse ZMQ socket monitor event"""
event_id = struct.unpack('<H', msg[:2])[0]
event_names = {
zmq.EVENT_ACCEPTED: 'EVENT_ACCEPTED',
zmq.EVENT_ACCEPT_FAILED: 'EVENT_ACCEPT_FAILED',
zmq.EVENT_BIND_FAILED: 'EVENT_BIND_FAILED',
zmq.EVENT_CLOSED: 'EVENT_CLOSED',
zmq.EVENT_CLOSE_FAILED: 'EVENT_CLOSE_FAILED',
zmq.EVENT_CONNECTED: 'EVENT_CONNECTED',
zmq.EVENT_CONNECT_DELAYED: 'EVENT_CONNECT_DELAYED',
zmq.EVENT_CONNECT_RETRIED: 'EVENT_CONNECT_RETRIED',
zmq.EVENT_DISCONNECTED: 'EVENT_DISCONNECTED',
zmq.EVENT_LISTENING: 'EVENT_LISTENING',
zmq.EVENT_MONITOR_STOPPED: 'EVENT_MONITOR_STOPPED',
}
event_name = event_names.get(event_id, f'{event_id}')
event_value = struct.unpack('<I', msg[2:])[0]
return (event_id, event_name, event_value)
def check_monitor(monitor):
"""Check monitor socket for events"""
try:
event_monitor = monitor.recv(zmq.NOBLOCK)
except zmq.Again:
return None
event_id, event_name, event_value = read_socket_event(event_monitor)
event_endpoint = monitor.recv(zmq.NOBLOCK)
logger.debug(f'Monitor: {event_name} {event_value} endpoint {event_endpoint}')
return (event_id, event_value)
class RconConnection:
"""RCON connection to Quake Live server"""
def __init__(self, host, password, identity):
self.host = host
self.password = password
self.identity = identity
self.context = None
self.socket = None
self.monitor = None
def connect(self):
"""Initialize connection"""
logger.info('Initializing ZMQ context...')
self.context = zmq.Context()
logger.info('Creating DEALER socket...')
self.socket = self.context.socket(zmq.DEALER)
logger.info('Setting up socket monitor...')
self.monitor = self.socket.get_monitor_socket(zmq.EVENT_ALL)
if self.password:
logger.info('Setting password for access')
self.socket.plain_username = b'rcon'
self.socket.plain_password = self.password.encode('utf-8')
self.socket.zap_domain = b'rcon'
logger.info(f'Setting socket identity: {self.identity}')
self.socket.setsockopt(zmq.IDENTITY, self.identity.encode('utf-8'))
self.socket.connect(self.host)
logger.info('Connection initiated, waiting for events...')
def send_command(self, command):
"""Send RCON command"""
if isinstance(command, str):
command = command.encode('utf-8')
self.socket.send(command)
logger.info(f'Sent command: {command}')
def poll(self, timeout):
"""Poll for messages"""
return self.socket.poll(timeout)
def recv_message(self):
"""Receive a message (non-blocking)"""
try:
return self.socket.recv(zmq.NOBLOCK).decode('utf-8', errors='replace')
except zmq.error.Again:
return None
def check_monitor(self):
"""Check monitor for events"""
return check_monitor(self.monitor)
def close(self):
"""Close connection"""
if self.socket:
self.socket.setsockopt(zmq.LINGER, 0) # Don't wait for unsent messages
self.socket.close()
if self.context:
self.context.term()
class StatsConnection:
"""Stats stream connection (ZMQ SUB socket)"""
def __init__(self, host, port, password):
self.host = host
self.port = port
self.password = password
self.context = None
self.socket = None
self.connected = False
def connect(self):
"""Connect to stats stream"""
stats_host = f'tcp://{self.host}:{self.port}'
logger.info(f'Connecting to stats stream: {stats_host}')
self.context = zmq.Context()
self.socket = self.context.socket(zmq.SUB)
logger.debug('Stats socket created (SUB type)')
if self.password and self.password.strip():
logger.debug('Setting PLAIN authentication')
self.socket.setsockopt(zmq.PLAIN_USERNAME, b'stats')
self.socket.setsockopt(zmq.PLAIN_PASSWORD, self.password.encode('utf-8'))
self.socket.setsockopt_string(zmq.ZAP_DOMAIN, 'stats')
logger.debug(f'Connecting to {stats_host}')
self.socket.connect(stats_host)
logger.debug('Setting ZMQ_SUBSCRIBE to empty (all messages)')
self.socket.setsockopt(zmq.SUBSCRIBE, b'')
time.sleep(0.5)
self.connected = True
logger.info('Stats stream connected')
def recv_message(self):
"""Receive stats message (non-blocking)"""
if not self.connected:
return None
try:
msg = self.socket.recv(zmq.NOBLOCK)
return msg.decode('utf-8', errors='replace')
except zmq.error.Again:
return None
def close(self):
"""Close connection"""
if self.socket:
self.socket.setsockopt(zmq.LINGER, 0) # Don't wait for unsent messages
self.socket.close()
if self.context:
self.context.term()
lib/parser.py
+365
View File
@@ -0,0 +1,365 @@
#!/usr/bin/env python3
"""
JSON game event parsing for QLPyCon
Parses events from Quake Live stats stream
"""
import json
import logging
import time
from .constants import WEAPON_NAMES, WEAPON_KILL_NAMES, DEATH_MESSAGES
from .formatter import get_team_prefix, strip_color_codes
logger = logging.getLogger('parser')
def calculate_weapon_accuracies(weapon_data):
"""Calculate accuracy percentages for all weapons"""
accuracies = {}
for weapon, stats in weapon_data.items():
shots_fired = int(stats.get('S', 0))
shots_hit = int(stats.get('H', 0))
accuracy = shots_hit / shots_fired if shots_fired > 0 else 0
accuracies[weapon] = accuracy
return accuracies
class EventParser:
"""Parses JSON game events into formatted messages"""
def __init__(self, game_state, json_logger=None, unknown_logger=None):
self.game_state = game_state
self.json_logger = json_logger
self.unknown_logger = unknown_logger
def parse_event(self, message):
"""
Parse JSON event and return formatted message string
Returns None if event should not be displayed
"""
try:
event = json.loads(message)
# Log all JSON if logger is configured
if self.json_logger:
self.json_logger.info('JSON Event received:')
self.json_logger.info(json.dumps(event, indent=2))
self.json_logger.info('---')
if 'TYPE' not in event or 'DATA' not in event:
logger.debug('JSON missing TYPE or DATA')
return None
event_type = event['TYPE']
data = event['DATA']
if 'WARMUP' in data:
self.game_state.server_info.warmup = data['WARMUP']
# Route to appropriate handler
handler_map = {
'PLAYER_SWITCHTEAM': self._handle_switchteam,
'PLAYER_DEATH': self._handle_death,
'PLAYER_KILL': self._handle_death, # Same handler
'PLAYER_MEDAL': self._handle_medal,
'MATCH_STARTED': self._handle_match_started,
'MATCH_REPORT': self._handle_match_report,
'PLAYER_STATS': self._handle_player_stats,
'PLAYER_CONNECT': lambda d: None,
'PLAYER_DISCONNECT': lambda d: None,
'ROUND_OVER': self._handle_round_over,
}
handler = handler_map.get(event_type)
if handler:
return handler(data)
else:
# Unknown event
logger.debug(f'Unknown event type: {event_type}')
if self.unknown_logger:
self.unknown_logger.info(f'Unknown event type: {event_type}')
self.unknown_logger.info(f'Full JSON: {json.dumps(event, indent=2)}')
return None
except json.JSONDecodeError as e:
logger.debug(f'JSON decode error: {e}')
return None
except (KeyError, TypeError) as e:
logger.debug(f'Error parsing event: {e}')
return None
def _handle_switchteam(self, data):
"""Handle PLAYER_SWITCHTEAM event"""
# Get Match Time
if 'TIME' in data:
self.game_state.server_info.match_time = int(data['TIME'])
self.game_state.server_info.match_time_last_sync = time.time()
if 'KILLER' not in data:
return None
killer = data['KILLER']
name = killer.get('NAME', 'Unknown')
team = killer.get('TEAM', '')
old_team = killer.get('OLD_TEAM', '')
# Update player team
self.game_state.player_tracker.update_team(name, team)
self.game_state.player_tracker.add_player(name)
if team == old_team:
return None
warmup = " ^8^3(Warmup)^0" if data.get('WARMUP', False) else ""
team_messages = {
'FREE': ' ^7joined the ^8fight^0',
'SPECTATOR': ' ^7joined the ^3Spectators^7',
'RED': ' ^7joined the ^1RED Team^7',
'BLUE': ' ^7joined the ^4BLUE Team^7'
}
old_team_messages = {
'FREE': 'the ^8fight^0',
'SPECTATOR': 'the ^3Spectators^7',
'RED': '^7the ^1RED Team^7',
'BLUE': '^7the ^4BLUE Team^7'
}
team_msg = team_messages.get(team, f' ^7joined team {team}^7')
old_team_msg = old_team_messages.get(old_team, f'team {old_team}')
team_prefix = get_team_prefix(name, self.game_state.player_tracker)
return f"^8^5[SWITCH]^7 {team_prefix}^8{name}^0{team_msg} from {old_team_msg}{warmup}\n"
def _handle_death(self, data):
"""Handle PLAYER_DEATH and PLAYER_KILL events"""
# Get Match Time
if 'TIME' in data:
self.game_state.server_info.match_time = int(data['TIME'])
self.game_state.server_info.match_time_last_sync = time.time()
if 'VICTIM' not in data:
return None
victim = data['VICTIM']
victim_name = victim.get('NAME', 'Unknown')
# Check for duplicate
time_val = data.get('TIME', 0)
killer_name = data.get('KILLER', {}).get('NAME', '') if data.get('KILLER') else ''
if self.game_state.event_deduplicator.is_duplicate('PLAYER_DEATH', time_val, killer_name, victim_name):
return None
# Update victim team
if 'TEAM' in victim:
self.game_state.player_tracker.update_team(victim_name, victim['TEAM'])
self.game_state.player_tracker.add_player(victim_name)
# Mark as dead
if not data.get('WARMUP', False):
self.game_state.server_info.dead_players[victim_name] = time.time()
victim_prefix = get_team_prefix(victim_name, self.game_state.player_tracker)
warmup = " ^8^3(Warmup)^0" if data.get('WARMUP', False) else ""
score_prefix = ""
# Environmental death (no killer)
if 'KILLER' not in data or not data['KILLER']:
# -1 for environmental death
if not data.get('WARMUP', False):
self.game_state.player_tracker.update_score(victim_name, -1)
score_prefix = "^8^1[-1]^7^0 "
mod = data.get('MOD', 'UNKNOWN')
msg_template = DEATH_MESSAGES.get(mod, "%s^8%s^0 ^1DIED FROM %s^7")
if mod in DEATH_MESSAGES:
msg = msg_template % (victim_prefix, victim_name)
else:
msg = msg_template % (victim_prefix, victim_name, mod)
return f"{score_prefix}{msg}{warmup}\n"
# Player killed by another player
killer = data['KILLER']
killer_name = killer.get('NAME', 'Unknown')
# Update killer team
if 'TEAM' in killer:
self.game_state.player_tracker.update_team(killer_name, killer['TEAM'])
self.game_state.player_tracker.add_player(killer_name)
killer_prefix = get_team_prefix(killer_name, self.game_state.player_tracker)
# Suicide
if killer_name == victim_name:
# -1 for suicide
if not data.get('WARMUP', False):
self.game_state.player_tracker.update_score(victim_name, -1)
score_prefix = "^8^1[-1]^7^0 "
weapon = killer.get('WEAPON', 'OTHER_WEAPON')
if weapon == 'ROCKET':
return f"{score_prefix}{killer_prefix}^8{killer_name}^0 ^7blew herself up.{warmup}\n"
elif weapon == 'GRENADE':
return f"{score_prefix}{killer_prefix}^8{killer_name}^0 ^7tripped on her own grenade.{warmup}\n"
elif weapon == 'PLASMA':
return f"{score_prefix}{killer_prefix}^8{killer_name}^0 ^7melted herself.{warmup}\n"
else:
weapon_name = WEAPON_NAMES.get(weapon, weapon)
return f"{score_prefix}{killer_prefix}^8{killer_name}^0 ^7committed suicide with the ^7{weapon_name}{warmup}\n"
return None
# Regular kill: +1 for killer
if not data.get('WARMUP', False):
self.game_state.player_tracker.update_score(killer_name, 1)
score_prefix = "^8^2[+1]^7^0 "
else:
score_prefix = ""
weapon = killer.get('WEAPON', 'UNKNOWN')
weapon_name = WEAPON_KILL_NAMES.get(weapon, f'the {weapon}')
hp_left = int(killer.get('HEALTH', 0))
hp_left_colored = ""
if hp_left <= 0: # from the grave
hp_left_colored = f"^8^5From the Grave^0"
elif hp_left < 25: # red
hp_left_colored = f"^8^1{hp_left}^0 ^7HP"
elif hp_left < 80: # yellow
hp_left_colored = f"^8^3{hp_left}^0 ^7HP"
elif hp_left < 126: # white
hp_left_colored = f"^8^7{hp_left}^0 ^7HP"
else: # green
hp_left_colored = f"^8^2{hp_left}^0 ^7HP"
return f"{score_prefix}{killer_prefix}^8{killer_name}^0 ^7fragged^7 {victim_prefix}^8{victim_name}^0 ^7with {weapon_name}^0 ^7({hp_left_colored}^7){warmup}\n"
def _handle_round_over(self, data):
"""Handle ROUND_OVER events for CA"""
# Get Match Time
if 'TIME' in data:
self.game_state.server_info.match_time = int(data['TIME'])
self.game_state.server_info.match_time_last_sync = time.time()
team_won = data.get('TEAM_WON')
round_num = data.get('ROUND', 0)
if team_won == 'RED':
self.game_state.server_info.red_rounds += 1
logger.info(f"Round {round_num}: RED wins (RED: {self.game_state.server_info.red_rounds}, BLUE: {self.game_state.server_info.blue_rounds})")
elif team_won == 'BLUE':
self.game_state.server_info.blue_rounds += 1
logger.info(f"Round {round_num}: BLUE wins (RED: {self.game_state.server_info.red_rounds}, BLUE: {self.game_state.server_info.blue_rounds})")
self.game_state.server_info.round_end_time = time.time()
return None # Don't display in chat
def _handle_medal(self, data):
"""Handle PLAYER_MEDAL event"""
# Get Match Time
if 'TIME' in data:
self.game_state.server_info.match_time = int(data['TIME'])
self.game_state.server_info.match_time_last_sync = time.time()
name = data.get('NAME', 'Unknown')
medal = data.get('MEDAL', 'UNKNOWN')
warmup = " ^8^3(Warmup)^7^0" if data.get('WARMUP', False) else ""
medal_prefix = "^8^6[MEDAL]^7^0 "
team_prefix = get_team_prefix(name, self.game_state.player_tracker)
# RED Medals (^1)
if medal in ["FIRSTFRAG", "HUMILIATION", "REVENGE"]:
return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^1{medal}^0{warmup}\n"
# GREEN Medals (^2)
elif medal in ["MIDAIR", "PERFECT"]:
return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^2{medal}^0{warmup}\n"
# YELLOW Medals (^3)
elif medal in ["EXCELLENT", "HEADSHOT", "RAMPAGE"]:
return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^3{medal}^0{warmup}\n"
# BLUE Medals (^4)
elif medal in ["ASSIST", "DEFENSE", "QUADGOD"]:
return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^4{medal}^0{warmup}\n"
# CYAN Medals (^5)
elif medal in ["CAPTURE", "COMBOKILL", "IMPRESSIVE"]:
return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^5{medal}^0{warmup}\n"
# PINK Medals (^6)
elif medal in ["ACCURACY", "PERFORATED"]:
return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^6{medal}^0{warmup}\n"
else:
return f"{medal_prefix}{team_prefix}^8{name}^0 ^7got ^8^7{medal}^0{warmup}\n"
def _handle_match_started(self, data):
"""Handle MATCH_STARTED event"""
# Get Match Time
if 'TIME' in data:
self.game_state.server_info.match_time = int(data['TIME'])
self.game_state.server_info.match_time_last_sync = time.time()
if self.game_state.server_info.is_team_mode():
return f"^8^2[GAME ON]^0 ^7Match has started - ^1^8RED ^0^7vs. ^4^8BLUE\n"
players = []
for player in data.get('PLAYERS', []):
name = player.get('NAME', 'Unknown')
players.append(name)
if players:
formatted = "^0 vs. ^8".join(players)
return f"^8^2[GAME ON]^0 ^7Match has started - ^8^7{formatted}\n"
return None
def _handle_match_report(self, data):
"""Handle MATCH_REPORT event"""
# Get Match Time
if 'TIME' in data:
self.game_state.server_info.match_time = int(data['TIME'])
self.game_state.server_info.match_time_last_sync = time.time()
if not self.game_state.server_info.is_team_mode():
return None
red_score = int(data.get('TSCORE0', '0'))
blue_score = int(data.get('TSCORE1', '0'))
report_prefix = "^8^1[GAME OVER]"
if red_score > blue_score:
return f"{report_prefix} ^7The ^1RED TEAM ^7WINS^0 by a score of ^8^1{red_score} ^0^7to ^8^4{blue_score}\n"
elif blue_score > red_score:
return f"{report_prefix} ^7The ^4BLUE TEAM ^7WINS^0 by a score of ^8^4{blue_score} ^0^7to ^8^1{red_score}\n"
else:
return f"{report_prefix} ^7The match is a TIE^0 with a score of ^8^1{red_score} ^0^7to ^8^4{blue_score}\n"
def _handle_player_stats(self, data):
"""Handle PLAYER_STATS event"""
name = data.get('NAME', 'Unknown')
team_prefix = get_team_prefix(name, self.game_state.player_tracker)
kills = int(data.get('KILLS', '0'))
deaths = int(data.get('DEATHS', '0'))
weapon_data = data.get('WEAPONS', {})
accuracies = calculate_weapon_accuracies(weapon_data)
if not accuracies:
return None
best_weapon = max(accuracies, key=accuracies.get)
best_accuracy = accuracies[best_weapon] * 100
weapon_stats = weapon_data.get(best_weapon, {})
best_weapon_kills = int(weapon_stats.get('K', 0))
weapon_name = WEAPON_NAMES.get(best_weapon, best_weapon)
return f"^8^5[PLAYER STATS]^7^0 {team_prefix}^8^7{name}^0^7 K/D: {kills}/{deaths} | Best Weapon: {weapon_name} - Acc: {best_accuracy:.2f}% - Kills: {best_weapon_kills}\n"
lib/settings.py
+209
View File
@@ -0,0 +1,209 @@
#!/usr/bin/env python3
"""
Configuration file handler for QLPyCon
Supports loading from ~/.qlpycon.conf or ./qlpycon.conf
"""
import os
import configparser
import logging
logger = logging.getLogger('config_loader')
class ConfigLoader:
"""Load configuration from INI file"""
def __init__(self):
self.config = configparser.ConfigParser()
self.config_loaded = False
def load(self):
"""
Try to load config from (in order):
1. ./qlpycon.conf (current directory)
2. ~/.qlpycon.conf (home directory)
"""
config_paths = [
'qlpycon.conf',
os.path.expanduser('~/.qlpycon.conf')
]
for path in config_paths:
if os.path.exists(path):
try:
self.config.read(path)
self.config_loaded = True
logger.info(f'Loaded configuration from: {path}')
return True
except Exception as e:
logger.warning(f'Failed to load config from {path}: {e}')
logger.debug('No configuration file found, using defaults')
return False
def get(self, section, key, fallback=None):
"""Get a configuration value"""
if not self.config_loaded:
return fallback
try:
return self.config.get(section, key, fallback=fallback)
except (configparser.NoSectionError, configparser.NoOptionError):
return fallback
def get_int(self, section, key, fallback=0):
"""Get an integer configuration value"""
value = self.get(section, key)
if value is None:
return fallback
try:
return int(value)
except ValueError:
logger.warning(f'Invalid integer value for [{section}] {key}: {value}')
return fallback
def get_bool(self, section, key, fallback=False):
"""Get a boolean configuration value"""
value = self.get(section, key)
if value is None:
return fallback
return value.lower() in ('true', 'yes', '1', 'on')
def get_host(self):
"""Get connection host"""
return self.get('connection', 'host')
def get_password(self):
"""Get connection password (supports ${ENV_VAR} syntax)"""
password = self.get('connection', 'password')
if not password:
return None
return self._resolve_password(password)
def get_servers(self):
"""Return dict of name -> host:port from [servers]"""
if not self.config.has_section('servers'):
return {}
return dict(self.config.items('servers'))
def get_server(self, name):
"""
Resolve a named server to (host, password).
host comes from [servers], password from [server:name] or [connection].
Returns (host, password) or (None, None) if name not found.
"""
servers = self.get_servers()
if name not in servers:
return None, None
host = servers[name]
if not host.startswith('tcp://'):
host = f'tcp://{host}'
# Per-server password override
section = f'server:{name}'
if self.config.has_section(section):
password = self.config.get(section, 'password', fallback=None)
if password:
password = self._resolve_password(password)
else:
password = self.get_password()
return host, password
def _resolve_password(self, password):
"""Resolve a password string, expanding ${VAR:-default} if needed"""
if not password:
return None
if password.startswith('${') and password.endswith('}'):
inner = password[2:-1]
if ':-' in inner:
env_var, default = inner.split(':-', 1)
else:
env_var, default = inner, None
return os.environ.get(env_var, default)
return password
def get_log_level(self):
"""Get logging level"""
level_str = self.get('logging', 'level', 'INFO')
levels = {
'DEBUG': logging.DEBUG,
'INFO': logging.INFO,
'WARNING': logging.WARNING,
'ERROR': logging.ERROR,
'CRITICAL': logging.CRITICAL
}
return levels.get(level_str.upper(), logging.INFO)
def create_example_config():
"""Create an example configuration file"""
config_content = """# qlpycon.conf
# Edit this file as needed.
#
# Connect by server name: qlpycon ffa
# Connect directly: qlpycon --host tcp://1.2.3.4:28960 --password secret
# List servers: qlpycon --list
# === Connection defaults ===>
# Default host if no server name or --host is given
[connection]
host = tcp://127.0.0.1:28960
# Password for all servers unless overridden in [server:name]
# Use ${ENV_VAR} to read from environment variable (recommended)
# Or set directly (less secure):
password = ${QLPYCON_PASSWORD:-secret}
# === Named servers ===>
# Simple entries: name = host:port
# These use the password from [connection] above.
[servers]
# Example:
#ffa = 10.13.12.161:28960
# === Per-server overrides ===>
# Use [server:name] to override any setting for a specific server.
# The name must match an entry in [servers] above.
# Example:
#[server:ffa]
#password = ${FFA_PASSWORD:-secret}
# === Logging ===>
[logging]
# Log level: DEBUG, INFO, WARNING, ERROR, CRITICAL
level = WARNING
# === UI ===>
[ui]
# Number of commands to remember in history
max_history = 10
# === Behaviour ===>
[behavior]
# Seconds to confirm quit (press Ctrl-C twice within this window)
quit_timeout = 3.0
# Seconds before players respawn after death
respawn_delay = 3.0
"""
example_path = os.path.expanduser('~/.qlpycon.conf.example')
try:
with open(example_path, 'w') as f:
f.write(config_content)
print(f'Created example config: {example_path}')
print(f'Copy to ~/.qlpycon.conf and edit as needed')
return True
except Exception as e:
print(f'Failed to create example config: {e}')
return False
if __name__ == '__main__':
# Create example config when run directly
create_example_config()
lib/state.py
+231
View File
@@ -0,0 +1,231 @@
#!/usr/bin/env python3
"""
Game state management for QLPyCon
Tracks server info, players, and teams
"""
import logging
from .constants import TEAM_MODES, TEAM_MAP, MAX_RECENT_EVENTS
from .formatter import strip_color_codes
logger = logging.getLogger('state')
class ServerInfo:
"""Tracks current server information"""
def __init__(self):
self.hostname = 'Unknown'
self.map = 'Unknown'
self.gametype = 'Unknown'
self.timelimit = '0'
self.fraglimit = '0'
self.roundlimit = '0'
self.capturelimit = '0'
self.maxclients = '0'
self.curclients = '0'
self.red_score = 0
self.red_rounds = 0
self.blue_score = 0
self.blue_rounds = 0
self.players = {} # Changed to dict: {name: {'score': str, 'ping': str}}
self.last_update = 0
self.warmup = False
self.dead_players = {}
self.round_end_time = None
self.match_time = 0
self.match_time_last_sync = 0 # Timestamp of last TIME update from server
def is_team_mode(self):
"""Check if current gametype is a team mode"""
return self.gametype in TEAM_MODES
def reset_round_scores(self):
"""Reset round scores (for new matches)"""
self.red_rounds = 0
self.blue_rounds = 0
self.dead_players.clear()
def update_from_cvar(self, cvar_name, value):
"""Update server info from a cvar response"""
# Normalize cvar name to lowercase for case-insensitive matching
cvar_lower = cvar_name.lower()
mapping = {
'qlx_serverbrandname': 'hostname',
'g_factorytitle': 'gametype',
'mapname': 'map',
'timelimit': 'timelimit',
'fraglimit': 'fraglimit',
'roundlimit': 'roundlimit',
'capturelimit': 'capturelimit',
'sv_maxclients': 'maxclients'
}
attr = mapping.get(cvar_lower)
if attr:
# Only strip color codes for non-hostname fields
if attr != 'hostname':
value = strip_color_codes(value)
setattr(self, attr, value)
logger.info(f'Updated {attr}: {value}')
return True
return False
class PlayerTracker:
"""Tracks player teams and information"""
def __init__(self, server_info):
self.server_info = server_info
self.player_teams = {}
def update_team(self, name, team):
"""Update player team. Team can be int or string"""
# Convert numeric team to string
if isinstance(team, int):
team = TEAM_MAP.get(team, 'FREE')
if team not in ['RED', 'BLUE', 'FREE', 'SPECTATOR']:
team = 'FREE'
# Store both original name and color-stripped version
self.player_teams[name] = team
clean_name = strip_color_codes(name)
if clean_name != name:
self.player_teams[clean_name] = team
logger.debug(f'Updated team for {name} (clean: {clean_name}): {team}')
def get_team(self, name):
"""Get player's team"""
return self.player_teams.get(name)
def add_player(self, name, score='0', ping='0'):
"""Add player to server's player dict if not exists"""
# Use original name with color codes as key
if name not in self.server_info.players:
self.server_info.players[name] = {
'score': score,
'ping': ping
}
logger.debug(f'Added player: {name}')
def get_players_by_team(self):
"""Get players organized by team"""
teams = {'RED': [], 'BLUE': [], 'SPECTATOR': [], 'FREE': []}
for name in self.server_info.players.keys():
team = self.player_teams.get(name, 'FREE')
if team not in teams:
team = 'FREE'
teams[team].append(name)
return teams
def remove_player(self, name):
"""Remove player from tracking"""
clean_name = strip_color_codes(name)
# Try to remove by exact name first
removed = self.server_info.players.pop(name, None)
# If not found, try to find by clean name
if not removed:
for player_name in list(self.server_info.players.keys()):
if strip_color_codes(player_name) == clean_name:
removed = self.server_info.players.pop(player_name)
logger.info(f'Removed player: {player_name} (matched clean name: {clean_name})')
break
else:
logger.info(f'Removed player: {name}')
if not removed:
logger.warning(f'Player not found for removal: {name} (clean: {clean_name})')
# Remove from team tracking
self.player_teams.pop(name, None)
self.player_teams.pop(clean_name, None)
def rename_player(self, old_name, new_name):
"""Rename a player while maintaining their team and score"""
old_clean = strip_color_codes(old_name)
# Get current team (try both names)
team = self.player_teams.get(old_name) or self.player_teams.get(old_clean, 'SPECTATOR')
# Find player data by old name
player_data = self.server_info.players.pop(old_name, None)
# If not found by exact name, try clean name
if not player_data:
for player_name in list(self.server_info.players.keys()):
if strip_color_codes(player_name) == old_clean:
player_data = self.server_info.players.pop(player_name)
break
# Add player with new name
if player_data:
self.server_info.players[new_name] = player_data
# Remove old team entries
self.player_teams.pop(old_name, None)
self.player_teams.pop(old_clean, None)
# Add new team entries with color codes preserved
self.update_team(new_name, team)
logger.debug(f'Renamed player: {old_name} -> {new_name} (team: {team})')
def update_score(self, name, delta):
"""Update player's score by delta (+1 for kill, -1 for death/suicide)"""
# Try exact name first (O(1) lookup)
if name in self.server_info.players:
current_score = int(self.server_info.players[name].get('score', 0))
self.server_info.players[name]['score'] = str(current_score + delta)
logger.debug(f"Score update: {name} {delta:+d} -> {self.server_info.players[name]['score']}")
return
# Fallback: search by clean name (rare case)
clean_name = strip_color_codes(name)
for player_name, player_data in self.server_info.players.items():
if strip_color_codes(player_name) == clean_name:
current_score = int(player_data.get('score', 0))
player_data['score'] = str(current_score + delta)
logger.debug(f"Score update: {player_name} {delta:+d} -> {player_data['score']}")
return
logger.warning(f"Could not update score for {name} - player not found")
class EventDeduplicator:
"""Prevents duplicate kill/death events"""
def __init__(self):
self.recent_events = []
def is_duplicate(self, event_type, time_val, killer_name, victim_name):
"""Check if this kill/death event is a duplicate"""
if event_type not in ('PLAYER_DEATH', 'PLAYER_KILL'):
return False
signature = f"KILL:{time_val}:{killer_name}:{victim_name}"
if signature in self.recent_events:
logger.debug(f'Duplicate event: {signature}')
return True
# Add to recent events
self.recent_events.append(signature)
if len(self.recent_events) > MAX_RECENT_EVENTS:
self.recent_events.pop(0)
return False
class GameState:
"""Main game state container"""
def __init__(self):
self.server_info = ServerInfo()
self.player_tracker = PlayerTracker(self.server_info)
self.event_deduplicator = EventDeduplicator()
self.pending_background_commands = set()
self.status_buffer = []
lib/ui.py
+807
View File
@@ -0,0 +1,807 @@
#!/usr/bin/env python3
"""
Curses-based UI for QLPyCon
Handles terminal display, windows, and color rendering
"""
import curses
import curses.textpad
import threading
import queue
import logging
import time
from .constants import COLOR_PAIRS, INFO_WINDOW_HEIGHT, INFO_WINDOW_Y, OUTPUT_WINDOW_Y, INPUT_WINDOW_HEIGHT, TEAM_MODES, MAX_COMMAND_HISTORY
from .cvars import autocomplete, COMMAND_SIGNATURES, get_signature_with_highlight, get_argument_suggestions, COMMAND_ARGUMENTS
logger = logging.getLogger('ui')
class CursesHandler(logging.Handler):
"""Logging handler that outputs to curses window"""
def __init__(self, window):
logging.Handler.__init__(self)
self.window = window
def emit(self, record):
try:
msg = self.format(record)
fs = "%s\n"
try:
print_colored(self.window, fs % msg, 0)
self.window.noutrefresh()
curses.doupdate()
except UnicodeError:
print_colored(self.window, fs % msg.encode("UTF-8"), 0)
self.window.noutrefresh()
curses.doupdate()
except (KeyboardInterrupt, SystemExit):
raise
except Exception:
self.handleError(record)
def print_colored(window, message, attributes=0):
"""
Print message with Quake color codes (^N)
^0 = reset, ^1 = red, ^2 = green, ^3 = yellow, ^4 = blue, ^5 = cyan, ^6 = magenta, ^7 = white, ^8 = bold, ^9 = underline
"""
if not curses.has_colors:
try:
window.addstr(message)
except curses.error:
pass
return
color = 0
bold = False
underline = False
parse_color = False
for ch in message:
val = ord(ch)
if parse_color:
if ch == '8':
bold = True
elif ch == '9':
underline = True
elif ch == '0':
bold = False
underline = False
elif ch == '7':
color = 0
elif ord('1') <= val <= ord('6'):
color = val - ord('0')
else:
try:
window.addch('^', curses.color_pair(color) | (curses.A_BOLD if bold else 0) | (curses.A_UNDERLINE if underline else 0) | attributes)
window.addch(ch, curses.color_pair(color) | (curses.A_BOLD if bold else 0) | (curses.A_UNDERLINE if underline else 0) | attributes)
except curses.error:
return
parse_color = False
elif ch == '^':
parse_color = True
else:
try:
window.addch(ch, curses.color_pair(color) | (curses.A_BOLD if bold else 0) | (curses.A_UNDERLINE if underline else 0) | attributes)
except curses.error:
return
def update_autocomplete_display(window, current_input, first_word, words, ends_with_space):
"""
Update autocomplete display based on current input state.
Returns (suggestions, suggestion_index, original_word) tuple for Tab cycling.
Handles three display modes:
1. Command autocomplete (typing partial command)
2. Signature display (command recognized, showing arguments)
3. Argument value suggestions (typing argument values)
"""
suggestions = []
suggestion_index = -1
original_word = ""
# Check if this is a command with argument definitions
if first_word in COMMAND_ARGUMENTS:
# Determine if user is typing arguments (not just the command)
if len(words) == 1 and not ends_with_space:
# Just command, no space yet → show signature with first arg highlighted
sig_parts = get_signature_with_highlight(first_word, 0)
if sig_parts:
x_pos = 0
for arg_text, is_current in sig_parts:
try:
if is_current:
window.addstr(1, x_pos, arg_text, curses.A_REVERSE)
else:
window.addstr(1, x_pos, arg_text, curses.A_DIM)
x_pos += len(arg_text) + 1
except curses.error:
pass
else:
# User is typing arguments
if ends_with_space:
# Starting new argument (empty so far)
arg_position = len(words) - 1 # -1 for command
current_value = ''
else:
# Typing current argument
arg_position = len(words) - 2 # -1 for command, -1 for 0-indexed
current_value = words[-1]
# Get argument suggestions
arg_suggestions = get_argument_suggestions(
first_word,
arg_position,
current_value,
player_list=None # TODO: pass player list from game_state
)
if arg_suggestions:
# Show argument value suggestions with label (limit to 10 for performance)
arg_type = COMMAND_ARGUMENTS[first_word][arg_position]['type']
display_suggestions = arg_suggestions[:10]
more_indicator = f' (+{len(arg_suggestions)-10} more)' if len(arg_suggestions) > 10 else ''
match_line = f'<{arg_type}>: {" ".join(display_suggestions)}{more_indicator}'
try:
window.addstr(1, 0, match_line, curses.A_DIM)
except curses.error:
pass
suggestions = arg_suggestions # Store for Tab cycling
suggestion_index = -1
original_word = current_value
else:
# No suggestions (freetext, player without list, etc.) → show signature
sig_parts = get_signature_with_highlight(first_word, arg_position)
if sig_parts:
x_pos = 0
for arg_text, is_current in sig_parts:
try:
if is_current:
window.addstr(1, x_pos, arg_text, curses.A_REVERSE)
else:
window.addstr(1, x_pos, arg_text, curses.A_DIM)
x_pos += len(arg_text) + 1
except curses.error:
pass
elif first_word in COMMAND_SIGNATURES and COMMAND_SIGNATURES[first_word]:
# Command with signature but no argument definitions
sig_parts = get_signature_with_highlight(first_word, 0)
if sig_parts:
x_pos = 0
for arg_text, is_current in sig_parts:
try:
if is_current:
window.addstr(1, x_pos, arg_text, curses.A_REVERSE)
else:
window.addstr(1, x_pos, arg_text, curses.A_DIM)
x_pos += len(arg_text) + 1
except curses.error:
pass
else:
# Not a recognized command → show command autocomplete
current_word = words[-1]
if len(current_word) >= 2:
suggestions = autocomplete(current_word, max_results=5)
suggestion_index = -1
original_word = current_word
if suggestions:
match_line = ' '.join(suggestions)
try:
window.addstr(1, 0, match_line, curses.A_DIM)
except curses.error:
pass
return suggestions, suggestion_index, original_word
class UIManager:
"""Manages curses windows and display"""
def __init__(self, screen, host):
self.screen = screen
self.host = host
self.info_window = None
self.output_window = None
self.input_window = None
self.divider_window = None
self.input_queue = None
self.command_history = []
self.history_index = -1
self.cursor_pos = 0 # Track cursor position in input
self._init_curses()
self._create_windows()
def _init_curses(self):
"""Initialize curses settings"""
curses.endwin()
curses.initscr()
self.screen.nodelay(1)
curses.start_color()
curses.use_default_colors()
curses.cbreak()
curses.curs_set(1) # Show cursor in input window
self.screen.addstr(f"Quake Live PyCon: {self.host}")
self.screen.noutrefresh()
# Initialize color pairs
for i in range(1, 7):
curses.init_pair(i, i, 0)
# Swap cyan and magenta (5 and 6)
curses.init_pair(5, 6, 0)
curses.init_pair(6, 5, 0)
def _create_windows(self):
"""Create all UI windows"""
maxy, maxx = self.screen.getmaxyx()
# Minimum terminal size check
if maxy < 20 or maxx < 80:
return False
# Server info window (top)
self.info_window = curses.newwin(
INFO_WINDOW_HEIGHT,
maxx - 4,
INFO_WINDOW_Y,
2
)
self.info_window.scrollok(False)
self.info_window.idlok(False)
self.info_window.leaveok(True)
self.info_window.noutrefresh()
# Output window (middle - main display)
self.output_window = curses.newwin(
maxy - 17,
maxx - 4,
OUTPUT_WINDOW_Y,
2
)
self.output_window.scrollok(True)
self.output_window.idlok(False)
self.output_window.idcok(False)
self.output_window.leaveok(True)
self.output_window.noutrefresh()
# Divider line
self.divider_window = curses.newwin(
1,
maxx - 4,
maxy - 3,
2
)
self.divider_window.hline(curses.ACS_HLINE, maxx - 4)
self.divider_window.scrollok(False)
self.divider_window.idlok(False)
self.divider_window.leaveok(True)
self.divider_window.noutrefresh()
# Input window (bottom)
self.input_window = curses.newwin(
INPUT_WINDOW_HEIGHT,
maxx - 6,
maxy - 2,
4
)
self.input_window.keypad(True)
self.input_window.nodelay(False)
self.screen.addstr(maxy - 2, 2, '$ ')
self.input_window.idlok(True)
self.input_window.idcok(True)
self.input_window.leaveok(False)
self.input_window.noutrefresh()
self.screen.noutrefresh()
curses.doupdate()
return True
def handle_resize(self):
"""Handle terminal resize event"""
try:
# Get new terminal dimensions
maxy, maxx = self.screen.getmaxyx()
# Minimum size check
if maxy < 20 or maxx < 80:
return False
# Update screen
curses.update_lines_cols()
self.screen.clear()
self.screen.addstr(0, 0, f"Quake Live PyCon: {self.host}")
self.screen.noutrefresh()
# Recreate windows with new dimensions
self.info_window.resize(INFO_WINDOW_HEIGHT, maxx - 4)
self.info_window.mvwin(INFO_WINDOW_Y, 2)
self.output_window.resize(maxy - 17, maxx - 4)
self.output_window.mvwin(OUTPUT_WINDOW_Y, 2)
self.divider_window.resize(1, maxx - 4)
self.divider_window.mvwin(maxy - 3, 2)
self.divider_window.clear()
self.divider_window.hline(curses.ACS_HLINE, maxx - 4)
self.input_window.resize(INPUT_WINDOW_HEIGHT, maxx - 6)
self.input_window.mvwin(maxy - 2, 4)
self.screen.addstr(maxy - 2, 2, '$ ')
# Refresh all windows
self.info_window.noutrefresh()
self.output_window.noutrefresh()
self.divider_window.noutrefresh()
self.input_window.noutrefresh()
self.screen.noutrefresh()
curses.doupdate()
return True
except curses.error:
return False
def setup_input_queue(self):
"""Setup threaded input queue with command history and autocomplete"""
def wait_stdin(q, window, manager):
current_input = ""
cursor_pos = 0
manager.cursor_pos = 0 # Keep manager in sync
temp_history_index = -1
temp_input = "" # Temp storage when navigating history
quit_confirm = False
# Autocomplete state
suggestions = []
suggestion_index = -1
original_word = "" # Store original word before cycling
while True:
try:
key = window.getch()
if key == -1: # No input
continue
# Handle terminal resize
if key == curses.KEY_RESIZE:
manager.handle_resize()
# Redraw input
window.erase()
window.addstr(0, 0, current_input)
window.move(0, cursor_pos)
window.noutrefresh()
curses.doupdate()
continue
# Tab key - cycle through suggestions
if key == ord('\t') or key == 9:
if suggestions:
# Cycle to next suggestion
suggestion_index = (suggestion_index + 1) % len(suggestions)
# Replace or append suggestion
words = current_input.split()
if words:
# If original_word is empty, we had trailing space - append new word
# Otherwise, replace current word
if original_word == '':
words.append(suggestions[suggestion_index])
# Update original_word so next Tab replaces instead of appending
original_word = suggestions[suggestion_index]
else:
words[-1] = suggestions[suggestion_index]
current_input = ' '.join(words)
cursor_pos = len(current_input)
manager.cursor_pos = cursor_pos
# Update display
window.erase()
window.addstr(0, 0, current_input)
# Determine display format
first_word = words[0].lower()
selected_value = suggestions[suggestion_index]
# Check if we're cycling argument values or commands
if first_word in COMMAND_ARGUMENTS and len(words) > 1:
# Cycling argument values - show with label
ends_with_space = current_input.endswith(' ')
if ends_with_space:
arg_position = len(words) - 1
else:
arg_position = len(words) - 2
# Bounds check to prevent index out of range
if arg_position < len(COMMAND_ARGUMENTS[first_word]):
arg_type = COMMAND_ARGUMENTS[first_word][arg_position]['type']
# Show only first 10 suggestions for performance
display_suggestions = suggestions[:10]
more_indicator = f' (+{len(suggestions)-10} more)' if len(suggestions) > 10 else ''
display_line = f'<{arg_type}>: {" ".join(display_suggestions)}{more_indicator}'
else:
# Fallback if position out of bounds
display_line = ' '.join(suggestions[:10])
else:
# Cycling commands - show with signature if available
display_suggestions = suggestions[:10]
match_line = ' '.join(display_suggestions)
if selected_value in COMMAND_SIGNATURES:
signature = COMMAND_SIGNATURES[selected_value]
if signature:
display_line = f"{match_line} → {signature}"
else:
display_line = match_line
else:
display_line = match_line
try:
window.addstr(1, 0, display_line, curses.A_DIM)
except curses.error:
pass
window.move(0, cursor_pos)
window.noutrefresh()
curses.doupdate() # Actually push the refresh to screen
continue
# Enter key
if key in (curses.KEY_ENTER, 10, 13):
if len(current_input) > 0:
# Add to history
manager.command_history.append(current_input)
if len(manager.command_history) > MAX_COMMAND_HISTORY:
manager.command_history.pop(0)
q.put(current_input)
current_input = ""
cursor_pos = 0
manager.cursor_pos = cursor_pos
temp_history_index = -1
temp_input = ""
suggestions = []
suggestion_index = -1
original_word = ""
window.erase()
window.noutrefresh()
# Arrow UP - previous command
elif key == curses.KEY_UP:
if len(manager.command_history) > 0:
# Save current input when first entering history
if temp_history_index == -1:
temp_input = current_input
temp_history_index = len(manager.command_history)
if temp_history_index > 0:
temp_history_index -= 1
current_input = manager.command_history[temp_history_index]
cursor_pos = len(current_input)
manager.cursor_pos = cursor_pos
suggestions = []
suggestion_index = -1
original_word = ""
window.erase()
window.addstr(0, 0, current_input)
window.noutrefresh()
# Arrow DOWN - next command
elif key == curses.KEY_DOWN:
if temp_history_index != -1:
temp_history_index += 1
if temp_history_index >= len(manager.command_history):
# Restore temp input
current_input = temp_input
temp_history_index = -1
temp_input = ""
else:
current_input = manager.command_history[temp_history_index]
cursor_pos = len(current_input)
manager.cursor_pos = cursor_pos
suggestions = []
suggestion_index = -1
original_word = ""
window.erase()
window.addstr(0, 0, current_input)
window.noutrefresh()
# Arrow LEFT - move cursor left
elif key == curses.KEY_LEFT:
if cursor_pos > 0:
cursor_pos -= 1
manager.cursor_pos = cursor_pos
window.move(0, cursor_pos)
window.noutrefresh()
# Arrow RIGHT - move cursor right
elif key == curses.KEY_RIGHT:
if cursor_pos < len(current_input):
cursor_pos += 1
manager.cursor_pos = cursor_pos
window.move(0, cursor_pos)
window.noutrefresh()
# Backspace
elif key in (curses.KEY_BACKSPACE, 127, 8):
if cursor_pos > 0:
current_input = current_input[:cursor_pos-1] + current_input[cursor_pos:]
cursor_pos -= 1
manager.cursor_pos = cursor_pos
temp_history_index = -1 # Exit history mode
window.erase()
window.addstr(0, 0, current_input)
# Parse input and update autocomplete display
words = current_input.split()
ends_with_space = current_input.endswith(' ')
if words:
first_word = words[0].lower()
suggestions, suggestion_index, original_word = update_autocomplete_display(
window, current_input, first_word, words, ends_with_space
)
window.move(0, cursor_pos)
window.noutrefresh()
curses.doupdate() # Immediate screen update
# Regular character
elif 32 <= key <= 126:
char = chr(key)
current_input = current_input[:cursor_pos] + char + current_input[cursor_pos:]
cursor_pos += 1
manager.cursor_pos = cursor_pos
temp_history_index = -1 # Exit history mode
window.erase()
window.addstr(0, 0, current_input)
# Parse input and update autocomplete display
words = current_input.split()
ends_with_space = current_input.endswith(' ')
if words:
first_word = words[0].lower()
suggestions, suggestion_index, original_word = update_autocomplete_display(
window, current_input, first_word, words, ends_with_space
)
window.move(0, cursor_pos)
window.noutrefresh()
curses.doupdate() # Immediate screen update
except Exception as e:
logger.error(f'Input error: {e}')
# Log but continue - input thread should stay alive
window.move(0, cursor_pos)
curses.doupdate()
self.input_queue = queue.Queue()
t = threading.Thread(target=wait_stdin, args=(self.input_queue, self.input_window, self))
t.daemon = True
t.start()
return self.input_queue
def setup_logging(self):
"""Setup logging handler for output window"""
handler = CursesHandler(self.output_window)
formatter = logging.Formatter('%(asctime)-8s|%(name)-12s|%(levelname)-6s|%(message)-s', '%H:%M:%S')
handler.setFormatter(formatter)
return handler
def print_message(self, message, attributes=0):
"""Print formatted message to output window"""
print_colored(self.output_window, message, attributes)
self.output_window.noutrefresh()
# Restore cursor to input window at current position
self.input_window.move(0, self.cursor_pos)
self.input_window.noutrefresh()
curses.doupdate()
def update_server_info(self, game_state):
"""Update server info window"""
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)
score = int(player_data.get('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']:
score = int(server_info.players.get(player_name, {}).get('score', 0))
red_players_with_scores.append((player_name, score))
for player_name in teams['BLUE']:
score = int(server_info.players.get(player_name, {}).get('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 ''
from .formatter import strip_color_codes
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:
score = int(server_info.players.get(player_name, {}).get('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 ''
from .formatter import strip_color_codes
col1_clean = strip_color_codes(col1)
col2_clean = strip_color_codes(col2)
col1_pad = 24 - len(col1_clean)
line = f"^8{col1}^0{' ' * col1_pad}^8{col2}^0\n"
print_colored(self.info_window, line, 0)
# Blank lines to fill
try:
self.info_window.addstr("\n")
except curses.error:
pass
# List spectators on one line
spec_list = " ".join(spec_players)
line = f"^8^3Spectators:^7 {spec_list}\n"
print_colored(self.info_window, line, 0)
# Blank lines to fill
try:
self.info_window.addstr("\n")
except curses.error:
pass
# Separator
separator = "^7" + "═" * (max_x - 1) + "^7"
print_colored(self.info_window, separator, 0)
self.info_window.noutrefresh()
# Restore cursor to input window at current position
self.input_window.move(0, self.cursor_pos)
self.input_window.noutrefresh()
curses.doupdate()