first commit

src/q3rcon_cli/__about__.py

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

src/q3rcon_cli/__init__.py

@@ -0,0 +1,3 @@
+# SPDX-FileCopyrightText: 2026-present onyx-and-iris <code@onyxandiris.online>
+#
+# SPDX-License-Identifier: MIT

src/q3rcon_cli/cli.py

@@ -0,0 +1,102 @@
+import clypi
+from aioq3rcon import Client
+from clypi import Command, arg
+from typing_extensions import override
+
+from .commands import (
+    Fastrestart,
+    Gametype,
+    Hostname,
+    Map,
+    Mapname,
+    Maprotate,
+    Plugins,
+    Status,
+)
+from .console import Console
+
+Subcommands = (
+    Status | Mapname | Maprotate | Fastrestart | Gametype | Hostname | Map | Plugins
+)
+
+
+class Q3rconCli(Command):
+    subcommand: Subcommands | None = None
+    host: str = arg(
+        'localhost',
+        short='h',
+        help='The host to connect to',
+        env='Q3RCON_CLI_HOST',
+        group='Connection',
+    )
+    port: int = arg(
+        27960,
+        short='p',
+        help='The port to connect to',
+        env='Q3RCON_CLI_PORT',
+        group='Connection',
+    )
+    password: str = arg(
+        '',
+        short='P',
+        help='The password for authentication',
+        env='Q3RCON_CLI_PASSWORD',
+        group='Connection',
+    )
+    interactive: bool = arg(
+        False,
+        short='i',
+        help='Whether to start in interactive mode (defaults to false)',
+    )
+
+    @override
+    async def run(self):
+        if self.interactive:
+            await self.run_interactive()
+        else:
+            await Status.run(self)
+
+    async def run_interactive(self):
+        print(
+            clypi.style('Entering interactive mode. Type', fg='blue'),
+            clypi.style("'Q'", fg='red'),
+            clypi.style('to quit.', fg='blue'),
+        )
+
+        DEFAULT_FRAGMENT_READ_TIMEOUT = 0.25
+        while command := input(clypi.style('cmd: ', fg='green')):
+            if command.lower() == 'q':
+                break
+
+            fragment_read_timeout = None
+            if command in (
+                'status',
+                'fast_restart',
+                'map_restart',
+                'map',
+                'map_rotate',
+            ):
+                fragment_read_timeout = 1
+
+            async with Client(
+                self.host,
+                self.port,
+                self.password,
+                fragment_read_timeout=fragment_read_timeout
+                or DEFAULT_FRAGMENT_READ_TIMEOUT,
+            ) as client:
+                try:
+                    if response := await client.send_command(command):
+                        Console.print_response(response)
+                except TimeoutError:
+                    print(
+                        clypi.style(
+                            'Timeout waiting for response for command:', fg='red'
+                        ),
+                        clypi.style(command, fg='yellow'),
+                    )
+
+
+def main():
+    cli = Q3rconCli().parse()
+    cli.start()

src/q3rcon_cli/commands/__init__.py

@@ -0,0 +1,19 @@
+from .fastrestart import Fastrestart
+from .gametype import Gametype
+from .hostname import Hostname
+from .map import Map
+from .mapname import Mapname
+from .maprotate import Maprotate
+from .plugins import Plugins
+from .status import Status
+
+__all__ = [
+    'Status',
+    'Mapname',
+    'Maprotate',
+    'Fastrestart',
+    'Gametype',
+    'Hostname',
+    'Map',
+    'Plugins',
+]

src/q3rcon_cli/commands/fastrestart.py

@@ -0,0 +1,19 @@
+from aioq3rcon import Client
+from clypi import Command, arg
+from typing_extensions import override
+
+from q3rcon_cli.console import Console
+
+
+class Fastrestart(Command):
+    """Executes a fast restart of the server."""
+
+    host: str = arg(inherited=True)
+    port: int = arg(inherited=True)
+    password: str = arg(inherited=True)
+
+    @override
+    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)

src/q3rcon_cli/commands/gametype.py

@@ -0,0 +1,43 @@
+from aioq3rcon import Client
+from clypi import Command, Positional, Spinner, arg
+from typing_extensions import override
+
+from q3rcon_cli.console import Console
+
+
+class Gametype(Command):
+    """Get or set the current gametype of the server."""
+
+    new_gametype: Positional[str] = arg(
+        help='The new gametype to change to (optional). If not provided, the current gametype will be printed.',
+        default='',
+    )
+    host: str = arg(inherited=True)
+    port: int = arg(inherited=True)
+    password: str = arg(inherited=True)
+    force: bool = arg(
+        False,
+        short='f',
+        help='Whether to force the gametype change even if players are currently in the server.',
+    )
+
+    @override
+    async def run(self):
+        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)
+            return
+
+        async with Client(self.host, self.port, self.password) as client:
+            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...'):
+                    client.fragment_read_timeout = 1
+                    await client.send_command('map_restart')
+
+                client.fragment_read_timeout = DEFAULT_FRAGMENT_READ_TIMEOUT
+                if response := await client.send_command('g_gametype'):
+                    Console.print_cvar(response)

src/q3rcon_cli/commands/hostname.py

@@ -0,0 +1,30 @@
+from aioq3rcon import Client
+from clypi import Command, Positional, arg
+from typing_extensions import override
+
+from q3rcon_cli.console import Console
+
+
+class Hostname(Command):
+    """Get or set the current hostname of the server."""
+
+    new_hostname: Positional[str] = arg(
+        help='The new hostname to change to (optional). If not provided, the current hostname will be printed.',
+        default='',
+    )
+    host: str = arg(inherited=True)
+    port: int = arg(inherited=True)
+    password: str = arg(inherited=True)
+
+    @override
+    async def run(self):
+        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)
+            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)

src/q3rcon_cli/commands/map.py

@@ -0,0 +1,33 @@
+from aioq3rcon import Client
+from clypi import Command, Positional, Spinner, arg
+from typing_extensions import override
+
+from q3rcon_cli.console import Console
+
+
+class Map(Command):
+    """Get the current map or change to a new one."""
+
+    new_map: Positional[str] = arg(
+        help='The new map to change to (optional). If not provided, the current map will be printed.',
+        default='',
+    )
+    host: str = arg(inherited=True)
+    port: int = arg(inherited=True)
+    password: str = arg(inherited=True)
+
+    @override
+    async def run(self):
+        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)
+            return
+
+        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_")}')
+                if response := await client.send_command('mapname'):
+                    Console.print_cvar(response)

src/q3rcon_cli/commands/mapname.py

@@ -0,0 +1,19 @@
+from aioq3rcon import Client
+from clypi import Command, arg
+from typing_extensions import override
+
+from q3rcon_cli.console import Console
+
+
+class Mapname(Command):
+    """Prints the current map name of the server."""
+
+    host: str = arg(inherited=True)
+    port: int = arg(inherited=True)
+    password: str = arg(inherited=True)
+
+    @override
+    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)

src/q3rcon_cli/commands/maprotate.py

@@ -0,0 +1,22 @@
+from aioq3rcon import Client
+from clypi import Command, Spinner, arg
+from typing_extensions import override
+
+from q3rcon_cli.console import Console
+
+
+class Maprotate(Command):
+    """Rotates the map to the next one in the map rotation list."""
+
+    host: str = arg(inherited=True)
+    port: int = arg(inherited=True)
+    password: str = arg(inherited=True)
+
+    @override
+    async def run(self):
+        async with Spinner('Rotating map...'):
+            async with Client(
+                self.host, self.port, self.password, fragment_read_timeout=1
+            ) as client:
+                if response := await client.send_command('map_rotate'):
+                    Console.print_response(response)

src/q3rcon_cli/commands/plugins.py

@@ -0,0 +1,19 @@
+from aioq3rcon import Client
+from clypi import Command, arg
+from typing_extensions import override
+
+from q3rcon_cli.console import Console
+
+
+class Plugins(Command):
+    """Prints the currently loaded plugins of the server."""
+
+    host: str = arg(inherited=True)
+    port: int = arg(inherited=True)
+    password: str = arg(inherited=True)
+
+    @override
+    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)

src/q3rcon_cli/commands/status.py

@@ -0,0 +1,22 @@
+from aioq3rcon import Client
+from clypi import Command, Spinner, arg
+from typing_extensions import override
+
+from q3rcon_cli.console import Console
+
+
+class Status(Command):
+    """Prints the status of the server."""
+
+    host: str = arg(inherited=True)
+    port: int = arg(inherited=True)
+    password: str = arg(inherited=True)
+
+    @override
+    async def run(self):
+        async with Spinner('Fetching status...'):
+            async with Client(
+                self.host, self.port, self.password, fragment_read_timeout=0.5
+            ) as client:
+                if response := await client.send_command('status'):
+                    Console.print_status(response)

src/q3rcon_cli/console.py

@@ -0,0 +1,95 @@
+import re
+
+import clypi
+from clypi import cprint
+
+
+class Console:
+    COLOUR_CODE_REGEX = re.compile(r'\^[0-9]')
+    STATUS_PLAYER_REGEX = re.compile(
+        r'^\s*(?P<slot>[0-9]+)\s+'
+        r'(?P<score>[0-9-]+)\s+'
+        r'(?P<ping>[0-9]+)\s+'
+        r'(?P<guid>[0-9a-f]+)\s+'
+        r'(?P<name>.*?)\s+'
+        r'(?P<last>[0-9]+?)\s*'
+        r'(?P<ip>(?:(?:25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])\.){3}'
+        r'(?:25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])):?'
+        r'(?P<port>-?[0-9]{1,5})\s*'
+        r'(?P<qport>-?[0-9]{1,5})\s+'
+        r'(?P<rate>[0-9]+)$',
+        re.IGNORECASE | re.VERBOSE,
+    )
+    CVAR_REGEX = re.compile(
+        r'^["](?P<name>[a-z_]+)["]\sis[:]\s'
+        r'["](?P<value>.*?)["]\s'
+        r'default[:]\s'
+        r'["](?P<default>.*?)["]\s'
+        r'info[:]\s'
+        r'["](?P<info>.*?)["]$'
+    )
+
+    @staticmethod
+    def remove_colour_codes(s: str) -> str:
+        """Remove Quake 3 colour codes from a string."""
+        return Console.COLOUR_CODE_REGEX.sub('', s)
+
+    @staticmethod
+    def print_response(response: str):
+        response = Console.remove_colour_codes(response).removeprefix('print\n')
+
+        cprint(f'\n{response}\n', fg='yellow')
+
+    @staticmethod
+    def print_status(response: str):
+        _slots = []
+        _scores = []
+        _pings = []
+        _guids = []
+        _names = []
+        _ips = []
+
+        lines = response.splitlines()
+        for line in lines:
+            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(m.group('name'))
+                _ips.append(m.group('ip'))
+
+        if not _slots:
+            cprint('\nNo players connected.\n', fg='yellow')
+            return
+
+        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)}')
+
+    @staticmethod
+    def print_cvar(response: str):
+        response = Console.remove_colour_codes(response).removeprefix('print\n')
+
+        if m := Console.CVAR_REGEX.match(response):
+            name = clypi.boxed(
+                [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, 30),
+            )
+            default = clypi.boxed(
+                [m.group('default')],
+                title='Default',
+                width=max(len(m.group('default')) + 4, 30),
+            )
+            info = clypi.boxed(
+                [m.group('info')], title='Info', width=max(len(m.group('info')) + 4, 30)
+            )
+            print(f'\n{clypi.stack(name, value, default, info, padding=0)}')