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
69
70
71
72
73
74
75
|
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]
|
