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