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