[tool.ruff]
line-length = 79
-#lint.select = ["E", "F", "UP", "B", "SIM", "I", "ARG", "N", "D", "ANN", "PYI"]
-lint.select = ["E", "F", "UP", "B", "SIM", "I", "ARG", "N", "ANN", "PYI"]
+lint.select = ["E", "F", "UP", "B", "SIM", "I", "ARG", "N", "ANN", "PYI", "D"]
[tool.ruff.lint.pydocstyle]
convention = "google"
def main() -> None:
+ run_game()
+ return None
try:
run_game()
except Exception as e:
-from pathlib import Path
+"""The core ECS engine."""
-import pygame
-
-from pacman.ecs.assets import Assets
-from pacman.ecs.coord import (
- Coord2D,
- CoordAbs2D,
- coord_update_plugins,
-)
-from pacman.ecs.graphics import (
- ShouldRender,
- Sprite2D,
- WindowEvents,
- graphics_plugins,
-)
-from pacman.ecs.hierarchy import (
- ChildOf,
- hierachy_plugins,
-)
-from pacman.ecs.schedule import (
- StartupSchedule,
- UpdateSchedule,
-)
+from pacman.ecs.coord import coord_update_plugins
+from pacman.ecs.hierarchy import hierachy_plugins
from pacman.ecs.update import schedule_plugins
-from pacman.ecs.world import (
- Component,
- Query,
- World,
-)
+from pacman.ecs.world import World
def minimal_plugins(world: World) -> None:
+ """Setup the minimum plugins for general game functionality."""
world.with_plugins(
schedule_plugins, hierachy_plugins, coord_update_plugins
)
-
-
-class WiggleSprite(Component):
- pass
-
-
-def move_sprites(sprites: Query[Coord2D, WiggleSprite]) -> None:
- import math
-
- tick = pygame.time.get_ticks() / 100
- for coord in sprites:
- pos = max(tick - coord.z * 0.3, 0.0)
- coord.y = 500.0 + math.sin(pos) * 30.0
-
-
-class WiggleRoot(Component):
- pass
-
-
-def move_root(root: Query[Coord2D, WiggleRoot]) -> None:
- import math
-
- tick = pygame.time.get_ticks() / 100
- for coord in root:
- coord.x = 30.0 + math.cos(tick) * 30.0
-
-
-def print_clicks(events: WindowEvents) -> None:
- for pos, click in events.clicks():
- print(pos, click)
-
-
-def setup_world(world: World, assets: Assets[Sprite2D]) -> None:
- asset_paths = list(Path("./asset").rglob("*.png"))
- root = world.spawn(
- tuple[Coord2D, CoordAbs2D, WiggleRoot],
- (
- Coord2D(100.0, 0.0, 0.0),
- CoordAbs2D(),
- WiggleRoot(),
- ),
- )
- world.spawn_many(
- tuple[
- Coord2D,
- CoordAbs2D,
- Sprite2D,
- ShouldRender,
- WiggleSprite,
- ChildOf,
- ],
- [
- (
- Coord2D(
- 30.0 * (n := (i + j) % len(asset_paths)),
- 500.0 + 10.0,
- 1.0 * n + j,
- ),
- CoordAbs2D(),
- assets.load(asset_path.as_posix()),
- ShouldRender(),
- WiggleSprite(),
- ChildOf(root),
- )
- for j in range(30)
- for i, asset_path in enumerate(asset_paths)
- ],
- )
-
-
-if __name__ == "__main__":
- world = (
- World()
- .with_plugins(minimal_plugins, graphics_plugins)
- .with_systems(StartupSchedule, setup_world)
- .with_systems(UpdateSchedule, move_sprites, move_root, print_clicks)
- )
- world.run_main()
+"""Asset loading in ECS."""
+
from abc import ABC, abstractmethod
from collections.abc import Callable
from os import PathLike
class Asset(ABC):
+ """A data type which may be loaded from a specific identifier."""
+
@classmethod
@abstractmethod
- def load(cls, ty: type, path: str) -> Self: ...
+ def load(cls, ty: type, path: str) -> Self:
+ """Load the type from the specified identifier.
+
+ Args:
+ ty:
+ The type, including generics, of cls.
+ path:
+ The identifier of the asset, usually a file path.
+ """
class Assets[T: Asset](Resource):
+ """A collection of cached assets to be loaded."""
+
def __init__(self, ty: type) -> None:
+ """Create a new empty asset collection."""
self._lookup: dict[str, T] = {}
self._asset_ty: type[T] = get_args(resolve_type_aliases(ty))[0]
def load(self, s: str | PathLike[str]) -> T:
+ """Load the given asset or fetch it from cache."""
s = abspath(s)
if s not in self._lookup:
self._lookup[s] = self._asset_ty.load(self._asset_ty, s)
+"""ECS coordinate logic."""
+
from collections.abc import Iterator
from typing import override
class CoordAbs2D(Component):
+ """Coord absolute in the window space."""
+
_vec: Vector3
def __init__(self) -> None:
+ """Create a new CoordAbs2D."""
self._vec = Vector3(0.0)
class Coord2D(Vector3, Component):
- pass
+ """Coord relative to the parent entity."""
class CoordBundle(Bundle):
+ """A bundle of coords that allows for easily setting up coordinates."""
+
def __init__(self, x: float = 0, y: float = 0, depth: float = -1) -> None:
+ """Create a new coord bundle."""
self.coord = Coord2D(x, y, depth)
self.coord_abs = CoordAbs2D()
def update_coord(entities: Query[tuple[CoordAbs2D, Coord2D]]) -> None:
+ """Update absolute coords to match relative coords."""
for abs_coord, coord in entities:
abs_coord._vec = coord.xyz
def update_coord_hierarchy(hierarchy: Hierarchy[CoordAbs2D]) -> None:
+ """Update absolute coords to be offset by parent's absolute coords."""
for parent, child in hierarchy.top_down():
if parent is None:
continue
def coord_update_plugins(world: World) -> None:
+ """Coordinate related logic."""
world.with_systems(
UpdateSchedule,
Systems(update_coord, update_coord_hierarchy)
+"""Simple 2D graphics and input in ECS."""
+
from collections.abc import Iterator
from dataclasses import dataclass
from time import time
class ShouldRender(Component):
- pass
+ """A marker component to avoid rendering a given entity."""
class StopAnimation(Component):
- pass
+ """A marker component to avoid animating a given entity."""
@dataclass
class Window(Resource):
+ """A window with clock to keep refresh rate."""
+
surface: pygame.Surface
clock: pygame.time.Clock
@dataclass
class WindowPositionned(Component):
+ """A component to position an entity relative to the window."""
+
pos: Vector2
offset: Vector2
@dataclass
class WindowEvents(Resource):
+ """A simple current tick list of window events."""
+
events: list[pygame.event.Event]
def clicks(self) -> Iterator[tuple[int, pygame.Vector2]]:
+ """Iterate over each click event on screen."""
return (
cast(tuple[int, pygame.Vector2], (e.pos, e.button))
for e in self.events
)
def keys_itr(self) -> Iterator[tuple[int, bool]]:
+ """Iterate over each key evenet and its pressed status."""
return (
(cast(int, (e.key)), e.type == pygame.KEYDOWN)
for e in self.events
@dataclass
class Inputs(Resource):
+ """An input tracking resource, including smart direction."""
+
inputs: dict[int, float]
def get_dir(self) -> int | None:
+ """Get which direction has been most recently pressed."""
res = sorted(
((k, v) for k, v in self.inputs.items() if k in dir_lookup),
key=lambda e: e[1],
return dir_lookup[res[0][0]]
def activate(self) -> bool:
+ """Get whether an activate key is pressed."""
return K_RETURN in self.inputs
@dataclass
class Sprite2D(Component, Asset):
+ """A 2D sprite for use in rendering."""
+
surface: pygame.Surface
@override
return Sprite2D(pygame.image.load(path))
def copy(self) -> "Sprite2D":
+ """Make a shallow copy of this sprite."""
return Sprite2D(self.surface)
def key_just_pressed(key: int) -> System[bool]:
+ """Return a system which returns whether the key was pressed this frame."""
+
def inner(events: WindowEvents) -> bool:
return any(key == k and pressed for k, pressed in events.keys_itr())
window: Maybe[Window],
world: World,
) -> RawComponentHook[WindowPositionned]:
+ """Hook when WindowPositionned components are inserted."""
+
def inner(added: RawComponentArg[WindowPositionned]) -> None:
dims = Vector2(0, 0)
if (win := window.v) is not None:
def update_window_positionned(
elems: Query[tuple[WindowPositionned, Coord2D]], window: Window
) -> None:
+ """Update window-positionned coordinates based on window size."""
dims = Vector2(window.surface.get_size())
for pos, e in elems:
e.xy = dims.elementwise() * pos.pos + pos.offset
def init_graphics(world: World) -> None:
+ """Initialize the graphics data."""
pygame.init()
world.spawn_resource(
Window(
def poll_events(events: WindowEvents) -> None:
+ """Poll the window events from the backend."""
events.events = pygame.event.get()
def close_on_close(world: World, events: WindowEvents) -> None:
+ """Signal to exit the world based on window events."""
if pygame.QUIT in (event.type for event in events.events):
world.spawn(WorldShouldExit, WorldShouldExit())
window: Window,
sprites_q: Query[tuple[CoordAbs2D, Sprite2D], ShouldRender],
) -> None:
+ """Render depth-ordered 2d sprites."""
sprites_raw = sorted(
sprites_q,
key=lambda e: e[0]._vec.z,
@dataclass
class AnimationSet(Component):
+ """An animation set, allowing for simple cyclic animations."""
+
anims: list[list[Sprite2D]]
ticks_per_frame: int
idx: int = 0
@property
def curr_frame(self) -> Sprite2D:
+ """Return the current animation frame."""
return self.anims[self.idx][self.frame]
def tick(self) -> None:
+ """Update the animation by one frame."""
self.curr_tick += 1
self.frame += self.curr_tick // self.ticks_per_frame
self.curr_tick %= self.ticks_per_frame
def insert_animation_sprites(
world: World, query: Query[tuple[Entity, AnimationSet], Not[Sprite2D]]
) -> None:
+ """Insert a sprite on entities which have an animation but no sprite."""
inserts = []
for ent, anim in query:
inserts.append((ent, anim.curr_frame.copy()))
tuple[Sprite2D, AnimationSet], tuple[ShouldRender, Not[StopAnimation]]
],
) -> None:
+ """Update animations."""
for sprite, anim in query:
anim.tick()
sprite.surface = anim.curr_frame.surface
def update_inputs(inputs: Inputs, events: WindowEvents) -> None:
+ """Update the inputs based on window events."""
for key, state in events.keys_itr():
if state:
inputs.inputs[key] = time()
def graphics_plugins(world: World) -> None:
+ """Setup graphics related logic."""
from pacman.ecs.gui import gui_plugins
world.with_systems(StartupSchedule, init_graphics).with_systems(
+"""GUI in ECS, with buttons, text, and more."""
+
from collections.abc import Callable
from dataclasses import dataclass, field
from enum import Enum, auto
@dataclass
class Font(Asset, Component):
+ """A component for displaying text, newlines not supported."""
+
font: pygame.font.Font
@override
return Font(pygame.font.Font(path, int(size)))
def center_text_for(self, text: str, dims: Vector2) -> CoordBundle:
+ """Compute how to center the text in the given rectangle."""
width, _ = self.font.size(text)
height = self.font.get_ascent() + self.font.get_descent()
return CoordBundle(
@dataclass
class TextDisplay(Component):
+ """A component for displaying text, newlines not supported."""
+
text: str
font: Font
color: Color = field(default_factory=lambda: Color(255, 255, 255))
@dataclass
class ComputedText(Component):
+ """A component that continuously updates which text the button has."""
+
cb: System[str]
class ButtonState(Enum):
+ """The states a button can be in."""
+
unselected = auto()
selected = auto()
pressed = auto()
@dataclass
class Button(Component):
+ """A button component for simple menus."""
+
cb: System[None]
state: ButtonState = ButtonState.unselected
@dataclass
class SelectedButton(Resource):
+ """A resource indicating a given button is currently selected."""
+
button: Entity
selected: SelectedButton,
events: WindowEvents,
) -> None:
+ """Update which button is selected based on inputs and button location."""
for key, active in events.keys_itr():
if not active:
continue
inputs: Inputs,
world: World,
) -> None:
+ """Update the button states based on inputs and selected button."""
cbs = []
for entity, button in buttons:
if entity is not selected.button:
elif not inputs.activate():
button.state = ButtonState.selected
for cb in cbs:
- world.run_raw_system(cb)
+ world.run_simple_system(cb)
def update_button_texture(buttons: Query[tuple[Button, AnimationSet]]) -> None:
+ """Select an animation set for buttons."""
for button, anim in buttons:
anim.idx = button.state.value - 1
elems: Query[tuple[TextDisplay, ComputedText]],
world: World,
) -> None:
+ """Update TextDisplay based on ComputedText components."""
for display, compute in elems:
- display.text = world.run_raw_system(
+ display.text = world.run_simple_system(
compute.cb,
cast(Callable[[], str], lambda display=display: display.text),
)
def update_text_sprites(
elems: Query[tuple[TextDisplay, Sprite2D | Entity]], queue: CommandQueue
) -> None:
+ """Update sprites for TextDisplay, inserting a Sprite2D if needed."""
inserts = []
for display, e in elems:
if (
def setup_gui_resources() -> None:
+ """Setup GUI related resources."""
pygame.font.init()
def gui_plugins(world: World) -> None:
+ """Setup GUI related logic."""
world.with_systems(StartupSchedule, setup_gui_resources).with_systems(
UpdateSchedule,
Systems(
+"""Hierarchy in ECS, with invariant maintenance."""
+
from collections.abc import Callable, Generator, Iterable
from typing import cast, get_args, override
class Children(Component):
+ """A component storing the children of this entity."""
+
def __init__(self, children: Iterable[Entity]) -> None:
+ """Create a parent-children relation."""
self._children = set(children)
@property
def children(self) -> Generator[Entity]:
+ """The children of this entity."""
yield from self._children
class ChildOf(Component):
+ """A component storing the parent of this entity."""
+
def __init__(self, parent: Entity) -> None:
+ """Create a child-parent relation."""
self._parent = parent
@property
def parent(self) -> Entity:
+ """The parent of this entity."""
return self._parent
-def parent_insert_hook(
+def children_insert_hook(
queue: CommandQueue, children: Query[ChildOf]
) -> RawComponentHook[Children]:
+ """The hook called when a Children component is inserted."""
+
def inner(itr: RawComponentArg[Children]) -> None:
inserts = [
(child, ChildOf(ent))
return inner
-def parent_remove_hook(queue: CommandQueue) -> RawComponentHook[Children]:
+def children_remove_hook(queue: CommandQueue) -> RawComponentHook[Children]:
+ """The hook called when a Children component is removed."""
+
def inner(itr: RawComponentArg[Children]) -> None:
removes = [e for _, children in itr for e in children.children]
queue.queue(lambda world: world.remove_many(ChildOf, removes))
return inner
-def child_insert_hook(
+def childof_insert_hook(
queue: CommandQueue,
parents: Query[Children],
) -> RawComponentHook[ChildOf]:
+ """The hook called when a ChildOf component is inserted."""
+
def inner(itr: RawComponentArg[ChildOf]) -> None:
inserts = {}
for ent, inserted in itr:
return inner
-def child_remove_hook(parents: Query[Children]) -> RawComponentHook[ChildOf]:
+def childof_remove_hook(
+ parents: Query[Children],
+) -> RawComponentHook[ChildOf]:
+ """The hook called when a ChildOf component is removed."""
+
def inner(itr: RawComponentArg[ChildOf]) -> None:
for ent, removed in itr:
parent = parents.get(removed.parent)
def hierachy_plugins(world: World) -> None:
- world.with_hook("insert", Children, parent_insert_hook).with_hook(
- "remove", Children, parent_remove_hook
- ).with_hook("insert", ChildOf, child_insert_hook).with_hook(
- "remove", ChildOf, child_remove_hook
+ """Setup hierarchy related logic."""
+ world.with_hook("insert", Children, children_insert_hook).with_hook(
+ "remove", Children, children_remove_hook
+ ).with_hook("insert", ChildOf, childof_insert_hook).with_hook(
+ "remove", ChildOf, childof_remove_hook
)
class Hierarchy[Items, Filters = tuple[()]](FromWorld):
+ """A hierarchy allowing for convenient manipulation."""
+
def __init__(
self,
roots: RootsQuery[Items, Filters],
nodes: NodesQuery[Items, Filters],
) -> None:
+ """Create a hierarchy from its queries."""
self._roots = roots
self._nodes = nodes
self._nodes.cleanup()
def top_down(self) -> Generator[tuple[Items | None, Items]]:
+ """Traverse this hierarchy top down."""
queue: list[tuple[Items | None, Entity]] = [
(None, e) for e in self._roots
]
queue.extend((child, e) for e in children.children)
def despawn_with_children(self, root: Entity) -> Callable[[World], None]:
+ """Despawn an entity and its children which are a part of the query."""
queue = [root]
entities = set()
while len(queue) != 0:
+"""Schedule logic for systems, with run conditions and labels."""
+
from collections import deque
from collections.abc import Callable, Generator, Iterable
from dataclasses import dataclass
@dataclass
class DAGNode[T]:
+ """A node in a DAG."""
+
parents: set[T]
children: set[T]
def is_root(self) -> bool:
+ """Whether this node has no parents."""
return len(self.parents) == 0
class DAG[T]:
+ """A Directed Acyclic Graph, used for dependency ordering."""
+
def __init__(self) -> None:
+ """Initialize an empty DAG."""
self._nodes: dict[T, DAGNode[T]] = {}
def _get_node(self, node: T) -> DAGNode[T]:
return self._nodes[node]
def add_edge(self, parent: T, child: T) -> None:
+ """Add an edge between two nodes."""
parents = self.node_parents(parent)
children = self.node_children(child)
shared = parents & children
self._get_node(child).parents.add(parent)
def node_parents(self, node: T) -> set[T]:
+ """The set of parents of a node in the graph."""
if node not in self._nodes:
return set()
res = set()
return res
def node_children(self, node: T) -> set[T]:
+ """The set of children of a node in the graph."""
if node not in self._nodes:
return set()
res = set()
return res
def traverse_from(self, root: T) -> Generator[T]:
+ """Traverse the DAG from dependents to dependencies."""
visited: set[T] = set()
queue = deque([root])
while len(queue) != 0:
class DAGLabel:
+ """A DAG label, used to structure the DAG."""
+
def __init__(self, name: str) -> None:
+ """Create a named DAGLabel."""
self.name = name
@override
class ScheduleLabel:
+ """A schedule label, used to name schedules in the world."""
+
def __init__(self, name: str) -> None:
+ """Create a named ScheduleLabel."""
self.name = name
@override
class SystemSet:
+ """A system set, used to label systems as being in a group."""
+
def __init__(self, name: str) -> None:
+ """Create a named SystemSet."""
self.name = name
self._pre = DAGLabel(name + "-pre")
self._post = DAGLabel(name + "-post")
class Schedule[T]:
+ """A schedule represented by a DAG, with run predicates and system sets."""
+
def __init__(self) -> None:
+ """Initialize an empty schedule."""
self._dag: DAG[T | DAGLabel | None] = DAG()
self._group_appartenance: dict[T | SystemSet, set[SystemSet]] = {}
self._run_predicates: dict[T | SystemSet, Callable[[], bool]] = {}
group: Iterable[SystemSet],
cond: Callable[[], bool] | None,
) -> None:
+ """Add ordering items to this schedule."""
+
def get_pre(e: T | SystemSet) -> T | DAGLabel:
if isinstance(e, SystemSet):
return e._pre
self._run_predicates[item] = cond
def traverse(self) -> Generator[T]:
+ """Traverse the schedule, dependencies first."""
cond_cache: dict[T | SystemSet, bool] = {}
def should_run(val: T | SystemSet) -> bool:
+"""World update and schedule logic."""
+
from pacman.ecs.schedule import (
MainSchedule,
StartupSchedule,
class WorldIsInit(Resource):
+ """A marker to signal the world has ran the startup schedule."""
+
pass
def run_startup(world: World, _: Not[WorldIsInit]) -> None:
+ """Run the startup schedule."""
world.tick(StartupSchedule)
world.spawn(WorldIsInit, WorldIsInit())
def run_update(world: World, _: WorldIsInit) -> None:
+ """Run the udpate schedule."""
world.tick(UpdateSchedule)
def schedule_plugins(world: World) -> None:
+ """Setup the schedule related functionality."""
world.with_systems(MainSchedule, run_startup, run_update).with_systems(
UpdateSchedule, Systems(PreUpdate, Update, PostUpdate).chain()
)
+"""The core of the ECS engine, where everything is bridged together."""
+
import inspect
from abc import ABC, abstractmethod
from collections.abc import Callable, Container, Iterable, Iterator
class Bundle(ABC):
+ """A class which acts as a set of components."""
+
@abstractmethod
- def components_iter(self) -> Iterator["Component"]: ...
+ def components_iter(self) -> Iterator["Component"]:
+ """Iterate over the components of this instance."""
+
@classmethod
@abstractmethod
- def components_type_iter(cls, ty: type) -> Iterator[type["Component"]]: ...
+ def components_type_iter(cls, ty: type) -> Iterator[type["Component"]]:
+ """Iterate over the types of compnents this type contains.
+
+ It is required to match components_iter.
+ """
type Components = RecursiveTuple["Component | Bundle"]
def components_iter(components: Components) -> Iterator["Component"]:
+ """Iterate over the components of nested components."""
if isinstance(components, Component):
yield components
elif isinstance(components, tuple):
def components_type_iter(
components: type[Components],
) -> Iterator[type["Component"]]:
+ """Iterate over the component types of nested components."""
if origin_or_cls(components) is tuple:
yield from (
e
class Archetype:
+ """An archetype, storing entities which all have the same components.
+
+ Archetypes are never deleted, and cache operation intermediates.
+ """
+
def __init__(self, world: "World", types: Iterable[ComponentID]) -> None:
+ """Create a new archetype from the given component types."""
self._world = world
self._types = frozenset(types)
self._positions: dict[Entity, int] = {}
def _compute_add[T: Components](
self, ty: type[T]
) -> tuple["Archetype", list[ComponentID]]:
+ """Compute the destination archetype and list of component types."""
ptr = id(ty)
if res := self._add_cache.get(ptr):
return (res[1], res[2])
def _compute_remove[T: Components](
self, ty: type[T]
) -> tuple["Archetype", list[ComponentID]]:
+ """Compute the destination archetype and list of component types."""
ptr = id(ty)
if res := self._remove_cache.get(ptr):
return (res[1], res[2])
def add_components[T: Components](
self, ty: type[T], components: list[tuple["Entity", T]]
) -> Callable[[], None]:
+ """Add components to entities, moving them to other archetypes."""
dst, comps_ty = self._compute_add(ty)
comps = [components_iter(e) for _, e in components]
positions = [self._positions[e] for e, _ in components]
def remove_components[T: Components](
self, ty: type[T], entities: list["Entity"]
) -> Callable[[], None]:
+ """Remove components from entities, moving them to other archetypes."""
dst, comps_ty = self._compute_remove(ty)
positions = [self._positions[e] for e in entities]
return res
def remove(self, entities: list["Entity"]) -> Callable[[], None]:
+ """Remove the given entities from self, despawning them."""
entities = [e for e in entities if e in self._positions]
positions = [self._positions[e] for e in entities]
comps = [
self,
cb: Callable[[ComponentID | None], Iterable[Any]],
) -> None:
+ """Extends self, calling cb to fetch tables of each required type.
+
+ None is used as a marker to request the entities.
+ """
for entity in cb(None):
self._positions[entity] = len(self._entities)
self._entities.append(entity)
), "Callaback's return lists weren't all of equal size"
def _discard(self, itr: Iterable[int]) -> None:
+ """Discard the given entities in self.
+
+ This does not maintain invariants by itself.
+ """
sort = sorted(itr, reverse=True)
def remove_idx[T](
class FromWorld(ABC):
+ """A type which may be extracted from a world to be given to a system."""
+
@classmethod
@abstractmethod
def extract_from_world(
cls, ty: type, world: "World"
- ) -> Callable[[], Self | None]: ...
+ ) -> Callable[[], Self | None]:
+ """Extract this type to pass it to a system.
+
+ A callable is used to allow for easy caching of intermediate
+ computations.
+ """
def cleanup(self) -> None:
+ """Cleanup this system argument after system use."""
return None
class QueryFilter(ABC):
+ """A query filter, that is, a type which may be filtered for by a query."""
+
@classmethod
@abstractmethod
def filter(
cls, ty: type
- ) -> Callable[[Archetype], bool | Callable[[int], bool]]: ...
+ ) -> Callable[[Archetype], bool | Callable[[int], bool]]:
+ """Filter for an archetype.
+
+ Computations about archetype appartenance and type extraction are done
+ at each step to improve performance.
+ """
type QueryFilters = RecursiveTuple[QueryFilter]
class _NoFetchCls:
- pass
+ """A sentinel class indicating a query item failed to fetch."""
_NoFetch = _NoFetchCls()
class QueryItem(QueryFilter):
+ """A query item, that is, a type which may be fetched by a query."""
+
@override
@classmethod
def filter(
@classmethod
def fetch_prio(cls, ty: type) -> int:
+ """Returns the priority of this type in case of unions."""
del ty
return 0
@abstractmethod
def filter_fetch(
cls, ty: type
- ) -> Callable[[Archetype], None | Callable[[int], Self | _NoFetchCls]]: ...
+ ) -> Callable[[Archetype], None | Callable[[int], Self | _NoFetchCls]]:
+ """Fetch this item for an archetype.
+
+ Computations about archetype appartenance and type extraction are done
+ at each step to improve performance.
+ """
type QueryItems = RecursiveTuple[QueryItem | None]
class Component(QueryItem):
+ """A component, attachable to an entity, fundamental to ECS."""
+
@override
@classmethod
def filter(cls, ty: type) -> Callable[[Archetype], bool]:
class Entity(QueryItem):
+ """An entity, unique identifier in a world, also a query item."""
+
@override
@classmethod
def filter(cls, ty: type) -> Callable[[Archetype], bool]:
class Not[T: FromWorld | Any](FromWorld, QueryItem):
+ """A simple system argument and query item, matching if T does not."""
+
@override
@classmethod
def filter_fetch(
class Maybe[T: FromWorld](FromWorld):
+ """A simple system argument which always instantiates fallible args."""
+
def __init__(self, v: T | None) -> None:
+ """Initialize a new maybe value."""
self.v = v
@override
class Query[Items, Filters = tuple[()]](
FromWorld, Iterable[Items], Container[Entity]
):
+ """A query system arg, that allows operations on components."""
+
def __init__(
self,
world: "World",
fetchers: type[Items],
filters: type[Filters] | None = None,
) -> None:
+ """Construct a query for the world with the given items and filters."""
if filters is None:
filters = cast(type[Filters], tuple[()])
resolved_fetchers = resolve_type_aliases(fetchers)
[Archetype], Callable[[int], QueryItems | _NoFetchCls] | None
],
]:
+ """Process the query fetchers as much as possible at each step."""
args = list(map(Query.preprocess_fetchers, get_args(fetcher)))
def tuple_constructor(
def preprocess_filters(
filter: type[QueryFilters],
) -> Callable[[Archetype], Callable[[int], bool] | bool]:
+ """Process the query filters as much as possible at each step."""
args = list(map(Query.preprocess_filters, get_args(filter)))
def tuple_constructor(
return cast(QueryFilter, filter).filter(filter)
def add_archetype(self, archetype: Archetype) -> None:
+ """Observe a new archetype for consideration in iteration."""
filt = self._filters_raw(archetype)
if filt is False:
return
self._fetchers.append(fetcher)
def updated_from_world(self) -> Self:
+ """Update the query in place, observing new archetypes."""
for arch in self._world._archetypes[self._seen_archetypes :]:
self.add_archetype(arch)
self._seen_archetypes = len(self._world._archetypes)
return self._fetchers[arch_idx](components_idx) is not _NoFetch
def get(self, entity: Entity) -> Items | None:
+ """Get the items for the given entity, or None if absent."""
arch: Archetype | None = self._world._entities.get(entity)
if arch is None:
return None
return cast(Items, res)
def __getitem__(self, entity: Entity) -> Items:
+ """Get the items for the given entity.
+
+ Raises:
+ KeyError: if the entity is not in the query.
+ """
arch: Archetype | None = self._world._entities.get(entity)
if arch is None:
raise KeyError("Entity not in world", entity)
return cast(Items, res)
def single(self) -> Items | None:
+ """Fetches exactly one element from the query if it exists.
+
+ Raises:
+ Exception: if more than one entity matches the query.
+ """
itr: Iterator[Items] = self.__iter__()
res = next(itr, None)
if next(itr, None) is not None:
raise Exception("More than one entity matches single query")
return res
- def single_e(self) -> Items:
- if (res := self.single()) is None:
- raise Exception("single_e must contain exactly one matching item")
- return res
-
class Resource(FromWorld, Component):
+ """A resource system argument.
+
+ Resources exist in only one instance in the world.
+ """
+
@override
@classmethod
def extract_from_world(
def extract_default(
cls, ty: type, world: "World"
) -> Callable[[], Self | None]:
+ """Return a callable that optionally returns a default value.
+
+ Override this to allow for resources to be initialized if not present.
+ """
del ty, world
return lambda: None
class CommandQueue(FromWorld):
+ """A command queue system argument for defered execution.
+
+ It stores functions to be called after system exit during cleanup.
+ Use it to avoid invalidating queries while still in use.
+ """
+
def __init__(self, world: "World") -> None:
+ """Create an emtpy CommandQueue."""
self._world = world
self._queue: list[Callable[[World], Any]] = []
sys(self._world)
def queue(self, sys: Callable[["World"], Any]) -> None:
+ """Enqueue the given sys to be executed after system exit."""
self._queue.append(sys)
def _system_failed_va[*P = *tuple[()]](*_: *P) -> Never:
+ """A default system fail callback for systems with arguments."""
raise ValueError("Failed to fetch arguments for system")
def _system_failed() -> Never:
+ """A default system fail callback."""
raise ValueError("Failed to fetch arguments for system")
@dataclass
class ComponentHooks[T: Component]:
+ """Component lifecycle hooks."""
+
_insert: RawComponentHook[T] = lambda _: None
_remove: RawComponentHook[T] = lambda _: None
def add_insert(self, cb: RawComponentHook[T]) -> None:
+ """Add an insert hook to these hooks."""
self._insert = composed_by(self._insert, cb, lambda _1, _2: None)
def add_remove(self, cb: RawComponentHook[T]) -> None:
+ """Add a remove hook to these hooks."""
self._remove = composed_by(self._remove, cb, lambda _1, _2: None)
class World(FromWorld):
+ """The world, core of the ECS.
+
+ It contains every entity, their components, stored in archetypes,
+ as well as the schedules.
+ """
+
def __init__(self) -> None:
+ """Create an empty world."""
self._entities: dict[Entity, Archetype] = {}
self._archetypes_lookup: dict[frozenset[ComponentID], Archetype] = {}
self._archetypes: list[Archetype] = []
] = {}
def component_id(self, comp: type) -> ComponentID:
+ """Get a unique ComponentID for a given type."""
if (res := self._components_lookup.get(comp)) is None:
res = ComponentID(len(self._components_cache))
self._components_cache.append((comp, ComponentHooks()))
return res
def spawn_empty(self) -> Entity:
+ """Spawn a single empty entity.
+
+ If called repeatedly, it is preferable to call spawn_many_empty.
+ """
return self.spawn_many_empty(1)[0]
def spawn_many_empty(self, n: int) -> list[Entity]:
+ """Spawn many empty entities efficiently."""
res = [Entity() for _ in range(n)]
self._base_archetype._extend(
lambda ty: (
)
return res
- def spawn[T: Components](self, ty: type[T], comp: T) -> Entity:
- return self.spawn_many(ty, [comp])[0]
-
def spawn_resource[T: Resource](
self, comp: T, ty: type[T] | None = None
) -> Entity:
+ """Spawn a resource or overwrite it if it exists."""
if ty is None:
ty = type(comp)
return self.spawn(ty, comp)
def despawn_resource[T: Resource](self, ty: type[T]) -> None:
+ """Despawn a resource if it exists."""
type Ty = ty # type: ignore
def remove(query: Query[Entity, Ty], world: World) -> None:
world.remove_many(ty, list(query))
- self.run_raw_system(remove)
+ self.run_simple_system(remove)
+
+ def spawn[T: Components](self, ty: type[T], comp: T) -> Entity:
+ """Spawn an entity, with initial components.
+
+ If called repeatedly, it is preferable to call spawn_many.
+ """
+ return self.spawn_many(ty, [comp])[0]
def spawn_many[T: Components](
self, ty: type[T], comps: list[T]
) -> list[Entity]:
+ """Spawn multiple entities efficiently, with initial components."""
ents = self.spawn_many_empty(len(comps))
self._base_archetype.add_components(
cast(type[Components], ty),
return ents
def despawn(self, ent: Entity) -> None:
+ """Despawn a single entity.
+
+ If called repeatedly, it is preferable to call despawn_many.
+ """
self.despawn_many([ent])
def despawn_many(self, ents: Iterable[Entity]) -> None:
+ """Despawn many entities efficiently."""
archs: dict[Archetype, list[Entity]] = {}
for ent in ents:
arch = self._entities[ent]
cb()
def insert[T: Components](self, ty: type[T], ent: Entity, comp: T) -> None:
+ """Insert components into a single entities.
+
+ If called repeatedly, it is preferable to call insert_many.
+ """
self.insert_many(ty, [(ent, comp)])
def insert_many[T: Components](
self, ty: type[T], comps: Iterable[tuple[Entity, T]]
) -> None:
+ """Insert components into many entities efficiently."""
archs: dict[Archetype, list[tuple[Entity, T]]] = {}
for ent, comp in comps:
arch = self._entities[ent]
cb()
def remove[T: Components](self, ty: type[T], ent: Entity) -> None:
+ """Remove components from a single entities.
+
+ If called repeatedly, it is preferable to call remove_many.
+ """
self.remove_many(ty, [ent])
def remove_many[T: Components](
- self, ty: type[T], comps: Iterable[Entity]
+ self, ty: type[T], entities: Iterable[Entity]
) -> None:
+ """Remove components from many entities efficiently."""
archs: dict[Archetype, list[Entity]] = {}
- for ent in comps:
+ for ent in entities:
if ent not in self._entities:
continue
arch = self._entities[ent]
def _sys_extractors[T](
self, system: System[T]
) -> tuple[Callable[[], FromWorld | None], ...]:
+ """Create extractors for each arg of a system."""
if system in self._extractors_cache:
return self._extractors_cache[system]
sig = inspect.signature(system, eval_str=True)
system: System[Callable[[*P], T]],
default: Callable[[*P], T] = _system_failed_va,
) -> Callable[[*P], T]:
+ """Add a system with arguments and return the raw system."""
extractors = self._sys_extractors(system)
def inner(*cb_args: *P) -> T:
return inner
- def add_raw_system[T](
+ def add_simple_system[T](
self,
system: System[T],
default: Callable[[], T] = _system_failed,
) -> RawSystem[T]:
+ """Add a simple system and return the raw system."""
extractors = self._sys_extractors(system)
def inner() -> T:
*args: *P,
default: Callable[[*P], T] = _system_failed_va,
) -> T:
+ """Run a system with arguments."""
return self.add_system(system, default)(*args)
- def run_raw_system[T](
+ def run_simple_system[T](
self,
system: System[T],
default: Callable[[], T] = _system_failed,
) -> T:
- return self.add_raw_system(system, default)()
+ """Run a simple system."""
+ return self.add_simple_system(system, default)()
@classmethod
@override
return lambda: world
def tick(self, schedule: ScheduleLabel) -> None:
+ """Execute the given schedule once."""
if schedule not in self._schedules:
return
for system in self._schedules[schedule].traverse():
schedule: ScheduleLabel,
stop_cond: RawSystem[bool] = lambda: False,
) -> None:
+ """Repeatedly run the schedule until stop_cond returns true."""
while not stop_cond():
self.tick(schedule)
def _register_schedule_system[T](self, system: System[T]) -> System[T]:
+ """Add a system to the scheduler cache, for faster later lookup."""
if system not in self._systems:
- self._systems[system] = self.add_raw_system(system, lambda: None)
+ self._systems[system] = self.add_simple_system(
+ system, lambda: None
+ )
return system
def with_systems(
schedule: ScheduleLabel,
*systems: "System | Systems",
) -> Self:
+ """Add the systems to the given schedule."""
if schedule not in self._schedules:
self._schedules[schedule] = Schedule()
Systems(*systems)._apply(self, self._schedules[schedule])
ty: type[T],
hook: ComponentHook[T],
) -> Self:
+ """Add the component hook to the world for the given component."""
comp_id = self.component_id(resolve_type_aliases(ty))
hooks = cast(ComponentHooks[T], self._components_cache[comp_id][1])
sys = self.add_system(hook, default=lambda _: None)
return self
def with_plugins(self, *plugins: Plugin) -> Self:
+ """Add the plugins to the world."""
for plugin in plugins:
if isinstance(plugin, tuple):
self.with_plugins(plugin)
return self
def run_main(self) -> None:
- self.run(MainSchedule, self.add_raw_system(should_exit))
+ """Run the default main schedule on the world."""
+ self.run(MainSchedule, self.add_simple_system(should_exit))
_order_groups: int = 0
class Systems:
+ """A builder to create complex system ordering and set apparentance."""
+
def __init__(self, *systems: "System | SystemSet | Systems") -> None:
+ """Create a builder with these systems."""
self._systems = systems
self._chain = False
self._pre: list[System | SystemSet] = []
self._conds: list[System[bool]] = []
def chain(self) -> Self:
+ """Mark these systems to be executed in order."""
self._chain = True
return self
def after(self, *systems: System | SystemSet) -> Self:
+ """Order these systems to execute after other schedule elements."""
self._pre.extend(systems)
return self
def before(self, *systems: System | SystemSet) -> Self:
+ """Order these systems to execute before other schedule elements."""
self._post.extend(systems)
return self
def in_set(self, *sets: SystemSet) -> Self:
+ """Add these systems to zero or more SystemSet."""
self._in_set.extend(sets)
return self
def run_if(self, *conds: System[bool]) -> Self:
+ """Add zero or more run conditions to these systems."""
self._conds.extend(conds)
return self
def _apply(self, world: World, sched: Schedule[System[Any]]) -> SystemSet:
+ """Applie this builder to the world and schedule."""
global _order_groups
_order_groups += 1
label = SystemSet(f"order-group-{_order_groups}")
)
for cond in self._conds:
sched.add_ordering(
- (label,), (), (), (), world.add_raw_system(cond, lambda: False)
+ (label,),
+ (),
+ (),
+ (),
+ world.add_simple_system(cond, lambda: False),
)
sched.add_ordering((label,), self._pre, self._post, self._in_set, None)
if self._chain:
class WorldShouldExit(Resource):
- pass
+ """A marker resource to signal the world should exit at this tick."""
def should_exit(e: Maybe[WorldShouldExit]) -> bool:
+ """Returns whether this world has the exit marker resource."""
return e.v is not None
class GhostVariant(Enum):
- """
- Enumeration for Ghost variant and behavior.
- """
+ """Enumeration for Ghost variant and behavior."""
A = auto()
B = auto()
slf: "Ghost",
blinky_pos: tuple[int, int],
) -> tuple[int, int]:
- """
- Return ghost target coordinates.
- """
+ """Return ghost target coordinates."""
match self:
case GhostVariant.A:
return player
class GhostState(Enum):
- """
- Enumeration for Ghost state and attitudes.
- """
+ """Enumeration for Ghost state and attitudes."""
chase = auto()
wander = auto()
@dataclass
class InitialPauseTimer(Resource):
- """
- Resource use for the initial pause before starting a level.
- """
+ """Resource use for the initial pause before starting a level."""
tick: int = 0
class Config(Resource, BaseModel):
- """
- Resource storing the configuration of the game.
- """
+ """Resource storing the configuration of the game."""
ghost_chase_speed: float = Field(default=2.5, ge=0.1, le=100.0)
ghost_flee_speed: float = Field(default=1.5, ge=0.1, le=100.0)
class Highscores(Resource, BaseModel):
- """
- Store the hichscores in a dictionnary.
- """
+ """Store the hichscores in a dictionnary."""
scores: dict[str, int]
class GameIsOver(Resource):
- """
- Resource set when game is over.
- """
+ """Resource set when game is over."""
pass
@dataclass
class GameState(Resource):
- """
- Resource storing current score, level, timer and lives.
- """
+ """Resource storing current score, level, timer and lives."""
score: int = 0
level: int = 0
class CheatMode(Component):
- """
- Component for the cheat mod.
- """
+ """Component for the cheat mod."""
pass
class MainMenu(Component):
- """
- Component for main menu items.
- """
+ """Component for main menu items."""
pass
class GameMenu(Component):
- """
- Component for game menu items.
- """
+ """Component for game menu items."""
pass
class GameGui(Component):
- """
- Component for game graphical user interface items.
- """
+ """Component for game graphical user interface items."""
pass
class PacGum(Component):
- """
- Component for pacgums items.
- """
+ """Component for pacgums items."""
pass
class SuperPacGum(Component):
- """
- Component for super pacgums items.
- """
+ """Component for super pacgums items."""
pass
@dataclass
class Pacman(Component):
- """
- Component for pacman items storing the home coordinates and entity.
- """
+ """Component for pacman items storing the home coordinates and entity."""
home: tuple[int, int]
home_entity: Entity
@dataclass
class GhostStatus(Component):
- """
- Component for ghost state and cycle.
- """
+ """Component for ghost state and cycle."""
tick: int = 0
cycle_wander: int = 60 * 10
@dataclass
class Ghost(Component):
- """
- Component for ghost items storing the ghost variant and the home
+ """Component for ghost items storing the ghost variant and the home
coordinates and entity.
"""
@dataclass
class TargetTile(Component):
- """
- Component for the coordinates of a target used for ghost movement.
- """
+ """Component for the coordinates of a target used for ghost movement."""
pos: tuple[int, int]
@dataclass
class TilePos(Component):
- """
- Component tile position as entity.
- """
+ """Component tile position as entity."""
pos: Entity
@dataclass
class InMotion(Component):
- """
- Component for pacman or ghost movement storing the position, the provenance
+ """Component for pacman or ghost movement storing the position, the provenance
and the direction.
"""
@dataclass
class MoveSpeed(Component):
- """
- Component for movement storing the movement speed by pixels per frames.
- """
+ """Component for movement storing the movement speed by pixels per frames."""
pixels_per_frame: float
class Maze(Component):
- """
- Component for maze storing the x and y dimension and the cells in a
+ """Component for maze storing the x and y dimension and the cells in a
dictionnary (coordonate entity).
"""
@dataclass
class Cell(Component):
- """
- Component for the cells of the maze storing the value and the position.
- """
+ """Component for the cells of the maze storing the value and the position."""
n: int
grid_pos: tuple[int, int]
def neighbor(self, idx: int) -> tuple[int, int]:
- """
- Get neighbor position by direction (1: up, 2: right ...).
- """
+ """Get neighbor position by direction (1: up, 2: right ...)."""
dx, dy = _DIR_OFFSETS[idx]
return (self.grid_pos[0] + dx, self.grid_pos[1] + dy)
def walls(self) -> list[bool]:
- """
- Get list of walls.
- """
+ """Get list of walls."""
return [(1 & (self.n >> i)) == 1 for i in range(4)]
def remove_comment(s: str) -> str:
- """
- Return s with all comment lines removed.
- """
+ """Return s with all comment lines removed."""
return "\n".join(line for line in s.split("\n") if not s.startswith("#"))
def load_config(world: World) -> None:
- """
- Load the configuration json file given in first argument and spawn the
+ """Load the configuration json file given in first argument and spawn the
Config resource.
"""
if len(sys.argv) != 2:
def load_highscores(world: World, config: Config) -> None:
- """
- Load the highscores json file and spawn the Highscores resource.
- """
+ """Load the highscores json file and spawn the Highscores resource."""
try:
with open(config.highscores_path) as f:
scores = Highscores.model_validate_json(f.read())
def save_highscores(config: Config, scores: Highscores) -> None:
- """
- Write the json highscores file from the Highscores resource.
- """
+ """Write the json highscores file from the Highscores resource."""
path = config.highscores_path
with open(path, "w+") as f:
f.write(scores.model_dump_json(indent=4))
def create_maze(
world: World, assets: Assets[Sprite2D]
) -> Callable[[MazeGenerator], Entity]:
- """
- System creating and adding the maze to the world.
- """
+ """System creating and adding the maze to the world."""
def inner(maze: MazeGenerator) -> Entity:
root_maze = Maze(len(maze.maze[0]), len(maze.maze))
def spawn_superpacgums(
world: World, mazes: Query[Maze], assets: Assets[Sprite2D]
) -> None:
- """
- System creating and adding super pacgums to the maze.
- """
+ """System creating and adding super pacgums to the maze."""
pacgums = []
for maze in mazes:
for pos in [
superpacgums: Query[ChildOf, SuperPacGum],
assets: Assets[Sprite2D],
) -> None:
- """
- System creating and adding pacgums to the maze.
- """
+ """System creating and adding pacgums to the maze."""
ignore = {e.parent for e in superpacgums}
pacgums = []
for ent, cell in cells:
def setup_maze(world: World) -> None:
- """
- System running the 'create_maze' system with a give MazeGenerator.
- """
+ """System running the 'create_maze' system with a give MazeGenerator."""
world.run_system(create_maze, MazeGenerator((14, 14)))
entities: Query[tuple[InMotion, Coord2D, MoveSpeed]],
positions: Query[Coord2D],
) -> None:
- """
- System enabling movement operation for entity wit InMotion component.
- """
+ """System enabling movement operation for entity wit InMotion component."""
for motion, pos, speed in entities:
if (dst := positions.get(motion.pos)) is None:
continue
entities: Query[tuple[InMotion, Coord2D, TilePos]],
cells: Query[Coord2D],
) -> None:
- """
- System updating the tile position for entity wit InMotion component.
- """
+ """System updating the tile position for entity wit InMotion component."""
for motion, coord, pos in entities:
src = cells[motion.src]
dst = cells[motion.pos]
cells: Query[Cell],
config: Config,
) -> None:
- """
- System creating and adding pacman to the maze.
- """
+ """System creating and adding pacman to the maze."""
if (maze_t := maze_q.single()) is None:
return
maze, maze_ent = maze_t
maze_q: Query[tuple[Maze, Entity]],
config: Config,
) -> None:
- """
- System creating and adding ghosts to the maze.
- """
+ """System creating and adding ghosts to the maze."""
if (maze_t := maze_q.single()) is None:
return
maze, maze_ent = maze_t
cells: Query[tuple[Cell, Coord2D]],
maze_q: Query[Maze],
) -> None:
- """
- System giving target movement to pacman with the input given by the user.
- """
+ """System giving target movement to pacman with the input given by the user."""
if (maze := maze_q.single()) is None:
return
for coord, motion, pos in pacman_q:
score: GameState,
config: Config,
) -> None:
- """
- System managing the collision and collection of pacgum by pacman.
- """
+ """System managing the collision and collection of pacgum by pacman."""
collided = [
pacgum
for coord, pacgum in pacgums
score: GameState,
config: Config,
) -> None:
- """
- System managing the collision and collection of superpacgum by pacman and
+ """System managing the collision and collection of superpacgum by pacman and
starting the ghost flee state if a superpacgum is collected.
"""
collided = [
def respawn_pacman(pacmans: Query[tuple[Pacman, Coord2D, InMotion]]) -> None:
- """
- System reseting pacman position when touched by ghost.
- """
+ """System reseting pacman position when touched by ghost."""
for pacman, pos, motion in pacmans:
pos.xy = Vector2(*pacman.home).elementwise() * 64
motion.pos = pacman.home_entity
def respawn_ghost(ghosts: Query[tuple[Ghost, Coord2D, InMotion]]) -> None:
- """
- System reseting ghost fleeing position when touched by pacman.
- """
+ """System reseting ghost fleeing position when touched by pacman."""
for ghost, pos, motion in ghosts:
pos.xy = Vector2(*ghost.home).elementwise() * 64
motion.pos = ghost.home_entity
state: GameState,
config: Config,
) -> None:
- """
- System managing the collision between by pacman and ghost.
- """
+ """System managing the collision between by pacman and ghost."""
for pacman_pos in pacmans:
for ghost_pos, status, ghost, motion, pos in ghosts:
if (
continue
match status.state:
case GhostState.wander | GhostState.chase:
- if world.run_raw_system(cheat_mode_enabled):
+ if world.run_simple_system(cheat_mode_enabled):
continue
- world.run_raw_system(respawn_pacman)
- world.run_raw_system(respawn_ghost)
- world.run_raw_system(start_pause_timer)
+ world.run_simple_system(respawn_pacman)
+ world.run_simple_system(respawn_ghost)
+ world.run_simple_system(start_pause_timer)
state.lives -= 1
break
cells: Query[tuple[Cell, Coord2D]],
maze_q: Query[Maze],
) -> None:
- """
- System giving target movement to ghost.
- """
+ """System giving target movement to ghost."""
if (maze := maze_q.single()) is None:
return
for target, motion, pos, status in entities:
pacman_q: Query[tuple[InMotion, TilePos], Pacman],
cells: Query[Cell],
) -> None:
- """
- System updating target to ghost.
- """
+ """System updating target to ghost."""
player = pacman_q.single()
if player is None:
return
def update_chase_status(status_q: Query[GhostStatus], config: Config) -> None:
- """
- System updating ghost status and cycle.
- """
+ """System updating ghost status and cycle."""
for status in status_q:
target_time: int
next_state: GhostState
def update_ghost_speed(
ghost: Query[tuple[MoveSpeed, GhostStatus]], config: Config
) -> None:
- """
- System updating ghost speed according to his state.
- """
+ """System updating ghost speed according to his state."""
for speed, status in ghost:
match status.state:
case GhostState.chase | GhostState.wander:
def update_pacman_anim(
pacman_q: Query[tuple[InMotion, AnimationSet], Pacman],
) -> None:
- """
- System updating pacman animation according to his direction.
- """
+ """System updating pacman animation according to his direction."""
for motion, anim in pacman_q:
anim.idx = motion.direction
ghost_q: Query[tuple[InMotion, AnimationSet, GhostStatus], Ghost],
config: Config,
) -> None:
- """
- System updating pacman animation according to his direction and state.
- """
+ """System updating pacman animation according to his direction and state."""
for motion, anim, status in ghost_q:
match status.state:
case GhostState.chase | GhostState.wander:
def start_pause_timer(timer: "InitialPauseTimer") -> None:
- """
- System starting the initial pause timer.
- """
+ """System starting the initial pause timer."""
timer.tick = 0
def setup_new_maze(world: World) -> None:
- """
- System setting up the maze by running 'setup_maze', 'spawn_superpacgums',
+ """System setting up the maze by running 'setup_maze', 'spawn_superpacgums',
'spawn_pacgums', 'spawn_pacman', 'spawn_ghosts' and
'start_pause_timer' systems.
"""
spawn_ghosts,
start_pause_timer,
):
- world.run_raw_system(system)
+ world.run_simple_system(system)
def maze_exists(itr: Query[Maze]) -> bool:
- """
- System cheking maze existance.
- """
+ """System cheking maze existance."""
return next(itr.__iter__(), None) is not None
def in_game(world: World) -> bool:
- """
- System cheking that the game is running.
- """
- return world.run_raw_system(maze_exists) and not world.run_raw_system(
- in_game_menu
- )
+ """System cheking that the game is running."""
+ return world.run_simple_system(
+ maze_exists
+ ) and not world.run_simple_system(in_game_menu)
def level_is_finished(
pacgums: Query[SuperPacGum | PacGum], _: Not[GameIsOver]
) -> bool:
- """
- System cheking the level completion.
- """
+ """System cheking the level completion."""
return next(iter(pacgums), None) is None
queue: CommandQueue,
state: GameState,
) -> None:
- """
- System called when level is completed, generating a new level or finishing
+ """System called when level is completed, generating a new level or finishing
the game if all levels are completed.
"""
for maze in mazes:
queue.queue(hierarchy.despawn_with_children(maze))
if state.level == 2:
- queue.queue(lambda world: world.run_raw_system(game_over))
+ queue.queue(lambda world: world.run_simple_system(game_over))
return
state.level += 1
state.lives += 1
def level_is_lost(
config: Config, state: GameState, _: Not[GameIsOver]
) -> bool:
- """
- System cheking the level is lost.
- """
+ """System cheking the level is lost."""
return config.level_timer < state.timer or state.lives <= 0
mazes: Query[Entity, Maze],
queue: CommandQueue,
) -> None:
- """
- System called when the player lose the game.
- """
+ """System called when the player lose the game."""
world.despawn_resource(GameState)
world.despawn_many(list(gui))
for maze in mazes:
queue.queue(hierarchy.despawn_with_children(maze))
- queue.queue(lambda world: world.add_raw_system(game_over)())
- world.run_raw_system(spawn_game_over)
+ queue.queue(lambda world: world.add_simple_system(game_over)())
+ world.run_simple_system(spawn_game_over)
def game_over(queue: CommandQueue) -> None:
# FIXME
- """
- System called when game is finish.
- """
+ """System called when game is finish."""
queue.queue(lambda world: world.spawn_resource(GameIsOver()))
queue.queue(lambda world: world.spawn_resource(PlayerName()))
def update_game_timer(world: World, state: GameState) -> None:
- """
- System updating the levels timer.
- """
- if world.run_raw_system(cheat_mode_enabled):
+ """System updating the levels timer."""
+ if world.run_simple_system(cheat_mode_enabled):
return
state.timer += 1
def initial_pause_done(timer: InitialPauseTimer, config: Config) -> bool:
- """
- System cheking if the inital timer is done.
- """
+ """System cheking if the inital timer is done."""
if timer.tick < config.initial_pos_timer:
timer.tick += 1
return False
def print_score(state: GameState) -> str:
- """
- Return a string with the player score.
- """
+ """Return a string with the player score."""
return f"Score: {state.score}"
def print_lives(state: GameState) -> str:
- """
- Return a string with the player lives.
- """
+ """Return a string with the player lives."""
return f"Lives: {state.lives}"
def print_timer(state: GameState, config: Config) -> str:
- """
- Return a string with the level timer.
- """
+ """Return a string with the level timer."""
time = config.level_timer - state.timer
return f"Time left: {time // 60}"
def spawn_hud(world: World, fonts: Assets[Font]) -> None:
- """
- System that print the Heads-Up Display in game with lives, levels and timer
+ """System that print the Heads-Up Display in game with lives, levels and timer
informations.
"""
world.spawn_many(
hierarchy: Hierarchy[tuple[()]],
menus: Query[Entity, MainMenu],
) -> None:
- """
- System closing main menu and starting the game and Heads-Up Display.
- """
+ """System closing main menu and starting the game and Heads-Up Display."""
for menu in menus:
hierarchy.despawn_with_children(menu)(world)
world.despawn_resource(SelectedButton)
world.spawn_resource(GameState())
- world.run_raw_system(setup_new_maze)
- world.run_raw_system(spawn_hud)
+ world.run_simple_system(setup_new_maze)
+ world.run_simple_system(spawn_hud)
def quit_game(world: World) -> None:
- """
- System closing the game and quiting the program.
- """
+ """System closing the game and quiting the program."""
world.spawn_resource(WorldShouldExit())
def spawn_button(
world: World, fonts: Assets[Font], sprites: Assets[Sprite2D]
) -> Callable[[WindowPositionned, str, System], Entity]:
- """
- System creating and adding a button to the world.
- """
+ """System creating and adding a button to the world."""
def inner(pos: WindowPositionned, s: str, cb: System) -> Entity:
parent = world.spawn(
def enable_cheat_mode(world: World) -> None:
- world.run_raw_system(spawn_cheat_info)
+ world.run_simple_system(spawn_cheat_info)
world.spawn(CheatMode, CheatMode())
def toggle_cheat_mode(world: World) -> None:
- if world.run_raw_system(cheat_mode_enabled):
- world.run_raw_system(disable_cheat_mode)
+ if world.run_simple_system(cheat_mode_enabled):
+ world.run_simple_system(disable_cheat_mode)
else:
- world.run_raw_system(enable_cheat_mode)
+ world.run_simple_system(enable_cheat_mode)
def close_current_game(
gui: Query[Entity, GameGui],
queue: CommandQueue,
) -> None:
- """
- System removing maze with childrens and game graphical user interface.
- """
- world.run_raw_system(close_game_menu)
+ """System removing maze with childrens and game graphical user interface."""
+ world.run_simple_system(close_game_menu)
world.despawn_resource(GameState)
world.despawn_many(list(gui))
for maze in mazes:
queue.queue(hierarchy.despawn_with_children(maze))
- if world.run_raw_system(cheat_mode_enabled):
- world.run_raw_system(disable_cheat_mode)
+ if world.run_simple_system(cheat_mode_enabled):
+ world.run_simple_system(disable_cheat_mode)
def back_to_main_menu(world: World) -> None:
- """
- System closing the current game and opening the main menu.
- """
- world.run_raw_system(close_current_game)
- world.run_raw_system(setup_main_menu)
+ """System closing the current game and opening the main menu."""
+ world.run_simple_system(close_current_game)
+ world.run_simple_system(setup_main_menu)
def in_game_menu(menus: Query[GameMenu]) -> bool:
- """
- System cheking if the main menu is running.
- """
+ """System cheking if the main menu is running."""
return next(menus.__iter__(), None) is not None
def toggle_game_menu(world: World) -> None:
- """
- System opening or closing the game menu.
- """
- if world.run_raw_system(in_game_menu):
- world.run_raw_system(close_game_menu)
+ """System opening or closing the game menu."""
+ if world.run_simple_system(in_game_menu):
+ world.run_simple_system(close_game_menu)
else:
- world.run_raw_system(setup_game_menu)
+ world.run_simple_system(setup_game_menu)
def close_game_menu(
menus: Query[Entity, GameMenu],
animated: Query[Entity, Pacman | Ghost],
) -> None:
- """
- System closing the game menu.
- """
+ """System closing the game menu."""
world.remove_many(StopAnimation, list(animated))
for menu in menus:
hierarchy.despawn_with_children(menu)(world)
def setup_game_menu(
world: World, animated: Query[Entity, Pacman | Ghost]
) -> None:
- """
- System opening the game menu.
- """
+ """System opening the game menu."""
world.insert_many(StopAnimation, [(e, StopAnimation()) for e in animated])
selected = world.run_system(
spawn_button,
def set_up_instructions(world: World, fonts: Assets[Font]) -> list[Entity]:
- """
- System adding instructions to main menu.
- """
+ """System adding instructions to main menu."""
text = """Use 'WASD'to move pacman in the maze.
Collect all pacgums to go to the next level.
Complete all levels to win the game.
def set_up_highsores(
world: World, scores: Highscores, fonts: Assets[Font]
) -> list[Entity]:
- """
- System adding highscores to main menu.
- """
+ """System adding highscores to main menu."""
highscores = [(name, value) for name, value in scores.scores.items()]
highscores.sort(key=lambda x: x[1], reverse=True)
entities = [
def setup_main_menu(world: World) -> None:
- """
- System creating the main menu by adding button, instruction and highscores.
- """
+ """System creating the main menu by adding button, instruction and highscores."""
selected = world.run_system(
spawn_button,
WindowPositionned(Vector2(0.5), Vector2(-150, -100)),
)
)
- entities.extend(world.run_raw_system(set_up_highsores))
- entities.extend(world.run_raw_system(set_up_instructions))
+ entities.extend(world.run_simple_system(set_up_highsores))
+ entities.extend(world.run_simple_system(set_up_instructions))
world.spawn(
tuple[MainMenu, Children, CoordBundle],
def run_game() -> None:
- """
- Start with all necessary plugins and system to run the game.
- """
+ """Start with all necessary plugins and system to run the game."""
World().with_plugins(graphics_plugins, minimal_plugins).with_systems(
StartupSchedule,
Systems(load_config, load_highscores, setup_main_menu)
+"""General utilities used mostly in the game engine."""
+"""Utilities to handle exceptions in lambas."""
+
from typing import Never
def panic(e: Exception) -> Never:
+ """Raises the exception, in a way that is usable in lambdas."""
raise e
# ruff: noqa: ANN401
+"""Utilities to resolve type aliases and generics."""
+
import types
import typing
from collections.abc import Iterable
def map_type_generics(ty: Any, mapping: dict[Any, type]) -> Any:
- """Map generics for this type, maybe creating a new type."""
+ """Resolve generics for this type, using the mapping."""
if ty in mapping:
return mapping[ty]
if (orig := get_origin(ty)) is None:
def resolve_type_aliases[T](ty: Any) -> Any:
- """Resolve any type aliases for a given type, recursively."""
+ """Resolve type aliases for a given type, recursively."""
while hasattr(ty, "__value__"):
if not (
hasattr(ty, "__type_params__")
def origin_or_cls(ty: Any) -> Any:
+ """Return the base class of ty, without any generics if they exist."""
res = get_origin(ty)
if res is None:
res = ty
-from collections.abc import Callable, Generator
-from typing import cast
-
-type RecursiveTuple[T] = T | tuple[RecursiveTuple[T], ...]
-
-
-def recursive_tuple_map[T, U](
- tup: RecursiveTuple[T], map: Callable[[T], U]
-) -> RecursiveTuple[U]:
- if not isinstance(tup, tuple):
- return map(tup)
- return tuple(recursive_tuple_map(e, map) for e in tup)
+"""Simple utilities to work with variadic generic tuples efficiently."""
+from collections.abc import Callable
-def recursive_tuple_create[T, U](
- tup: T, map: Callable[[T], U | tuple[T, ...]]
-) -> RecursiveTuple[U]:
- if not isinstance((res := map(tup)), tuple):
- return res
- return tuple(
- cast(RecursiveTuple[U], recursive_tuple_create(tup, map))
- for tup in res
- )
-
-
-def recursive_tuple_iter[T](tup: RecursiveTuple[T]) -> Generator[T]:
- if not isinstance(tup, tuple):
- yield tup
- else:
- for sub in cast(tuple[RecursiveTuple[T], ...], tup):
- yield from recursive_tuple_iter(sub)
+type RecursiveTuple[T] = T | tuple[RecursiveTuple[T], ...]
def fast_tuple_constructor[T, U](
cbs: tuple[Callable[[U], T], ...],
) -> Callable[[U], tuple[T, ...]]:
- match cbs:
- case (e1,):
- return lambda arg: (e1(arg),)
- case (e1, e2):
- return lambda arg: (e1(arg), e2(arg))
- case (e1, e2, e3):
- return lambda arg: (e1(arg), e2(arg), e3(arg))
- case (e1, e2, e3, e4):
- return lambda arg: (e1(arg), e2(arg), e3(arg), e4(arg))
- case (e1, e2, e3, e4, e5):
- return lambda arg: (e1(arg), e2(arg), e3(arg), e4(arg), e5(arg))
- case (e1, e2, e3, e4, e5, e6):
- return lambda arg: (
- e1(arg),
- e2(arg),
- e3(arg),
- e4(arg),
- e5(arg),
- e6(arg),
- )
- case (e1, e2, e3, e4, e5, e6, e7):
- return lambda arg: (
- e1(arg),
- e2(arg),
- e3(arg),
- e4(arg),
- e5(arg),
- e6(arg),
- e7(arg),
- )
- case (e1, e2, e3, e4, e5, e6, e7, e8):
- return lambda arg: (
- e1(arg),
- e2(arg),
- e3(arg),
- e4(arg),
- e5(arg),
- e6(arg),
- e7(arg),
- e8(arg),
- )
- case (e1, e2, e3, e4, e5, e6, e7, e8, e9):
- return lambda arg: (
- e1(arg),
- e2(arg),
- e3(arg),
- e4(arg),
- e5(arg),
- e6(arg),
- e7(arg),
- e8(arg),
- e9(arg),
- )
- case (e1, e2, e3, e4, e5, e6, e7, e8, e9, e10):
- return lambda arg: (
- e1(arg),
- e2(arg),
- e3(arg),
- e4(arg),
- e5(arg),
- e6(arg),
- e7(arg),
- e8(arg),
- e9(arg),
- e10(arg),
- )
- case (e1, e2, e3, e4, e5, e6, e7, e8, e9, e10, e11):
- return lambda arg: (
- e1(arg),
- e2(arg),
- e3(arg),
- e4(arg),
- e5(arg),
- e6(arg),
- e7(arg),
- e8(arg),
- e9(arg),
- e10(arg),
- e11(arg),
- )
- case (e1, e2, e3, e4, e5, e6, e7, e8, e9, e10, e11, e12):
- return lambda arg: (
- e1(arg),
- e2(arg),
- e3(arg),
- e4(arg),
- e5(arg),
- e6(arg),
- e7(arg),
- e8(arg),
- e9(arg),
- e10(arg),
- e11(arg),
- e12(arg),
- )
- case (e1, e2, e3, e4, e5, e6, e7, e8, e9, e10, e11, e12, e13):
- return lambda arg: (
- e1(arg),
- e2(arg),
- e3(arg),
- e4(arg),
- e5(arg),
- e6(arg),
- e7(arg),
- e8(arg),
- e9(arg),
- e10(arg),
- e11(arg),
- e12(arg),
- e13(arg),
- )
- case (e1, e2, e3, e4, e5, e6, e7, e8, e9, e10, e11, e12, e13, e14):
- return lambda arg: (
- e1(arg),
- e2(arg),
- e3(arg),
- e4(arg),
- e5(arg),
- e6(arg),
- e7(arg),
- e8(arg),
- e9(arg),
- e10(arg),
- e11(arg),
- e12(arg),
- e13(arg),
- e14(arg),
- )
- case (
- e1,
- e2,
- e3,
- e4,
- e5,
- e6,
- e7,
- e8,
- e9,
- e10,
- e11,
- e12,
- e13,
- e14,
- e15,
- ):
- return lambda arg: (
- e1(arg),
- e2(arg),
- e3(arg),
- e4(arg),
- e5(arg),
- e6(arg),
- e7(arg),
- e8(arg),
- e9(arg),
- e10(arg),
- e11(arg),
- e12(arg),
- e13(arg),
- e14(arg),
- e15(arg),
- )
- case e:
- return lambda arg: tuple(cb(arg) for cb in e)
+ """Map a tuple of callables to a callable of tuples."""
+ return lambda arg: tuple(cb(arg) for cb in cbs)
def composed_by[T, U](
b: Callable[[T], U],
op: Callable[[U, U], U],
) -> Callable[[T], U]:
+ """Compose two functions together, calling both on the same argument."""
if a is None:
return b
return lambda i: op(a(i), b(i))
projects / axy / ft / pacman.git / commitdiff