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