mirror of
https://github.com/onyx-and-iris/q3rcon-cli.git
synced 2026-03-23 18:29:18 +00:00
Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 60afda97dc | |||
| 968cd2dc8a | |||
| d7ea4d9fad | |||
| d0519bb8e9 |
@ -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'
|
||||||
|
|||||||
@ -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():
|
||||||
|
|||||||
@ -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)
|
||||||
|
|||||||
@ -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'
|
||||||
|
)
|
||||||
|
|||||||
@ -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')
|
||||||
|
|||||||
@ -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'
|
||||||
|
)
|
||||||
|
|||||||
@ -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)
|
||||||
|
|||||||
@ -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)
|
||||||
|
|||||||
@ -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)
|
||||||
|
|||||||
@ -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)
|
||||||
|
|||||||
@ -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)
|
||||||
|
|||||||
@ -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')
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user