import inspect
from abc import ABC, abstractmethod
from collections.abc import Callable, Container, Iterable, Iterator
+from dataclasses import dataclass
from typing import (
Any,
Literal,
+ NewType,
Self,
Union,
cast,
from pacman.utils.type_resolve import origin_or_cls, resolve_type_aliases
from pacman.utils.variadics_please import (
RecursiveTuple,
+ composed_by,
fast_tuple_constructor,
)
yield from cast(type[Bundle], components).components_type_iter()
+ComponentID = NewType("ComponentID", int)
+
+
class Archetype:
- def __init__(
- self, world: "World", types: Iterable[type["Component"]]
- ) -> None:
+ def __init__(self, world: "World", types: Iterable[ComponentID]) -> None:
self._world = world
self._types = frozenset(types)
self._positions: dict[Entity, int] = {}
self._entities: list[Entity] = []
- self._components: dict[type[Component], list[Any]] = {
+ self._components: dict[ComponentID, list[Any]] = {
ty: [] for ty in types
}
self._add_cache: dict[
- int, tuple[Any, Archetype, list[type[Component]]]
+ int, tuple[Any, Archetype, list[ComponentID]]
] = {}
self._remove_cache: dict[
- int, tuple[Any, Archetype, list[type[Component]]]
+ int, tuple[Any, Archetype, list[ComponentID]]
] = {}
self._world._archetypes.append(self)
def _compute_add(
self, ty: type[Components]
- ) -> tuple["Archetype", list[type["Component"]]]:
+ ) -> tuple["Archetype", list[ComponentID]]:
ptr = id(ty)
if res := self._add_cache.get(ptr):
return (res[1], res[2])
- comps = list(components_type_iter(resolve_type_aliases(ty)))
+ comps = list(
+ map(
+ self._world.component_id,
+ components_type_iter(resolve_type_aliases(ty)),
+ )
+ )
res_arch: Archetype
types = frozenset(comps) | self._types
if prev := self._world._archetypes_lookup.get(types):
def _compute_remove(
self, ty: type[Components]
- ) -> tuple["Archetype", list[type["Component"]]]:
+ ) -> tuple["Archetype", list[ComponentID]]:
ptr = id(ty)
if res := self._remove_cache.get(ptr):
return (res[1], res[2])
- comps = list(components_type_iter(resolve_type_aliases(ty)))
+ comps = list(
+ map(
+ self._world.component_id,
+ components_type_iter(resolve_type_aliases(ty)),
+ )
+ )
res_arch: Archetype
types = self._types - frozenset(comps)
if prev := self._world._archetypes_lookup.get(types):
for k in comps_ty:
kv[k] = [next(e) for e in comps]
- def cb(ty: type[Entity] | type[Component]) -> Iterable[Any]:
- if issubclass(ty, Entity):
+ def cb(ty: ComponentID | None) -> Iterable[Any]:
+ if ty is None:
return (e for e, _ in components)
if ty in kv:
return kv[ty]
dst, _ = self._compute_remove(ty)
positions = [self._positions[e] for e in entities]
- def cb(ty: type[Entity] | type[Component]) -> Iterable[Any]:
- if issubclass(ty, Entity):
+ def cb(ty: ComponentID | None) -> Iterable[Any]:
+ if ty is None:
return entities
return map(self._components[ty].__getitem__, positions)
def _extend(
self,
- cb: Callable[[type["Component"] | type["Entity"]], Iterable[Any]],
+ cb: Callable[[ComponentID | None], Iterable[Any]],
) -> None:
- for entity in cb(Entity):
+ for entity in cb(None):
self._positions[entity] = len(self._entities)
self._entities.append(entity)
for ty, comp in self._components.items():
type Plugin = RecursiveTuple[Callable[["World"], None]]
-type ComponentHook[T] = Callable[["World", "Entity", T], None]
+type RawComponentHook[T] = Callable[[list[tuple["Entity", T]]], None]
+type ComponentHook[T] = System[RawComponentHook[T]]
class FromWorld(ABC):
else lambda i: filter_fetch(i) is not None
)
+ @classmethod
+ def fetch_prio(cls, ty: type) -> int:
+ return 0
+
@classmethod
@abstractmethod
def filter_fetch(
@override
@classmethod
def filter(cls, ty: type) -> Callable[[Archetype], bool]:
- return lambda archetype: ty in archetype._components
+ return lambda archetype: (
+ archetype._world.component_id(ty) in archetype._components
+ )
@override
@classmethod
) -> Callable[[Archetype], None | Callable[[int], Self | _NoFetchCls]]:
return lambda archetype: (
None
- if (idx := archetype._components.get(ty)) is None
+ if (
+ idx := archetype._components.get(
+ archetype._world.component_id(ty)
+ )
+ )
+ is None
else cast(Callable[[int], Self], lambda i: idx[i])
)
self._world: World = world
self._fetchers_raw = cast(
Callable[[Archetype], Callable[[int], Items | _NoFetchCls] | None],
- Query.preprocess_fetchers(resolved_fetchers),
+ Query.preprocess_fetchers(resolved_fetchers)[1],
)
self._filters_raw = Query.preprocess_filters(resolved_filters)
self._seen_archetypes: int = 0
@staticmethod
def preprocess_fetchers(
fetcher: type[QueryItems],
- ) -> Callable[
- [Archetype], Callable[[int], QueryItems | _NoFetchCls] | None
+ ) -> tuple[
+ int,
+ Callable[
+ [Archetype], Callable[[int], QueryItems | _NoFetchCls] | None
+ ],
]:
args = list(map(Query.preprocess_fetchers, get_args(fetcher)))
archetype: Archetype,
) -> Callable[[int], QueryItems] | None:
lst = []
- for arg in args:
+ for _, arg in args:
curr = arg(archetype)
if curr is None:
return None
archetype: Archetype,
) -> Callable[[int], QueryItems | _NoFetchCls] | None:
lst: list[Callable[[int], QueryItems | _NoFetchCls]] = []
- for arg in args:
+ for _, arg in args:
curr = arg(archetype)
if curr is None:
continue
orig = origin_or_cls(fetcher)
if orig is tuple:
- return tuple_constructor
+ return (0, tuple_constructor)
if orig is Union:
- return union_constructor
+ args.sort(key=lambda e: e[0], reverse=True)
+ return (0, union_constructor)
if orig is None or orig is type(None):
- return lambda _: None
+ return (-1, lambda _: lambda _: None)
- return cast(QueryItem, fetcher).filter_fetch(fetcher)
+ fetcher_cls = cast(QueryItem, fetcher)
+ return (
+ fetcher_cls.fetch_prio(fetcher),
+ fetcher_cls.filter_fetch(fetcher),
+ )
@staticmethod
def preprocess_filters(
) -> Callable[[Archetype], Callable[[int], bool] | bool]:
args = list(map(Query.preprocess_filters, get_args(filter)))
- def composed_by(
- a: Callable[[int], bool] | None,
- b: Callable[[int], bool],
- op: Callable[[bool, bool], bool],
- ) -> Callable[[int], bool]:
- if a is None:
- return b
- return lambda i: op(a(i), b(i))
-
def tuple_constructor(
archetype: Archetype,
) -> Callable[[int], bool] | bool:
raise ValueError("Failed to fetch arguments for system")
+@dataclass
+class ComponentHooks[T: Component]:
+ _insert: RawComponentHook[T] = lambda _: None
+ _remove: RawComponentHook[T] = lambda _: None
+
+ def add_insert(self, cb: RawComponentHook[T]) -> None:
+ self._insert = composed_by(self._insert, cb, lambda _1, _2: None)
+
+ def add_remove(self, cb: RawComponentHook[T]) -> None:
+ self._remove = composed_by(self._remove, cb, lambda _1, _2: None)
+
+
class World(FromWorld):
def __init__(self) -> None:
self._entities: dict[Entity, Archetype] = {}
- self._archetypes_lookup: dict[
- frozenset[type[Component]], Archetype
- ] = {}
+ self._archetypes_lookup: dict[frozenset[ComponentID], Archetype] = {}
self._archetypes: list[Archetype] = []
+ self._components_cache: list[
+ tuple[type, ComponentHooks[Component]]
+ ] = []
+ self._components_lookup: dict[type, ComponentID] = {}
self._schedules: dict[ScheduleLabel, Schedule[RawSystem]] = {}
- self._insert_hooks: dict[type, ComponentHook[Any]] = {}
- self._remove_hooks: dict[type, ComponentHook[Any]] = {}
self._base_archetype: Archetype = Archetype(self, ())
- def _run_cond(self, _: RawSystem | SystemSet) -> bool:
- return True
+ def component_id(self, comp: type) -> ComponentID:
+ if (res := self._components_lookup.get(comp)) is None:
+ print(f"added comp {comp}")
+ res = ComponentID(len(self._components_cache))
+ self._components_cache.append((comp, ComponentHooks()))
+ self._components_lookup[comp] = res
+ return res
def spawn_empty(self) -> Entity:
return self.spawn_many_empty(1)[0]
self._base_archetype._extend(
lambda ty: (
res
- if ty is Entity
+ if ty is None
else panic(Exception("Should be unreachable"))
)
)
def tick(self, schedule: ScheduleLabel) -> None:
if schedule not in self._schedules:
return
- for system in self._schedules[schedule].traverse(self._run_cond):
+ for system in self._schedules[schedule].traverse(lambda _: True):
system()
def run(
Systems(*systems).in_set(*sets)._apply(self, self._schedules[schedule])
return self
- def with_hook[T](
+ def with_hook[T: Component](
self,
cycle: Literal["insert", "remove"],
ty: type[T],
hook: ComponentHook[T],
) -> "World":
+ comp_id = self.component_id(resolve_type_aliases(ty))
+ hooks = self._components_cache[comp_id][1]
+ sys = self.add_system_raw(hook)
match cycle:
case "insert":
- d = self._insert_hooks
+ hooks.add_insert(lambda e: sys()(e))
case "remove":
- d = self._remove_hooks
- if ty not in d:
- d[ty] = hook
- else:
- old = d[ty]
-
- def new(world: World, entity: Entity, component: T) -> None:
- old(world, entity, component)
- hook(world, entity, component)
- pass
-
- d[ty] = new
+ hooks.add_remove(lambda e: sys()(e))
return self
def with_plugins(self, *plugins: Plugin) -> "World":
projects / axy / ft / pacman.git / commitdiff