bastd.actor.playerspaz

  1# Released under the MIT License. See LICENSE for details.
  2#
  3"""Functionality related to player-controlled Spazzes."""
  4
  5from __future__ import annotations
  6
  7from typing import TYPE_CHECKING, TypeVar, overload
  8
  9import ba
 10from bastd.actor.spaz import Spaz
 11
 12if TYPE_CHECKING:
 13    from typing import Any, Sequence, Literal
 14
 15# pylint: disable=invalid-name
 16PlayerType = TypeVar('PlayerType', bound=ba.Player)
 17TeamType = TypeVar('TeamType', bound=ba.Team)
 18# pylint: enable=invalid-name
 19
 20
 21class PlayerSpazHurtMessage:
 22    """A message saying a PlayerSpaz was hurt.
 23
 24    Category: **Message Classes**
 25    """
 26
 27    spaz: PlayerSpaz
 28    """The PlayerSpaz that was hurt"""
 29
 30    def __init__(self, spaz: PlayerSpaz):
 31        """Instantiate with the given ba.Spaz value."""
 32        self.spaz = spaz
 33
 34
 35class PlayerSpaz(Spaz):
 36    """A Spaz subclass meant to be controlled by a ba.Player.
 37
 38    Category: **Gameplay Classes**
 39
 40    When a PlayerSpaz dies, it delivers a ba.PlayerDiedMessage
 41    to the current ba.Activity. (unless the death was the result of the
 42    player leaving the game, in which case no message is sent)
 43
 44    When a PlayerSpaz is hurt, it delivers a PlayerSpazHurtMessage
 45    to the current ba.Activity.
 46    """
 47
 48    def __init__(
 49        self,
 50        player: ba.Player,
 51        color: Sequence[float] = (1.0, 1.0, 1.0),
 52        highlight: Sequence[float] = (0.5, 0.5, 0.5),
 53        character: str = 'Spaz',
 54        powerups_expire: bool = True,
 55    ):
 56        """Create a spaz for the provided ba.Player.
 57
 58        Note: this does not wire up any controls;
 59        you must call connect_controls_to_player() to do so.
 60        """
 61
 62        super().__init__(
 63            color=color,
 64            highlight=highlight,
 65            character=character,
 66            source_player=player,
 67            start_invincible=True,
 68            powerups_expire=powerups_expire,
 69        )
 70        self.last_player_attacked_by: ba.Player | None = None
 71        self.last_attacked_time = 0.0
 72        self.last_attacked_type: tuple[str, str] | None = None
 73        self.held_count = 0
 74        self.last_player_held_by: ba.Player | None = None
 75        self._player = player
 76        self._drive_player_position()
 77
 78    # Overloads to tell the type system our return type based on doraise val.
 79
 80    @overload
 81    def getplayer(
 82        self, playertype: type[PlayerType], doraise: Literal[False] = False
 83    ) -> PlayerType | None:
 84        ...
 85
 86    @overload
 87    def getplayer(
 88        self, playertype: type[PlayerType], doraise: Literal[True]
 89    ) -> PlayerType:
 90        ...
 91
 92    def getplayer(
 93        self, playertype: type[PlayerType], doraise: bool = False
 94    ) -> PlayerType | None:
 95        """Get the ba.Player associated with this Spaz.
 96
 97        By default this will return None if the Player no longer exists.
 98        If you are logically certain that the Player still exists, pass
 99        doraise=False to get a non-optional return type.
100        """
101        player: Any = self._player
102        assert isinstance(player, playertype)
103        if not player.exists() and doraise:
104            raise ba.PlayerNotFoundError()
105        return player if player.exists() else None
106
107    def connect_controls_to_player(
108        self,
109        enable_jump: bool = True,
110        enable_punch: bool = True,
111        enable_pickup: bool = True,
112        enable_bomb: bool = True,
113        enable_run: bool = True,
114        enable_fly: bool = True,
115    ) -> None:
116        """Wire this spaz up to the provided ba.Player.
117
118        Full control of the character is given by default
119        but can be selectively limited by passing False
120        to specific arguments.
121        """
122        player = self.getplayer(ba.Player)
123        assert player
124
125        # Reset any currently connected player and/or the player we're
126        # wiring up.
127        if self._connected_to_player:
128            if player != self._connected_to_player:
129                player.resetinput()
130            self.disconnect_controls_from_player()
131        else:
132            player.resetinput()
133
134        player.assigninput(ba.InputType.UP_DOWN, self.on_move_up_down)
135        player.assigninput(ba.InputType.LEFT_RIGHT, self.on_move_left_right)
136        player.assigninput(
137            ba.InputType.HOLD_POSITION_PRESS, self.on_hold_position_press
138        )
139        player.assigninput(
140            ba.InputType.HOLD_POSITION_RELEASE, self.on_hold_position_release
141        )
142        intp = ba.InputType
143        if enable_jump:
144            player.assigninput(intp.JUMP_PRESS, self.on_jump_press)
145            player.assigninput(intp.JUMP_RELEASE, self.on_jump_release)
146        if enable_pickup:
147            player.assigninput(intp.PICK_UP_PRESS, self.on_pickup_press)
148            player.assigninput(intp.PICK_UP_RELEASE, self.on_pickup_release)
149        if enable_punch:
150            player.assigninput(intp.PUNCH_PRESS, self.on_punch_press)
151            player.assigninput(intp.PUNCH_RELEASE, self.on_punch_release)
152        if enable_bomb:
153            player.assigninput(intp.BOMB_PRESS, self.on_bomb_press)
154            player.assigninput(intp.BOMB_RELEASE, self.on_bomb_release)
155        if enable_run:
156            player.assigninput(intp.RUN, self.on_run)
157        if enable_fly:
158            player.assigninput(intp.FLY_PRESS, self.on_fly_press)
159            player.assigninput(intp.FLY_RELEASE, self.on_fly_release)
160
161        self._connected_to_player = player
162
163    def disconnect_controls_from_player(self) -> None:
164        """
165        Completely sever any previously connected
166        ba.Player from control of this spaz.
167        """
168        if self._connected_to_player:
169            self._connected_to_player.resetinput()
170            self._connected_to_player = None
171
172            # Send releases for anything in case its held.
173            self.on_move_up_down(0)
174            self.on_move_left_right(0)
175            self.on_hold_position_release()
176            self.on_jump_release()
177            self.on_pickup_release()
178            self.on_punch_release()
179            self.on_bomb_release()
180            self.on_run(0.0)
181            self.on_fly_release()
182        else:
183            print(
184                'WARNING: disconnect_controls_from_player() called for'
185                ' non-connected player'
186            )
187
188    def handlemessage(self, msg: Any) -> Any:
189        # FIXME: Tidy this up.
190        # pylint: disable=too-many-branches
191        # pylint: disable=too-many-statements
192        # pylint: disable=too-many-nested-blocks
193        assert not self.expired
194
195        # Keep track of if we're being held and by who most recently.
196        if isinstance(msg, ba.PickedUpMessage):
197            # Augment standard behavior.
198            super().handlemessage(msg)
199            self.held_count += 1
200            picked_up_by = msg.node.source_player
201            if picked_up_by:
202                self.last_player_held_by = picked_up_by
203        elif isinstance(msg, ba.DroppedMessage):
204            # Augment standard behavior.
205            super().handlemessage(msg)
206            self.held_count -= 1
207            if self.held_count < 0:
208                print('ERROR: spaz held_count < 0')
209
210            # Let's count someone dropping us as an attack.
211            picked_up_by = msg.node.source_player
212            if picked_up_by:
213                self.last_player_attacked_by = picked_up_by
214                self.last_attacked_time = ba.time()
215                self.last_attacked_type = ('picked_up', 'default')
216        elif isinstance(msg, ba.StandMessage):
217            super().handlemessage(msg)  # Augment standard behavior.
218
219            # Our Spaz was just moved somewhere. Explicitly update
220            # our associated player's position in case it is being used
221            # for logic (otherwise it will be out of date until next step)
222            self._drive_player_position()
223
224        elif isinstance(msg, ba.DieMessage):
225
226            # Report player deaths to the game.
227            if not self._dead:
228
229                # Immediate-mode or left-game deaths don't count as 'kills'.
230                killed = (
231                    not msg.immediate and msg.how is not ba.DeathType.LEFT_GAME
232                )
233
234                activity = self._activity()
235
236                player = self.getplayer(ba.Player, False)
237                if not killed:
238                    killerplayer = None
239                else:
240                    # If this player was being held at the time of death,
241                    # the holder is the killer.
242                    if self.held_count > 0 and self.last_player_held_by:
243                        killerplayer = self.last_player_held_by
244                    else:
245                        # Otherwise, if they were attacked by someone in the
246                        # last few seconds, that person is the killer.
247                        # Otherwise it was a suicide.
248                        # FIXME: Currently disabling suicides in Co-Op since
249                        #  all bot kills would register as suicides; need to
250                        #  change this from last_player_attacked_by to
251                        #  something like last_actor_attacked_by to fix that.
252                        if (
253                            self.last_player_attacked_by
254                            and ba.time() - self.last_attacked_time < 4.0
255                        ):
256                            killerplayer = self.last_player_attacked_by
257                        else:
258                            # ok, call it a suicide unless we're in co-op
259                            if activity is not None and not isinstance(
260                                activity.session, ba.CoopSession
261                            ):
262                                killerplayer = player
263                            else:
264                                killerplayer = None
265
266                # We should never wind up with a dead-reference here;
267                # we want to use None in that case.
268                assert killerplayer is None or killerplayer
269
270                # Only report if both the player and the activity still exist.
271                if killed and activity is not None and player:
272                    activity.handlemessage(
273                        ba.PlayerDiedMessage(
274                            player, killed, killerplayer, msg.how
275                        )
276                    )
277
278            super().handlemessage(msg)  # Augment standard behavior.
279
280        # Keep track of the player who last hit us for point rewarding.
281        elif isinstance(msg, ba.HitMessage):
282            source_player = msg.get_source_player(type(self._player))
283            if source_player:
284                self.last_player_attacked_by = source_player
285                self.last_attacked_time = ba.time()
286                self.last_attacked_type = (msg.hit_type, msg.hit_subtype)
287            super().handlemessage(msg)  # Augment standard behavior.
288            activity = self._activity()
289            if activity is not None and self._player.exists():
290                activity.handlemessage(PlayerSpazHurtMessage(self))
291        else:
292            return super().handlemessage(msg)
293        return None
294
295    def _drive_player_position(self) -> None:
296        """Drive our ba.Player's official position
297
298        If our position is changed explicitly, this should be called again
299        to instantly update the player position (otherwise it would be out
300        of date until the next sim step)
301        """
302        player = self._player
303        if player:
304            assert self.node
305            assert player.node
306            self.node.connectattr('torso_position', player.node, 'position')