import asyncio
import json
import threading
import time
from typing import Any, Optional, Dict, TYPE_CHECKING
import websockets
if TYPE_CHECKING:
from .base_bot_internals import BaseBotInternals
from ..base_bot_abc import BaseBotABC
from ..bot_info import BotInfo
from ..events import (
ConnectedEvent,
DisconnectedEvent,
ConnectionErrorEvent,
GameStartedEvent,
GameEndedEvent,
RoundStartedEvent,
RoundEndedEvent,
)
from .bot_event_handlers import BotEventHandlers
from .internal_event_handlers import InternalEventHandlers
from .json_util import to_json, from_json
from ..initial_position import InitialPosition
from ..bot_exception import BotException
from ..mapper.event_mapper import EventMapper
from ..mapper.game_setup_mapper import GameSetupMapper
from ..mapper.results_mapper import ResultsMapper
from robocode_tank_royale.schema import (
Message,
TickEventForBot,
BotReady,
GameStartedEventForBot,
GameEndedEventForBot,
RoundEndedEventForBot,
SkippedTurnEvent,
ServerHandshake,
RoundStartedEvent as RoundStartedEventForBot,
)
[docs]
class WebSocketHandler:
"""
Websocket handler for Robocode Tank Royale Bot API that handles websocket connections
and messages from the server.
"""
[docs]
def __init__(
self,
base_bot_internals: "BaseBotInternals",
server_url: str,
server_secret: Optional[str],
base_bot: BaseBotABC,
bot_info: BotInfo,
bot_event_handlers: BotEventHandlers,
internal_event_handlers: InternalEventHandlers,
closed_event: threading.Event,
event_queue: 'EventQueue',
):
"""Initialize the websocket handler."""
self.base_bot_internals = base_bot_internals
self.server_url = server_url
self.server_secret: Optional[str] = server_secret
self.base_bot = base_bot
self.bot_info = bot_info
self.bot_event_handlers = bot_event_handlers
self.internal_event_handlers = internal_event_handlers
self.closed_event = closed_event
self.event_queue = event_queue
self.websocket: None | websockets.ClientConnection = None
[docs]
async def connect(self):
"""Connect to the WebSocket server."""
try:
self.websocket = await websockets.connect(self.server_url)
# Publish connected event
self.bot_event_handlers.on_connected.publish(
ConnectedEvent(self.server_url)
)
return self.websocket
except Exception as e:
self.bot_event_handlers.on_connection_error.publish(
ConnectionErrorEvent(self.server_url, e)
)
self.closed_event.set()
raise
[docs]
async def disconnect(self, code: int = 1000, reason: str = ""):
"""Disconnect from the WebSocket server."""
if self.websocket:
await self.websocket.close(code, reason)
[docs]
async def on_close(
self, websocket: websockets.ClientConnection, code: int, reason: str
) -> None:
"""Handle WebSocket close event.""" # Unused parameter, but kept for compatibility
# Publish to both event handlers
disconnected_event = DisconnectedEvent(self.server_url, True, code, reason)
self.bot_event_handlers.on_disconnected.publish(disconnected_event)
self.internal_event_handlers.on_disconnected.publish(disconnected_event)
self.closed_event.set()
[docs]
async def on_error(self, websocket: websockets.ClientConnection, error: Exception):
"""Handle WebSocket error."""
del websocket # Unused parameter, but kept for compatibility
self.bot_event_handlers.on_connection_error.publish(
ConnectionErrorEvent(self.server_url, error)
)
self.closed_event.set()
[docs]
async def receive_messages(self):
"""Main loop for receiving messages from the WebSocket server."""
assert self.websocket is not None, "WebSocket connection is not established."
try:
async for message in self.websocket:
if isinstance(message, bytes):
message = message.decode("utf-8")
assert isinstance(message, str), "Received message is not a string."
await self.process_message(message)
except websockets.exceptions.ConnectionClosed as e:
assert e.rcvd is not None, "ConnectionClosed without received data."
await self.on_close(self.websocket, e.rcvd.code, e.rcvd.reason)
except Exception as e:
print(f'Unexpected error: {e}')
await self.on_error(self.websocket, e)
[docs]
async def process_message(self, message: str):
"""Process the received WebSocket message."""
json_msg = json.loads(message)
if "type" in json_msg:
msg_type = json_msg["type"]
if msg_type == "TickEventForBot":
await self.handle_tick(json_msg)
elif msg_type == "RoundStartedEvent":
await self.handle_round_started(json_msg)
elif msg_type == "RoundEndedEventForBot":
await self.handle_round_ended(json_msg)
elif msg_type == "GameStartedEventForBot":
await self.handle_game_started(json_msg)
elif msg_type == "GameEndedEventForBot":
await self.handle_game_ended(json_msg)
elif msg_type == "SkippedTurnEvent":
await self.handle_skipped_turn(json_msg)
elif msg_type == "ServerHandshake":
await self.handle_server_handshake(json_msg)
elif msg_type == "GameAbortedEvent":
await self.handle_game_aborted()
else:
raise BotException(f"Unsupported WebSocket message type: {msg_type}")
def _is_event_handling_disabled(self, current_turn: int) -> bool:
disabled_turn = self.base_bot_internals.event_handling_disabled_turn
return disabled_turn != 0 and disabled_turn < (int(current_turn) - 1)
[docs]
async def handle_tick(self, json_msg: Dict[Any, Any]) -> None:
"""Handle a tick event from the server."""
# Determine turn number early to apply correct disabled-handling semantics
turn_number = json_msg.get("turn_number") or json_msg.get("turnNumber")
if turn_number is not None and self._is_event_handling_disabled(int(turn_number)):
return
self.base_bot_internals.tick_start_nano_time = time.monotonic_ns()
tick_event_for_bot: TickEventForBot = from_json(json_msg) # type: ignore
mapped_tick_event = EventMapper.map_tick_event(
tick_event_for_bot, self.base_bot
)
self.base_bot_internals.tick_event = mapped_tick_event
# Stage events from this tick into the event queue (Java parity)
self.event_queue.add_events_from_tick(mapped_tick_event)
# mapped_tick_event.events should still be iterable
for event in mapped_tick_event.events:
self.internal_event_handlers.fire_event(event)
# Trigger next turn (not tick-event!)
self.internal_event_handlers.on_next_turn.publish(mapped_tick_event)
[docs]
async def handle_round_started(self, json_msg: Dict[Any, Any]) -> None:
"""Handle a round started event from the server."""
schema_evt: RoundStartedEventForBot = from_json(json_msg) # type: ignore
round_started_event = RoundStartedEvent(schema_evt.round_number)
self.internal_event_handlers.on_round_started.publish(round_started_event)
self.bot_event_handlers.on_round_started.publish(round_started_event)
[docs]
async def handle_round_ended(self, json_msg: Dict[Any, Any]):
"""Handle a round ended event from the server."""
schema_evt: RoundEndedEventForBot = from_json(json_msg) # type: ignore
results = ResultsMapper.map(schema_evt.results)
round_ended_event = RoundEndedEvent(
schema_evt.round_number, schema_evt.turn_number, results
)
self.bot_event_handlers.on_round_ended.publish(round_ended_event)
self.internal_event_handlers.on_round_ended.publish(round_ended_event) # triggers stop_thread()
# Dispatch any queued events (e.g. WonRoundEvent from the last tick). Bot thread is now
# stopped so there is no concurrent dispatch race. Must run before ROUND_STARTED clears
# the event queue.
self.event_queue.dispatch_events(schema_evt.turn_number)
# Transfer any remaining stdout/stderr from event handlers (e.g. on_won_round) before the round ends
self._transfer_std_out_to_bot_intent()
[docs]
async def handle_game_started(self, json_msg: Dict[Any, Any]) -> None:
"""Handle a game started event from the server."""
assert self.websocket is not None, "WebSocket connection is not established."
game_started_event: GameStartedEventForBot = from_json(json_msg) # type: ignore
self.base_bot_internals.my_id = game_started_event.my_id
if game_started_event.teammate_ids is not None:
self.base_bot_internals.teammate_ids = set(
id for id in game_started_event.teammate_ids if id is not None
)
self.base_bot_internals.game_setup = GameSetupMapper.map(
game_started_event.game_setup
)
initial_position = InitialPosition(
game_started_event.start_x,
game_started_event.start_y,
game_started_event.start_direction,
)
self.base_bot_internals.initial_position = initial_position
self.bot_event_handlers.on_game_started.publish(
GameStartedEvent(
game_started_event.my_id,
initial_position,
self.base_bot_internals.game_setup,
)
)
# Send ready signal
await self.websocket.send(to_json(BotReady(type=Message.Type.BOT_READY)))
[docs]
async def handle_game_ended(self, json_msg: Dict[Any, Any]) -> None:
"""Handle a game ended event from the server."""
schema_evt: GameEndedEventForBot = from_json(json_msg) # type: ignore
game_ended_event = GameEndedEvent()
game_ended_event.number_of_rounds = schema_evt.number_of_rounds
game_ended_event.results = ResultsMapper.map(schema_evt.results)
self.bot_event_handlers.on_game_ended.publish(game_ended_event)
self.internal_event_handlers.on_game_ended.publish(game_ended_event)
[docs]
async def handle_game_aborted(self) -> None:
"""Handle a game aborted event from the server."""
self.bot_event_handlers.on_game_aborted.publish(None)
self.internal_event_handlers.on_game_aborted.publish(None)
[docs]
async def handle_skipped_turn(self, json_msg: Dict[Any, Any]) -> None:
"""Handle a skipped turn event from the server."""
schema_evt: SkippedTurnEvent = from_json(json_msg) # type: ignore
skipped_turn_event = EventMapper.map_skipped_turn_event(schema_evt)
self.event_queue.add_event(skipped_turn_event)
[docs]
async def handle_server_handshake(self, json_msg: Dict[Any, Any]) -> None:
"""Handle a server handshake from the server."""
assert self.websocket is not None, "WebSocket connection is not established."
server_handshake: ServerHandshake = from_json(json_msg) # type: ignore
self.base_bot_internals.server_handshake = server_handshake
self._verify_server_version_compatibility(server_handshake.version)
# Validate bot info before sending bot handshake
self._validate_bot_info()
# Reply by sending bot handshake
# Infer droid status by marker interface inheritance (Java parity), with fallback to explicit flag for backward compatibility
try:
from ..droid_abc import DroidABC # type: ignore
except Exception:
DroidABC = None # type: ignore
is_droid: bool = False
if 'DroidABC' in locals() and DroidABC is not None and isinstance(self.base_bot, DroidABC): # type: ignore
is_droid = True
elif hasattr(self.base_bot, "is_droid"):
# Allow legacy bots explicitly setting the flag
is_droid = bool(getattr(self.base_bot, "is_droid"))
assert isinstance(is_droid, bool), "is_droid must be a boolean value"
# Create bot handshake message
from ..internal.bot_handshake_factory import BotHandshakeFactory
bot_handshake = BotHandshakeFactory.create(
server_handshake.session_id, self.bot_info, is_droid, self.server_secret
)
# Send handshake message
# Ensure backward compatibility for tests expecting 'session_id' (snake_case)
payload_str = to_json(bot_handshake)
try:
payload = json.loads(payload_str)
if "sessionId" in payload and "session_id" not in payload:
payload["session_id"] = payload["sessionId"]
await self.websocket.send(json.dumps(payload))
except Exception:
# Fallback to original payload if any unexpected error occurs
await self.websocket.send(payload_str)
def _transfer_std_out_to_bot_intent(self) -> None:
"""Transfer captured stdout/stderr to bot intent for sending to server."""
if self.base_bot_internals.recording_stdout:
output = self.base_bot_internals.recording_stdout.read_next()
if output:
self.base_bot_internals.bot_intent.std_out = output
else:
self.base_bot_internals.bot_intent.std_out = None
if self.base_bot_internals.recording_stderr:
error = self.base_bot_internals.recording_stderr.read_next()
if error:
self.base_bot_internals.bot_intent.std_err = error
else:
self.base_bot_internals.bot_intent.std_err = None
def _verify_server_version_compatibility(self, server_version: Optional[str]) -> None:
"""Verifies that the server uses a protocol version compatible with this Bot API.
Per SemVer, versions are compatible when the major versions are equal; for the 0.x
range anything may change between minor versions, so there the minor versions must be
equal as well. Without this check, an incompatible server and Bot API silently
misinterpret each other's messages, and the bot appears to join the battle but stands
idle without ever scoring.
"""
try:
from importlib.metadata import version as _package_version
api_version: Optional[str] = _package_version("robocode-tank-royale")
except Exception:
return # version unavailable; cannot verify
api = self._parse_major_minor(api_version)
server = self._parse_major_minor(server_version)
if api is None or server is None:
return
incompatible = api[0] != server[0] or (api[0] == 0 and api[1] != server[1])
if incompatible:
message = (
f"Protocol version mismatch: Bot API version {api_version} is not compatible "
f"with server version {server_version}. The major versions must be equal "
f"(and the minor versions as well for major version 0)."
)
import sys
print(message, file=sys.stderr)
raise BotException(message)
@staticmethod
def _parse_major_minor(version: Optional[str]) -> Optional[tuple[int, int]]:
if not version:
return None
import re
match = re.match(r"\s*(\d+)(?:\.(\d+))?", version)
if not match:
return None
return int(match.group(1)), int(match.group(2) or 0)
def _validate_bot_info(self) -> None:
"""Validate bot info before sending handshake to server."""
if self._is_blank(self.bot_info.name):
self._throw_missing_property_exception("name")
if self._is_blank(self.bot_info.version):
self._throw_missing_property_exception("version")
if not self.bot_info.authors or self._is_all_blank(self.bot_info.authors):
self._throw_missing_property_exception("authors")
def _throw_missing_property_exception(self, property_name: str) -> None:
"""Throw a BotException for a missing required property."""
raise BotException(
f"Required bot property '{property_name}' is missing. "
f"This property is required in order for the bot to be recognized when booting it up and "
f"when it needs to join the game. You must set this property in your bot code "
f"or provide a .json configuration file."
)
def _is_blank(self, s: Optional[str]) -> bool:
"""Check if a string is None or whitespace-only."""
return s is None or not s.strip()
def _is_all_blank(self, strings: list[str]) -> bool:
"""Check if all strings in a list are blank."""
return all(self._is_blank(s) for s in strings)