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
76
77
78
79
80
81
|
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,
ClientEventTransformer,
Event,
EventType,
)
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: Event) -> None:
if event.event_type == EventType.ENTITY_POSITION_UPDATE:
self._process_entity_position_update(entity_manager, event)
def _process_entity_position_update(
self, entity_manager: EntityManager, event: Event
) -> None:
entity = entity_manager.get_entity(event.data["entity_id"])
if entity is None:
logger.error(f"Entity {event.data['entity_id']} does not exist")
return
if event.data["client_id"] is None:
logger.error("Client ID is required for position updates")
return
controllable = entity.get_component(ComponentType.CONTROLLABLE)
if controllable is None:
logger.error(f"Entity {entity} is not controllable")
return
if controllable.by != event.data["client_id"]:
logger.error(
f"Entity {entity} is not controllable by client {event.data['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)
class KennelClientEventTransformer(ClientEventTransformer):
def apply(self, entity_manager: EntityManager, event: Event) -> Event:
if event.event_type == EventType.ENTITY_BORN:
id = event.data["id"]
entity = entity_manager.get_entity(id)
if entity is None:
logger.error(f"Entity {id} does not exist")
return event
for component_type in entity.components:
if component_type == ComponentType.CONTROLLABLE:
# Do not send controllable components to clients
continue
component = entity.get_component(component_type)
if component is not None:
event.data[component_type] = component.dict()
return event
system_manager.add_system(
NetworkSystem(EventProcessor(), KennelClientEventTransformer())
)
kennel = Game(entity_manager, system_manager, config.MIN_TIME_STEP)
def create_session_controllable_entities(session: str) -> List[Entity]:
laser = Laser(uuid.uuid4().hex, session)
return [laser]
|
