# Released under the MIT License. See LICENSE for details.
#
"""V2 account ui bits."""
from __future__ import annotations # Docs-generation hack.
import time
import asyncio
import logging
from efro.util import strip_exception_tracebacks
import bacommon.cloud
import bauiv1 as bui
from bauiv1 import _commonassets, classicassets
from bauiv1 import builtinassets
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
self._idprefix = bui.app.ui_v1.new_id_prefix('resourcetypeinfo')
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}', _commonassets.strings.status.loading)],
# ),
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=_commonassets.strings.actions.cancel,
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.WeakCallStrict(self._update), repeat=True
)
bui.pushcall(bui.WeakCallStrict(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
self._message_in_flight = True
bui.app.create_async_task(self._run_sign_in(), name='v2proxy sign-in')
def _get_server_address(self) -> str:
plus = bui.app.plus
assert plus is not None
out = plus.get_master_server_address()
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,
)
async def _run_sign_in(self) -> None:
"""Drive the full sign-in flow: request, show UI, poll status."""
plus = bui.app.plus
assert plus is not None
try:
response = await plus.cloud.send_message_async(
bacommon.cloud.LoginProxyRequestMessage()
)
except Exception as exc:
# Something went wrong; show an error message.
self._message_in_flight = False
self._complete = True
if self._root_widget:
self._set_error_state(f'response exc ({type(exc).__name__})')
# We're done with the exception, so strip its tracebacks
# to avoid reference cycles.
strip_exception_tracebacks(exc)
return
self._message_in_flight = False
self._complete = True
# Stop here if our window died while we were waiting.
if not self._root_widget:
return
# 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=_commonassets.strings.status.please_wait,
)
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, poll for results now.
self._proxyid = response.proxyid
self._proxykey = response.proxykey
await self._poll_sign_in_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=classicassets.strings.account.v2_link_instructions,
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,
id=f'{self._idprefix}|address',
position=(
(self._width * 0.5 - button_width * 0.5),
self._height - 185,
),
autoselect=True,
size=(button_width, 60),
label=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=address_pretty,
flatness=1.0,
maxwidth=self._width,
scale=0.75,
h_align='center',
v_align='center',
autoselect=True,
on_activate_call=bui.CallStrict(
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),
)
async def _poll_sign_in_status(self) -> None:
"""Poll the login-proxy until it succeeds, fails, or we die."""
plus = bui.app.plus
assert plus is not None
assert self._proxyid is not None
assert self._proxykey is not None
while True:
await asyncio.sleep(STATUS_CHECK_INTERVAL_SECONDS)
# Stop if our window has died or is on its way out.
if not self._root_widget or self._root_widget.transitioning_out:
return
try:
response = await plus.cloud.send_message_async(
bacommon.cloud.LoginProxyStateQueryMessage(
proxyid=self._proxyid, proxykey=self._proxykey
)
)
except Exception as exc:
# Errors here are generally transient; keep polling.
# We're done with the exception, so strip its
# tracebacks to avoid reference cycles.
strip_exception_tracebacks(exc)
continue
if not self._root_widget:
return
if response.state is response.State.FAIL:
logging.info('LoginProxy failed.')
builtinassets.audio.error.get().play()
bui.screenmessage(
_commonassets.strings.values.error, color=(1, 0, 0)
)
self._done()
return
# If we got a token, set ourself as signed in. Hooray!
if response.state is response.State.SUCCESS:
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. Fire-and-forget; not a huge
# deal if it fails.
plus.cloud.send_message_cb(
bacommon.cloud.LoginProxyCompleteMessage(
proxyid=self._proxyid
),
on_response=bui.WeakCallPartial(
self._proxy_complete_response
),
)
self._done()
return
# Otherwise (State.WAITING) just poll again.
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(
_commonassets.strings.status.copied_to_clipboard,
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