pgnet.devclient

  1"""Developer client - command line tool to interface with a server."""
  2
  3from typing import Optional, Any
  4import sys
  5import arrow
  6import asyncio
  7import aioconsole
  8from .server import Server
  9from .client import Client
 10from .util import (
 11    Packet,
 12    Response,
 13    DEFAULT_PORT,
 14    ADMIN_USERNAME,
 15    DEFAULT_ADMIN_PASSWORD,
 16)
 17from .examples import ExampleGame, ExampleClient
 18
 19
 20class DevCLI:
 21    """A CLI for the pgnet client."""
 22
 23    async def async_run(self, remote: bool):
 24        """Run the command-line devclient.
 25
 26        Args:
 27            remote: Connect to a remote server, otherwise run a local server using
 28                `DevGame`.
 29        """
 30        self.client = self._get_client(remote)
 31        conn_coro = self.client.async_connect()
 32        conn_task = asyncio.create_task(conn_coro)
 33        cli_task = asyncio.create_task(self._cli())
 34        combined_task = asyncio.wait(
 35            (conn_task, cli_task),
 36            return_when=asyncio.FIRST_COMPLETED,
 37        )
 38        await combined_task
 39        if not conn_task.done():
 40            self.client.disconnect()
 41            await asyncio.wait_for(conn_task, timeout=1)
 42
 43    def _get_client(self, remote: bool):
 44        username = ADMIN_USERNAME
 45        password = DEFAULT_ADMIN_PASSWORD
 46        username = input("Enter username (leave blank for admin): ") or username
 47        password = input("Enter password (leave blank for admin default): ") or password
 48        if not remote:
 49            return ExampleClient.local(
 50                game=ExampleGame,
 51                username=username,
 52                password=password,
 53            )
 54        address = input("Enter address (leave blank for localhost): ") or "localhost"
 55        port = int(input("Enter port (leave blank for default): ") or DEFAULT_PORT)
 56        pubkey = input("Enter pubkey to verify (leave blank to ignore): ") or ""
 57        return Client.remote(
 58            username=username,
 59            password=password,
 60            address=address,
 61            port=port,
 62            verify_server_pubkey=pubkey,
 63        )
 64
 65    async def _cli(self):
 66        while not self.client.connected:
 67            await asyncio.sleep(0.1)
 68        while True:
 69            uinput = await aioconsole.ainput(">> ")
 70            if uinput == "quit":
 71                return
 72            packet = self._parse_cli_packet(uinput)
 73            if packet:
 74                await self._send_packet(packet)
 75
 76    async def _send_packet(self, packet):
 77        print(packet.debug_repr)
 78        print(f"    SENT: {arrow.now().for_json()}")
 79        response = asyncio.Future()
 80        self.client.send(packet, lambda sr, r=response: r.set_result(sr))
 81        await response
 82        self._log_response(response.result())
 83
 84    @classmethod
 85    def _parse_cli_packet(cls, s: str, /) -> Optional[Packet]:
 86        try:
 87            parts = s.split(";")
 88            message = parts.pop(0)
 89            if message.startswith(".."):
 90                message = message[1:]
 91            elif message.startswith("."):
 92                message = f"__pgnet__{message}"
 93            payload = {}
 94            for p in parts:
 95                key, value = cls._parse_part(p)
 96                payload[key] = value
 97        except ValueError as e:
 98            print(
 99                "Bad CLI request format, expected: "
100                f"'message_str;key1=value1;key2=value2'\n{e}"
101            )
102            return None
103        return Packet(message, payload)
104
105    @staticmethod
106    def _parse_part(part: str, /) -> tuple[str, Any]:
107        key, value = part.split("=", 1)
108        if value == "True":
109            value = True
110        elif value == "False":
111            value = False
112        elif value.isnumeric():
113            value = int(value)
114        else:
115            try:
116                value = float(value)
117            except ValueError:
118                pass
119        return key, value
120
121    @staticmethod
122    def _log_response(response: Response):
123        strs = [
124            f"RECEIVED: {arrow.now().for_json()}",
125            response.debug_repr,
126            "-" * 20,
127            f"MESSAGE:  {response.message}",
128        ]
129        if len(tuple(response.payload.keys())):
130            strs.extend([
131                "PAYLOAD:",
132                *(f"{k:>20} : {v}" for k, v in response.payload.items()),
133            ])
134        print("\n".join(strs))
135        print("=" * 20)
136
137
138def run():
139    """Main script entry point for the devclient.
140
141    Will parse the first argument from `sys.argv`:
142    * `no argument`: run locally using `pgnet.ExampleClient` and `pgnet.ExampleGame`
143    * `"-s"` or `"--server"`: run a server using `pgnet.ExampleGame`
144    * `"-r"` or `"--remote"`: connect to a remote server using `pgnet.Client`
145    """
146    arg = None
147    if len(sys.argv) > 1:
148        arg = sys.argv[1]
149    if arg in {"-s", "--server"}:
150        asyncio.run(Server(ExampleGame).async_run())
151    elif arg in {"-r", "--remote"}:
152        asyncio.run(DevCLI().async_run(remote=True))
153    else:
154        asyncio.run(DevCLI().async_run(remote=False))
155
156
157__all__ = (
158    "run",
159)