from collections.abc import Callable, Generator, Iterable
-from typing import get_args, override
+from typing import cast, get_args, override
from pacman.ecs.world import (
CommandQueue,
cls, ty: type, world: "World"
) -> Callable[[], "Hierarchy[Items, Filters]"]:
items, filters = get_args(ty)
- roots_ty: type[RootsQuery[Items, Filters]] = resolve_type_aliases(
- RootsQuery[items, filters]
+ type _Items = items # type: ignore [valid-type]
+ type _Filters = filters # type: ignore [valid-type]
+ roots_ty = cast(
+ type[RootsQuery[Items, Filters]],
+ resolve_type_aliases(
+ RootsQuery[
+ _Items,
+ _Filters,
+ ]
+ ),
)
- nodes_ty: type[NodesQuery[Items, Filters]] = resolve_type_aliases(
- NodesQuery[items, filters]
+ nodes_ty = cast(
+ type[NodesQuery[Items, Filters]],
+ resolve_type_aliases(NodesQuery[_Items, _Filters]),
)
roots = roots_ty.extract_from_world(roots_ty, world)
nodes = nodes_ty.extract_from_world(nodes_ty, world)
from typing import (
Any,
Literal,
+ Never,
NewType,
Self,
Union,
def components_iter(self) -> Iterator["Component"]: ...
@classmethod
@abstractmethod
- def components_type_iter(cls) -> Iterator[type["Component"]]: ...
+ def components_type_iter(cls, ty: type) -> Iterator[type["Component"]]: ...
type Components = RecursiveTuple["Component | Bundle"]
elif issubclass(components, Component):
yield components
else:
- yield from cast(type[Bundle], components).components_type_iter()
+ yield from cast(type[Bundle], components).components_type_iter(
+ components
+ )
ComponentID = NewType("ComponentID", int)
self._world._archetypes.append(self)
self._world._archetypes_lookup[self._types] = self
- def _compute_add(
- self, ty: type[Components]
+ def _compute_add[T: Components](
+ self, ty: type[T]
) -> tuple["Archetype", list[ComponentID]]:
ptr = id(ty)
if res := self._add_cache.get(ptr):
self._add_cache[ptr] = (ty, res_arch, comps)
return (res_arch, comps)
- def _compute_remove(
- self, ty: type[Components]
+ def _compute_remove[T: Components](
+ self, ty: type[T]
) -> tuple["Archetype", list[ComponentID]]:
ptr = id(ty)
if res := self._remove_cache.get(ptr):
for ent in entities:
del self._world._entities[ent]
for hook, comp in comps:
- hook._remove(zip(entities, comp))
+ hook._remove(zip(entities, comp, strict=True))
def _extend(
self,
remove_idx(comp)
-type RawSystem[T] = Callable[[], T]
+type RawSystem[T = Any] = Callable[[], T]
-type System[T] = Callable[..., T]
+type System[T = Any] = Callable[..., T]
type Plugin = RecursiveTuple[Callable[["World"], None]]
type RawComponentArg[T] = Iterable[tuple["Entity", T]]
-type RawComponentHook[T] = Callable[[RawComponentArg], None]
+type RawComponentHook[T] = Callable[[RawComponentArg[T]], None]
type ComponentHook[T] = System[RawComponentHook[T]]
) -> Callable[[], Self | None]: ...
def cleanup(self) -> None:
- pass
+ return None
class QueryFilter(ABC):
@classmethod
def fetch_prio(cls, ty: type) -> int:
+ del ty
return 0
@classmethod
@override
@classmethod
def filter(cls, ty: type) -> Callable[[Archetype], bool]:
+ del ty
return lambda _: True
@override
def filter_fetch(
cls, ty: type
) -> Callable[[Archetype], None | Callable[[int], "Entity"]]:
+ del ty
return lambda archetype: lambda i: archetype._entities[i]
@classmethod
def filter_fetch(
cls, ty: type
- ) -> Callable[[Archetype], None | Callable[[int], Self | _NoFetchCls]]:
+ ) -> Callable[[Archetype], None | Callable[[int], "Not[T] | _NoFetchCls"]]:
preprocessed = cls.filter(ty)
return lambda archetype: (
(lambda _: Not())
- if (filt := preprocessed(archetype)) == False
+ if (filt := preprocessed(archetype)) is False
else (
None
- if filt == True
+ if filt is True
else (lambda i: _NoFetch if filt(i) else Not())
)
)
def extract_from_world(
cls, ty: type, world: "World"
) -> Callable[[], "Not[T] | None"]:
- inner = cast(type[FromWorld], get_args(ty)[0])
- extractor = inner.extract_from_world(inner, world)
+ inner_ty = cast(type[FromWorld], get_args(ty)[0])
+ extractor = inner_ty.extract_from_world(inner_ty, world)
def inner() -> Not[T] | None:
if (res := extractor()) is None:
def extract_from_world(
cls, ty: type, world: "World"
) -> Callable[[], "Maybe[T]"]:
- inner = cast(type[T], get_args(ty)[0])
- extractor = inner.extract_from_world(inner, world)
+ inner_ty = cast(type[T], get_args(ty)[0])
+ extractor = inner_ty.extract_from_world(inner_ty, world)
def inner() -> Maybe[T]:
return Maybe(extractor())
return cast(QueryFilter, filter).filter(filter)
def add_archetype(self, archetype: Archetype) -> None:
- cond: Callable[[int], bool]
- match self._filters_raw(archetype):
- case True:
- cond = lambda _: True
- case False:
- return
- case e:
- cond = e
+ filt = self._filters_raw(archetype)
+ if filt is False:
+ return
+ cond: Callable[[int], bool] = (
+ (lambda _: True) if filt is True else filt
+ )
fetcher = self._fetchers_raw(archetype)
if fetcher is None:
)
@override
- def __contains__(self, entity: Entity) -> bool:
+ def __contains__(self, entity: object) -> bool:
+ if not isinstance(entity, Entity):
+ return False
arch: Archetype | None = self._world._entities.get(entity)
if arch is None:
return False
) -> Callable[[], Self | None]:
_query = Query[ty] # type: ignore
query: Callable[[], Query[Self]] = cast(
- type[Query[Self]], _query
- ).extract_from_world(_query, world)
+ Callable[[], Query[Self]], _query.extract_from_world(_query, world)
+ )
return lambda: query().single()
def extract_from_world(
cls, ty: type, world: "World"
) -> Callable[[], Self]:
+ del ty
return lambda: cls(world)
@override
self._queue.append(sys)
-def _system_failed_va[T, *P = *tuple[()]](*_: *P) -> T:
+def _system_failed_va[*P = *tuple[()]](*_: *P) -> Never:
raise ValueError("Failed to fetch arguments for system")
-def _system_failed[T]() -> T:
+def _system_failed() -> Never:
raise ValueError("Failed to fetch arguments for system")
tuple[type, ComponentHooks[Component]]
] = []
self._components_lookup: dict[type, ComponentID] = {}
- self._schedules: dict[ScheduleLabel, Schedule[RawSystem]] = {}
+ self._schedules: dict[ScheduleLabel, Schedule[RawSystem[Any]]] = {}
self._base_archetype: Archetype = Archetype(self, ())
def component_id(self, comp: type) -> ComponentID:
for extract in extractors:
curr = extract()
if curr is None:
- return default()
+ return default(*cb_args)
args.append(curr)
res = system(*args)(*cb_args)
for arg in args:
def extract_from_world(
cls, ty: type, world: "World"
) -> Callable[[], "World"]:
+ del ty
return lambda: world
def tick(self, schedule: ScheduleLabel) -> None:
hook: ComponentHook[T],
) -> "World":
comp_id = self.component_id(resolve_type_aliases(ty))
- hooks = self._components_cache[comp_id][1]
+ hooks = cast(ComponentHooks[T], self._components_cache[comp_id][1])
sys = self.add_system(hook, default=lambda _: None)
match cycle:
case "insert":
- hooks.add_insert(lambda e: sys(e))
+ hooks.add_insert(sys)
case "remove":
- hooks.add_remove(lambda e: sys(e))
+ hooks.add_remove(sys)
return self
def with_plugins(self, *plugins: Plugin) -> "World":
projects / axy / ft / pacman.git / commitdiff
