import pygame.sprite
from pygame import Vector2
-
-class Entity:
- pass
+from pacman.ecs.schedule import Schedule, ScheduleLabel, SystemSet
-class Schedule:
+class Entity:
pass
self._entities: dict[Entity, set[type]] = {}
self._components: dict[type, dict[Entity, Any]] = {}
self._resources: dict[type, Any] = {}
- self._schedules: dict[Schedule, set[System]] = {}
+ self._schedules: dict[ScheduleLabel, Schedule[System]] = {}
def entity(self, entity: Entity) -> EntityThunk:
return EntityThunk(self, entity)
return item.storage in self._resources
return item in self._entities
- def tick(self, schedule: Schedule) -> None:
+ 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]:
+ for system in self._schedules[schedule].traverse(self._run_cond):
try:
system(self)
except SkipSystem:
def run(
self,
- schedule: Schedule,
+ schedule: ScheduleLabel,
stop_cond: Callable[["World"], bool] = lambda _: False,
) -> None:
while not stop_cond(self):
self.tick(schedule)
- def with_systems(self, schedule: Schedule, *systems: System) -> "World":
+ def with_systems(
+ self, schedule: ScheduleLabel, *systems: System
+ ) -> "World":
if schedule not in self._schedules:
- self._schedules[schedule] = set()
- self._schedules[schedule].update(systems)
+ self._schedules[schedule] = Schedule()
+ self._schedules[schedule].add_ordering(systems, [], [], [])
return self
def with_plugins(self, *plugins: Plugin) -> "World":
)
-MainSchedule = Schedule()
-StartupSchedule = Schedule()
-UpdateSchedule = Schedule()
+MainSchedule = ScheduleLabel("main")
+StartupSchedule = ScheduleLabel("startup")
+UpdateSchedule = ScheduleLabel("update")
class WorldIsInit:
coord.vec.y = 500.0 + math.sin(pos) * 30.0
-world = (
- World()
- .with_plugins(minimal_plugins, graphics_plugins)
- .with_systems(UpdateSchedule, move_sprites)
-)
-assets = list(Path("./asset").rglob("*.png"))
-for i, asset in enumerate(assets):
- world[Entity()] = (
- ScreenCoord2D(Vector2(30.0 * i, 500.0), 1.0 * i),
- Sprite2D(pygame.image.load(asset)),
- ShouldRender(),
+if __name__ == "__main__":
+ world = (
+ World()
+ .with_plugins(minimal_plugins, graphics_plugins)
+ .with_systems(UpdateSchedule, move_sprites)
)
-world.run_main()
+ assets = list(Path("./asset").rglob("*.png"))
+ for i, asset in enumerate(assets):
+ world[Entity()] = (
+ ScreenCoord2D(Vector2(30.0 * i, 500.0), 1.0 * i),
+ Sprite2D(pygame.image.load(asset)),
+ ShouldRender(),
+ )
+ world.run_main()
--- /dev/null
+from collections import deque
+from collections.abc import Callable, Generator, Iterable
+from dataclasses import dataclass
+from typing import override
+
+
+@dataclass
+class DAGNode[T]:
+ parents: set[T]
+ children: set[T]
+
+ def is_root(self) -> bool:
+ return len(self.parents) == 0
+
+
+class DAG[T]:
+ def __init__(self) -> None:
+ self._nodes: dict[T, DAGNode[T]] = {}
+
+ def _get_node(self, node: T) -> DAGNode[T]:
+ if node not in self._nodes:
+ self._nodes[node] = DAGNode(set(), set())
+ return self._nodes[node]
+
+ def add_edge(self, parent: T, child: T) -> None:
+ self._get_node(parent).children.add(child)
+ self._get_node(child).parents.add(parent)
+
+ def traverse_from(self, root: T) -> Generator[T]:
+ visited: set[T] = set()
+ queue = deque([root])
+ while len(queue) != 0:
+ curr_key = queue.popleft()
+ if curr_key in visited:
+ continue
+ curr = self._nodes[curr_key]
+ if not all(
+ e in visited or self._nodes[e].is_root() for e in curr.parents
+ ):
+ continue
+ yield curr_key
+ visited.add(curr_key)
+ queue.extend(e for e in curr.children if e not in visited)
+
+
+class DAGLabel:
+ def __init__(self, name: str) -> None:
+ self.name = name
+
+ @override
+ def __repr__(self) -> str:
+ return self.name
+
+
+class ScheduleLabel:
+ def __init__(self, name: str) -> None:
+ self.name = name
+
+ @override
+ def __repr__(self) -> str:
+ return self.name
+
+
+class SystemSet:
+ def __init__(self, name: str) -> None:
+ self.name = name
+ self._pre = DAGLabel(name + "-pre")
+ self._post = DAGLabel(name + "-post")
+
+
+class Schedule[T]:
+ def __init__(self) -> None:
+ self._dag: DAG[T | DAGLabel | None] = DAG()
+ self._group_appartenance: dict[T | SystemSet, set[SystemSet]] = {}
+
+ def add_ordering(
+ self,
+ items: Iterable[T | SystemSet],
+ pre: Iterable[T | SystemSet],
+ post: Iterable[T | SystemSet],
+ group: Iterable[SystemSet],
+ ):
+ def get_pre(e: T | SystemSet) -> T | DAGLabel:
+ if isinstance(e, SystemSet):
+ return e._pre
+ return e
+
+ def get_post(e: T | SystemSet) -> T | DAGLabel:
+ if isinstance(e, SystemSet):
+ return e._post
+ return e
+
+ for item in items:
+ self._dag.add_edge(None, get_pre(item))
+ if item not in self._group_appartenance:
+ self._group_appartenance[item] = set()
+ for e in pre:
+ self._dag.add_edge(get_post(e), get_pre(item))
+ for e in post:
+ self._dag.add_edge(get_post(item), get_pre(e))
+ for e in group:
+ self._dag.add_edge(get_pre(e), get_pre(item))
+ self._dag.add_edge(get_post(item), get_post(e))
+ self._group_appartenance[item].add(e)
+
+ def traverse(
+ self, run_cond: Callable[[T | SystemSet], bool]
+ ) -> Generator[T]:
+ cond_cache: dict[T | SystemSet, bool] = {}
+
+ def should_run(val: T | SystemSet) -> bool:
+ if val in cond_cache:
+ return cond_cache[val]
+ res = all(
+ should_run(parent)
+ for parent in self._group_appartenance.get(val, set())
+ ) and run_cond(val)
+ cond_cache[val] = res
+ return res
+
+ for e in self._dag.traverse_from(None):
+ if isinstance(e, DAGLabel) or e is None:
+ continue
+ if e not in self._group_appartenance:
+ continue
+ if not should_run(e):
+ continue
+ yield e
+
+
+if __name__ == "__main__":
+ graph = Schedule[str]()
+
+ 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)
projects / axy / ft / pacman.git / commitdiff