1# Released under the MIT License. See LICENSE for details.
  2#
  3"""Defines a capture-the-flag game."""
  4
  5# ba_meta require api 8
  6# (see https://ballistica.net/wiki/meta-tag-system)
  7
  8from __future__ import annotations
  9
 10import logging
 11from typing import TYPE_CHECKING
 12
 13from bascenev1lib.actor.playerspaz import PlayerSpaz
 14from bascenev1lib.actor.scoreboard import Scoreboard
 15from bascenev1lib.actor.flag import (
 16    FlagFactory,
 17    Flag,
 18    FlagPickedUpMessage,
 19    FlagDroppedMessage,
 20    FlagDiedMessage,
 21)
 22import bascenev1 as bs
 23
 24if TYPE_CHECKING:
 25    from typing import Any, Sequence
 26
 27
 28class CTFFlag(Flag):
 29    """Special flag type for CTF games."""
 30
 31    activity: CaptureTheFlagGame
 32
 33    def __init__(self, team: Team):
 34        assert team.flagmaterial is not None
 35        super().__init__(
 36            materials=[team.flagmaterial],
 37            position=team.base_pos,
 38            color=team.color,
 39        )
 40        self._team = team
 41        self.held_count = 0
 42        self.counter = bs.newnode(
 43            'text',
 44            owner=self.node,
 45            attrs={'in_world': True, 'scale': 0.02, 'h_align': 'center'},
 46        )
 47        self.reset_return_times()
 48        self.last_player_to_hold: Player | None = None
 49        self.time_out_respawn_time: int | None = None
 50        self.touch_return_time: float | None = None
 51
 52    def reset_return_times(self) -> None:
 53        """Clear flag related times in the activity."""
 54        self.time_out_respawn_time = int(self.activity.flag_idle_return_time)
 55        self.touch_return_time = float(self.activity.flag_touch_return_time)
 56
 57    @property
 58    def team(self) -> Team:
 59        """The flag's team."""
 60        return self._team
 61
 62
 63class Player(bs.Player['Team']):
 64    """Our player type for this game."""
 65
 66    def __init__(self) -> None:
 67        self.touching_own_flag = 0
 68
 69
 70class Team(bs.Team[Player]):
 71    """Our team type for this game."""
 72
 73    def __init__(
 74        self,
 75        base_pos: Sequence[float],
 76        base_region_material: bs.Material,
 77        base_region: bs.Node,
 78        spaz_material_no_flag_physical: bs.Material,
 79        spaz_material_no_flag_collide: bs.Material,
 80        flagmaterial: bs.Material,
 81    ):
 82        self.base_pos = base_pos
 83        self.base_region_material = base_region_material
 84        self.base_region = base_region
 85        self.spaz_material_no_flag_physical = spaz_material_no_flag_physical
 86        self.spaz_material_no_flag_collide = spaz_material_no_flag_collide
 87        self.flagmaterial = flagmaterial
 88        self.score = 0
 89        self.flag_return_touches = 0
 90        self.home_flag_at_base = True
 91        self.touch_return_timer: bs.Timer | None = None
 92        self.enemy_flag_at_base = False
 93        self.flag: CTFFlag | None = None
 94        self.last_flag_leave_time: float | None = None
 95        self.touch_return_timer_ticking: bs.NodeActor | None = None
 96
 97
 98# ba_meta export bascenev1.GameActivity
 99class CaptureTheFlagGame(bs.TeamGameActivity[Player, Team]):
100    """Game of stealing other team's flag and returning it to your base."""
101
102    name = 'Capture the Flag'
103    description = 'Return the enemy flag to score.'
104    available_settings = [
105        bs.IntSetting('Score to Win', min_value=1, default=3),
106        bs.IntSetting(
107            'Flag Touch Return Time',
108            min_value=0,
109            default=0,
110            increment=1,
111        ),
112        bs.IntSetting(
113            'Flag Idle Return Time',
114            min_value=5,
115            default=30,
116            increment=5,
117        ),
118        bs.IntChoiceSetting(
119            'Time Limit',
120            choices=[
121                ('None', 0),
122                ('1 Minute', 60),
123                ('2 Minutes', 120),
124                ('5 Minutes', 300),
125                ('10 Minutes', 600),
126                ('20 Minutes', 1200),
127            ],
128            default=0,
129        ),
130        bs.FloatChoiceSetting(
131            'Respawn Times',
132            choices=[
133                ('Shorter', 0.25),
134                ('Short', 0.5),
135                ('Normal', 1.0),
136                ('Long', 2.0),
137                ('Longer', 4.0),
138            ],
139            default=1.0,
140        ),
141        bs.BoolSetting('Epic Mode', default=False),
142    ]
143
144    @classmethod
145    def supports_session_type(cls, sessiontype: type[bs.Session]) -> bool:
146        return issubclass(sessiontype, bs.DualTeamSession)
147
148    @classmethod
149    def get_supported_maps(cls, sessiontype: type[bs.Session]) -> list[str]:
150        assert bs.app.classic is not None
151        return bs.app.classic.getmaps('team_flag')
152
153    def __init__(self, settings: dict):
154        super().__init__(settings)
155        self._scoreboard = Scoreboard()
156        self._alarmsound = bs.getsound('alarm')
157        self._ticking_sound = bs.getsound('ticking')
158        self._score_sound = bs.getsound('score')
159        self._swipsound = bs.getsound('swip')
160        self._last_score_time = 0
161        self._all_bases_material = bs.Material()
162        self._last_home_flag_notice_print_time = 0.0
163        self._score_to_win = int(settings['Score to Win'])
164        self._epic_mode = bool(settings['Epic Mode'])
165        self._time_limit = float(settings['Time Limit'])
166
167        self.flag_touch_return_time = float(settings['Flag Touch Return Time'])
168        self.flag_idle_return_time = float(settings['Flag Idle Return Time'])
169
170        # Base class overrides.
171        self.slow_motion = self._epic_mode
172        self.default_music = (
173            bs.MusicType.EPIC if self._epic_mode else bs.MusicType.FLAG_CATCHER
174        )
175
176    def get_instance_description(self) -> str | Sequence:
177        if self._score_to_win == 1:
178            return 'Steal the enemy flag.'
179        return 'Steal the enemy flag ${ARG1} times.', self._score_to_win
180
181    def get_instance_description_short(self) -> str | Sequence:
182        if self._score_to_win == 1:
183            return 'return 1 flag'
184        return 'return ${ARG1} flags', self._score_to_win
185
186    def create_team(self, sessionteam: bs.SessionTeam) -> Team:
187        # Create our team instance and its initial values.
188
189        base_pos = self.map.get_flag_position(sessionteam.id)
190        Flag.project_stand(base_pos)
191
192        bs.newnode(
193            'light',
194            attrs={
195                'position': base_pos,
196                'intensity': 0.6,
197                'height_attenuated': False,
198                'volume_intensity_scale': 0.1,
199                'radius': 0.1,
200                'color': sessionteam.color,
201            },
202        )
203
204        base_region_mat = bs.Material()
205        pos = base_pos
206        base_region = bs.newnode(
207            'region',
208            attrs={
209                'position': (pos[0], pos[1] + 0.75, pos[2]),
210                'scale': (0.5, 0.5, 0.5),
211                'type': 'sphere',
212                'materials': [base_region_mat, self._all_bases_material],
213            },
214        )
215
216        spaz_mat_no_flag_physical = bs.Material()
217        spaz_mat_no_flag_collide = bs.Material()
218        flagmat = bs.Material()
219
220        team = Team(
221            base_pos=base_pos,
222            base_region_material=base_region_mat,
223            base_region=base_region,
224            spaz_material_no_flag_physical=spaz_mat_no_flag_physical,
225            spaz_material_no_flag_collide=spaz_mat_no_flag_collide,
226            flagmaterial=flagmat,
227        )
228
229        # Some parts of our spazzes don't collide physically with our
230        # flags but generate callbacks.
231        spaz_mat_no_flag_physical.add_actions(
232            conditions=('they_have_material', flagmat),
233            actions=(
234                ('modify_part_collision', 'physical', False),
235                (
236                    'call',
237                    'at_connect',
238                    lambda: self._handle_touching_own_flag(team, True),
239                ),
240                (
241                    'call',
242                    'at_disconnect',
243                    lambda: self._handle_touching_own_flag(team, False),
244                ),
245            ),
246        )
247
248        # Other parts of our spazzes don't collide with our flags at all.
249        spaz_mat_no_flag_collide.add_actions(
250            conditions=('they_have_material', flagmat),
251            actions=('modify_part_collision', 'collide', False),
252        )
253
254        # We wanna know when *any* flag enters/leaves our base.
255        base_region_mat.add_actions(
256            conditions=('they_have_material', FlagFactory.get().flagmaterial),
257            actions=(
258                ('modify_part_collision', 'collide', True),
259                ('modify_part_collision', 'physical', False),
260                (
261                    'call',
262                    'at_connect',
263                    lambda: self._handle_flag_entered_base(team),
264                ),
265                (
266                    'call',
267                    'at_disconnect',
268                    lambda: self._handle_flag_left_base(team),
269                ),
270            ),
271        )
272
273        return team
274
275    def on_team_join(self, team: Team) -> None:
276        # Can't do this in create_team because the team's color/etc. have
277        # not been wired up yet at that point.
278        self._spawn_flag_for_team(team)
279        self._update_scoreboard()
280
281    def on_begin(self) -> None:
282        super().on_begin()
283        self.setup_standard_time_limit(self._time_limit)
284        self.setup_standard_powerup_drops()
285        bs.timer(1.0, call=self._tick, repeat=True)
286
287    def _spawn_flag_for_team(self, team: Team) -> None:
288        team.flag = CTFFlag(team)
289        team.flag_return_touches = 0
290        self._flash_base(team, length=1.0)
291        assert team.flag.node
292        self._swipsound.play(position=team.flag.node.position)
293
294    def _handle_flag_entered_base(self, team: Team) -> None:
295        try:
296            flag = bs.getcollision().opposingnode.getdelegate(CTFFlag, True)
297        except bs.NotFoundError:
298            # Don't think this should logically ever happen.
299            print('Error getting CTFFlag in entering-base callback.')
300            return
301
302        if flag.team is team:
303            team.home_flag_at_base = True
304
305            # If the enemy flag is already here, score!
306            if team.enemy_flag_at_base:
307                # And show team name which scored (but actually we could
308                # show here player who returned enemy flag).
309                self.show_zoom_message(
310                    bs.Lstr(
311                        resource='nameScoresText', subs=[('${NAME}', team.name)]
312                    ),
313                    color=team.color,
314                )
315                self._score(team)
316        else:
317            team.enemy_flag_at_base = True
318            if team.home_flag_at_base:
319                # Award points to whoever was carrying the enemy flag.
320                player = flag.last_player_to_hold
321                if player and player.team is team:
322                    self.stats.player_scored(player, 50, big_message=True)
323
324                # Update score and reset flags.
325                self._score(team)
326
327            # If the home-team flag isn't here, print a message to that effect.
328            else:
329                # Don't want slo-mo affecting this
330                curtime = bs.basetime()
331                if curtime - self._last_home_flag_notice_print_time > 5.0:
332                    self._last_home_flag_notice_print_time = curtime
333                    bpos = team.base_pos
334                    tval = bs.Lstr(resource='ownFlagAtYourBaseWarning')
335                    tnode = bs.newnode(
336                        'text',
337                        attrs={
338                            'text': tval,
339                            'in_world': True,
340                            'scale': 0.013,
341                            'color': (1, 1, 0, 1),
342                            'h_align': 'center',
343                            'position': (bpos[0], bpos[1] + 3.2, bpos[2]),
344                        },
345                    )
346                    bs.timer(5.1, tnode.delete)
347                    bs.animate(
348                        tnode, 'scale', {0.0: 0, 0.2: 0.013, 4.8: 0.013, 5.0: 0}
349                    )
350
351    def _tick(self) -> None:
352        # If either flag is away from base and not being held, tick down its
353        # respawn timer.
354        for team in self.teams:
355            flag = team.flag
356            assert flag is not None
357
358            if not team.home_flag_at_base and flag.held_count == 0:
359                time_out_counting_down = True
360                if flag.time_out_respawn_time is None:
361                    flag.reset_return_times()
362                assert flag.time_out_respawn_time is not None
363                flag.time_out_respawn_time -= 1
364                if flag.time_out_respawn_time <= 0:
365                    flag.handlemessage(bs.DieMessage())
366            else:
367                time_out_counting_down = False
368
369            if flag.node and flag.counter:
370                pos = flag.node.position
371                flag.counter.position = (pos[0], pos[1] + 1.3, pos[2])
372
373                # If there's no self-touches on this flag, set its text
374                # to show its auto-return counter.  (if there's self-touches
375                # its showing that time).
376                if team.flag_return_touches == 0:
377                    flag.counter.text = (
378                        str(flag.time_out_respawn_time)
379                        if (
380                            time_out_counting_down
381                            and flag.time_out_respawn_time is not None
382                            and flag.time_out_respawn_time <= 10
383                        )
384                        else ''
385                    )
386                    flag.counter.color = (1, 1, 1, 0.5)
387                    flag.counter.scale = 0.014
388
389    def _score(self, team: Team) -> None:
390        team.score += 1
391        self._score_sound.play()
392        self._flash_base(team)
393        self._update_scoreboard()
394
395        # Have teammates celebrate.
396        for player in team.players:
397            if player.actor:
398                player.actor.handlemessage(bs.CelebrateMessage(2.0))
399
400        # Reset all flags/state.
401        for reset_team in self.teams:
402            if not reset_team.home_flag_at_base:
403                assert reset_team.flag is not None
404                reset_team.flag.handlemessage(bs.DieMessage())
405            reset_team.enemy_flag_at_base = False
406        if team.score >= self._score_to_win:
407            self.end_game()
408
409    def end_game(self) -> None:
410        results = bs.GameResults()
411        for team in self.teams:
412            results.set_team_score(team, team.score)
413        self.end(results=results, announce_delay=0.8)
414
415    def _handle_flag_left_base(self, team: Team) -> None:
416        cur_time = bs.time()
417        try:
418            flag = bs.getcollision().opposingnode.getdelegate(CTFFlag, True)
419        except bs.NotFoundError:
420            # This can happen if the flag stops touching us due to being
421            # deleted; that's ok.
422            return
423
424        if flag.team is team:
425            # Check times here to prevent too much flashing.
426            if (
427                team.last_flag_leave_time is None
428                or cur_time - team.last_flag_leave_time > 3.0
429            ):
430                self._alarmsound.play(position=team.base_pos)
431                self._flash_base(team)
432            team.last_flag_leave_time = cur_time
433            team.home_flag_at_base = False
434        else:
435            team.enemy_flag_at_base = False
436
437    def _touch_return_update(self, team: Team) -> None:
438        # Count down only while its away from base and not being held.
439        assert team.flag is not None
440        if team.home_flag_at_base or team.flag.held_count > 0:
441            team.touch_return_timer_ticking = None
442            return  # No need to return when its at home.
443        if team.touch_return_timer_ticking is None:
444            team.touch_return_timer_ticking = bs.NodeActor(
445                bs.newnode(
446                    'sound',
447                    attrs={
448                        'sound': self._ticking_sound,
449                        'positional': False,
450                        'loop': True,
451                    },
452                )
453            )
454        flag = team.flag
455        if flag.touch_return_time is not None:
456            flag.touch_return_time -= 0.1
457            if flag.counter:
458                flag.counter.text = f'{flag.touch_return_time:.1f}'
459                flag.counter.color = (1, 1, 0, 1)
460                flag.counter.scale = 0.02
461
462            if flag.touch_return_time <= 0.0:
463                self._award_players_touching_own_flag(team)
464                flag.handlemessage(bs.DieMessage())
465
466    def _award_players_touching_own_flag(self, team: Team) -> None:
467        for player in team.players:
468            if player.touching_own_flag > 0:
469                return_score = 10 + 5 * int(self.flag_touch_return_time)
470                self.stats.player_scored(
471                    player, return_score, screenmessage=False
472                )
473
474    def _handle_touching_own_flag(self, team: Team, connecting: bool) -> None:
475        """Called when a player touches or stops touching their own team flag.
476
477        We keep track of when each player is touching their own flag so we
478        can award points when returned.
479        """
480        player: Player | None
481        try:
482            spaz = bs.getcollision().sourcenode.getdelegate(PlayerSpaz, True)
483        except bs.NotFoundError:
484            return
485
486        player = spaz.getplayer(Player, True)
487
488        if player:
489            player.touching_own_flag += 1 if connecting else -1
490
491        # If return-time is zero, just kill it immediately.. otherwise keep
492        # track of touches and count down.
493        if float(self.flag_touch_return_time) <= 0.0:
494            assert team.flag is not None
495            if (
496                connecting
497                and not team.home_flag_at_base
498                and team.flag.held_count == 0
499            ):
500                self._award_players_touching_own_flag(team)
501                bs.getcollision().opposingnode.handlemessage(bs.DieMessage())
502
503        # Takes a non-zero amount of time to return.
504        else:
505            if connecting:
506                team.flag_return_touches += 1
507                if team.flag_return_touches == 1:
508                    team.touch_return_timer = bs.Timer(
509                        0.1,
510                        call=bs.Call(self._touch_return_update, team),
511                        repeat=True,
512                    )
513                    team.touch_return_timer_ticking = None
514            else:
515                team.flag_return_touches -= 1
516                if team.flag_return_touches == 0:
517                    team.touch_return_timer = None
518                    team.touch_return_timer_ticking = None
519            if team.flag_return_touches < 0:
520                logging.exception('CTF flag_return_touches < 0')
521
522    def _flash_base(self, team: Team, length: float = 2.0) -> None:
523        light = bs.newnode(
524            'light',
525            attrs={
526                'position': team.base_pos,
527                'height_attenuated': False,
528                'radius': 0.3,
529                'color': team.color,
530            },
531        )
532        bs.animate(light, 'intensity', {0.0: 0, 0.25: 2.0, 0.5: 0}, loop=True)
533        bs.timer(length, light.delete)
534
535    def spawn_player_spaz(
536        self,
537        player: Player,
538        position: Sequence[float] | None = None,
539        angle: float | None = None,
540    ) -> PlayerSpaz:
541        """Intercept new spazzes and add our team material for them."""
542        spaz = super().spawn_player_spaz(player, position, angle)
543        player = spaz.getplayer(Player, True)
544        team: Team = player.team
545        player.touching_own_flag = 0
546        no_physical_mats: list[bs.Material] = [
547            team.spaz_material_no_flag_physical
548        ]
549        no_collide_mats: list[bs.Material] = [
550            team.spaz_material_no_flag_collide
551        ]
552
553        # Our normal parts should still collide; just not physically
554        # (so we can calc restores).
555        assert spaz.node
556        spaz.node.materials = list(spaz.node.materials) + no_physical_mats
557        spaz.node.roller_materials = (
558            list(spaz.node.roller_materials) + no_physical_mats
559        )
560
561        # Pickups and punches shouldn't hit at all though.
562        spaz.node.punch_materials = (
563            list(spaz.node.punch_materials) + no_collide_mats
564        )
565        spaz.node.pickup_materials = (
566            list(spaz.node.pickup_materials) + no_collide_mats
567        )
568        spaz.node.extras_material = (
569            list(spaz.node.extras_material) + no_collide_mats
570        )
571        return spaz
572
573    def _update_scoreboard(self) -> None:
574        for team in self.teams:
575            self._scoreboard.set_team_value(
576                team, team.score, self._score_to_win
577            )
578
579    def handlemessage(self, msg: Any) -> Any:
580        if isinstance(msg, bs.PlayerDiedMessage):
581            super().handlemessage(msg)  # Augment standard behavior.
582            self.respawn_player(msg.getplayer(Player))
583
584        elif isinstance(msg, FlagDiedMessage):
585            assert isinstance(msg.flag, CTFFlag)
586            bs.timer(0.1, bs.Call(self._spawn_flag_for_team, msg.flag.team))
587
588        elif isinstance(msg, FlagPickedUpMessage):
589            # Store the last player to hold the flag for scoring purposes.
590            assert isinstance(msg.flag, CTFFlag)
591            try:
592                msg.flag.last_player_to_hold = msg.node.getdelegate(
593                    PlayerSpaz, True
594                ).getplayer(Player, True)
595            except bs.NotFoundError:
596                pass
597
598            msg.flag.held_count += 1
599            msg.flag.reset_return_times()
600
601        elif isinstance(msg, FlagDroppedMessage):
602            # Store the last player to hold the flag for scoring purposes.
603            assert isinstance(msg.flag, CTFFlag)
604            msg.flag.held_count -= 1
605
606        else:
607            super().handlemessage(msg)