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
 23    canceling does not prevent the engine from continuing its attempt to
 24    establish connectivity; it simply cancels the operation that is
 25    waiting for connectivity.
 26    """
 27    plus = bui.app.plus
 28    assert plus is not None
 29
 30    # Quick-out: if we're already connected, don't bother with the UI.
 31    # We do, however, push this call instead of calling it immediately
 32    # so as to be consistent with the waiting path.
 33    if plus.cloud.connected:
 34        bui.pushcall(on_connected)
 35        return
 36
 37    WaitForConnectivityWindow(on_connected=on_connected, on_cancel=on_cancel)
 38
 39
 40class WaitForConnectivityWindow(bui.Window):
 41    """Window informing the user that the game is establishing connectivity."""
 42
 43    def __init__(
 44        self,
 45        on_connected: Callable[[], Any],
 46        on_cancel: Callable[[], Any] | None,
 47    ) -> None:
 48        self._on_connected = on_connected
 49        self._on_cancel = on_cancel
 50        self._width = 650
 51        self._height = 300
 52        super().__init__(
 53            root_widget=bui.containerwidget(
 54                size=(self._width, self._height),
 55                transition='in_scale',
 56                parent=bui.get_special_widget('overlay_stack'),
 57            )
 58        )
 59        bui.textwidget(
 60            parent=self._root_widget,
 61            position=(self._width * 0.5, self._height * 0.7),
 62            size=(0, 0),
 63            scale=1.2,
 64            h_align='center',
 65            v_align='center',
 66            text=bui.Lstr(resource='internal.connectingToPartyText'),
 67            maxwidth=self._width * 0.9,
 68        )
 69
 70        self._spinner = bui.spinnerwidget(
 71            parent=self._root_widget,
 72            position=(self._width * 0.5, self._height * 0.54),
 73            style='bomb',
 74            size=48,
 75        )
 76
 77        self._info_text = bui.textwidget(
 78            parent=self._root_widget,
 79            position=(self._width * 0.5, self._height * 0.4),
 80            size=(0, 0),
 81            color=(0.6, 0.5, 0.6),
 82            flatness=1.0,
 83            shadow=0.0,
 84            scale=0.75,
 85            h_align='center',
 86            v_align='center',
 87            text='',
 88            maxwidth=self._width * 0.9,
 89        )
 90        self._info_text_str = ''
 91        cancel_button = bui.buttonwidget(
 92            parent=self._root_widget,
 93            autoselect=True,
 94            position=(50, 30),
 95            size=(150, 50),
 96            label=bui.Lstr(resource='cancelText'),
 97            on_activate_call=self._cancel,
 98        )
 99        bui.containerwidget(edit=self._root_widget, cancel_button=cancel_button)
100        self._update_timer = bui.AppTimer(
101            0.113, bui.WeakCall(self._update), repeat=True
102        )
103
104    def _update(self) -> None:
105
106        plus = bui.app.plus
107        assert plus is not None
108
109        if plus.cloud.connected:
110            self._connected()
111            return
112
113        # Show what connectivity is up to if we don't have any published
114        # zone-pings yet (or if we do but there's no transport state to
115        # show yet).
116        if not bui.app.net.zone_pings or not bui.app.net.transport_state:
117            infotext = bui.app.net.connectivity_state
118        else:
119            infotext = bui.app.net.transport_state
120        if infotext != self._info_text_str:
121            self._info_text_str = infotext
122            bui.textwidget(edit=self._info_text, text=infotext)
123
124    def _connected(self) -> None:
125        if not self._root_widget or self._root_widget.transitioning_out:
126            return
127
128        # Show 'connected.' and kill the spinner for the brief moment
129        # we're visible on our way out.
130        bui.textwidget(
131            edit=self._info_text, text=bui.Lstr(resource='remote_app.connected')
132        )
133        if self._spinner:
134            self._spinner.delete()
135
136        bui.containerwidget(
137            edit=self._root_widget,
138            transition=('out_scale'),
139        )
140        bui.pushcall(self._on_connected)
141
142    def _cancel(self) -> None:
143        if not self._root_widget or self._root_widget.transitioning_out:
144            return
145        bui.containerwidget(
146            edit=self._root_widget,
147            transition=('out_scale'),
148        )
149        if self._on_cancel is not None:
150            bui.pushcall(self._on_cancel)