RYDE-WORK/ballistica
1
0
Fork 0
mirror of https://github.com/RYDE-WORK/ballistica.git synced 2026-02-05 15:03:33 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

more tidying

Browse Source
This commit is contained in:
Eric 2023-06-10 19:20:43 -07:00
parent b645c883c2
commit 2d0364c8e6
No known key found for this signature in database
GPG Key ID: 89C93F0F8D6D5A98
4 changed files with 121 additions and 123 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
src/assets/ba_data/python/baclassic/_subsystem.py
View File

@ -222,7 +222,6 @@ class ClassicSubsystem(AppSubsystem):
# Run a test in a few seconds to see if we should pop up an existing # Run a test in a few seconds to see if we should pop up an existing
# pending special offer. # pending special offer.
def check_special_offer() -> None: def check_special_offer() -> None:
plus = bui.app.plus
assert plus is not None assert plus is not None
from bauiv1lib.specialoffer import show_offer from bauiv1lib.specialoffer import show_offer

98
src/assets/ba_data/python/bascenev1/_activity.py
View File

@ -4,16 +4,10 @@
from __future__ import annotations from __future__ import annotations
import weakref import weakref
import logging
from typing import TYPE_CHECKING, Generic, TypeVar from typing import TYPE_CHECKING, Generic, TypeVar
import _babase import babase
from babase._error import (
print_exception,
SessionTeamNotFoundError,
SessionPlayerNotFoundError,
NodeNotFoundError,
)
from babase._general import Call, verify_object_death
import _bascenev1 import _bascenev1
from bascenev1._dependency import DependencyComponent from bascenev1._dependency import DependencyComponent
from bascenev1._team import Team from bascenev1._team import Team
@ -179,7 +173,7 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
# is dying. # is dying.
self._actor_refs: list[bascenev1.Actor] = [] self._actor_refs: list[bascenev1.Actor] = []
self._actor_weak_refs: list[weakref.ref[bascenev1.Actor]] = [] self._actor_weak_refs: list[weakref.ref[bascenev1.Actor]] = []
self._last_prune_dead_actors_time = _babase.apptime() self._last_prune_dead_actors_time = babase.apptime()
self._prune_dead_actors_timer: bascenev1.Timer | None = None self._prune_dead_actors_timer: bascenev1.Timer | None = None
self.teams = [] self.teams = []
@ -193,15 +187,15 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
# If the activity has been run then we should have already cleaned # If the activity has been run then we should have already cleaned
# it up, but we still need to run expire calls for un-run activities. # it up, but we still need to run expire calls for un-run activities.
if not self._expired: if not self._expired:
with _babase.ContextRef.empty(): with babase.ContextRef.empty():
self._expire() self._expire()
# Inform our owner that we officially kicked the bucket. # Inform our owner that we officially kicked the bucket.
if self._transitioning_out: if self._transitioning_out:
session = self._session() session = self._session()
if session is not None: if session is not None:
_babase.pushcall( babase.pushcall(
Call( babase.Call(
session.transitioning_out_activity_was_freed, session.transitioning_out_activity_was_freed,
self.can_show_ad_on_death, self.can_show_ad_on_death,
) )
@ -219,7 +213,7 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
""" """
node = self._globalsnode node = self._globalsnode
if not node: if not node:
raise NodeNotFoundError() raise babase.NodeNotFoundError()
return node return node
@property @property
@ -290,10 +284,12 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
# and reports any lingering references keeping it alive. # and reports any lingering references keeping it alive.
# We store the timer on the activity so as soon as the activity dies # We store the timer on the activity so as soon as the activity dies
# it gets cleaned up. # it gets cleaned up.
with _babase.ContextRef.empty(): with babase.ContextRef.empty():
ref = weakref.ref(self) ref = weakref.ref(self)
self._activity_death_check_timer = _babase.AppTimer( self._activity_death_check_timer = babase.AppTimer(
5.0, Call(self._check_activity_death, ref, [0]), repeat=True 5.0,
babase.Call(self._check_activity_death, ref, [0]),
repeat=True,
) )
# Run _expire in an empty context; nothing should be happening in # Run _expire in an empty context; nothing should be happening in
@ -302,7 +298,7 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
# and we can't properly provide context in that situation anyway; might # and we can't properly provide context in that situation anyway; might
# as well be consistent). # as well be consistent).
if not self._expired: if not self._expired:
with _babase.ContextRef.empty(): with babase.ContextRef.empty():
self._expire() self._expire()
else: else:
raise RuntimeError( raise RuntimeError(
@ -460,7 +456,7 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
try: try:
self.on_transition_in() self.on_transition_in()
except Exception: except Exception:
print_exception(f'Error in on_transition_in for {self}.') logging.exception('Error in on_transition_in for %s.', self)
# Tell the C++ layer that this activity is the main one, so it uses # Tell the C++ layer that this activity is the main one, so it uses
# settings from our globals, directs various events to us, etc. # settings from our globals, directs various events to us, etc.
@ -474,7 +470,7 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
try: try:
self.on_transition_out() self.on_transition_out()
except Exception: except Exception:
print_exception(f'Error in on_transition_out for {self}.') logging.exception('Error in on_transition_out for %s.', self)
def begin(self, session: bascenev1.Session) -> None: def begin(self, session: bascenev1.Session) -> None:
"""Begin the activity. """Begin the activity.
@ -570,7 +566,7 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
try: try:
self.on_player_join(player) self.on_player_join(player)
except Exception: except Exception:
print_exception(f'Error in on_player_join for {self}.') logging.exception('Error in on_player_join for %s.', self)
def remove_player(self, sessionplayer: bascenev1.SessionPlayer) -> None: def remove_player(self, sessionplayer: bascenev1.SessionPlayer) -> None:
"""Remove a player from the Activity while it is running. """Remove a player from the Activity while it is running.
@ -600,11 +596,11 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
try: try:
self.on_player_leave(player) self.on_player_leave(player)
except Exception: except Exception:
print_exception(f'Error in on_player_leave for {self}.') logging.exception('Error in on_player_leave for %s.', self)
try: try:
player.leave() player.leave()
except Exception: except Exception:
print_exception(f'Error on leave for {player} in {self}.') logging.exception('Error on leave for %s in %s.', player, self)
self._reset_session_player_for_no_activity(sessionplayer) self._reset_session_player_for_no_activity(sessionplayer)
@ -629,7 +625,7 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
try: try:
self.on_team_join(team) self.on_team_join(team)
except Exception: except Exception:
print_exception(f'Error in on_team_join for {self}.') logging.exception('Error in on_team_join for %s.', self)
def remove_team(self, sessionteam: bascenev1.SessionTeam) -> None: def remove_team(self, sessionteam: bascenev1.SessionTeam) -> None:
"""Remove a team from a Running Activity """Remove a team from a Running Activity
@ -651,11 +647,11 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
try: try:
self.on_team_leave(team) self.on_team_leave(team)
except Exception: except Exception:
print_exception(f'Error in on_team_leave for {self}.') logging.exception('Error in on_team_leave for %s.', self)
try: try:
team.leave() team.leave()
except Exception: except Exception:
print_exception(f'Error on leave for {team} in {self}.') logging.exception('Error on leave for %s in %s.', team, self)
sessionteam.activityteam = None sessionteam.activityteam = None
@ -675,16 +671,18 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
try: try:
sessionplayer.setnode(None) sessionplayer.setnode(None)
except Exception: except Exception:
print_exception( logging.exception(
f'Error resetting SessionPlayer node on {sessionplayer}' 'Error resetting SessionPlayer node on %s for %s.',
f' for {self}.' sessionplayer,
self,
) )
try: try:
sessionplayer.resetinput() sessionplayer.resetinput()
except Exception: except Exception:
print_exception( logging.exception(
f'Error resetting SessionPlayer input on {sessionplayer}' 'Error resetting SessionPlayer input on %s for %s.',
f' for {self}.' sessionplayer,
self,
) )
# These should never fail I think... # These should never fail I think...
@ -749,10 +747,10 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
counter[0] += 1 counter[0] += 1
if counter[0] == 4: if counter[0] == 4:
print('Killing app due to stuck activity... :-(') print('Killing app due to stuck activity... :-(')
_babase.quit() babase.quit()
except Exception: except Exception:
print_exception('Error on _check_activity_death/') logging.exception('Error on _check_activity_death.')
def _expire(self) -> None: def _expire(self) -> None:
"""Put the activity in a state where it can be garbage-collected. """Put the activity in a state where it can be garbage-collected.
@ -766,12 +764,12 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
try: try:
self.on_expire() self.on_expire()
except Exception: except Exception:
print_exception(f'Error in Activity on_expire() for {self}.') logging.exception('Error in Activity on_expire() for %s.', self)
try: try:
self._customdata = None self._customdata = None
except Exception: except Exception:
print_exception(f'Error clearing customdata for {self}.') logging.exception('Error clearing customdata for %s.', self)
# Don't want to be holding any delay-delete refs at this point. # Don't want to be holding any delay-delete refs at this point.
self._prune_delay_deletes() self._prune_delay_deletes()
@ -786,19 +784,19 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
try: try:
self._activity_data.expire() self._activity_data.expire()
except Exception: except Exception:
print_exception(f'Error expiring _activity_data for {self}.') logging.exception('Error expiring _activity_data for %s.', self)
def _expire_actors(self) -> None: def _expire_actors(self) -> None:
# Expire all Actors. # Expire all Actors.
for actor_ref in self._actor_weak_refs: for actor_ref in self._actor_weak_refs:
actor = actor_ref() actor = actor_ref()
if actor is not None: if actor is not None:
verify_object_death(actor) babase.verify_object_death(actor)
try: try:
actor.on_expire() actor.on_expire()
except Exception: except Exception:
print_exception( logging.exception(
f'Error in Actor.on_expire()' f' for {actor_ref()}.' 'Error in Actor.on_expire() for %s.', actor_ref()
) )
def _expire_players(self) -> None: def _expire_players(self) -> None:
@ -806,57 +804,57 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
# get freed soon. # get freed soon.
for ex_player in (p() for p in self._players_that_left): for ex_player in (p() for p in self._players_that_left):
if ex_player is not None: if ex_player is not None:
verify_object_death(ex_player) babase.verify_object_death(ex_player)
for player in self.players: for player in self.players:
# This should allow our bascenev1.Player instance to be freed. # This should allow our bascenev1.Player instance to be freed.
# Complain if that doesn't happen. # Complain if that doesn't happen.
verify_object_death(player) babase.verify_object_death(player)
try: try:
player.expire() player.expire()
except Exception: except Exception:
print_exception(f'Error expiring {player}') logging.exception('Error expiring %s.', player)
# Reset the SessionPlayer to a not-in-an-activity state. # Reset the SessionPlayer to a not-in-an-activity state.
try: try:
sessionplayer = player.sessionplayer sessionplayer = player.sessionplayer
self._reset_session_player_for_no_activity(sessionplayer) self._reset_session_player_for_no_activity(sessionplayer)
except SessionPlayerNotFoundError: except babase.SessionPlayerNotFoundError:
# Conceivably, someone could have held on to a Player object # Conceivably, someone could have held on to a Player object
# until now whos underlying SessionPlayer left long ago... # until now whos underlying SessionPlayer left long ago...
pass pass
except Exception: except Exception:
print_exception(f'Error expiring {player}.') logging.exception('Error expiring %s.', player)
def _expire_teams(self) -> None: def _expire_teams(self) -> None:
# Issue warnings for any teams that left the game but don't # Issue warnings for any teams that left the game but don't
# get freed soon. # get freed soon.
for ex_team in (p() for p in self._teams_that_left): for ex_team in (p() for p in self._teams_that_left):
if ex_team is not None: if ex_team is not None:
verify_object_death(ex_team) babase.verify_object_death(ex_team)
for team in self.teams: for team in self.teams:
# This should allow our bascenev1.Team instance to die. # This should allow our bascenev1.Team instance to die.
# Complain if that doesn't happen. # Complain if that doesn't happen.
verify_object_death(team) babase.verify_object_death(team)
try: try:
team.expire() team.expire()
except Exception: except Exception:
print_exception(f'Error expiring {team}') logging.exception('Error expiring %s.', team)
try: try:
sessionteam = team.sessionteam sessionteam = team.sessionteam
sessionteam.activityteam = None sessionteam.activityteam = None
except SessionTeamNotFoundError: except babase.SessionTeamNotFoundError:
# It is expected that Team objects may last longer than # It is expected that Team objects may last longer than
# the SessionTeam they came from (game objects may hold # the SessionTeam they came from (game objects may hold
# team references past the point at which the underlying # team references past the point at which the underlying
# player/team has left the game) # player/team has left the game)
pass pass
except Exception: except Exception:
print_exception(f'Error expiring Team {team}.') logging.exception('Error expiring Team %s.', team)
def _prune_delay_deletes(self) -> None: def _prune_delay_deletes(self) -> None:
self._delay_delete_players.clear() self._delay_delete_players.clear()
@ -871,7 +869,7 @@ class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
] ]
def _prune_dead_actors(self) -> None: def _prune_dead_actors(self) -> None:
self._last_prune_dead_actors_time = _babase.apptime() self._last_prune_dead_actors_time = babase.apptime()
# Prune our strong refs when the Actor's exists() call gives False # Prune our strong refs when the Actor's exists() call gives False
self._actor_refs = [a for a in self._actor_refs if a.exists()] self._actor_refs = [a for a in self._actor_refs if a.exists()]

124
src/assets/ba_data/python/bascenev1/_gameactivity.py
View File

@ -6,12 +6,10 @@
from __future__ import annotations from __future__ import annotations
import random import random
import logging
from typing import TYPE_CHECKING, TypeVar from typing import TYPE_CHECKING, TypeVar
import _babase import babase
from babase._language import Lstr
from babase._error import MapNotFoundError, print_error, print_exception
from babase._general import WeakCall
import _bascenev1 import _bascenev1
from bascenev1._activity import Activity from bascenev1._activity import Activity
from bascenev1._player import PlayerInfo from bascenev1._player import PlayerInfo
@ -25,7 +23,6 @@ if TYPE_CHECKING:
from bascenev1lib.actor.playerspaz import PlayerSpaz from bascenev1lib.actor.playerspaz import PlayerSpaz
from bascenev1lib.actor.bomb import TNTSpawner from bascenev1lib.actor.bomb import TNTSpawner
import babase
import bascenev1 import bascenev1
PlayerT = TypeVar('PlayerT', bound='bascenev1.Player') PlayerT = TypeVar('PlayerT', bound='bascenev1.Player')
@ -89,8 +86,8 @@ class GameActivity(Activity[PlayerT, TeamT]):
bascenev1.GameActivity.get_supported_maps() they can just rely on bascenev1.GameActivity.get_supported_maps() they can just rely on
the default implementation here which calls those methods. the default implementation here which calls those methods.
""" """
assert _babase.app.classic is not None assert babase.app.classic is not None
delegate = _babase.app.classic.delegate delegate = babase.app.classic.delegate
assert delegate is not None assert delegate is not None
delegate.create_default_game_settings_ui( delegate.create_default_game_settings_ui(
cls, sessiontype, settings, completion_call cls, sessiontype, settings, completion_call
@ -115,18 +112,18 @@ class GameActivity(Activity[PlayerT, TeamT]):
Subclasses should override getname(); not this. Subclasses should override getname(); not this.
""" """
name = Lstr(translate=('gameNames', cls.getname())) name = babase.Lstr(translate=('gameNames', cls.getname()))
# A few substitutions for 'Epic', 'Solo' etc. modes. # A few substitutions for 'Epic', 'Solo' etc. modes.
# FIXME: Should provide a way for game types to define filters of # FIXME: Should provide a way for game types to define filters of
# their own and should not rely on hard-coded settings names. # their own and should not rely on hard-coded settings names.
if settings is not None: if settings is not None:
if 'Solo Mode' in settings and settings['Solo Mode']: if 'Solo Mode' in settings and settings['Solo Mode']:
name = Lstr( name = babase.Lstr(
resource='soloNameFilterText', subs=[('${NAME}', name)] resource='soloNameFilterText', subs=[('${NAME}', name)]
) )
if 'Epic Mode' in settings and settings['Epic Mode']: if 'Epic Mode' in settings and settings['Epic Mode']:
name = Lstr( name = babase.Lstr(
resource='epicNameFilterText', subs=[('${NAME}', name)] resource='epicNameFilterText', subs=[('${NAME}', name)]
) )
@ -135,7 +132,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
@classmethod @classmethod
def get_team_display_string(cls, name: str) -> babase.Lstr: def get_team_display_string(cls, name: str) -> babase.Lstr:
"""Given a team name, returns a localized version of it.""" """Given a team name, returns a localized version of it."""
return Lstr(translate=('teamNames', name)) return babase.Lstr(translate=('teamNames', name))
@classmethod @classmethod
def get_description(cls, sessiontype: type[bascenev1.Session]) -> str: def get_description(cls, sessiontype: type[bascenev1.Session]) -> str:
@ -157,7 +154,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
Sub-classes should override get_description(); not this. Sub-classes should override get_description(); not this.
""" """
description = cls.get_description(sessiontype) description = cls.get_description(sessiontype)
return Lstr(translate=('gameDescriptions', description)) return babase.Lstr(translate=('gameDescriptions', description))
@classmethod @classmethod
def get_available_settings( def get_available_settings(
@ -179,8 +176,8 @@ class GameActivity(Activity[PlayerT, TeamT]):
for this game-type for the given bascenev1.Session type. for this game-type for the given bascenev1.Session type.
""" """
del sessiontype # Unused arg. del sessiontype # Unused arg.
assert _babase.app.classic is not None assert babase.app.classic is not None
return _babase.app.classic.getmaps('melee') return babase.app.classic.getmaps('melee')
@classmethod @classmethod
def get_settings_display_string(cls, config: dict[str, Any]) -> babase.Lstr: def get_settings_display_string(cls, config: dict[str, Any]) -> babase.Lstr:
@ -194,7 +191,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
# In newer configs, map is in settings; it used to be in the # In newer configs, map is in settings; it used to be in the
# config root. # config root.
if 'map' in config['settings']: if 'map' in config['settings']:
sval = Lstr( sval = babase.Lstr(
value='${NAME} @ ${MAP}', value='${NAME} @ ${MAP}',
subs=[ subs=[
('${NAME}', name), ('${NAME}', name),
@ -209,7 +206,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
], ],
) )
elif 'map' in config: elif 'map' in config:
sval = Lstr( sval = babase.Lstr(
value='${NAME} @ ${MAP}', value='${NAME} @ ${MAP}',
subs=[ subs=[
('${NAME}', name), ('${NAME}', name),
@ -223,7 +220,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
) )
else: else:
print('invalid game config - expected map entry under settings') print('invalid game config - expected map entry under settings')
sval = Lstr(value='???') sval = babase.Lstr(value='???')
return sval return sval
@classmethod @classmethod
@ -240,7 +237,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
"""Instantiate the Activity.""" """Instantiate the Activity."""
super().__init__(settings) super().__init__(settings)
plus = _babase.app.plus plus = babase.app.plus
# Holds some flattened info about the player set at the point # Holds some flattened info about the player set at the point
# when on_begin() is called. # when on_begin() is called.
@ -297,7 +294,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
exist. exist.
""" """
if self._map is None: if self._map is None:
raise MapNotFoundError raise babase.MapNotFoundError
return self._map return self._map
def get_instance_display_string(self) -> babase.Lstr: def get_instance_display_string(self) -> babase.Lstr:
@ -323,7 +320,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
self.session.campaign_level_name self.session.campaign_level_name
).displayname ).displayname
except Exception: except Exception:
print_error('error getting campaign level name') logging.exception('Error getting campaign level name.')
return self.get_instance_display_string() return self.get_instance_display_string()
def get_instance_description(self) -> str | Sequence: def get_instance_description(self) -> str | Sequence:
@ -401,7 +398,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
""" """
def _continue_choice(self, do_continue: bool) -> None: def _continue_choice(self, do_continue: bool) -> None:
plus = _babase.app.plus plus = babase.app.plus
assert plus is not None assert plus is not None
self._is_waiting_for_continue = False self._is_waiting_for_continue = False
if self.has_ended(): if self.has_ended():
@ -436,11 +433,11 @@ class GameActivity(Activity[PlayerT, TeamT]):
# pylint: disable=cyclic-import # pylint: disable=cyclic-import
from bascenev1._coopsession import CoopSession from bascenev1._coopsession import CoopSession
classic = _babase.app.classic classic = babase.app.classic
assert classic is not None assert classic is not None
continues_window = classic.continues_window continues_window = classic.continues_window
plus = _babase.app.plus plus = babase.app.plus
try: try:
if plus is not None and plus.get_v1_account_misc_read_val( if plus is not None and plus.get_v1_account_misc_read_val(
'enableContinues', False 'enableContinues', False
@ -459,20 +456,20 @@ class GameActivity(Activity[PlayerT, TeamT]):
# Only attempt this if we're not currently paused # Only attempt this if we're not currently paused
# and there appears to be no UI. # and there appears to be no UI.
assert _babase.app.classic is not None assert babase.app.classic is not None
hmmw = _babase.app.ui_v1.has_main_menu_window() hmmw = babase.app.ui_v1.has_main_menu_window()
if not gnode.paused and not hmmw: if not gnode.paused and not hmmw:
self._is_waiting_for_continue = True self._is_waiting_for_continue = True
with _babase.ContextRef.empty(): with babase.ContextRef.empty():
_babase.apptimer( babase.apptimer(
0.5, 0.5,
lambda: continues_window( lambda: continues_window(
self, self,
self._continue_cost, self._continue_cost,
continue_call=WeakCall( continue_call=babase.WeakCall(
self._continue_choice, True self._continue_choice, True
), ),
cancel_call=WeakCall( cancel_call=babase.WeakCall(
self._continue_choice, False self._continue_choice, False
), ),
), ),
@ -480,15 +477,15 @@ class GameActivity(Activity[PlayerT, TeamT]):
return return
except Exception: except Exception:
print_exception('Error handling continues.') logging.exception('Error handling continues.')
self.end_game() self.end_game()
def on_begin(self) -> None: def on_begin(self) -> None:
super().on_begin() super().on_begin()
if _babase.app.classic is not None: if babase.app.classic is not None:
_babase.app.classic.game_begin_analytics() babase.app.classic.game_begin_analytics()
# We don't do this in on_transition_in because it may depend on # We don't do this in on_transition_in because it may depend on
# players/teams which aren't available until now. # players/teams which aren't available until now.
@ -510,13 +507,13 @@ class GameActivity(Activity[PlayerT, TeamT]):
# time is left. # time is left.
tournament_id = self.session.tournament_id tournament_id = self.session.tournament_id
if tournament_id is not None: if tournament_id is not None:
assert _babase.app.plus is not None assert babase.app.plus is not None
_babase.app.plus.tournament_query( babase.app.plus.tournament_query(
args={ args={
'tournamentIDs': [tournament_id], 'tournamentIDs': [tournament_id],
'source': 'in-game time remaining query', 'source': 'in-game time remaining query',
}, },
callback=WeakCall(self._on_tournament_query_response), callback=babase.WeakCall(self._on_tournament_query_response),
) )
def _on_tournament_query_response( def _on_tournament_query_response(
@ -526,8 +523,8 @@ class GameActivity(Activity[PlayerT, TeamT]):
data_t = data['t'] # This used to be the whole payload. data_t = data['t'] # This used to be the whole payload.
# Keep our cached tourney info up to date # Keep our cached tourney info up to date
assert _babase.app.classic is not None assert babase.app.classic is not None
_babase.app.classic.accounts.cache_tournament_info(data_t) babase.app.classic.accounts.cache_tournament_info(data_t)
self._setup_tournament_time_limit( self._setup_tournament_time_limit(
max(5, data_t[0]['timeRemaining']) max(5, data_t[0]['timeRemaining'])
) )
@ -598,11 +595,11 @@ class GameActivity(Activity[PlayerT, TeamT]):
subs = [] subs = []
for i in range(len(sb_desc_l) - 1): for i in range(len(sb_desc_l) - 1):
subs.append(('${ARG' + str(i + 1) + '}', str(sb_desc_l[i + 1]))) subs.append(('${ARG' + str(i + 1) + '}', str(sb_desc_l[i + 1])))
translation = Lstr( translation = babase.Lstr(
translate=('gameDescriptions', sb_desc_l[0]), subs=subs translate=('gameDescriptions', sb_desc_l[0]), subs=subs
) )
sb_desc = translation sb_desc = translation
vrmode = _babase.app.vr_mode vrmode = babase.app.vr_mode
yval = -34 if is_empty else -20 yval = -34 if is_empty else -20
yval -= 16 yval -= 16
sbpos = ( sbpos = (
@ -698,15 +695,17 @@ class GameActivity(Activity[PlayerT, TeamT]):
subs = [] subs = []
for i in range(len(desc_l) - 1): for i in range(len(desc_l) - 1):
subs.append(('${ARG' + str(i + 1) + '}', str(desc_l[i + 1]))) subs.append(('${ARG' + str(i + 1) + '}', str(desc_l[i + 1])))
translation = Lstr(translate=('gameDescriptions', desc_l[0]), subs=subs) translation = babase.Lstr(
translate=('gameDescriptions', desc_l[0]), subs=subs
)
# Do some standard filters (epic mode, etc). # Do some standard filters (epic mode, etc).
if self.settings_raw.get('Epic Mode', False): if self.settings_raw.get('Epic Mode', False):
translation = Lstr( translation = babase.Lstr(
resource='epicDescriptionFilterText', resource='epicDescriptionFilterText',
subs=[('${DESCRIPTION}', translation)], subs=[('${DESCRIPTION}', translation)],
) )
vrmode = _babase.app.vr_mode vrmode = babase.app.vr_mode
dnode = _bascenev1.newnode( dnode = _bascenev1.newnode(
'text', 'text',
attrs={ attrs={
@ -741,8 +740,8 @@ class GameActivity(Activity[PlayerT, TeamT]):
# If there's any tips left on the list, display one. # If there's any tips left on the list, display one.
if self.tips: if self.tips:
tip = self.tips.pop(random.randrange(len(self.tips))) tip = self.tips.pop(random.randrange(len(self.tips)))
tip_title = Lstr( tip_title = babase.Lstr(
value='${A}:', subs=[('${A}', Lstr(resource='tipText'))] value='${A}:', subs=[('${A}', babase.Lstr(resource='tipText'))]
) )
icon: bascenev1.Texture | None = None icon: bascenev1.Texture | None = None
sound: bascenev1.Sound | None = None sound: bascenev1.Sound | None = None
@ -753,14 +752,14 @@ class GameActivity(Activity[PlayerT, TeamT]):
assert isinstance(tip, str) assert isinstance(tip, str)
# Do a few substitutions. # Do a few substitutions.
tip_lstr = Lstr( tip_lstr = babase.Lstr(
translate=('tips', tip), translate=('tips', tip),
subs=[('${PICKUP}', _babase.charstr(SpecialChar.TOP_BUTTON))], subs=[('${PICKUP}', babase.charstr(SpecialChar.TOP_BUTTON))],
) )
base_position = (75, 50) base_position = (75, 50)
tip_scale = 0.8 tip_scale = 0.8
tip_title_scale = 1.2 tip_title_scale = 1.2
vrmode = _babase.app.vr_mode vrmode = babase.app.vr_mode
t_offs = -350.0 t_offs = -350.0
tnode = _bascenev1.newnode( tnode = _bascenev1.newnode(
@ -909,7 +908,8 @@ class GameActivity(Activity[PlayerT, TeamT]):
from bascenev1lib.actor.respawnicon import RespawnIcon from bascenev1lib.actor.respawnicon import RespawnIcon
player.customdata['respawn_timer'] = _bascenev1.Timer( player.customdata['respawn_timer'] = _bascenev1.Timer(
respawn_time, WeakCall(self.spawn_player_if_exists, player) respawn_time,
babase.WeakCall(self.spawn_player_if_exists, player),
) )
player.customdata['respawn_icon'] = RespawnIcon( player.customdata['respawn_icon'] = RespawnIcon(
player, respawn_time player, respawn_time
@ -958,7 +958,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
playerspaztype = PlayerSpaz playerspaztype = PlayerSpaz
light_color = _math.normalized_color(color) light_color = _math.normalized_color(color)
display_color = _babase.safecolor(color, target_intensity=0.75) display_color = babase.safecolor(color, target_intensity=0.75)
spaz = playerspaztype( spaz = playerspaztype(
color=color, color=color,
highlight=highlight, highlight=highlight,
@ -1006,7 +1006,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
self._powerup_drop_timer = _bascenev1.Timer( self._powerup_drop_timer = _bascenev1.Timer(
DEFAULT_POWERUP_INTERVAL, DEFAULT_POWERUP_INTERVAL,
WeakCall(self._standard_drop_powerups), babase.WeakCall(self._standard_drop_powerups),
repeat=True, repeat=True,
) )
self._standard_drop_powerups() self._standard_drop_powerups()
@ -1030,7 +1030,9 @@ class GameActivity(Activity[PlayerT, TeamT]):
# Drop one powerup per point. # Drop one powerup per point.
points = self.map.powerup_spawn_points points = self.map.powerup_spawn_points
for i in range(len(points)): for i in range(len(points)):
_bascenev1.timer(i * 0.4, WeakCall(self._standard_drop_powerup, i)) _bascenev1.timer(
i * 0.4, babase.WeakCall(self._standard_drop_powerup, i)
)
def _setup_standard_tnt_drops(self) -> None: def _setup_standard_tnt_drops(self) -> None:
"""Standard tnt drop.""" """Standard tnt drop."""
@ -1055,7 +1057,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
return return
self._standard_time_limit_time = int(duration) self._standard_time_limit_time = int(duration)
self._standard_time_limit_timer = _bascenev1.Timer( self._standard_time_limit_timer = _bascenev1.Timer(
1.0, WeakCall(self._standard_time_limit_tick), repeat=True 1.0, babase.WeakCall(self._standard_time_limit_tick), repeat=True
) )
self._standard_time_limit_text = NodeActor( self._standard_time_limit_text = NodeActor(
_bascenev1.newnode( _bascenev1.newnode(
@ -1121,7 +1123,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
'color': (1, 0.7, 0, 1), 'color': (1, 0.7, 0, 1),
'position': (0, -90), 'position': (0, -90),
'scale': 1.2, 'scale': 1.2,
'text': Lstr(resource='timeExpiredText'), 'text': babase.Lstr(resource='timeExpiredText'),
}, },
) )
_bascenev1.getsound('refWhistle').play() _bascenev1.getsound('refWhistle').play()
@ -1145,7 +1147,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
# then we have to mess with contexts and whatnot since its currently # then we have to mess with contexts and whatnot since its currently
# not available in activity contexts. :-/ # not available in activity contexts. :-/
self._tournament_time_limit_timer = _bascenev1.BaseTimer( self._tournament_time_limit_timer = _bascenev1.BaseTimer(
1.0, WeakCall(self._tournament_time_limit_tick), repeat=True 1.0, babase.WeakCall(self._tournament_time_limit_tick), repeat=True
) )
self._tournament_time_limit_title_text = NodeActor( self._tournament_time_limit_title_text = NodeActor(
_bascenev1.newnode( _bascenev1.newnode(
@ -1161,7 +1163,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
'position': (60, 50), 'position': (60, 50),
'flatness': 1.0, 'flatness': 1.0,
'scale': 0.5, 'scale': 0.5,
'text': Lstr(resource='tournamentText'), 'text': babase.Lstr(resource='tournamentText'),
}, },
) )
) )
@ -1233,7 +1235,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
if self._tournament_time_limit <= 0: if self._tournament_time_limit <= 0:
self._tournament_time_limit_timer = None self._tournament_time_limit_timer = None
self.end_game() self.end_game()
tval = Lstr( tval = babase.Lstr(
resource='tournamentTimeExpiredText', resource='tournamentTimeExpiredText',
fallback_resource='timeExpiredText', fallback_resource='timeExpiredText',
) )
@ -1275,7 +1277,7 @@ class GameActivity(Activity[PlayerT, TeamT]):
# Reserve a spot on the screen (in case we get multiple of these so # Reserve a spot on the screen (in case we get multiple of these so
# they don't overlap). # they don't overlap).
i = 0 i = 0
cur_time = _babase.apptime() cur_time = babase.apptime()
while True: while True:
if ( if (
i not in self._zoom_message_times i not in self._zoom_message_times
@ -1303,8 +1305,8 @@ class GameActivity(Activity[PlayerT, TeamT]):
# If settings doesn't specify a map, pick a random one from the # If settings doesn't specify a map, pick a random one from the
# list of supported ones. # list of supported ones.
unowned_maps: list[str] = ( unowned_maps: list[str] = (
_babase.app.classic.store.get_unowned_maps() babase.app.classic.store.get_unowned_maps()
if _babase.app.classic is not None if babase.app.classic is not None
else [] else []
) )
valid_maps: list[str] = [ valid_maps: list[str] = [
@ -1313,7 +1315,9 @@ class GameActivity(Activity[PlayerT, TeamT]):
if m not in unowned_maps if m not in unowned_maps
] ]
if not valid_maps: if not valid_maps:
_babase.screenmessage(Lstr(resource='noValidMapsErrorText')) babase.screenmessage(
babase.Lstr(resource='noValidMapsErrorText')
)
raise RuntimeError('No valid maps') raise RuntimeError('No valid maps')
map_name = valid_maps[random.randrange(len(valid_maps))] map_name = valid_maps[random.randrange(len(valid_maps))]
return map_name return map_name

21
src/assets/ba_data/python/bascenev1/_multiteamsession.py
View File

@ -5,17 +5,16 @@ from __future__ import annotations
import copy import copy
import random import random
import logging
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
import _babase import babase
import _bascenev1 import _bascenev1
from babase._error import NotFoundError, print_error
from bascenev1._session import Session from bascenev1._session import Session
if TYPE_CHECKING: if TYPE_CHECKING:
from typing import Any, Sequence from typing import Any, Sequence
import babase
import bascenev1 import bascenev1
DEFAULT_TEAM_COLORS = ((0.1, 0.25, 1.0), (1.0, 0.25, 0.2)) DEFAULT_TEAM_COLORS = ((0.1, 0.25, 1.0), (1.0, 0.25, 0.2))
@ -43,7 +42,7 @@ class MultiTeamSession(Session):
from bascenev1 import _playlist from bascenev1 import _playlist
from bascenev1lib.activity.multiteamjoin import MultiTeamJoinActivity from bascenev1lib.activity.multiteamjoin import MultiTeamJoinActivity
app = _babase.app app = babase.app
classic = app.classic classic = app.classic
assert classic is not None assert classic is not None
cfg = app.config cfg = app.config
@ -163,8 +162,8 @@ class MultiTeamSession(Session):
def get_max_players(self) -> int: def get_max_players(self) -> int:
"""Return max number of Players allowed to join the game at once.""" """Return max number of Players allowed to join the game at once."""
if self.use_teams: if self.use_teams:
return _babase.app.config.get('Team Game Max Players', 8) return babase.app.config.get('Team Game Max Players', 8)
return _babase.app.config.get('Free-for-All Max Players', 8) return babase.app.config.get('Free-for-All Max Players', 8)
def _instantiate_next_game(self) -> None: def _instantiate_next_game(self) -> None:
self._next_game_instance = _bascenev1.newactivity( self._next_game_instance = _bascenev1.newactivity(
@ -229,7 +228,7 @@ class MultiTeamSession(Session):
# (ie: no longer sitting in the lobby). # (ie: no longer sitting in the lobby).
try: try:
has_team = player.sessionteam is not None has_team = player.sessionteam is not None
except NotFoundError: except babase.NotFoundError:
has_team = False has_team = False
if has_team: if has_team:
self.stats.register_sessionplayer(player) self.stats.register_sessionplayer(player)
@ -245,7 +244,7 @@ class MultiTeamSession(Session):
def _switch_to_score_screen(self, results: Any) -> None: def _switch_to_score_screen(self, results: Any) -> None:
"""Switch to a score screen after leaving a round.""" """Switch to a score screen after leaving a round."""
del results # Unused arg. del results # Unused arg.
print_error('this should be overridden') logging.error('This should be overridden.', stack_info=True)
def announce_game_results( def announce_game_results(
self, self,
@ -262,8 +261,6 @@ class MultiTeamSession(Session):
announcement of the same. announcement of the same.
""" """
# pylint: disable=cyclic-import # pylint: disable=cyclic-import
from babase._math import normalized_color
from babase._language import Lstr
from bascenev1._gameutils import cameraflash from bascenev1._gameutils import cameraflash
from bascenev1._freeforallsession import FreeForAllSession from bascenev1._freeforallsession import FreeForAllSession
from bascenev1._messages import CelebrateMessage from bascenev1._messages import CelebrateMessage
@ -286,14 +283,14 @@ class MultiTeamSession(Session):
wins_resource = 'winsPlayerText' wins_resource = 'winsPlayerText'
else: else:
wins_resource = 'winsTeamText' wins_resource = 'winsTeamText'
wins_text = Lstr( wins_text = babase.Lstr(
resource=wins_resource, resource=wins_resource,
subs=[('${NAME}', winning_sessionteam.name)], subs=[('${NAME}', winning_sessionteam.name)],
) )
activity.show_zoom_message( activity.show_zoom_message(
wins_text, wins_text,
scale=0.85, scale=0.85,
color=normalized_color(winning_sessionteam.color), color=babase.normalized_color(winning_sessionteam.color),
) )