# 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 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