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

Add timestamps to game events and connection messages

Browse Source 
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <[email protected]>
This commit is contained in:
Debian
2026-09-16 21:31:31 +02:00
co-authored by Sisyphus
parent b8864ca8dd
commit 190152733e
2 changed files with 15 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/parser.py
+5 -1
View File
@@ -72,7 +72,11 @@ class EventParser:
handler = handler_map.get(event_type) handler = handler_map.get(event_type)
if handler: 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: else:
# Unknown event # Unknown event
logger.debug(f'Unknown event type: {event_type}') logger.debug(f'Unknown event type: {event_type}')
main.py
+10 -5
View File
@@ -361,7 +361,8 @@ def main_loop(screen, args):
# Initialize network connections # Initialize network connections
password_display = '***' if args.password else '(none)' password_display = '***' if args.password else '(none)'
ui.print_message(f"Connecting with host={args.host} password={password_display}\n") 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 = RconConnection(args.host, args.password, args.identity)
rcon.connect() rcon.connect()
@@ -403,11 +404,13 @@ def main_loop(screen, args):
# Check monitor for connection events # Check monitor for connection events
monitor_event = rcon.check_monitor() monitor_event = rcon.check_monitor()
if monitor_event and monitor_event[0] == zmq.EVENT_CONNECTED: 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\n")
rcon.send_command(b'register') rcon.send_command(b'register')
logger.info('Registration message sent') 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'zmq_stats_password')
rcon.send_command(b'net_port') rcon.send_command(b'net_port')
name_capture.request(rcon) name_capture.request(rcon)
@@ -491,14 +494,16 @@ def main_loop(screen, args):
# Connect to stats if we have both credentials # Connect to stats if we have both credentials
if stats_port and stats_password and stats_conn is None: if stats_port and stats_password and stats_conn is None:
try: 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] host_ip = args.host.split('//')[1].split(':')[0]
stats_conn = StatsConnection(host_ip, stats_port, stats_password) stats_conn = StatsConnection(host_ip, stats_port, stats_password)
stats_conn.connect() stats_conn.connect()
poller.register(stats_conn.socket, zmq.POLLIN) 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 # Request initial server info
logger.info('Sending initial server info queries') logger.info('Sending initial server info queries')