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