1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
from kennel.engine.game import Game
from kennel.engine.components.component import ComponentType
from kennel.engine.entities.entity import EntityManager, Entity
from kennel.engine.entities.laser import Laser
from kennel.engine.systems.system import SystemManager
from kennel.engine.systems.network import (
NetworkSystem,
EventProcessor,
Event,
EventType,
)
from kennel.engine.systems.world import WorldSystem
from kennel.config import config
from .app import logger
from typing import List
import uuid
entity_manager = EntityManager()
system_manager = SystemManager()
class KennelEventProcessor(EventProcessor):
def accept(
self, entity_manager: EntityManager, event: type[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: Event, 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:
logger.error(f"Entity {entity} is not controllable")
return
if 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
entity.add_component(position)
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]
|
