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.7),
 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
 69        self._spinner = bui.spinnerwidget(
 70            parent=self._root_widget,
 71            position=(self._width * 0.5, self._height * 0.54),
 72        )
 73
 74        self._info_text = bui.textwidget(
 75            parent=self._root_widget,
 76            position=(self._width * 0.5, self._height * 0.4),
 77            size=(0, 0),
 78            color=(0.6, 0.5, 0.6),
 79            flatness=1.0,
 80            shadow=0.0,
 81            scale=0.75,
 82            h_align='center',
 83            v_align='center',
 84            text='',
 85            maxwidth=self._width * 0.9,
 86        )
 87        self._info_text_str = ''
 88        cancel_button = bui.buttonwidget(
 89            parent=self._root_widget,
 90            autoselect=True,
 91            position=(50, 30),
 92            size=(150, 50),
 93            label=bui.Lstr(resource='cancelText'),
 94            on_activate_call=self._cancel,
 95        )
 96        bui.containerwidget(edit=self._root_widget, cancel_button=cancel_button)
 97        self._update_timer = bui.AppTimer(
 98            0.113, bui.WeakCall(self._update), repeat=True
 99        )
100
101    def _update(self) -> None:
102
103        plus = bui.app.plus
104        assert plus is not None
105
106        if plus.cloud.connected:
107            self._connected()
108            return
109
110        # Show what connectivity is up to if we don't have any published
111        # zone-pings yet (or if we do but there's no transport state to
112        # show yet).
113        if not bui.app.net.zone_pings or not bui.app.net.transport_state:
114            infotext = bui.app.net.connectivity_state
115        else:
116            infotext = bui.app.net.transport_state
117        if infotext != self._info_text_str:
118            self._info_text_str = infotext
119            bui.textwidget(edit=self._info_text, text=infotext)
120
121    def _connected(self) -> None:
122        if not self._root_widget or self._root_widget.transitioning_out:
123            return
124
125        # Show 'connected.' and kill the spinner for the brief moment
126        # we're visible on our way out.
127        bui.textwidget(
128            edit=self._info_text, text=bui.Lstr(resource='remote_app.connected')
129        )
130        if self._spinner:
131            self._spinner.delete()
132
133        bui.containerwidget(
134            edit=self._root_widget,
135            transition=('out_scale'),
136        )
137        bui.pushcall(self._on_connected)
138
139    def _cancel(self) -> None:
140        if not self._root_widget or self._root_widget.transitioning_out:
141            return
142        bui.containerwidget(
143            edit=self._root_widget,
144            transition=('out_scale'),
145        )
146        if self._on_cancel is not None:
147            bui.pushcall(self._on_cancel)