bauiv1lib.connectivity

  1# Released under the MIT License. See LICENSE for details.
  2#
  3"""UI functionality related to master-server connectivity."""
  4
  5from __future__ import annotations
  6
  7from typing import TYPE_CHECKING
  8
  9import bauiv1 as bui
 10
 11if TYPE_CHECKING:
 12    from typing import Callable, Any
 13
 14
 15def wait_for_connectivity(
 16    on_connected: Callable[[], Any],
 17    on_cancel: Callable[[], Any] | None = None,
 18) -> None:
 19    """Wait for the engine to establish a master-server connection.
 20
 21    If need be, shows a window to keep the user informed of connectivity
 22    state and allows the user to cancel the operation. Note that canceling
 23    does not prevent the engine from continuing its attempt to establish
 24    connectivity; it simply cancels the operation that depends on it.
 25    """
 26    plus = bui.app.plus
 27    assert plus is not None
 28
 29    # Quick-out: if we're already connected, don't bother with the UI.
 30    # We do, however, push this call instead of calling it immediately
 31    # so as to be consistent with the waiting path.
 32    if plus.cloud.connected:
 33        bui.pushcall(on_connected)
 34        return
 35
 36    WaitForConnectivityWindow(on_connected=on_connected, on_cancel=on_cancel)
 37
 38
 39class WaitForConnectivityWindow(bui.Window):
 40    """Window informing the user that the game is establishing connectivity."""
 41
 42    def __init__(
 43        self,
 44        on_connected: Callable[[], Any],
 45        on_cancel: Callable[[], Any] | None,
 46    ) -> None:
 47        self._on_connected = on_connected
 48        self._on_cancel = on_cancel
 49        self._width = 650
 50        self._height = 300
 51        super().__init__(
 52            root_widget=bui.containerwidget(
 53                size=(self._width, self._height),
 54                transition='in_scale',
 55                parent=bui.get_special_widget('overlay_stack'),
 56            )
 57        )
 58        bui.textwidget(
 59            parent=self._root_widget,
 60            position=(self._width * 0.5, self._height * 0.65),
 61            size=(0, 0),
 62            scale=1.2,
 63            h_align='center',
 64            v_align='center',
 65            text=bui.Lstr(resource='internal.connectingToPartyText'),
 66            maxwidth=self._width * 0.9,
 67        )
 68        self._info_text = bui.textwidget(
 69            parent=self._root_widget,
 70            position=(self._width * 0.5, self._height * 0.45),
 71            size=(0, 0),
 72            color=(0.6, 0.5, 0.6),
 73            flatness=1.0,
 74            shadow=0.0,
 75            scale=0.75,
 76            h_align='center',
 77            v_align='center',
 78            text='',
 79            maxwidth=self._width * 0.9,
 80        )
 81        self._info_text_str = ''
 82        cancel_button = bui.buttonwidget(
 83            parent=self._root_widget,
 84            autoselect=True,
 85            position=(50, 30),
 86            size=(150, 50),
 87            label=bui.Lstr(resource='cancelText'),
 88            on_activate_call=self._cancel,
 89        )
 90        bui.containerwidget(edit=self._root_widget, cancel_button=cancel_button)
 91        self._update_timer = bui.AppTimer(
 92            0.113, bui.WeakCall(self._update), repeat=True
 93        )
 94
 95    def _update(self) -> None:
 96
 97        plus = bui.app.plus
 98        assert plus is not None
 99
100        if plus.cloud.connected:
101            self._connected()
102            return
103
104        # Show what connectivity is up to if we don't have any published
105        # zone-pings yet (or if we do but there's no transport state to
106        # show yet).
107        if not bui.app.net.zone_pings or not bui.app.net.transport_state:
108            infotext = bui.app.net.connectivity_state
109        else:
110            infotext = bui.app.net.transport_state
111        if infotext != self._info_text_str:
112            self._info_text_str = infotext
113            bui.textwidget(edit=self._info_text, text=infotext)
114
115    def _connected(self) -> None:
116        if not self._root_widget or self._root_widget.transitioning_out:
117            return
118        bui.containerwidget(
119            edit=self._root_widget,
120            transition=('out_scale'),
121        )
122        bui.pushcall(self._on_connected)
123
124    def _cancel(self) -> None:
125        if not self._root_widget or self._root_widget.transitioning_out:
126            return
127        bui.containerwidget(
128            edit=self._root_widget,
129            transition=('out_scale'),
130        )
131        if self._on_cancel is not None:
132            bui.pushcall(self._on_cancel)