from collections.abc import Callable, Generator, Iterable
from dataclasses import dataclass
from pathlib import Path
-from typing import Any, cast, get_args, overload
+from typing import Any, Literal, cast, get_args, overload
import pygame.sprite
from pygame import Vector2
-from pacman.ecs.schedule import Schedule, ScheduleLabel, SystemSet
-
-
-class Entity:
- pass
-
-
-@dataclass
-class Resource[T]:
- storage: type[T]
-
-
-class SkipSystem(Exception):
- pass
-
-
-type System = Callable[["World"], None]
-
-type Plugin = Callable[["World"], None] | tuple[Plugin, ...]
-
-
-@dataclass
-class EntityThunk:
- world: "World"
- entity: Entity
-
- def __getitem__[T](self, ty: type[T]) -> T:
- return cast(T, self.world._components[ty][self.entity])
-
- def __setitem__[T](self, ty: type[T], val: T) -> None:
- if self.entity not in self.world._entities:
- self.world._entities[self.entity] = set()
- if ty not in self.world._entities[self.entity]:
- self.world._entities[self.entity].add(ty)
- if ty not in self.world._components:
- self.world._components[ty] = {}
- self.world._components[ty][self.entity] = val
-
- def __delitem__[T](self, ty: type[T]) -> None:
- if ty not in self.world._entities[self.entity]:
- return
- self.world._entities[self.entity].remove(ty)
- del self.world._components[ty][self.entity]
-
- def __contains__[T](self, ty: type[T]) -> bool:
- return ty in self.world._entities[self.entity]
-
-
-class World:
- def __init__(self) -> None:
- self._entities: dict[Entity, set[type]] = {}
- self._components: dict[type, dict[Entity, Any]] = {}
- self._resources: dict[type, Any] = {}
- self._schedules: dict[ScheduleLabel, Schedule[System]] = {}
-
- def entity(self, entity: Entity) -> EntityThunk:
- return EntityThunk(self, entity)
-
- def query[*T](self, ty: type[tuple[*T]]) -> Generator[tuple[*T]]:
- args = get_args(ty)
- if len(args) == 0:
- return
- sets: list[dict[Entity, Any]] = sorted(
- (
- self._entities if arg is Entity else self._components[arg]
- for arg in args
- if arg is Entity or arg in self._components
- ),
- key=len,
- reverse=True,
- )
- if len(sets) == 0:
- return
- for entity in sets.pop():
- if any(entity not in e for e in sets):
- continue
- yield cast(
- tuple[*T],
- tuple(
- entity if e is Entity else self._components[e][entity]
- for e in args
- ),
- )
-
- @overload
- def __getitem__[T](self, arg: Resource[T]) -> T: ...
- @overload
- def __getitem__[*T](
- self, arg: type[tuple[*T]]
- ) -> Generator[tuple[*T]]: ...
- @overload
- def __getitem__(self, arg: Entity) -> EntityThunk: ...
- def __getitem__(
- self,
- arg: Any,
- ) -> Any:
- if isinstance(arg, Entity):
- return self.entity(arg)
- if isinstance(arg, Resource):
- return self._resources[arg.storage]
- return self.query(arg)
-
- @overload
- def __setitem__[T](self, key: Resource[T], val: T) -> None: ...
- @overload
- def __setitem__[*T](self, key: Entity, val: tuple[*T]) -> None: ...
- def __setitem__(self, key: Any, val: Any) -> None:
- if isinstance(key, Entity) and isinstance(val, tuple):
- if key in self:
- del self[key]
- self._entities[key] = set()
- for component in val:
- self[key][type(component)] = component
- elif isinstance(key, Resource) and isinstance(val, key.storage):
- self._resources[key.storage] = val
- else:
- raise TypeError()
-
- @overload
- def __delitem__[T](self, key: Resource[T]) -> None: ...
- @overload
- def __delitem__(self, key: Entity) -> None: ...
- def __delitem__[T](self, key: Entity | Resource[T]) -> None:
- if isinstance(key, Entity):
- components = self._entities[key]
- del self._entities[key]
- for component in components:
- del self._components[component][key]
- else:
- del self._resources[key.storage]
-
- @overload
- def __contains__[T](self, item: Resource[T]) -> bool: ...
- @overload
- def __contains__(self, item: Entity) -> bool: ...
-
- def __contains__[T](self, item: Entity | Resource[T]) -> bool:
- if isinstance(item, Resource):
- return item.storage in self._resources
- return item in self._entities
-
- def _run_cond(self, _: System | SystemSet) -> bool:
- return True
-
- def tick(self, schedule: ScheduleLabel) -> None:
- if schedule not in self._schedules:
- return
- for system in self._schedules[schedule].traverse(self._run_cond):
- try:
- system(self)
- except SkipSystem:
- pass
-
- def res_s[T](self, ty: type[T]) -> T:
- if Resource(ty) not in self:
- raise SkipSystem()
- return self[Resource(ty)]
-
- def run(
- self,
- schedule: ScheduleLabel,
- stop_cond: Callable[["World"], bool] = lambda _: False,
- ) -> None:
- while not stop_cond(self):
- self.tick(schedule)
-
- def with_systems(
- self, schedule: ScheduleLabel, *systems: System
- ) -> "World":
- if schedule not in self._schedules:
- self._schedules[schedule] = Schedule()
- self._schedules[schedule].add_ordering(systems, [], [], [])
- return self
-
- def with_plugins(self, *plugins: Plugin) -> "World":
- for plugin in plugins:
- if isinstance(plugin, tuple):
- self.with_plugins(plugin)
- continue
- plugin(self)
- return self
-
- def run_main(self) -> None:
- self.run(
- MainSchedule, lambda world: Resource(WorldShouldExit) in world
- )
-
-
-MainSchedule = ScheduleLabel("main")
-StartupSchedule = ScheduleLabel("startup")
-UpdateSchedule = ScheduleLabel("update")
-
-
-class WorldIsInit:
- pass
-
-
-class WorldShouldExit:
- pass
+from pacman.ecs.hierarchy import hierachy_plugins
+from pacman.ecs.schedule import (
+ MainSchedule,
+ Schedule,
+ ScheduleLabel,
+ StartupSchedule,
+ SystemSet,
+ UpdateSchedule,
+ WorldIsInit,
+ WorldShouldExit,
+)
+from pacman.ecs.world import Entity, Resource, Systems, World
def run_minimal_subschedules(world: World) -> None:
world.tick(UpdateSchedule)
-def minimal_plugins(world: World) -> None:
+def schedule_plugins(world: World) -> None:
world.with_systems(MainSchedule, run_minimal_subschedules)
+def minimal_plugins(world: World) -> None:
+ world.with_plugins(schedule_plugins, hierachy_plugins)
+
+
@dataclass
class ScreenCoord2D:
vec: Vector2
def graphics_plugins(world: World) -> None:
world.with_systems(StartupSchedule, init_window).with_systems(
- UpdateSchedule, poll_events, close_on_close, render_sprite2d
+ UpdateSchedule,
+ Systems(poll_events, close_on_close, render_sprite2d).chain(),
)
--- /dev/null
+from collections.abc import Generator, Iterable
+
+from pacman.ecs.world import Entity, World
+
+
+class Children:
+ def __init__(self, children: Iterable[Entity]) -> None:
+ self._children = set(children)
+
+ @property
+ def children(self) -> Generator[Entity]:
+ for child in self._children:
+ yield child
+
+
+__no_hook = False
+
+
+def get_parents(world: World, entity: Entity) -> Generator[Entity]:
+ """Generate parents of entity including self"""
+ yield entity
+ while Parent in world[entity]:
+ entity: Entity = world[entity][Parent]._parent
+ yield entity
+
+
+def children_insert_hook(
+ world: World, entity: Entity, component: Children
+) -> None:
+ global __no_hook
+ if __no_hook:
+ return
+ invalid_children = set(get_parents(world, entity))
+ if any(child in invalid_children for child in component.children):
+ raise Exception("Cyclic child hierarchy")
+ for child in map(lambda e: world[e], component.children):
+ if Parent in child:
+ del child[Parent]
+ __no_hook = True
+ child[Parent] = Parent(entity)
+ __no_hook = False
+
+ pass
+
+
+def children_remove_hook(
+ world: World, _entity: Entity, component: Children
+) -> None:
+ global __no_hook
+ if __no_hook:
+ return
+ __no_hook = True
+ for child in component.children:
+ del world[child][Parent]
+ __no_hook = False
+
+
+class Parent:
+ def __init__(self, parent: Entity) -> None:
+ self._parent = parent
+
+ @property
+ def parent(self) -> Entity:
+ return self._parent
+
+
+def parent_insert_hook(
+ world: World, entity: Entity, component: Parent
+) -> None:
+ global __no_hook
+ if __no_hook:
+ return
+ parent = world[component.parent]
+ if Parent not in parent:
+ __no_hook = True
+ parent[Children] = Children((entity,))
+ __no_hook = False
+ else:
+ parent[Children]._children.add(entity)
+
+
+def parent_remove_hook(
+ world: World, entity: Entity, component: Parent
+) -> None:
+ global __no_hook
+ if __no_hook:
+ return
+ world[component.parent][Children]._children.remove(entity)
+
+
+def hierachy_plugins(world: World) -> None:
+ world.with_hook("insert", Children, children_insert_hook).with_hook(
+ "remove", Children, children_remove_hook
+ ).with_hook("insert", Parent, parent_insert_hook).with_hook(
+ "remove", Parent, parent_remove_hook
+ )
yield e
-if __name__ == "__main__":
- graph = Schedule[str]()
+MainSchedule = ScheduleLabel("main")
+StartupSchedule = ScheduleLabel("startup")
+UpdateSchedule = ScheduleLabel("update")
- group_a = SystemSet("Group A")
- group_b = SystemSet("Group B")
- graph.add_ordering(["a", "b", "c"], ["z"], ["w"], [group_a])
- graph.add_ordering(["q"], [group_a], [], [group_b])
- graph.add_ordering([group_b], [], ["w"], [])
- for e in graph.traverse(lambda _: True):
- print(e)
+class WorldIsInit:
+ pass
+
+
+class WorldShouldExit:
+ pass
--- /dev/null
+from collections.abc import Callable, Generator
+from dataclasses import dataclass
+from typing import Any, Literal, cast, get_args, overload
+
+from pacman.ecs.schedule import (
+ MainSchedule,
+ Schedule,
+ ScheduleLabel,
+ SystemSet,
+ WorldShouldExit,
+)
+
+
+class Entity:
+ pass
+
+
+@dataclass
+class Resource[T]:
+ storage: type[T]
+
+
+class SkipSystem(Exception):
+ pass
+
+
+type System = Callable[["World"], None]
+
+type Plugin = Callable[["World"], None] | tuple[Plugin, ...]
+
+type ComponentHook[T] = Callable[["World", "Entity", T], None]
+
+
+@dataclass
+class EntityThunk:
+ world: "World"
+ entity: Entity
+
+ def __getitem__[T](self, ty: type[T]) -> T:
+ return cast(T, self.world._components[ty][self.entity])
+
+ def __setitem__[T](self, ty: type[T], val: T) -> None:
+ if self.entity not in self.world._entities:
+ self.world._entities[self.entity] = set()
+ if ty not in self.world._entities[self.entity]:
+ self.world._entities[self.entity].add(ty)
+ if ty not in self.world._components:
+ self.world._components[ty] = {}
+ self.world._components[ty][self.entity] = val
+
+ def __delitem__[T](self, ty: type[T]) -> None:
+ if ty not in self.world._entities[self.entity]:
+ return
+ self.world._entities[self.entity].remove(ty)
+ del self.world._components[ty][self.entity]
+
+ def __contains__[T](self, ty: type[T]) -> bool:
+ return ty in self.world._entities[self.entity]
+
+
+class World:
+ def __init__(self) -> None:
+ self._entities: dict[Entity, set[type]] = {}
+ self._components: dict[type, dict[Entity, Any]] = {}
+ self._resources: dict[type, Any] = {}
+ self._schedules: dict[ScheduleLabel, Schedule[System]] = {}
+ self._insert_hooks: dict[type, ComponentHook[Any]] = {}
+ self._remove_hooks: dict[type, ComponentHook[Any]] = {}
+
+ def entity(self, entity: Entity) -> EntityThunk:
+ return EntityThunk(self, entity)
+
+ def query[*T, *U](
+ self, ty: type[tuple[*T]], without: type[tuple[*U]] | None = None
+ ) -> Generator[tuple[*T]]:
+ args = get_args(ty)
+ neg_args = get_args(without)
+ if len(args) == 0:
+ return
+ sets: list[dict[Entity, Any]] = sorted(
+ (
+ self._entities if arg is Entity else self._components[arg]
+ for arg in args
+ if arg is Entity or arg in self._components
+ ),
+ key=len,
+ reverse=True,
+ )
+ neg_sets: list[dict[Entity, Any]] = sorted(
+ (
+ self._entities if arg is Entity else self._components[arg]
+ for arg in neg_args
+ if arg is Entity or arg in self._components
+ ),
+ key=len,
+ reverse=True,
+ )
+ if len(sets) == 0:
+ return
+ for entity in sets.pop():
+ if any(entity not in e for e in sets) or any(
+ entity in e for e in neg_sets
+ ):
+ continue
+ yield cast(
+ tuple[*T],
+ tuple(
+ entity if e is Entity else self._components[e][entity]
+ for e in args
+ ),
+ )
+
+ @overload
+ def __getitem__[T](self, arg: Resource[T]) -> T: ...
+ @overload
+ def __getitem__[*T](
+ self, arg: type[tuple[*T]]
+ ) -> Generator[tuple[*T]]: ...
+ @overload
+ def __getitem__(self, arg: Entity) -> EntityThunk: ...
+ def __getitem__(
+ self,
+ arg: Any,
+ ) -> Any:
+ if isinstance(arg, Entity):
+ return self.entity(arg)
+ if isinstance(arg, Resource):
+ return self._resources[arg.storage]
+ return self.query(arg)
+
+ @overload
+ def __setitem__[T](self, key: Resource[T], val: T) -> None: ...
+ @overload
+ def __setitem__[*T](self, key: Entity, val: tuple[*T]) -> None: ...
+ def __setitem__(self, key: Any, val: Any) -> None:
+ if isinstance(key, Entity) and isinstance(val, tuple):
+ if key in self:
+ del self[key]
+ self._entities[key] = set()
+ for component in val:
+ self[key][type(component)] = component
+ elif isinstance(key, Resource) and isinstance(val, key.storage):
+ self._resources[key.storage] = val
+ else:
+ raise TypeError()
+
+ @overload
+ def __delitem__[T](self, key: Resource[T]) -> None: ...
+ @overload
+ def __delitem__(self, key: Entity) -> None: ...
+ def __delitem__[T](self, key: Entity | Resource[T]) -> None:
+ if isinstance(key, Entity):
+ components = self._entities[key]
+ del self._entities[key]
+ for component in components:
+ del self._components[component][key]
+ else:
+ del self._resources[key.storage]
+
+ @overload
+ def __contains__[T](self, item: Resource[T]) -> bool: ...
+ @overload
+ def __contains__(self, item: Entity) -> bool: ...
+
+ def __contains__[T](self, item: Entity | Resource[T]) -> bool:
+ if isinstance(item, Resource):
+ return item.storage in self._resources
+ return item in self._entities
+
+ def _run_cond(self, _: System | SystemSet) -> bool:
+ return True
+
+ def tick(self, schedule: ScheduleLabel) -> None:
+ if schedule not in self._schedules:
+ return
+ for system in self._schedules[schedule].traverse(self._run_cond):
+ try:
+ system(self)
+ except SkipSystem:
+ pass
+
+ def res_s[T](self, ty: type[T]) -> T:
+ if Resource(ty) not in self:
+ raise SkipSystem()
+ return self[Resource(ty)]
+
+ def run(
+ self,
+ schedule: ScheduleLabel,
+ stop_cond: Callable[["World"], bool] = lambda _: False,
+ ) -> None:
+ while not stop_cond(self):
+ self.tick(schedule)
+
+ def with_systems(
+ self, schedule: ScheduleLabel, *systems: "System | Systems"
+ ) -> "World":
+ if schedule not in self._schedules:
+ self._schedules[schedule] = Schedule()
+ Systems(*systems)._apply(self._schedules[schedule])
+ return self
+
+ def with_hook[T](
+ self,
+ cycle: Literal["insert", "remove"],
+ ty: type[T],
+ hook: ComponentHook[T],
+ ) -> "World":
+ match cycle:
+ case "insert":
+ d = self._insert_hooks
+ case "remove":
+ d = self._insert_hooks
+ if ty not in d:
+ d[ty] = hook
+ else:
+ old = d[ty]
+ d[ty] = lambda world, entity, component: (
+ old(world, entity, component),
+ hook(world, entity, component),
+ )[1]
+ return self
+
+ def with_plugins(self, *plugins: Plugin) -> "World":
+ for plugin in plugins:
+ if isinstance(plugin, tuple):
+ self.with_plugins(plugin)
+ continue
+ plugin(self)
+ return self
+
+ def run_main(self) -> None:
+ self.run(
+ MainSchedule, lambda world: Resource(WorldShouldExit) in world
+ )
+
+
+_order_groups: int = 0
+
+
+class Systems:
+ def __init__(self, *systems: "System | SystemSet | Systems") -> None:
+ self._systems = systems
+ self._chain = False
+ self._pre: list[System | SystemSet] = []
+ self._post: list[System | SystemSet] = []
+ self._in_set: list[SystemSet] = []
+
+ def chain(self) -> "Systems":
+ self._chain = True
+ return self
+
+ def after(self, *systems: System | SystemSet) -> "Systems":
+ self._pre.extend(systems)
+ return self
+
+ def before(self, *systems: System | SystemSet) -> "Systems":
+ self._post.extend(systems)
+ return self
+
+ def in_set(self, *sets: SystemSet) -> "Systems":
+ self._in_set.extend(sets)
+ return self
+
+ def _apply(self, sched: Schedule[System]) -> SystemSet:
+ global _order_groups
+ _order_groups += 1
+ label = SystemSet(f"order-group-{_order_groups}")
+ systems = (
+ e._apply(sched) if isinstance(e, Systems) else e
+ for e in self._systems
+ )
+ sched.add_ordering((label,), self._pre, self._post, self._in_set)
+ if self._chain:
+ prev = ()
+ for system in systems:
+ sched.add_ordering((system,), prev, (), (label,))
+ prev = (system,)
+ else:
+ sched.add_ordering(systems, (), (), (label,))
+ return label
projects / axy / ft / pacman.git / commitdiff