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