mirror of
https://github.com/hypervortex/VH-Bombsquad-Modded-Server-Files
synced 2026-08-15 13:04:50 +00:00
1419 lines
49 KiB
Python
1419 lines
49 KiB
Python
# Released under the MIT License. See LICENSE for details.
|
|
#
|
|
# pylint: disable=too-many-lines
|
|
"""Defines the public tab in the gather UI."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import time
|
|
import threading
|
|
from enum import Enum
|
|
from dataclasses import dataclass
|
|
from typing import TYPE_CHECKING, cast
|
|
|
|
import ba
|
|
import ba.internal
|
|
from bastd.ui.gather import GatherTab
|
|
|
|
if TYPE_CHECKING:
|
|
from typing import Callable, Any
|
|
from bastd.ui.gather import GatherWindow
|
|
|
|
# Print a bit of info about pings, queries, etc.
|
|
DEBUG_SERVER_COMMUNICATION = False
|
|
DEBUG_PROCESSING = False
|
|
|
|
|
|
class SubTabType(Enum):
|
|
"""Available sub-tabs."""
|
|
|
|
JOIN = 'join'
|
|
HOST = 'host'
|
|
|
|
|
|
@dataclass
|
|
class PartyEntry:
|
|
"""Info about a public party."""
|
|
|
|
address: str
|
|
index: int
|
|
queue: str | None = None
|
|
port: int = -1
|
|
name: str = ''
|
|
size: int = -1
|
|
size_max: int = -1
|
|
claimed: bool = False
|
|
ping: float | None = None
|
|
ping_interval: float = -1.0
|
|
next_ping_time: float = -1.0
|
|
ping_attempts: int = 0
|
|
ping_responses: int = 0
|
|
stats_addr: str | None = None
|
|
clean_display_index: int | None = None
|
|
|
|
def get_key(self) -> str:
|
|
"""Return the key used to store this party."""
|
|
return f'{self.address}_{self.port}'
|
|
|
|
|
|
class UIRow:
|
|
"""Wrangles UI for a row in the party list."""
|
|
|
|
def __init__(self) -> None:
|
|
self._name_widget: ba.Widget | None = None
|
|
self._size_widget: ba.Widget | None = None
|
|
self._ping_widget: ba.Widget | None = None
|
|
self._stats_button: ba.Widget | None = None
|
|
|
|
def __del__(self) -> None:
|
|
self._clear()
|
|
|
|
def _clear(self) -> None:
|
|
for widget in [
|
|
self._name_widget,
|
|
self._size_widget,
|
|
self._ping_widget,
|
|
self._stats_button,
|
|
]:
|
|
if widget:
|
|
widget.delete()
|
|
|
|
def update(
|
|
self,
|
|
index: int,
|
|
party: PartyEntry,
|
|
sub_scroll_width: float,
|
|
sub_scroll_height: float,
|
|
lineheight: float,
|
|
columnwidget: ba.Widget,
|
|
join_text: ba.Widget,
|
|
filter_text: ba.Widget,
|
|
existing_selection: Selection | None,
|
|
tab: PublicGatherTab,
|
|
) -> None:
|
|
"""Update for the given data."""
|
|
# pylint: disable=too-many-locals
|
|
|
|
# Quick-out: if we've been marked clean for a certain index and
|
|
# we're still at that index, we're done.
|
|
if party.clean_display_index == index:
|
|
return
|
|
|
|
ping_good = ba.internal.get_v1_account_misc_read_val('pingGood', 100)
|
|
ping_med = ba.internal.get_v1_account_misc_read_val('pingMed', 500)
|
|
|
|
self._clear()
|
|
hpos = 20
|
|
vpos = sub_scroll_height - lineheight * index - 50
|
|
self._name_widget = ba.textwidget(
|
|
text=ba.Lstr(value=party.name),
|
|
parent=columnwidget,
|
|
size=(sub_scroll_width * 0.63, 20),
|
|
position=(0 + hpos, 4 + vpos),
|
|
selectable=True,
|
|
on_select_call=ba.WeakCall(
|
|
tab.set_public_party_selection,
|
|
Selection(party.get_key(), SelectionComponent.NAME),
|
|
),
|
|
on_activate_call=ba.WeakCall(tab.on_public_party_activate, party),
|
|
click_activate=True,
|
|
maxwidth=sub_scroll_width * 0.45,
|
|
corner_scale=1.4,
|
|
autoselect=True,
|
|
color=(1, 1, 1, 0.3 if party.ping is None else 1.0),
|
|
h_align='left',
|
|
v_align='center',
|
|
)
|
|
ba.widget(
|
|
edit=self._name_widget,
|
|
left_widget=join_text,
|
|
show_buffer_top=64.0,
|
|
show_buffer_bottom=64.0,
|
|
)
|
|
if existing_selection == Selection(
|
|
party.get_key(), SelectionComponent.NAME
|
|
):
|
|
ba.containerwidget(
|
|
edit=columnwidget, selected_child=self._name_widget
|
|
)
|
|
if party.stats_addr:
|
|
url = party.stats_addr.replace(
|
|
'${ACCOUNT}',
|
|
ba.internal.get_v1_account_misc_read_val_2(
|
|
'resolvedAccountID', 'UNKNOWN'
|
|
),
|
|
)
|
|
self._stats_button = ba.buttonwidget(
|
|
color=(0.3, 0.6, 0.94),
|
|
textcolor=(1.0, 1.0, 1.0),
|
|
label=ba.Lstr(resource='statsText'),
|
|
parent=columnwidget,
|
|
autoselect=True,
|
|
on_activate_call=ba.Call(ba.open_url, url),
|
|
on_select_call=ba.WeakCall(
|
|
tab.set_public_party_selection,
|
|
Selection(party.get_key(), SelectionComponent.STATS_BUTTON),
|
|
),
|
|
size=(120, 40),
|
|
position=(sub_scroll_width * 0.66 + hpos, 1 + vpos),
|
|
scale=0.9,
|
|
)
|
|
if existing_selection == Selection(
|
|
party.get_key(), SelectionComponent.STATS_BUTTON
|
|
):
|
|
ba.containerwidget(
|
|
edit=columnwidget, selected_child=self._stats_button
|
|
)
|
|
|
|
self._size_widget = ba.textwidget(
|
|
text=str(party.size) + '/' + str(party.size_max),
|
|
parent=columnwidget,
|
|
size=(0, 0),
|
|
position=(sub_scroll_width * 0.86 + hpos, 20 + vpos),
|
|
scale=0.7,
|
|
color=(0.8, 0.8, 0.8),
|
|
h_align='right',
|
|
v_align='center',
|
|
)
|
|
|
|
if index == 0:
|
|
ba.widget(edit=self._name_widget, up_widget=filter_text)
|
|
if self._stats_button:
|
|
ba.widget(edit=self._stats_button, up_widget=filter_text)
|
|
|
|
self._ping_widget = ba.textwidget(
|
|
parent=columnwidget,
|
|
size=(0, 0),
|
|
position=(sub_scroll_width * 0.94 + hpos, 20 + vpos),
|
|
scale=0.7,
|
|
h_align='right',
|
|
v_align='center',
|
|
)
|
|
if party.ping is None:
|
|
ba.textwidget(
|
|
edit=self._ping_widget, text='-', color=(0.5, 0.5, 0.5)
|
|
)
|
|
else:
|
|
ba.textwidget(
|
|
edit=self._ping_widget,
|
|
text=str(int(party.ping)),
|
|
color=(0, 1, 0)
|
|
if party.ping <= ping_good
|
|
else (1, 1, 0)
|
|
if party.ping <= ping_med
|
|
else (1, 0, 0),
|
|
)
|
|
|
|
party.clean_display_index = index
|
|
|
|
|
|
@dataclass
|
|
class State:
|
|
"""State saved/restored only while the app is running."""
|
|
|
|
sub_tab: SubTabType = SubTabType.JOIN
|
|
parties: list[tuple[str, PartyEntry]] | None = None
|
|
next_entry_index: int = 0
|
|
filter_value: str = ''
|
|
have_server_list_response: bool = False
|
|
have_valid_server_list: bool = False
|
|
|
|
|
|
class SelectionComponent(Enum):
|
|
"""Describes what part of an entry is selected."""
|
|
|
|
NAME = 'name'
|
|
STATS_BUTTON = 'stats_button'
|
|
|
|
|
|
@dataclass
|
|
class Selection:
|
|
"""Describes the currently selected list element."""
|
|
|
|
entry_key: str
|
|
component: SelectionComponent
|
|
|
|
|
|
class AddrFetchThread(threading.Thread):
|
|
"""Thread for fetching an address in the bg."""
|
|
|
|
def __init__(self, call: Callable[[Any], Any]):
|
|
super().__init__()
|
|
self._call = call
|
|
|
|
def run(self) -> None:
|
|
try:
|
|
# FIXME: Update this to work with IPv6 at some point.
|
|
import socket
|
|
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.connect(('8.8.8.8', 80))
|
|
val = sock.getsockname()[0]
|
|
sock.close()
|
|
ba.pushcall(ba.Call(self._call, val), from_other_thread=True)
|
|
except Exception as exc:
|
|
from efro.error import is_udp_communication_error
|
|
|
|
# Ignore expected network errors; log others.
|
|
if is_udp_communication_error(exc):
|
|
pass
|
|
else:
|
|
ba.print_exception()
|
|
|
|
|
|
class PingThread(threading.Thread):
|
|
"""Thread for sending out game pings."""
|
|
|
|
def __init__(
|
|
self,
|
|
address: str,
|
|
port: int,
|
|
call: Callable[[str, int, float | None], int | None],
|
|
):
|
|
super().__init__()
|
|
self._address = address
|
|
self._port = port
|
|
self._call = call
|
|
|
|
def run(self) -> None:
|
|
ba.app.ping_thread_count += 1
|
|
sock: socket.socket | None = None
|
|
try:
|
|
import socket
|
|
from ba.internal import get_ip_address_type
|
|
|
|
socket_type = get_ip_address_type(self._address)
|
|
sock = socket.socket(socket_type, socket.SOCK_DGRAM)
|
|
sock.connect((self._address, self._port))
|
|
|
|
accessible = False
|
|
starttime = time.time()
|
|
|
|
# Send a few pings and wait a second for
|
|
# a response.
|
|
sock.settimeout(1)
|
|
for _i in range(3):
|
|
sock.send(b'\x0b')
|
|
result: bytes | None
|
|
try:
|
|
# 11: BA_PACKET_SIMPLE_PING
|
|
result = sock.recv(10)
|
|
except Exception:
|
|
result = None
|
|
if result == b'\x0c':
|
|
# 12: BA_PACKET_SIMPLE_PONG
|
|
accessible = True
|
|
break
|
|
time.sleep(1)
|
|
ping = (time.time() - starttime) * 1000.0
|
|
ba.pushcall(
|
|
ba.Call(
|
|
self._call,
|
|
self._address,
|
|
self._port,
|
|
ping if accessible else None,
|
|
),
|
|
from_other_thread=True,
|
|
)
|
|
except Exception as exc:
|
|
from efro.error import is_udp_communication_error
|
|
|
|
if is_udp_communication_error(exc):
|
|
pass
|
|
else:
|
|
ba.print_exception('Error on gather ping', once=True)
|
|
finally:
|
|
try:
|
|
if sock is not None:
|
|
sock.close()
|
|
except Exception:
|
|
ba.print_exception('Error on gather ping cleanup', once=True)
|
|
|
|
ba.app.ping_thread_count -= 1
|
|
|
|
|
|
class PublicGatherTab(GatherTab):
|
|
"""The public tab in the gather UI"""
|
|
|
|
def __init__(self, window: GatherWindow) -> None:
|
|
super().__init__(window)
|
|
self._container: ba.Widget | None = None
|
|
self._join_text: ba.Widget | None = None
|
|
self._host_text: ba.Widget | None = None
|
|
self._filter_text: ba.Widget | None = None
|
|
self._local_address: str | None = None
|
|
self._last_connect_attempt_time: float | None = None
|
|
self._sub_tab: SubTabType = SubTabType.JOIN
|
|
self._selection: Selection | None = None
|
|
self._refreshing_list = False
|
|
self._update_timer: ba.Timer | None = None
|
|
self._host_scrollwidget: ba.Widget | None = None
|
|
self._host_name_text: ba.Widget | None = None
|
|
self._host_toggle_button: ba.Widget | None = None
|
|
self._last_server_list_query_time: float | None = None
|
|
self._join_list_column: ba.Widget | None = None
|
|
self._join_status_text: ba.Widget | None = None
|
|
self._host_max_party_size_value: ba.Widget | None = None
|
|
self._host_max_party_size_minus_button: (ba.Widget | None) = None
|
|
self._host_max_party_size_plus_button: (ba.Widget | None) = None
|
|
self._host_status_text: ba.Widget | None = None
|
|
self._signed_in = False
|
|
self._ui_rows: list[UIRow] = []
|
|
self._refresh_ui_row = 0
|
|
self._have_user_selected_row = False
|
|
self._first_valid_server_list_time: float | None = None
|
|
|
|
# Parties indexed by id:
|
|
self._parties: dict[str, PartyEntry] = {}
|
|
|
|
# Parties sorted in display order:
|
|
self._parties_sorted: list[tuple[str, PartyEntry]] = []
|
|
self._party_lists_dirty = True
|
|
|
|
# Sorted parties with filter applied:
|
|
self._parties_displayed: dict[str, PartyEntry] = {}
|
|
|
|
self._next_entry_index = 0
|
|
self._have_server_list_response = False
|
|
self._have_valid_server_list = False
|
|
self._filter_value = ''
|
|
self._pending_party_infos: list[dict[str, Any]] = []
|
|
self._last_sub_scroll_height = 0.0
|
|
|
|
def on_activate(
|
|
self,
|
|
parent_widget: ba.Widget,
|
|
tab_button: ba.Widget,
|
|
region_width: float,
|
|
region_height: float,
|
|
region_left: float,
|
|
region_bottom: float,
|
|
) -> ba.Widget:
|
|
c_width = region_width
|
|
c_height = region_height - 20
|
|
self._container = ba.containerwidget(
|
|
parent=parent_widget,
|
|
position=(
|
|
region_left,
|
|
region_bottom + (region_height - c_height) * 0.5,
|
|
),
|
|
size=(c_width, c_height),
|
|
background=False,
|
|
selection_loops_to_parent=True,
|
|
)
|
|
v = c_height - 30
|
|
self._join_text = ba.textwidget(
|
|
parent=self._container,
|
|
position=(c_width * 0.5 - 245, v - 13),
|
|
color=(0.6, 1.0, 0.6),
|
|
scale=1.3,
|
|
size=(200, 30),
|
|
maxwidth=250,
|
|
h_align='left',
|
|
v_align='center',
|
|
click_activate=True,
|
|
selectable=True,
|
|
autoselect=True,
|
|
on_activate_call=lambda: self._set_sub_tab(
|
|
SubTabType.JOIN,
|
|
region_width,
|
|
region_height,
|
|
playsound=True,
|
|
),
|
|
text=ba.Lstr(
|
|
resource='gatherWindow.' 'joinPublicPartyDescriptionText'
|
|
),
|
|
)
|
|
self._host_text = ba.textwidget(
|
|
parent=self._container,
|
|
position=(c_width * 0.5 + 45, v - 13),
|
|
color=(0.6, 1.0, 0.6),
|
|
scale=1.3,
|
|
size=(200, 30),
|
|
maxwidth=250,
|
|
h_align='left',
|
|
v_align='center',
|
|
click_activate=True,
|
|
selectable=True,
|
|
autoselect=True,
|
|
on_activate_call=lambda: self._set_sub_tab(
|
|
SubTabType.HOST,
|
|
region_width,
|
|
region_height,
|
|
playsound=True,
|
|
),
|
|
text=ba.Lstr(
|
|
resource='gatherWindow.' 'hostPublicPartyDescriptionText'
|
|
),
|
|
)
|
|
ba.widget(edit=self._join_text, up_widget=tab_button)
|
|
ba.widget(
|
|
edit=self._host_text,
|
|
left_widget=self._join_text,
|
|
up_widget=tab_button,
|
|
)
|
|
ba.widget(edit=self._join_text, right_widget=self._host_text)
|
|
|
|
# Attempt to fetch our local address so we have it for error messages.
|
|
if self._local_address is None:
|
|
AddrFetchThread(ba.WeakCall(self._fetch_local_addr_cb)).start()
|
|
|
|
self._set_sub_tab(self._sub_tab, region_width, region_height)
|
|
self._update_timer = ba.Timer(
|
|
0.1,
|
|
ba.WeakCall(self._update),
|
|
repeat=True,
|
|
timetype=ba.TimeType.REAL,
|
|
)
|
|
return self._container
|
|
|
|
def on_deactivate(self) -> None:
|
|
self._update_timer = None
|
|
|
|
def save_state(self) -> None:
|
|
|
|
# Save off a small number of parties with the lowest ping; we'll
|
|
# display these immediately when our UI comes back up which should
|
|
# be enough to make things feel nice and crisp while we do a full
|
|
# server re-query or whatnot.
|
|
ba.app.ui.window_states[type(self)] = State(
|
|
sub_tab=self._sub_tab,
|
|
parties=[(i, copy.copy(p)) for i, p in self._parties_sorted[:40]],
|
|
next_entry_index=self._next_entry_index,
|
|
filter_value=self._filter_value,
|
|
have_server_list_response=self._have_server_list_response,
|
|
have_valid_server_list=self._have_valid_server_list,
|
|
)
|
|
|
|
def restore_state(self) -> None:
|
|
state = ba.app.ui.window_states.get(type(self))
|
|
if state is None:
|
|
state = State()
|
|
assert isinstance(state, State)
|
|
self._sub_tab = state.sub_tab
|
|
|
|
# Restore the parties we stored.
|
|
if state.parties:
|
|
self._parties = {
|
|
key: copy.copy(party) for key, party in state.parties
|
|
}
|
|
self._parties_sorted = list(self._parties.items())
|
|
self._party_lists_dirty = True
|
|
|
|
self._next_entry_index = state.next_entry_index
|
|
|
|
# FIXME: should save/restore these too?..
|
|
self._have_server_list_response = state.have_server_list_response
|
|
self._have_valid_server_list = state.have_valid_server_list
|
|
self._filter_value = state.filter_value
|
|
|
|
def _set_sub_tab(
|
|
self,
|
|
value: SubTabType,
|
|
region_width: float,
|
|
region_height: float,
|
|
playsound: bool = False,
|
|
) -> None:
|
|
assert self._container
|
|
if playsound:
|
|
ba.playsound(ba.getsound('click01'))
|
|
|
|
# Reset our selection.
|
|
# (prevents selecting something way down the list if we switched away
|
|
# and came back)
|
|
self._selection = None
|
|
self._have_user_selected_row = False
|
|
|
|
# Reset refresh to the top and make sure everything refreshes.
|
|
self._refresh_ui_row = 0
|
|
for party in self._parties.values():
|
|
party.clean_display_index = None
|
|
|
|
self._sub_tab = value
|
|
active_color = (0.6, 1.0, 0.6)
|
|
inactive_color = (0.5, 0.4, 0.5)
|
|
ba.textwidget(
|
|
edit=self._join_text,
|
|
color=active_color if value is SubTabType.JOIN else inactive_color,
|
|
)
|
|
ba.textwidget(
|
|
edit=self._host_text,
|
|
color=active_color if value is SubTabType.HOST else inactive_color,
|
|
)
|
|
|
|
# Clear anything existing in the old sub-tab.
|
|
for widget in self._container.get_children():
|
|
if widget and widget not in {self._host_text, self._join_text}:
|
|
widget.delete()
|
|
|
|
if value is SubTabType.JOIN:
|
|
self._build_join_tab(region_width, region_height)
|
|
|
|
if value is SubTabType.HOST:
|
|
self._build_host_tab(region_width, region_height)
|
|
|
|
def _build_join_tab(
|
|
self, region_width: float, region_height: float
|
|
) -> None:
|
|
c_width = region_width
|
|
c_height = region_height - 20
|
|
sub_scroll_height = c_height - 125
|
|
sub_scroll_width = 830
|
|
v = c_height - 35
|
|
v -= 60
|
|
filter_txt = ba.Lstr(resource='filterText')
|
|
self._filter_text = ba.textwidget(
|
|
parent=self._container,
|
|
text=self._filter_value,
|
|
size=(350, 45),
|
|
position=(290, v - 10),
|
|
h_align='left',
|
|
v_align='center',
|
|
editable=True,
|
|
maxwidth=310,
|
|
description=filter_txt,
|
|
)
|
|
ba.widget(edit=self._filter_text, up_widget=self._join_text)
|
|
ba.textwidget(
|
|
text=filter_txt,
|
|
parent=self._container,
|
|
size=(0, 0),
|
|
position=(270, v + 13),
|
|
maxwidth=150,
|
|
scale=0.8,
|
|
color=(0.5, 0.46, 0.5),
|
|
flatness=1.0,
|
|
h_align='right',
|
|
v_align='center',
|
|
)
|
|
|
|
ba.textwidget(
|
|
text=ba.Lstr(resource='nameText'),
|
|
parent=self._container,
|
|
size=(0, 0),
|
|
position=(90, v - 8),
|
|
maxwidth=60,
|
|
scale=0.6,
|
|
color=(0.5, 0.46, 0.5),
|
|
flatness=1.0,
|
|
h_align='center',
|
|
v_align='center',
|
|
)
|
|
ba.textwidget(
|
|
text=ba.Lstr(resource='gatherWindow.partySizeText'),
|
|
parent=self._container,
|
|
size=(0, 0),
|
|
position=(755, v - 8),
|
|
maxwidth=60,
|
|
scale=0.6,
|
|
color=(0.5, 0.46, 0.5),
|
|
flatness=1.0,
|
|
h_align='center',
|
|
v_align='center',
|
|
)
|
|
ba.textwidget(
|
|
text=ba.Lstr(resource='gatherWindow.pingText'),
|
|
parent=self._container,
|
|
size=(0, 0),
|
|
position=(825, v - 8),
|
|
maxwidth=60,
|
|
scale=0.6,
|
|
color=(0.5, 0.46, 0.5),
|
|
flatness=1.0,
|
|
h_align='center',
|
|
v_align='center',
|
|
)
|
|
v -= sub_scroll_height + 23
|
|
self._host_scrollwidget = scrollw = ba.scrollwidget(
|
|
parent=self._container,
|
|
simple_culling_v=10,
|
|
position=((c_width - sub_scroll_width) * 0.5, v),
|
|
size=(sub_scroll_width, sub_scroll_height),
|
|
claims_up_down=False,
|
|
claims_left_right=True,
|
|
autoselect=True,
|
|
)
|
|
self._join_list_column = ba.containerwidget(
|
|
parent=scrollw,
|
|
background=False,
|
|
size=(400, 400),
|
|
claims_left_right=True,
|
|
)
|
|
self._join_status_text = ba.textwidget(
|
|
parent=self._container,
|
|
text='',
|
|
size=(0, 0),
|
|
scale=0.9,
|
|
flatness=1.0,
|
|
shadow=0.0,
|
|
h_align='center',
|
|
v_align='top',
|
|
maxwidth=c_width,
|
|
color=(0.6, 0.6, 0.6),
|
|
position=(c_width * 0.5, c_height * 0.5),
|
|
)
|
|
|
|
def _build_host_tab(
|
|
self, region_width: float, region_height: float
|
|
) -> None:
|
|
c_width = region_width
|
|
c_height = region_height - 20
|
|
v = c_height - 35
|
|
v -= 25
|
|
is_public_enabled = ba.internal.get_public_party_enabled()
|
|
v -= 30
|
|
|
|
ba.textwidget(
|
|
parent=self._container,
|
|
size=(0, 0),
|
|
h_align='center',
|
|
v_align='center',
|
|
maxwidth=c_width * 0.9,
|
|
scale=0.7,
|
|
flatness=1.0,
|
|
color=(0.5, 0.46, 0.5),
|
|
position=(region_width * 0.5, v + 10),
|
|
text=ba.Lstr(resource='gatherWindow.publicHostRouterConfigText'),
|
|
)
|
|
v -= 30
|
|
|
|
party_name_text = ba.Lstr(
|
|
resource='gatherWindow.partyNameText',
|
|
fallback_resource='editGameListWindow.nameText',
|
|
)
|
|
ba.textwidget(
|
|
parent=self._container,
|
|
size=(0, 0),
|
|
h_align='right',
|
|
v_align='center',
|
|
maxwidth=200,
|
|
scale=0.8,
|
|
color=ba.app.ui.infotextcolor,
|
|
position=(210, v - 9),
|
|
text=party_name_text,
|
|
)
|
|
self._host_name_text = ba.textwidget(
|
|
parent=self._container,
|
|
editable=True,
|
|
size=(535, 40),
|
|
position=(230, v - 30),
|
|
text=ba.app.config.get('Public Party Name', ''),
|
|
maxwidth=494,
|
|
shadow=0.3,
|
|
flatness=1.0,
|
|
description=party_name_text,
|
|
autoselect=True,
|
|
v_align='center',
|
|
corner_scale=1.0,
|
|
)
|
|
|
|
v -= 60
|
|
ba.textwidget(
|
|
parent=self._container,
|
|
size=(0, 0),
|
|
h_align='right',
|
|
v_align='center',
|
|
maxwidth=200,
|
|
scale=0.8,
|
|
color=ba.app.ui.infotextcolor,
|
|
position=(210, v - 9),
|
|
text=ba.Lstr(
|
|
resource='maxPartySizeText',
|
|
fallback_resource='maxConnectionsText',
|
|
),
|
|
)
|
|
self._host_max_party_size_value = ba.textwidget(
|
|
parent=self._container,
|
|
size=(0, 0),
|
|
h_align='center',
|
|
v_align='center',
|
|
scale=1.2,
|
|
color=(1, 1, 1),
|
|
position=(240, v - 9),
|
|
text=str(ba.internal.get_public_party_max_size()),
|
|
)
|
|
btn1 = self._host_max_party_size_minus_button = ba.buttonwidget(
|
|
parent=self._container,
|
|
size=(40, 40),
|
|
on_activate_call=ba.WeakCall(
|
|
self._on_max_public_party_size_minus_press
|
|
),
|
|
position=(280, v - 26),
|
|
label='-',
|
|
autoselect=True,
|
|
)
|
|
btn2 = self._host_max_party_size_plus_button = ba.buttonwidget(
|
|
parent=self._container,
|
|
size=(40, 40),
|
|
on_activate_call=ba.WeakCall(
|
|
self._on_max_public_party_size_plus_press
|
|
),
|
|
position=(350, v - 26),
|
|
label='+',
|
|
autoselect=True,
|
|
)
|
|
v -= 50
|
|
v -= 70
|
|
if is_public_enabled:
|
|
label = ba.Lstr(
|
|
resource='gatherWindow.makePartyPrivateText',
|
|
fallback_resource='gatherWindow.stopAdvertisingText',
|
|
)
|
|
else:
|
|
label = ba.Lstr(
|
|
resource='gatherWindow.makePartyPublicText',
|
|
fallback_resource='gatherWindow.startAdvertisingText',
|
|
)
|
|
self._host_toggle_button = ba.buttonwidget(
|
|
parent=self._container,
|
|
label=label,
|
|
size=(400, 80),
|
|
on_activate_call=self._on_stop_advertising_press
|
|
if is_public_enabled
|
|
else self._on_start_advertizing_press,
|
|
position=(c_width * 0.5 - 200, v),
|
|
autoselect=True,
|
|
up_widget=btn2,
|
|
)
|
|
ba.widget(edit=self._host_name_text, down_widget=btn2)
|
|
ba.widget(edit=btn2, up_widget=self._host_name_text)
|
|
ba.widget(edit=btn1, up_widget=self._host_name_text)
|
|
ba.widget(edit=self._join_text, down_widget=self._host_name_text)
|
|
v -= 10
|
|
self._host_status_text = ba.textwidget(
|
|
parent=self._container,
|
|
text=ba.Lstr(resource='gatherWindow.' 'partyStatusNotPublicText'),
|
|
size=(0, 0),
|
|
scale=0.7,
|
|
flatness=1.0,
|
|
h_align='center',
|
|
v_align='top',
|
|
maxwidth=c_width * 0.9,
|
|
color=(0.6, 0.56, 0.6),
|
|
position=(c_width * 0.5, v),
|
|
)
|
|
v -= 90
|
|
ba.textwidget(
|
|
parent=self._container,
|
|
text=ba.Lstr(resource='gatherWindow.dedicatedServerInfoText'),
|
|
size=(0, 0),
|
|
scale=0.7,
|
|
flatness=1.0,
|
|
h_align='center',
|
|
v_align='center',
|
|
maxwidth=c_width * 0.9,
|
|
color=(0.5, 0.46, 0.5),
|
|
position=(c_width * 0.5, v),
|
|
)
|
|
|
|
# If public sharing is already on,
|
|
# launch a status-check immediately.
|
|
if ba.internal.get_public_party_enabled():
|
|
self._do_status_check()
|
|
|
|
def _on_public_party_query_result(
|
|
self, result: dict[str, Any] | None
|
|
) -> None:
|
|
starttime = time.time()
|
|
self._have_server_list_response = True
|
|
|
|
if result is None:
|
|
self._have_valid_server_list = False
|
|
return
|
|
|
|
if not self._have_valid_server_list:
|
|
self._first_valid_server_list_time = time.time()
|
|
|
|
self._have_valid_server_list = True
|
|
parties_in = result['l']
|
|
|
|
assert isinstance(parties_in, list)
|
|
self._pending_party_infos += parties_in
|
|
|
|
# To avoid causing a stutter here, we do most processing of
|
|
# these entries incrementally in our _update() method.
|
|
# The one thing we do here is prune parties not contained in
|
|
# this result.
|
|
for partyval in list(self._parties.values()):
|
|
partyval.claimed = False
|
|
for party_in in parties_in:
|
|
addr = party_in['a']
|
|
assert isinstance(addr, str)
|
|
port = party_in['p']
|
|
assert isinstance(port, int)
|
|
party_key = f'{addr}_{port}'
|
|
party = self._parties.get(party_key)
|
|
if party is not None:
|
|
party.claimed = True
|
|
self._parties = {
|
|
key: val for key, val in list(self._parties.items()) if val.claimed
|
|
}
|
|
self._parties_sorted = [p for p in self._parties_sorted if p[1].claimed]
|
|
self._party_lists_dirty = True
|
|
|
|
# self._update_server_list()
|
|
if DEBUG_PROCESSING:
|
|
print(
|
|
f'Handled public party query results in '
|
|
f'{time.time()-starttime:.5f}s.'
|
|
)
|
|
|
|
def _update(self) -> None:
|
|
"""Periodic updating."""
|
|
|
|
# Special case: if a party-queue window is up, don't do any of this
|
|
# (keeps things smoother).
|
|
# if ba.app.ui.have_party_queue_window:
|
|
# return
|
|
|
|
if self._sub_tab is SubTabType.JOIN:
|
|
|
|
# Keep our filter-text up to date from the UI.
|
|
text = self._filter_text
|
|
if text:
|
|
filter_value = cast(str, ba.textwidget(query=text))
|
|
if filter_value != self._filter_value:
|
|
self._filter_value = filter_value
|
|
self._party_lists_dirty = True
|
|
|
|
# Also wipe out party clean-row states.
|
|
# (otherwise if a party disappears from a row due to
|
|
# filtering and then reappears on that same row when
|
|
# the filter is removed it may not update)
|
|
for party in self._parties.values():
|
|
party.clean_display_index = None
|
|
|
|
self._query_party_list_periodically()
|
|
self._ping_parties_periodically()
|
|
|
|
# If any new party infos have come in, apply some of them.
|
|
self._process_pending_party_infos()
|
|
|
|
# Anytime we sign in/out, make sure we refresh our list.
|
|
signed_in = ba.internal.get_v1_account_state() == 'signed_in'
|
|
if self._signed_in != signed_in:
|
|
self._signed_in = signed_in
|
|
self._party_lists_dirty = True
|
|
|
|
# Update sorting to account for ping updates, new parties, etc.
|
|
self._update_party_lists()
|
|
|
|
# If we've got a party-name text widget, keep its value plugged
|
|
# into our public host name.
|
|
text = self._host_name_text
|
|
if text:
|
|
name = cast(str, ba.textwidget(query=self._host_name_text))
|
|
ba.internal.set_public_party_name(name)
|
|
|
|
# Update status text.
|
|
status_text = self._join_status_text
|
|
if status_text:
|
|
if not signed_in:
|
|
ba.textwidget(
|
|
edit=status_text, text=ba.Lstr(resource='notSignedInText')
|
|
)
|
|
else:
|
|
# If we have a valid list, show no status; just the list.
|
|
# Otherwise show either 'loading...' or 'error' depending
|
|
# on whether this is our first go-round.
|
|
if self._have_valid_server_list:
|
|
ba.textwidget(edit=status_text, text='')
|
|
else:
|
|
if self._have_server_list_response:
|
|
ba.textwidget(
|
|
edit=status_text, text=ba.Lstr(resource='errorText')
|
|
)
|
|
else:
|
|
ba.textwidget(
|
|
edit=status_text,
|
|
text=ba.Lstr(
|
|
value='${A}...',
|
|
subs=[
|
|
(
|
|
'${A}',
|
|
ba.Lstr(resource='store.loadingText'),
|
|
)
|
|
],
|
|
),
|
|
)
|
|
|
|
self._update_party_rows()
|
|
|
|
def _update_party_rows(self) -> None:
|
|
columnwidget = self._join_list_column
|
|
if not columnwidget:
|
|
return
|
|
|
|
assert self._join_text
|
|
assert self._filter_text
|
|
|
|
# Janky - allow escaping when there's nothing in our list.
|
|
assert self._host_scrollwidget
|
|
ba.containerwidget(
|
|
edit=self._host_scrollwidget,
|
|
claims_up_down=(len(self._parties_displayed) > 0),
|
|
)
|
|
|
|
# Clip if we have more UI rows than parties to show.
|
|
clipcount = len(self._ui_rows) - len(self._parties_displayed)
|
|
if clipcount > 0:
|
|
clipcount = max(clipcount, 50)
|
|
self._ui_rows = self._ui_rows[:-clipcount]
|
|
|
|
# If we have no parties to show, we're done.
|
|
if not self._parties_displayed:
|
|
return
|
|
|
|
sub_scroll_width = 830
|
|
lineheight = 42
|
|
sub_scroll_height = lineheight * len(self._parties_displayed) + 50
|
|
ba.containerwidget(
|
|
edit=columnwidget, size=(sub_scroll_width, sub_scroll_height)
|
|
)
|
|
|
|
# Any time our height changes, reset the refresh back to the top
|
|
# so we don't see ugly empty spaces appearing during initial list
|
|
# filling.
|
|
if sub_scroll_height != self._last_sub_scroll_height:
|
|
self._refresh_ui_row = 0
|
|
self._last_sub_scroll_height = sub_scroll_height
|
|
|
|
# Also note that we need to redisplay everything since its pos
|
|
# will have changed.. :(
|
|
for party in self._parties.values():
|
|
party.clean_display_index = None
|
|
|
|
# Ew; this rebuilding generates deferred selection callbacks
|
|
# so we need to push deferred notices so we know to ignore them.
|
|
def refresh_on() -> None:
|
|
self._refreshing_list = True
|
|
|
|
ba.pushcall(refresh_on)
|
|
|
|
# Ok, now here's the deal: we want to avoid creating/updating this
|
|
# entire list at one time because it will lead to hitches. So we
|
|
# refresh individual rows quickly in a loop.
|
|
rowcount = min(12, len(self._parties_displayed))
|
|
|
|
party_vals_displayed = list(self._parties_displayed.values())
|
|
while rowcount > 0:
|
|
refresh_row = self._refresh_ui_row % len(self._parties_displayed)
|
|
if refresh_row >= len(self._ui_rows):
|
|
self._ui_rows.append(UIRow())
|
|
refresh_row = len(self._ui_rows) - 1
|
|
|
|
# For the first few seconds after getting our first server-list,
|
|
# refresh only the top section of the list; this allows the lowest
|
|
# ping servers to show up more quickly.
|
|
if self._first_valid_server_list_time is not None:
|
|
if time.time() - self._first_valid_server_list_time < 4.0:
|
|
if refresh_row > 40:
|
|
refresh_row = 0
|
|
|
|
self._ui_rows[refresh_row].update(
|
|
refresh_row,
|
|
party_vals_displayed[refresh_row],
|
|
sub_scroll_width=sub_scroll_width,
|
|
sub_scroll_height=sub_scroll_height,
|
|
lineheight=lineheight,
|
|
columnwidget=columnwidget,
|
|
join_text=self._join_text,
|
|
existing_selection=self._selection,
|
|
filter_text=self._filter_text,
|
|
tab=self,
|
|
)
|
|
self._refresh_ui_row = refresh_row + 1
|
|
rowcount -= 1
|
|
|
|
# So our selection callbacks can start firing..
|
|
def refresh_off() -> None:
|
|
self._refreshing_list = False
|
|
|
|
ba.pushcall(refresh_off)
|
|
|
|
def _process_pending_party_infos(self) -> None:
|
|
starttime = time.time()
|
|
|
|
# We want to do this in small enough pieces to not cause UI hitches.
|
|
chunksize = 30
|
|
parties_in = self._pending_party_infos[:chunksize]
|
|
self._pending_party_infos = self._pending_party_infos[chunksize:]
|
|
for party_in in parties_in:
|
|
addr = party_in['a']
|
|
assert isinstance(addr, str)
|
|
port = party_in['p']
|
|
assert isinstance(port, int)
|
|
party_key = f'{addr}_{port}'
|
|
party = self._parties.get(party_key)
|
|
if party is None:
|
|
# If this party is new to us, init it.
|
|
party = PartyEntry(
|
|
address=addr,
|
|
next_ping_time=ba.time(ba.TimeType.REAL)
|
|
+ 0.001 * party_in['pd'],
|
|
index=self._next_entry_index,
|
|
)
|
|
self._parties[party_key] = party
|
|
self._parties_sorted.append((party_key, party))
|
|
self._party_lists_dirty = True
|
|
self._next_entry_index += 1
|
|
assert isinstance(party.address, str)
|
|
assert isinstance(party.next_ping_time, float)
|
|
|
|
# Now, new or not, update its values.
|
|
party.queue = party_in.get('q')
|
|
assert isinstance(party.queue, (str, type(None)))
|
|
party.port = port
|
|
party.name = party_in['n']
|
|
assert isinstance(party.name, str)
|
|
party.size = party_in['s']
|
|
assert isinstance(party.size, int)
|
|
party.size_max = party_in['sm']
|
|
assert isinstance(party.size_max, int)
|
|
|
|
# Server provides this in milliseconds; we use seconds.
|
|
party.ping_interval = 0.001 * party_in['pi']
|
|
assert isinstance(party.ping_interval, float)
|
|
party.stats_addr = party_in['sa']
|
|
assert isinstance(party.stats_addr, (str, type(None)))
|
|
|
|
# Make sure the party's UI gets updated.
|
|
party.clean_display_index = None
|
|
|
|
if DEBUG_PROCESSING and parties_in:
|
|
print(
|
|
f'Processed {len(parties_in)} raw party infos in'
|
|
f' {time.time()-starttime:.5f}s.'
|
|
)
|
|
|
|
def _update_party_lists(self) -> None:
|
|
if not self._party_lists_dirty:
|
|
return
|
|
starttime = time.time()
|
|
assert len(self._parties_sorted) == len(self._parties)
|
|
|
|
self._parties_sorted.sort(
|
|
key=lambda p: (
|
|
p[1].ping if p[1].ping is not None else 999999.0,
|
|
p[1].index,
|
|
)
|
|
)
|
|
|
|
# If signed out or errored, show no parties.
|
|
if (
|
|
ba.internal.get_v1_account_state() != 'signed_in'
|
|
or not self._have_valid_server_list
|
|
):
|
|
self._parties_displayed = {}
|
|
else:
|
|
if self._filter_value:
|
|
filterval = self._filter_value.lower()
|
|
self._parties_displayed = {
|
|
k: v
|
|
for k, v in self._parties_sorted
|
|
if filterval in v.name.lower()
|
|
}
|
|
else:
|
|
self._parties_displayed = dict(self._parties_sorted)
|
|
|
|
# Any time our selection disappears from the displayed list, go back to
|
|
# auto-selecting the top entry.
|
|
if (
|
|
self._selection is not None
|
|
and self._selection.entry_key not in self._parties_displayed
|
|
):
|
|
self._have_user_selected_row = False
|
|
|
|
# Whenever the user hasn't selected something, keep the first visible
|
|
# row selected.
|
|
if not self._have_user_selected_row and self._parties_displayed:
|
|
firstpartykey = next(iter(self._parties_displayed))
|
|
self._selection = Selection(firstpartykey, SelectionComponent.NAME)
|
|
|
|
self._party_lists_dirty = False
|
|
if DEBUG_PROCESSING:
|
|
print(
|
|
f'Sorted {len(self._parties_sorted)} parties in'
|
|
f' {time.time()-starttime:.5f}s.'
|
|
)
|
|
|
|
def _query_party_list_periodically(self) -> None:
|
|
now = ba.time(ba.TimeType.REAL)
|
|
|
|
# Fire off a new public-party query periodically.
|
|
if (
|
|
self._last_server_list_query_time is None
|
|
or now - self._last_server_list_query_time
|
|
> 0.001
|
|
* ba.internal.get_v1_account_misc_read_val(
|
|
'pubPartyRefreshMS', 10000
|
|
)
|
|
):
|
|
self._last_server_list_query_time = now
|
|
if DEBUG_SERVER_COMMUNICATION:
|
|
print('REQUESTING SERVER LIST')
|
|
if ba.internal.get_v1_account_state() == 'signed_in':
|
|
ba.internal.add_transaction(
|
|
{
|
|
'type': 'PUBLIC_PARTY_QUERY',
|
|
'proto': ba.app.protocol_version,
|
|
'lang': ba.app.lang.language,
|
|
},
|
|
callback=ba.WeakCall(self._on_public_party_query_result),
|
|
)
|
|
ba.internal.run_transactions()
|
|
else:
|
|
self._on_public_party_query_result(None)
|
|
|
|
def _ping_parties_periodically(self) -> None:
|
|
now = ba.time(ba.TimeType.REAL)
|
|
|
|
# Go through our existing public party entries firing off pings
|
|
# for any that have timed out.
|
|
for party in list(self._parties.values()):
|
|
if party.next_ping_time <= now and ba.app.ping_thread_count < 15:
|
|
|
|
# Crank the interval up for high-latency or non-responding
|
|
# parties to save us some useless work.
|
|
mult = 1
|
|
if party.ping_responses == 0:
|
|
if party.ping_attempts > 4:
|
|
mult = 10
|
|
elif party.ping_attempts > 2:
|
|
mult = 5
|
|
if party.ping is not None:
|
|
mult = (
|
|
10 if party.ping > 300 else 5 if party.ping > 150 else 2
|
|
)
|
|
|
|
interval = party.ping_interval * mult
|
|
if DEBUG_SERVER_COMMUNICATION:
|
|
print(
|
|
f'pinging #{party.index} cur={party.ping} '
|
|
f'interval={interval} '
|
|
f'({party.ping_responses}/{party.ping_attempts})'
|
|
)
|
|
|
|
party.next_ping_time = now + party.ping_interval * mult
|
|
party.ping_attempts += 1
|
|
|
|
PingThread(
|
|
party.address, party.port, ba.WeakCall(self._ping_callback)
|
|
).start()
|
|
|
|
def _ping_callback(
|
|
self, address: str, port: int | None, result: float | None
|
|
) -> None:
|
|
# Look for a widget corresponding to this target.
|
|
# If we find one, update our list.
|
|
party_key = f'{address}_{port}'
|
|
party = self._parties.get(party_key)
|
|
if party is not None:
|
|
if result is not None:
|
|
party.ping_responses += 1
|
|
|
|
# We now smooth ping a bit to reduce jumping around in the list
|
|
# (only where pings are relatively good).
|
|
current_ping = party.ping
|
|
if current_ping is not None and result is not None and result < 150:
|
|
smoothing = 0.7
|
|
party.ping = (
|
|
smoothing * current_ping + (1.0 - smoothing) * result
|
|
)
|
|
else:
|
|
party.ping = result
|
|
|
|
# Need to re-sort the list and update the row display.
|
|
party.clean_display_index = None
|
|
self._party_lists_dirty = True
|
|
|
|
def _fetch_local_addr_cb(self, val: str) -> None:
|
|
self._local_address = str(val)
|
|
|
|
def _on_public_party_accessible_response(
|
|
self, data: dict[str, Any] | None
|
|
) -> None:
|
|
|
|
# If we've got status text widgets, update them.
|
|
text = self._host_status_text
|
|
if text:
|
|
if data is None:
|
|
ba.textwidget(
|
|
edit=text,
|
|
text=ba.Lstr(
|
|
resource='gatherWindow.' 'partyStatusNoConnectionText'
|
|
),
|
|
color=(1, 0, 0),
|
|
)
|
|
else:
|
|
if not data.get('accessible', False):
|
|
ex_line: str | ba.Lstr
|
|
if self._local_address is not None:
|
|
ex_line = ba.Lstr(
|
|
value='\n${A} ${B}',
|
|
subs=[
|
|
(
|
|
'${A}',
|
|
ba.Lstr(
|
|
resource='gatherWindow.'
|
|
'manualYourLocalAddressText'
|
|
),
|
|
),
|
|
('${B}', self._local_address),
|
|
],
|
|
)
|
|
else:
|
|
ex_line = ''
|
|
ba.textwidget(
|
|
edit=text,
|
|
text=ba.Lstr(
|
|
value='${A}\n${B}${C}',
|
|
subs=[
|
|
(
|
|
'${A}',
|
|
ba.Lstr(
|
|
resource='gatherWindow.'
|
|
'partyStatusNotJoinableText'
|
|
),
|
|
),
|
|
(
|
|
'${B}',
|
|
ba.Lstr(
|
|
resource='gatherWindow.'
|
|
'manualRouterForwardingText',
|
|
subs=[
|
|
(
|
|
'${PORT}',
|
|
str(
|
|
ba.internal.get_game_port()
|
|
),
|
|
)
|
|
],
|
|
),
|
|
),
|
|
('${C}', ex_line),
|
|
],
|
|
),
|
|
color=(1, 0, 0),
|
|
)
|
|
else:
|
|
ba.textwidget(
|
|
edit=text,
|
|
text=ba.Lstr(
|
|
resource='gatherWindow.' 'partyStatusJoinableText'
|
|
),
|
|
color=(0, 1, 0),
|
|
)
|
|
|
|
def _do_status_check(self) -> None:
|
|
from ba.internal import master_server_get
|
|
|
|
ba.textwidget(
|
|
edit=self._host_status_text,
|
|
color=(1, 1, 0),
|
|
text=ba.Lstr(resource='gatherWindow.' 'partyStatusCheckingText'),
|
|
)
|
|
master_server_get(
|
|
'bsAccessCheck',
|
|
{'b': ba.app.build_number},
|
|
callback=ba.WeakCall(self._on_public_party_accessible_response),
|
|
)
|
|
|
|
def _on_start_advertizing_press(self) -> None:
|
|
from bastd.ui.account import show_sign_in_prompt
|
|
|
|
if ba.internal.get_v1_account_state() != 'signed_in':
|
|
show_sign_in_prompt()
|
|
return
|
|
|
|
name = cast(str, ba.textwidget(query=self._host_name_text))
|
|
if name == '':
|
|
ba.screenmessage(
|
|
ba.Lstr(resource='internal.invalidNameErrorText'),
|
|
color=(1, 0, 0),
|
|
)
|
|
ba.playsound(ba.getsound('error'))
|
|
return
|
|
ba.internal.set_public_party_name(name)
|
|
cfg = ba.app.config
|
|
cfg['Public Party Name'] = name
|
|
cfg.commit()
|
|
ba.playsound(ba.getsound('shieldUp'))
|
|
ba.internal.set_public_party_enabled(True)
|
|
|
|
# In GUI builds we want to authenticate clients only when hosting
|
|
# public parties.
|
|
ba.internal.set_authenticate_clients(True)
|
|
|
|
self._do_status_check()
|
|
ba.buttonwidget(
|
|
edit=self._host_toggle_button,
|
|
label=ba.Lstr(
|
|
resource='gatherWindow.makePartyPrivateText',
|
|
fallback_resource='gatherWindow.stopAdvertisingText',
|
|
),
|
|
on_activate_call=self._on_stop_advertising_press,
|
|
)
|
|
|
|
def _on_stop_advertising_press(self) -> None:
|
|
ba.internal.set_public_party_enabled(False)
|
|
|
|
# In GUI builds we want to authenticate clients only when hosting
|
|
# public parties.
|
|
ba.internal.set_authenticate_clients(False)
|
|
ba.playsound(ba.getsound('shieldDown'))
|
|
text = self._host_status_text
|
|
if text:
|
|
ba.textwidget(
|
|
edit=text,
|
|
text=ba.Lstr(
|
|
resource='gatherWindow.' 'partyStatusNotPublicText'
|
|
),
|
|
color=(0.6, 0.6, 0.6),
|
|
)
|
|
ba.buttonwidget(
|
|
edit=self._host_toggle_button,
|
|
label=ba.Lstr(
|
|
resource='gatherWindow.makePartyPublicText',
|
|
fallback_resource='gatherWindow.startAdvertisingText',
|
|
),
|
|
on_activate_call=self._on_start_advertizing_press,
|
|
)
|
|
|
|
def on_public_party_activate(self, party: PartyEntry) -> None:
|
|
"""Called when a party is clicked or otherwise activated."""
|
|
self.save_state()
|
|
if party.queue is not None:
|
|
from bastd.ui.partyqueue import PartyQueueWindow
|
|
|
|
ba.playsound(ba.getsound('swish'))
|
|
PartyQueueWindow(party.queue, party.address, party.port)
|
|
else:
|
|
address = party.address
|
|
port = party.port
|
|
|
|
# Rate limit this a bit.
|
|
now = time.time()
|
|
last_connect_time = self._last_connect_attempt_time
|
|
if last_connect_time is None or now - last_connect_time > 2.0:
|
|
ba.internal.connect_to_party(address, port=port)
|
|
self._last_connect_attempt_time = now
|
|
|
|
def set_public_party_selection(self, sel: Selection) -> None:
|
|
"""Set the sel."""
|
|
if self._refreshing_list:
|
|
return
|
|
self._selection = sel
|
|
self._have_user_selected_row = True
|
|
|
|
def _on_max_public_party_size_minus_press(self) -> None:
|
|
val = max(1, ba.internal.get_public_party_max_size() - 1)
|
|
ba.internal.set_public_party_max_size(val)
|
|
ba.textwidget(edit=self._host_max_party_size_value, text=str(val))
|
|
|
|
def _on_max_public_party_size_plus_press(self) -> None:
|
|
val = ba.internal.get_public_party_max_size()
|
|
val += 1
|
|
ba.internal.set_public_party_max_size(val)
|
|
ba.textwidget(edit=self._host_max_party_size_value, text=str(val))
|