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
|
import uuid
from typing import List, Optional
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,
UpstreamEventProcessor,
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(UpstreamEventProcessor):
def accept(
self, entity_manager: EntityManager, event: tuple[Event, str]
) -> Optional[Event]:
client_event, client_id = event
if client_event.event_type == EventType.ENTITY_POSITION_UPDATE:
return self._process_entity_position_update(
entity_manager, client_event, client_id
)
def _process_entity_position_update(
self,
entity_manager: EntityManager,
event: EntityPositionUpdateEvent,
client_id: str,
) -> Optional[Event]:
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)
return event
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]
|
