mousefox.app.clientframe

  1"""Home of `ClientFrame`."""
  2
  3from typing import Optional
  4import asyncio
  5import functools
  6import kvex as kx
  7import pgnet
  8from .connectpanel import ConnectPanel
  9from .userframe import UserFrame
 10
 11
 12class ClientFrame(kx.XAnchor):
 13    """Main client frame.
 14
 15    This frame contains the `ConnectPanel` and spawns a `UserFrame` for each
 16    successful connection.
 17    """
 18
 19    _conpath = "client"
 20
 21    def __init__(self, app_config):
 22        """Initialize the class with an `AppConfig`."""
 23        super().__init__()
 24        self._game_widget_factory = app_config.game_widget
 25        self._make_widgets(app_config)
 26        self._client: Optional[pgnet.Client] = None
 27        self.app.menu.set_callback("app", "disconnect", self._disconnect)
 28        self.app.controller.bind("client.user.disconnect", self._disconnect)
 29        self.app.controller.set_active_callback(self._conpath, self._on_screen)
 30        self._on_screen()
 31
 32    def update(self, *args):
 33        """Background refresh tasks."""
 34        if self._user_container.content:
 35            self._user_container.content.update()
 36
 37    def _make_widgets(self, app_config):
 38        self._connect_panel = ConnectPanel(app_config, self._set_client)
 39        placeholder = kx.XPlaceholder(label_text="Error: not connected.")
 40        self._user_container = kx.XContainer(placeholder)
 41        self._sm = kx.XScreenManager.from_widgets(
 42            dict(
 43                connect=self._connect_panel,
 44                user=self._user_container,
 45            ),
 46            transition=kx.FadeTransition(duration=0.2),
 47        )
 48        self._sm.bind(current=self._on_screen)
 49        self.add_widget(self._sm)
 50
 51    def _on_screen(self, *args):
 52        if self._sm.current == "connect":
 53            assert self._user_container.content is None
 54            self.app.menu.get_button("app", "disconnect").disabled = True
 55            self.app.menu.get_button("app", "leave_game").disabled = True
 56            if self.app.controller.active.startswith(self._conpath):
 57                self.app.controller.active = f"{self._conpath}.connect"
 58            self._connect_panel.set_focus()
 59        elif self._sm.current == "user":
 60            self.app.menu.get_button("app", "disconnect").disabled = False
 61            if self.app.controller.active.startswith(self._conpath):
 62                self.app.controller.active = f"{self._conpath}.user"
 63            userframe = self._user_container.content
 64            if userframe:
 65                userframe.set_focus()
 66        else:
 67            raise RuntimeError(f"Unknown screen: {self._sm.current}")
 68
 69    def _set_client(self, client: pgnet.Client, /):
 70        """Set a client for the app to use."""
 71        asyncio.create_task(self._async_set_client(client))
 72
 73    async def _async_set_client(self, client: pgnet.Client, /):
 74        assert not self._client
 75        self._client = client
 76        self.app.set_feedback(client.status)
 77        user_frame = UserFrame(client, self._game_widget_factory)
 78        self._user_container.content = user_frame
 79        client.on_status = functools.partial(self._on_client_status, client)
 80        client.on_connection = functools.partial(self._on_client_connected, client)
 81        await client.async_connect()
 82        assert not client.connected
 83        self._user_container.content = None
 84        self._client = None
 85        self._sm.current = "connect"
 86
 87    def _on_client_connected(self, client: pgnet.Client, connected: bool):
 88        assert client is self._client
 89        if not connected:
 90            return
 91        client.on_connection = None
 92        self._sm.current = "user"
 93
 94    def _on_client_status(self, client: pgnet.Client, status_message: str):
 95        assert client is self._client
 96        status = pgnet.Status.OK if client.connected else pgnet.Status.BAD
 97        self.app.set_feedback(status_message, status)
 98
 99    def _disconnect(self, *args):
100        if not self._client:
101            return
102        self._client.disconnect()