Compare commits

...

4 Commits
v0.2.4 ... main

Author SHA1 Message Date
60afda97dc defaults are dealt with by CMD_CONFIG 2026-03-23 14:53:33 +00:00
968cd2dc8a rename TIMEOUTS to CMD_CONFIG 2026-03-23 14:48:09 +00:00
d7ea4d9fad patch bump 2026-03-23 14:45:50 +00:00
d0519bb8e9 add spinners for all commands (including interactive mode).
pass send_command(interpret=True) for commands that return long responses.
note, it strips the first `"` from cvar responses which causes CVAR_REGEX to fail.
2026-03-23 14:45:35 +00:00
12 changed files with 99 additions and 68 deletions

View File

@ -1,4 +1,4 @@
# SPDX-FileCopyrightText: 2026-present onyx-and-iris <code@onyxandiris.online> # SPDX-FileCopyrightText: 2026-present onyx-and-iris <code@onyxandiris.online>
# #
# SPDX-License-Identifier: MIT # SPDX-License-Identifier: MIT
__version__ = '0.2.4' __version__ = '0.2.5'

View File

@ -1,6 +1,6 @@
import clypi import clypi
from aioq3rcon import Client, IncorrectPasswordError from aioq3rcon import Client, IncorrectPasswordError
from clypi import Command, arg from clypi import Command, Spinner, arg
from typing_extensions import override from typing_extensions import override
from . import console from . import console
@ -72,35 +72,41 @@ class Q3rconCli(Command):
clypi.style('to quit.', fg='blue'), clypi.style('to quit.', fg='blue'),
) )
DEFAULT_TIMEOUT = 2
DEFAULT_FRAGMENT_READ_TIMEOUT = 0.25 DEFAULT_FRAGMENT_READ_TIMEOUT = 0.25
while command := input(clypi.style('cmd: ', fg='green')): while command := input(clypi.style('cmd: ', fg='green')):
if command.lower() == 'q': if command.lower() == 'q':
break break
fragment_read_timeout = None CMD_CONFIG = {
if command in ( 'status': (2, 1, False),
'status', 'fast_restart': (3, 1, True),
'fast_restart', 'map_restart': (3, 1, True),
'map_restart', 'map': (3, 1, True),
'map', 'map_rotate': (3, 1, True),
'map_rotate', }
): timeout, fragment_read_timeout, interpret = CMD_CONFIG.get(
fragment_read_timeout = 1 command.split()[0].lower(),
(DEFAULT_TIMEOUT, DEFAULT_FRAGMENT_READ_TIMEOUT, False),
)
async with Client( async with Spinner(f"Sending command: '{command}'", suffix='...'):
self.host, async with Client(
self.port, self.host,
self.password, self.port,
fragment_read_timeout=fragment_read_timeout self.password,
or DEFAULT_FRAGMENT_READ_TIMEOUT, timeout=timeout,
) as client: fragment_read_timeout=fragment_read_timeout,
try: ) as client:
if response := await client.send_command(command): try:
console.out.print_response(response) if response := await client.send_command(
except TimeoutError: command, interpret=interpret
console.err.print( ):
f"Timeout waiting for response for command: '{command}'" console.out.print_response(response)
) except TimeoutError:
console.err.print(
f"Timeout waiting for response for command: '{command}'"
)
def main(): def main():

View File

@ -1,5 +1,5 @@
from aioq3rcon import Client from aioq3rcon import Client
from clypi import Command, arg from clypi import Command, Spinner, arg
from typing_extensions import override from typing_extensions import override
from q3rcon_cli import console from q3rcon_cli import console
@ -14,6 +14,8 @@ class Fastrestart(Command):
@override @override
async def run(self): async def run(self):
async with Client(self.host, self.port, self.password) as client: async with Spinner('Executing fast restart', suffix='...'):
if response := await client.send_command('fast_restart'): async with Client(self.host, self.port, self.password) as client:
console.out.print_response(response) response = await client.send_command('fast_restart', interpret=True)
console.out.print_response(response)

View File

@ -1,5 +1,5 @@
from aioq3rcon import Client from aioq3rcon import Client
from clypi import Command, Positional, Spinner, arg, cprint from clypi import Command, Positional, Spinner, arg
from typing_extensions import override from typing_extensions import override
from q3rcon_cli import console from q3rcon_cli import console
@ -24,16 +24,23 @@ class Gametype(Command):
@override @override
async def run(self): async def run(self):
if not self.new_gametype: if not self.new_gametype:
async with Client(self.host, self.port, self.password) as client: async with Spinner('Fetching current gametype', suffix='...'):
if response := await client.send_command('g_gametype'): async with Client(self.host, self.port, self.password) as client:
console.out.print_cvar(response) response = await client.send_command('g_gametype')
console.out.print_cvar(response)
return return
async with Client(self.host, self.port, self.password) as client: async with Spinner(f'Changing gametype to {self.new_gametype}', suffix='...'):
await client.send_command(f'g_gametype {self.new_gametype}') async with Client(self.host, self.port, self.password) as client:
if self.force: await client.send_command(f'g_gametype {self.new_gametype}')
async with Spinner('Forcing gametype change', suffix='...'):
if self.force:
async with Spinner('Forcing gametype change', suffix='...'):
async with Client(self.host, self.port, self.password) as client:
client.timeout = 3
client.fragment_read_timeout = 1 client.fragment_read_timeout = 1
await client.send_command('map_restart') await client.send_command('map_restart')
cprint(f'Gametype changed successfully to {self.new_gametype}.', fg='green') console.out.print(
f'Gametype changed successfully to {self.new_gametype}.', style='green'
)

View File

@ -1,5 +1,5 @@
from aioq3rcon import Client from aioq3rcon import Client
from clypi import Command, Positional, arg, cprint from clypi import Command, Positional, Spinner, arg
from typing_extensions import override from typing_extensions import override
from q3rcon_cli import console from q3rcon_cli import console
@ -19,12 +19,14 @@ class Hostname(Command):
@override @override
async def run(self): async def run(self):
if not self.new_hostname: if not self.new_hostname:
async with Client(self.host, self.port, self.password) as client: async with Spinner('Fetching current hostname', suffix='...'):
if response := await client.send_command('sv_hostname'): async with Client(self.host, self.port, self.password) as client:
console.out.print_cvar(response) response = await client.send_command('sv_hostname')
console.out.print_cvar(response)
return return
async with Client(self.host, self.port, self.password) as client: async with Spinner(f'Changing hostname to {self.new_hostname}', suffix='...'):
await client.send_command(f'sv_hostname {self.new_hostname}') async with Client(self.host, self.port, self.password) as client:
await client.send_command(f'sv_hostname {self.new_hostname}')
cprint(f'Hostname changed to: {self.new_hostname}', fg='green') console.out.print(f'Hostname changed to: {self.new_hostname}', style='green')

View File

@ -1,5 +1,5 @@
from aioq3rcon import Client from aioq3rcon import Client
from clypi import Command, Positional, Spinner, arg, cprint from clypi import Command, Positional, Spinner, arg
from typing_extensions import override from typing_extensions import override
from q3rcon_cli import console from q3rcon_cli import console
@ -19,9 +19,10 @@ class Map(Command):
@override @override
async def run(self): async def run(self):
if not self.new_map: if not self.new_map:
async with Client(self.host, self.port, self.password) as client: async with Spinner('Getting current map', suffix='...'):
if response := await client.send_command('mapname'): async with Client(self.host, self.port, self.password) as client:
console.out.print_cvar(response) response = await client.send_command('mapname')
console.out.print_cvar(response)
return return
async with Spinner('Changing map', suffix='...'): async with Spinner('Changing map', suffix='...'):
@ -30,4 +31,6 @@ class Map(Command):
) as client: ) as client:
await client.send_command(f'map mp_{self.new_map.removeprefix("mp_")}') await client.send_command(f'map mp_{self.new_map.removeprefix("mp_")}')
cprint(f'Map changed to {self.new_map.removeprefix("mp_")}', fg='green') console.out.print(
f'Map changed to {self.new_map.removeprefix("mp_")}', style='green'
)

View File

@ -1,5 +1,5 @@
from aioq3rcon import Client from aioq3rcon import Client
from clypi import Command, arg from clypi import Command, Spinner, arg
from typing_extensions import override from typing_extensions import override
from q3rcon_cli import console from q3rcon_cli import console
@ -14,6 +14,8 @@ class Mapname(Command):
@override @override
async def run(self): async def run(self):
async with Client(self.host, self.port, self.password) as client: async with Spinner('Getting map name...'):
if response := await client.send_command('mapname'): async with Client(self.host, self.port, self.password) as client:
console.out.print_cvar(response) response = await client.send_command('mapname')
console.out.print_cvar(response)

View File

@ -16,7 +16,8 @@ class Maprestart(Command):
async def run(self): async def run(self):
async with Spinner('Restarting map', suffix='...'): async with Spinner('Restarting map', suffix='...'):
async with Client( async with Client(
self.host, self.port, self.password, fragment_read_timeout=1 self.host, self.port, self.password, timeout=3, fragment_read_timeout=1
) as client: ) as client:
if response := await client.send_command('map_restart'): response = await client.send_command('map_restart', interpret=True)
console.out.print_response(response)
console.out.print_response(response)

View File

@ -16,7 +16,12 @@ class Maprotate(Command):
async def run(self): async def run(self):
async with Spinner('Rotating map', suffix='...'): async with Spinner('Rotating map', suffix='...'):
async with Client( async with Client(
self.host, self.port, self.password, fragment_read_timeout=1 self.host,
self.port,
self.password,
timeout=3,
fragment_read_timeout=1,
) as client: ) as client:
if response := await client.send_command('map_rotate'): response = await client.send_command('map_rotate', interpret=True)
console.out.print_response(response)
console.out.print_response(response)

View File

@ -1,5 +1,5 @@
from aioq3rcon import Client from aioq3rcon import Client
from clypi import Command, arg from clypi import Command, Spinner, arg
from typing_extensions import override from typing_extensions import override
from q3rcon_cli import console from q3rcon_cli import console
@ -14,6 +14,8 @@ class Plugins(Command):
@override @override
async def run(self): async def run(self):
async with Client(self.host, self.port, self.password) as client: async with Spinner('Fetching plugins...'):
if response := await client.send_command('plugins'): async with Client(self.host, self.port, self.password) as client:
console.out.print_response(response) response = await client.send_command('plugins')
console.out.print_response(response)

View File

@ -16,7 +16,8 @@ class Status(Command):
async def run(self): async def run(self):
async with Spinner('Fetching status', suffix='...'): async with Spinner('Fetching status', suffix='...'):
async with Client( async with Client(
self.host, self.port, self.password, fragment_read_timeout=0.5 self.host, self.port, self.password, fragment_read_timeout=1
) as client: ) as client:
if response := await client.send_command('status'): response = await client.send_command('status')
console.out.print_status(response)
console.out.print_status(response)

View File

@ -72,7 +72,7 @@ class OutConsole(Console):
_names.append(self._remove_colour_codes(m.group('name'))) _names.append(self._remove_colour_codes(m.group('name')))
_ips.append(m.group('ip')) _ips.append(m.group('ip'))
elif m := OutConsole.STATUS_MAP_REGEX.match(line): elif m := OutConsole.STATUS_MAP_REGEX.match(line):
cprint(f'\nCurrent map: {m.group("mapname")}', fg=self.style) cprint(f'Current map: {m.group("mapname")}', fg=self.style)
if not _slots: if not _slots:
cprint('No players connected.', fg='red') cprint('No players connected.', fg='red')