Python Starter Bot

build.md

@@ -1,10 +1,10 @@
-##Структура проекта
+## Структура проекта
 
 - `common` - модуль с общими классами, используемыми и на стороне сервера, и на стороне бота
 - `server` - модуль с кодом сервера. Дорабатывается исключительно администраторами.
 - `starter-bot` - модуль-заготовка для бота. Он же - простейший пример. Можно брать за основу для работы над ботом.
 
-##Сборка и запуск проекта
+## Сборка и запуск проекта
 
 Сборка проекта осуществляется с помощью [Maven](https://ru.wikipedia.org/wiki/Apache_Maven).
 Шаги сборки:

starter-bot-python/README.md

@@ -0,0 +1,12 @@
+## 🐍 Python Starter bot
+
+### Как использовать
+
+[`main.py`](main.py) - настройки подключения к серверу
+
+[`bot.py`](bot.py) - код бота и его параметры
+
+Требования: **Python 3.10+**
+
+---
+_[Артем **nGragas** Корников](https://t.me/Rush_iam) для [Croc Bot Battle](https://brainz.croc.ru/hello-work)_

starter-bot-python/bot.py

@@ -0,0 +1,38 @@
+from client.bot_base import BotBase
+from client.message.extra_types import Mode
+from client.message.messages import MatchStarted, Update
+
+# Параметры регистрации бота на сервере
+bot_name = 'SuperStarter'
+bot_secret = ''
+mode = Mode.FRIENDLY
+
+
+class Bot(BotBase):
+    # Старт
+    def on_match_start(self, match_info: MatchStarted):
+        # Правила матча
+        self.match_info = match_info
+        self.id = match_info.your_id
+        print(match_info)
+        print(f'Матч стартовал! Бот <{self.name}> готов')
+
+    # Каждый ход
+    def on_update(self, update: Update) -> tuple[int, int]:
+        # Данные раунда: что бот "видит"
+        round_number = update.round
+        coins = update.coin
+        blocks = update.block
+        my_bot = next((bot for bot in update.bot if bot.id == self.id), None)
+        opponents = [bot for bot in update.bot if bot.id != self.id]
+
+        # Выбираем направление движения
+        import random
+        dx = random.choice([-1, 0, 1])
+        dy = random.choice([-1, 0, 1])
+
+        return dx, dy  # Отправляем ход серверу
+
+    # Конец матча
+    def on_match_over(self) -> None:
+        print('Матч окончен')

starter-bot-python/client/bot_base.py

@@ -0,0 +1,26 @@
+from dataclasses import dataclass, field
+from typing import TypeVar
+
+from client.message.extra_types import Mode
+from client.message.messages import MatchStarted, Update
+
+
+BotImpl = TypeVar('BotImpl', bound='BotBase')
+
+
+@dataclass
+class BotBase:
+    name: str
+    secret: str = None
+    mode: Mode = Mode.FRIENDLY
+    match_info: MatchStarted = field(init=False)
+    id: int = field(init=False)
+
+    def on_match_start(self, match_info: MatchStarted):
+        raise NotImplementedError
+
+    def on_update(self, update: Update) -> tuple[int, int]:
+        raise NotImplementedError
+
+    def on_match_over(self) -> None:
+        raise NotImplementedError

starter-bot-python/client/bot_match_runner.py

@@ -0,0 +1,21 @@
+from .bot_base import BotImpl
+from .client import HypernullClient
+from .message import messages
+
+
+class BotMatchRunner:
+    def __init__(self, bot: BotImpl, client: HypernullClient):
+        self.bot = bot
+        self.client = client
+
+    def run(self) -> None:
+        self.client.register(self.bot)
+
+        match_info: messages.MatchStarted = self.client.get()
+        self.bot.on_match_start(match_info)
+
+        while update := self.client.get_update():
+            dx, dy = self.bot.on_update(update)
+            self.client.move(dx, dy)
+
+        self.bot.on_match_over()

starter-bot-python/client/client.py

@@ -0,0 +1,48 @@
+from .bot_base import BotImpl
+from .socket_session import SocketSession
+from .message import factory, messages, extra_types
+
+
+class HypernullClient:
+    version: int = 1
+
+    def __init__(self, host: str = 'localhost', port: int = 2021):
+        self.session = SocketSession(host, port)
+        msg = self.get()
+        if not isinstance(msg, messages.Hello):
+            raise Exception(
+                f'Wanted message {messages.Hello.__name__}, got: {type(msg)}'
+            )
+
+        if msg.protocol_version != self.version:
+            raise Exception(
+                f'Client v{self.version}, but Server v{msg.protocol_version}'
+            )
+
+    def get(self) -> factory.Message:
+        data = self.session.read()
+        return factory.MessageFactory.load(data)
+
+    def send(self, msg: messages.MessageBase) -> None:
+        data = msg.dump()
+        self.session.write(data)
+
+    def register(self, bot: BotImpl) -> None:
+        register = messages.Register(
+            bot_name=bot.name,
+            bot_secret=bot.secret,
+            mode=bot.mode,
+        )
+        self.send(register)
+
+    def get_update(self) -> messages.Update | None:
+        update: messages.Update | messages.MatchOver = self.get()
+        if isinstance(update, messages.MatchOver):
+            return None
+        return update
+
+    def move(self, dx: int, dy: int) -> None:
+        move = messages.Move(
+            offset=extra_types.XY(dx, dy)
+        )
+        self.send(move)

starter-bot-python/client/message/base.py

@@ -0,0 +1,18 @@
+from dataclasses import dataclass, asdict
+
+
+@dataclass
+class MessageBase:
+    @classmethod
+    def type(cls) -> str:
+        return cls.__name__.lower()
+
+    def dump(self) -> str:
+        command = self.type()
+        params = '\n'.join(
+            f'{k} {" ".join(map(str, v.values())) if isinstance(v, dict) else v}'
+            for k, v in asdict(self).items() if v not in [None, '']
+        )
+        return f'{command}\n' \
+               f'{params}\n' \
+               f'end\n'

starter-bot-python/client/message/extra_types.py

@@ -0,0 +1,19 @@
+from dataclasses import dataclass
+from enum import Enum
+
+
+class Mode(str, Enum):
+    FRIENDLY = 'FRIENDLY'
+    DEATHMATCH = 'DEATHMATCH'
+
+
+@dataclass
+class XY:
+    x: int
+    y: int
+
+
+@dataclass
+class BotInfo(XY):
+    coins: int
+    id: int

starter-bot-python/client/message/factory.py

@@ -0,0 +1,60 @@
+import dataclasses
+import inspect
+import types
+from collections import defaultdict
+from typing import TypeVar, Type
+
+from . import messages
+
+
+Message = TypeVar('Message', bound=messages.MessageBase)
+
+
+class MessageFactory:
+    _known_message_types: dict[str, Type[Message]] = dict(
+        inspect.getmembers(messages, inspect.isclass)
+    )
+
+    @classmethod
+    def load(cls, data: list[str]) -> Message:
+        if not data:
+            raise Exception('got empty data')
+
+        command = cls._to_camel_case(data[0])
+        if command not in cls._known_message_types:
+            raise Exception(f'unknown command: {command}')
+
+        message_class = cls._known_message_types[command]
+        field_type_mapping: dict[str, tuple[type | None, type]] = {
+            field.name: cls._get_field_types(field)
+            for field in dataclasses.fields(message_class)
+        }
+
+        params = defaultdict(list)
+        for row in data[1:-1]:
+            name, *value = row.split()
+
+            container, real_type = field_type_mapping[name]
+            if container is None:
+                params[name] = real_type(*value)
+            elif container is list:
+                # map(int, ) assumes that all nested types has only int fields
+                params[name].append(real_type(*map(int, value)))
+            else:
+                raise Exception(f'cannot handle {command}:{name}:{container}')
+
+        return message_class(**params)
+
+    @staticmethod
+    def _get_field_types(field: dataclasses.Field) -> tuple[type | None, type]:
+        if isinstance(field.type, types.GenericAlias):
+            container = field.type.__origin__
+            real_type = field.type.__args__[0]
+        else:
+            container = None
+            real_type = field.type
+        return container, real_type
+
+    @staticmethod
+    def _to_camel_case(snake_case: str) -> str:
+        return ''.join(t.title() for t in snake_case.split('_'))

starter-bot-python/client/message/messages.py

@@ -0,0 +1,48 @@
+from dataclasses import dataclass, field
+
+from . import extra_types
+from .base import MessageBase
+
+
+@dataclass
+class Hello(MessageBase):
+    protocol_version: int
+
+
+@dataclass
+class Register(MessageBase):
+    bot_name: str
+    bot_secret: str
+    mode: extra_types.Mode
+
+
+@dataclass
+class MatchStarted(MessageBase):
+    num_rounds: int
+    mode: extra_types.Mode
+    map_size: extra_types.XY
+    your_id: int
+    view_radius: int
+    mining_radius: int
+    attack_radius: int
+    move_time_limit: int
+    match_id: int = 0
+    num_bots: int = 0
+
+
+@dataclass
+class Update(MessageBase):
+    round: int
+    bot: list[extra_types.BotInfo] = field(default_factory=list)
+    block: list[extra_types.XY] = field(default_factory=list)
+    coin: list[extra_types.XY] = field(default_factory=list)
+
+
+@dataclass
+class Move(MessageBase):
+    offset: extra_types.XY
+
+
+@dataclass
+class MatchOver(MessageBase):
+    pass

starter-bot-python/client/socket_session.py

@@ -0,0 +1,36 @@
+import logging
+import socket
+
+
+class SocketSession:
+    _buffer_size = 8192
+
+    def __init__(self, host: str, port: int):
+        self.socket = socket.socket()
+        self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+        self.socket.connect((host, port))
+        self.buffer = bytearray()
+
+    def __del__(self):
+        self.socket.close()
+
+    def read(self) -> list[str]:
+        end_index = self._find_end_index()
+        while end_index == -1:
+            self.buffer += self.socket.recv(self._buffer_size)
+            end_index = self._find_end_index()
+
+        data = self.buffer[:end_index + 3]
+        self.buffer = self.buffer[end_index + 4:]
+
+        if len(self.buffer) > 0:
+            logging.warning('skipping round, seems like your bot had timed out')
+            return self.read()
+
+        return data.decode().split('\n')
+
+    def _find_end_index(self):
+        return self.buffer.find(b'end\n', len(self.buffer) - self._buffer_size)
+
+    def write(self, data: str) -> None:
+        self.socket.sendall(data.encode())

starter-bot-python/main.py

@@ -0,0 +1,12 @@
+from bot import Bot, bot_name, bot_secret, mode
+from client.bot_match_runner import BotMatchRunner
+from client.client import HypernullClient
+
+server_host = 'localhost'
+server_port = 2021
+
+if __name__ == '__main__':
+    BotMatchRunner(
+        bot=Bot(bot_name, bot_secret, mode),
+        client=HypernullClient(server_host, server_port),
+    ).run()