ninemine/Convention-Python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

BS 0.1.0

Browse Source
This commit is contained in:
LiuBai095
2025-07-11 01:36:10 +08:00
parent b259b499d4
commit 56e16ca707
4 changed files with 201 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
Convention/Runtime/Architecture.md
View File

@@ -1,107 +0,0 @@
[返回](./Runtime-README.md)
# /Convention/Runtime/Architecture
---
核心架构模块，提供依赖注入、事件系统和时间线管理功能
## 接口定义
系统定义了以下核心接口来规范架构设计：
### ISignal
信号接口，用于事件系统的消息传递
### IModel
模型接口，定义数据模型的序列化和反序列化能力：
- `Save()` 保存数据到字符串
- `Load(string data)` 从字符串加载数据
### IConvertable<T>
转换接口，提供类型转换能力：
- `ConvertTo()` 转换为目标类型
### IConvertModel<T>
组合接口，同时继承 `IModel` 和 `IConvertable<T>`
## 核心类型
### SingletonModel<T>
泛型单例模型，提供类型安全的单例管理：
- 支持依赖注入和实例替换
- 提供隐式转换操作符
- 集成模型序列化功能
### DependenceModel
依赖模型，管理多个条件的组合验证：
- 支持多个 `IConvertModel<bool>` 的依赖关系
- 提供枚举接口遍历所有依赖
- 所有依赖条件均满足时返回true
## 注册系统
### 对象注册
- `Register(Type slot, object target, Action completer, params Type[] dependences)`
- `Register<T>(T target, Action completer, params Type[] dependences)`
提供基于类型的依赖注入容器：
- 支持依赖关系管理
- 支持注册完成回调
- 自动解析依赖顺序
### 注册查询
- `Contains(Type type)` / `Contains<T>()`
- `Get(Type type)` / `Get<T>()`
## 事件系统
### 消息监听
- `AddListener<Signal>(Type slot, Action<Signal> listener)`
- `SendMessage(Type slot, ISignal signal)`
- `SendMessage<Signal>(Signal signal)`
提供类型安全的事件分发机制：
- 支持按类型注册监听器
- 支持泛型消息类型
- 自动管理监听器生命周期
### Listening类
监听器管理类，提供：
- `StopListening()` 停止监听
## 时间线系统
### 时间线管理
- `CreateTimeline()` 创建新的时间线
- `AddStep(int timelineId, Func<bool> predicate, params Action[] actions)` 添加步骤
- `UpdateTimeline()` 更新所有时间线
- `ResetTimelineContext(int timelineId)` 重置时间线上下文
提供基于条件的任务调度：
- 支持多条时间线并行执行
- 基于谓词的条件触发
- 支持动态添加执行步骤
## 类型格式化
### 类型序列化
- `FormatType(Type type)` 格式化类型为字符串
- `LoadFromFormat(string data)` 从字符串加载类型
- `LoadFromFormat(string data, out Exception exception)` 安全加载类型
使用格式：`Assembly::FullTypeName`
## 内部管理
### 系统重置
- `InternalReset()` 重置所有内部状态
清理以下组件：
- 注册历史
- 未完成的目标
- 完成器回调
- 依赖关系
- 子对象容器
- 事件监听器
- 时间线队列

215
Convention/Runtime/Architecture.py
View File

@@ -1,24 +1,24 @@
from .Config import *
from .Reflection import *
from abs import ABC, abstractmethod
from abc import ABC, abstractmethod
class ISignal(ABC):
pass
class IModel(ABC):
@absstractmethod
@abstractmethod
def Save(self) -> str:
pass
@absstractmethod
@abstractmethod
def Load(self, data:str) -> None:
pass
class IConvertable[T](ABC):
@absstractmethod
def ConvertTo() -> T:
@abstractmethod
def ConvertTo(self) -> T:
pass
class IConvertModel[T](ABC, IConventable[T], IModel):
class IConvertModel[T](IConvertable[T], IModel):
pass
class SingletonModel[T](IModel):
@@ -26,22 +26,23 @@ class SingletonModel[T](IModel):
@staticmethod
def GetInstance(t:Typen[T]) -> T:
return _InjectInstances[t]
return SingletonModel._InjectInstances[t]
@staticmethod
def SetInstance(t:Typen[T], obj:T) -> None:
_InjectInstances[t] = obj
SingletonModel._InjectInstances[t] = obj
def __init__(self, t:Typen[T]) -> None:
self.typen: type = t
@override
def Save() -> str:
return GetInstance(self.typen).Save()
def Save(self) -> str:
return SingletonModel.GetInstance(self.typen).Save()
def DependenceModel(IConvertModel[bool]):
class DependenceModel(IConvertModel[bool]):
def __init__(self, queries:Sequence[IConvertModel[bool]]) -> None:
self.queries:list[IConventModel] = list(queries)
self.queries:list[IConvertModel[bool]] = list(queries)
@override
def ConvertTo(self) -> bool:
for query in self.queries:
if query.ConvertTo() == False:
@@ -49,38 +50,54 @@ def DependenceModel(IConvertModel[bool]):
return True
def __iter__(self):
return self.queries
return iter(self.queries)
def Load(self, data:str):
raise NotImplementedError()
def Save(self) -> str:
return NotImplement
raise NotImplementedError()
class Achitecture:
SignalListener = Callable[[ISignal], None]
class Architecture:
@staticmethod
def FormatType(t:type) -> str:
return f"{t.__module__}::{t.__name__}"
@staticmethod
def LoadFromFormat(data:str) -> type|None:
module,name = data.split("::")
return StringWithModule2Type(module,name)
def LoadFromFormat(data:str, exception:Exception|None=None) -> type|None:
try:
module,name = data.split("::")
return StringWithModel2Type(name, module=module)
except Exception as ex:
if exception is not None:
exception = ex
return None
@classmethod
def InternalReset(cls) -> None:
# Register System
cls._RegisterHistory.clear()
cls._UncompleteTargets.clear()
cls._Completer.clear()
cls._Dependences.clear()
cls._Childs.clear()
# Event Listener
cls._SignalListener.clear()
# Linear Chain for Dependence
cls._TimelineQueues.clear()
cls._TimelineContentID = 0
#region Objects Registered
class TypeQuery(IConvertModel[bool]):
def __init__(self, queryType:type) -> None:
self._quertType = quertType
self._queryType = queryType
@override
def ConvertTo(self) -> bool:
return self._queryType in Architectrue.Childs
return self._queryType in Architecture._Childs
def Load(self, data:str) -> None:
raise NotImplementedError()
@@ -100,31 +117,167 @@ class Achitecture:
@override
def ConvertTo(self) -> bool:
return self._registerSlot in Architecture.Childs
return self._registerSlot in Architecture._Childs
@override
def Load(self,data:str) -> None:
raise InvalidOperationError()
raise InvalidOperationError(f"Cannot use {self.__class__.__name__} to load type")
@override
def Save(self) -> str:
raise InalidOperationError()
return f"{Architecture.FormatType(self._registerSlot)}[{self.ConvertTo()}]"
@classmethod
def _InternalRegisteringComplete(cls) -> bool,Set[type]:
def _InternalRegisteringComplete(cls) -> tuple[bool,Set[type]]:
resultSet: Set[type] = set()
stats: bool = False
for dependence in cls._Dependences.keys:
if cls._Dependences[dependence].ConventTo():
resultSet.insert(dependence)
for dependence in cls._Dependences.keys():
if cls._Dependences[dependence].ConvertTo():
resultSet.add(dependence)
stats = True
return stats,resultSet
@classmethod
def _InternalRegisteringUpdate(cls, internalUpdateBuffer:Set[type]):
for complete in cls._InternalUpdateBuffer:
cls._Dependences.remove(complete)
# TODO
for complete in internalUpdateBuffer:
cls._Dependences.pop(complete, None)
for complete in internalUpdateBuffer:
cls._Completer[complete]()
cls._Completer.pop(complete, None)
for complete in internalUpdateBuffer:
cls._Childs[complete] = cls._UncompleteTargets[complete]
cls._UncompleteTargets.pop(complete, None)
@classmethod
def Register(cls, slot:type, target:object, completer:Action, *dependences:type) -> 'Architecture.Registering':
if slot in cls._RegisterHistory:
raise InvalidOperationError("Illegal duplicate registrations")
cls._RegisterHistory.add(slot)
cls._Completer[slot] = completer
cls._UncompleteTargets[slot] = target
# 过滤掉自身依赖
filtered_deps = [dep for dep in dependences if dep != slot]
type_queries = [cls.TypeQuery(dep) for dep in filtered_deps]
cls._Dependences[slot] = DependenceModel(type_queries)
while True:
has_complete, buffer = cls._InternalRegisteringComplete()
if not has_complete:
break
cls._InternalRegisteringUpdate(buffer)
return cls.Registering(slot)
@classmethod
def RegisterGeneric[T](cls, target:T, completer:Action, *dependences:type) -> 'Architecture.Registering':
return cls.Register(type(target), target, completer, *dependences)
@classmethod
def Contains(cls, type_:type) -> bool:
return type_ in cls._Childs
@classmethod
def ContainsGeneric[T](cls) -> bool:
return cls.Contains(type(T))
@classmethod
def InternalGet(cls, type_:type) -> object:
return cls._Childs[type_]
@classmethod
def Get(cls, type_:type) -> object:
return cls.InternalGet(type_)
@classmethod
def GetGeneric[T](cls) -> T:
return cls.Get(type(T))
#endregion
#region Signal & Update
_SignalListener: Dict[type, Set[SignalListener]] = {}
class Listening:
def __init__(self, action:SignalListener, type_:type):
self._action = action
self._type = type_
def StopListening(self):
if self._type in Architecture._SignalListener:
Architecture._SignalListener[self._type].discard(self._action)
@classmethod
def AddListenerGeneric[Signal](cls, slot:type, listener:SignalListener) -> 'Architecture.Listening':
if slot not in cls._SignalListener:
cls._SignalListener[slot] = set()
def action(signal:ISignal):
if isinstance(signal, slot):
listener(signal)
result = cls.Listening(action, slot)
cls._SignalListener[slot].add(action)
return result
@classmethod
def SendMessage(cls, slot:type, signal:ISignal):
if slot in cls._SignalListener:
for action in cls._SignalListener[slot]:
action(signal)
#endregion
#region Timeline/Chain & Update
class TimelineQueueEntry:
def __init__(self):
self.predicate: Callable[[], bool] = lambda: False
self.actions: List[Action] = []
class Timeline:
def __init__(self):
self.predicate_mapper: Dict[Callable[[], bool], int] = {}
self.queue: List[Architecture.TimelineQueueEntry] = []
self.context: int = 0
_TimelineQueues: Dict[int, Timeline] = {}
_TimelineContentID: int = 0
@classmethod
def CreateTimeline(cls) -> int:
cls._TimelineQueues[cls._TimelineContentID] = cls.Timeline()
cls._TimelineContentID += 1
return cls._TimelineContentID - 1
@classmethod
def AddStep(cls, timeline_id:int, predicate:Callable[[], bool], *actions:Action):
timeline = cls._TimelineQueues[timeline_id]
if predicate in timeline.predicate_mapper:
time = timeline.predicate_mapper[predicate]
timeline.queue[time].actions.extend(actions)
else:
time = len(timeline.queue)
timeline.predicate_mapper[predicate] = time
entry = cls.TimelineQueueEntry()
entry.predicate = predicate
entry.actions = list(actions)
timeline.queue.append(entry)
@classmethod
def UpdateTimeline(cls):
for timeline in cls._TimelineQueues.values():
if timeline.context < len(timeline.queue):
if timeline.queue[timeline.context].predicate():
for action in timeline.queue[timeline.context].actions:
action()
timeline.context += 1
@classmethod
def ResetTimelineContext(cls, timeline_id:int):
cls._TimelineQueues[timeline_id].context = 0
#endregion

21
Convention/Runtime/Config.py
View File

@@ -10,6 +10,20 @@ import time
import os
from colorama import Fore as ConsoleFrontColor, Back as ConsoleBackgroundColor, Style as ConsoleStyle
class NotImplementedError(Exception):
def __init__(self, message:Optional[str]=None) -> None:
if message is not None:
super().__init__(message)
else:
super().__init__()
class InvalidOperationError(Exception):
def __init__(self, message:Optional[str]=None) -> None:
if message is not None:
super().__init__(message)
else:
super().__init__()
def format_traceback_info(char:str='\n', back:int=1):
return char.join(traceback.format_stack()[:-back])
@@ -77,12 +91,7 @@ except ImportError:
type Typen[_T] = type
type Action[_T] = Callable[[_T], None]
type Action2[_T1, _T2] = Callable[[_T1, _T2], None]
type Action3[_T1, _T2, _T3] = Callable[[_T1, _T2, _T3], None]
type Action4[_T1, _T2, _T3, _T4] = Callable[[_T1, _T2, _T3, _T4], None]
type Action5[_T1, _T2, _T3, _T4, _T5] = Callable[[_T1, _T2, _T3, _T4, _T5], None]
type ActionW = Callable[[Sequence[Any]], None]
type Action = Callable[[], None]
type ClosuresCallable[_T] = Union[Callable[[Optional[None]], _T], Typen[_T]]
def AssemblyTypen(obj:Any) -> str:

10
Convention/Runtime/String.py
View File

@@ -1,13 +1,7 @@
from pathlib import Path
from .Config import *
import re
from pathlib import Path
import xml.etree.ElementTree as ET
from xml.dom import minidom
import math
def LimitStringLength(data, max_length:int=50) -> str:
s:str = data if data is str else str(data)
s:str = data if isinstance(data, str) else str(data)
if len(s) <= max_length:
return s
else:
@@ -24,7 +18,7 @@ def FillString(data:Any,
fill_char: str = " ",
side: Literal["left", "right", "center"] = "right"
) -> str:
s:str = data if data is str else str(data)
s:str = data if isinstance(data, str) else str(data)
char = fill_char[0]
if len(s) >= max_length:
return s