Немного теории
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: boolunique_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
