Реализация и применение Entity Component System на примере python

от автора

Немного теории

Entity Component System (ECS) — это паттерн, используемый при разработке видеоигр, для хранения игровых объектов.

Компоненты (Components)

Все характеристики объектов находятся в минимальных структурах данных — компонентах, хранящих схожие смысловые величины. Например здоровье может являться компонентом, в котором будет храниться его максимальное и текущее значения.

Хорошей практикой является создание наиболее ёмких смысловых компонентов. Никому не нужны раздутые компоненты, которые можно использовать для крайне малого числа сущностей.

Сущности (Entity)

Сами же игровые объекты (сущности) являются ни чем иным как совокупностью различных компонентов.

Ничего кроме id сущность не имеет, однако по этому id можно получить соответствующие компоненты.

В зависимости от компонентов из которых составлены сущности, они могут быть самыми различными объектами, от стрелы до мишени и многого другого. Добавив или убрав компонент можно значительно изменить поведение сущности.

Системы (Systems)

Вот у нас уже есть данные, но они никак не взаимодействуют друг с другом. Для этого и предназначены системы.

Система — набор правил, влияющий на данные компонентов. Системы поочерёдно проходятся по каждому из зависимых компонентов, изменяя их.

При этом каждая система независима от друг друга, и не влияет на компоненты которые к ней не относятся.

Так, создав сущность с несколькими компонентами (например здоровье и координаты), на неё могут влиять две системы (регенерация и передвижение соответственно).

При этом, системы могут быть безболезненно отключены. Убрав систему систему смерти, сущности не будут исчезать и с нулевым здоровьем, при этом всё остальное продолжит работать как прежде.

Преимущества

Производительность

За счёт плоской структуры компонентов открываются широкие возможности оптимизации работы с памятью. Например в python можно использовать слоты для объектов, а в более низкоуровневых языках — эффективнее занимать блоки памяти под целые массивы компонентов.

Расширяемость

Так как каждая система независима, можно легко добавлять новые системы, не ломая старые.

Гибкость

Функционал объекта может быть изменён простым изменением состава компонентов.

Система может быть выключена и это не сломает работу остальных систем.

Пример реализации

В данной статье хочу сконцентрироваться именно на использовании ECS, а не на реализации этого паттерна. Поэтому исходники класса EntityComponentSystem представлены в спойлерах. При желании можете подробнее изучить исходный код, он снабжён достаточным для понимания количеством комментариев.

ecs_types.py
from dataclasses import dataclass from typing import Any, Type  EntityId = str Component = object   @dataclass class StoredSystem:     variables: dict[str, Any]     components: dict[str, Type[Component]]  # key is argument name     has_entity_id_argument: bool     has_ecs_argument: bool

unique_id.py
# Класс простенький, можно спокойно заменить на uuid class UniqueIdGenerator:     last_id = 0      @classmethod     def generate_id(cls) -> str:         cls.last_id += 1         return str(cls.last_id)

entity_component_system.py
import inspect from typing import Callable, Type, Any, Iterator  from ecs_types import EntityId, Component, StoredSystem from unique_id import UniqueIdGenerator   class EntityComponentSystem:     def __init__(self, on_create: Callable[[EntityId, list[Component]], None] = None,                  on_remove: Callable[[EntityId], None] = None):         """         :param on_create:         Хук, отрабатывающий при создании сущности,         например может пригодиться, если сервер сообщает клиентам о появлении новых сущностей          :param on_remove:         Хук, отрабатывающий перед удалением сущности         """          # Здесь хранятся все системы вместе с полученными от них сигнатурами         self._systems: dict[Callable, StoredSystem] = {}         # По типу компонента хранятся словари, содержащие сами компоненты по ключам entity_id         self._components: dict[Type[Component], dict[EntityId, Component]] = {}         self._entities: list[EntityId] = []         self._vars = {}         self.on_create = on_create         self.on_remove = on_remove      def _unsafe_get_component(self, entity_id: EntityId, component_class: Type[Component]) -> Component:         """         Возвращает компонент сущности с типом переданного класса component_class         Кидает KeyError если сущность не существует или не имеет такого компонента         """          return self._components[component_class][entity_id]      def init_component(self, component_class: Type[Component]) -> None:         """         Инициализация класса компонента. Следует вызвать до создания сущностей         """          self._components[component_class] = {}      def add_variable(self, variable_name: str, variable_value: Any) -> None:         """         Инициализация переменной. Далее может быть запрошена любой системой.         """          self._vars[variable_name] = variable_value      def init_system(self, system: Callable):         """         Инициализация системы. Если система зависит от внешней переменной - передайте её в add_variable до инициализации.          Внешние переменные и специальные аргументы (ecs: EntityComponentSystem и entity_id: EntityId) запрашиваются         через указание имени аргумента в функции системы.          Запрашиваемые компоненты указываются через указание типа аргумента (например dummy_health: HealthComponent).         Название аргумента в таком случае может быть названо как угодно.         Запрашиваемый компонент должен быть инициализирован до инициализации системы         """          stored_system = StoredSystem(             components={},             variables={},             has_entity_id_argument=False,             has_ecs_argument=False         )          # Через сигнатуру функции системы узнаем какие данные и компоненты она запрашивает.         # Сохраним в StoredSystem чтобы не перепроверять сигнатуру каждый кадр.         system_params = inspect.signature(system).parameters         for param_name, param in system_params.items():             if param_name == 'entity_id':  # Система может требовать конкретный entity_id для переданных компонентов                 stored_system.has_entity_id_argument = True              elif param_name == 'ecs':  # Системе может потребоваться ссылка на ecs. Например, для удаления сущностей                 stored_system.has_ecs_argument = True              elif param.annotation in self._components:                 stored_system.components[param_name] = param.annotation              elif param_name in self._vars:                 stored_system.variables[param_name] = self._vars[param_name]              else:                 raise Exception(f'Wrong argument: {param_name}')          self._systems[system] = stored_system      def create_entity(self, components: list[Component], entity_id=None) -> EntityId:         """         Создание сущности на основе списка его компонентов         Можно задавать свой entity_id но он обязан быть уникальным         """          if entity_id is None:             entity_id = UniqueIdGenerator.generate_id()         else:             assert entity_id not in self._entities, f"Entity with id {entity_id} already exists"          for component in components:             self._components[component.__class__][entity_id] = component         self._entities.append(entity_id)          if self.on_create:             self.on_create(entity_id, components)          return entity_id      def get_entity_ids_with_components(self, *component_classes) -> set[EntityId]:         """         Получить все entity_id у которых есть каждый из компонентов, указанных в component_classes         """          if not component_classes:             return set(self._entities)          # Если запрошено несколько компонентов - то следует вернуть сущности, обладающие каждым из них         # Это достигается пересечением множеств entity_id по классу компонента         entities = set.intersection(*[set(self._components[component_class].keys())                                       for component_class in component_classes])         return entities      def get_entities_with_components(self, *component_classes) -> Iterator[tuple[EntityId, list[Component]]]:         """         Получить все entity_id вместе с указанными компонентами         """          for entity_id in self.get_entity_ids_with_components(*component_classes):             components = tuple(self._unsafe_get_component(entity_id, component_class)                                for component_class in component_classes)             yield entity_id, components      def update(self) -> None:         """         Вызывает все системы.         Следует вызывать в игровом цикле.         """          for system_function, system in self._systems.items():             for entity_id in self.get_entity_ids_with_components(*system.components.values()):                 special_args = {}                 if system.has_ecs_argument:                     special_args['ecs'] = self                 if system.has_entity_id_argument:                     special_args['entity_id'] = entity_id                                      # Сделано для того чтобы в системе можно было указывать любые имена для запрашиваемых компонентов                 required_components_arguments = {param: self._unsafe_get_component(entity_id, component_name) for                                                  param, component_name in                                                  system.components.items()}                                  system_function(**(required_components_arguments | system.variables | special_args))      def remove_entity(self, entity_id: EntityId):         """         Удаляет сущность         """                  if self.on_remove is not None:             self.on_remove(entity_id)         for components in self._components.values():             components.pop(entity_id, None)         self._entities.remove(entity_id)      def get_component(self, entity_id: EntityId, component_class: Type[Component]):         """         :return         Возвращает компонент сущности с типом переданного класса component_class         Возвращает None если сущность не существует или не имеет такого компонента         """                  return self._components[component_class].get(entity_id, None)      def get_components(self, entity_id: EntityId, component_classes):         """         :return         Возвращает требуемые компоненты сущности.         Возвращает None если сущность не существует или не имеет всех этих компонентов         """                  try:             return tuple(self._unsafe_get_component(entity_id, component_class)                           for component_class in component_classes)         except KeyError:             return None

Если вы собираетесь пользоваться моими исходниками — рекомендую добавить файл с аннотациями типов для удобной работы.

entity_component_system.pyi
from typing import Protocol, Type, TypeVar, overload, Callable, Any, Iterator  from ecs_types import EntityId, Component, StoredSystem  Component1 = TypeVar('Component1') Component2 = TypeVar('Component2') Component3 = TypeVar('Component3') Component4 = TypeVar('Component4')   class EntityComponentSystem(Protocol):     _systems: dict[Callable, StoredSystem]     _components: dict[Type[Component], dict[EntityId, Component]]     _entities: list[EntityId]      _vars: dict[str, Any]     on_create: Callable[[EntityId, list[Component]], None]     on_remove: Callable[[EntityId], None]      def __init__(self, on_create: Callable[[EntityId, list[Component]], None] = None,                  on_remove: Callable[[EntityId], None] = None): ...      @overload     def _unsafe_get_component(self, entity_id: str, component_class: Type[Component1]) -> Component1: ...      @overload     def init_component(self, component_class: Type[Component1]) -> None: ...      @overload     def init_system(self, system: Callable): ...      @overload     def add_variable(self, variable_name: str, variable_value: Any) -> None: ...      @overload     def create_entity(self, components: list[Component1], entity_id=None) -> EntityId: ...      @overload     def get_entity_ids_with_components(self, class1: Type[Component1]) -> set[EntityId]: ...      @overload     def get_entity_ids_with_components(self, class1: Type[Component1], class2: Type[Component2]) -> set[EntityId]: ...      @overload     def get_entity_ids_with_components(self, class1: Type[Component1], class2: Type[Component2], class3: Type[Component3]) -> set[EntityId]: ...      @overload     def get_entity_ids_with_components(self, class1: Type[Component1], class2: Type[Component2], class3: Type[Component3], class4: Type[Component4]) -> set[EntityId]: ...      @overload     def get_entities_with_components(self, class1: Type[Component1]) -> Iterator[tuple[         EntityId, tuple[Component1]]]: ...      @overload     def get_entities_with_components(self, class1: Type[Component1], class2: Type[Component2]) -> Iterator[         tuple[             EntityId, tuple[Component1, Component2]]]: ...      @overload     def get_entities_with_components(self, class1: Type[Component1], class2: Type[Component2], class3: Type[Component3]) -> \     Iterator[tuple[EntityId, tuple[Component1, Component2, Component3]]]: ...      @overload     def get_entities_with_components(self, class1: Type[Component1], class2: Type[Component2], class3: Type[Component3], class4: Type[Component4]) -> Iterator[tuple[         EntityId, tuple[Component1, Component2, Component3, Component4]]]: ...      def update(self) -> None: ...      def remove_entity(self, entity_id: EntityId): ...      def get_component(self, entity_id: EntityId, component_class: Type[Component1]) -> Component1: ...      @overload     def get_components(self, entity_id: EntityId,                        component_classes: tuple[Type[Component1]]) -> tuple[Component1]: ...      @overload     def get_components(self, entity_id: EntityId,                        component_classes: tuple[Type[Component1], Type[Component2]]) -> tuple[         Component1, Component2]: ...      @overload     def get_components(self, entity_id: EntityId,                        component_classes: tuple[Type[Component1], Type[Component2], Type[Component3]]) -> tuple[         Component1, Component2, Component3]: ... 

Пример использования

Давайте посмотрим, как с помощью ECS можно описать простую стрелу, врезающуюся в мишень.

Начнём с импортов:

from entity_component_system import EntityComponentSystem from ecs_types import EntityId from dataclasses import dataclass, field import math

Будем описывать наши компоненты через датаклассы. Ведь в python это наиболее удобный способ описать такие простые объекты.

Во-первых у стрелы должны быть координаты и размеры:

@dataclass(slots=True) class ColliderComponent:     x: float     y: float     radius: float      def distance(self, other: 'ColliderComponent'):         return math.sqrt((self.x - other.x) ** 2 + (self.y - other.y) ** 2)      def is_intersecting(self, other: 'ColliderComponent'):         return self.distance(other) <= self.radius + other.radius

Во-вторых нашей стреле необходима скорость с которой она будет двигаться:

@dataclass(slots=True) class VelocityComponent:     speed_x: float = 0.0     speed_y: float = 0.0

Ещё не забудем указать что она пропадает при контакте, нанося урон:

@dataclass(slots=True) class DamageOnContactComponent:     damage: int     die_on_contact: bool = True

Теперь у нас достаточно компонентов чтобы составить из них стрелу. Укажем как можно её собрать, зная необходимые характеристики:

def create_arrow(x: float, y: float, angle: int, speed: float, damage: int):     arrow_radius = 15     return [         ColliderComponent(x, y, arrow_radius),         # Вектор скорости вычисляется на основе величины скорости и угла под которым пустили стрелу.         VelocityComponent(             speed_x=math.cos(math.radians(angle)) * speed,             speed_y=math.sin(math.radians(angle)) * speed         ),         DamageOnContactComponent(damage)     ]

Перейдём к мишени. Для её создания не хватает компонента здоровья. Опишем его:

@dataclass(slots=True) class HealthComponent:     max_amount: int     amount: int = field(default=None)      def __post_init__(self):         if self.amount is None:             self.amount = self.max_amount      def apply_damage(self, damage: int):         self.amount = max(0, self.amount - damage)

Создадим же фабрику мишеней:

def create_dummy(x: float, y: float, health: int):     dummy_radius = 50     return [         ColliderComponent(x, y, dummy_radius),         HealthComponent(             max_amount=health,         )     ]

Вот мы и подготовили всё данные сущностей и компонентов. Опишем, что с ними надо делать.

Заставим нашу стрелу двигаться. Для этого напишем систему, перемещающую объекты у которых есть скорость:

def velocity_system(velocity: VelocityComponent, collider: ColliderComponent):     collider.x += velocity.speed_x     collider.y += velocity.speed_y

Теперь наша стрела может летать. Скажем, что она должна делать при соприкосновении с мишенью:

def damage_on_contact_system(entity_id: EntityId,                              # Запрашиваем EntityComponentSystem для удаления стрелы при попадании                              ecs: EntityComponentSystem,                              damage_on_contact: DamageOnContactComponent,                              collider: ColliderComponent):     # Проходимся по всем компонентам с координатами и здоровьем     for enemy_id, (enemy_health, enemy_collider) in ecs.get_entities_with_components(HealthComponent,                                                                                      ColliderComponent):         # Пусть стрела и не обладает здоровьем, но эта проверка нужна на тот случай если компонент окажется на сущности где оно есть                if entity_id == enemy_id:             continue                  if collider.is_intersecting(enemy_collider):             enemy_health.apply_damage(damage_on_contact.damage)             if damage_on_contact.die_on_contact:                 ecs.remove_entity(entity_id)                 return

Будем уничтожать сущности, здоровье которых упало до нуля:

def death_system(entity_id: EntityId, health: HealthComponent, ecs: EntityComponentSystem):     if health.amount <= 0:         ecs.remove_entity(entity_id)

Наконец все сущности, компоненты и системы описаны, осталось только убедиться что всё будет работать вместе.

Для начала инициализируем все компоненты и системы:

ecs = EntityComponentSystem()  ecs.init_component(ColliderComponent) ecs.init_component(VelocityComponent) ecs.init_component(DamageOnContactComponent) ecs.init_component(HealthComponent)  ecs.init_system(velocity_system) ecs.init_system(damage_on_contact_system) ecs.init_system(death_system)

Теперь создадим мишень и стрелы, которые её уничтожат:

ecs.create_entity(create_arrow(x=0, y=0, angle=45, speed=2, damage=50)) ecs.create_entity(create_arrow(x=500, y=0, angle=135, speed=1.5, damage=50)) ecs.create_entity(create_arrow(x=0, y=500, angle=-45, speed=1.1, damage=50)) ecs.create_entity(create_arrow(x=500, y=500, angle=-135, speed=1, damage=50)) ecs.create_entity(create_dummy(x=250, y=250, health=200))

Для наглядной демонстрации результата используем pygame (на момент написания документация доступна только через веб архив):

import pygame from pygame import Color from pygame.time import Clock  screen = pygame.display.set_mode((500, 500)) running = True clock = Clock()  while running:     for event in pygame.event.get():          if event.type == pygame.QUIT:             running = False      ecs.update()      screen.fill((93, 161, 48))     for entity_id, (collider,) in ecs.get_entities_with_components(ColliderComponent):         pygame.draw.circle(screen, Color('gray'), (collider.x, collider.y), collider.radius)      pygame.display.flip()     clock.tick(60)

Заключение

Весь исходный код, представленный в статье собран в гисте.

Реализация этого паттерна использована мной при создании онлайн стратегии в реальном времени на pygame. Если эта статья вам понравится, то я напишу статью, описывающую её работу.

За помощь в написании спасибо @AlexandrovRoman


ссылка на оригинал статьи https://habr.com/ru/post/702598/


Комментарии

Добавить комментарий

Ваш адрес email не будет опубликован. Обязательные поля помечены *