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