onyx_online/q3rcon-cli
1
0
Fork 0
mirror of https://github.com/onyx-and-iris/q3rcon-cli.git synced 2026-03-23 18:29:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

updates to the Console classes:

Browse Source 
- added ErrConsole
- import console as namespace throughout the package
- swap out the static methods for instance methods.

Wrap the entry point in try/except.
Note, catching IncorrectPasswordError is currently not working. See https://github.com/Iapetus-11/aio-q3-rcon/issues/4
This commit is contained in:
onyx-and-iris 2026-03-22 04:10:53 +00:00
parent 9a4fefcbbb
commit bfb45efb32
11 changed files with 64 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
src/q3rcon_cli/cli.py
View File

@ -1,8 +1,9 @@
import clypi
from aioq3rcon import Client
from aioq3rcon import Client, IncorrectPasswordError
from clypi import Command, arg
from typing_extensions import override
from . import console
from .commands import (
Fastrestart,
Gametype,
@ -14,7 +15,6 @@ from .commands import (
Plugins,
Status,
)
from .console import Console
Subcommands = (
Fastrestart
@ -96,16 +96,20 @@ class Q3rconCli(Command):
) as client:
try:
if response := await client.send_command(command):
Console.print_response(response)
console.out.print_response(response)
except TimeoutError:
print(
clypi.style(
'Timeout waiting for response for command:', fg='red'
),
clypi.style(command, fg='yellow'),
console.err.print(
f"Timeout waiting for response for command: '{command}'"
)
def main():
cli = Q3rconCli().parse()
cli.start()
try:
cli = Q3rconCli().parse()
cli.start()
except IncorrectPasswordError:
console.err.print('Incorrect password provided.')
except TimeoutError:
console.err.print(
f"Timeout waiting for response for command: '{type(cli.subcommand).__name__.lower()}'"
)

4
src/q3rcon_cli/commands/fastrestart.py
View File

@ -2,7 +2,7 @@ from aioq3rcon import Client
from clypi import Command, arg
from typing_extensions import override
from q3rcon_cli.console import Console
from q3rcon_cli import console
class Fastrestart(Command):
@ -16,4 +16,4 @@ class Fastrestart(Command):
async def run(self):
async with Client(self.host, self.port, self.password) as client:
if response := await client.send_command('fast_restart'):
Console.print_response(response)
console.out.print_response(response)

6
src/q3rcon_cli/commands/gametype.py
View File

@ -2,7 +2,7 @@ from aioq3rcon import Client
from clypi import Command, Positional, Spinner, arg
from typing_extensions import override
from q3rcon_cli.console import Console
from q3rcon_cli import console
class Gametype(Command):
@ -26,7 +26,7 @@ class Gametype(Command):
if not Gametype.new_gametype:
async with Client(self.host, self.port, self.password) as client:
if response := await client.send_command('g_gametype'):
Console.print_cvar(response)
console.out.print_cvar(response)
return
async with Client(self.host, self.port, self.password) as client:
@ -40,4 +40,4 @@ class Gametype(Command):
client.fragment_read_timeout = DEFAULT_FRAGMENT_READ_TIMEOUT
if response := await client.send_command('g_gametype'):
Console.print_cvar(response)
console.out.print_cvar(response)

6
src/q3rcon_cli/commands/hostname.py
View File

@ -2,7 +2,7 @@ from aioq3rcon import Client
from clypi import Command, Positional, arg
from typing_extensions import override
from q3rcon_cli.console import Console
from q3rcon_cli import console
class Hostname(Command):
@ -21,10 +21,10 @@ class Hostname(Command):
if not self.new_hostname:
async with Client(self.host, self.port, self.password) as client:
if response := await client.send_command('sv_hostname'):
Console.print_cvar(response)
console.out.print_cvar(response)
return
async with Client(self.host, self.port, self.password) as client:
await client.send_command(f'sv_hostname {self.new_hostname}')
if response := await client.send_command('sv_hostname'):
Console.print_cvar(response)
console.out.print_cvar(response)

6
src/q3rcon_cli/commands/map.py
View File

@ -2,7 +2,7 @@ from aioq3rcon import Client
from clypi import Command, Positional, Spinner, arg
from typing_extensions import override
from q3rcon_cli.console import Console
from q3rcon_cli import console
class Map(Command):
@ -21,7 +21,7 @@ class Map(Command):
if not self.new_map:
async with Client(self.host, self.port, self.password) as client:
if response := await client.send_command('mapname'):
Console.print_cvar(response)
console.out.print_cvar(response)
return
async with Spinner('Changing map...'):
@ -30,4 +30,4 @@ class Map(Command):
) as client:
await client.send_command(f'map mp_{self.new_map.removeprefix("mp_")}')
if response := await client.send_command('mapname'):
Console.print_cvar(response)
console.out.print_cvar(response)

4
src/q3rcon_cli/commands/mapname.py
View File

@ -2,7 +2,7 @@ from aioq3rcon import Client
from clypi import Command, arg
from typing_extensions import override
from q3rcon_cli.console import Console
from q3rcon_cli import console
class Mapname(Command):
@ -16,4 +16,4 @@ class Mapname(Command):
async def run(self):
async with Client(self.host, self.port, self.password) as client:
if response := await client.send_command('mapname'):
Console.print_cvar(response)
console.out.print_cvar(response)

4
src/q3rcon_cli/commands/maprestart.py
View File

@ -2,7 +2,7 @@ from aioq3rcon import Client
from clypi import Command, Spinner, arg
from typing_extensions import override
from q3rcon_cli.console import Console
from q3rcon_cli import console
class Maprestart(Command):
@ -19,4 +19,4 @@ class Maprestart(Command):
self.host, self.port, self.password, fragment_read_timeout=1
) as client:
if response := await client.send_command('map_restart'):
Console.print_response(response)
console.out.print_response(response)

4
src/q3rcon_cli/commands/maprotate.py
View File

@ -2,7 +2,7 @@ from aioq3rcon import Client
from clypi import Command, Spinner, arg
from typing_extensions import override
from q3rcon_cli.console import Console
from q3rcon_cli import console
class Maprotate(Command):
@ -19,4 +19,4 @@ class Maprotate(Command):
self.host, self.port, self.password, fragment_read_timeout=1
) as client:
if response := await client.send_command('map_rotate'):
Console.print_response(response)
console.out.print_response(response)

4
src/q3rcon_cli/commands/plugins.py
View File

@ -2,7 +2,7 @@ from aioq3rcon import Client
from clypi import Command, arg
from typing_extensions import override
from q3rcon_cli.console import Console
from q3rcon_cli import console
class Plugins(Command):
@ -16,4 +16,4 @@ class Plugins(Command):
async def run(self):
async with Client(self.host, self.port, self.password) as client:
if response := await client.send_command('plugins'):
Console.print_response(response)
console.out.print_response(response)

4
src/q3rcon_cli/commands/status.py
View File

@ -2,7 +2,7 @@ from aioq3rcon import Client
from clypi import Command, Spinner, arg
from typing_extensions import override
from q3rcon_cli.console import Console
from q3rcon_cli import console
class Status(Command):
@ -19,4 +19,4 @@ class Status(Command):
self.host, self.port, self.password, fragment_read_timeout=0.5
) as client:
if response := await client.send_command('status'):
Console.print_status(response)
console.out.print_status(response)

44
src/q3rcon_cli/console.py
View File

@ -5,6 +5,19 @@ from clypi import cprint
class Console:
def __init__(self, style: str = 'yellow'):
self.style = style
def print(self, message: str, style: str | None = None):
cprint(message, fg=style or self.style)
class ErrorConsole(Console):
def __init__(self):
super().__init__(style='red')
class OutConsole(Console):
COLOUR_CODE_REGEX = re.compile(r'\^[0-9]')
STATUS_PLAYER_REGEX = re.compile(
r'^\s*(?P<slot>[0-9]+)\s+'
@ -30,18 +43,16 @@ class Console:
)
@staticmethod
def remove_colour_codes(s: str) -> str:
def _remove_colour_codes(s: str) -> str:
"""Remove Quake 3 colour codes from a string."""
return Console.COLOUR_CODE_REGEX.sub('', s)
return OutConsole.COLOUR_CODE_REGEX.sub('', s)
@staticmethod
def print_response(response: str):
response = Console.remove_colour_codes(response).removeprefix('print\n')
def print_response(self, response: str):
response = self._remove_colour_codes(response).removeprefix('print\n')
cprint(f'\n{response}\n', fg='yellow')
cprint(f'\n{response}\n', fg=self.style)
@staticmethod
def print_status(response: str):
def print_status(self, response: str):
_slots = []
_scores = []
_pings = []
@ -51,16 +62,16 @@ class Console:
lines = response.splitlines()
for line in lines:
if m := Console.STATUS_PLAYER_REGEX.match(line):
if m := OutConsole.STATUS_PLAYER_REGEX.match(line):
_slots.append(m.group('slot'))
_scores.append(m.group('score'))
_pings.append(m.group('ping'))
_guids.append(m.group('guid'))
_names.append(m.group('name'))
_names.append(self._remove_colour_codes(m.group('name')))
_ips.append(m.group('ip'))
if not _slots:
cprint('\nNo players connected.\n', fg='yellow')
cprint('\nNo players connected.\n', fg=self.style)
return
slots = clypi.boxed(_slots, title='Slot', width=15)
@ -71,11 +82,10 @@ class Console:
ips = clypi.boxed(_ips, title='IP', width=30)
print(f'\n{clypi.stack(slots, scores, pings, guids, names, ips, padding=0)}')
@staticmethod
def print_cvar(response: str):
response = Console.remove_colour_codes(response).removeprefix('print\n')
def print_cvar(self, response: str):
response = self._remove_colour_codes(response).removeprefix('print\n')
if m := Console.CVAR_REGEX.match(response):
if m := self.CVAR_REGEX.match(response):
name = clypi.boxed(
[m.group('name')], title='Name', width=max(len(m.group('name')) + 4, 30)
)
@ -93,3 +103,7 @@ class Console:
[m.group('info')], title='Info', width=max(len(m.group('info')) + 4, 30)
)
print(f'\n{clypi.stack(name, value, default, info, padding=0)}')
out = OutConsole()
err = ErrorConsole()