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

salidu

Browse Source
This commit is contained in:
xbl
2026-01-09 13:17:30 +01:00
parent 8d8a729490
commit 7d5210214a
12 changed files with 1790 additions and 620 deletions
Download Patch File Download Diff File Expand all files Collapse all files

240
AUTOCOMPLETE.md Normal file
View File

@ -0,0 +1,240 @@
# Autocomplete Feature
## Overview
QLPyCon includes intelligent autocomplete for console variables (cvars) and commands with fuzzy matching support.
## Features
- **Real-time Suggestions**: Fuzzy matches appear below the input line as you type
- **Tab Cycling**: Press Tab to cycle through suggestions
- **Smart Argument Highlighting**: Current argument position highlighted in reverse video
- **Argument Value Suggestions**: Intelligent suggestions for command arguments (bot names, maps, gametypes, etc.)
- **Command Signatures**: Automatic display when typing commands with arguments
- **Fuzzy Matching**: Finds matches even with partial or misspelled input
- **Smart Scoring**: Best matches appear first (exact > prefix > substring > fuzzy)
## Usage
Type a partial command (2+ characters) and fuzzy matches appear below the prompt:
```
$ sv_m
sv_maxclients sv_minrate sv_maxrate sv_timeout sv_floodprotect
```
Press **Tab** to cycle through the suggestions:
```
$ sv_m [Tab] → sv_maxclients
$ sv_maxclients [Tab] → sv_minrate
$ sv_minrate [Tab] → sv_maxrate
$ sv_maxrate [Tab] → sv_timeout
$ sv_timeout [Tab] → sv_floodprotect
$ sv_floodprotect [Tab] → sv_maxclients (cycles back)
```
**Command signatures with argument highlighting:**
When you type a command that has arguments, the signature appears automatically with the **current argument highlighted**:
```
$ addbot
[<botname>] [skill 1-5] [team] [msec delay] [altname]
↑ highlighted (currently typing this)
$ addbot sarge
<botname> [[skill 1-5]] [team] [msec delay] [altname]
↑ highlighted (now typing skill level)
$ addbot sarge 5
<botname> [skill 1-5] [[team]] [msec delay] [altname]
↑ highlighted (now typing team)
$ kick
[<player>]
↑ highlighted
$ g_gametype
[<0=FFA 1=Duel 2=TDM 3=CA 4=CTF...>]
↑ highlighted
```
The highlighted argument (shown with `[[ ]]` above, displayed in reverse video) shows you **exactly what to type next**.
Matches update in real-time as you type or delete characters.
**Argument value suggestions:**
After typing a command with a space, the system suggests valid values for each argument with fuzzy matching:
```
$ addbot
Anarki Angel Biker Bitterman Bones
↑ Shows bot names
$ addbot sar
Sarge
↑ Fuzzy matches 'sar' → 'Sarge'
$ addbot Sarge
1 2 3 4 5
↑ Shows skill levels (1-5)
$ addbot Sarge 4
red blue free spectator any
↑ Shows team values
$ map
aerowalk almostlost arenagate asylum battleforged
↑ Shows map names
$ map blood
bloodrun
↑ Fuzzy matches 'blood' → 'bloodrun'
$ g_gametype
0 1 2 3 4 FFA Duel TDM CA CTF
↑ Shows numeric and string gametype values
$ callvote
map kick shuffle teamsize
↑ Shows vote types
```
The system knows valid values for **25+ 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
- **Skill levels**: 1-5 for bots
- **Boolean values**: 0, 1, true, false, enabled, disabled
- **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:**
### Commands with Signatures
The following commands show usage help when selected:
**Bot commands:**
- `addbot` - <botname> [skill 1-5] [team] [msec delay] [altname]
- `removebot` - <botname>
**Player management:**
- `kick` - <player>
- `kickban` - <player>
- `ban` - <player>
- `tempban` - <player> <seconds>
- `tell` - <player> <message>
**Game settings:**
- `g_gametype` - <0=FFA 1=Duel 2=TDM 3=CA 4=CTF...>
- `timelimit` - <minutes>
- `fraglimit` - <frags>
- `capturelimit` - <captures>
**Map & voting:**
- `map` - <mapname>
- `callvote` - <vote type> [args...]
- `say` - <message>
### Additional Cvars
**Server Configuration:**
- sv_hostname, sv_maxclients, sv_fps, sv_pure, etc.
**Network & ZMQ:**
- net_port, zmq_rcon_enable, zmq_stats_enable, etc.
**QLX (minqlx):**
- qlx_serverBrandName, qlx_owner, qlx_redditAuth
## Technical Details
### Fuzzy Matching Algorithm
Scoring system (higher = better match):
- **1000**: Exact match
- **500+**: Prefix match (prioritizes shorter results)
- **100-**: Substring match (earlier = better)
- **50**: Contains all characters in order
### Examples:
```python
autocomplete('sv_')
→ ['sv_fps', 'sv_pure', 'sv_maxrate', 'sv_minrate', 'sv_timeout']
autocomplete('time')
→ ['timelimit', 'sv_timeout']
autocomplete('stat')
→ ['status', 'zmq_stats_enable', 'zmq_stats_ip', 'zmq_stats_password', 'zmq_stats_port']
```
## Customization
### Add Your Own Cvars
Edit `cvars.py` and add to the appropriate list:
```python
# Custom cvars
CUSTOM_CVARS = [
'my_custom_cvar',
'another_cvar',
]
# Add to ALL_CVARS
ALL_CVARS = (
SERVER_CVARS +
GAME_CVARS +
# ...
CUSTOM_CVARS
)
```
### Adjust Behavior
In `ui.py`, modify:
```python
# Minimum characters before showing suggestions
if len(current_word) >= 2: # Change to 1 or 3
# Maximum suggestions displayed
suggestions = autocomplete(current_word, max_results=5) # Change limit
```
## Keyboard Shortcuts
| Key | Action |
|-----|--------|
| **Tab** | Cycle through autocomplete suggestions |
| **↑** | Previous command in history |
| **↓** | Next command in history |
| **←/→** | Move cursor |
| **Backspace** | Delete (updates suggestions) |
| **Enter** | Send command |
## Notes
- Autocomplete starts after typing 2+ characters
- Suggestions appear on the line below the prompt
- Up to 5 matches shown at once (best matches first)
- Suggestions update in real-time as you type
- History navigation (↑/↓) works normally
## Testing Autocomplete
```bash
# Run cvars module directly to test matching
python3 cvars.py
# Output shows test queries and results
```

146
README.md
View File

@ -1,119 +1,81 @@
# QLPyCon - Quake Live Python Console
A modular, refactored terminal-based client for monitoring and remote controlling Quake Live servers via ZMQ.
Terminal-based client for monitoring and controlling Quake Live servers via ZMQ.
### Features
## Features
- **Real-time game monitoring** - Watch kills, deaths, team switches, medals, and more
- **Server info display** - Shows hostname, map, gametype, limits, and player count
- **Team-aware chat** - Color-coded messages with team prefixes
- **Powerup tracking** - Formatted pickup and carrier kill messages
- **JSON event logging** - Capture all game events to file for analysis
- **Colorized output** - Quake color code support (^0-^7)
- Real-time game monitoring (kills, deaths, medals, team switches)
- Server info display (map, gametype, scores, player list)
- Team-aware colorized chat with location tracking
- Powerup pickup and carrier kill notifications
- JSON event capture for analysis
- Quake color code support (^0-^7)
- **Autocomplete for cvars/commands** with fuzzy matching
- **Intelligent argument suggestions** for 25+ commands (bot names, maps, gametypes)
### Module Structure
```
qlpycon/
├── __init__.py # Package initialization
├── main.py # Entry point and main loop
├── config.py # Constants and configuration
├── state.py # Game state management (ServerInfo, PlayerTracker, etc)
├── network.py # ZMQ connections (RCON and stats stream)
├── parser.py # JSON event parsing
├── formatter.py # Message formatting and colorization
└── ui.py # Curses interface (windows, display, input)
```
### Installation
## Installation
```bash
# Requirements
pip install pyzmq
# Run directly
python -m qlpycon.main --host tcp://127.0.0.1:27961 --password YOUR_PASSWORD
python main.py --host tcp://SERVER_IP:PORT --password RCON_PASSWORD
```
### Usage
## Usage
```bash
# Basic connection
# Basic
python main.py --host tcp://SERVER_IP:PORT --password RCON_PASSWORD
# Verbose mode (show all communications)
python main.py --host tcp://SERVER_IP:PORT --password PASS -v
# Verbose logging
python main.py --host tcp://SERVER_IP:PORT --password RCON_PASSWORD -v
# Debug mode (detailed logging)
python main.py --host tcp://SERVER_IP:PORT --password PASS -vv
# Capture all JSON events to file
python main.py --host tcp://SERVER_IP:PORT --password PASS --json events.log
# Custom unknown events log
python main.py --unknown-log my_unknown.log
# Capture JSON events
python main.py --host tcp://SERVER_IP:PORT --password RCON_PASSWORD --json events.log
```
### Command Line Options
**Options:**
- `--host URI` - ZMQ RCON endpoint (default: tcp://127.0.0.1:27961)
- `--password PASS` - RCON password (required)
- `-v` / `-vv` - Verbose (INFO) or debug (DEBUG) logging
- `--json FILE` - Log all events as JSON
- `--unknown-log FILE` - Log unparsed events (default: unknown_events.log)
- `--host` - ZMQ URI (default: tcp://127.0.0.1:27961)
- `--password` - RCON password
- `--identity` - Socket identity (random UUID by default)
- `-v, --verbose` - Increase verbosity (use -v for INFO, -vv for DEBUG)
- `--json FILE` - Log all JSON events to FILE
- `--unknown-log FILE` - Log unknown JSON events (default: unknown_events.log)
**Configuration File (Optional):**
Create `~/.qlpycon.conf` or `./qlpycon.conf`:
```ini
[connection]
host = tcp://SERVER_IP:PORT
password = your_password
### Architecture Overview
[logging]
level = INFO
```
#### State Management (`state.py`)
- **ServerInfo** - Tracks server configuration and metadata
- **PlayerTracker** - Manages player teams and information
- **EventDeduplicator** - Prevents duplicate kill/death events
- **GameState** - Main container for all state
**Input Features:**
- **Smart autocomplete** - Type commands and see arguments highlighted in real-time
- **Argument value suggestions** - Intelligent suggestions for bot names, maps, gametypes, teams, etc.
- **Tab** - Cycle through autocomplete suggestions
- **↑/↓** - Command history navigation
- **Argument highlighting** - Current argument position shown in reverse video
- **Command signatures** - Automatic display (e.g., `addbot <botname> [skill 1-5] [team]`)
#### Network Layer (`network.py`)
- **RconConnection** - Handles DEALER socket for RCON commands
- **StatsConnection** - Handles SUB socket for game event stream
See [AUTOCOMPLETE.md](AUTOCOMPLETE.md) for details.
#### Event Parsing (`parser.py`)
- **EventParser** - Parses JSON game events into formatted messages
- Modular handlers for each event type (deaths, medals, team switches, etc)
## Architecture
#### Message Formatting (`formatter.py`)
- Color code handling (Quake's ^N system)
- Team prefix generation
- Timestamp logic
- Chat message formatting
```
main.py - Main loop, argument parsing, signal handling
config.py - Constants (weapons, teams, colors, limits)
state.py - Game state (ServerInfo, PlayerTracker, EventDeduplicator)
network.py - ZMQ connections (RCON DEALER, Stats SUB sockets)
parser.py - JSON event parsing (deaths, medals, switches, stats)
formatter.py - Message formatting, color codes, team prefixes
ui.py - Curses interface (3-panel: info, output, input)
```
#### UI Layer (`ui.py`)
- **UIManager** - Manages all curses windows
- Three-panel layout: server info, output, input
- Threaded input queue for non-blocking commands
- Color rendering with curses
**Supported Events:**
PLAYER_SWITCHTEAM, PLAYER_DEATH/KILL, PLAYER_MEDAL, PLAYER_STATS, MATCH_STARTED/REPORT, PLAYER_CONNECT/DISCONNECT, ROUND_OVER
### Event Types Supported
- `PLAYER_SWITCHTEAM` - Team changes
- `PLAYER_DEATH` / `PLAYER_KILL` - Frag events (deduplicated)
- `PLAYER_MEDAL` - Medal awards
- `PLAYER_STATS` - End-game statistics with weapon accuracy
- `MATCH_STARTED` - Match initialization
- `MATCH_REPORT` - Final scores
- `PLAYER_CONNECT` / `PLAYER_DISCONNECT` - Connection events
- `ROUND_OVER` - Round completion
### Color Codes
Quake Live uses `^N` color codes where N is 0-7:
- `^0` - Black
- `^1` - Red
- `^2` - Green
- `^3` - Yellow
- `^4` - Blue
- `^5` - Cyan
- `^6` - Magenta
- `^7` - White (default)
### License
## License
WTFPL

8
config.py
View File

@ -9,6 +9,11 @@ VERSION = "0.8.1"
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
@ -18,6 +23,9 @@ 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",

559
cvars.py Normal file
View File

@ -0,0 +1,559 @@
#!/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',
]
# 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}")

235
main.py
View File

@ -13,13 +13,26 @@ import curses
import zmq
import signal
import sys
import threading
from config import VERSION, DEFAULT_HOST, POLL_TIMEOUT
from config import VERSION, DEFAULT_HOST, POLL_TIMEOUT, QUIT_CONFIRM_TIMEOUT, RESPAWN_DELAY, MAX_COMMAND_HISTORY
from state import GameState
from network import RconConnection, StatsConnection
from parser import EventParser
from formatter import format_message, format_chat_message, format_powerup_message
from formatter import format_message, format_chat_message, format_powerup_message, strip_color_codes
from ui import UIManager
from qlpycon_config import ConfigLoader
# Pre-compiled regex patterns
CVAR_RESPONSE_PATTERN = re.compile(r'"([^"]+)"\s+is:"([^"]*)"')
PORT_PATTERN = re.compile(r'(\d+)')
BROADCAST_PATTERN = re.compile(r'^broadcast:\s*print\s*"(.+?)(?:\\n)?"\s*$')
TIMESTAMP_PATTERN = re.compile(r'\^\d\[\^\d[0-9:]+\^\d\]\^\d\s*|\[[0-9:]+\]\s*')
CONNECT_PATTERN = re.compile(r'^(.+?)\s+connected')
DISCONNECT_PATTERN = re.compile(r'^(.+?)\s+disconnected')
KICK_PATTERN = re.compile(r'^(.+?)\s+was kicked')
INACTIVITY_PATTERN = re.compile(r'^(.+?)\s+Dropped due to inactivity')
RENAME_PATTERN = re.compile(r'^(.+?)\s+renamed to\s+(.+?)$')
# Configure logging
logger = logging.getLogger('main')
@ -31,22 +44,23 @@ all_json_logger.setLevel(logging.DEBUG)
unknown_json_logger = logging.getLogger('unknown_json')
unknown_json_logger.setLevel(logging.DEBUG)
# Global flag for quit confirmation
# Global flag for quit confirmation (thread-safe)
quit_confirm_time = None
quit_confirm_lock = threading.Lock()
def signal_handler(sig, frame):
"""Handle Ctrl+C with confirmation"""
global quit_confirm_time
import time
current_time = time.time()
if quit_confirm_time is None or (current_time - quit_confirm_time) > 3:
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("^1^8Press Ctrl-C again within 3 seconds to quit^0")
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 3 seconds
# Second Ctrl-C within timeout
logger.warning("^1^8Quittin'^0")
sys.exit(0)
@ -63,7 +77,7 @@ def parse_cvar_response(message, game_state, ui):
return True
# Parse cvar responses (format: "cvar_name" is:"value" default:...)
cvar_match = re.search(r'"([^"]+)"\s+is:"([^"]*)"', message)
cvar_match = CVAR_RESPONSE_PATTERN.search(message)
if cvar_match:
cvar_name = cvar_match.group(1)
value = cvar_match.group(2)
@ -85,25 +99,85 @@ def handle_stats_connection(message, rcon, ui, game_state):
# Extract stats port
if 'net_port' in message and ' is:' in message and '"net_port"' in message:
match = re.search(r' is:"([^"]+)"', message)
match = CVAR_RESPONSE_PATTERN.search(message)
if match:
port_str = match.group(1).strip()
digit_match = re.search(r'(\d+)', port_str)
port_str = match.group(2).strip()
digit_match = PORT_PATTERN.search(port_str)
if digit_match:
stats_port = digit_match.group(1)
logger.info(f'Got stats port: {stats_port}')
# Extract stats password
if 'zmq_stats_password' in message and ' is:' in message and '"zmq_stats_password"' in message:
match = re.search(r' is:"([^"]+)"', message)
match = CVAR_RESPONSE_PATTERN.search(message)
if match:
password_str = match.group(1)
password_str = re.sub(r'\^\d', '', password_str) # Strip color codes
password_str = match.group(2)
password_str = strip_color_codes(password_str)
stats_password = password_str.strip()
logger.info(f'Got stats password: {stats_password}')
return stats_port, stats_password
def handle_user_input(input_queue, rcon, ui):
"""Process user command input"""
while not input_queue.empty():
command = input_queue.get()
logger.info(f'Sending command: {repr(command.strip())}')
# Display command with timestamp
timestamp = time.strftime('%H:%M:%S')
ui.print_message(f"^5[^7{timestamp}^5] >>> {command.strip()}^7\n")
rcon.send_command(command)
def handle_stats_stream(stats_conn, stats_check_counter, event_parser, ui, game_state):
"""Poll and process stats stream events"""
if not stats_conn or not stats_conn.connected:
return stats_check_counter
stats_check_counter += 1
if stats_check_counter % 100 == 0:
logger.debug(f'Stats polling active (check #{stats_check_counter // 100})')
stats_msg = stats_conn.recv_message()
if stats_msg:
logger.info(f'Stats event received ({len(stats_msg)} bytes)')
# Parse game event
parsed = event_parser.parse_event(stats_msg)
if parsed:
# Format with timestamp before displaying
formatted_msg, attributes = format_message(parsed)
ui.print_message(formatted_msg)
ui.update_server_info(game_state)
return stats_check_counter
def handle_player_respawns(game_state, ui):
"""Check and revive dead players after respawn delay"""
if game_state.server_info.gametype == 'Clan Arena':
# CA: revive all players after round end
if game_state.server_info.round_end_time:
if time.time() - game_state.server_info.round_end_time >= RESPAWN_DELAY:
game_state.server_info.dead_players.clear()
game_state.server_info.round_end_time = None
ui.update_server_info(game_state)
else:
# Other modes: revive individual players after death
current_time = time.time()
players_to_revive = [
name for name, death_time in game_state.server_info.dead_players.items()
if current_time - death_time >= RESPAWN_DELAY
]
if players_to_revive:
for name in players_to_revive:
del game_state.server_info.dead_players[name]
ui.update_server_info(game_state)
def parse_player_events(message, game_state, ui):
"""
Parse connect, disconnect, kick, and rename messages
@ -113,13 +187,12 @@ def parse_player_events(message, game_state, ui):
msg = message
# Strip broadcast: print "..." wrapper with regex
broadcast_match = re.match(r'^broadcast:\s*print\s*"(.+?)(?:\\n)?"\s*$', msg)
broadcast_match = BROADCAST_PATTERN.match(msg)
if broadcast_match:
msg = broadcast_match.group(1)
# Strip timestamp: [HH:MM:SS] or ^3[^7HH:MM:SS^3]^7
msg = re.sub(r'\^\d\[\^\d[0-9:]+\^\d\]\^\d\s*', '', msg)
msg = re.sub(r'\[[0-9:]+\]\s*', '', msg)
msg = TIMESTAMP_PATTERN.sub('', msg)
msg = msg.strip()
if not msg:
@ -128,15 +201,14 @@ def parse_player_events(message, game_state, ui):
logger.debug(f'parse_player_events: {repr(msg)}')
# Strip color codes for matching
from formatter import strip_color_codes
clean_msg = strip_color_codes(msg)
# Match connects: "NAME connected" or "NAME connected with Steam ID"
connect_match = re.match(r'^(.+?)\s+connected', clean_msg)
connect_match = CONNECT_PATTERN.match(clean_msg)
if connect_match:
player_name_match = re.match(r'^(.+?)\s+connected', msg)
player_name_match = CONNECT_PATTERN.match(msg)
player_name = player_name_match.group(1).strip() if player_name_match else connect_match.group(1).strip()
player_name = re.sub(r'\^\d+$', '', player_name)
player_name = strip_color_codes(player_name)
logger.info(f'CONNECT: {repr(player_name)}')
game_state.player_tracker.update_team(player_name, 'SPECTATOR')
@ -151,12 +223,11 @@ def parse_player_events(message, game_state, ui):
return True
# Regular disconnect
disconnect_match = re.match(r'^(.+?)\s+disconnected', clean_msg)
disconnect_match = DISCONNECT_PATTERN.match(clean_msg)
if disconnect_match:
original_match = re.match(r'^(.+?)\s+disconnected', msg)
original_match = DISCONNECT_PATTERN.match(msg)
player_name = original_match.group(1).strip() if original_match else disconnect_match.group(1).strip()
player_name = re.sub(r'\^\d+$', '', player_name)
player_name = re.sub(r'^\^\d+', '', player_name)
player_name = strip_color_codes(player_name)
logger.info(f'DISCONNECT: {repr(player_name)}')
game_state.player_tracker.remove_player(player_name)
@ -167,12 +238,11 @@ def parse_player_events(message, game_state, ui):
return True
# Kick
kick_match = re.match(r'^(.+?)\s+was kicked', clean_msg)
kick_match = KICK_PATTERN.match(clean_msg)
if kick_match:
original_match = re.match(r'^(.+?)\s+was kicked', msg)
original_match = KICK_PATTERN.match(msg)
player_name = original_match.group(1).strip() if original_match else kick_match.group(1).strip()
player_name = re.sub(r'\^\d+$', '', player_name)
player_name = re.sub(r'^\^\d+', '', player_name)
player_name = strip_color_codes(player_name)
logger.info(f'KICK: {repr(player_name)}')
game_state.player_tracker.remove_player(player_name)
@ -183,12 +253,11 @@ def parse_player_events(message, game_state, ui):
return True
# Inactivity
inactivity_match = re.match(r'^(.+?)\s+Dropped due to inactivity', clean_msg)
inactivity_match = INACTIVITY_PATTERN.match(clean_msg)
if inactivity_match:
original_match = re.match(r'^(.+?)\s+Dropped due to inactivity', msg)
original_match = INACTIVITY_PATTERN.match(msg)
player_name = original_match.group(1).strip() if original_match else inactivity_match.group(1).strip()
player_name = re.sub(r'\^\d+$', '', player_name)
player_name = re.sub(r'^\^\d+', '', player_name)
player_name = strip_color_codes(player_name)
logger.info(f'INACTIVITY DROP: {repr(player_name)}')
game_state.player_tracker.remove_player(player_name)
@ -199,10 +268,10 @@ def parse_player_events(message, game_state, ui):
return True
# Match renames: "OldName renamed to NewName"
rename_match = re.match(r'^(.+?)\s+renamed to\s+(.+?)$', clean_msg)
rename_match = RENAME_PATTERN.match(clean_msg)
if rename_match:
# Extract from original message
original_match = re.match(r'^(.+?)\s+renamed to\s+(.+?)$', msg)
original_match = RENAME_PATTERN.match(msg)
if original_match:
old_name = original_match.group(1).strip()
new_name = original_match.group(2).strip()
@ -210,10 +279,9 @@ def parse_player_events(message, game_state, ui):
old_name = rename_match.group(1).strip()
new_name = rename_match.group(2).strip()
# Remove trailing color codes from both names
old_name = re.sub(r'\^\d+$', '', old_name)
new_name = re.sub(r'^\^\d+', '', new_name) # Remove leading ^7 from new name
new_name = re.sub(r'\^\d+$', '', new_name) # Remove trailing too
# Remove color codes from both names
old_name = strip_color_codes(old_name)
new_name = strip_color_codes(new_name)
old_name = old_name.rstrip('\n\r') # Remove trailing newline
new_name = new_name.rstrip('\n\r') # Remove trailing newline
@ -235,14 +303,24 @@ def main_loop(screen):
# Setup signal handler for Ctrl+C with confirmation
signal.signal(signal.SIGINT, signal_handler)
# Parse arguments
# Load configuration file (optional)
config = ConfigLoader()
config.load()
# Parse arguments (command line overrides config file)
parser = argparse.ArgumentParser(description='Verbose QuakeLive server statistics')
parser.add_argument('--host', default=DEFAULT_HOST, help=f'ZMQ URI to connect to. Defaults to {DEFAULT_HOST}')
parser.add_argument('--password', required=False, help='RCON password')
parser.add_argument('--identity', default=uuid.uuid1().hex, help='Socket identity (random UUID by default)')
parser.add_argument('-v', '--verbose', action='count', default=0, help='Increase verbosity (-v INFO, -vv DEBUG)')
parser.add_argument('--unknown-log', default='unknown_events.log', help='File to log unknown JSON events')
parser.add_argument('-j', '--json', dest='json_log', default=None, help='File to log all JSON events')
parser.add_argument('--host', default=config.get_host() or DEFAULT_HOST,
help=f'ZMQ URI to connect to. Defaults to {DEFAULT_HOST}')
parser.add_argument('--password', default=config.get_password(), required=False,
help='RCON password')
parser.add_argument('--identity', default=uuid.uuid1().hex,
help='Socket identity (random UUID by default)')
parser.add_argument('-v', '--verbose', action='count', default=0,
help='Increase verbosity (-v INFO, -vv DEBUG)')
parser.add_argument('--unknown-log', default='unknown_events.log',
help='File to log unknown JSON events')
parser.add_argument('-j', '--json', dest='json_log', default=None,
help='File to log all JSON events')
args = parser.parse_args()
# Set logging level
@ -300,7 +378,8 @@ def main_loop(screen):
# Create event parser
event_parser = EventParser(game_state, json_logger, unknown_json_logger)
# Main event loop
# Main event loop with resource cleanup
try:
while not shutdown:
# Poll RCON socket
event = rcon.poll(POLL_TIMEOUT)
@ -317,54 +396,13 @@ def main_loop(screen):
rcon.send_command(b'net_port')
# Handle user input
while not input_queue.empty():
command = input_queue.get()
logger.info(f'Sending command: {repr(command.strip())}')
# Display command with timestamp
timestamp = time.strftime('%H:%M:%S')
ui.print_message(f"^5[^7{timestamp}^5] >>> {command.strip()}^7\n")
rcon.send_command(command)
handle_user_input(input_queue, rcon, ui)
# Poll stats stream if connected
if stats_conn and stats_conn.connected:
stats_check_counter += 1
if stats_check_counter % 100 == 0:
logger.debug(f'Stats polling active (check #{stats_check_counter // 100})')
stats_msg = stats_conn.recv_message()
if stats_msg:
logger.info(f'Stats event received ({len(stats_msg)} bytes)')
# Parse game event
parsed = event_parser.parse_event(stats_msg)
if parsed:
# Format with timestamp before displaying
formatted_msg, attributes = format_message(parsed)
ui.print_message(formatted_msg)
ui.update_server_info(game_state)
stats_check_counter = handle_stats_stream(stats_conn, stats_check_counter, event_parser, ui, game_state)
# Check if we need to revive players
if game_state.server_info.gametype == 'Clan Arena':
# CA: revive all players 3s after round end
if game_state.server_info.round_end_time:
if time.time() - game_state.server_info.round_end_time >= 3.0:
game_state.server_info.dead_players.clear()
game_state.server_info.round_end_time = None
ui.update_server_info(game_state)
else:
# Other modes: revive individual players 3s after death
current_time = time.time()
players_to_revive = [
name for name, death_time in game_state.server_info.dead_players.items()
if current_time - death_time >= 3.0
]
if players_to_revive:
for name in players_to_revive:
del game_state.server_info.dead_players[name]
ui.update_server_info(game_state)
handle_player_respawns(game_state, ui)
# Process RCON messages
if event > 0:
@ -405,8 +443,8 @@ def main_loop(screen):
rcon.send_command(b'capturelimit')
rcon.send_command(b'sv_maxclients')
# Clear player list since map changed
game_state.server_info.players = []
# Clear player dict since map changed
game_state.server_info.players = {}
game_state.player_tracker.player_teams = {}
ui.update_server_info(game_state)
@ -485,5 +523,16 @@ def main_loop(screen):
formatted_msg, attributes = format_message(message)
ui.print_message(formatted_msg)
finally:
# Clean up resources
logger.info("Shutting down...")
if rcon:
logger.debug("Closing RCON connection")
rcon.close()
if stats_conn:
logger.debug("Closing stats connection")
stats_conn.close()
logger.info("Shutdown complete")
if __name__ == '__main__':
curses.wrapper(main_loop)

2
parser.py
View File

@ -223,7 +223,7 @@ class EventParser:
weapon = killer.get('WEAPON', 'UNKNOWN')
weapon_name = WEAPON_KILL_NAMES.get(weapon, f'the {weapon}')
hp_left = killer.get('HEALTH', '0 HP')
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"

16
qlpycon.bash
View File

@ -1,10 +1,20 @@
#/usr/bin/env bash
#!/usr/bin/env bash
#
# Helper script to connect to different Quake Live servers
# Set QLPYCON_PASSWORD environment variable or create ~/.qlpycon.conf
workdir="/home/xbl/gits/qlpycon"
password="$(cat $workdir/rconpw.txt)"
workdir="/home/xbl/gitz/qlpycon"
password="${QLPYCON_PASSWORD:-}"
serverip="10.13.12.93"
# Check if password is set
if [ -z "$password" ]; then
echo "Error: QLPYCON_PASSWORD environment variable not set"
echo "Set it with: export QLPYCON_PASSWORD='your_password'"
echo "Or create ~/.qlpycon.conf with [connection] section"
exit 1
fi
cd "$workdir"
source venv/bin/activate

147
qlpycon_config.py Normal file
View File

@ -0,0 +1,147 @@
#!/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
# Support environment variable substitution: ${VAR_NAME}
if password.startswith('${') and password.endswith('}'):
env_var = password[2:-1]
return os.environ.get(env_var)
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 Configuration File
# Place this file as ~/.qlpycon.conf or ./qlpycon.conf
[connection]
# Server connection settings
host = tcp://10.13.12.93:28969
# Use ${ENV_VAR} to read from environment
password = ${QLPYCON_PASSWORD}
[logging]
# Log level: DEBUG, INFO, WARNING, ERROR, CRITICAL
level = INFO
[ui]
# Max command history entries
max_history = 10
# Color scheme (future feature)
color_scheme = quake
[behavior]
# Quit confirmation timeout (seconds)
quit_timeout = 3.0
# Player respawn delay (seconds)
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()

87
state.py
View File

@ -28,7 +28,7 @@ class ServerInfo:
self.red_rounds = 0
self.blue_score = 0
self.blue_rounds = 0
self.players = []
self.players = {} # Changed to dict: {name: {'score': str, 'ping': str}}
self.last_update = 0
self.warmup = False
self.dead_players = {}
@ -101,26 +101,19 @@ class PlayerTracker:
return self.player_teams.get(name)
def add_player(self, name, score='0', ping='0'):
"""Add player to server's player list if not exists"""
clean_name = re.sub(r'\^\d', '', name)
# Check if player already exists (by either name or clean name)
for existing in self.server_info.players:
existing_clean = re.sub(r'\^\d', '', existing['name'])
if existing['name'] == name or existing_clean == clean_name:
return # Already exists
self.server_info.players.append({
'name': name,
"""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 player in self.server_info.players:
name = player['name']
for name in self.server_info.players.keys():
team = self.player_teams.get(name, 'FREE')
if team not in teams:
team = 'FREE'
@ -131,39 +124,47 @@ class PlayerTracker:
"""Remove player from tracking"""
clean_name = re.sub(r'\^\d', '', name)
# Count before removal
before_count = len(self.server_info.players)
# Try to remove by exact name first
removed = self.server_info.players.pop(name, None)
# Remove from player list - check both original and clean names
self.server_info.players = [
p for p in self.server_info.players
if p['name'] != name and re.sub(r'\^\d', '', p['name']) != clean_name
]
# Log if anything was actually removed
after_count = len(self.server_info.players)
if before_count != after_count:
logger.info(f'Removed player: {name} (clean: {clean_name}) - {before_count} -> {after_count}')
# If not found, try to find by clean name
if not removed:
for player_name in list(self.server_info.players.keys()):
if re.sub(r'\^\d', '', 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 - both versions
# 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"""
"""Rename a player while maintaining their team and score"""
old_clean = re.sub(r'\^\d', '', old_name)
# Get current team (try both names)
team = self.player_teams.get(old_name) or self.player_teams.get(old_clean, 'SPECTATOR')
# Find and update player in server list
for player in self.server_info.players:
if player['name'] == old_name or re.sub(r'\^\d', '', player['name']) == old_clean:
player['name'] = new_name
# 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 re.sub(r'\^\d', '', 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)
@ -175,14 +176,20 @@ class PlayerTracker:
def update_score(self, name, delta):
"""Update player's score by delta (+1 for kill, -1 for death/suicide)"""
clean_name = re.sub(r'\^\d', '', name)
# 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
for player in self.server_info.players:
player_clean = re.sub(r'\^\d', '', player['name'])
if player['name'] == name or player_clean == clean_name:
current_score = int(player.get('score', 0))
player['score'] = str(current_score + delta)
logger.debug(f"Score update: {name} {delta:+d} -> {player['score']}")
# Fallback: search by clean name (rare case)
clean_name = re.sub(r'\^\d', '', name)
for player_name, player_data in self.server_info.players.items():
if re.sub(r'\^\d', '', 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")

286
ui.py
View File

@ -9,7 +9,8 @@ import curses.textpad
import threading
import queue
import logging
from config import COLOR_PAIRS, INFO_WINDOW_HEIGHT, INFO_WINDOW_Y, OUTPUT_WINDOW_Y, INPUT_WINDOW_HEIGHT, TEAM_MODES
from config 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')
@ -35,7 +36,7 @@ class CursesHandler(logging.Handler):
curses.doupdate()
except (KeyboardInterrupt, SystemExit):
raise
except:
except Exception:
self.handleError(record)
def print_colored(window, message, attributes=0):
@ -75,6 +76,116 @@ def print_colored(window, message, attributes=0):
else:
window.addch(ch, curses.color_pair(color) | (curses.A_BOLD if bold else 0) | (curses.A_UNDERLINE if underline else 0) | attributes)
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:
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:
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:
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:
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:
pass
return suggestions, suggestion_index, original_word
class UIManager:
"""Manages curses windows and display"""
@ -100,7 +211,7 @@ class UIManager:
curses.start_color()
curses.use_default_colors()
curses.cbreak()
curses.curs_set(0)
curses.curs_set(1) # Show cursor in input window
self.screen.addstr(f"Quake Live PyCon: {self.host}")
self.screen.noutrefresh()
@ -174,7 +285,7 @@ class UIManager:
curses.doupdate()
def setup_input_queue(self):
"""Setup threaded input queue with command history"""
"""Setup threaded input queue with command history and autocomplete"""
def wait_stdin(q, window, manager):
current_input = ""
cursor_pos = 0
@ -182,6 +293,11 @@ class UIManager:
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()
@ -189,12 +305,82 @@ class UIManager:
if key == -1: # No input
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)
# 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:
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) > 10:
if len(manager.command_history) > MAX_COMMAND_HISTORY:
manager.command_history.pop(0)
q.put(current_input)
@ -202,6 +388,9 @@ class UIManager:
cursor_pos = 0
temp_history_index = -1
temp_input = ""
suggestions = []
suggestion_index = -1
original_word = ""
window.erase()
window.noutrefresh()
@ -217,6 +406,9 @@ class UIManager:
temp_history_index -= 1
current_input = manager.command_history[temp_history_index]
cursor_pos = len(current_input)
suggestions = []
suggestion_index = -1
original_word = ""
window.erase()
window.addstr(0, 0, current_input)
window.noutrefresh()
@ -234,6 +426,9 @@ class UIManager:
current_input = manager.command_history[temp_history_index]
cursor_pos = len(current_input)
suggestions = []
suggestion_index = -1
original_word = ""
window.erase()
window.addstr(0, 0, current_input)
window.noutrefresh()
@ -258,10 +453,23 @@ class UIManager:
current_input = current_input[:cursor_pos-1] + current_input[cursor_pos:]
cursor_pos -= 1
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:
@ -269,14 +477,28 @@ class UIManager:
current_input = current_input[:cursor_pos] + char + current_input[cursor_pos:]
cursor_pos += 1
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:
import logging
logging.getLogger('ui').error(f'Input error: {e}')
# Log but continue - input thread should stay alive
window.move(0, cursor_pos)
curses.doupdate()
@ -313,8 +535,7 @@ class UIManager:
hostname = server_info.hostname
timer_display = ""
if not server_info.warmup:
#if server_info.match_time > 0 and not server_info.warmup:
if server_info.match_time > 0 and not server_info.warmup:
mins = server_info.match_time // 60
secs = server_info.match_time % 60
timer_display = f"^3^0Time:^8^7 {mins}:{secs:02d}^0"
@ -365,10 +586,9 @@ class UIManager:
else:
red_total = 0
blue_total = 0
for player in server_info.players:
player_name = player['name']
for player_name, player_data in server_info.players.items():
team = game_state.player_tracker.get_team(player_name)
score = int(player.get('score', 0))
score = int(player_data.get('score', 0))
if team == 'RED':
red_total += score
@ -386,19 +606,11 @@ class UIManager:
spec_players = []
for player_name in teams['RED']:
score = 0
for player in server_info.players:
if player['name'] == player_name:
score = int(player.get('score', 0))
break
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 = 0
for player in server_info.players:
if player['name'] == player_name:
score = int(player.get('score', 0))
break
score = int(server_info.players.get(player_name, {}).get('score', 0))
blue_players_with_scores.append((player_name, score))
# Sort by score descending
@ -414,18 +626,8 @@ class UIManager:
blue_name = blue_players[i] if i < len(blue_players) else ''
# Get scores for team players
red_score = ''
blue_score = ''
if red_name:
for player in server_info.players:
if player['name'] == red_name:
red_score = player.get('score', '0')
break
if blue_name:
for player in server_info.players:
if player['name'] == blue_name:
blue_score = player.get('score', '0')
break
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
@ -449,11 +651,7 @@ class UIManager:
free_players = teams['FREE']
free_players_with_scores = []
for player_name in free_players:
score = 0
for player in server_info.players:
if player['name'] == player_name:
score = int(player.get('score', 0))
break
score = int(server_info.players.get(player_name, {}).get('score', 0))
free_players_with_scores.append((player_name, score))
# Sort by score descending
@ -469,18 +667,8 @@ class UIManager:
col2_name = free_col2[i] if i < len(free_col2) else ''
# Get scores for FREE players
col1_score = ''
col2_score = ''
if col1_name:
for player in server_info.players:
if player['name'] == col1_name:
col1_score = player.get('score', '0')
break
if col2_name:
for player in server_info.players:
if player['name'] == col2_name:
col2_score = player.get('score', '0')
break
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