Source code for pyecs.managers.EntityManager

import threading
from random import randbytes
from typing import Literal

from pyecs.common.Types import UUID4, Entity
from pyecs.helpers.Statuses import StatusCodes


[docs] class EntityManager(object):
[docs] def __init__(self): self.alive_entities: set[UUID4] = set() self._lock: threading.Lock = threading.Lock()
def _unique_id(self) -> UUID4: """ Generate a Version 4 UUID according to RFC 9562 specification. UUIDv4 uses 122 random bits with 6 reserved bits for version and variant: - Byte 7 (time_hi_and_version): Sets version to 4 by clearing upper 4 bits and setting bit 6 (0x40) through (byte & 0x0F) | 0x40 - Byte 9 (clock_seq_hi_and_reserved): Sets variant to RFC 9562 by clearing upper 2 bits and setting bit 7 (0x80) through (byte & 0x3F) | 0x80 Returns a canonical UUID string in 8-4-4-4-12 format. Reference: https://digitalbunker.dev/understanding-how-uuids-are-generated/ """ rb: bytearray = bytearray(randbytes(16)) rb[7] = (rb[7] & 0x0F) | 0x40 rb[9] = (rb[9] & 0x3F) | 0x80 return "-".join( [rb[:4].hex(), rb[4:6].hex(), rb[6:8].hex(), rb[8:10].hex(), rb[10:16].hex()] )
[docs] def create_entity( self, ) -> tuple[Literal[StatusCodes.ENTITY_CREATED], Entity] | Literal[StatusCodes.FAILURE]: """ Create a new entity in the entity component system. This method generates a unique UUID4 identifier for the entity and registers it in the alive_entities set. Returns a tuple containing ENTITY_CREATED status and the new entity UUID on success. FAILURE is typed for future expansion purposes only; currently cannot be returned. """ with self._lock: new_entity: UUID4 = self._unique_id() self.alive_entities.add(new_entity) return (StatusCodes.ENTITY_CREATED, new_entity)
[docs] def destroy_entity( self, entity: Entity ) -> Literal[StatusCodes.ENTITY_DESTROYED, StatusCodes.FAILURE]: """ Remove an entity from the entity component system. This method removes the specified entity UUID from the alive_entities set and triggers cleanup of all associated components. The entity becomes invalid after this operation and should not be used in subsequent operations. Returns ENTITY_DESTROYED status on successful removal, or FAILURE status if the entity does not exist or destruction fails. """ with self._lock: if entity not in self.alive_entities: return StatusCodes.FAILURE self.alive_entities.remove(entity) return StatusCodes.ENTITY_DESTROYED
[docs] def is_alive(self, entity: Entity) -> bool: """ Check if an entity is currently alive in the system. Returns True if the entity exists in the alive_entities set, False otherwise. """ with self._lock: return entity in self.alive_entities