Edit on GitHub

mousefox.app.app

MouseFox GUI app.

View Source

  1"""MouseFox GUI app."""
  2
  3from typing import Optional, Type, Callable
  4from loguru import logger
  5import asyncio
  6from dataclasses import dataclass
  7import pathlib
  8import kvex as kx
  9import kvex.kivy
 10import pgnet
 11from .. import util
 12from .clientframe import ClientFrame
 13from .serverframe import ServerFrame
 14
 15
 16COLOR_PALETTE = "default"
 17HOTKEYS_FILE = pathlib.Path(__file__).parent / "hotkeys.toml"
 18MINIMUM_SIZE = (1024, 768)
 19
 20
 21@dataclass
 22class AppConfig:
 23    """Configuration for MouseFox app."""
 24
 25    game_widget: kvex.kivy.Widget
 26    """Kivy widget for the game."""
 27    game_class: Type[pgnet.Game]
 28    """Game subclass for the local client and server (see `pgnet.Server`)."""
 29    client_class: Type[pgnet.Client] = pgnet.Client
 30    """Client subclass for the client (see `pgnet.Client`)."""
 31    server_factory: Optional[Callable] = None
 32    """The server factory for the local client (see `pgnet.Client.local`)."""
 33    disable_local: bool = False
 34    """Disable local clients."""
 35    disable_remote: bool = False
 36    """Disable remote clients."""
 37    maximize: bool = False
 38    """If app should start maximized."""
 39    borderless: bool = False
 40    """If app should not have window borders."""
 41    size: Optional[tuple[int, int]] = (1280, 768)
 42    """App window size in pixels."""
 43    offset: Optional[tuple[int, int]] = None
 44    """App window offset in pixels."""
 45    title: str = "MouseFox"
 46    """App window title."""
 47    info_text: str = "No info available."
 48    """Text to show when ready to connect."""
 49    online_info_text: str = "No online info available."
 50    """Text to show when ready to connect remotely."""
 51    allow_quit: bool = True
 52    """Allow MouseFox to quit or restart the script."""
 53
 54    def __getitem__(self, item):
 55        """Get item."""
 56        return getattr(self, item)
 57
 58    def keys(self):
 59        """Enables mapping.
 60
 61        For example:
 62        ```python3
 63        config = mousefox.AppConfig(...)
 64        mousefox.run(**config)
 65        ```
 66        """
 67        return self.__dataclass_fields__.keys()
 68
 69
 70class App(kx.XApp):
 71    """MouseFox GUI app."""
 72
 73    def __init__(self, app_config: AppConfig, /):
 74        """Initialize the app. It is recommended to run the app with `mousefox.run`."""
 75        super().__init__()
 76        self._client: Optional[pgnet.Client] = None
 77        if app_config.borderless:
 78            self.toggle_borderless(True)
 79        if app_config.size:
 80            size = tuple(max(c) for c in zip(MINIMUM_SIZE, app_config.size))
 81            self.set_size(*size)
 82        else:
 83            self.set_size(*MINIMUM_SIZE)
 84        if app_config.offset:
 85            kx.schedule_once(lambda *a: self.set_position(*app_config.offset))
 86        if app_config.maximize:
 87            kx.schedule_once(lambda *a: self.maximize())
 88        self.title = app_config.title
 89        """App title."""
 90        self.controller = kx.XHotkeyController(
 91            name="app",
 92            logger=logger.debug,
 93            log_register=True,
 94            log_bind=True,
 95            log_callback=True,
 96            log_active=True,
 97        )
 98        self._register_controller(self.controller)
 99        self.game_controller = kx.XHotkeyController(
100            name="game",
101            logger=logger.debug,
102            log_register=True,
103            log_bind=True,
104            log_callback=True,
105            log_active=True,
106        )
107        self._make_widgets(app_config)
108        self.hook(self._update, 20)
109        self.set_feedback("Welcome")
110
111    async def async_run(self):
112        """Override base method."""
113        r = await super().async_run()
114        if self._client:
115            self._client.disconnect()
116        await _close_remaining_tasks()
117        return r
118
119    def set_feedback(self, text: str, status: pgnet.Status = pgnet.Status.OK, /):
120        """Set feedback in the status bar.
121
122        Args:
123            text: Text to show.
124            status: pgnet.Status, used for colors.
125        """
126        color = _STATUS_COLORS[status]
127        self._status.color = color.rgba
128        self._status.text = text
129
130    def feedback_response(
131        self,
132        response: pgnet.Response,
133        *,
134        only_statuses: bool = True,
135    ):
136        """Send a response to status bar.
137
138        Args:
139            response: pgnet Response.
140            only_statuses: Ignore responses with OK status.
141        """
142        if only_statuses and response.status == pgnet.Status.OK:
143            return
144        self.set_feedback(response.message, response.status)
145
146    def _show_client(self, *args):
147        self._sm.current = "client"
148        self.controller.active = "client"
149        self.menu.get_button("app", "host_server").disabled = False
150
151    def _show_server(self, *args):
152        self._sm.current = "server"
153        self.controller.active = "server"
154        self.menu.get_button("app", "host_server").disabled = True
155
156    def _make_widgets(self, app_config):
157        self._make_menu()
158        self._status = kx.XLabel(
159            halign="left",
160            valign="middle",
161            italic=True,
162            padding=(10, 10),
163            outline_width=2,
164            outline_color=(0, 0, 0),
165            font_size="18sp",
166        )
167        self._client_frame = ClientFrame(app_config)
168        self._server_frame = ServerFrame(app_config)
169        self._sm = kx.XScreenManager.from_widgets(
170            dict(
171                client=self._client_frame,
172                server=self._server_frame,
173            ),
174            transition=kx.FadeTransition(duration=0.2),
175        )
176        # Assemble
177        self.top_bar = kx.XBox()
178        self.top_bar.add_widgets(self.menu, self._status)
179        self.top_bar.set_size(y="32sp")
180        main_frame = kx.XBox(orientation="vertical")
181        main_frame.add_widgets(self.top_bar, self._sm)
182        self.root.clear_widgets()
183        self.root.add_widget(main_frame)
184        self._refresh_background()
185        self._show_client()
186
187    def on_theme(self, *args):
188        """Override base method."""
189        self.set_feedback(f"Set theme: {self.theme_name}")
190        self._refresh_background()
191
192    def _refresh_background(self, *args):
193        self.root.make_bg(self.theme.primary.bg)
194
195    def _make_menu(self):
196        self.menu = kx.XButtonBar()
197        self.menu.set_size(x="500dp")
198        self.menu.add_category("app")
199        menu_add = self.menu.add_button
200        menu_get = self.menu.get_button
201        menu_add("app", "quit", self.stop)
202        menu_add("app", "restart", self.restart)
203        menu_add("app", "host_server", self._show_server)
204        menu_add("app", "disconnect")
205        menu_add("app", "leave_game")
206        menu_add("app", "lobby")
207        menu_add("app", "game")
208        menu_add("app", "admin_panel")
209        menu_get("app", "disconnect").disabled = True
210        menu_get("app", "leave_game").disabled = True
211        menu_get("app", "lobby").disabled = True
212        menu_get("app", "game").disabled = True
213        menu_get("app", "admin_panel").disabled = True
214        self.menu.add_theme_selectors(prefix="")
215
216    def _register_controller(self, controller: kx.XHotkeyController):
217        loaded_dict = util.toml_load(HOTKEYS_FILE)
218        hotkeys = _flatten_hotkey_paths(loaded_dict)
219        for control, hotkeys in hotkeys.items():
220            if not isinstance(hotkeys, list):
221                hotkeys = [hotkeys]
222            for hk in hotkeys:
223                controller.register(control, hk)
224        controller.bind("quit", self.stop)
225        controller.bind("restart", self.restart)
226        controller.bind("debug", controller.debug)
227        controller.bind("show_client", self._show_client)
228        controller.bind("show_server", self._show_server)
229        for i, tname in enumerate(kx.THEME_NAMES):
230            self.controller.register(
231                f"Change theme to {tname}",
232                f"numpad{i+1}",
233                bind=lambda *a, t=tname: self.set_theme(t),
234            )
235
236    def _update(self, *args):
237        self._client_frame.update()
238
239
240def _flatten_hotkey_paths(nested: dict, prefix: str = "") -> dict:
241    new_dict = dict()
242    for k, v in nested.items():
243        if isinstance(v, dict):
244            new_dict |= _flatten_hotkey_paths(v, f"{prefix}{k}.")
245        else:
246            new_dict[f"{prefix}{k}"] = v
247    return new_dict
248
249
250async def _close_remaining_tasks(debug: bool = True):
251    remaining_tasks = asyncio.all_tasks() - {asyncio.current_task(), }
252    if not remaining_tasks:
253        return
254    for t in remaining_tasks:
255        t.cancel()
256    if debug:
257        logger.debug(
258            f"Remaining {len(remaining_tasks)} tasks:\n"
259            + "\n".join(f"  -- {t}" for t in remaining_tasks)
260        )
261    for coro in asyncio.as_completed(list(remaining_tasks)):
262        try:
263            await coro
264        except asyncio.CancelledError:
265            removed_tasks = remaining_tasks - asyncio.all_tasks()
266            remaining_tasks -= removed_tasks
267            if removed_tasks and not debug:
268                logger.debug(f"Removed {len(removed_tasks)} tasks: {removed_tasks}")
269                logger.debug(
270                    f"Remaining {len(remaining_tasks)} tasks:\n"
271                    + "\n".join(f"  -- {t}" for t in remaining_tasks)
272                )
273            continue
274
275
276_STATUS_COLORS = {
277    pgnet.Status.OK.value: kx.XColor.from_hex("00bb00"),
278    pgnet.Status.UNEXPECTED.value: kx.XColor.from_hex("bbbb00"),
279    pgnet.Status.BAD.value: kx.XColor.from_hex("ff0000"),
280}

@dataclass

class AppConfig: View Source

22@dataclass
23class AppConfig:
24    """Configuration for MouseFox app."""
25
26    game_widget: kvex.kivy.Widget
27    """Kivy widget for the game."""
28    game_class: Type[pgnet.Game]
29    """Game subclass for the local client and server (see `pgnet.Server`)."""
30    client_class: Type[pgnet.Client] = pgnet.Client
31    """Client subclass for the client (see `pgnet.Client`)."""
32    server_factory: Optional[Callable] = None
33    """The server factory for the local client (see `pgnet.Client.local`)."""
34    disable_local: bool = False
35    """Disable local clients."""
36    disable_remote: bool = False
37    """Disable remote clients."""
38    maximize: bool = False
39    """If app should start maximized."""
40    borderless: bool = False
41    """If app should not have window borders."""
42    size: Optional[tuple[int, int]] = (1280, 768)
43    """App window size in pixels."""
44    offset: Optional[tuple[int, int]] = None
45    """App window offset in pixels."""
46    title: str = "MouseFox"
47    """App window title."""
48    info_text: str = "No info available."
49    """Text to show when ready to connect."""
50    online_info_text: str = "No online info available."
51    """Text to show when ready to connect remotely."""
52    allow_quit: bool = True
53    """Allow MouseFox to quit or restart the script."""
54
55    def __getitem__(self, item):
56        """Get item."""
57        return getattr(self, item)
58
59    def keys(self):
60        """Enables mapping.
61
62        For example:
63        ```python3
64        config = mousefox.AppConfig(...)
65        mousefox.run(**config)
66        ```
67        """
68        return self.__dataclass_fields__.keys()

Configuration for MouseFox app.

AppConfig( game_widget: kivy.uix.widget.Widget, game_class: Type[pgnet.util.Game], client_class: Type[pgnet.client.Client] = <class 'pgnet.client.Client'>, server_factory: Optional[Callable] = None, disable_local: bool = False, disable_remote: bool = False, maximize: bool = False, borderless: bool = False, size: Optional[tuple[int, int]] = (1280, 768), offset: Optional[tuple[int, int]] = None, title: str = 'MouseFox', info_text: str = 'No info available.', online_info_text: str = 'No online info available.', allow_quit: bool = True)

game_widget: kivy.uix.widget.Widget

Kivy widget for the game.

game_class: Type[pgnet.util.Game]

Game subclass for the local client and server (see pgnet.Server).

client_class: Type[pgnet.client.Client] = <class 'pgnet.client.Client'>

Client subclass for the client (see pgnet.Client).

server_factory: Optional[Callable] = None

The server factory for the local client (see pgnet.Client.local).

disable_local: bool = False

Disable local clients.

disable_remote: bool = False

Disable remote clients.

maximize: bool = False

If app should start maximized.

borderless: bool = False

If app should not have window borders.

size: Optional[tuple[int, int]] = (1280, 768)

App window size in pixels.

offset: Optional[tuple[int, int]] = None

App window offset in pixels.

title: str = 'MouseFox'

App window title.

info_text: str = 'No info available.'

Text to show when ready to connect.

online_info_text: str = 'No online info available.'

Text to show when ready to connect remotely.

allow_quit: bool = True

Allow MouseFox to quit or restart the script.

def keys(self): View Source

59    def keys(self):
60        """Enables mapping.
61
62        For example:
63        ```python3
64        config = mousefox.AppConfig(...)
65        mousefox.run(**config)
66        ```
67        """
68        return self.__dataclass_fields__.keys()

Enables mapping.

For example:

config = mousefox.AppConfig(...)
mousefox.run(**config)

class App(kvex.app.XApp): View Source

 71class App(kx.XApp):
 72    """MouseFox GUI app."""
 73
 74    def __init__(self, app_config: AppConfig, /):
 75        """Initialize the app. It is recommended to run the app with `mousefox.run`."""
 76        super().__init__()
 77        self._client: Optional[pgnet.Client] = None
 78        if app_config.borderless:
 79            self.toggle_borderless(True)
 80        if app_config.size:
 81            size = tuple(max(c) for c in zip(MINIMUM_SIZE, app_config.size))
 82            self.set_size(*size)
 83        else:
 84            self.set_size(*MINIMUM_SIZE)
 85        if app_config.offset:
 86            kx.schedule_once(lambda *a: self.set_position(*app_config.offset))
 87        if app_config.maximize:
 88            kx.schedule_once(lambda *a: self.maximize())
 89        self.title = app_config.title
 90        """App title."""
 91        self.controller = kx.XHotkeyController(
 92            name="app",
 93            logger=logger.debug,
 94            log_register=True,
 95            log_bind=True,
 96            log_callback=True,
 97            log_active=True,
 98        )
 99        self._register_controller(self.controller)
100        self.game_controller = kx.XHotkeyController(
101            name="game",
102            logger=logger.debug,
103            log_register=True,
104            log_bind=True,
105            log_callback=True,
106            log_active=True,
107        )
108        self._make_widgets(app_config)
109        self.hook(self._update, 20)
110        self.set_feedback("Welcome")
111
112    async def async_run(self):
113        """Override base method."""
114        r = await super().async_run()
115        if self._client:
116            self._client.disconnect()
117        await _close_remaining_tasks()
118        return r
119
120    def set_feedback(self, text: str, status: pgnet.Status = pgnet.Status.OK, /):
121        """Set feedback in the status bar.
122
123        Args:
124            text: Text to show.
125            status: pgnet.Status, used for colors.
126        """
127        color = _STATUS_COLORS[status]
128        self._status.color = color.rgba
129        self._status.text = text
130
131    def feedback_response(
132        self,
133        response: pgnet.Response,
134        *,
135        only_statuses: bool = True,
136    ):
137        """Send a response to status bar.
138
139        Args:
140            response: pgnet Response.
141            only_statuses: Ignore responses with OK status.
142        """
143        if only_statuses and response.status == pgnet.Status.OK:
144            return
145        self.set_feedback(response.message, response.status)
146
147    def _show_client(self, *args):
148        self._sm.current = "client"
149        self.controller.active = "client"
150        self.menu.get_button("app", "host_server").disabled = False
151
152    def _show_server(self, *args):
153        self._sm.current = "server"
154        self.controller.active = "server"
155        self.menu.get_button("app", "host_server").disabled = True
156
157    def _make_widgets(self, app_config):
158        self._make_menu()
159        self._status = kx.XLabel(
160            halign="left",
161            valign="middle",
162            italic=True,
163            padding=(10, 10),
164            outline_width=2,
165            outline_color=(0, 0, 0),
166            font_size="18sp",
167        )
168        self._client_frame = ClientFrame(app_config)
169        self._server_frame = ServerFrame(app_config)
170        self._sm = kx.XScreenManager.from_widgets(
171            dict(
172                client=self._client_frame,
173                server=self._server_frame,
174            ),
175            transition=kx.FadeTransition(duration=0.2),
176        )
177        # Assemble
178        self.top_bar = kx.XBox()
179        self.top_bar.add_widgets(self.menu, self._status)
180        self.top_bar.set_size(y="32sp")
181        main_frame = kx.XBox(orientation="vertical")
182        main_frame.add_widgets(self.top_bar, self._sm)
183        self.root.clear_widgets()
184        self.root.add_widget(main_frame)
185        self._refresh_background()
186        self._show_client()
187
188    def on_theme(self, *args):
189        """Override base method."""
190        self.set_feedback(f"Set theme: {self.theme_name}")
191        self._refresh_background()
192
193    def _refresh_background(self, *args):
194        self.root.make_bg(self.theme.primary.bg)
195
196    def _make_menu(self):
197        self.menu = kx.XButtonBar()
198        self.menu.set_size(x="500dp")
199        self.menu.add_category("app")
200        menu_add = self.menu.add_button
201        menu_get = self.menu.get_button
202        menu_add("app", "quit", self.stop)
203        menu_add("app", "restart", self.restart)
204        menu_add("app", "host_server", self._show_server)
205        menu_add("app", "disconnect")
206        menu_add("app", "leave_game")
207        menu_add("app", "lobby")
208        menu_add("app", "game")
209        menu_add("app", "admin_panel")
210        menu_get("app", "disconnect").disabled = True
211        menu_get("app", "leave_game").disabled = True
212        menu_get("app", "lobby").disabled = True
213        menu_get("app", "game").disabled = True
214        menu_get("app", "admin_panel").disabled = True
215        self.menu.add_theme_selectors(prefix="")
216
217    def _register_controller(self, controller: kx.XHotkeyController):
218        loaded_dict = util.toml_load(HOTKEYS_FILE)
219        hotkeys = _flatten_hotkey_paths(loaded_dict)
220        for control, hotkeys in hotkeys.items():
221            if not isinstance(hotkeys, list):
222                hotkeys = [hotkeys]
223            for hk in hotkeys:
224                controller.register(control, hk)
225        controller.bind("quit", self.stop)
226        controller.bind("restart", self.restart)
227        controller.bind("debug", controller.debug)
228        controller.bind("show_client", self._show_client)
229        controller.bind("show_server", self._show_server)
230        for i, tname in enumerate(kx.THEME_NAMES):
231            self.controller.register(
232                f"Change theme to {tname}",
233                f"numpad{i+1}",
234                bind=lambda *a, t=tname: self.set_theme(t),
235            )
236
237    def _update(self, *args):
238        self._client_frame.update()

MouseFox GUI app.

App(app_config: mousefox.app.app.AppConfig, /) View Source

 74    def __init__(self, app_config: AppConfig, /):
 75        """Initialize the app. It is recommended to run the app with `mousefox.run`."""
 76        super().__init__()
 77        self._client: Optional[pgnet.Client] = None
 78        if app_config.borderless:
 79            self.toggle_borderless(True)
 80        if app_config.size:
 81            size = tuple(max(c) for c in zip(MINIMUM_SIZE, app_config.size))
 82            self.set_size(*size)
 83        else:
 84            self.set_size(*MINIMUM_SIZE)
 85        if app_config.offset:
 86            kx.schedule_once(lambda *a: self.set_position(*app_config.offset))
 87        if app_config.maximize:
 88            kx.schedule_once(lambda *a: self.maximize())
 89        self.title = app_config.title
 90        """App title."""
 91        self.controller = kx.XHotkeyController(
 92            name="app",
 93            logger=logger.debug,
 94            log_register=True,
 95            log_bind=True,
 96            log_callback=True,
 97            log_active=True,
 98        )
 99        self._register_controller(self.controller)
100        self.game_controller = kx.XHotkeyController(
101            name="game",
102            logger=logger.debug,
103            log_register=True,
104            log_bind=True,
105            log_callback=True,
106            log_active=True,
107        )
108        self._make_widgets(app_config)
109        self.hook(self._update, 20)
110        self.set_feedback("Welcome")

Initialize the app. It is recommended to run the app with mousefox.run.

title

App title.

async def async_run(self): View Source

112    async def async_run(self):
113        """Override base method."""
114        r = await super().async_run()
115        if self._client:
116            self._client.disconnect()
117        await _close_remaining_tasks()
118        return r

Override base method.

def set_feedback(self, text: str, status: pgnet.util.Status = <Status.OK: 0>, /): View Source

120    def set_feedback(self, text: str, status: pgnet.Status = pgnet.Status.OK, /):
121        """Set feedback in the status bar.
122
123        Args:
124            text: Text to show.
125            status: pgnet.Status, used for colors.
126        """
127        color = _STATUS_COLORS[status]
128        self._status.color = color.rgba
129        self._status.text = text

Set feedback in the status bar.

Arguments:

text: Text to show.
status: pgnet.Status, used for colors.

def feedback_response(self, response: pgnet.util.Response, *, only_statuses: bool = True): View Source

131    def feedback_response(
132        self,
133        response: pgnet.Response,
134        *,
135        only_statuses: bool = True,
136    ):
137        """Send a response to status bar.
138
139        Args:
140            response: pgnet Response.
141            only_statuses: Ignore responses with OK status.
142        """
143        if only_statuses and response.status == pgnet.Status.OK:
144            return
145        self.set_feedback(response.message, response.status)

Send a response to status bar.

Arguments:

response: pgnet Response.
only_statuses: Ignore responses with OK status.

def on_theme(self, *args): View Source

188    def on_theme(self, *args):
189        """Override base method."""
190        self.set_feedback(f"Set theme: {self.theme_name}")
191        self._refresh_background()

Override base method.

Inherited Members

kvex.app.XApp: current_focus; block_input; window; theme_name; theme; set_theme; subtheme_name; subtheme_context; run; restart; hook; add_widget; mouse_pos; maximize; set_size; toggle_fullscreen; toggle_borderless; set_position; enable_escape_exit; disable_multitouch; enable_resize; open_settings; overlay; with_overlay
kivy.app.App: icon; use_kivy_settings; settings_cls; kv_directory; kv_file; build; build_config; build_settings; load_kv; get_application_name; get_application_icon; get_application_config; root_window; load_config; directory; user_data_dir; name; stop; on_start; on_stop; on_pause; on_resume; get_running_app; on_config_change; display_settings; close_settings; create_settings; destroy_settings; on_title; on_icon
kivy._event.EventDispatcher: register_event_type; unregister_event_types; unregister_event_type; is_event_type; bind; unbind; fbind; funbind; unbind_uid; get_property_observers; events; dispatch; dispatch_generic; dispatch_children; setter; getter; property; properties; create_property; apply_property; proxy_ref