Compare commits

..
33 Commits
Author SHA1 Message Date
DebianandSisyphus ec64804fec Infer kill event teams without relying on the livedamage plugin
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 14:39:21 +02:00
DebianandSisyphus c1b5f6d86a Populate team rosters from a LIVESTATS team token and tag the damage list
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 14:26:06 +02:00
DebianandSisyphus cbeab1ae45 Detect team modes by GAME_TYPE code and sync gametype at match start
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 14:19:37 +02:00
DebianandSisyphus 016384ef12 Display responses to cvars the user queries from the console
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 13:57:19 +02:00
DebianandSisyphus f8a71e7906 Render locally what rcon say broadcasts print on the server
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 13:32:49 +02:00
DebianandSisyphus c636557849 Sync rosters at match start, harden dedup and TIME handling, add scoreboard and damage summary
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 13:04:34 +02:00
DebianandSisyphus 43758f980c Run the roster seeder after cvar handling so credential responses survive
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 12:43:19 +02:00
DebianandSisyphus c9487b954a Seed the player roster from a status dump on connect and map change
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 12:33:59 +02:00
DebianandSisyphus c5e0bc3f01 Track roster connects and disconnects from the stats stream
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 12:30:12 +02:00
DebianandSisyphus b69cc38cce Consume every LIVESTATS heartbeat variant and show them all only with -v
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 12:26:42 +02:00
DebianandSisyphus 17dd04857f Announce the winner of aborted matches too, only restarts stay silent
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 12:20:46 +02:00
DebianandSisyphus ca1666f3a5 Ignore docs build artifacts
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 12:19:29 +02:00
DebianandSisyphus be7c3aec56 Track vote subjects and authoritative player scores in game state
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 12:19:29 +02:00
DebianandSisyphus 38f1446cde Announce forfeited match winners and show LIVESTATS lines only with -v
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-18 12:19:23 +02:00
DebianandSisyphus c08db27b57 Suppress all status debris in the window instead of shape-matching chunks
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-16 23:36:53 +02:00
DebianandSisyphus 73e503e863 Hide status output chunks behind a suppression window opened by the status echo
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-16 23:31:46 +02:00
DebianandSisyphus 3bc84ba3f2 Catch status rows in all arrival modes: wrapped, colored and split fragments
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-16 23:24:46 +02:00
DebianandSisyphus d188c6a506 Match RCON output shapes on color-stripped text so status rows hide correctly
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-16 23:07:41 +02:00
DebianandSisyphus 8eccfefc75 Show client id and endpoint on connect/disconnect, gate command echoes behind -v
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-16 22:04:47 +02:00
DebianandSisyphus e834b44ba6 Hide RCON command output but keep command echoes from other clients
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-16 21:47:25 +02:00
DebianandSisyphus 9ea4aa122a Make server name learning opt-in to avoid lagging the game server
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-16 21:35:10 +02:00
DebianandSisyphus 190152733e Add timestamps to game events and connection messages
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-16 21:31:31 +02:00
DebianandSisyphus b8864ca8dd Stop double-formatting stats stream events
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-16 21:27:28 +02:00
DebianandSisyphus b641eb4e45 Fix scroll-hold drift, ignore input while too small and redraw scrolls immediately
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-16 21:27:28 +02:00
DebianandSisyphus 71bd740d78 Bound listing capture and accept all server count-line formats
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-16 21:27:27 +02:00
pfl 3d1dc19d9b Learn cvar and command names from the server for autocomplete
Send cvarlist and cmdlist on connect and merge the answer into the
autocomplete candidates. Number the server menu from 0, keep suggesting
sibling commands while the command name is typed, mask the password in
the startup line, ignore venv/, fix README claims.
2026-09-16 18:25:48 +02:00
pfl db123993c0 Rework TUI: single-threaded curses, scrollback buffer, server menu
- All curses drawing on the main thread; stdin polled together with the
  ZMQ sockets, Ctrl-C only sets a flag
- Output kept in a 2000-line buffer and redrawn from it, so resizes keep
  the history; PgUp/PgDn and mouse wheel scroll it
- Messages without trailing newline continue the current line (status
  table columns arrive as separate messages)
- Too-small terminal shows a warning instead of exiting
- Start without arguments shows a server selection menu
- Colors on the terminal default background; version 0.9.0
2026-09-16 17:01:51 +02:00
DebianandSisyphus 0613d380a4 Add graceful shutdown, reconnect handling and config file integration
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-13 14:00:16 +02:00
DebianandSisyphus 3213bb38b4 Handle undersized terminals and wire autocomplete/history options
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-13 14:00:16 +02:00
DebianandSisyphus 271637ee71 Add config getters for logging, history and behavior settings
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-13 14:00:16 +02:00
DebianandSisyphus 5b90d66276 Remove dead code, fix cvar argument type and add player name helper
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-13 14:00:13 +02:00
DebianandSisyphus e47badb7b3 Harden JSON event parsing and broadcast message formatting
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-13 14:00:13 +02:00
DebianandSisyphus 15d461c234 Fix ZMQ socket monitor leak and guard monitor recv against races
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
2026-09-13 14:00:13 +02:00
13 changed files with 1652 additions and 725 deletions
+2
View File
@@ -1,4 +1,6 @@
__pycache__/
docs/
venv/
venv3/
*test*
*.json
+53 -30
View File
@@ -34,18 +34,29 @@ duel = 10.13.12.93:28961
## Usage
```bash
qlpycon # pick a server from a menu
qlpycon ffa # connect by name
qlpycon --host tcp://10.13.12.93:28960 --password secret # connect directly
qlpycon --list # list configured servers
```
Without arguments, qlpycon shows the servers from `[servers]` numbered from 0 and connects to the one you select (Up/Down or j/k, Enter or the number key 0-9; q quits). If no servers are configured it connects to `[connection] host`.
**Keys in the console:**
- `Enter`: send the command
- `Tab`: cycle autocomplete suggestions
- `Up`/`Down`: command history
- `PgUp`/`PgDn` or mouse wheel: scroll the output back and forth (the last 2000 lines are kept)
- Hold `Shift` while selecting text with the mouse (the wheel is reported to qlpycon)
- `Ctrl-C` twice: quit
**Options:**
- `--host URI` — ZMQ RCON endpoint
- `--password PASS` — RCON password (or set `QLPYCON_PASSWORD` env var)
- `--list` — list configured servers and exit
- `-v` / `-vv` — verbose (INFO) or debug (DEBUG) logging
- `--json FILE` — log all JSON events to file
- `--unknown-log FILE` — log unparsed events (default: unknown_events.log)
- `--host URI`: ZMQ RCON endpoint
- `--password PASS`: RCON password (or set `QLPYCON_PASSWORD` env var)
- `--list`: list configured servers and exit
- `-v` / `-vv`: verbose (INFO) or debug (DEBUG) logging
- `--json FILE`: log all JSON events to file
- `--unknown-log FILE`: log unparsed events (default: unknown_events.log)
## Features
@@ -53,24 +64,28 @@ qlpycon --list # list configured servers
- Team-aware colorized output with Quake color code support
- Powerup pickup and carrier kill notifications
- Server info panel (map, gametype, scores, players)
- Tab autocomplete for cvars and commands with fuzzy matching
- Argument suggestions for 25+ commands (bot names, maps, gametypes)
- Tab autocomplete for cvars and commands with fuzzy matching; on connect the
server's own `cvarlist` and `cmdlist` output is read so every cvar and
command the server knows is suggested, not only the built-in list
- Argument suggestions for 22 commands (bot names, maps, gametypes)
- Command history (↑/↓)
- Output scrollback (PgUp/PgDn) that survives terminal resizes
## Architecture
```
main.py — entry point, arg parsing, signal handling
qlpycon.conf — user configuration
main.py - entry point, arg parsing, signal handling
qlpycon.conf - user configuration
lib/
constants.py — weapons, teams, colors, limits
settings.py — config file loader
state.py — game state (server info, players, teams)
network.py — ZMQ connections (RCON DEALER, stats SUB)
parser.py — JSON event parsing
formatter.py — message formatting and colorization
ui.py — curses interface (info / output / input panels)
cvars.py — cvar/command database and autocomplete
constants.py - weapons, teams, colors, limits
settings.py - config file loader
state.py - game state (server info, players, teams)
network.py - ZMQ connections (RCON DEALER, stats SUB)
parser.py - JSON event parsing
formatter.py - message formatting and colorization
ui.py - curses interface (info / output / input panels)
cvars.py - built-in cvar/command database and autocomplete
namelist.py - learns cvar/command names from the server (cvarlist, cmdlist)
```
## License
@@ -163,7 +178,7 @@ $ addbot Sarge
↑ Shows skill levels (1-5)
$ addbot Sarge 4
red blue free spectator any
red blue free spectator r
↑ Shows team values
$ map
@@ -175,29 +190,35 @@ bloodrun
↑ Fuzzy matches 'blood' → 'bloodrun'
$ g_gametype
0 1 2 3 4 FFA Duel TDM CA CTF
↑ Shows numeric and string gametype values
ffa duel race tdm ca
↑ Shows gametype values (string names first, then the numbers 0-11)
$ callvote
map kick shuffle teamsize
map map_restart nextmap gametype kick
↑ Shows vote types
```
The system knows valid values for **25+ commands** including:
The system knows valid values for **22 commands** including:
- **32 bot names**: Sarge, Ranger, Visor, Xaero, Anarki, etc.
- **40+ maps**: bloodrun, campgrounds, toxicity, aerowalk, etc.
- **12 game types**: FFA, Duel, TDM, CA, CTF, etc. (numeric and string forms)
- **Vote types**: map, kick, shuffle, teamsize, g_gametype, etc.
- **Team values**: red, blue, free, spectator, any
- **38 maps**: bloodrun, campgrounds, toxicity, aerowalk, etc.
- **12 game types**: ffa, duel, tdm, ca, ctf, etc. (string and numeric forms)
- **Vote types**: map, kick, shuffle, teamsize, gametype, etc.
- **Team values**: red, blue, free, spectator (and r, b, f, s)
- **Skill levels**: 1-5 for bots
- **Boolean values**: 0, 1, true, false, enabled, disabled
- **Boolean values**: 0, 1
- **Common settings**: timelimits, fraglimits, sv_fps values, etc.
Arguments with freetext (like player names or custom messages) fall back to showing the signature with highlighting.
## Supported Commands
**76 cvars and commands included:**
Right after connecting, qlpycon sends `cvarlist` and `cmdlist` to the server
and adds every name from the answer to the autocomplete candidates (the
listing itself is hidden from the output). The built-in list below is the
fallback before that answer arrives and the source of signatures and argument
values.
**Built-in cvars and commands:**
### Commands with Signatures
@@ -263,7 +284,9 @@ autocomplete('stat')
### Add Your Own Cvars
Edit `cvars.py` and add to the appropriate list:
Names the server reports via `cvarlist`/`cmdlist` are picked up automatically.
To add a signature or argument values, or a name the server does not report,
edit `cvars.py` and add to the appropriate list:
```python
# Custom cvars
+13 -10
View File
@@ -5,7 +5,7 @@ Configuration and constants for QLPyCon
import re
VERSION = "0.8.1"
VERSION = "0.9.0"
# Pattern matching
COLOR_CODE_PATTERN = re.compile(r'\^\d') # Quake color codes (^0-^9)
@@ -18,13 +18,19 @@ 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
STATUS_OUTPUT_WINDOW = 1.5 # Seconds a 'status' command echo suppresses its chunked table output
STATUS_SEED_WINDOW = 2.0 # Seconds the roster seeder collects the 'status' dump after connect
CVAR_RESPONSE_TTL = 10.0 # Seconds a user-queried cvar response may arrive after the query
STATS_CONNECTION_DELAY = 0.5 # Initial stats connection delay
# UI dimensions
MIN_ROWS = 20
MIN_COLS = 80
INFO_WINDOW_HEIGHT = 12
INFO_WINDOW_Y = 2
OUTPUT_WINDOW_Y = 14
INPUT_WINDOW_HEIGHT = 2
OUTPUT_SCROLLBACK_LINES = 2000 # Output lines kept for PgUp/PgDn and redraw after resize
# Event deduplication
MAX_RECENT_EVENTS = 10
@@ -51,6 +57,12 @@ TEAM_MAP = {
3: 'SPECTATOR'
}
# GAME_TYPE codes from MATCH_STARTED that are team-based; matching on the code
# is immune to customized factory titles (g_factoryTitle)
TEAM_GAME_TYPE_CODES = {
'TDM', 'CA', 'CTF', '1FLAG', 'HARVESTER', 'FT', 'DOM', 'RR'
}
TEAM_COLORS = {
'RED': '^1(RED)^7',
'BLUE': '^4(BLUE)^7',
@@ -104,12 +116,3 @@ POWERUP_COLORS = {
'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)
}
+38 -6
View File
@@ -4,6 +4,8 @@ Quake Live console variables (cvars) database
Common cvars for autocomplete and fuzzy search
"""
import re
# Server configuration
SERVER_CVARS = [
'sv_hostname',
@@ -38,6 +40,14 @@ GAME_CVARS = [
'g_quadHog',
'g_training',
'g_instagib',
'g_password',
'g_needpass',
'g_overtime',
'g_startingHealth',
'g_startingArmor',
'g_infiniteAmmo',
'g_loadout',
'sv_privatePassword',
]
# Map and rotation
@@ -90,6 +100,8 @@ QLX_CVARS = [
'qlx_serverBrandName',
'qlx_owner',
'qlx_redditAuth',
'qlx_plugins',
'qlx_pluginsPath',
]
# Bot cvars
@@ -149,9 +161,21 @@ COMMANDS = [
'scores',
# Info commands
'serverinfo',
'systeminfo',
'players',
'maplist',
'configstrings',
'cvarlist',
'cmdlist',
'dumpuser',
# Console
'exec',
'set',
'seta',
'sets',
'echo',
'clientkick',
'cp',
]
# Bot names for addbot command (complete list)
@@ -269,6 +293,9 @@ COMMAND_SIGNATURES = {
'sv_maxclients': '<1-64>',
'sv_hostname': '<name>',
'sv_fps': '<20|30|40|60|125>',
'g_password': '<password>',
'g_needpass': '<0|1>',
'sv_privatePassword': '<password>',
# Network
'net_port': '<port number>',
@@ -329,7 +356,7 @@ COMMAND_ARGUMENTS = {
{'type': 'botname', 'required': True},
{'type': 'skill', 'required': False},
{'type': 'team', 'required': False},
{'type': 'msec delay number', 'required': False}, # msec delay
{'type': 'number', 'required': False}, # msec delay
{'type': 'freetext', 'required': False}, # altname
],
'removebot': [
@@ -525,12 +552,18 @@ def fuzzy_match(query, candidates, max_results=5):
return [match for match, score in matches[:max_results]]
def autocomplete(partial, max_results=5):
def autocomplete(partial, learned_names=(), max_results=5):
"""
Autocomplete a partial cvar/command
Returns list of suggestions
Autocomplete a partial cvar/command name.
Candidates are the built-in lists plus the names learned from the server
(see lib/namelist.py); a learned name that is already built in is skipped.
"""
return fuzzy_match(partial, ALL_CVARS, max_results)
candidates = list(ALL_CVARS)
builtin_lower = {name.lower() for name in ALL_CVARS}
for name in sorted(learned_names):
if name.lower() not in builtin_lower:
candidates.append(name)
return fuzzy_match(partial, candidates, max_results)
def parse_signature(signature):
@@ -541,7 +574,6 @@ def parse_signature(signature):
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)
+6 -3
View File
@@ -78,12 +78,15 @@ def format_message(message, add_timestamp=True):
# Handle broadcast messages
attributes = 0
if message[:10] == "broadcast:":
message = message[11:]
message = message[10:].lstrip()
attributes = 1 # Bold
# Handle print messages
if message[:7] == "print \"":
message = message[7:-2] + "\n"
message = message[7:]
if message.endswith("\""):
message = message[:-1]
message = message.rstrip() + "\n"
# Add timestamp if requested and appropriate
if add_timestamp and should_add_timestamp(message):
@@ -173,7 +176,7 @@ def format_powerup_message(message, player_tracker):
import time
if message.startswith("broadcast:"):
message = message[11:].strip()
message = message[10:].strip()
# Strip print " wrapper
if message.startswith('print "'):
+119
View File
@@ -0,0 +1,119 @@
#!/usr/bin/env python3
"""
Cvar and command names learned from the server, for autocomplete.
After connecting, main.py sends `cvarlist` and `cmdlist`. The server answers
with one line per name and a count line at the end:
cvarlist: S A sv_hostname "My Server" ... 1234 total cvars
cmdlist: addbot ... 321 commands
NameCapture collects the names from those lines and hides the listing from
the output window while it is running.
"""
import logging
import re
import time
logger = logging.getLogger('namelist')
# Capture stops when no listing output arrived for this many seconds.
# Guards against the count line looking different than expected, which would
# otherwise hide all later output forever. Measured from the last message, not
# from the request: a large cvarlist (5000+ messages) takes seconds to process.
CAPTURE_TIMEOUT = 5.0
# Absolute cap on a capture, measured from the request. The timeout above is
# refreshed by every incoming message, so regular server output can keep a
# stuck capture (count line never matches) alive indefinitely without this bound.
CAPTURE_MAX_DURATION = 30.0
# Cvar line: the name is the word directly before the quoted value.
# Example: 'S A sv_hostname "My Server"' -> sv_hostname
CVARLIST_LINE = re.compile(r'([A-Za-z_]\w*)\s+"')
# Command line: one bare word.
# Example: 'addbot' -> addbot
CMDLIST_LINE = re.compile(r'^\s*([A-Za-z_]\w*)\s*$')
# Count line that ends a listing.
# Examples: '1234 total cvars', '321 commands', '321 total commands'
END_LINE = re.compile(r'^\s*\d+\s+(total cvars|total commands|commands)\s*$')
def unwrap(message):
"""
Turn one RCON message into plain text with real newlines.
Example: 'print "addbot\\n"' -> 'addbot\n'
"""
text = message
if text.startswith('print "'):
text = text[7:]
if text.endswith('"'):
text = text[:-1]
return text.replace('\\n', '\n')
class NameCapture:
"""Collects cvar and command names from the server's cvarlist and cmdlist output"""
def __init__(self):
self.names = set()
self.listings_pending = 0 # count lines still expected; 0 = not capturing
self.deadline = 0
self.deadline_max = 0
self.partial_line = ''
def request(self, rcon):
"""Send cvarlist and cmdlist; their output is captured until both count lines arrived"""
self.listings_pending = 2
self.deadline = time.time() + CAPTURE_TIMEOUT
self.deadline_max = time.time() + CAPTURE_MAX_DURATION
self.partial_line = ''
rcon.send_command(b'cvarlist')
rcon.send_command(b'cmdlist')
logger.info('Requested cvarlist and cmdlist for autocomplete')
def handle(self, message):
"""
Feed one RCON message.
Returns True when the message belongs to a running listing (hide it from the output).
"""
if self.listings_pending == 0:
return False
now = time.time()
if now > self.deadline_max:
logger.warning(f'Capture exceeded {CAPTURE_MAX_DURATION:.0f}s, '
f'stopping capture with {len(self.names)} names')
self.listings_pending = 0
return False
if now > self.deadline:
logger.warning(f'No listing output for {CAPTURE_TIMEOUT:.0f}s, '
f'stopping capture with {len(self.names)} names')
self.listings_pending = 0
return False
self.deadline = now + CAPTURE_TIMEOUT
# A line can arrive in several messages (the server sends each flag column of
# a cvarlist row as its own 1-byte message), so keep the unfinished tail
text = self.partial_line + unwrap(message)
lines = text.split('\n')
self.partial_line = lines.pop()
for line in lines:
self._learn_line(line)
# The last count line may come without a trailing newline
if END_LINE.match(self.partial_line):
self._learn_line(self.partial_line)
self.partial_line = ''
return True
def _learn_line(self, line):
if END_LINE.match(line):
self.listings_pending -= 1
if self.listings_pending == 0:
logger.info(f'Learned {len(self.names)} cvar and command names from the server')
return
match = CVARLIST_LINE.search(line) or CMDLIST_LINE.match(line)
if match:
self.names.add(match.group(1))
+9 -8
View File
@@ -9,6 +9,8 @@ import struct
import logging
import time
from .constants import STATS_CONNECTION_DELAY
logger = logging.getLogger('network')
@@ -37,13 +39,13 @@ def check_monitor(monitor):
"""Check monitor socket for events"""
try:
event_monitor = monitor.recv(zmq.NOBLOCK)
event_id, event_name, event_value = read_socket_event(event_monitor)
event_endpoint = 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)
return (event_id, event_value, event_endpoint.decode('utf-8', 'replace'))
class RconConnection:
@@ -87,10 +89,6 @@ class RconConnection:
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:
@@ -104,6 +102,9 @@ class RconConnection:
def close(self):
"""Close connection"""
if self.monitor:
self.monitor.setsockopt(zmq.LINGER, 0) # Don't wait for unsent messages
self.monitor.close()
if self.socket:
self.socket.setsockopt(zmq.LINGER, 0) # Don't wait for unsent messages
self.socket.close()
@@ -143,7 +144,7 @@ class StatsConnection:
logger.debug('Setting ZMQ_SUBSCRIBE to empty (all messages)')
self.socket.setsockopt(zmq.SUBSCRIBE, b'')
time.sleep(0.5)
time.sleep(STATS_CONNECTION_DELAY)
self.connected = True
logger.info('Stats stream connected')
+183 -41
View File
@@ -32,6 +32,15 @@ class EventParser:
self.json_logger = json_logger
self.unknown_logger = unknown_logger
def _sync_match_time(self, data):
"""Sync the live match timer; a malformed TIME value must not discard
the whole event (the sync runs inside the handler try blocks)"""
try:
self.game_state.server_info.match_time = int(data['TIME'])
self.game_state.server_info.match_time_last_sync = time.time()
except (TypeError, ValueError):
logger.debug(f'Malformed TIME value: {data.get("TIME")!r}')
def parse_event(self, message):
"""
Parse JSON event and return formatted message string
@@ -65,14 +74,18 @@ class EventParser:
'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,
'PLAYER_CONNECT': self._handle_player_connect,
'PLAYER_DISCONNECT': self._handle_player_disconnect,
'ROUND_OVER': self._handle_round_over,
}
handler = handler_map.get(event_type)
if handler:
return handler(data)
formatted = handler(data)
if formatted:
timestamp = time.strftime('%H:%M:%S')
return f"^3[^7{timestamp}^3]^7 {formatted}"
return None
else:
# Unknown event
logger.debug(f'Unknown event type: {event_type}')
@@ -84,17 +97,38 @@ class EventParser:
except json.JSONDecodeError as e:
logger.debug(f'JSON decode error: {e}')
return None
except (KeyError, TypeError) as e:
except (KeyError, TypeError, ValueError) as e:
logger.debug(f'Error parsing event: {e}')
return None
def _handle_player_connect(self, data):
"""Track a connecting player so the roster is complete; connects to the
server land as spectators until they join a team — except reconnects of
already-tracked players, whose known team is kept. No feed line: the
RCON connect broadcast already displays one."""
name = data.get('NAME')
if not name:
return None
team = data.get('TEAM')
if team is None:
team = self.game_state.player_tracker.get_team(name) or 'SPECTATOR'
self.game_state.player_tracker.update_team(name, team)
self.game_state.player_tracker.add_player(name)
return None
def _handle_player_disconnect(self, data):
"""Remove a disconnecting player from the roster. No feed line: the
RCON disconnect broadcast already displays one."""
name = data.get('NAME')
if name:
self.game_state.player_tracker.remove_player(name)
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()
self._sync_match_time(data)
if 'KILLER' not in data:
return None
@@ -133,13 +167,39 @@ class EventParser:
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 _opposite_team(self, team):
"""The other active team in a two-team mode; None when not applicable"""
return {1: 2, 2: 1, 'RED': 'BLUE', 'BLUE': 'RED'}.get(team)
def _resolve_event_teams(self, data, killer, victim, killer_name, victim_name):
"""
Resolve (killer_team, victim_team) for a kill/death event. Event TEAM
fields win; when a server omits them (some do during warmup), the
missing side is derived from the counterpart player and the
TEAMKILL/SUICIDE flags. Solo modes always resolve to FREE — every
participant is FREE by definition. None means unknowable from this
event (leave the tracked team untouched).
"""
if not self.game_state.server_info.is_team_mode():
return ('FREE', 'FREE')
killer_team = killer.get('TEAM') if killer else None
victim_team = victim.get('TEAM') if victim else None
same_team = (bool(data.get('SUICIDE', False))
or strip_color_codes(killer_name).strip() == strip_color_codes(victim_name).strip()
or bool(data.get('TEAMKILL', False)))
if killer_team is not None and victim_team is None:
victim_team = killer_team if same_team else self._opposite_team(killer_team)
elif victim_team is not None and killer_team is None:
killer_team = victim_team if same_team else self._opposite_team(victim_team)
return (killer_team, victim_team)
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()
self._sync_match_time(data)
if 'VICTIM' not in data:
return None
@@ -158,6 +218,11 @@ class EventParser:
if 'TEAM' in victim:
self.game_state.player_tracker.update_team(victim_name, victim['TEAM'])
self.game_state.player_tracker.add_player(victim_name)
elif not self.game_state.server_info.is_team_mode():
# Solo modes: every participant is FREE by definition, even when
# the event carries no TEAM field
self.game_state.player_tracker.update_team(victim_name, 'FREE')
self.game_state.player_tracker.add_player(victim_name)
# Mark as dead
if not data.get('WARMUP', False):
@@ -167,10 +232,14 @@ class EventParser:
warmup = " ^8^3(Warmup)^0" if data.get('WARMUP', False) else ""
score_prefix = ""
# CA scores are damage-based server-side (kills + damage/100): frag
# deltas would corrupt them, so only PLAYER_STATS/LIVESTATS sync there
ca_scoring = self.game_state.server_info.gametype == 'Clan Arena'
# Environmental death (no killer)
if 'KILLER' not in data or not data['KILLER']:
# -1 for environmental death
if not data.get('WARMUP', False):
if not data.get('WARMUP', False) and not ca_scoring:
self.game_state.player_tracker.update_score(victim_name, -1)
score_prefix = "^8^1[-1]^7^0 "
@@ -188,17 +257,34 @@ class EventParser:
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)
# Team resolution with inference: prefer the event's TEAM fields; when
# a server omits them (some do during warmup), derive the missing side
# from the counterpart and the TEAMKILL/SUICIDE flags
killer_team, victim_team = self._resolve_event_teams(
data, killer, victim, killer_name, victim_name)
if killer_team is not None:
self.game_state.player_tracker.update_team(killer_name, killer_team)
if victim_team is not None:
self.game_state.player_tracker.update_team(victim_name, victim_team)
self.game_state.player_tracker.add_player(killer_name)
self.game_state.player_tracker.add_player(victim_name)
killer_prefix = get_team_prefix(killer_name, self.game_state.player_tracker)
# Suicide
if killer_name == victim_name:
# Suicide: names can differ cosmetically (color codes/whitespace) between
# the KILLER and VICTIM objects, so compare stripped. Steamid equality is
# only trusted for humans - bot ids are shared and would flag every
# bot-vs-bot frag as a suicide.
killer_steamid = str(killer.get('STEAM_ID', '') or '')
victim_steamid = str(victim.get('STEAM_ID', '') or '')
suicide = (strip_color_codes(killer_name).strip()
== strip_color_codes(victim_name).strip())
if (not suicide and killer_steamid.startswith('7656119')
and victim_steamid.startswith('7656119')):
suicide = killer_steamid == victim_steamid
if suicide:
# -1 for suicide
if not data.get('WARMUP', False):
if not data.get('WARMUP', False) and not ca_scoring:
self.game_state.player_tracker.update_score(victim_name, -1)
score_prefix = "^8^1[-1]^7^0 "
@@ -212,10 +298,9 @@ class EventParser:
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):
if not data.get('WARMUP', False) and not ca_scoring:
self.game_state.player_tracker.update_score(killer_name, 1)
score_prefix = "^8^2[+1]^7^0 "
@@ -244,9 +329,7 @@ class EventParser:
"""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()
self._sync_match_time(data)
team_won = data.get('TEAM_WON')
round_num = data.get('ROUND', 0)
@@ -266,9 +349,7 @@ class EventParser:
"""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()
self._sync_match_time(data)
name = data.get('NAME', 'Unknown')
medal = data.get('MEDAL', 'UNKNOWN')
@@ -301,18 +382,41 @@ class EventParser:
"""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()
self._sync_match_time(data)
# GAME_TYPE/FACTORY_TITLE are the authoritative gametype for this match;
# the title alone can be customized and defeat exact team-mode matching
self.game_state.server_info.game_type_code = data.get('GAME_TYPE', '')
if data.get('FACTORY_TITLE'):
self.game_state.server_info.gametype = data['FACTORY_TITLE']
# The PLAYERS array is the authoritative roster at match start: sync
# teams (numeric codes via TEAM_MAP) and reset scores, clear dead
# markers, and drop stale dedup signatures (TIME restarts at 0)
players = []
for player in data.get('PLAYERS', []):
name = player.get('NAME')
if not name:
continue
players.append(name)
team = player.get('TEAM')
if team is None:
team = 'SPECTATOR'
self.game_state.player_tracker.update_team(name, team)
self.game_state.player_tracker.add_player(name, score='0', ping='0')
self.game_state.player_tracker.set_score(name, '0')
# The array is the complete match roster: drop tracked players absent
# from it (they left before the match started)
roster_names = {strip_color_codes(name) for name in players}
for tracked in list(self.game_state.server_info.players.keys()):
if strip_color_codes(tracked) not in roster_names:
self.game_state.player_tracker.remove_player(tracked)
self.game_state.server_info.dead_players.clear()
self.game_state.event_deduplicator.reset()
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"
@@ -323,23 +427,54 @@ class EventParser:
"""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()
self._sync_match_time(data)
# Restarted matches (server shutdown mid-game) have no winner; aborts
# and forfeits still have final scores that name one
try:
if int(data.get('RESTARTED', 0) or 0) != 0:
return None
except (TypeError, ValueError):
return None
# Final standings from the per-player STATS array
stats_rows = []
for entry in data.get('STATS') or []:
if not isinstance(entry, dict):
continue
entry_name = entry.get('NAME')
if not entry_name:
continue
try:
entry_score = int(entry.get('SCORE', 0))
except (TypeError, ValueError):
entry_score = 0
stats_rows.append((strip_color_codes(entry_name), entry_score,
entry.get('KILLS', '?'), entry.get('DEATHS', '?')))
stats_rows.sort(key=lambda x: x[1], reverse=True)
scoreboard = ''
if stats_rows:
formatted = ' '.join(f"{name} {score} ({kills}/{deaths})"
for name, score, kills, deaths in stats_rows[:5])
scoreboard = f"^8^1[SCORES]^7 {formatted}\n"
if not self.game_state.server_info.is_team_mode():
return None
return scoreboard or 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"
winner = 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"
winner = 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"
winner = f"{report_prefix} ^7The match is a TIE^0 with a score of ^8^1{red_score} ^0^7to ^8^4{blue_score}\n"
if scoreboard:
winner = f"{winner.rstrip(chr(10))}\n{scoreboard}"
return winner
def _handle_player_stats(self, data):
"""Handle PLAYER_STATS event"""
@@ -349,6 +484,13 @@ class EventParser:
kills = int(data.get('KILLS', '0'))
deaths = int(data.get('DEATHS', '0'))
# PLAYER_STATS SCORE is authoritative (the server's own counters) and
# in CA it is damage-based, so sync it instead of accumulating deltas
if data.get('SCORE') is not None:
self.game_state.player_tracker.set_score(name, data['SCORE'])
if 'TEAM' in data:
self.game_state.player_tracker.update_team(name, data['TEAM'])
weapon_data = data.get('WEAPONS', {})
accuracies = calculate_weapon_accuracies(weapon_data)
+31 -3
View File
@@ -72,6 +72,18 @@ class ConfigLoader:
return value.lower() in ('true', 'yes', '1', 'on')
def get_float(self, section, key, fallback=0.0):
"""Get a float configuration value"""
value = self.get(section, key)
if value is None:
return fallback
try:
return float(value)
except ValueError:
logger.warning(f'Invalid float value for [{section}] {key}: {value}')
return fallback
def get_host(self):
"""Get connection host"""
return self.get('connection', 'host')
@@ -129,8 +141,11 @@ class ConfigLoader:
return password
def get_log_level(self):
"""Get logging level"""
level_str = self.get('logging', 'level', 'INFO')
"""Get logging level (None if not configured or invalid)"""
level_str = self.get('logging', 'level')
if not level_str:
return None
levels = {
'DEBUG': logging.DEBUG,
'INFO': logging.INFO,
@@ -138,7 +153,19 @@ class ConfigLoader:
'ERROR': logging.ERROR,
'CRITICAL': logging.CRITICAL
}
return levels.get(level_str.upper(), logging.INFO)
return levels.get(level_str.upper())
def get_max_history(self, default):
"""Get command history length from [ui] section"""
return self.get_int('ui', 'max_history', default)
def get_quit_timeout(self, default):
"""Get quit confirmation timeout from [behavior] section"""
return self.get_float('behavior', 'quit_timeout', default)
def get_respawn_delay(self, default):
"""Get respawn delay from [behavior] section"""
return self.get_float('behavior', 'respawn_delay', default)
def create_example_config():
@@ -146,6 +173,7 @@ def create_example_config():
config_content = """# qlpycon.conf
# Edit this file as needed.
#
# Pick from a menu: qlpycon
# Connect by server name: qlpycon ffa
# Connect directly: qlpycon --host tcp://1.2.3.4:28960 --password secret
# List servers: qlpycon --list
+91 -15
View File
@@ -5,7 +5,7 @@ Tracks server info, players, and teams
"""
import logging
from .constants import TEAM_MODES, TEAM_MAP, MAX_RECENT_EVENTS
from .constants import TEAM_MODES, TEAM_MAP, MAX_RECENT_EVENTS, TEAM_GAME_TYPE_CODES
from .formatter import strip_color_codes
logger = logging.getLogger('state')
@@ -18,12 +18,12 @@ class ServerInfo:
self.hostname = 'Unknown'
self.map = 'Unknown'
self.gametype = 'Unknown'
self.game_type_code = ''
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
@@ -37,8 +37,9 @@ class ServerInfo:
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
"""Check if current gametype is a team mode, by GAME_TYPE code from
MATCH_STARTED or by factory title"""
return self.gametype in TEAM_MODES or self.game_type_code in TEAM_GAME_TYPE_CODES
def reset_round_scores(self):
"""Reset round scores (for new matches)"""
@@ -102,14 +103,25 @@ class PlayerTracker:
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}')
"""Add player to server's player dict if not exists. A seeded clean-name
entry is upgraded to the color-coded name (preserving score/ping) instead
of duplicating the roster entry."""
if name in self.server_info.players:
return
clean_name = strip_color_codes(name)
for player_name in list(self.server_info.players.keys()):
if player_name != name and strip_color_codes(player_name) == clean_name:
existing = self.server_info.players.pop(player_name)
existing.setdefault('score', score)
existing.setdefault('ping', ping)
self.server_info.players[name] = existing
logger.debug(f'Upgraded roster entry {player_name} -> {name}')
return
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"""
@@ -121,6 +133,42 @@ class PlayerTracker:
teams[team].append(name)
return teams
def get_player_names(self):
"""Get clean (color-code-stripped) names of all tracked players"""
return [strip_color_codes(name) for name in self.server_info.players]
def set_score(self, name, score):
"""Set a player's authoritative score (from PLAYER_STATS / LIVESTATS)"""
clean_name = strip_color_codes(name)
if name in self.server_info.players:
self.server_info.players[name]['score'] = str(score)
return True
for player_name in self.server_info.players:
if strip_color_codes(player_name) == clean_name:
self.server_info.players[player_name]['score'] = str(score)
return True
return False
def update_livestats(self, clean_name, score, dealt, taken, ping, team=None):
"""Update live score/damage/ping from a LIVESTATS console print (clean name).
With a team token (extended livedamage format) the team is authoritative —
it sorts tracked players and adds unknown ones to the roster, which is
what populates teams correctly during warmup."""
for player_name, player_data in self.server_info.players.items():
if strip_color_codes(player_name) == clean_name:
player_data['score'] = str(score)
player_data['ping'] = str(ping)
player_data['dealt'] = str(dealt)
player_data['taken'] = str(taken)
if team is not None:
self.update_team(player_name, team)
return True
if team is None:
return False
self.add_player(clean_name, score=str(score), ping=str(ping))
self.update_team(clean_name, team)
return True
def remove_player(self, name):
"""Remove player from tracking"""
clean_name = strip_color_codes(name)
@@ -206,7 +254,12 @@ class EventDeduplicator:
if event_type not in ('PLAYER_DEATH', 'PLAYER_KILL'):
return False
signature = f"KILL:{time_val}:{killer_name}:{victim_name}"
# Signatures use stripped names: the double-emitted PLAYER_KILL and
# PLAYER_DEATH for one frag can carry cosmetically different name
# strings (color codes/whitespace), which would defeat the match
signature = (f"KILL:{time_val}:"
f"{strip_color_codes(killer_name).strip()}:"
f"{strip_color_codes(victim_name).strip()}")
if signature in self.recent_events:
logger.debug(f'Duplicate event: {signature}')
@@ -219,6 +272,30 @@ class EventDeduplicator:
return False
def reset(self):
"""Clear the signature buffer (match start: TIME restarts at 0)"""
self.recent_events = []
class VoteTracker:
"""Tracks cast subjects of the running vote; reset on each vote call"""
def __init__(self):
self.subjects = []
def on_call(self):
self.subjects = []
def on_cast(self, subject):
self.subjects.append(subject)
def result_annotation(self):
"""The cast subject when every cast agreed, else None (multi-option votes:
the winner is unknowable from broadcasts)"""
if self.subjects and all(s == self.subjects[0] for s in self.subjects):
return self.subjects[0]
return None
class GameState:
"""Main game state container"""
@@ -227,5 +304,4 @@ class GameState:
self.server_info = ServerInfo()
self.player_tracker = PlayerTracker(self.server_info)
self.event_deduplicator = EventDeduplicator()
self.pending_background_commands = set()
self.status_buffer = []
self.vote_tracker = VoteTracker()
+662 -540
View File
File diff suppressed because it is too large Load Diff
+440 -69
View File
@@ -14,15 +14,15 @@ import zmq
import signal
import sys
import os
import threading
from lib.constants import VERSION, DEFAULT_HOST, POLL_TIMEOUT, QUIT_CONFIRM_TIMEOUT, RESPAWN_DELAY, MAX_COMMAND_HISTORY
from lib.constants import VERSION, DEFAULT_HOST, POLL_TIMEOUT, QUIT_CONFIRM_TIMEOUT, RESPAWN_DELAY, MAX_COMMAND_HISTORY, STATUS_OUTPUT_WINDOW, STATUS_SEED_WINDOW, CVAR_RESPONSE_TTL
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
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:"([^"]*)"')
@@ -34,6 +34,26 @@ 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+([A-Za-z]+|\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')
CVAR_QUERY_PATTERN = re.compile(r'^[A-Za-z_][A-Za-z0-9_]*$')
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')
@@ -45,53 +65,275 @@ all_json_logger.setLevel(logging.DEBUG)
unknown_json_logger = logging.getLogger('unknown_json')
unknown_json_logger.setLevel(logging.DEBUG)
# Global flag for quit confirmation (thread-safe)
# 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
quit_confirm_lock = threading.Lock()
# Configurable timeouts (set from config in __main__)
_QUIT_CONFIRM_TIMEOUT = QUIT_CONFIRM_TIMEOUT
_RESPAWN_DELAY = RESPAWN_DELAY
# Cvars the user queried from the console (lowercased name -> request time):
# their responses display verbatim; everything else stays suppressed
expected_cvar_responses = {}
def consume_expected_cvar(cvar_name):
"""True when the user queried this cvar from the console recently"""
requested_at = expected_cvar_responses.pop(cvar_name.lower(), None)
return requested_at is not None and (time.time() - requested_at) < CVAR_RESPONSE_TTL
# Global shutdown flag (set by handle_ctrl_c, checked by main_loop)
shutdown_requested = False
def signal_handler(sig, frame):
"""Handle Ctrl+C with confirmation"""
global quit_confirm_time
"""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()
with quit_confirm_lock:
if quit_confirm_time is None or (current_time - quit_confirm_time) > QUIT_CONFIRM_TIMEOUT:
# First Ctrl-C or timeout expired
logger.warning(f"^1^8Press Ctrl-C again within {QUIT_CONFIRM_TIMEOUT:.0f} seconds to quit^0")
quit_confirm_time = current_time
else:
# Second Ctrl-C within timeout
logger.warning("^1^8Quittin'^0")
curses.endwin() # Restore terminal before force exit
os._exit(0)
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 server cvar responses: update state for known cvars, and display the
response when the user queried it from the console. Returns True when the
message is consumed (known-cvar update or a displayed query result).
"""
# Suppress command echo lines
if message.startswith('zmq RCON command') and 'from' in message:
suppress_cmds = ['status', 'roundlimit', 'qlx_serverBrandName', 'g_factoryTitle',
'mapname', 'timelimit', 'fraglimit', 'capturelimit', 'sv_maxclients']
if any(f': {cmd}' in message for cmd in suppress_cmds):
return True
match = CVAR_RESPONSE_PATTERN.search(message)
if not match:
return False
# 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)
cvar_name = match.group(1)
value = match.group(2)
if game_state.server_info.update_from_cvar(cvar_name, value):
ui.update_server_info(game_state)
return True
state_updated = game_state.server_info.update_from_cvar(cvar_name, value)
if state_updated:
ui.update_server_info(game_state)
displayed = consume_expected_cvar(cvar_name)
if displayed:
timestamp = time.strftime('%H:%M:%S')
ui.print_message(f"^3[^7{timestamp}^3]^7 {message.rstrip()}\n")
return state_updated or displayed
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 handle_stats_connection(message, rcon, ui, game_state):
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. An optional trailing
team token (extended livedamage format) is authoritative and populates
the roster with correct teams — including during warmup. 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, team = match.groups()
if team is not None and team.isdigit():
team = int(team)
game_state.player_tracker.update_livestats(
name, int(score), int(dealt), int(taken), int(ping), team)
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)
@@ -116,14 +358,13 @@ def handle_stats_connection(message, rcon, ui, game_state):
password_str = match.group(2)
password_str = strip_color_codes(password_str)
stats_password = password_str.strip()
logger.info(f'Got stats password: {stats_password}')
logger.info('Got stats password')
return stats_port, stats_password
def handle_user_input(input_queue, rcon, ui):
"""Process user command input"""
while not input_queue.empty():
command = input_queue.get()
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
@@ -132,6 +373,18 @@ def handle_user_input(input_queue, rcon, ui):
rcon.send_command(command)
# The server does not echo rcon-originated says back on the console
# stream, so render locally what players see in-game
stripped = command.strip()
if stripped.lower().startswith('say '):
timestamp = time.strftime('%H:%M:%S')
ui.print_message(f"^3[^7{timestamp}^3]^7 Server: {stripped[4:]}\n")
# A bare cvar name queries its value; the response displays verbatim
# when it arrives (other clients' query output stays suppressed)
if CVAR_QUERY_PATTERN.match(stripped):
expected_cvar_responses[stripped.lower()] = time.time()
def handle_stats_stream(stats_conn, stats_check_counter, event_parser, ui, game_state):
"""Poll and process stats stream events"""
@@ -147,12 +400,10 @@ def handle_stats_stream(stats_conn, stats_check_counter, event_parser, ui, game_
if stats_msg:
logger.info(f'Stats event received ({len(stats_msg)} bytes)')
# Parse game event
# Parse game event (parse_event returns a display-ready string)
parsed = event_parser.parse_event(stats_msg)
if parsed:
# Format with timestamp before displaying
formatted_msg, attributes = format_message(parsed)
ui.print_message(formatted_msg)
ui.print_message(parsed)
ui.update_server_info(game_state)
return stats_check_counter
@@ -163,7 +414,7 @@ def handle_player_respawns(game_state, ui):
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:
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)
@@ -172,7 +423,7 @@ def handle_player_respawns(game_state, ui):
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 current_time - death_time >= _RESPAWN_DELAY
]
if players_to_revive:
for name in players_to_revive:
@@ -205,6 +456,12 @@ def parse_player_events(message, game_state, ui):
# 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:
@@ -300,14 +557,18 @@ def parse_player_events(message, game_state, ui):
return False
def main_loop(screen, args):
"""Main application loop"""
"""
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(logging.WARNING)
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:
@@ -321,32 +582,46 @@ def main_loop(screen, args):
unknown_json_logger.propagate = False
# Initialize components
ui = UIManager(screen, args.host)
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)
# Setup input queue
input_queue = ui.setup_input_queue()
# 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
ui.print_message(f"Connecting with host={args.host} password={args.password}\n")
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
# Shutdown flag
shutdown = False
status_suppress_until = 0.0
# Timer refresh tracking (update UI once per second)
last_ui_update = 0
@@ -366,23 +641,42 @@ def main_loop(screen, args):
# Main event loop with resource cleanup
try:
while not shutdown:
# Poll RCON socket
event = rcon.poll(POLL_TIMEOUT)
while not shutdown_requested:
# Wait for RCON data, a connection event, a stats event or a key press
ready = dict(poller.poll(POLL_TIMEOUT))
handle_ctrl_c()
# Check monitor for connection events
monitor_event = rcon.check_monitor()
if monitor_event and monitor_event[0] == zmq.EVENT_CONNECTED:
ui.print_message("Connected to server\n")
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')
ui.print_message("Requesting connection info...\n")
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(input_queue, rcon, ui)
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)
@@ -391,7 +685,7 @@ def main_loop(screen, args):
handle_player_respawns(game_state, ui)
# Process RCON messages
if event > 0:
if rcon.socket in ready:
logger.debug('Socket has data available')
msg_count = 0
@@ -414,7 +708,36 @@ def main_loop(screen, args):
if parse_player_events(message, game_state, ui):
continue
if '------- Game Initialization -------' in message or 'Game Initialization' in message:
# 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
# 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')
@@ -432,6 +755,10 @@ def main_loop(screen, args):
# 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
@@ -440,7 +767,7 @@ def main_loop(screen, args):
continue
# Check for stats connection info
port, password = handle_stats_connection(message, rcon, ui, game_state)
port, password = handle_stats_connection(message)
if port:
stats_port = port
if password:
@@ -449,13 +776,16 @@ def main_loop(screen, args):
# Connect to stats if we have both credentials
if stats_port and stats_password and stats_conn is None:
try:
ui.print_message("Connecting to stats stream...\n")
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)
ui.print_message("Stats stream connected - ready for game events\n")
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')
@@ -476,6 +806,27 @@ def main_loop(screen, args):
ui.print_message(f"^1[^7{timestamp}^1] Error: Stats connection failed: {e}^7\n")
logger.error(f'Stats connection failed: {e}')
# Seed the roster from the status dump after connect/map change.
# Must run after the cvar/credential handlers above (their
# responses share the window and are never table rows) and
# before the shape-based suppressor (rows must buffer, not
# vanish). Table chunks stay suppressed, real content flows.
if status_seeder.is_active():
if not is_real_content(message):
status_seeder.feed(message)
continue
# 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:
@@ -501,12 +852,13 @@ def main_loop(screen, args):
logger.debug(f'Filtered bot debug: {message[:50]}')
continue
# Check if it's a chat message
if ':' in message and not message.startswith(('print', 'broadcast', 'zmq')):
# 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, attributes = format_message(message)
formatted_msg, _ = format_message(message)
ui.print_message(formatted_msg)
# Update server info panel every second (for live timer display)
@@ -531,6 +883,10 @@ if __name__ == '__main__':
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,
@@ -551,6 +907,11 @@ if __name__ == '__main__':
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()
@@ -562,6 +923,14 @@ if __name__ == '__main__':
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)
@@ -571,10 +940,12 @@ if __name__ == '__main__':
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)
+5
View File
@@ -1,6 +1,7 @@
# qlpycon.conf
# Edit this file as needed.
#
# Pick from a menu: qlpycon
# Connect by server name: qlpycon ffa
# Connect directly: qlpycon --host tcp://1.2.3.4:28960 --password secret
# List servers: qlpycon --list
@@ -43,5 +44,9 @@ max_history = 10
[behavior]
# Seconds to confirm quit (press Ctrl-C twice within this window)
quit_timeout = 3.0
# Learn extra cvar/command names for autocomplete from the server.
# Sends cvarlist + cmdlist on connect: a large burst of messages that can
# lag the game server. Off by default; the built-in names always work.
learn_names = false
# Seconds before players respawn after death
respawn_delay = 3.0