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