""" Système de gestion d'entités, composants et ressources. """ 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 Game: """ Permet de faire une simple boucle de jeu. """ def __init__(self) -> None: """ Créer une un jeu. """ self._running = False self._startup_tasks: dict[int, list[Callable[[World], None]]] = {} self._update_tasks: dict[int, list[Callable[[World], None]]] = {} self._shutdown_tasks: dict[int, list[Callable[[World], None]]] = {} def add_startup_tasks(self, priority: int, *tasks: Callable[[World], None]) -> None: """ Ajoute des taches qui s'executeront au démarrage du jeu. Paramètres: priority: La priorité de la tache. *tasks: Les taches à ajouter. """ if self._running: raise RuntimeError("Cannot add startup task while the loop is running") self._startup_tasks.setdefault(priority, []).extend(tasks) def add_tasks(self, priority: int, *tasks: Callable[[World], None]) -> None: """ Ajoute des taches qui s'executeront a chaque mise à jour du jeu. Paramètres: priority: La priorité de la tache. *tasks: Les taches à ajouter. """ if self._running: raise RuntimeError("Cannot add task while the loop is running") self._update_tasks.setdefault(priority, []).extend(tasks) def add_shutdown_tasks( self, priority: int, *tasks: Callable[[World], None] ) -> None: """ Ajoute des taches qui s'executeront à la fin de la boucle de jeu. Paramètres: priority: La priorité de la tache. *tasks: Les taches à ajouter. """ if self._running: raise RuntimeError("Cannot add shutdown task while the loop is running") self._shutdown_tasks.setdefault(priority, []).extend(tasks) def _run_tasks( self, tasks: dict[int, list[Callable[[World], None]]], world: World ) -> None: """ Execute toutes les taches donnes en paramètres en respectant la priorité. """ priorities = sorted(list(tasks.keys())) for priority in priorities: for task in tasks[priority]: try: task(world) except Exception as e: error(f"Error in task: {e}") def run(self) -> World: """ Lance la boucle de jeu. """ if self._running: raise RuntimeError("The loop is already running") self._running = True # On initialize le monde world: World = World() world.set(self) # On applique les moddifications pour l'ajout de la ressource world.apply() # On execute les taches d'initialisation du monde self._run_tasks(self._startup_tasks, world) # On applique les changements world.apply() while self._running: # On exécute les taches de mise a jour du monde self._run_tasks(self._update_tasks, world) # On applique les changements world.apply() # On exécute les taches de fin du monde self._run_tasks(self._shutdown_tasks, world) # On applique les changements 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