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

Release/v0.4.1 #1

Merged
rus07tam merged 12 commits from release/v0.4.1 into main 2025-10-27 15:05:33 +03:00
Conversation 2 Commits 12 Files changed 14 +103 -678
14 changed files with 103 additions and 678 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -13,5 +13,6 @@ wheels/
.direnv
.mypy_cache
.python-version
.pytest_cache
.vscode
_autosummary

678
README.md
View file

@ -1,29 +1,41 @@
<div align="center">
# 🐍 Snakia Framework
![Code Quality](https://img.shields.io/codacy/grade/508e02449dd145e7a93c7ba08ac756a4?logo=codacy)
![Code Size](https://img.shields.io/github/languages/code-size/ruject/snakia)
![License](https://img.shields.io/github/license/ruject/snakia)
![Open Issues](https://img.shields.io/github/issues-raw/ruject/snakia)
![Commit Activity](https://img.shields.io/github/commit-activity/m/ruject/snakia)
[API Reference](https://ruject.github.io/snakia/)
&nbsp;•&nbsp;
[Telegram Chat](https://t.me/RuJect_Community)
</div>
**Snakia** is a modern Python framework for creating applications with Entity-Component-System (ECS) architecture, event system, and reactive programming. Built with performance (maybe) and modularity in mind, Snakia provides a clean API for developing complex applications ranging from games to terminal user interfaces.
## 📋 Table of Contents
- [🎯 Roadmap & TODO](#-roadmap--todo)
- [🚀 Installation](#-installation)
- [🚀 Quick Start](#-quick-start)
- [🏗️ Architecture](#️-architecture)
- [⚙️ Core](#️-core)
- [🎯 ECS System](#-ecs-system)
- [📡 Event System (ES)](#-event-system-es)
- [🔌 Plugin System](#-plugin-system)
- [🎨 TUI System](#-tui-system)
- [⚡ Reactive Programming (RX)](#-reactive-programming-rx)
- [🛠️ Utilities](#️-utilities)
- [🎭 Decorators](#-decorators)
- [🏷️ Properties](#-properties)
- [🌐 Platform Abstraction](#-platform-abstraction)
- [📦 Examples](#-examples)
- [🤝 Contributing](#-contributing)
- [🆘 Support](#-support)
- [📄 License](#-license)
- [🐍 Snakia Framework](#-snakia-framework)
- [📋 Table of Contents](#-table-of-contents)
- [✨ Key Features](#-key-features)
- [🚀 Installation](#-installation)
- [Prerequisites](#prerequisites)
- [Install from PyPi (recommended)](#install-from-pypi-recommended)
- [Install from Source](#install-from-source)
- [🎯 Roadmap \& TODO](#-roadmap--todo)
- [🚀 Quick Start](#-quick-start)
- [🏗️ Architecture](#️-architecture)
- [📦 Examples](#-examples)
- [Health System](#health-system)
- [TUI Application](#tui-application)
- [🤝 Contributing](#-contributing)
- [How to Contribute](#how-to-contribute)
- [Development Guidelines](#development-guidelines)
### ✨ Key Features
## ✨ Key Features
- 🏗️ **ECS Architecture** - Flexible entity-component-system for scalable game/app logic
- 📡 **Event System** - Asynchronous event handling with filters and priorities
@ -52,15 +64,9 @@ pip install snakia
### Install from Source
```bash
# Clone the repository
git clone https://github.com/RuJect/Snakia.git
cd Snakia
# Install with pip
pip install -e .
# Or with uv (recommended)
uv sync
```
## 🎯 Roadmap & TODO
@ -133,616 +139,6 @@ Snakia/
└── types/ # Special types
```
## ⚙️ Core
### Engine
The central component of the framework that coordinates all systems:
```python
from snakia.core.engine import Engine
engine = Engine()
# Systems:
# - engine.system - ECS system
# - engine.dispatcher - Event system
# - engine.loader - Plugin loader
engine.start() # Start all systems
engine.stop() # Stop all systems
engine.update() # Update systems
```
## 🎯 ECS System
Entity-Component-System architecture for creating flexible and performant applications.
### Component
Base class for all components:
```python
from snakia.core.ecs import Component
from pydantic import Field
class PositionComponent(Component):
x: float = Field(default=0.0)
y: float = Field(default=0.0)
class VelocityComponent(Component):
vx: float = Field(default=0.0)
vy: float = Field(default=0.0)
```
### Processor
Processors handle components in the system:
```python
from snakia.core.ecs import Processor, System
class MovementProcessor(Processor):
def process(self, system: System) -> None:
# Get all entities with Position and Velocity
for entity, (pos, vel) in system.get_components(
PositionComponent, VelocityComponent
):
pos.x += vel.vx
pos.y += vel.vy
```
### System
Entity and component management:
```python
# Creating an entity with components
entity = system.create_entity(
PositionComponent(x=10, y=20),
VelocityComponent(vx=1, vy=0)
)
# Adding a component to an existing entity
system.add_component(entity, HealthComponent(value=100))
# Getting entity components
pos, vel = system.get_components_of_entity(
entity, PositionComponent, VelocityComponent
)
# Checking for components
if system.has_components(entity, PositionComponent, VelocityComponent):
print("Entity has position and velocity")
# Removing a component
system.remove_component(entity, VelocityComponent)
# Deleting an entity
system.delete_entity(entity)
```
## 📡 Event System (ES)
Asynchronous event system with filter and priority support.
### Event
Base class for events:
```python
from snakia.core.es import Event
from pydantic import Field
class PlayerDiedEvent(Event):
player_id: int = Field()
cause: str = Field(default="unknown")
ttl: int = Field(default=10) # Event lifetime
```
### Handler
Event handlers:
```python
from snakia.core.es import Handler, Action
def on_player_died(event: PlayerDiedEvent) -> Action | None:
print(f"Player {event.player_id} died from {event.cause}")
return Action.move(1) # Move to next handler
```
### Filter
Event filters:
```python
from snakia.core.es import Filter
def only_important_deaths(event: PlayerDiedEvent) -> bool:
return event.cause in ["boss", "pvp"]
# Using a filter
@dispatcher.on(PlayerDiedEvent, filter=only_important_deaths)
def handle_important_death(event: PlayerDiedEvent):
print("Important death occurred!")
```
### Dispatcher
Central event dispatcher:
```python
from snakia.core.es import Dispatcher, Subscriber
dispatcher = Dispatcher()
# Subscribing to an event
dispatcher.subscribe(PlayerDiedEvent, Subscriber(
handler=on_player_died,
filter=only_important_deaths,
priority=10
))
# Decorator for subscription
@dispatcher.on(PlayerDiedEvent, priority=5)
def handle_death(event: PlayerDiedEvent):
print("Death handled!")
# Publishing an event
dispatcher.publish(PlayerDiedEvent(player_id=123, cause="boss"))
```
## 🔌 Plugin System
Modular system for loading and managing plugins.
### Plugin
Base class for plugins:
```python
from snakia.core.loader import Meta, Plugin, PluginProcessor
from snakia.types import Version
class MyProcessor(PluginProcessor):
def process(self, system):
# Processor logic
pass
class MyPlugin(Plugin, meta=Meta(
name="my_plugin",
author="developer",
version=Version.from_args(1, 0, 0),
processors=(MyProcessor,),
subscribers=()
)):
def on_load(self):
print("Plugin loaded!")
def on_unload(self):
print("Plugin unloaded!")
```
### Meta
Plugin metadata:
```python
from snakia.core.loader import Meta
from snakia.core.es import Subscriber
meta = Meta(
name="plugin_name",
author="author_name",
version=Version.from_args(1, 0, 0),
processors=(Processor1, Processor2),
subscribers=(
(EventType, Subscriber(handler, filter, priority)),
)
)
```
### Loader
Plugin loader:
```python
from snakia.core.loader import Loader
loader = Loader(engine)
# Registering a plugin
loader.register(MyPlugin)
# Loading all plugins
loader.load_all()
# Unloading all plugins
loader.unload_all()
```
## 🎨 TUI System
System for creating text-based user interfaces.
### Widget
Base class for widgets:
```python
from snakia.core.tui import Widget, Canvas, CanvasChar
from snakia.core.rx import Bindable
class MyWidget(Widget):
def __init__(self):
super().__init__()
self.text = self.state("Hello World")
self.color = self.state(CanvasChar(fg_color="red"))
def on_render(self) -> Canvas:
canvas = Canvas(20, 5)
canvas.draw_text(0, 0, self.text.value, self.color.value)
return canvas
```
### Canvas
Drawing canvas:
```python
from snakia.core.tui import Canvas, CanvasChar
canvas = Canvas(80, 24)
# Drawing text
canvas.draw_text(10, 5, "Hello", CanvasChar(fg_color="blue"))
# Drawing rectangle
canvas.draw_rect(0, 0, 20, 10, CanvasChar("█", fg_color="green"))
# Filling area
canvas.draw_filled_rect(5, 5, 10, 5, CanvasChar(" ", bg_color="red"))
# Lines
canvas.draw_line_h(0, 0, 20, CanvasChar("-"))
canvas.draw_line_v(0, 0, 10, CanvasChar("|"))
```
### CanvasChar
Character with attributes:
```python
from snakia.core.tui import CanvasChar
char = CanvasChar(
char="A",
fg_color="red", # Text color
bg_color="blue", # Background color
bold=True, # Bold
italic=False, # Italic
underline=True # Underline
)
```
### Renderer
Screen rendering:
```python
from snakia.core.tui import RenderContext
from snakia.core.tui.render import ANSIRenderer
import sys
class StdoutTarget:
def write(self, text: str): sys.stdout.write(text)
def flush(self): sys.stdout.flush()
renderer = ANSIRenderer(StdoutTarget())
with RenderContext(renderer) as ctx:
ctx.render(widget.render())
```
### Ready-made Widgets
```python
from snakia.core.tui.widgets import (
TextWidget, BoxWidget,
HorizontalSplitWidget, VerticalSplitWidget
)
# Text widget
text = TextWidget("Hello", CanvasChar(fg_color="red", bold=True))
# Box widget
box = BoxWidget(10, 5, CanvasChar("█", fg_color="yellow"))
# Splitters
h_split = HorizontalSplitWidget([text1, text2], "|")
v_split = VerticalSplitWidget([h_split, box], "-")
```
## ⚡ Reactive Programming (RX)
Reactive programming system for creating responsive interfaces.
### Bindable
Reactive variables:
```python
from snakia.core.rx import Bindable, ValueChanged
# Creating a reactive variable
counter = Bindable(0)
# Subscribing to changes
def on_change(event: ValueChanged[int]):
print(f"Counter changed from {event.old_value} to {event.new_value}")
counter.subscribe(on_change)
# Changing value
counter.set(5) # Will call on_change
counter(10) # Alternative syntax
```
### AsyncBindable
Asynchronous reactive variables:
```python
from snakia.core.rx import AsyncBindable
async_counter = AsyncBindable(0)
async def async_handler(event: ValueChanged[int]):
print(f"Async counter: {event.new_value}")
await async_counter.subscribe(async_handler, run_now=True)
await async_counter.set(42)
```
### Operators
```python
from snakia.core.rx import map, filter, combine, merge
# Transformation
doubled = map(counter, lambda x: x * 2)
# Filtering
even_only = filter(counter, lambda x: x % 2 == 0)
# Combining
combined = combine(counter, doubled, lambda a, b: a + b)
# Merging streams
merged = merge(counter, async_counter)
```
## 🛠️ Utilities
### to_async
Converting synchronous functions to asynchronous:
```python
from snakia.utils import to_async
def sync_function(x):
return x * 2
async_function = to_async(sync_function)
result = await async_function(5)
```
### nolock
Performance optimization:
```python
from snakia.utils import nolock
def busy_loop():
while running:
# Work
nolock() # Release GIL
```
### inherit
Simplified inheritance:
```python
from snakia.utils import inherit
class Base:
def method(self): pass
class Derived(inherit(Base)):
def method(self):
super().method()
# Additional logic
```
### this
Reference to current object:
```python
from snakia.utils import this
def func():
return this() # Returns `<function func at ...>`
```
### throw
Throwing exceptions:
```python
from snakia.utils import throw
def validate(value):
if value < 0:
throw(ValueError("Value must be positive"))
```
### frame
Working with frames:
```python
from snakia.utils import frame
def process_frame():
current_frame = frame()
# Process frame
```
## 🎭 Decorators
### inject_replace
Method replacement:
```python
from snakia.decorators import inject_replace
class Original:
def method(self): return "original"
@inject_replace(Original, "method")
def new_method(self): return "replaced"
```
### inject_before / inject_after
Hooks before and after execution:
```python
from snakia.decorators import inject_before, inject_after
@inject_before(MyClass, "method")
def before_hook(self): print("Before method")
@inject_after(MyClass, "method")
def after_hook(self): print("After method")
```
### singleton
Singleton pattern:
```python
from snakia.decorators import singleton
@singleton
class Database:
def __init__(self):
self.connection = "connected"
```
### pass_exceptions
Exception handling:
```python
from snakia.decorators import pass_exceptions
@pass_exceptions(ValueError, TypeError)
def risky_function():
# Code that might throw exceptions
pass
```
## 🏷️ Properties
### readonly
Read-only property:
```python
from snakia.property import readonly
class Currency:
@readonly
def rate(self) -> int:
return 100
currency = Currency()
currency.rate = 200
print(currency.rate) # Output: 100
```
### initonly
Initialization-only property:
```python
from snakia.property import initonly
class Person:
name = initonly[str]("name")
bob = Person()
bob.name = "Bob"
print(bob.name) # Output: "Bob"
bob.name = "not bob"
print(bob.name) # Output: "Bob"
```
### 🏛️ classproperty
Class property:
```python
from snakia.property import classproperty
class MyClass:
@classproperty
def class_value(cls):
return "class_value"
```
## 🌐 Platform Abstraction
### 🖥️ PlatformOS
Operating system abstraction:
```python
from snakia.platform import PlatformOS, OS
# Detecting current OS
current_os = OS.current()
if current_os == PlatformOS.LINUX:
print("Running on Linux")
elif current_os == PlatformOS.ANDROID:
print("Running on Android")
```
### 🏗️ PlatformLayer
Platform layers:
```python
from snakia.platform import LinuxLayer, AndroidLayer
# Linux layer
linux_layer = LinuxLayer()
# Android layer
android_layer = AndroidLayer()
```
## 📦 Examples
### Health System
@ -857,15 +253,3 @@ We welcome contributions to Snakia development! Whether you're fixing bugs, addi
- Write clear commit messages
- Update documentation for new features
- Test your changes thoroughly
## 🆘 Support
Need help? We're here to assist you!
- 🐛 **Bug Reports** - [GitHub Issues](https://github.com/RuJect/Snakia/issues)
- 💬 **Community Chat** - [RuJect Community Telegram](https://t.me/RuJect_Community)
- 📧 **Direct Contact** - mailto:rus07tam.uwu@gmail.com
## 📄 License
See the `LICENSE` file for details.

3
pyproject.toml
View file

@ -1,6 +1,6 @@
[project]
name = "snakia"
version = "0.4.0"
version = "0.4.1"
description = "Modern python framework"
readme = "README.md"
authors = [
@ -38,3 +38,4 @@ disable = ["C0114", "C0115", "C0116", "R0801"]
max-args = 8
max-positional-arguments = 7
min-public-methods = 1
fail-on = "error"

2
src/snakia/core/rx/__init__.py
View file

@ -4,6 +4,7 @@ from .bindable import Bindable
from .chain import chain
from .combine import combine
from .concat import concat
from .cond import cond
from .const import const
from .filter import filter # noqa: W0622 # pylint: disable=W0622
from .map import map # noqa: W0622 # pylint: disable=W0622
@ -18,6 +19,7 @@ __all__ = [
"chain",
"combine",
"concat",
"cond",
"const",
"filter",
"map",

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

@ -12,10 +12,6 @@ class AsyncBindable[T: Any](BaseBindable[T]):
super().__init__(default_value)
self.__subscribers: list[BindableSubscriber[T, Awaitable[Any]]] = []
@property
def value(self) -> T:
return self.__value
@property
def subscribers(
self,
@ -25,8 +21,8 @@ class AsyncBindable[T: Any](BaseBindable[T]):
async def set(self, value: T) -> None:
"""Set the value."""
e = ValueChanged(self.__value, value)
self.__value = value
e = ValueChanged(self.value, value)
self.set_silent(value)
for subscriber in self.__subscribers:
await subscriber(e)
@ -58,7 +54,7 @@ class AsyncBindable[T: Any](BaseBindable[T]):
async def _run() -> None:
await subscriber(
ValueChanged(self.__default_value, self.__value)
ValueChanged(self.__default_value, self.value)
)
return _run()
@ -96,7 +92,7 @@ class AsyncBindable[T: Any](BaseBindable[T]):
async def _run() -> None:
await subscriber(
ValueChanged(self.__default_value, self.__value)
ValueChanged(self.__default_value, self.value)
)
return _run()

2
src/snakia/core/rx/bindable.py
View file

@ -19,7 +19,7 @@ class Bindable[T: Any](BaseBindable[T]):
def set(self, value: T) -> None:
"""Set the value."""
e = ValueChanged(self.__value, value)
e = ValueChanged(self.value, value)
self.set_silent(value)
for subscriber in self.__subscribers:
subscriber(e)

13
src/snakia/core/rx/cond.py Normal file
View file

@ -0,0 +1,13 @@
from typing import Callable
def cond[**P, T, F](
codacy-production[bot] commented 2025-10-27 15:04:50 +03:00 (Migrated from github.com)
Copy link

🚫 Codacy found a high ErrorProne issue: expected '(' (F999)

The issue reported by the Prospector linter is due to the incorrect syntax used for the type parameters in the function definition. In Python's type hinting, you cannot use square brackets [] for defining type variables; instead, you should use parentheses () with the TypeVar function from the typing module.

To fix the issue, you need to change the definition of the cond function to use TypeVar for the type parameters. Here's the single line change needed:

def cond(P, T, F)(condition: Callable[P, bool], if_true: Callable[P, T], if_false: Callable[P, F]) -> Callable[P, T | F]:

This comment was generated by an experimental AI tool.

:no_entry_sign: **Codacy** found a **high ErrorProne** issue: [expected '(' (F999)](https://app.codacy.com/gh/RuJect/snakia/pull-requests/1) The issue reported by the Prospector linter is due to the incorrect syntax used for the type parameters in the function definition. In Python's type hinting, you cannot use square brackets `[]` for defining type variables; instead, you should use parentheses `()` with the `TypeVar` function from the `typing` module. To fix the issue, you need to change the definition of the `cond` function to use `TypeVar` for the type parameters. Here's the single line change needed: ```suggestion def cond(P, T, F)(condition: Callable[P, bool], if_true: Callable[P, T], if_false: Callable[P, F]) -> Callable[P, T | F]: ``` --- *This comment was generated by an experimental AI tool.*
condition: Callable[P, bool],
if_true: Callable[P, T],
if_false: Callable[P, F],
) -> Callable[P, T | F]:
return lambda *args, **kw: (
if_true(*args, **kw)
if condition(*args, **kw)
else if_false(*args, **kw)
)

14
src/snakia/field/field.py
View file

@ -36,14 +36,24 @@ class Field[T: Any](ABC, PrivProperty[T]):
@classmethod
def custom[R](
cls: type[Field[Any]],
serialize: Callable[[R], str],
deserialize: Callable[[str], R],
serialize: Callable[[Field[R], R], bytes],
deserialize: Callable[[Field[R], bytes], R],
) -> type[Field[R]]:
return inherit(
cls, {"serialize": serialize, "deserialize": deserialize}
)
@final
@staticmethod
def get_fields(class_: type[Any] | Any, /) -> dict[str, Field[Any]]:
if not isinstance(class_, type):
class_ = class_.__class__
return {
k: v for k, v in class_.__dict__.items() if isinstance(v, Field)
}
if TYPE_CHECKING:
@final
@classmethod
def type(cls) -> type[T]: ...

3
src/snakia/property/__init__.py
View file

@ -1,5 +1,5 @@
from .cell_property import CellProperty
from .classproperty import ClassProperty
from .classproperty import ClassProperty, classproperty
from .hook_property import HookProperty
from .initonly import Initonly, initonly
from .priv_property import PrivProperty
@ -9,6 +9,7 @@ from .readonly import Readonly, readonly
__all__ = [
"CellProperty",
"ClassProperty",
"classproperty",
"HookProperty",
"Initonly",
"initonly",

2
src/snakia/property/classproperty.py
View file

@ -48,7 +48,7 @@ class ClassProperty[T]:
def classproperty[T](
fget: Callable[[Any], T] = empty.func,
fget: Callable[[Any], T],
fset: Callable[[Any, T], None] = empty.func,
fdel: Callable[[Any], None] = empty.func,
) -> ClassProperty[T]:

2
src/snakia/property/priv_property.py
View file

@ -15,7 +15,7 @@ class PrivProperty[T]:
def __get__(self, instance: Any, owner: type | None = None, /) -> T:
if self.__default_value:
return getattr(instance, self.__name, self.__default_value)
return getattr(instance, self.__name)
return getattr(instance, self.__name) # type: ignore
def __set__(self, instance: Any, value: T, /) -> None:
setattr(instance, self.__name, value)

14
src/snakia/property/property.py
View file

@ -5,7 +5,12 @@ from snakia.types import empty
class Property[T]:
"""
A property that can be set, get, and deleted."""
A property that can be set, get, and deleted.
"""
__slots__ = "__fget", "__fset", "__fdel", "__name"
__name: str
def __init__(
self,
@ -17,6 +22,9 @@ class Property[T]:
self.__fset = fset
self.__fdel = fdel
def __set_name__(self, owner: type, name: str) -> None:
self.__name = name
def __get__(self, instance: Any, owner: type | None = None, /) -> T:
return self.__fget(instance)
@ -40,3 +48,7 @@ class Property[T]:
"""Descriptor deleter."""
self.__fdel = fdel
return self
@property
def name(self) -> str:
return self.__name

29
src/snakia/property/readonly.py
View file

@ -1,27 +1,32 @@
from typing import Any, Callable
from snakia.utils import throw
class Readonly[T]:
from .property import Property
class Readonly[T](Property[T]):
"""
codacy-production[bot] commented 2025-10-27 15:04:50 +03:00 (Migrated from github.com)
Copy link

🚫 Codacy found a high ErrorProne issue: invalid syntax (F999)

The issue arises from the use of generics in the class definition. The syntax Readonly[T](Property[T]) is incorrect in Python, as the use of generics requires the Generic base class from the typing module. To fix this, you need to inherit from Generic[T] in addition to Property[T].

Here's the code suggestion to fix the issue:

class Readonly(Property[T], Generic[T]):

This comment was generated by an experimental AI tool.

:no_entry_sign: **Codacy** found a **high ErrorProne** issue: [invalid syntax (F999)](https://app.codacy.com/gh/RuJect/snakia/pull-requests/1) The issue arises from the use of generics in the class definition. The syntax `Readonly[T](Property[T])` is incorrect in Python, as the use of generics requires the `Generic` base class from the `typing` module. To fix this, you need to inherit from `Generic[T]` in addition to `Property[T]`. Here's the code suggestion to fix the issue: ```suggestion class Readonly(Property[T], Generic[T]): ``` --- *This comment was generated by an experimental AI tool.*
Readonly property.
"""
__slots__ = ("__fget",)
def __init__(
self,
fget: Callable[[Any], T],
*,
strict: bool = False,
) -> None:
self.__fget = fget
def __get__(self, instance: Any, owner: type | None = None, /) -> T:
return self.__fget(instance)
def __set__(self, instance: Any, value: T, /) -> None:
pass
super().__init__(
fget=fget,
fset=(
(lambda *_: throw(TypeError("Cannot set readonly property")))
if strict
else lambda *_: None
),
)
def readonly[T](value: T) -> Readonly[T]:
def readonly[T](value: T, *, strict: bool = False) -> Readonly[T]:
"""Create a readonly property with the given value.
Args:
@ -30,4 +35,4 @@ def readonly[T](value: T) -> Readonly[T]:
Returns:
Readonly[T]: The readonly property.
"""
return Readonly(lambda _: value)
return Readonly(lambda _: value, strict=strict)

2
uv.lock generated
View file

@ -104,7 +104,7 @@ wheels = [
[[package]]
name = "snakia"
version = "0.4.0"
version = "0.4.1"
source = { editable = "." }
dependencies = [
{ name = "networkx" },