-from dataclasses import dataclass
from pathlib import Path
-from typing import override
import pygame
-from pygame import Vector3
-from pacman.ecs.assets import Asset, Assets
+from pacman.ecs.assets import Assets
+from pacman.ecs.coord import (
+ Coord2D,
+ CoordAbs2D,
+ coord_update_plugins,
+)
+from pacman.ecs.graphics import ShouldRender, Sprite2D, graphics_plugins
from pacman.ecs.hierarchy import (
ChildOf,
- Hierarchy,
hierachy_plugins,
)
from pacman.ecs.schedule import (
- MainSchedule,
StartupSchedule,
- SystemSet,
UpdateSchedule,
)
+from pacman.ecs.update import schedule_plugins
from pacman.ecs.world import (
Component,
- Not,
Query,
- Resource,
- Systems,
World,
- WorldShouldExit,
)
-class WorldIsInit(Resource):
- pass
-
-
-def run_startup(world: World, _: Not[WorldIsInit]) -> None:
- world.tick(StartupSchedule)
- world.spawn(WorldIsInit, WorldIsInit())
-
-
-def run_update(world: World, _: WorldIsInit) -> None:
- world.tick(UpdateSchedule)
-
-
-PreUpdate = SystemSet("preupdate")
-Update = SystemSet("update")
-PostUpdate = SystemSet("postupdate")
-
-
-def schedule_plugins(world: World) -> None:
- world.with_systems(MainSchedule, run_startup, run_update).with_systems(
- UpdateSchedule, Systems(PreUpdate, Update, PostUpdate).chain()
- )
-
-
-class CoordAbs2D(Component):
- _vec: Vector3
-
- def __init__(self) -> None:
- self._vec = Vector3(0.0)
-
-
-class Coord2D(Vector3, Component):
- pass
-
-
-def update_coord(entities: Query[tuple[CoordAbs2D, Coord2D]]) -> None:
- for abs_coord, coord in entities:
- abs_coord._vec = coord.xyz
-
-
-def update_coord_hierarchy(hierarchy: Hierarchy[CoordAbs2D]) -> None:
- for parent, child in hierarchy.top_down():
- if parent is None:
- continue
- child._vec += parent._vec
-
-
-PropagateTransform = SystemSet("propagate-transform")
-
-
-def coord_update_plugins(world: World) -> None:
- world.with_systems(
- UpdateSchedule,
- Systems(update_coord, update_coord_hierarchy)
- .chain()
- .in_set(PostUpdate, PropagateTransform),
- )
- world.with_systems(UpdateSchedule, update_coord)
-
-
def minimal_plugins(world: World) -> None:
world.with_plugins(
schedule_plugins, hierachy_plugins, coord_update_plugins
)
-class ShouldRender(Component):
- pass
-
-
-@dataclass
-class Window(Resource):
- surface: pygame.Surface
- clock: pygame.time.Clock
-
-
-@dataclass
-class WindowEvents(Resource):
- events: list[pygame.event.Event]
-
-
-def init_window(world: World) -> None:
- pygame.init()
- world.spawn(Window, Window(pygame.display.set_mode(), pygame.time.Clock()))
- world.spawn(WindowEvents, WindowEvents([]))
-
-
-@dataclass
-class Sprite2D(Component, Asset):
- surface: pygame.Surface
-
- @override
- @classmethod
- def load(cls, ty: type, path: str) -> "Sprite2D":
- del ty
- return Sprite2D(pygame.image.load(path))
-
-
-def poll_events(events: WindowEvents) -> None:
- events.events = pygame.event.get()
-
-
-def close_on_close(world: World, events: WindowEvents) -> None:
- if pygame.QUIT in (event.type for event in events.events):
- world.spawn(WorldShouldExit, WorldShouldExit())
-
-
-def render_sprite2d(
- window: Window,
- sprites_q: Query[tuple[CoordAbs2D, Sprite2D], ShouldRender],
-) -> None:
- sprites_raw = sorted(
- sprites_q,
- key=lambda e: e[0]._vec.z,
- reverse=True,
- )
- sprites = [
- (sprite.surface, coord._vec.xy) for coord, sprite in sprites_raw
- ]
- window.surface.fill((0, 0, 0))
- window.surface.blits(sprites)
- window.clock.tick(60)
- pygame.display.flip()
-
-
-def graphics_plugins(world: World) -> None:
- world.with_systems(StartupSchedule, init_window).with_systems(
- UpdateSchedule,
- Systems(render_sprite2d).in_set(PostUpdate).after(PropagateTransform),
- Systems(poll_events, close_on_close).chain().in_set(PreUpdate),
- )
-
-
class WiggleSprite(Component):
pass
for i, asset_path in enumerate(asset_paths)
],
)
- pass
if __name__ == "__main__":
from collections.abc import Callable
from os import PathLike
from os.path import abspath
-from typing import Self, cast, get_args, override
+from typing import Self, get_args, override
from pacman.ecs.world import Resource, World
from pacman.utils.type_resolve import resolve_type_aliases
@override
@classmethod
- def extract_from_world(
+ def extract_default(
cls, ty: type, world: World
) -> Callable[[], Self | None]:
- inner = cast(
- Callable[[], Self | None], Resource.extract_from_world(ty, world)
- )
- return lambda: cast(
- Self, (res if (res := inner()) is not None else Assets(ty))
- )
+ del world
+ return lambda: cls(ty)
--- /dev/null
+from pygame import Vector3
+
+from pacman.ecs.hierarchy import Hierarchy
+from pacman.ecs.schedule import SystemSet, UpdateSchedule
+from pacman.ecs.update import PostUpdate
+from pacman.ecs.world import Component, Query, Systems, World
+
+
+class CoordAbs2D(Component):
+ _vec: Vector3
+
+ def __init__(self) -> None:
+ self._vec = Vector3(0.0)
+
+
+class Coord2D(Vector3, Component):
+ pass
+
+
+def update_coord(entities: Query[tuple[CoordAbs2D, Coord2D]]) -> None:
+ for abs_coord, coord in entities:
+ abs_coord._vec = coord.xyz
+
+
+def update_coord_hierarchy(hierarchy: Hierarchy[CoordAbs2D]) -> None:
+ for parent, child in hierarchy.top_down():
+ if parent is None:
+ continue
+ child._vec += parent._vec
+
+
+PropagateTransform = SystemSet("propagate-transform")
+
+
+def coord_update_plugins(world: World) -> None:
+ world.with_systems(
+ UpdateSchedule,
+ Systems(update_coord, update_coord_hierarchy)
+ .chain()
+ .in_set(PostUpdate, PropagateTransform),
+ )
+ world.with_systems(UpdateSchedule, update_coord)
--- /dev/null
+from dataclasses import dataclass
+from typing import override
+
+import pygame
+
+from pacman.ecs.assets import Asset
+from pacman.ecs.coord import CoordAbs2D, PropagateTransform
+from pacman.ecs.schedule import StartupSchedule, UpdateSchedule
+from pacman.ecs.update import PostUpdate, PreUpdate
+from pacman.ecs.world import (
+ Component,
+ Query,
+ Resource,
+ Systems,
+ World,
+ WorldShouldExit,
+)
+
+
+class ShouldRender(Component):
+ pass
+
+
+@dataclass
+class Window(Resource):
+ surface: pygame.Surface
+ clock: pygame.time.Clock
+
+
+@dataclass
+class WindowEvents(Resource):
+ events: list[pygame.event.Event]
+
+
+def init_window(world: World) -> None:
+ pygame.init()
+ world.spawn(Window, Window(pygame.display.set_mode(), pygame.time.Clock()))
+ world.spawn(WindowEvents, WindowEvents([]))
+
+
+@dataclass
+class Sprite2D(Component, Asset):
+ surface: pygame.Surface
+
+ @override
+ @classmethod
+ def load(cls, ty: type, path: str) -> "Sprite2D":
+ del ty
+ return Sprite2D(pygame.image.load(path))
+
+
+def poll_events(events: WindowEvents) -> None:
+ events.events = pygame.event.get()
+
+
+def close_on_close(world: World, events: WindowEvents) -> None:
+ if pygame.QUIT in (event.type for event in events.events):
+ world.spawn(WorldShouldExit, WorldShouldExit())
+
+
+def render_sprite2d(
+ window: Window,
+ sprites_q: Query[tuple[CoordAbs2D, Sprite2D], ShouldRender],
+) -> None:
+ sprites_raw = sorted(
+ sprites_q,
+ key=lambda e: e[0]._vec.z,
+ reverse=True,
+ )
+ sprites = [
+ (sprite.surface, coord._vec.xy) for coord, sprite in sprites_raw
+ ]
+ window.surface.fill((0, 0, 0))
+ window.surface.blits(sprites)
+ window.clock.tick(60)
+ pygame.display.flip()
+
+
+def graphics_plugins(world: World) -> None:
+ world.with_systems(StartupSchedule, init_window).with_systems(
+ UpdateSchedule,
+ Systems(render_sprite2d).in_set(PostUpdate).after(PropagateTransform),
+ Systems(poll_events, close_on_close).chain().in_set(PreUpdate),
+ )
--- /dev/null
+from pacman.ecs.schedule import (
+ MainSchedule,
+ StartupSchedule,
+ SystemSet,
+ UpdateSchedule,
+)
+from pacman.ecs.world import Not, Resource, Systems, World
+
+
+class WorldIsInit(Resource):
+ pass
+
+
+def run_startup(world: World, _: Not[WorldIsInit]) -> None:
+ world.tick(StartupSchedule)
+ world.spawn(WorldIsInit, WorldIsInit())
+
+
+def run_update(world: World, _: WorldIsInit) -> None:
+ world.tick(UpdateSchedule)
+
+
+PreUpdate = SystemSet("preupdate")
+Update = SystemSet("update")
+PostUpdate = SystemSet("postupdate")
+
+
+def schedule_plugins(world: World) -> None:
+ world.with_systems(MainSchedule, run_startup, run_update).with_systems(
+ UpdateSchedule, Systems(PreUpdate, Update, PostUpdate).chain()
+ )
itr: Iterator[Items] = self.__iter__()
res = next(itr, None)
if next(itr, None) is not None:
- return None
+ raise Exception("More than one entity matches single query")
return res
query: Callable[[], Query[Self]] = cast(
Callable[[], Query[Self]], _query.extract_from_world(_query, world)
)
- return lambda: query().single()
+ default: Callable[[], Self | None] = cls.extract_default(ty, world)
+ return lambda: (
+ res if (res := query().single()) is not None else default()
+ )
+
+ @classmethod
+ def extract_default(
+ cls, ty: type, world: "World"
+ ) -> Callable[[], Self | None]:
+ del ty, world
+ return lambda: None
class CommandQueue(FromWorld):
projects / axy / ft / pacman.git / commitdiff