Source code for bascenev1._coopsession

# Released under the MIT License. See LICENSE for details.
#
"""Functionality related to coop-mode sessions."""
from __future__ import annotations

from typing import TYPE_CHECKING, override

import babase

import _bascenev1
from bascenev1._session import Session

if TYPE_CHECKING:
    from typing import Any, Callable, Sequence

    import bascenev1

TEAM_COLORS = [(0.2, 0.4, 1.6)]
TEAM_NAMES = ['Good Guys']


[docs] class CoopSession(Session): """A session which runs cooperative-mode games. These generally consist of 1-4 players against the computer and include functionality such as high score lists. """ use_teams = True use_team_colors = False allow_mid_activity_joins = False # Note: even though these are instance vars, we annotate them at the # class level so that docs generation can access their types. campaign: bascenev1.Campaign | None """The baclassic.Campaign instance this Session represents, or None if there is no associated Campaign.""" def __init__(self) -> None: """Instantiate a co-op mode session.""" # pylint: disable=cyclic-import from bascenev1lib.activity.coopjoin import CoopJoinActivity babase.increment_analytics_count('Co-op session start') app = babase.app classic = app.classic assert classic is not None # If they passed in explicit min/max, honor that. # Otherwise defer to user overrides or defaults. if 'min_players' in classic.coop_session_args: min_players = classic.coop_session_args['min_players'] else: min_players = 1 if 'max_players' in classic.coop_session_args: max_players = classic.coop_session_args['max_players'] else: max_players = app.config.get('Coop Game Max Players', 4) if 'submit_score' in classic.coop_session_args: submit_score = classic.coop_session_args['submit_score'] else: submit_score = True # print('FIXME: COOP SESSION WOULD CALC DEPS.') depsets: Sequence[bascenev1.DependencySet] = [] super().__init__( depsets, team_names=TEAM_NAMES, team_colors=TEAM_COLORS, min_players=min_players, max_players=max_players, submit_score=submit_score, ) # Tournament-ID if we correspond to a co-op tournament (otherwise None) self.tournament_id: str | None = classic.coop_session_args.get( 'tournament_id' ) self.campaign = classic.getcampaign( classic.coop_session_args['campaign'] ) self.campaign_level_name: str = classic.coop_session_args['level'] self._ran_tutorial_activity = False self._tutorial_activity: bascenev1.Activity | None = None self._custom_menu_ui: list[dict[str, Any]] = [] # Start our joining screen. self.setactivity(_bascenev1.newactivity(CoopJoinActivity)) self._next_game_instance: bascenev1.GameActivity | None = None self._next_game_level_name: str | None = None self._update_on_deck_game_instances()
[docs] def get_current_game_instance(self) -> bascenev1.GameActivity: """Get the game instance currently being played.""" return self._current_game_instance
[docs] @override def should_allow_mid_activity_joins( self, activity: bascenev1.Activity ) -> bool: # pylint: disable=cyclic-import from bascenev1._gameactivity import GameActivity # Disallow any joins in the middle of the game. if isinstance(activity, GameActivity): return False return True
def _update_on_deck_game_instances(self) -> None: # pylint: disable=cyclic-import from bascenev1._gameactivity import GameActivity classic = babase.app.classic assert classic is not None # Instantiate levels we may be running soon to let them load in the bg. # Build an instance for the current level. assert self.campaign is not None level = self.campaign.getlevel(self.campaign_level_name) gametype = level.gametype settings = level.get_settings() # Make sure all settings the game expects are present. neededsettings = gametype.get_available_settings(type(self)) for setting in neededsettings: if setting.name not in settings: settings[setting.name] = setting.default newactivity = _bascenev1.newactivity(gametype, settings) assert isinstance(newactivity, GameActivity) self._current_game_instance: GameActivity = newactivity # Find the next level and build an instance for it too. levels = self.campaign.levels level = self.campaign.getlevel(self.campaign_level_name) nextlevel: bascenev1.Level | None if level.index < len(levels) - 1: nextlevel = levels[level.index + 1] else: nextlevel = None if nextlevel: gametype = nextlevel.gametype settings = nextlevel.get_settings() # Make sure all settings the game expects are present. neededsettings = gametype.get_available_settings(type(self)) for setting in neededsettings: if setting.name not in settings: settings[setting.name] = setting.default # We wanna be in the activity's context while taking it down. newactivity = _bascenev1.newactivity(gametype, settings) assert isinstance(newactivity, GameActivity) self._next_game_instance = newactivity self._next_game_level_name = nextlevel.name else: self._next_game_instance = None self._next_game_level_name = None # Special case: # If our current level is 'onslaught training', instantiate # our tutorial so its ready to go. (if we haven't run it yet). if ( self.campaign_level_name == 'Onslaught Training' and self._tutorial_activity is None and not self._ran_tutorial_activity ): from bascenev1lib.tutorial import TutorialActivity self._tutorial_activity = _bascenev1.newactivity(TutorialActivity)
[docs] @override def get_custom_menu_entries(self) -> list[dict[str, Any]]: return self._custom_menu_ui
[docs] @override def on_player_leave(self, sessionplayer: bascenev1.SessionPlayer) -> None: super().on_player_leave(sessionplayer) _bascenev1.timer(2.0, babase.WeakCall(self._handle_empty_activity))
def _handle_empty_activity(self) -> None: """Handle cases where all players have left the current activity.""" from bascenev1._gameactivity import GameActivity activity = self.getactivity() if activity is None: return # Hmm what should we do in this case? # If there are still players in the current activity, we're good. if activity.players: return # If there are *not* players in the current activity but there # *are* in the session: if not activity.players and self.sessionplayers: # If we're in a game, we should restart to pull in players # currently waiting in the session. if isinstance(activity, GameActivity): # Never restart tourney games however; just end the session # if all players are gone. if self.tournament_id is not None: self.end() else: self.restart() # 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 babase.app.env.gui: self.end() else: if isinstance(activity, GameActivity): with activity.context: activity.end_game() def _on_tournament_restart_menu_press( self, resume_callback: Callable[[], Any] ) -> None: # pylint: disable=cyclic-import from bascenev1._gameactivity import GameActivity 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( tournament_id=self.tournament_id, tournament_activity=activity, on_close_call=resume_callback, )
[docs] def restart(self) -> None: """Restart the current game activity.""" # Tell the current activity to end with a 'restart' outcome. # We use 'force' so that we apply even if end has already been called # (but is in its delay period). # Make an exception if there's no players left. Otherwise this # can override the default session end that occurs in that case. if not self.sessionplayers: return # This method may get called from the UI context so make sure we # explicitly run in the activity's context. activity = self.getactivity() if activity is not None and not activity.expired: activity.can_show_ad_on_death = True with activity.context: activity.end(results={'outcome': 'restart'}, force=True)
# noinspection PyUnresolvedReferences
[docs] @override def on_activity_end( self, activity: bascenev1.Activity, results: Any ) -> None: """Method override for co-op sessions. Jumps between co-op games and score screens. """ # pylint: disable=too-many-branches # pylint: disable=too-many-locals # pylint: disable=too-many-statements # pylint: disable=cyclic-import from bascenev1lib.activity.coopscore import CoopScoreScreen from bascenev1lib.tutorial import TutorialActivity from bascenev1._gameresults import GameResults from bascenev1._player import PlayerInfo from bascenev1._activitytypes import JoinActivity, TransitionActivity from bascenev1._coopgame import CoopGameActivity from bascenev1._score import ScoreType app = babase.app env = app.env classic = app.classic assert classic is not None # If we're running a TeamGameActivity we'll have a GameResults # as results. Otherwise its an old CoopGameActivity so its giving # us a dict of random stuff. if isinstance(results, GameResults): outcome = 'defeat' # This can't be 'beaten'. else: outcome = '' if results is None else results.get('outcome', '') # 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 env.gui: active_players = [p for p in self.sessionplayers if p.in_game] if not active_players: self.end() return # If we're in a between-round activity or a restart-activity, # hop into a round. if isinstance( activity, (JoinActivity, CoopScoreScreen, TransitionActivity) ): if outcome == 'next_level': if self._next_game_instance is None: raise RuntimeError() assert self._next_game_level_name is not None self.campaign_level_name = self._next_game_level_name next_game = self._next_game_instance else: next_game = self._current_game_instance # Special case: if we're coming from a joining-activity # and will be going into onslaught-training, show the # tutorial first. if ( isinstance(activity, JoinActivity) and self.campaign_level_name == 'Onslaught Training' and not (env.demo or env.arcade) ): if self._tutorial_activity is None: raise RuntimeError('Tutorial not preloaded properly.') self.setactivity(self._tutorial_activity) self._tutorial_activity = None self._ran_tutorial_activity = True self._custom_menu_ui = [] # Normal case; launch the next round. else: # Reset stats for the new activity. self.stats.reset() for player in self.sessionplayers: # Skip players that are still choosing a team. if player.in_game: self.stats.register_sessionplayer(player) self.stats.setactivity(next_game) # Now flip the current activity.. self.setactivity(next_game) if not (env.demo or env.arcade): if ( self.tournament_id is not None and classic.coop_session_args['submit_score'] ): self._custom_menu_ui = [ { 'label': babase.Lstr(resource='restartText'), 'resume_on_call': False, 'call': babase.WeakCall( self._on_tournament_restart_menu_press ), } ] else: self._custom_menu_ui = [ { 'label': babase.Lstr(resource='restartText'), 'call': babase.WeakCall(self.restart), } ] # If we were in a tutorial, just pop a transition to get to the # actual round. elif isinstance(activity, TutorialActivity): self.setactivity(_bascenev1.newactivity(TransitionActivity)) else: playerinfos: list[bascenev1.PlayerInfo] # Generic team games. if isinstance(results, GameResults): playerinfos = results.playerinfos score = results.get_sessionteam_score(results.sessionteams[0]) fail_message = None score_order = ( 'decreasing' if results.lower_is_better else 'increasing' ) if results.scoretype in ( ScoreType.SECONDS, ScoreType.MILLISECONDS, ): scoretype = 'time' # ScoreScreen wants hundredths of a second. if score is not None: if results.scoretype is ScoreType.SECONDS: score *= 100 elif results.scoretype is ScoreType.MILLISECONDS: score //= 10 else: raise RuntimeError('FIXME') else: if results.scoretype is not ScoreType.POINTS: print(f'Unknown ScoreType:' f' "{results.scoretype}"') scoretype = 'points' # Old coop-game-specific results; should migrate away from these. else: playerinfos = results.get('playerinfos') score = results['score'] if 'score' in results else None fail_message = ( results['fail_message'] if 'fail_message' in results else None ) score_order = ( results['score_order'] if 'score_order' in results else 'increasing' ) activity_score_type = ( activity.get_score_type() if isinstance(activity, CoopGameActivity) else None ) assert activity_score_type is not None scoretype = activity_score_type # Validate types. if playerinfos is not None: assert isinstance(playerinfos, list) assert all(isinstance(i, PlayerInfo) for i in playerinfos) # Looks like we were in a round - check the outcome and # go from there. if outcome == 'restart': # This will pop up back in the same round. self.setactivity(_bascenev1.newactivity(TransitionActivity)) else: self.setactivity( _bascenev1.newactivity( CoopScoreScreen, { 'playerinfos': playerinfos, 'score': score, 'fail_message': fail_message, 'score_order': score_order, 'score_type': scoretype, 'outcome': outcome, 'campaign': self.campaign, 'level': self.campaign_level_name, }, ) ) # No matter what, get the next 2 levels ready to go. self._update_on_deck_game_instances()
# Docs-generation hack; import some stuff that we likely only forward-declared # in our actual source code so that docs tools can find it. from typing import (Coroutine, Any, Literal, Callable, Generator, Awaitable, Sequence, Self) import asyncio from concurrent.futures import Future from pathlib import Path from enum import Enum