Source code for bauiv1lib.account.v2proxy

# Released under the MIT License. See LICENSE for details.
#
"""V2 account ui bits."""

from __future__ import annotations

import time
import logging

from efro.error import CommunicationError
import bacommon.cloud
import bauiv1 as bui

STATUS_CHECK_INTERVAL_SECONDS = 2.0


[docs] class V2ProxySignInWindow(bui.Window): """A window allowing signing in to a v2 account.""" def __init__(self, origin_widget: bui.Widget): self._width = 600 self._height = 550 self._proxyid: str | None = None self._proxykey: str | None = None self._overlay_web_browser_open = False assert bui.app.classic is not None uiscale = bui.app.ui_v1.uiscale super().__init__( root_widget=bui.containerwidget( size=(self._width, self._height), transition='in_scale', scale_origin_stack_offset=( origin_widget.get_screen_space_center() ), scale=( 1.16 if uiscale is bui.UIScale.SMALL else 1.0 if uiscale is bui.UIScale.MEDIUM else 0.9 ), ) ) self._loading_spinner = bui.spinnerwidget( parent=self._root_widget, position=(self._width * 0.5, self._height * 0.5), size=60, style='bomb', ) self._state_text = bui.textwidget( parent=self._root_widget, position=(self._width * 0.5, self._height * 0.6), h_align='center', v_align='center', size=(0, 0), scale=1.4, maxwidth=0.9 * self._width, # text=bui.Lstr( # value='${A}...', # subs=[('${A}', bui.Lstr(resource='loadingText'))], # ), text='', color=(1, 1, 1), ) self._sub_state_text = bui.textwidget( parent=self._root_widget, position=(self._width * 0.5, self._height * 0.55), h_align='center', v_align='top', scale=0.85, size=(0, 0), maxwidth=0.9 * self._width, text='', ) self._sub_state_text2 = bui.textwidget( parent=self._root_widget, position=(self._width * 0.1, self._height * 0.3), h_align='left', v_align='top', scale=0.7, size=(0, 0), maxwidth=0.9 * self._width, text='', ) self._cancel_button = bui.buttonwidget( parent=self._root_widget, position=(30, self._height - 65), size=(130, 50), scale=0.8, label=bui.Lstr(resource='cancelText'), on_activate_call=self._done, autoselect=True, ) bui.containerwidget( edit=self._root_widget, cancel_button=self._cancel_button ) self._message_in_flight = False self._complete = False self._connection_wait_timeout_time = time.monotonic() + 10.0 self._update_timer = bui.AppTimer( 0.371, bui.WeakCall(self._update), repeat=True ) bui.pushcall(bui.WeakCall(self._update)) def _update(self) -> None: plus = bui.app.plus assert plus is not None # If we've opened an overlay web browser, all we do is kill # ourselves when it closes. if self._overlay_web_browser_open: if not bui.overlay_web_browser_is_open(): self._overlay_web_browser_open = False self._done() return if self._message_in_flight or self._complete: return now = time.monotonic() # Spin for a moment if it looks like we have no server # connection; it might still be getting on its feet. if ( not plus.cloud.connected and now < self._connection_wait_timeout_time ): return plus.cloud.send_message_cb( bacommon.cloud.LoginProxyRequestMessage(), on_response=bui.WeakCall(self._on_proxy_request_response), ) self._message_in_flight = True def _get_server_address(self) -> str: plus = bui.app.plus assert plus is not None out = plus.get_master_server_address(version=2) assert isinstance(out, str) return out def _set_error_state(self, error_location: str) -> None: msaddress = self._get_server_address() addr = msaddress.removeprefix('https://') bui.spinnerwidget(edit=self._loading_spinner, visible=False) bui.textwidget( edit=self._state_text, text=f'Unable to connect to {addr}.', color=(1, 0, 0), ) support_email = 'support@froemling.net' bui.textwidget( edit=self._sub_state_text, text=( f'Usually this means your internet is down.\n' f'Please contact {support_email} if this is not the case.' ), color=(1, 0, 0), ) bui.textwidget( edit=self._sub_state_text2, text=( f'debug-info:\n' f' error-location: {error_location}\n' f' connectivity: {bui.app.net.connectivity_state}\n' f' transport: {bui.app.net.transport_state}' ), color=(0.8, 0.2, 0.3), flatness=1.0, shadow=0.0, ) def _on_proxy_request_response( self, response: bacommon.cloud.LoginProxyRequestResponse | Exception ) -> None: plus = bui.app.plus assert plus is not None if not self._message_in_flight: logging.warning( 'v2proxy got _on_proxy_request_response' ' without _message_in_flight set; unexpected.' ) self._message_in_flight = False # Something went wrong. Show an error message and schedule retry. if isinstance(response, Exception): self._set_error_state(f'response exc ({type(response).__name__})') self._complete = True return self._complete = True # Clear out stuff we use to show progress/errors. self._loading_spinner.delete() self._sub_state_text.delete() self._sub_state_text2.delete() # If we have overlay-web-browser functionality, bring up # an inline sign-in dialog. if bui.overlay_web_browser_is_supported(): bui.textwidget( edit=self._state_text, text=bui.Lstr(resource='pleaseWaitText'), ) self._show_overlay_sign_in_ui(response) self._overlay_web_browser_open = True else: # Otherwise just show link-button/qr-code for the sign-in. self._state_text.delete() self._show_standard_sign_in_ui(response) # In either case, start querying for results now. self._proxyid = response.proxyid self._proxykey = response.proxykey bui.apptimer( STATUS_CHECK_INTERVAL_SECONDS, bui.WeakCall(self._ask_for_status) ) def _show_overlay_sign_in_ui( self, response: bacommon.cloud.LoginProxyRequestResponse ) -> None: msaddress = self._get_server_address() address = msaddress + response.url_overlay bui.overlay_web_browser_open_url(address) def _show_standard_sign_in_ui( self, response: bacommon.cloud.LoginProxyRequestResponse ) -> None: msaddress = self._get_server_address() # Show link(s) the user can use to sign in. address = msaddress + response.url address_pretty = address.removeprefix('https://') assert bui.app.classic is not None bui.textwidget( parent=self._root_widget, position=(self._width * 0.5, self._height - 95), size=(0, 0), text=bui.Lstr( resource='accountSettingsWindow.v2LinkInstructionsText' ), color=bui.app.ui_v1.title_color, maxwidth=self._width * 0.9, h_align='center', v_align='center', ) button_width = 450 if bui.is_browser_likely_available(): bui.buttonwidget( parent=self._root_widget, position=( (self._width * 0.5 - button_width * 0.5), self._height - 185, ), autoselect=True, size=(button_width, 60), label=bui.Lstr(value=address_pretty), color=(0.55, 0.5, 0.6), textcolor=(0.75, 0.7, 0.8), on_activate_call=lambda: bui.open_url(address), ) qroffs = 0.0 else: bui.textwidget( parent=self._root_widget, position=(self._width * 0.5 - 200, self._height - 180), size=(button_width - 50, 50), text=bui.Lstr(value=address_pretty), flatness=1.0, maxwidth=self._width, scale=0.75, h_align='center', v_align='center', autoselect=True, on_activate_call=bui.Call(self._copy_link, address_pretty), selectable=True, ) qroffs = 20.0 qr_size = 270 bui.imagewidget( parent=self._root_widget, position=( self._width * 0.5 - qr_size * 0.5, self._height * 0.36 + qroffs - qr_size * 0.5, ), size=(qr_size, qr_size), texture=bui.get_qrcode_texture(address), ) def _ask_for_status(self) -> None: assert self._proxyid is not None assert self._proxykey is not None assert bui.app.plus is not None bui.app.plus.cloud.send_message_cb( bacommon.cloud.LoginProxyStateQueryMessage( proxyid=self._proxyid, proxykey=self._proxykey ), on_response=bui.WeakCall(self._got_status), ) def _got_status( self, response: bacommon.cloud.LoginProxyStateQueryResponse | Exception ) -> None: if ( isinstance(response, bacommon.cloud.LoginProxyStateQueryResponse) and response.state is response.State.FAIL ): logging.info('LoginProxy failed.') bui.getsound('error').play() bui.screenmessage(bui.Lstr(resource='errorText'), color=(1, 0, 0)) self._done() return # If we got a token, set ourself as signed in. Hooray! if ( isinstance(response, bacommon.cloud.LoginProxyStateQueryResponse) and response.state is response.State.SUCCESS ): plus = bui.app.plus assert plus is not None assert response.credentials is not None plus.accounts.set_primary_credentials(response.credentials) # As a courtesy, tell the server we're done with this proxy # so it can clean up (not a huge deal if this fails) assert self._proxyid is not None try: plus.cloud.send_message_cb( bacommon.cloud.LoginProxyCompleteMessage( proxyid=self._proxyid ), on_response=bui.WeakCall(self._proxy_complete_response), ) except CommunicationError: pass except Exception: logging.warning( 'Unexpected error sending login-proxy-complete message', exc_info=True, ) self._done() return # If we're still waiting, ask again soon. if ( isinstance(response, Exception) or response.state is response.State.WAITING ): bui.apptimer( STATUS_CHECK_INTERVAL_SECONDS, bui.WeakCall(self._ask_for_status), ) def _proxy_complete_response(self, response: None | Exception) -> None: del response # Not used. # We could do something smart like retry on exceptions here, but # this isn't critical so we'll just let anything slide. def _copy_link(self, link: str) -> None: if bui.clipboard_is_supported(): bui.clipboard_set_text(link) bui.screenmessage( bui.Lstr(resource='copyConfirmText'), color=(0, 1, 0) ) def _done(self) -> None: # no-op if our underlying widget is dead or on its way out. if not self._root_widget or self._root_widget.transitioning_out: return # If we've got an inline browser up, tell it to close. if self._overlay_web_browser_open: bui.overlay_web_browser_close() bui.containerwidget(edit=self._root_widget, transition='out_scale')
# Docs-generation hack; import some stuff that we likely only forward-declared # in our actual source code so that docs tools can find it. from typing import (Coroutine, Any, Literal, Callable, Generator, Awaitable, Sequence, Self) import asyncio from concurrent.futures import Future from pathlib import Path from enum import Enum