RuJect/snakia
1
0
Fork 0
Code Issues 5 Pull requests Projects Releases 5 Packages Wiki Activity Actions

chore: format code

Browse source
This commit is contained in:
rus07tam 2025-11-23 10:49:54 +00:00
parent a49fe64be8
commit 7b34426c82
5 changed files with 26 additions and 63 deletions
Download patch file Download diff file Expand all files Collapse all files

8
examples/tui.py
View file

@ -2,8 +2,12 @@ import sys
from snakia.core.tui import CanvasChar, RenderContext
from snakia.core.tui.render import ANSIRenderer
from snakia.core.tui.widgets import (BoxWidget, HorizontalSplitWidget,
TextWidget, VerticalSplitWidget)
from snakia.core.tui.widgets import (
BoxWidget,
HorizontalSplitWidget,
TextWidget,
VerticalSplitWidget,
)
class StdoutTarget:

33
src/snakia/core/ecs/system.py
View file

@ -82,9 +82,7 @@ class System:
self.__processors.remove(processor)
@overload
def get_components(
self, c1: type[A], /
) -> Iterable[tuple[int, tuple[A]]]: ...
def get_components(self, c1: type[A], /) -> Iterable[tuple[int, tuple[A]]]: ...
@overload
def get_components(
@ -117,10 +115,7 @@ class System:
) -> Iterable[tuple[int, tuple[Component, ...]]]:
"""Returns all entities with the given components."""
entity_set = set.intersection(
*(
self.__components[component_type]
for component_type in component_types
)
*(self.__components[component_type] for component_type in component_types)
)
for entity in entity_set:
yield (
@ -181,9 +176,7 @@ class System:
),
)
def get_component(
self, component_type: type[C], /
) -> Iterable[tuple[int, C]]:
def get_component(self, component_type: type[C], /) -> Iterable[tuple[int, C]]:
"""Returns all entities with the given component."""
for entity in self.__components[component_type].copy():
yield entity, cast(C, self.__entitites[entity][component_type])
@ -214,24 +207,16 @@ class System:
self.__components[component_type].add(entity)
self.__entitites[entity][component_type] = component
def has_component(
self, entity: int, component_type: type[Component]
) -> bool:
def has_component(self, entity: int, component_type: type[Component]) -> bool:
"""Returns True if the entity has the given component."""
return component_type in self.__entitites[entity]
def has_components(
self, entity: int, *component_types: type[Component]
) -> bool:
def has_components(self, entity: int, *component_types: type[Component]) -> bool:
"""Returns True if the entity has all the given components."""
components_dict = self.__entitites[entity]
return all(
comp_type in components_dict for comp_type in component_types
)
return all(comp_type in components_dict for comp_type in component_types)
def remove_component(
self, entity: int, component_type: type[C]
) -> C | None:
def remove_component(self, entity: int, component_type: type[C]) -> C | None:
"""Removes a component from an entity."""
self.__components[component_type].discard(entity)
if not self.__components[component_type]:
@ -263,9 +248,7 @@ class System:
def entity_exists(self, entity: int) -> bool:
"""Returns True if the entity exists."""
return (
entity in self.__entitites and entity not in self.__dead_entities
)
return entity in self.__entitites and entity not in self.__dead_entities
def start(self) -> None:
"""Starts the system."""

4
src/snakia/core/engine.py
View file

@ -23,9 +23,7 @@ class Engine:
self.__dispatcher_thread: threading.Thread | None = None
def start(self) -> None:
self.__system_thread = threading.Thread(
target=self.system.start, daemon=False
)
self.__system_thread = threading.Thread(target=self.system.start, daemon=False)
self.__dispatcher_thread = threading.Thread(
target=self.dispatcher.start, daemon=False
)

18
src/snakia/core/es/dispatcher.py
View file

@ -23,9 +23,9 @@ class Dispatcher:
def __init__(self) -> None:
self.__queue: Final = queue.Queue[Event]()
self.__subscribers: Final[
dict[type[Event], list[Subscriber[Event]]]
] = defaultdict(list)
self.__subscribers: Final[dict[type[Event], list[Subscriber[Event]]]] = (
defaultdict(list)
)
self.__running: bool = False
@property
@ -33,15 +33,11 @@ class Dispatcher:
"""Returns True if the dispatcher is running."""
return self.__running
def subscribe(
self, event_type: type[T], subscriber: Subscriber[T]
) -> None:
def subscribe(self, event_type: type[T], subscriber: Subscriber[T]) -> None:
"""Subscribe to an event type."""
self.__subscribers[event_type].append(subscriber) # type: ignore
def unsubscribe(
self, event_type: type[T], subscriber: Subscriber[T]
) -> None:
def unsubscribe(self, event_type: type[T], subscriber: Subscriber[T]) -> None:
"""Unsubscribe from an event type."""
for sub in self.__subscribers[event_type].copy():
if sub.handler != subscriber.handler:
@ -97,9 +93,7 @@ class Dispatcher:
i = 0
while i < len(subscribers):
subscriber = subscribers[i]
if subscriber.filters is not None and not subscriber.filters(
event
):
if subscriber.filters is not None and not subscriber.filters(event):
continue
action = subscriber.handler(event)

26
src/snakia/core/rx/async_bindable.py
View file

@ -1,12 +1,4 @@
from typing import (
Any,
Awaitable,
Callable,
Generic,
Literal,
TypeVar,
overload,
)
from typing import Any, Awaitable, Callable, Generic, Literal, TypeVar, overload
from .base_bindable import BaseBindable, BindableSubscriber, ValueChanged
@ -66,25 +58,19 @@ class AsyncBindable(BaseBindable[T], Generic[T]):
if run_now:
async def _run() -> None:
await subscriber(
ValueChanged(self.__default_value, self.value)
)
await subscriber(ValueChanged(self.__default_value, self.value))
return _run()
return None
def unsubscribe(
self, subscriber: BindableSubscriber[T, Awaitable[Any]]
) -> None:
def unsubscribe(self, subscriber: BindableSubscriber[T, Awaitable[Any]]) -> None:
"""Unsubscribe from an value."""
self.__subscribers.remove(subscriber)
@overload
def on(
self, run_now: Literal[True]
) -> Callable[
[BindableSubscriber[T, Awaitable[Any]]], Awaitable[None]
]: ...
) -> Callable[[BindableSubscriber[T, Awaitable[Any]]], Awaitable[None]]: ...
@overload
def on(
@ -104,9 +90,7 @@ class AsyncBindable(BaseBindable[T], Generic[T]):
if run_now:
async def _run() -> None:
await subscriber(
ValueChanged(self.__default_value, self.value)
)
await subscriber(ValueChanged(self.__default_value, self.value))
return _run()
return None