DR_Team/Difficult-Rocket
DR_Team/Difficult-Rocket
6
2
Fork 0
Code Issues Pull Requests Projects Releases 25 Wiki Activity

add typing remove utils.new_thread

Browse Source
This commit is contained in:
shenjack 2023-05-12 21:06:56 +08:00
parent 2ae539974e
commit 582ce53c8a
7 changed files with 179 additions and 173 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
Difficult_Rocket/__init__.py
View File

@ -5,7 +5,6 @@
# -------------------------------
import sys
import warnings
import importlib
import traceback
import contextlib
@ -14,7 +13,7 @@ from pathlib import Path
from typing import Optional, List, Tuple
from Difficult_Rocket.api.types import Options
from Difficult_Rocket.utils.new_thread import new_thread
from Difficult_Rocket.utils.thread import new_thread
from libs.MCDR.version import Version

5
Difficult_Rocket/client/__init__.py
View File

@ -32,12 +32,11 @@ from Difficult_Rocket.utils import tools
from Difficult_Rocket.api.types import Options
from Difficult_Rocket.command import line, tree
from Difficult_Rocket.utils.translate import tr
from Difficult_Rocket import DR_runtime, DR_option
from Difficult_Rocket import DR_runtime
from Difficult_Rocket.api.screen import BaseScreen
from Difficult_Rocket.utils.new_thread import new_thread
from Difficult_Rocket.utils.thread import new_thread
from Difficult_Rocket.client.fps.fps_log import FpsLogger
from Difficult_Rocket.client.guis.widgets import InputBox
from Difficult_Rocket.exception.command import CommandError
from Difficult_Rocket.exception.language import LanguageNotFound
from Difficult_Rocket.client.screen import DRScreen, DRDEBUGScreen

2
Difficult_Rocket/command/line.py
View File

@ -27,7 +27,7 @@ from pyglet.graphics import Batch, Group
# from DR
from Difficult_Rocket.utils import translate
from Difficult_Rocket.command.api import CommandText
from Difficult_Rocket.utils.new_thread import new_thread
from Difficult_Rocket.utils.thread import new_thread
class CommandLineTextEntry(widgets.TextEntry):

155
Difficult_Rocket/utils/new_thread.py
View File

@ -1,155 +0,0 @@
# -------------------------------
# Difficult Rocket
# Copyright © 2020-2023 by shenjackyuanjie 3695888@qq.com
# All rights reserved
# -------------------------------
import functools
import inspect
import threading
from typing import Optional, Callable, Union, List
"""
This part of code come from MCDReforged(https://github.com/Fallen-Breath/MCDReforged)
Very thanks to Fallen_Breath and other coder who helped MCDR worked better
GNU Lesser General Public License v3.0（GNU LGPL v3)
(have some changes)
"""
__all__ = [
'new_thread',
'FunctionThread'
]
record_thread = False
record_destination: List[Callable[['FunctionThread'], None]] = []
def copy_signature(target: Callable, origin: Callable) -> Callable:
"""
Copy the function signature of origin into target
"""
# https://stackoverflow.com/questions/39926567/python-create-decorator-preserving-function-arguments
target.__signature__ = inspect.signature(origin)
return target
class FunctionThread(threading.Thread):
"""
A Thread subclass which is used in decorator :func:`new_thread` to wrap a synchronized function call
"""
__NONE = object()
def __init__(self, target, name, args, kwargs, daemon):
super().__init__(target=target, args=args, kwargs=kwargs, name=name, daemon=daemon)
self.__return_value = self.__NONE
self.__error = None
def wrapped_target(*args_, **kwargs_):
try:
self.__return_value = target(*args_, **kwargs_)
except Exception as e:
self.__error = e
raise e from None
self._target = wrapped_target
def get_return_value(self, block: bool = False, timeout: Optional[float] = None):
"""
Get the return value of the original function
If an exception has occurred during the original function call, the exception will be risen again here
Examples::
>>> import time
>>> @new_thread
... def do_something(text: str):
... time.sleep(1)
... return text
>>> do_something('task').get_return_value(block=True)
'task'
:param block: If it should join the thread before getting the return value to make sure the function invocation finishes
:param timeout: The maximum timeout for the thread join
:raise RuntimeError: If the thread is still alive when getting return value. Might be caused by ``block=False``
while the thread is still running, or thread join operation times out
:return: The return value of the original function
"""
if block:
self.join(timeout)
if self.__return_value is self.__NONE:
if self.is_alive():
raise RuntimeError('The thread is still running')
raise self.__error
return self.__return_value
def new_thread(arg: Optional[Union[str, Callable]] = None,
daemon: bool = False,
log_thread: bool = True):
"""
This is a one line solution to make your function executes in parallels.
When decorated with this decorator, functions will be executed in a new daemon thread
This decorator only changes the return value of the function to the created ``Thread`` object.
Beside the return value, it reserves all signatures of the decorated function,
so you can safely use the decorated function as if there's no decorating at all
It's also a simple compatible upgrade method for old MCDR 0.x plugins
The return value of the decorated function is changed to the ``Thread`` object that executes this function
The decorated function has 1 extra field:
* ``original`` field: The original undecorated function
Examples::
>>> import time
>>> @new_thread('My Plugin Thread')
... def do_something(text: str):
... time.sleep(1)
... print(threading.current_thread().name)
>>> callable(do_something.original)
True
>>> t = do_something('foo')
>>> isinstance(t, FunctionThread)
True
>>> t.join()
My Plugin Thread
:param arg: A :class:`str`, the name of the thread. It's recommend to specify the thread name, so when you
log something by ``server.logger``, a meaningful thread name will be displayed
instead of a plain and meaningless ``Thread-3``
:param daemon: If the thread should be a daemon thread
:param log_thread: If the thread should be logged to callback defined in record_destination
"""
def wrapper(func):
@functools.wraps(func) # to preserve the origin function information
def wrap(*args, **kwargs):
thread = FunctionThread(target=func, args=args, kwargs=kwargs, name=thread_name, daemon=daemon)
if record_thread:
for destination in record_destination:
destination(thread)
thread.start()
return thread
# bring the signature of the func to the wrap function
# so inspect.getfullargspec(func) works correctly
copy_signature(wrap, func)
wrap.original = func # access this field to get the original function
return wrap
# Directly use @new_thread without ending brackets case, e.g. @new_thread
if isinstance(arg, Callable):
thread_name = None
return wrapper(arg)
# Use @new_thread with ending brackets case, e.g. @new_thread('A'), @new_thread()
else:
thread_name = arg
return wrapper

26
Difficult_Rocket/utils/options.py
View File

@ -41,10 +41,14 @@ class Options:
"""
一个用于存储选项 / 提供 API 定义 的类
用法:
继承 Options 类
在类里定义 option: typing
(可选 定义 name: str = 'Option Base' 用于在打印的时候显示名字)
存储配置: 继承 Options 类
在类里定义 option: typing
(可选 定义 name: str = 'Option Base' 用于在打印的时候显示名字)
提供 API 接口: 继承 Options 类
在类里定义 option: typing
定义 一些需要的方法
子类: 继承 新的 Options 类
实现定义的方法
"""
name = 'Option Base'
cached_options: Dict[str, Union[str, Any]] = {}
@ -130,6 +134,10 @@ class Options:
return self.cached_options
def option_with_len(self) -> Tuple[List[Tuple[str, Union[Any, Type], Type]], int, int, int]:
"""
返回一个可以用于打印的 option 列表
:return:
"""
options = self.flush_option()
max_len_key = 1
max_len_value = 1
@ -144,6 +152,10 @@ class Options:
return option_list, max_len_key, max_len_value, max_len_value_t
def as_markdown(self) -> str:
"""
返回一个 markdown 格式的 option 字符串
:return: markdown 格式的 option 字符串
"""
value = self.option_with_len()
cache = StringIO()
option_len = max(value[1], len('Option'))
@ -161,6 +173,12 @@ class Options:
@classmethod
def add_option(cls, name: str, value: Union[Callable, object]) -> Dict:
"""
向配置类中添加一个额外的配置
:param name: 配置的名字
:param value: 用于获取配置的函数或者类
:return: 配置类的所有配置
"""
if not hasattr(cls, '_options'):
cls._options: Dict[str, Union[Callable, object]] = {}
cls._options[name] = value

157
Difficult_Rocket/utils/thread.py
View File

@ -11,20 +11,24 @@ github: @shenjackyuanjie
gitee: @shenjackyuanjie
"""
import functools
import inspect
import threading
from typing import Union
from threading import Lock
from typing import Optional, Callable, Union, List
from Difficult_Rocket import DR_option, crash
from Difficult_Rocket.exception.threading import LockTimeOutError
__all__ = [
'new_thread',
'FunctionThread',
'ThreadLock',
'record_thread',
'record_destination',
]
class Threads(threading.Thread):
def run(self):
if DR_option.record_thread:
crash.all_thread.append(self)
super().run()
record_thread = False
record_destination: List[Callable[['FunctionThread'], None]] = []
class ThreadLock:
@ -44,6 +48,143 @@ class ThreadLock:
self.lock.release()
"""
This part of code come from MCDReforged(https://github.com/Fallen-Breath/MCDReforged)
Very thanks to Fallen_Breath and other coder who helped MCDR worked better
GNU Lesser General Public License v3.0（GNU LGPL v3)
(have some changes)
"""
def copy_signature(target: Callable, origin: Callable) -> Callable:
"""
Copy the function signature of origin into target
"""
# https://stackoverflow.com/questions/39926567/python-create-decorator-preserving-function-arguments
target.__signature__ = inspect.signature(origin)
return target
class FunctionThread(threading.Thread):
"""
A Thread subclass which is used in decorator :func:`new_thread` to wrap a synchronized function call
"""
__NONE = object()
def __init__(self, target, name, args, kwargs, daemon):
super().__init__(target=target, args=args, kwargs=kwargs, name=name, daemon=daemon)
self.__return_value = self.__NONE
self.__error = None
def wrapped_target(*args_, **kwargs_):
try:
self.__return_value = target(*args_, **kwargs_)
except Exception as e:
self.__error = e
raise e from None
self._target = wrapped_target
def get_return_value(self, block: bool = False, timeout: Optional[float] = None):
"""
Get the return value of the original function
If an exception has occurred during the original function call, the exception will be risen again here
Examples::
>>> import time
>>> @new_thread
... def do_something(text: str):
... time.sleep(1)
... return text
>>> do_something('task').get_return_value(block=True)
'task'
:param block: If it should join the thread before getting the return value to make sure the function invocation finishes
:param timeout: The maximum timeout for the thread join
:raise RuntimeError: If the thread is still alive when getting return value. Might be caused by ``block=False``
while the thread is still running, or thread join operation times out
:return: The return value of the original function
"""
if block:
self.join(timeout)
if self.__return_value is self.__NONE:
if self.is_alive():
raise RuntimeError('The thread is still running')
raise self.__error
return self.__return_value
def new_thread(arg: Optional[Union[str, Callable]] = None,
daemon: bool = False,
log_thread: bool = True):
"""
This is a one line solution to make your function executes in parallels.
When decorated with this decorator, functions will be executed in a new daemon thread
This decorator only changes the return value of the function to the created ``Thread`` object.
Beside the return value, it reserves all signatures of the decorated function,
so you can safely use the decorated function as if there's no decorating at all
It's also a simple compatible upgrade method for old MCDR 0.x plugins
The return value of the decorated function is changed to the ``Thread`` object that executes this function
The decorated function has 1 extra field:
* ``original`` field: The original undecorated function
Examples::
>>> import time
>>> @new_thread('My Plugin Thread')
... def do_something(text: str):
... time.sleep(1)
... print(threading.current_thread().name)
>>> callable(do_something.original)
True
>>> t = do_something('foo')
>>> isinstance(t, FunctionThread)
True
>>> t.join()
My Plugin Thread
:param arg: A :class:`str`, the name of the thread. It's recommend to specify the thread name, so when you
log something by ``server.logger``, a meaningful thread name will be displayed
instead of a plain and meaningless ``Thread-3``
:param daemon: If the thread should be a daemon thread
:param log_thread: If the thread should be logged to callback defined in record_destination
"""
def wrapper(func):
@functools.wraps(func) # to preserve the origin function information
def wrap(*args, **kwargs):
thread = FunctionThread(target=func, args=args, kwargs=kwargs, name=thread_name, daemon=daemon)
if record_thread:
for destination in record_destination:
destination(thread)
thread.start()
return thread
# bring the signature of the func to the wrap function
# so inspect.getfullargspec(func) works correctly
copy_signature(wrap, func)
wrap.original = func # access this field to get the original function
return wrap
# Directly use @new_thread without ending brackets case, e.g. @new_thread
if isinstance(arg, Callable):
thread_name = None
return wrapper(arg)
# Use @new_thread with ending brackets case, e.g. @new_thread('A'), @new_thread()
else:
thread_name = arg
return wrapper
if __name__ == "__main__":
from Difficult_Rocket.exception import TestError

4
docs/src/update_logs.md
View File

@ -74,6 +74,10 @@
- Remove `write_options` method
- Remove `write_markdown_tablet` method
- Replace with `Option().as_markdown()`
- `Difficult_Rocket.utils.new_thread`
- Moved to `Diffiuclt_Rocket.utils.thread`
- `Difficult_Rocket.utils.thread`
- Remove `Threads`
### Changes