Source code for bot_api.internal.bot_event_handlers

"""
Class used for bot event handlers on the public API.

These handlers may or may not be triggered by the bot event queue, and might
not be handled immediately by the bot logic.
"""

from typing import Any, Dict, Type

from ..base_bot_abc import BaseBotABC
from ..events import (
    ConnectedEvent, DisconnectedEvent, ConnectionErrorEvent,
    GameStartedEvent, GameEndedEvent, RoundStartedEvent, RoundEndedEvent,
    TickEvent, SkippedTurnEvent, DeathEvent, BotDeathEvent,
    HitBotEvent, HitWallEvent, BulletFiredEvent, HitByBulletEvent,
    BulletHitBotEvent, BulletHitBulletEvent, BulletHitWallEvent,
    ScannedBotEvent, WonRoundEvent, CustomEvent, TeamMessageEvent
)
from ..events.bot_event import BotEvent
from ..events.event_abc import EventABC

from .event_handler import EventHandler


[docs] class BotEventHandlers: """ Class used for bot event handlers on the public API. These handlers may or may not be triggered by the bot event queue, and might not be handled immediately by the bot logic. """
[docs] def __init__(self, base_bot: BaseBotABC): """ Initialize a new BotEventHandlers instance. Args: base_bot: The base bot that will handle the events. """ self.on_connected = EventHandler[ConnectedEvent]() self.on_disconnected = EventHandler[DisconnectedEvent]() self.on_connection_error = EventHandler[ConnectionErrorEvent]() self.on_game_started = EventHandler[GameStartedEvent]() self.on_game_ended = EventHandler[GameEndedEvent]() self.on_game_aborted = EventHandler[Any]() self.on_round_started = EventHandler[RoundStartedEvent]() self.on_round_ended = EventHandler[RoundEndedEvent]() self.on_tick = EventHandler[TickEvent]() self.on_skipped_turn = EventHandler[SkippedTurnEvent]() self.on_death = EventHandler[DeathEvent]() self.on_bot_death = EventHandler[BotDeathEvent]() self.on_hit_bot = EventHandler[HitBotEvent]() self.on_hit_wall = EventHandler[HitWallEvent]() self.on_bullet_fired = EventHandler[BulletFiredEvent]() self.on_hit_by_bullet = EventHandler[HitByBulletEvent]() self.on_bullet_hit = EventHandler[BulletHitBotEvent]() self.on_bullet_hit_bullet = EventHandler[BulletHitBulletEvent]() self.on_bullet_hit_wall = EventHandler[BulletHitWallEvent]() self.on_scanned_bot = EventHandler[ScannedBotEvent]() self.on_won_round = EventHandler[WonRoundEvent]() self.on_custom_event = EventHandler[CustomEvent]() self.on_team_message = EventHandler[TeamMessageEvent]() self._event_handler_map: Dict[Type[EventABC], EventHandler[Any]] = {} self._initialize_event_handlers() self._subscribe_to_event_handlers(base_bot)
def _initialize_event_handlers(self) -> None: """Initialize the mapping between event types and their handlers.""" self._event_handler_map[ConnectedEvent] = self.on_connected self._event_handler_map[DisconnectedEvent] = self.on_disconnected self._event_handler_map[ConnectionErrorEvent] = self.on_connection_error self._event_handler_map[GameStartedEvent] = self.on_game_started self._event_handler_map[GameEndedEvent] = self.on_game_ended self._event_handler_map[self.GameAbortedEvent] = self.on_game_aborted self._event_handler_map[RoundStartedEvent] = self.on_round_started self._event_handler_map[RoundEndedEvent] = self.on_round_ended self._event_handler_map[TickEvent] = self.on_tick self._event_handler_map[SkippedTurnEvent] = self.on_skipped_turn self._event_handler_map[DeathEvent] = self.on_death self._event_handler_map[BotDeathEvent] = self.on_bot_death self._event_handler_map[HitBotEvent] = self.on_hit_bot self._event_handler_map[HitWallEvent] = self.on_hit_wall self._event_handler_map[BulletFiredEvent] = self.on_bullet_fired self._event_handler_map[HitByBulletEvent] = self.on_hit_by_bullet self._event_handler_map[BulletHitBotEvent] = self.on_bullet_hit self._event_handler_map[BulletHitBulletEvent] = self.on_bullet_hit_bullet self._event_handler_map[BulletHitWallEvent] = self.on_bullet_hit_wall self._event_handler_map[ScannedBotEvent] = self.on_scanned_bot self._event_handler_map[WonRoundEvent] = self.on_won_round self._event_handler_map[CustomEvent] = self.on_custom_event self._event_handler_map[TeamMessageEvent] = self.on_team_message def _subscribe_to_event_handlers(self, base_bot: BaseBotABC) -> None: """ Subscribe the base bot to all event handlers. Args: base_bot: The base bot that will handle the events. """ self.on_connected.subscribe(base_bot.on_connected) self.on_disconnected.subscribe(base_bot.on_disconnected) self.on_connection_error.subscribe(base_bot.on_connection_error) self.on_game_started.subscribe(base_bot.on_game_started) self.on_game_ended.subscribe(base_bot.on_game_ended) self.on_round_started.subscribe(base_bot.on_round_started) self.on_round_ended.subscribe(base_bot.on_round_ended) self.on_tick.subscribe(base_bot.on_tick) self.on_skipped_turn.subscribe(base_bot.on_skipped_turn) self.on_death.subscribe(base_bot.on_death) self.on_bot_death.subscribe(base_bot.on_bot_death) self.on_hit_bot.subscribe(base_bot.on_hit_bot) self.on_hit_wall.subscribe(base_bot.on_hit_wall) self.on_bullet_fired.subscribe(base_bot.on_bullet_fired) self.on_hit_by_bullet.subscribe(base_bot.on_hit_by_bullet) self.on_bullet_hit.subscribe(base_bot.on_bullet_hit) self.on_bullet_hit_bullet.subscribe(base_bot.on_bullet_hit_bullet) self.on_bullet_hit_wall.subscribe(base_bot.on_bullet_hit_wall) self.on_scanned_bot.subscribe(base_bot.on_scanned_bot) self.on_won_round.subscribe(base_bot.on_won_round) self.on_custom_event.subscribe(base_bot.on_custom_event) self.on_team_message.subscribe(base_bot.on_team_message)
[docs] def fire_event(self, event: BotEvent) -> None: """ Fire an event to its registered handler. Args: event: The event to fire. Raises: IllegalStateException: If there is no handler for the event type. """ event_type = type(event) handler = self._event_handler_map.get(event_type) if handler is not None: handler.publish(event) else: raise RuntimeError(f"Unhandled event type: {event}")
# Virtual (fake) event:
[docs] class GameAbortedEvent(EventABC): """Virtual event class representing a game aborted event.""" pass