511 lines
17 KiB
Python
511 lines
17 KiB
Python
"""
|
|
Système de gestion d'entités, composants et ressources.
|
|
"""
|
|
|
|
|
|
__all__ = [
|
|
"World",
|
|
"Entity",
|
|
"Plugin",
|
|
"Game",
|
|
]
|
|
|
|
|
|
from abc import abstractmethod
|
|
from typing import Iterator, Callable, Tuple, TypeVar, Type, Optional
|
|
from logging import error
|
|
|
|
|
|
_T = TypeVar("_T")
|
|
"""
|
|
Un type générique utilisé pour les composants et les ressources.
|
|
"""
|
|
|
|
|
|
class World:
|
|
"""
|
|
Un monde contenant des entités et des ressources.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
"""
|
|
Permet de créer un nouveau monde vide.
|
|
"""
|
|
self._to_apply: set[Entity] = set()
|
|
self._entities: set[Entity] = set()
|
|
self._mapping: dict[Type, set[Entity]] = {}
|
|
self._to_apply_resources: list[tuple[Type, Optional[object]]] = []
|
|
self._resources: dict[Type, object] = {}
|
|
|
|
def create_entity(self, *components: _T) -> "Entity":
|
|
"""
|
|
Crée une entité avec les composants donnés en paramètres.
|
|
|
|
Paramètres:
|
|
*components: Les composants de l'entité.
|
|
|
|
Retourne:
|
|
L'entité crée.
|
|
"""
|
|
return Entity(self, *components)
|
|
|
|
def remove_entity(self, entity: "Entity") -> None:
|
|
"""
|
|
Supprime une entité du monde.
|
|
|
|
Paramètres:
|
|
entity: L'entité a supprimer.
|
|
"""
|
|
entity._deleted = True
|
|
self._to_apply.add(entity)
|
|
|
|
def set(self, *resources: _T) -> None:
|
|
"""
|
|
Définit les ressources données en paramètres.
|
|
Si les ressources existent deja, elles seront remplacées.
|
|
|
|
Paramètres:
|
|
*resources: Les ressources a definir.
|
|
"""
|
|
for resource in resources:
|
|
self._to_apply_resources.append((type(resource), resource))
|
|
|
|
def remove(self, *resource_types: Type[_T]) -> None:
|
|
"""
|
|
Supprime les ressources données en paramètres.
|
|
|
|
Paramètres:
|
|
*resource_types: Les types des ressources à supprimer.
|
|
"""
|
|
for resource_type in resource_types:
|
|
if resource_type in self._resources:
|
|
self._to_apply_resources.append((resource_type, None))
|
|
|
|
def apply(self) -> None:
|
|
"""
|
|
Applique les changements réaliser dans le monde.
|
|
"""
|
|
for entity in self._to_apply:
|
|
if entity._deleted:
|
|
self._entities.remove(entity)
|
|
for component_type in entity._components:
|
|
self._mapping[component_type].remove(entity)
|
|
else:
|
|
self._entities.add(entity)
|
|
for component_type, component in entity._to_apply:
|
|
if component is None:
|
|
del entity._components[component_type]
|
|
self._mapping[component_type].remove(entity)
|
|
else:
|
|
entity._components[component_type] = component # type: ignore[assignment]
|
|
self._mapping.setdefault(component_type, set()).add(entity)
|
|
entity._to_apply.clear()
|
|
self._to_apply.clear()
|
|
for resource_type, resource in self._to_apply_resources:
|
|
if resource is None:
|
|
del self._resources[resource_type]
|
|
else:
|
|
self._resources[resource_type] = resource
|
|
self._to_apply_resources.clear()
|
|
|
|
def query(
|
|
self, *needed: Type[_T], without: Tuple[Type[_T], ...] = ()
|
|
) -> Iterator["Entity"]:
|
|
"""
|
|
Renvoie les entités qui ont les composants de *needed et sans les composants de *without.
|
|
|
|
Paramètres:
|
|
*needed: Le type de composants que les entités doivent avoir.
|
|
*without: Les type de composants que les entités ne doivent pas avoir.
|
|
|
|
Retourne:
|
|
Les entités qui ont les composants de *needed et sans les composants de *without.
|
|
Si *needed est vide, on retourne toutes les entités qui ont les composants de *without.
|
|
"""
|
|
if not needed:
|
|
for entity in self._entities:
|
|
if all(without_type not in entity for without_type in without):
|
|
yield entity
|
|
else:
|
|
for entity in self._mapping.get(needed[0], set()):
|
|
if all(
|
|
entity in self._mapping.get(component_type, set())
|
|
for component_type in needed[1:]
|
|
) and all(without_type not in entity for without_type in without):
|
|
yield entity
|
|
|
|
def __getitem__(self, resource_type: Type[_T]) -> _T:
|
|
"""
|
|
Renvoie la ressource de type *resource_type.
|
|
|
|
Paramètres:
|
|
resource_type: Le type de ressource à récupérer.
|
|
|
|
Retourne:
|
|
La ressource de type *resource_type.
|
|
"""
|
|
resource: _T = self._resources[resource_type] # type: ignore[assignment]
|
|
return resource
|
|
|
|
def __contains__(self, *resource_types: Type[_T]) -> bool:
|
|
"""
|
|
Renvoie True si le monde contient toutes les ressources de *resource_types.
|
|
|
|
Paramètres:
|
|
*resource_types: Les types de ressource à tester.
|
|
|
|
Retourne:
|
|
Si toutes les ressources de *resource_types sont dans le monde.
|
|
"""
|
|
for resource_type in resource_types:
|
|
if resource_type not in self._resources:
|
|
return False
|
|
return True
|
|
|
|
|
|
class Entity:
|
|
"""
|
|
Une entité du monde.
|
|
"""
|
|
|
|
def __init__(self, world: World, *components: _T):
|
|
"""
|
|
Créer une entité avec les composants en paramètres et l'ajoute au monde.
|
|
|
|
Paramètres:
|
|
world: Le monde auquel ajouter l'entité.
|
|
*components: Les composants de l'entité.
|
|
"""
|
|
self._world = world
|
|
self._to_apply: list[tuple[Type, Optional[object]]] = [
|
|
(type(component), component) for component in components
|
|
]
|
|
self._components: dict[Type[_T], _T] = {}
|
|
self._deleted = False
|
|
self._world._to_apply.add(self)
|
|
|
|
def set(self, *components: _T) -> None:
|
|
"""
|
|
Définit les composants de l'entité donnés en paramètres.
|
|
|
|
Paramètres:
|
|
*components: Les composants a definir.
|
|
"""
|
|
for component in components:
|
|
self._to_apply.append((type(component), component))
|
|
self._world._to_apply.add(self)
|
|
|
|
def remove(self, *component_types: Type[_T]) -> None:
|
|
"""
|
|
Supprime les composants de l'entité donnés en paramètres.
|
|
|
|
Paramètres:
|
|
*component_types: Le type des composants à supprimer.
|
|
"""
|
|
for component_type in component_types:
|
|
if component_type in self._components:
|
|
self._to_apply.append((component_type, None))
|
|
self._world._to_apply.add(self)
|
|
|
|
def __getitem__(self, component_type: Type[_T]) -> _T:
|
|
"""
|
|
Renvoie le composant de type *component_type.
|
|
|
|
Paramètres:
|
|
component_type: Le type du composant à récupérer.
|
|
|
|
Retourne:
|
|
Le composant de type *component_type.
|
|
"""
|
|
return self._components[component_type]
|
|
|
|
def __contains__(self, *component_types: Type[_T]) -> bool:
|
|
"""
|
|
Renvoie True si l'entité contient tous les composants de *component_types.
|
|
|
|
Paramètres:
|
|
component_type: Les types des composants à tester.
|
|
|
|
Retourne:
|
|
Si tous les composants de *component_types sont dans l'entité.
|
|
"""
|
|
for component_type in component_types:
|
|
if component_type not in self._components:
|
|
return False
|
|
return True
|
|
|
|
|
|
class Plugin:
|
|
"""
|
|
Un plugin qui ajoute des fonctionnalités a un jeu.
|
|
"""
|
|
|
|
@abstractmethod
|
|
def apply(self, game: "Game") -> None:
|
|
"""
|
|
Applique le plugin a un jeu.
|
|
|
|
Paramètres:
|
|
game: Le jeu auquel appliquer le plugin.
|
|
"""
|
|
pass
|
|
|
|
|
|
class Game:
|
|
"""
|
|
Permet de faire une simple boucle de jeu.
|
|
"""
|
|
|
|
def __init__(self, *plugins: Plugin) -> None:
|
|
"""
|
|
Créer une un jeu avec les plugins donnés en paramètres.
|
|
|
|
Paramètres:
|
|
*plugins: Les plugins a ajouter au jeu.
|
|
"""
|
|
self._running = False
|
|
self._pre_startup_tasks: list[Callable[[World], None]] = []
|
|
self._startup_tasks: list[Callable[[World], None]] = []
|
|
self._post_startup_tasks: list[Callable[[World], None]] = []
|
|
self._pre_update_tasks: list[Callable[[World], None]] = []
|
|
self._update_tasks: list[Callable[[World], None]] = []
|
|
self._post_update_tasks: list[Callable[[World], None]] = []
|
|
self._pre_render_tasks: list[Callable[[World], None]] = []
|
|
self._render_tasks: list[Callable[[World], None]] = []
|
|
self._post_render_tasks: list[Callable[[World], None]] = []
|
|
self._pre_shutdown_tasks: list[Callable[[World], None]] = []
|
|
self._shutdown_tasks: list[Callable[[World], None]] = []
|
|
self._post_shutdown_tasks: list[Callable[[World], None]] = []
|
|
for plugin in plugins:
|
|
plugin.apply(self)
|
|
|
|
def add_pre_startup_tasks(self, *tasks: Callable[[World], None]) -> None:
|
|
"""
|
|
Ajoute des taches qui s'executeront en premier avant le démarrage du jeu.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("Cannot add a task while the loop is running")
|
|
self._pre_startup_tasks.extend(tasks)
|
|
|
|
def add_startup_tasks(self, *tasks: Callable[[World], None]) -> None:
|
|
"""
|
|
Ajoute des taches qui s'executeront au démarrage du jeu.
|
|
|
|
Paramètres:
|
|
*tasks: Les taches à ajouter.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("Cannot add a task while the loop is running")
|
|
self._startup_tasks.extend(tasks)
|
|
|
|
def add_post_startup_tasks(self, *tasks: Callable[[World], None]) -> None:
|
|
"""
|
|
Ajoute des taches qui s'executeront en dernier après le démarrage du jeu.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("Cannot add a task while the loop is running")
|
|
self._post_startup_tasks.extend(tasks)
|
|
|
|
def add_pre_update_tasks(self, *tasks: Callable[[World], None]) -> None:
|
|
"""
|
|
Ajoute des taches qui s'executeront au debut de chaque mise à jour du jeu.
|
|
|
|
Paramètres:
|
|
*tasks: Les taches à ajouter.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("Cannot add a task while the loop is running")
|
|
self._pre_update_tasks.extend(tasks)
|
|
|
|
def add_update_tasks(self, *tasks: Callable[[World], None]) -> None:
|
|
"""
|
|
Ajoute des taches qui s'executeront a chaque mise à jour du jeu.
|
|
|
|
Paramètres:
|
|
*tasks: Les taches à ajouter.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("Cannot add a task while the loop is running")
|
|
self._update_tasks.extend(tasks)
|
|
|
|
def add_post_update_tasks(self, *tasks: Callable[[World], None]) -> None:
|
|
"""
|
|
Ajoute des taches qui s'executeront à la fin de chaque mise à jour du jeu.
|
|
|
|
Paramètres:
|
|
*tasks: Les taches à ajouter.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("Cannot add a task while the loop is running")
|
|
self._post_update_tasks.extend(tasks)
|
|
|
|
def add_pre_render_tasks(self, *tasks: Callable[[World], None]) -> None:
|
|
"""
|
|
Ajoute des taches qui s'executeront au début de chaque mise à jour du jeu pour le rendu.
|
|
|
|
Paramètres:
|
|
*tasks: Les taches à ajouter.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("Cannot add a task while the loop is running")
|
|
self._pre_render_tasks.extend(tasks)
|
|
|
|
def add_render_tasks(self, *tasks: Callable[[World], None]) -> None:
|
|
"""
|
|
Ajoute des taches qui s'executeront après chaque mise à jour du jeu pour le rendu.
|
|
|
|
Paramètres:
|
|
*tasks: Les taches à ajouter.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("Cannot add a task while the loop is running")
|
|
self._render_tasks.extend(tasks)
|
|
|
|
def add_post_render_tasks(self, *tasks: Callable[[World], None]) -> None:
|
|
"""
|
|
Ajoute des taches qui s'executeront après chaque mise à jour du jeu pour le rendu.
|
|
|
|
Paramètres:
|
|
*tasks: Les taches à ajouter.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("Cannot add a task while the loop is running")
|
|
self._post_render_tasks.extend(tasks)
|
|
|
|
def add_pre_shutdown_tasks(self, *tasks: Callable[[World], None]) -> None:
|
|
"""
|
|
Ajoute des taches qui s'executeront au début de la fin de la boucle de jeu.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("Cannot add a task while the loop is running")
|
|
self._pre_shutdown_tasks.extend(tasks)
|
|
|
|
def add_shutdown_tasks(self, *tasks: Callable[[World], None]) -> None:
|
|
"""
|
|
Ajoute des taches qui s'executeront à la fin de la boucle de jeu.
|
|
|
|
Paramètres:
|
|
*tasks: Les taches à ajouter.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("Cannot add a task while the loop is running")
|
|
self._shutdown_tasks.extend(tasks)
|
|
|
|
def add_post_shutdown_tasks(self, *tasks: Callable[[World], None]) -> None:
|
|
"""
|
|
Ajoute des taches qui s'executeront à la fin de la fin de la boucle de jeu.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("Cannot add a task while the loop is running")
|
|
self._post_shutdown_tasks.extend(tasks)
|
|
|
|
def run(self, world: World = World()) -> World:
|
|
"""
|
|
Lance la boucle de jeu.
|
|
"""
|
|
if self._running:
|
|
raise RuntimeError("The loop is already running")
|
|
self._running = True
|
|
|
|
# On initialize le monde
|
|
world.set(self)
|
|
world.apply()
|
|
|
|
# On execute les taches de pré initialisation du monde
|
|
for task in self._pre_startup_tasks:
|
|
try:
|
|
task(world)
|
|
except Exception as e:
|
|
error(f"Error during pre-startup task: {e}")
|
|
world.apply()
|
|
|
|
# On execute les taches d'initialisation du monde
|
|
for task in self._startup_tasks:
|
|
try:
|
|
task(world)
|
|
except Exception as e:
|
|
error(f"Error during startup task: {e}")
|
|
world.apply()
|
|
|
|
# On execute les taches de post initialisation du monde
|
|
for task in self._post_startup_tasks:
|
|
try:
|
|
task(world)
|
|
except Exception as e:
|
|
error(f"Error during post-startup task: {e}")
|
|
world.apply()
|
|
|
|
while self._running:
|
|
# On execute les taches de pré mise à jour du monde
|
|
for task in self._pre_update_tasks:
|
|
try:
|
|
task(world)
|
|
except Exception as e:
|
|
error(f"Error during pre-update task: {e}")
|
|
world.apply()
|
|
|
|
# On exécute les taches de mise a jour du monde
|
|
for task in self._update_tasks:
|
|
try:
|
|
task(world)
|
|
except Exception as e:
|
|
error(f"Error during update task: {e}")
|
|
world.apply()
|
|
|
|
# On execute les taches de post mise à jour du monde
|
|
for task in self._post_update_tasks:
|
|
try:
|
|
task(world)
|
|
except Exception as e:
|
|
error(f"Error during post-update task: {e}")
|
|
world.apply()
|
|
|
|
# On execute les taches de rendu du jeu
|
|
for task in self._render_tasks:
|
|
try:
|
|
task(world)
|
|
except Exception as e:
|
|
error(f"Error during render task: {e}")
|
|
world.apply()
|
|
|
|
# On execute les taches de fin de rendu du jeu
|
|
for task in self._post_render_tasks:
|
|
try:
|
|
task(world)
|
|
except Exception as e:
|
|
error(f"Error during post-render task: {e}")
|
|
world.apply()
|
|
|
|
# On execute les taches de pré fin de boucle
|
|
for task in self._pre_shutdown_tasks:
|
|
try:
|
|
task(world)
|
|
except Exception as e:
|
|
error(f"Error during pre-shutdown task: {e}")
|
|
world.apply()
|
|
|
|
# On exécute les taches de fin du monde
|
|
for task in self._shutdown_tasks:
|
|
try:
|
|
task(world)
|
|
except Exception as e:
|
|
error(f"Error during shutdown task: {e}")
|
|
world.apply()
|
|
|
|
# On execute les taches de post fin de boucle
|
|
for task in self._post_shutdown_tasks:
|
|
try:
|
|
task(world)
|
|
except Exception as e:
|
|
error(f"Error during post-shutdown task: {e}")
|
|
world.apply()
|
|
|
|
# On retourne le monde
|
|
return world
|
|
|
|
def stop(self) -> None:
|
|
"""
|
|
Demande la fin de la boucle de jeu. La boucle s'arretera a la prochaine mise à jour.
|
|
"""
|
|
self._running = False
|