Source code for bascenev1._activity

# Released under the MIT License. See LICENSE for details.
#
"""Defines Activity class."""
from __future__ import annotations

import weakref
import logging
from typing import TYPE_CHECKING, Generic, TypeVar

import babase
import _bascenev1
from bascenev1._dependency import DependencyComponent
from bascenev1._team import Team
from bascenev1._messages import UNHANDLED
from bascenev1._player import Player

if TYPE_CHECKING:
    from typing import Any
    import bascenev1

PlayerT = TypeVar('PlayerT', bound=Player)
TeamT = TypeVar('TeamT', bound=Team)


class Activity(DependencyComponent, Generic[PlayerT, TeamT]):
    """Units of execution wrangled by a bascenev1.Session.

    Category: Gameplay Classes

    Examples of Activities include games, score-screens, cutscenes, etc.
    A bascenev1.Session has one 'current' Activity at any time, though
    their existence can overlap during transitions.
    """

    # pylint: disable=too-many-public-methods

    settings_raw: dict[str, Any]
    """The settings dict passed in when the activity was made.
       This attribute is deprecated and should be avoided when possible;
       activities should pull all values they need from the 'settings' arg
       passed to the Activity __init__ call."""

    teams: list[TeamT]
    """The list of bascenev1.Team-s in the Activity. This gets populated just
       before on_begin() is called and is updated automatically as players
       join or leave the game. (at least in free-for-all mode where every
       player gets their own team; in teams mode there are always 2 teams
       regardless of the player count)."""

    players: list[PlayerT]
    """The list of bascenev1.Player-s in the Activity. This gets populated
       just before on_begin() is called and is updated automatically as
       players join or leave the game."""

    announce_player_deaths = False
    """Whether to print every time a player dies. This can be pertinent
       in games such as Death-Match but can be annoying in games where it
       doesn't matter."""

    is_joining_activity = False
    """Joining activities are for waiting for initial player joins.
       They are treated slightly differently than regular activities,
       mainly in that all players are passed to the activity at once
       instead of as each joins."""

    allow_pausing = False
    """Whether game-time should still progress when in menus/etc."""

    allow_kick_idle_players = True
    """Whether idle players can potentially be kicked (should not happen in
       menus/etc)."""

    use_fixed_vr_overlay = False
    """In vr mode, this determines whether overlay nodes (text, images, etc)
       are created at a fixed position in space or one that moves based on
       the current map. Generally this should be on for games and off for
       transitions/score-screens/etc. that persist between maps."""

    slow_motion = False
    """If True, runs in slow motion and turns down sound pitch."""

    inherits_slow_motion = False
    """Set this to True to inherit slow motion setting from previous
       activity (useful for transitions to avoid hitches)."""

    inherits_music = False
    """Set this to True to keep playing the music from the previous activity
       (without even restarting it)."""

    inherits_vr_camera_offset = False
    """Set this to true to inherit VR camera offsets from the previous
       activity (useful for preventing sporadic camera movement
       during transitions)."""

    inherits_vr_overlay_center = False
    """Set this to true to inherit (non-fixed) VR overlay positioning from
       the previous activity (useful for prevent sporadic overlay jostling
       during transitions)."""

    inherits_tint = False
    """Set this to true to inherit screen tint/vignette colors from the
       previous activity (useful to prevent sudden color changes during
       transitions)."""

    allow_mid_activity_joins: bool = True
    """Whether players should be allowed to join in the middle of this
       activity. Note that Sessions may not allow mid-activity-joins even
       if the activity says its ok."""

    transition_time = 0.0
    """If the activity fades or transitions in, it should set the length of
       time here so that previous activities will be kept alive for that
       long (avoiding 'holes' in the screen)
       This value is given in real-time seconds."""

    can_show_ad_on_death = False
    """Is it ok to show an ad after this activity ends before showing
       the next activity?"""

    def __init__(self, settings: dict):
        """Creates an Activity in the current bascenev1.Session.

        The activity will not be actually run until
        bascenev1.Session.setactivity is called. 'settings' should be a
        dict of key/value pairs specific to the activity.

        Activities should preload as much of their media/etc as possible in
        their constructor, but none of it should actually be used until they
        are transitioned in.
        """
        super().__init__()

        # Create our internal engine data.
        self._activity_data = _bascenev1.register_activity(self)

        assert isinstance(settings, dict)
        assert _bascenev1.getactivity() is self

        self._globalsnode: bascenev1.Node | None = None

        # Player/Team types should have been specified as type args;
        # grab those.
        self._playertype: type[PlayerT]
        self._teamtype: type[TeamT]
        self._setup_player_and_team_types()

        # FIXME: Relocate or remove the need for this stuff.
        self.paused_text: bascenev1.Actor | None = None

        self._session = weakref.ref(_bascenev1.getsession())

        # Preloaded data for actors, maps, etc; indexed by type.
        self.preloads: dict[type, Any] = {}

        # Hopefully can eventually kill this; activities should
        # validate/store whatever settings they need at init time
        # (in a more type-safe way).
        self.settings_raw = settings

        self._has_transitioned_in = False
        self._has_begun = False
        self._has_ended = False
        self._activity_death_check_timer: bascenev1.AppTimer | None = None
        self._expired = False
        self._delay_delete_players: list[PlayerT] = []
        self._delay_delete_teams: list[TeamT] = []
        self._players_that_left: list[weakref.ref[PlayerT]] = []
        self._teams_that_left: list[weakref.ref[TeamT]] = []
        self._transitioning_out = False

        # A handy place to put most actors; this list is pruned of dead
        # actors regularly and these actors are insta-killed as the activity
        # is dying.
        self._actor_refs: list[bascenev1.Actor] = []
        self._actor_weak_refs: list[weakref.ref[bascenev1.Actor]] = []
        self._last_prune_dead_actors_time = babase.apptime()
        self._prune_dead_actors_timer: bascenev1.Timer | None = None

        self.teams = []
        self.players = []

        self.lobby = None
        self._stats: bascenev1.Stats | None = None
        self._customdata: dict | None = {}

    def __del__(self) -> None:
        # 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.
        if not self._expired:
            with babase.ContextRef.empty():
                self._expire()

        # Inform our owner that we officially kicked the bucket.
        if self._transitioning_out:
            session = self._session()
            if session is not None:
                babase.pushcall(
                    babase.Call(
                        session.transitioning_out_activity_was_freed,
                        self.can_show_ad_on_death,
                    )
                )

    @property
    def context(self) -> bascenev1.ContextRef:
        """A context-ref pointing at this activity."""
        return self._activity_data.context()

    @property
    def globalsnode(self) -> bascenev1.Node:
        """The 'globals' bascenev1.Node for the activity. This contains various
        global controls and values.
        """
        node = self._globalsnode
        if not node:
            raise babase.NodeNotFoundError()
        return node

    @property
    def stats(self) -> bascenev1.Stats:
        """The stats instance accessible while the activity is running.

        If access is attempted before or after, raises a
        bascenev1.NotFoundError.
        """
        if self._stats is None:
            raise babase.NotFoundError()
        return self._stats

[docs] def on_expire(self) -> None: """Called when your activity is being expired. If your activity has created anything explicitly that may be retaining a strong reference to the activity and preventing it from dying, you should clear that out here. From this point on your activity's sole purpose in life is to hit zero references and die so the next activity can begin. """
@property def customdata(self) -> dict: """Entities needing to store simple data with an activity can put it here. This dict will be deleted when the activity expires, so contained objects generally do not need to worry about handling expired activities. """ assert not self._expired assert isinstance(self._customdata, dict) return self._customdata @property def expired(self) -> bool: """Whether the activity is expired. An activity is set as expired when shutting down. At this point no new nodes, timers, etc should be made, run, etc, and the activity should be considered a 'zombie'. """ return self._expired @property def playertype(self) -> type[PlayerT]: """The type of bascenev1.Player this Activity is using.""" return self._playertype @property def teamtype(self) -> type[TeamT]: """The type of bascenev1.Team this Activity is using.""" return self._teamtype
[docs] def set_has_ended(self, val: bool) -> None: """(internal)""" self._has_ended = val
[docs] def expire(self) -> None: """Begin the process of tearing down the activity. (internal) """ # Create an app-timer that watches a weak-ref of this activity # and reports any lingering references keeping it alive. # We store the timer on the activity so as soon as the activity dies # it gets cleaned up. with babase.ContextRef.empty(): ref = weakref.ref(self) self._activity_death_check_timer = babase.AppTimer( 5.0, babase.Call(self._check_activity_death, ref, [0]), repeat=True, ) # Run _expire in an empty context; nothing should be happening in # there except deleting things which requires no context. # (plus, _expire() runs in the destructor for un-run activities # and we can't properly provide context in that situation anyway; might # as well be consistent). if not self._expired: with babase.ContextRef.empty(): self._expire() else: raise RuntimeError( f'destroy() called when already expired for {self}.' )
[docs] def retain_actor(self, actor: bascenev1.Actor) -> None: """Add a strong-reference to a bascenev1.Actor to this Activity. The reference will be lazily released once bascenev1.Actor.exists() returns False for the Actor. The bascenev1.Actor.autoretain() method is a convenient way to access this same functionality. """ if __debug__: from bascenev1._actor import Actor assert isinstance(actor, Actor) self._actor_refs.append(actor)
[docs] def add_actor_weak_ref(self, actor: bascenev1.Actor) -> None: """Add a weak-reference to a bascenev1.Actor to the bascenev1.Activity. (called by the bascenev1.Actor base class) """ if __debug__: from bascenev1._actor import Actor assert isinstance(actor, Actor) self._actor_weak_refs.append(weakref.ref(actor))
@property def session(self) -> bascenev1.Session: """The bascenev1.Session this bascenev1.Activity belongs to. Raises a babase.SessionNotFoundError if the Session no longer exists. """ session = self._session() if session is None: raise babase.SessionNotFoundError() return session
[docs] def on_player_join(self, player: PlayerT) -> None: """Called when a new bascenev1.Player has joined the Activity. (including the initial set of Players) """
[docs] def on_player_leave(self, player: PlayerT) -> None: """Called when a bascenev1.Player is leaving the Activity."""
[docs] def on_team_join(self, team: TeamT) -> None: """Called when a new bascenev1.Team joins the Activity. (including the initial set of Teams) """
[docs] def on_team_leave(self, team: TeamT) -> None: """Called when a bascenev1.Team leaves the Activity."""
[docs] def on_transition_in(self) -> None: """Called when the Activity is first becoming visible. Upon this call, the Activity should fade in backgrounds, start playing music, etc. It does not yet have access to players or teams, however. They remain owned by the previous Activity up until bascenev1.Activity.on_begin() is called. """
[docs] def on_transition_out(self) -> None: """Called when your activity begins transitioning out. Note that this may happen at any time even if bascenev1.Activity.end() has not been called. """
[docs] def on_begin(self) -> None: """Called once the previous Activity has finished transitioning out. At this point the activity's initial players and teams are filled in and it should begin its actual game logic. """
[docs] def handlemessage(self, msg: Any) -> Any: """General message handling; can be passed any message object.""" del msg # Unused arg. return UNHANDLED
[docs] def has_transitioned_in(self) -> bool: """Return whether bascenev1.Activity.on_transition_in() has run.""" return self._has_transitioned_in
[docs] def has_begun(self) -> bool: """Return whether bascenev1.Activity.on_begin() has run.""" return self._has_begun
[docs] def has_ended(self) -> bool: """Return whether the activity has commenced ending.""" return self._has_ended
[docs] def is_transitioning_out(self) -> bool: """Return whether bascenev1.Activity.on_transition_out() has run.""" return self._transitioning_out
[docs] def transition_in(self, prev_globals: bascenev1.Node | None) -> None: """Called by Session to kick off transition-in. (internal) """ assert not self._has_transitioned_in self._has_transitioned_in = True # Set up the globals node based on our settings. with self.context: glb = self._globalsnode = _bascenev1.newnode('globals') # Now that it's going to be front and center, # set some global values based on what the activity wants. glb.use_fixed_vr_overlay = self.use_fixed_vr_overlay glb.allow_kick_idle_players = self.allow_kick_idle_players if self.inherits_slow_motion and prev_globals is not None: glb.slow_motion = prev_globals.slow_motion else: glb.slow_motion = self.slow_motion if self.inherits_music and prev_globals is not None: glb.music_continuous = True # Prevent restarting same music. glb.music = prev_globals.music glb.music_count += 1 if self.inherits_vr_camera_offset and prev_globals is not None: glb.vr_camera_offset = prev_globals.vr_camera_offset if self.inherits_vr_overlay_center and prev_globals is not None: glb.vr_overlay_center = prev_globals.vr_overlay_center glb.vr_overlay_center_enabled = ( prev_globals.vr_overlay_center_enabled ) # If they want to inherit tint from the previous self. if self.inherits_tint and prev_globals is not None: glb.tint = prev_globals.tint glb.vignette_outer = prev_globals.vignette_outer glb.vignette_inner = prev_globals.vignette_inner # Start pruning our various things periodically. self._prune_dead_actors() self._prune_dead_actors_timer = _bascenev1.Timer( 5.17, self._prune_dead_actors, repeat=True ) _bascenev1.timer(13.3, self._prune_delay_deletes, repeat=True) # Also start our low-level scene running. self._activity_data.start() try: self.on_transition_in() except Exception: logging.exception('Error in on_transition_in for %s.', self) # Tell the C++ layer that this activity is the main one, so it uses # settings from our globals, directs various events to us, etc. self._activity_data.make_foreground()
[docs] def transition_out(self) -> None: """Called by the Session to start us transitioning out.""" assert not self._transitioning_out self._transitioning_out = True with self.context: try: self.on_transition_out() except Exception: logging.exception('Error in on_transition_out for %s.', self)
[docs] def begin(self, session: bascenev1.Session) -> None: """Begin the activity. (internal) """ assert not self._has_begun # Inherit stats from the session. self._stats = session.stats # Add session's teams in. for team in session.sessionteams: self.add_team(team) # Add session's players in. for player in session.sessionplayers: self.add_player(player) self._has_begun = True # Let the activity do its thing. with self.context: # Note: do we want to catch errors here? # Currently I believe we wind up canceling the # activity launch; just wanna be sure that is intentional. self.on_begin()
[docs] def end( self, results: Any = None, delay: float = 0.0, force: bool = False ) -> None: """Commences Activity shutdown and delivers results to the Session. 'delay' is the time delay before the Activity actually ends (in seconds). Further calls to end() will be ignored up until this time, unless 'force' is True, in which case the new results will replace the old. """ # Ask the session to end us. self.session.end_activity(self, results, delay, force)
[docs] def create_player(self, sessionplayer: bascenev1.SessionPlayer) -> PlayerT: """Create the Player instance for this Activity. Subclasses can override this if the activity's player class requires a custom constructor; otherwise it will be called with no args. Note that the player object should not be used at this point as it is not yet fully wired up; wait for bascenev1.Activity.on_player_join() for that. """ del sessionplayer # Unused. player = self._playertype() return player
[docs] def create_team(self, sessionteam: bascenev1.SessionTeam) -> TeamT: """Create the Team instance for this Activity. Subclasses can override this if the activity's team class requires a custom constructor; otherwise it will be called with no args. Note that the team object should not be used at this point as it is not yet fully wired up; wait for on_team_join() for that. """ del sessionteam # Unused. team = self._teamtype() return team
[docs] def add_player(self, sessionplayer: bascenev1.SessionPlayer) -> None: """(internal)""" assert sessionplayer.sessionteam is not None sessionplayer.resetinput() sessionteam = sessionplayer.sessionteam assert sessionplayer in sessionteam.players team = sessionteam.activityteam assert team is not None sessionplayer.setactivity(self) with self.context: sessionplayer.activityplayer = player = self.create_player( sessionplayer ) player.postinit(sessionplayer) assert player not in team.players team.players.append(player) assert player in team.players assert player not in self.players self.players.append(player) assert player in self.players try: self.on_player_join(player) except Exception: logging.exception('Error in on_player_join for %s.', self)
[docs] def remove_player(self, sessionplayer: bascenev1.SessionPlayer) -> None: """Remove a player from the Activity while it is running. (internal) """ assert not self.expired player: Any = sessionplayer.activityplayer assert isinstance(player, self._playertype) team: Any = sessionplayer.sessionteam.activityteam assert isinstance(team, self._teamtype) assert player in team.players team.players.remove(player) assert player not in team.players assert player in self.players self.players.remove(player) assert player not in self.players # This should allow our bascenev1.Player instance to die. # Complain if that doesn't happen. # verify_object_death(player) with self.context: try: self.on_player_leave(player) except Exception: logging.exception('Error in on_player_leave for %s.', self) try: player.leave() except Exception: logging.exception('Error on leave for %s in %s.', player, self) self._reset_session_player_for_no_activity(sessionplayer) # Add the player to a list to keep it around for a while. This is # to discourage logic from firing on player object death, which # may not happen until activity end if something is holding refs # to it. self._delay_delete_players.append(player) self._players_that_left.append(weakref.ref(player))
[docs] def add_team(self, sessionteam: bascenev1.SessionTeam) -> None: """Add a team to the Activity (internal) """ assert not self.expired with self.context: sessionteam.activityteam = team = self.create_team(sessionteam) team.postinit(sessionteam) self.teams.append(team) try: self.on_team_join(team) except Exception: logging.exception('Error in on_team_join for %s.', self)
[docs] def remove_team(self, sessionteam: bascenev1.SessionTeam) -> None: """Remove a team from a Running Activity (internal) """ assert not self.expired assert sessionteam.activityteam is not None team: Any = sessionteam.activityteam assert isinstance(team, self._teamtype) assert team in self.teams self.teams.remove(team) assert team not in self.teams with self.context: # Make a decent attempt to persevere if user code breaks. try: self.on_team_leave(team) except Exception: logging.exception('Error in on_team_leave for %s.', self) try: team.leave() except Exception: logging.exception('Error on leave for %s in %s.', team, self) sessionteam.activityteam = None # Add the team to a list to keep it around for a while. This is # to discourage logic from firing on team object death, which # may not happen until activity end if something is holding refs # to it. self._delay_delete_teams.append(team) self._teams_that_left.append(weakref.ref(team))
def _reset_session_player_for_no_activity( self, sessionplayer: bascenev1.SessionPlayer ) -> None: # Let's be extra-defensive here: killing a node/input-call/etc # could trigger user-code resulting in errors, but we would still # like to complete the reset if possible. try: sessionplayer.setnode(None) except Exception: logging.exception( 'Error resetting SessionPlayer node on %s for %s.', sessionplayer, self, ) try: sessionplayer.resetinput() except Exception: logging.exception( 'Error resetting SessionPlayer input on %s for %s.', sessionplayer, self, ) # These should never fail I think... sessionplayer.setactivity(None) sessionplayer.activityplayer = None # noinspection PyUnresolvedReferences def _setup_player_and_team_types(self) -> None: """Pull player and team types from our typing.Generic params.""" # TODO: There are proper calls for pulling these in Python 3.8; # should update this code when we adopt that. # NOTE: If we get Any as PlayerT or TeamT (generally due # to no generic params being passed) we automatically use the # base class types, but also warn the user since this will mean # less type safety for that class. (its better to pass the base # player/team types explicitly vs. having them be Any) if not TYPE_CHECKING: self._playertype = type(self).__orig_bases__[-1].__args__[0] if not isinstance(self._playertype, type): self._playertype = Player print( f'ERROR: {type(self)} was not passed a Player' f' type argument; please explicitly pass bascenev1.Player' f' if you do not want to override it.' ) self._teamtype = type(self).__orig_bases__[-1].__args__[1] if not isinstance(self._teamtype, type): self._teamtype = Team print( f'ERROR: {type(self)} was not passed a Team' f' type argument; please explicitly pass bascenev1.Team' f' if you do not want to override it.' ) assert issubclass(self._playertype, Player) assert issubclass(self._teamtype, Team) @classmethod def _check_activity_death( cls, activity_ref: weakref.ref[Activity], counter: list[int] ) -> None: """Sanity check to make sure an Activity was destroyed properly. Receives a weakref to a bascenev1.Activity which should have torn itself down due to no longer being referenced anywhere. Will complain and/or print debugging info if the Activity still exists. """ try: activity = activity_ref() print( 'ERROR: Activity is not dying when expected:', activity, '(warning ' + str(counter[0] + 1) + ')', ) print( 'This means something is still strong-referencing it.\n' 'Check out methods such as efro.debug.printrefs() to' ' help debug this sort of thing.' ) # Note: no longer calling gc.get_referrers() here because it's # usage can bork stuff. (see notes at top of efro.debug) counter[0] += 1 if counter[0] == 4: print('Killing app due to stuck activity... :-(') babase.quit() except Exception: logging.exception('Error on _check_activity_death.') def _expire(self) -> None: """Put the activity in a state where it can be garbage-collected. This involves clearing anything that might be holding a reference to it, etc. """ assert not self._expired self._expired = True try: self.on_expire() except Exception: logging.exception('Error in Activity on_expire() for %s.', self) try: self._customdata = None except Exception: logging.exception('Error clearing customdata for %s.', self) # Don't want to be holding any delay-delete refs at this point. self._prune_delay_deletes() self._expire_actors() self._expire_players() self._expire_teams() # This will kill all low level stuff: Timers, Nodes, etc., which # should clear up any remaining refs to our Activity and allow us # to die peacefully. try: self._activity_data.expire() except Exception: logging.exception('Error expiring _activity_data for %s.', self) def _expire_actors(self) -> None: # Expire all Actors. for actor_ref in self._actor_weak_refs: actor = actor_ref() if actor is not None: babase.verify_object_death(actor) try: actor.on_expire() except Exception: logging.exception( 'Error in Actor.on_expire() for %s.', actor_ref() ) def _expire_players(self) -> None: # Issue warnings for any players that left the game but don't # get freed soon. for ex_player in (p() for p in self._players_that_left): if ex_player is not None: babase.verify_object_death(ex_player) for player in self.players: # This should allow our bascenev1.Player instance to be freed. # Complain if that doesn't happen. babase.verify_object_death(player) try: player.expire() except Exception: logging.exception('Error expiring %s.', player) # Reset the SessionPlayer to a not-in-an-activity state. try: sessionplayer = player.sessionplayer self._reset_session_player_for_no_activity(sessionplayer) except babase.SessionPlayerNotFoundError: # Conceivably, someone could have held on to a Player object # until now whos underlying SessionPlayer left long ago... pass except Exception: logging.exception('Error expiring %s.', player) def _expire_teams(self) -> None: # Issue warnings for any teams that left the game but don't # get freed soon. for ex_team in (p() for p in self._teams_that_left): if ex_team is not None: babase.verify_object_death(ex_team) for team in self.teams: # This should allow our bascenev1.Team instance to die. # Complain if that doesn't happen. babase.verify_object_death(team) try: team.expire() except Exception: logging.exception('Error expiring %s.', team) try: sessionteam = team.sessionteam sessionteam.activityteam = None except babase.SessionTeamNotFoundError: # It is expected that Team objects may last longer than # the SessionTeam they came from (game objects may hold # team references past the point at which the underlying # player/team has left the game) pass except Exception: logging.exception('Error expiring Team %s.', team) def _prune_delay_deletes(self) -> None: self._delay_delete_players.clear() self._delay_delete_teams.clear() # Clear out any dead weak-refs. self._teams_that_left = [ t for t in self._teams_that_left if t() is not None ] self._players_that_left = [ p for p in self._players_that_left if p() is not None ] def _prune_dead_actors(self) -> None: self._last_prune_dead_actors_time = babase.apptime() # 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()] # Prune our weak refs once the Actor object has been freed. self._actor_weak_refs = [ a for a in self._actor_weak_refs if a() is not None ]