删除了不常用的内容, 并使得Config能与3.10兼容

This commit is contained in:
2025-12-21 04:22:14 +08:00
parent efd826f677
commit 058975c37d
4 changed files with 33 additions and 89 deletions

View File

@@ -204,18 +204,17 @@ try:
except ImportError:
InternalImportingThrow("Internal", ["pydantic"])
type Typen[_T] = type
type Action = Callable[[], None]
type ClosuresCallable[_T] = Union[Callable[[Optional[None]], _T], Typen[_T]]
T = TypeVar("T")
Action = TypeVar("Action", bound=Callable[[], None])
ClosuresCallable = Union[Callable[[Optional[None]], Any], type]
def AssemblyTypen(obj:Any) -> str:
if isinstance(obj, type):
return f"{obj.__module__}.{obj.__name__}, "\
f"{obj.Assembly() if hasattr(obj, "Assembly") else "Global"}"
f"{obj.Assembly() if hasattr(obj, 'Assembly') else 'Global'}"
else:
return f"{obj.__class__.__module__}.{obj.__class__.__name__}, "\
f"{obj.GetAssembly() if hasattr(obj, "GetAssembly") else "Global"}"
f"{obj.GetAssembly() if hasattr(obj, 'GetAssembly') else 'Global'}"
def ReadAssemblyTypen(
assembly_typen: str,
*,
@@ -232,68 +231,13 @@ def ReadAssemblyTypen(
target_type = getattr(importlib.import_module(module_name), class_name)
return target_type, assembly_name
# using as c#: event
class ActionEvent[_Call:Callable]:
def __init__(self, actions:Sequence[_Call]):
super().__init__()
self._actions: List[Callable] = [action for action in actions]
self.call_indexs: List[int] = [i for i in range(len(actions))]
self.last_result: List[Any] = []
def CallFuncWithoutCallIndexControl(self, index:int, *args, **kwargs) -> Union[Any, Exception]:
try:
return self._actions[index](*args, **kwargs)
except Exception as ex:
return ex
def CallFunc(self, index:int, *args, **kwargs) -> Union[Any, Exception]:
return self.CallFuncWithoutCallIndexControl(self.call_indexs[index], *args, **kwargs)
def _InjectInvoke(self, *args, **kwargs):
result:List[Any] = []
for index in range(self.CallMaxCount):
result.append(self.CallFunc(index, *args, **kwargs))
return result
def Invoke(self, *args, **kwargs) -> Union[Self, bool]:
self.last_result = self._InjectInvoke(*args, **kwargs)
return self
def InitCallIndex(self):
self.call_indexs = [i for i in range(len(self._actions))]
def AddAction(self, action:_Call):
self._actions.append(action)
self.call_indexs.append(len(self._actions)-1)
return self
def AddActions(self, actions:Sequence[_Call]):
for action in actions:
self.AddAction(action)
return self
def _InternalRemoveAction(self, action:_Call):
if action in self._actions:
index = self._actions.index(action)
self._actions.remove(action)
self.call_indexs.remove(index)
for i in range(len(self.call_indexs)):
if self.call_indexs[i] > index:
self.call_indexs[i] -= 1
return True
return False
def RemoveAction(self, action:_Call):
while self._InternalRemoveAction(action):
pass
return self
def IsValid(self):
return not any(isinstance(x, Exception) for x in self.last_result)
def __bool__(self):
return self.IsValid()
@property
def CallMaxCount(self):
return len(self.call_indexs)
@property
def ActionCount(self):
return len(self._actions)
# region instance
# threads
class atomic[_T]:
_T = TypeVar("_T")
class atomic(Generic[_T]):
def __init__(
self,
value: _T,
@@ -320,13 +264,13 @@ class atomic[_T]:
return self.FetchAdd(value)
def __sub__(self, value:_T):
return self.FetchSub(value)
def __iadd__(self, value:_T) -> Self:
def __iadd__(self, value:_T) -> 'atomic[_T]':
self.FetchAdd(value)
return self
def __isub__(self, value:_T) -> Self:
def __isub__(self, value:_T) -> 'atomic[_T]':
self.FetchSub(value)
return self
def __enter__(self) -> Self:
def __enter__(self) -> 'atomic[_T]':
self._is_in_with = True
self.locker.acquire()
return self
@@ -482,7 +426,7 @@ class PlatformIndicator:
"""
return "Assets/"
class DescriptiveIndicator[T]:
class DescriptiveIndicator(Generic[T]):
def __init__(self, description:str, value:T) -> None:
self.descripion : str = description
self.value : T = value