import uuid from typing import List from kennel.config import config from kennel.engine.components.component import ComponentType from kennel.engine.entities.entity import Entity, EntityManager from kennel.engine.entities.laser import Laser from kennel.engine.game import Game from kennel.engine.systems.network import ( EntityPositionUpdateEvent, Event, EventProcessor, EventType, NetworkSystem, ) from kennel.engine.systems.system import SystemManager from kennel.engine.systems.world import WorldSystem from .app import logger entity_manager = EntityManager() system_manager = SystemManager() class KennelEventProcessor(EventProcessor): def accept( self, entity_manager: EntityManager, event: Event | tuple[Event, str] ) -> None: if isinstance(event, tuple): client_event, client_id = event self._process_client_event(entity_manager, client_event, client_id) return def _process_client_event( self, entity_manager: EntityManager, event: Event, client_id: str ) -> None: if event.event_type == EventType.ENTITY_POSITION_UPDATE: self._process_entity_position_update(entity_manager, event, client_id) def _process_entity_position_update( self, entity_manager: EntityManager, event: EntityPositionUpdateEvent, client_id: str, ) -> None: entity = entity_manager.get_entity(event.data["id"]) if entity is None: logger.error(f"Entity(id={event.data['id']}) does not exist") return controllable = entity.get_component(ComponentType.CONTROLLABLE) if controllable is None or controllable.by != client_id: logger.error(f"Entity {entity} is not controllable by client {client_id}") return position = entity.get_component(ComponentType.POSITION) if position is None: logger.error(f"Entity {entity} has no position") return position.x = event.data["position"]["x"] position.y = event.data["position"]["y"] entity.add_component(position) entity_manager.add_entity(entity) system_manager.add_system(WorldSystem(config.WORLD_WIDTH, config.WORLD_HEIGHT)) system_manager.add_system( NetworkSystem(KennelEventProcessor()), ) kennel = Game(entity_manager, system_manager, config.MIN_TIME_STEP) def create_session_controllable_entities(session: str) -> List[Entity]: laser = Laser(uuid.uuid4().hex[:10], session) return [laser]