Compare commits

..

No commits in common. "main" and "v0.1.0" have entirely different histories.
main ... v0.1.0

18 changed files with 141 additions and 337 deletions

2
.gitignore vendored
View File

@ -210,5 +210,3 @@ marimo/_lsp/
__marimo__/
# End of ignr
test-*.py

View File

@ -1,15 +0,0 @@
repos:
- repo: https://github.com/pre-commit/pre-commit-hooks
rev: v2.3.0
hooks:
- id: check-yaml
- id: end-of-file-fixer
- id: trailing-whitespace
- repo: local
hooks:
- id: update-readme
name: update-readme
entry: python tools/update_readme_help.py
language: python
pass_filenames: false

View File

@ -10,9 +10,6 @@
## Table of Contents
- [Installation](#installation)
- [Configuration](#configuration)
- [Use](#use)
- [Shell Completion](#shell-completion)
- [License](#license)
## Installation
@ -52,8 +49,8 @@ example .envrc:
```env
#!/usr/bin/env bash
export Q3RCON_CLI_HOST=localhost
export Q3RCON_CLI_PORT=28960
export Q3RCON_CLI_HOST="localhost"
export Q3RCON_CLI_PORT="28960"
export Q3RCON_CLI_PASSWORD="<rcon password>"
```
@ -62,36 +59,26 @@ export Q3RCON_CLI_PASSWORD="<rcon password>"
```console
Usage: q3rcon-cli [OPTIONS] COMMAND
┏━ Subcommands ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┏━ Subcommands ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ status Prints the status of the server ┃
┃ mapname Prints the current map name of the server ┃
┃ maprotate Rotates the map to the next one in the map rotation list ┃
┃ fastrestart Executes a fast restart of the server ┃
┃ gametype Get or set the current gametype of the server ┃
┃ hostname Get or set the current hostname of the server ┃
┃ map Get the current map or change to a new one ┃
┃ mapname Prints the current map name of the server ┃
┃ maprestart Restarts the current map ┃
┃ maprotate Rotates the map to the next one in the map rotation list ┃
┃ plugins Prints the currently loaded plugins of the server ┃
┃ status Prints the status of the server ┃
┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
┏━ Options ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
┏━ Options ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ -i, --interactive Whether to start in interactive mode (defaults to false) ┃
┃ -v, --version Show the version and exit ┃
┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
┏━ Connection options ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
┏━ Connection options ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ -h, --host <HOST> The host to connect to ┃
┃ -p, --port <PORT> The port to connect to ┃
┃ -P, --password <PASSWORD> The password for authentication ┃
┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
```
## Shell Completion
Shell completion scripts are available for *bash*, *zsh*, and *fish*.
```console
q3rcon-cli --install-autocomplete
┗━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┛
```
## Special Thanks

View File

@ -20,7 +20,7 @@ classifiers = [
"Programming Language :: Python :: Implementation :: CPython",
"Programming Language :: Python :: Implementation :: PyPy",
]
dependencies = ["aio-q3-rcon>=1.0.1", "clypi>=1.8.2"]
dependencies = ["aio-q3-rcon>=1.0.0", "clypi>=1.8.2"]
[project.scripts]
q3rcon-cli = "q3rcon_cli.cli:main"
@ -33,9 +33,6 @@ Source = "https://github.com/onyx-and-iris/q3rcon-cli"
[tool.hatch.version]
path = "src/q3rcon_cli/__about__.py"
[tool.hatch.envs.default]
workspace.members = [{ path = "../aio-q3-rcon" }]
[tool.hatch.envs.types]
extra-dependencies = ["mypy>=1.0.0"]
[tool.hatch.envs.types.scripts]

View File

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

View File

@ -1,33 +1,22 @@
import clypi
from aioq3rcon import Client, IncorrectPasswordError
from clypi import Command, Spinner, arg
from clypi import parsers as cp
from aioq3rcon import Client
from clypi import Command, arg
from typing_extensions import override
from . import console
from .__about__ import __version__
from .commands import (
Fastrestart,
Gametype,
Hostname,
Map,
Mapname,
Maprestart,
Maprotate,
Plugins,
Status,
)
from .console import Console
Subcommands = (
Fastrestart
| Gametype
| Hostname
| Map
| Mapname
| Maprestart
| Maprotate
| Plugins
| Status
Status | Mapname | Maprotate | Fastrestart | Gametype | Hostname | Map | Plugins
)
@ -39,7 +28,6 @@ class Q3rconCli(Command):
help='The host to connect to',
env='Q3RCON_CLI_HOST',
group='Connection',
parser=cp.Str(min=1),
)
port: int = arg(
28960,
@ -47,7 +35,6 @@ class Q3rconCli(Command):
help='The port to connect to',
env='Q3RCON_CLI_PORT',
group='Connection',
parser=cp.Int(min=1, max=65535),
)
password: str = arg(
'',
@ -55,82 +42,61 @@ class Q3rconCli(Command):
help='The password for authentication',
env='Q3RCON_CLI_PASSWORD',
group='Connection',
parser=cp.Str(min=8),
)
interactive: bool = arg(
False,
short='i',
help='Whether to start in interactive mode (defaults to false)',
)
version: bool = arg(
False,
short='v',
help='Show the version and exit',
)
@override
async def run(self):
if self.version:
print(f'q3rcon-cli version: {clypi.style(__version__, fg="green")}')
return
if self.interactive:
await self.run_interactive()
else:
await Status(self.host, self.port, self.password).run()
await Status.run(self)
async def run_interactive(self):
print(
clypi.style('Entering interactive mode. Type', fg='blue'),
clypi.style("'Q'", fg='yellow'),
clypi.style("'Q'", fg='red'),
clypi.style('to quit.', fg='blue'),
)
DEFAULT_TIMEOUT = 2
DEFAULT_FRAGMENT_READ_TIMEOUT = 0.25
while command := input(clypi.style('cmd: ', fg='green')):
if command.lower() == 'q':
break
CMD_CONFIG = {
'status': (2, 1, False),
'fast_restart': (3, 1, True),
'map_restart': (3, 1, True),
'map': (3, 1, True),
'map_rotate': (3, 1, True),
}
timeout, fragment_read_timeout, interpret = CMD_CONFIG.get(
command.split()[0].lower(),
(DEFAULT_TIMEOUT, DEFAULT_FRAGMENT_READ_TIMEOUT, False),
)
fragment_read_timeout = None
if command in (
'status',
'fast_restart',
'map_restart',
'map',
'map_rotate',
):
fragment_read_timeout = 1
async with Spinner(f"Sending command: '{command}'", suffix='...'):
async with Client(
self.host,
self.port,
self.password,
timeout=timeout,
fragment_read_timeout=fragment_read_timeout,
fragment_read_timeout=fragment_read_timeout
or DEFAULT_FRAGMENT_READ_TIMEOUT,
) as client:
try:
response = await client.send_command(
command, interpret=interpret
)
if response := await client.send_command(command):
Console.print_response(response)
except TimeoutError:
console.err.print(
f"Timeout waiting for response for command: '{command}'"
print(
clypi.style(
'Timeout waiting for response for command:', fg='red'
),
clypi.style(command, fg='yellow'),
)
console.out.print_response(response)
def main():
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()}'"
)

View File

@ -3,7 +3,6 @@ from .gametype import Gametype
from .hostname import Hostname
from .map import Map
from .mapname import Mapname
from .maprestart import Maprestart
from .maprotate import Maprotate
from .plugins import Plugins
from .status import Status
@ -17,5 +16,4 @@ __all__ = [
'Hostname',
'Map',
'Plugins',
'Maprestart',
]

View File

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

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 import console
from q3rcon_cli.console import Console
class Gametype(Command):
@ -23,24 +23,21 @@ class Gametype(Command):
@override
async def run(self):
if not self.new_gametype:
async with Spinner('Fetching current gametype', suffix='...'):
if not Gametype.new_gametype:
async with Client(self.host, self.port, self.password) as client:
response = await client.send_command('g_gametype')
console.out.print_cvar(response)
if response := await client.send_command('g_gametype'):
Console.print_cvar(response)
return
async with Spinner(f'Changing gametype to {self.new_gametype}', suffix='...'):
async with Client(self.host, self.port, self.password) as client:
await client.send_command(f'g_gametype {self.new_gametype}')
DEFAULT_FRAGMENT_READ_TIMEOUT = client.fragment_read_timeout
await client.send_command(f'g_gametype {self.new_gametype}')
if self.force:
async with Spinner('Forcing gametype change', suffix='...'):
async with Client(self.host, self.port, self.password) as client:
client.timeout = 3
async with Spinner('Forcing gametype change...'):
client.fragment_read_timeout = 1
await client.send_command('map_restart')
console.out.print(
f'Gametype changed successfully to {self.new_gametype}.', style='green'
)
client.fragment_read_timeout = DEFAULT_FRAGMENT_READ_TIMEOUT
if response := await client.send_command('g_gametype'):
Console.print_cvar(response)

View File

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

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 import console
from q3rcon_cli.console import Console
class Map(Command):
@ -19,18 +19,15 @@ class Map(Command):
@override
async def run(self):
if not self.new_map:
async with Spinner('Getting current map', suffix='...'):
async with Client(self.host, self.port, self.password) as client:
response = await client.send_command('mapname')
console.out.print_cvar(response)
if response := await client.send_command('mapname'):
Console.print_cvar(response)
return
async with Spinner('Changing map', suffix='...'):
async with Spinner('Changing map...'):
async with Client(
self.host, self.port, self.password, fragment_read_timeout=1
) as client:
await client.send_command(f'map mp_{self.new_map.removeprefix("mp_")}')
console.out.print(
f'Map changed to {self.new_map.removeprefix("mp_")}', style='green'
)
if response := await client.send_command('mapname'):
Console.print_cvar(response)

View File

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

View File

@ -1,23 +0,0 @@
from aioq3rcon import Client
from clypi import Command, Spinner, arg
from typing_extensions import override
from q3rcon_cli import console
class Maprestart(Command):
"""Restarts the current map."""
host: str = arg(inherited=True)
port: int = arg(inherited=True)
password: str = arg(inherited=True)
@override
async def run(self):
async with Spinner('Restarting map', suffix='...'):
async with Client(
self.host, self.port, self.password, timeout=3, fragment_read_timeout=1
) as client:
response = await client.send_command('map_restart', interpret=True)
console.out.print_response(response)

View File

@ -2,7 +2,7 @@ from aioq3rcon import Client
from clypi import Command, Spinner, arg
from typing_extensions import override
from q3rcon_cli import console
from q3rcon_cli.console import Console
class Maprotate(Command):
@ -14,14 +14,9 @@ class Maprotate(Command):
@override
async def run(self):
async with Spinner('Rotating map', suffix='...'):
async with Spinner('Rotating map...'):
async with Client(
self.host,
self.port,
self.password,
timeout=3,
fragment_read_timeout=1,
self.host, self.port, self.password, fragment_read_timeout=1
) as client:
response = await client.send_command('map_rotate', interpret=True)
console.out.print_response(response)
if response := await client.send_command('map_rotate'):
Console.print_response(response)

View File

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

View File

@ -2,7 +2,7 @@ from aioq3rcon import Client
from clypi import Command, Spinner, arg
from typing_extensions import override
from q3rcon_cli import console
from q3rcon_cli.console import Console
class Status(Command):
@ -14,10 +14,9 @@ class Status(Command):
@override
async def run(self):
async with Spinner('Fetching status', suffix='...'):
async with Spinner('Fetching status...'):
async with Client(
self.host, self.port, self.password, fragment_read_timeout=1
self.host, self.port, self.password, fragment_read_timeout=0.5
) as client:
response = await client.send_command('status')
console.out.print_status(response)
if response := await client.send_command('status'):
Console.print_status(response)

View File

@ -1,23 +1,10 @@
import re
import clypi
from clypi import Boxes, cprint
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+'
@ -33,8 +20,6 @@ class OutConsole(Console):
r'(?P<rate>[0-9]+)$',
re.IGNORECASE | re.VERBOSE,
)
STATUS_MAP_REGEX = re.compile(r'^map: (?P<mapname>mp_[a-z_]+)$')
CVAR_REGEX = re.compile(
r'^["](?P<name>[a-z_]+)["]\sis[:]\s'
r'["](?P<value>.*?)["]\s'
@ -45,15 +30,18 @@ class OutConsole(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 OutConsole.COLOUR_CODE_REGEX.sub('', s)
return Console.COLOUR_CODE_REGEX.sub('', s)
def print_response(self, response: str):
if response := self._remove_colour_codes(response).removeprefix('print\n'):
cprint(response, fg=self.style)
@staticmethod
def print_response(response: str):
response = Console.remove_colour_codes(response).removeprefix('print\n')
def print_status(self, response: str):
cprint(f'\n{response}\n', fg='yellow')
@staticmethod
def print_status(response: str):
_slots = []
_scores = []
_pings = []
@ -63,76 +51,45 @@ class OutConsole(Console):
lines = response.splitlines()
for line in lines:
if m := OutConsole.STATUS_PLAYER_REGEX.match(line):
if m := Console.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(self._remove_colour_codes(m.group('name')))
_names.append(m.group('name'))
_ips.append(m.group('ip'))
elif m := OutConsole.STATUS_MAP_REGEX.match(line):
cprint(f'Current map: {m.group("mapname")}', fg=self.style)
if not _slots:
cprint('No players connected.', fg='red')
cprint('\nNo players connected.\n', fg='yellow')
return
slots = clypi.boxed(
_slots, title='Slot', width=10, align='center', style=Boxes.ROUNDED
)
scores = clypi.boxed(
_scores, title='Score', width=10, align='center', style=Boxes.ROUNDED
)
pings = clypi.boxed(
_pings, title='Ping', width=10, align='center', style=Boxes.ROUNDED
)
guids = clypi.boxed(
_guids,
title='GUID',
width=len(max(_guids, key=len)) + 4,
style=Boxes.ROUNDED,
)
names = clypi.boxed(
_names,
title='Name',
width=len(max(_names, key=len)) + 4,
style=Boxes.ROUNDED,
)
ips = clypi.boxed(
_ips, title='IP', width=len(max(_ips, key=len)) + 4, style=Boxes.ROUNDED
)
print(f'{clypi.stack(slots, scores, pings, guids, names, ips, padding=0)}')
slots = clypi.boxed(_slots, title='Slot', width=15)
scores = clypi.boxed(_scores, title='Score', width=15)
pings = clypi.boxed(_pings, title='Ping', width=15)
guids = clypi.boxed(_guids, title='GUID', width=40)
names = clypi.boxed(_names, title='Name', width=30)
ips = clypi.boxed(_ips, title='IP', width=30)
print(f'\n{clypi.stack(slots, scores, pings, guids, names, ips, padding=0)}')
def print_cvar(self, response: str):
response = self._remove_colour_codes(response).removeprefix('print\n')
@staticmethod
def print_cvar(response: str):
response = Console.remove_colour_codes(response).removeprefix('print\n')
if m := self.CVAR_REGEX.match(response):
if m := Console.CVAR_REGEX.match(response):
name = clypi.boxed(
[m.group('name')],
title='Name',
width=max(len(m.group('name')) + 4, 15),
style=Boxes.ROUNDED,
[m.group('name')], title='Name', width=max(len(m.group('name')) + 4, 30)
)
value = clypi.boxed(
[m.group('value')],
title='Value',
width=max(len(m.group('value')) + 4, 15),
style=Boxes.ROUNDED,
width=max(len(m.group('value')) + 4, 30),
)
default = clypi.boxed(
[m.group('default')],
title='Default',
width=max(len(m.group('default')) + 4, 15),
style=Boxes.ROUNDED,
width=max(len(m.group('default')) + 4, 30),
)
info = clypi.boxed(
[m.group('info')],
title='Info',
width=max(len(m.group('info')) + 4, 15),
style=Boxes.ROUNDED,
[m.group('info')], title='Info', width=max(len(m.group('info')) + 4, 30)
)
print(f'{clypi.stack(name, value, default, info, padding=0)}')
out = OutConsole()
err = ErrorConsole()
print(f'\n{clypi.stack(name, value, default, info, padding=0)}')

View File

@ -1,41 +0,0 @@
#!/usr/bin/env python3
import re
import subprocess
from pathlib import Path
REPO_ROOT = Path(__file__).parent.parent
README_PATH = REPO_ROOT / 'README.md'
HELP_CMD = ['hatch', 'run', 'q3rcon-cli', '--help']
def get_help_output():
result = subprocess.run(HELP_CMD, capture_output=True, text=True, check=True)
output = result.stdout.strip()
ansi_escape = re.compile(r'\x1B\[[0-?]*[ -/]*[@-~]')
output = ansi_escape.sub('', output)
return output
def update_readme(help_text):
readme = README_PATH.read_text()
pattern = re.compile(
r'```console\nUsage: q3rcon-cli \[OPTIONS\] COMMAND.*?```', re.DOTALL
)
new_block = f'```console\n{help_text}\n```'
new_readme, n = pattern.subn(new_block, readme, count=1)
if n == 0:
raise RuntimeError(
'Could not find the q3rcon-cli usage console block to replace in README.md'
)
README_PATH.write_text(new_readme)
print('README.md updated with latest --help output.')
def main():
help_text = get_help_output()
update_readme(help_text)
if __name__ == '__main__':
main()