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

yup still cleaning

Browse Source
This commit is contained in:
Eric 2023-06-10 19:59:15 -07:00
parent ba287dc5ac
commit 8e127a7c48
No known key found for this signature in database
GPG Key ID: 89C93F0F8D6D5A98
10 changed files with 95 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/assets/ba_data/python/baclassic/_store.py
View File

@ -11,8 +11,8 @@ import _babase
if TYPE_CHECKING:
from typing import Any
import babase
import baclassic
import bascenev1

14
src/assets/ba_data/python/bascenev1/_actor.py
View File

@ -5,10 +5,11 @@
from __future__ import annotations
import weakref
import logging
from typing import TYPE_CHECKING, TypeVar, overload
import babase
from babase._error import print_exception, ActivityNotFoundError
import _bascenev1
from bascenev1._messages import (
DieMessage,
@ -20,7 +21,6 @@ from bascenev1._messages import (
if TYPE_CHECKING:
from typing import Any, Literal
import babase
import bascenev1
ActorT = TypeVar('ActorT', bound='Actor')
@ -93,7 +93,9 @@ class Actor:
if not self.expired:
self.handlemessage(DieMessage())
except Exception:
print_exception('exception in bascenev1.Actor.__del__() for', self)
logging.exception(
'Error in bascenev1.Actor.__del__() for %s.', self
)
def handlemessage(self, msg: Any) -> Any:
"""General message handling; can be passed any message object."""
@ -120,7 +122,7 @@ class Actor:
"""
activity = self._activity()
if activity is None:
raise ActivityNotFoundError()
raise babase.ActivityNotFoundError()
activity.retain_actor(self)
return self
@ -190,7 +192,7 @@ class Actor:
"""
activity = self._activity()
if activity is None:
raise ActivityNotFoundError()
raise babase.ActivityNotFoundError()
return activity
# Overloads to convey our exact return type depending on 'doraise' value.
@ -212,5 +214,5 @@ class Actor:
"""
activity = self._activity()
if activity is None and doraise:
raise ActivityNotFoundError()
raise babase.ActivityNotFoundError()
return activity

31
src/assets/ba_data/python/bascenev1/_coopgame.py
View File

@ -6,15 +6,16 @@ from __future__ import annotations
import logging
from typing import TYPE_CHECKING, TypeVar
import _babase
import babase
import _bascenev1
from babase._general import WeakCall
from bascenev1._gameactivity import GameActivity
if TYPE_CHECKING:
from typing import Sequence
from bascenev1lib.actor.playerspaz import PlayerSpaz
import babase
import bascenev1
PlayerT = TypeVar('PlayerT', bound='bascenev1.Player')
@ -52,11 +53,13 @@ class CoopGameActivity(GameActivity[PlayerT, TeamT]):
super().on_begin()
# Show achievements remaining.
if not (_babase.app.demo_mode or _babase.app.arcade_mode):
_bascenev1.timer(3.8, WeakCall(self._show_remaining_achievements))
if not (babase.app.demo_mode or babase.app.arcade_mode):
_bascenev1.timer(
3.8, babase.WeakCall(self._show_remaining_achievements)
)
# Preload achievement images in case we get some.
_bascenev1.timer(2.0, WeakCall(self._preload_achievements))
_bascenev1.timer(2.0, babase.WeakCall(self._preload_achievements))
# FIXME: this is now redundant with activityutils.getscoreconfig();
# need to kill this.
@ -84,8 +87,8 @@ class CoopGameActivity(GameActivity[PlayerT, TeamT]):
player.actor.handlemessage(CelebrateMessage(duration))
def _preload_achievements(self) -> None:
assert _babase.app.classic is not None
achievements = _babase.app.classic.ach.achievements_for_coop_level(
assert babase.app.classic is not None
achievements = babase.app.classic.ach.achievements_for_coop_level(
self._get_coop_level_name()
)
for ach in achievements:
@ -96,17 +99,17 @@ class CoopGameActivity(GameActivity[PlayerT, TeamT]):
from babase._language import Lstr
from bascenev1lib.actor.text import Text
assert _babase.app.classic is not None
assert babase.app.classic is not None
ts_h_offs = 30
v_offs = -200
achievements = [
a
for a in _babase.app.classic.ach.achievements_for_coop_level(
for a in babase.app.classic.ach.achievements_for_coop_level(
self._get_coop_level_name()
)
if not a.complete
]
vrmode = _babase.app.vr_mode
vrmode = babase.app.vr_mode
if achievements:
Text(
Lstr(resource='achievementsRemainingText'),
@ -158,8 +161,8 @@ class CoopGameActivity(GameActivity[PlayerT, TeamT]):
False otherwise
"""
classic = _babase.app.classic
plus = _babase.app.plus
classic = babase.app.classic
plus = babase.app.plus
if classic is None or plus is None:
logging.warning(
'_award_achievement is a no-op without classic and plus.'
@ -222,7 +225,7 @@ class CoopGameActivity(GameActivity[PlayerT, TeamT]):
"""Set up a beeping noise to play when any players are near death."""
self._life_warning_beep = None
self._life_warning_beep_timer = _bascenev1.Timer(
1.0, WeakCall(self._update_life_warning), repeat=True
1.0, babase.WeakCall(self._update_life_warning), repeat=True
)
def _update_life_warning(self) -> None:

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

@ -5,13 +5,14 @@ from __future__ import annotations
from typing import TYPE_CHECKING
import _babase
import babase
import _bascenev1
from bascenev1._session import Session
if TYPE_CHECKING:
from typing import Any, Callable, Sequence
import babase
import bascenev1
TEAM_COLORS = [(0.2, 0.4, 1.6)]
@ -44,8 +45,8 @@ class CoopSession(Session):
# pylint: disable=cyclic-import
from bascenev1lib.activity.coopjoin import CoopJoinActivity
_babase.increment_analytics_count('Co-op session start')
app = _babase.app
babase.increment_analytics_count('Co-op session start')
app = babase.app
classic = app.classic
assert classic is not None
@ -112,7 +113,7 @@ class CoopSession(Session):
# pylint: disable=cyclic-import
from bascenev1._gameactivity import GameActivity
classic = _babase.app.classic
classic = babase.app.classic
assert classic is not None
# Instantiate levels we may be running soon to let them load in the bg.
@ -212,7 +213,7 @@ class CoopSession(Session):
# Hmm; no players anywhere. Let's end the entire session if we're
# running a GUI (or just the current game if we're running headless).
else:
if not _babase.app.headless_mode:
if not babase.app.headless_mode:
self.end()
else:
if isinstance(activity, GameActivity):
@ -225,12 +226,12 @@ class CoopSession(Session):
# pylint: disable=cyclic-import
from bascenev1._gameactivity import GameActivity
assert _babase.app.classic is not None
assert babase.app.classic is not None
activity = self.getactivity()
if activity is not None and not activity.expired:
assert self.tournament_id is not None
assert isinstance(activity, GameActivity)
_babase.app.classic.tournament_entry_window(
babase.app.classic.tournament_entry_window(
tournament_id=self.tournament_id,
tournament_activity=activity,
on_close_call=resume_callback,
@ -278,7 +279,7 @@ class CoopSession(Session):
from bascenev1lib.activity.coopscore import CoopScoreScreen
from bascenev1lib.tutorial import TutorialActivity
app = _babase.app
app = babase.app
classic = app.classic
assert classic is not None
@ -293,7 +294,7 @@ class CoopSession(Session):
# If we're running with a gui and at any point we have no
# in-game players, quit out of the session (this can happen if
# someone leaves in the tutorial for instance).
if not _babase.app.headless_mode:
if not babase.app.headless_mode:
active_players = [p for p in self.sessionplayers if p.in_game]
if not active_players:
self.end()

13
src/assets/ba_data/python/bascenev1/_debug.py
View File

@ -5,12 +5,11 @@ from __future__ import annotations
from typing import TYPE_CHECKING
import _babase
import babase
if TYPE_CHECKING:
from typing import Any
import babase
import bascenev1
@ -31,7 +30,7 @@ def print_live_object_warnings(
from bascenev1._actor import Actor
from bascenev1._activity import Activity
assert _babase.app.classic is not None
assert babase.app.classic is not None
sessions: list[bascenev1.Session] = []
activities: list[bascenev1.Activity] = []
@ -39,7 +38,7 @@ def print_live_object_warnings(
# Once we come across leaked stuff, printing again is probably
# redundant.
if _babase.app.classic.printed_live_object_warning:
if babase.app.classic.printed_live_object_warning:
return
for obj in gc.get_objects():
if isinstance(obj, Actor):
@ -53,17 +52,17 @@ def print_live_object_warnings(
for session in sessions:
if session is ignore_session:
continue
_babase.app.classic.printed_live_object_warning = True
babase.app.classic.printed_live_object_warning = True
print(f'ERROR: Session found {when}: {session}')
# Complain about any remaining activities.
for activity in activities:
if activity is ignore_activity:
continue
_babase.app.classic.printed_live_object_warning = True
babase.app.classic.printed_live_object_warning = True
print(f'ERROR: Activity found {when}: {activity}')
# Complain about any remaining actors.
for actor in actors:
_babase.app.classic.printed_live_object_warning = True
babase.app.classic.printed_live_object_warning = True
print(f'ERROR: Actor found {when}: {actor}')

6
src/assets/ba_data/python/bascenev1/_dualteamsession.py
View File

@ -5,12 +5,12 @@ from __future__ import annotations
from typing import TYPE_CHECKING
import _babase
import babase
import _bascenev1
from bascenev1._multiteamsession import MultiTeamSession
if TYPE_CHECKING:
import babase
import bascenev1
@ -29,7 +29,7 @@ class DualTeamSession(MultiTeamSession):
_playlists_var = 'Team Tournament Playlists'
def __init__(self) -> None:
_babase.increment_analytics_count('Teams session start')
babase.increment_analytics_count('Teams session start')
super().__init__()
def _switch_to_score_screen(self, results: bascenev1.GameResults) -> None:

6
src/assets/ba_data/python/bascenev1/_freeforallsession.py
View File

@ -6,12 +6,12 @@ from __future__ import annotations
from typing import TYPE_CHECKING
import _babase
import babase
import _bascenev1
from bascenev1._multiteamsession import MultiTeamSession
if TYPE_CHECKING:
import babase
import bascenev1
@ -50,7 +50,7 @@ class FreeForAllSession(MultiTeamSession):
return point_awards
def __init__(self) -> None:
_babase.increment_analytics_count('Free-for-all session start')
babase.increment_analytics_count('Free-for-all session start')
super().__init__()
def _switch_to_score_screen(self, results: bascenev1.GameResults) -> None:

8
src/assets/ba_data/python/bascenev1/_messages.py
View File

@ -8,11 +8,11 @@ from dataclasses import dataclass
from typing import TYPE_CHECKING, TypeVar
from enum import Enum
import _babase
import babase
if TYPE_CHECKING:
from typing import Sequence, Any
import babase
import bascenev1
@ -257,8 +257,8 @@ class HitMessage:
"""Instantiate a message with given values."""
self.srcnode = srcnode
self.pos = pos if pos is not None else _babase.Vec3()
self.velocity = velocity if velocity is not None else _babase.Vec3()
self.pos = pos if pos is not None else babase.Vec3()
self.velocity = velocity if velocity is not None else babase.Vec3()
self.magnitude = magnitude
self.velocity_magnitude = velocity_magnitude
self.radius = radius

2
src/assets/ba_data/python/bascenev1/_powerup.py
View File

@ -9,7 +9,7 @@ from dataclasses import dataclass
if TYPE_CHECKING:
from typing import Sequence
import babase
import bascenev1

80
src/assets/ba_data/python/bascenev1/_session.py
View File

@ -4,18 +4,17 @@
from __future__ import annotations
import weakref
import logging
from typing import TYPE_CHECKING
import _babase
from babase._error import print_error, print_exception, NodeNotFoundError
from babase._language import Lstr
import babase
import _bascenev1
from bascenev1._player import Player
if TYPE_CHECKING:
from typing import Sequence, Any
import babase
import bascenev1
@ -196,7 +195,7 @@ class Session:
with self.context:
self.on_team_join(team)
except Exception:
print_exception(f'Error in on_team_join for {self}.')
logging.exception('Error in on_team_join for %s.', self)
self.lobby = Lobby()
self.stats = Stats()
@ -214,7 +213,7 @@ class Session:
"""The sessionglobals bascenev1.Node for the session."""
node = self._sessionglobalsnode
if not node:
raise NodeNotFoundError()
raise babase.NodeNotFoundError()
return node
def should_allow_mid_activity_joins(
@ -236,15 +235,15 @@ class Session:
"""
# Limit player counts *unless* we're in a stress test.
if (
_babase.app.classic is not None
and _babase.app.classic.stress_test_reset_timer is None
babase.app.classic is not None
and babase.app.classic.stress_test_reset_timer is None
):
if len(self.sessionplayers) >= self.max_players:
# Print a rejection message *only* to the client trying to
# join (prevents spamming everyone else in the game).
_bascenev1.getsound('error').play()
_bascenev1.screenmessage(
Lstr(
babase.Lstr(
resource='playerLimitReachedText',
subs=[('${COUNT}', str(self.max_players))],
),
@ -277,15 +276,15 @@ class Session:
try:
self.lobby.remove_chooser(sessionplayer)
except Exception:
print_exception('Error in Lobby.remove_chooser().')
logging.exception('Error in Lobby.remove_chooser().')
else:
# Ok, they've already entered the game. Remove them from
# teams/activities/etc.
sessionteam = sessionplayer.sessionteam
assert sessionteam is not None
_babase.screenmessage(
Lstr(
babase.screenmessage(
babase.Lstr(
resource='playerLeftText',
subs=[('${PLAYER}', sessionplayer.getname(full=True))],
)
@ -342,17 +341,18 @@ class Session:
self.sessionteams.remove(sessionteam)
self.on_team_leave(sessionteam)
except Exception:
print_exception(
f'Error in on_team_leave for Session {self}.'
logging.exception(
'Error in on_team_leave for Session %s.', self
)
else:
print('Team no in Session teams in on_player_leave.')
try:
sessionteam.leave()
except Exception:
print_exception(
f'Error clearing sessiondata'
f' for team {sessionteam} in session {self}.'
logging.exception(
'Error clearing sessiondata for team %s in session %s.',
sessionteam,
self,
)
def end(self) -> None:
@ -370,17 +370,16 @@ class Session:
from bascenev1._activitytypes import EndSessionActivity
with self.context:
curtime = _babase.apptime()
curtime = babase.apptime()
if self._ending:
# Ignore repeats unless its been a while.
assert self._launch_end_session_activity_time is not None
since_last = curtime - self._launch_end_session_activity_time
if since_last < 30.0:
return
print_error(
'_launch_end_session_activity called twice (since_last='
+ str(since_last)
+ ')'
logging.error(
'_launch_end_session_activity called twice (since_last=%s)',
since_last,
)
self._launch_end_session_activity_time = curtime
self.setactivity(_bascenev1.newactivity(EndSessionActivity))
@ -482,7 +481,7 @@ class Session:
return
if activity is self._activity_retained:
print_error('Activity set to already-current activity.')
logging.error('Activity set to already-current activity.')
return
if self._next_activity is not None:
@ -523,8 +522,8 @@ class Session:
# the activity should have no refs left to it and should die (which
# will trigger the next activity to run).
if prev_activity is not None:
with _babase.ContextRef.empty():
_babase.apptimer(
with babase.ContextRef.empty():
babase.apptimer(
max(0.0, activity.transition_time), prev_activity.expire
)
self._in_set_activity = False
@ -551,9 +550,12 @@ class Session:
with self.context:
self.on_activity_end(activity, results)
except Exception:
print_exception(
f'Error in on_activity_end() for session {self}'
f' activity {activity} with results {results}'
logging.error(
'Error in on_activity_end() for session %s'
' activity %s with results %s',
self,
activity,
results,
)
def _request_player(self, sessionplayer: bascenev1.SessionPlayer) -> bool:
@ -568,7 +570,7 @@ class Session:
with self.context:
result = self.on_player_request(sessionplayer)
except Exception:
print_exception(f'Error in on_player_request for {self}')
logging.exception('Error in on_player_request for %s.', self)
result = False
# If they said yes, add the player to the lobby.
@ -578,7 +580,7 @@ class Session:
try:
self.lobby.add_chooser(sessionplayer)
except Exception:
print_exception('Error in lobby.add_chooser().')
logging.exception('Error in lobby.add_chooser().')
return result
@ -598,7 +600,7 @@ class Session:
"""
if self._next_activity is None:
# Should this ever happen?
print_error('begin_next_activity() called with no _next_activity')
logging.error('begin_next_activity() called with no _next_activity')
return
# We store both a weak and a strong ref to the new activity;
@ -654,8 +656,8 @@ class Session:
# Get our next activity going.
self._complete_end_activity(activity, {})
else:
_babase.screenmessage(
Lstr(
babase.screenmessage(
babase.Lstr(
resource='notEnoughPlayersText',
subs=[('${COUNT}', str(min_players))],
),
@ -681,12 +683,12 @@ class Session:
# hitches.
garbage_collect()
assert _babase.app.classic is not None
assert babase.app.classic is not None
with self.context:
if can_show_ad_on_death:
_babase.app.classic.ads.call_after_ad(self.begin_next_activity)
babase.app.classic.ads.call_after_ad(self.begin_next_activity)
else:
_babase.pushcall(self.begin_next_activity)
babase.pushcall(self.begin_next_activity)
def _add_chosen_player(
self, chooser: bascenev1.Chooser
@ -721,8 +723,8 @@ class Session:
):
pass_to_activity = False
with self.context:
_babase.screenmessage(
Lstr(
babase.screenmessage(
babase.Lstr(
resource='playerDelayedJoinText',
subs=[
('${PLAYER}', sessionplayer.getname(full=True))
@ -751,7 +753,7 @@ class Session:
try:
self.on_team_join(sessionteam)
except Exception:
print_exception(f'Error in on_team_join for {self}.')
logging.exception('Error in on_team_join for %s.', self)
# Add player's team to the Activity.
if pass_to_activity: