Skip to content

Description

Control Panel is used as an entry point to ECS storages and runner.

API

ControlPanel

Source code in encosy/controlpanel.py 
 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
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
class ControlPanel:
    def __init__(
        self,
        system_storage: SystemStorageMeta | None = None,
        entity_storage: EntityStorageMeta | None = None,
        resource_storage: ResourceStorageMeta | None = None,
        runner: RunnerMeta | None = None,
    ):
        """
        ECS control panel

        Each storage unit represents and fulfill the purpose for distributing
        storing and querying items that associate with it

        Args:
            system_storage: Object that implements SystemStorageMeta
            entity_storage: Object that implements ResourceStorageMeta
            resource_storage: Object that implements ResourceStorageMeta
            runner: Object that implements RunnerMeta
        """

        if system_storage is None:
            system_storage = DefaultSystemStorage()
        if entity_storage is None:
            entity_storage = DefaultEntityStorage()
        if resource_storage is None:
            resource_storage = DefaultResourceStorage()
        if runner is None:
            runner = DefaultRunner()

        self.system_storage = system_storage
        self.entity_storage = entity_storage
        self.resource_storage = resource_storage
        self.runner = runner

        self._systems_to_drop: dict[Callable[[Any], Any], None] = {}
        self._systems_stop: dict[Callable[[Any], Any], None] = {}
        self._stop = False

        self._commands = Commands(self)

    def register_systems(
        self, *systems: Callable[[Any], Any]
    ) -> "ControlPanel":
        """
        Register any Callable.
        Input params of function can contain:
            - commands: Commands (name and type is reserved)
            - resource: Any - same as entity, but only one can exist
            - entity: Entity[ComponentType1, ComponentType2, ...]

        Args:
            *systems: Callable that contain definition or typing for input
                [Commands, Entities[ ListOfComponents ], Any],
                Any is considered as resource

        Returns:
            self
        """
        for system in systems:
            self.system_storage.add(system)
        return self

    def register_resources(self, *resources: Any) -> "ControlPanel":
        """
        Register resources. Each resource should be unique type
        as it is accessed through it

        Args:
            *resources: Object of any type

        Returns:
            self
        """
        for resource in resources:
            self.resource_storage.add(resource)
        return self

    def register_entities(self, *entities: Any) -> "ControlPanel":
        """
        Register entity. Each entity assigned a unique integer

        Args:
            *entities: Entity of any type that suits entity_storage.add
                by default the type is 'Entity'

        Returns:
            self

        """
        for entity in entities:
            self.entity_storage.add(entity)
        return self

    def register_plugins(
        self, *plugins: Callable[["ControlPanel"], Any]
    ) -> "ControlPanel":
        """
        Register plugins. Plugin is a simple function that takes ControlPanel

        Args:
            *plugins: simple function to that takes ControlPanel and apply
                onto self immediately

        Returns:
            self

        """
        for plugin in plugins:
            plugin(self)
        return self

    def _remove_system(self, *systems: Callable[[Any], Any]) -> "ControlPanel":
        """
        Drop given systems

        Args:
            *systems: same callable as in system register

        Returns:
            self

        """
        for system in systems:
            self.system_storage.remove(system)
        return self

    def remove_entities(self, *component_types: type) -> "ControlPanel":
        """
        Drop entities based on its components types

        Args:
            *component_types: remove entities by type of components

        Returns:
            self

        """
        entities = self.entity_storage.get(*component_types)
        for entity in entities:
            self.entity_storage.remove(entity)
        return self

    def drop_entities_with_expression(
        self, expression: Callable[[Entity], bool]
    ) -> "ControlPanel":
        """
        Drops entities based on expression of type (entity: Entity) -> bool
        Ex:
            - lambda entity: Entity[Name] == "MyName"

        Args:
            expression: callable that takes entity as input and return if it is
                True or False

        Returns:
            self

        """
        entities = self.entity_storage.query_expression(expression)
        for entity in entities:
            self.entity_storage.remove(entity)
        return self

    def stop_systems(self, *systems: Callable[[Any], Any]) -> "ControlPanel":
        """
        Add systems to stop dictionary

        Args:
            *systems: used as hash to stop system

        Returns:
            self

        """
        for system in systems:
            self._systems_stop[system] = None
        return self

    def start_systems(self, *systems: Callable[[Any], Any]) -> "ControlPanel":
        """
        Remove systems from stop dictionary

        Args:
            *systems: used as hash to resume systems

        Returns:
            self

        """
        for system in systems:
            if system in self._systems_stop:
                self._systems_stop.pop(system)
        return self

    def schedule_drop_systems(
        self, *systems: Callable[[Any], Any]
    ) -> "ControlPanel":
        """
        Schedules drop of a given systems
        Add system to drop queue and call _run_scheduled_drop_systems
        at the end of a tick

        Args:
            *systems: used as hash to drop system from storage

        Returns:
            self

        """
        for system in systems:
            self._systems_to_drop[system] = None
        return self

    def _run_scheduled_drop_systems(self) -> "ControlPanel":
        """
        Runs drop on a system drop queue and clear it

        Returns:
            self

        """
        self._remove_system(*self._systems_to_drop.keys())
        self._systems_to_drop = {}
        return self

    def pause(self) -> "ControlPanel":
        """
        Set a pause for a tick function (stop = True). Stop works as a gate
        in tick function

        Returns:
            self

        """
        self._stop = True
        return self

    def resume(self) -> "ControlPanel":
        """
        Set a resume for a tick function (stop = False).
        stop works as a gate in tick function

        Returns:
            self

        """
        self._stop = False
        return self

    def _extract_system_input(
        self, system_config: SystemConfig
    ) -> dict[str, Any]:
        """
        Extracts input values for given system and returns basic kwargs
        If any of the resources does not exist or isn't registered - KeyError

        Args:
            system_config: to connects storages with systems using defined class

        Returns:
            key word arguments or kwargs
        """
        key_word_arguments: dict[str, Any] = {}
        system_config = self.system_storage.get(system_config.callable)
        for command, name in system_config.commands.items():
            key_word_arguments[name] = self._commands
        for resource, name in system_config.resources.items():
            key_word_arguments[name] = self.resource_storage.get(resource)
        for component_types, name in system_config.components.items():
            key_word_arguments[name] = self.entity_storage.get(
                *component_types
            )
        return key_word_arguments

    def _function_generator(self) -> Generator:
        """
        Generator for runner

        Returns:
            generator of functions

        """
        for system_config in self.system_storage.get_all():
            if system_config.callable in self._systems_stop:
                continue
            yield lambda: system_config.callable(
                **self._extract_system_input(system_config)
            )

    def tick(self) -> bool:
        """
        By default, equivalent to one step where each system
        in a given order (register_system(1->2->3->...) executes
        with a requested params [Any, Commands, Entities(list[tuple[Any]]])
        If requested resource does not exist then KeyError raised

        Returns:
            False if stopped and True if not

        """
        if self._stop:
            return False
        self.runner.run(self._function_generator())
        self._run_scheduled_drop_systems()
        return True

    def run(self) -> "ControlPanel":
        """
        Simple loop for executing ticks until commands.pause()
        called within any system then after current tick new won't start

        Returns:
            self

        """
        self.resume()
        while self.tick():
            pass
        return self

    def __repr__(self):
        return "Control Panel"

__init__(system_storage=None, entity_storage=None, resource_storage=None, runner=None)

ECS control panel

Each storage unit represents and fulfill the purpose for distributing storing and querying items that associate with it

Parameters:

Name Type Description Default
system_storage SystemStorageMeta | None

Object that implements SystemStorageMeta

 None
entity_storage EntityStorageMeta | None

Object that implements ResourceStorageMeta

 None
resource_storage ResourceStorageMeta | None

Object that implements ResourceStorageMeta

 None
runner RunnerMeta | None

Object that implements RunnerMeta

 None
Source code in encosy/controlpanel.py 
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
def __init__(
    self,
    system_storage: SystemStorageMeta | None = None,
    entity_storage: EntityStorageMeta | None = None,
    resource_storage: ResourceStorageMeta | None = None,
    runner: RunnerMeta | None = None,
):
    """
    ECS control panel

    Each storage unit represents and fulfill the purpose for distributing
    storing and querying items that associate with it

    Args:
        system_storage: Object that implements SystemStorageMeta
        entity_storage: Object that implements ResourceStorageMeta
        resource_storage: Object that implements ResourceStorageMeta
        runner: Object that implements RunnerMeta
    """

    if system_storage is None:
        system_storage = DefaultSystemStorage()
    if entity_storage is None:
        entity_storage = DefaultEntityStorage()
    if resource_storage is None:
        resource_storage = DefaultResourceStorage()
    if runner is None:
        runner = DefaultRunner()

    self.system_storage = system_storage
    self.entity_storage = entity_storage
    self.resource_storage = resource_storage
    self.runner = runner

    self._systems_to_drop: dict[Callable[[Any], Any], None] = {}
    self._systems_stop: dict[Callable[[Any], Any], None] = {}
    self._stop = False

    self._commands = Commands(self)

drop_entities_with_expression(expression)

Drops entities based on expression of type (entity: Entity) -> bool

Ex
  • lambda entity: Entity[Name] == "MyName"

Parameters:

Name Type Description Default
expression Callable[[Entity], bool]

callable that takes entity as input and return if it is True or False

 required

Returns:

Type Description
ControlPanel

self
Source code in encosy/controlpanel.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def drop_entities_with_expression(
    self, expression: Callable[[Entity], bool]
) -> "ControlPanel":
    """
    Drops entities based on expression of type (entity: Entity) -> bool
    Ex:
        - lambda entity: Entity[Name] == "MyName"

    Args:
        expression: callable that takes entity as input and return if it is
            True or False

    Returns:
        self

    """
    entities = self.entity_storage.query_expression(expression)
    for entity in entities:
        self.entity_storage.remove(entity)
    return self

pause()

Set a pause for a tick function (stop = True). Stop works as a gate in tick function

Returns:

Type Description
ControlPanel

self
Source code in encosy/controlpanel.py 
241
242
243
244
245
246
247
248
249
250
251
def pause(self) -> "ControlPanel":
    """
    Set a pause for a tick function (stop = True). Stop works as a gate
    in tick function

    Returns:
        self

    """
    self._stop = True
    return self

register_entities(*entities)

Register entity. Each entity assigned a unique integer

Parameters:

Name Type Description Default
*entities Any

Entity of any type that suits entity_storage.add by default the type is 'Entity'

 ()

Returns:

Type Description
ControlPanel

self
Source code in encosy/controlpanel.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def register_entities(self, *entities: Any) -> "ControlPanel":
    """
    Register entity. Each entity assigned a unique integer

    Args:
        *entities: Entity of any type that suits entity_storage.add
            by default the type is 'Entity'

    Returns:
        self

    """
    for entity in entities:
        self.entity_storage.add(entity)
    return self

register_plugins(*plugins)

Register plugins. Plugin is a simple function that takes ControlPanel

Parameters:

Name Type Description Default
*plugins Callable[[ControlPanel], Any]

simple function to that takes ControlPanel and apply onto self immediately

 ()

Returns:

Type Description
ControlPanel

self
Source code in encosy/controlpanel.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def register_plugins(
    self, *plugins: Callable[["ControlPanel"], Any]
) -> "ControlPanel":
    """
    Register plugins. Plugin is a simple function that takes ControlPanel

    Args:
        *plugins: simple function to that takes ControlPanel and apply
            onto self immediately

    Returns:
        self

    """
    for plugin in plugins:
        plugin(self)
    return self

register_resources(*resources)

Register resources. Each resource should be unique type as it is accessed through it

Parameters:

Name Type Description Default
*resources Any

Object of any type

 ()

Returns:

Type Description
ControlPanel

self
Source code in encosy/controlpanel.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def register_resources(self, *resources: Any) -> "ControlPanel":
    """
    Register resources. Each resource should be unique type
    as it is accessed through it

    Args:
        *resources: Object of any type

    Returns:
        self
    """
    for resource in resources:
        self.resource_storage.add(resource)
    return self

register_systems(*systems)

Register any Callable.

Input params of function can contain
  • commands: Commands (name and type is reserved)
  • resource: Any - same as entity, but only one can exist
  • entity: Entity[ComponentType1, ComponentType2, ...]

Parameters:

Name Type Description Default
*systems Callable[[Any], Any]

Callable that contain definition or typing for input [Commands, Entities[ ListOfComponents ], Any], Any is considered as resource

 ()

Returns:

Type Description
ControlPanel

self
Source code in encosy/controlpanel.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def register_systems(
    self, *systems: Callable[[Any], Any]
) -> "ControlPanel":
    """
    Register any Callable.
    Input params of function can contain:
        - commands: Commands (name and type is reserved)
        - resource: Any - same as entity, but only one can exist
        - entity: Entity[ComponentType1, ComponentType2, ...]

    Args:
        *systems: Callable that contain definition or typing for input
            [Commands, Entities[ ListOfComponents ], Any],
            Any is considered as resource

    Returns:
        self
    """
    for system in systems:
        self.system_storage.add(system)
    return self

remove_entities(*component_types)

Drop entities based on its components types

Parameters:

Name Type Description Default
*component_types type

remove entities by type of components

 ()

Returns:

Type Description
ControlPanel

self
Source code in encosy/controlpanel.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def remove_entities(self, *component_types: type) -> "ControlPanel":
    """
    Drop entities based on its components types

    Args:
        *component_types: remove entities by type of components

    Returns:
        self

    """
    entities = self.entity_storage.get(*component_types)
    for entity in entities:
        self.entity_storage.remove(entity)
    return self

resume()

Set a resume for a tick function (stop = False). stop works as a gate in tick function

Returns:

Type Description
ControlPanel

self
Source code in encosy/controlpanel.py 
253
254
255
256
257
258
259
260
261
262
263
def resume(self) -> "ControlPanel":
    """
    Set a resume for a tick function (stop = False).
    stop works as a gate in tick function

    Returns:
        self

    """
    self._stop = False
    return self

run()

Simple loop for executing ticks until commands.pause() called within any system then after current tick new won't start

Returns:

Type Description
ControlPanel

self
Source code in encosy/controlpanel.py 
322
323
324
325
326
327
328
329
330
331
332
333
334
def run(self) -> "ControlPanel":
    """
    Simple loop for executing ticks until commands.pause()
    called within any system then after current tick new won't start

    Returns:
        self

    """
    self.resume()
    while self.tick():
        pass
    return self

schedule_drop_systems(*systems)

Schedules drop of a given systems Add system to drop queue and call _run_scheduled_drop_systems at the end of a tick

Parameters:

Name Type Description Default
*systems Callable[[Any], Any]

used as hash to drop system from storage

 ()

Returns:

Type Description
ControlPanel

self
Source code in encosy/controlpanel.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
def schedule_drop_systems(
    self, *systems: Callable[[Any], Any]
) -> "ControlPanel":
    """
    Schedules drop of a given systems
    Add system to drop queue and call _run_scheduled_drop_systems
    at the end of a tick

    Args:
        *systems: used as hash to drop system from storage

    Returns:
        self

    """
    for system in systems:
        self._systems_to_drop[system] = None
    return self

start_systems(*systems)

Remove systems from stop dictionary

Parameters:

Name Type Description Default
*systems Callable[[Any], Any]

used as hash to resume systems

 ()

Returns:

Type Description
ControlPanel

self
Source code in encosy/controlpanel.py 
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def start_systems(self, *systems: Callable[[Any], Any]) -> "ControlPanel":
    """
    Remove systems from stop dictionary

    Args:
        *systems: used as hash to resume systems

    Returns:
        self

    """
    for system in systems:
        if system in self._systems_stop:
            self._systems_stop.pop(system)
    return self

stop_systems(*systems)

Add systems to stop dictionary

Parameters:

Name Type Description Default
*systems Callable[[Any], Any]

used as hash to stop system

 ()

Returns:

Type Description
ControlPanel

self
Source code in encosy/controlpanel.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def stop_systems(self, *systems: Callable[[Any], Any]) -> "ControlPanel":
    """
    Add systems to stop dictionary

    Args:
        *systems: used as hash to stop system

    Returns:
        self

    """
    for system in systems:
        self._systems_stop[system] = None
    return self

tick()

By default, equivalent to one step where each system in a given order (register_system(1->2->3->...) executes with a requested params [Any, Commands, Entities(list[tuple[Any]]]) If requested resource does not exist then KeyError raised

Returns:

Type Description
bool

False if stopped and True if not
Source code in encosy/controlpanel.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
def tick(self) -> bool:
    """
    By default, equivalent to one step where each system
    in a given order (register_system(1->2->3->...) executes
    with a requested params [Any, Commands, Entities(list[tuple[Any]]])
    If requested resource does not exist then KeyError raised

    Returns:
        False if stopped and True if not

    """
    if self._stop:
        return False
    self.runner.run(self._function_generator())
    self._run_scheduled_drop_systems()
    return True