File size: 2,110 Bytes
3d3d712 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 |
from typing import Any, Callable, Dict, List, Optional, Type, Union
from .base import Plugin
__all__: List[str] = [
"register_plugin",
"test_plugin",
]
register_plugin_inner: Optional[Callable[[Type[Plugin]], None]] = None
def register_plugin(func: Union[Callable[..., Any], Type[Plugin]]):
"""
register a plugin, the plugin could be a class or a callable function
:param func: the plugin class or a callable function
"""
global register_plugin_inner
if "register_plugin_inner" not in globals() or register_plugin_inner is None:
print("no registry for loading plugin")
elif isinstance(func, type) and issubclass(func, Plugin):
register_plugin_inner(func)
elif callable(func):
func_name = func.__name__
def callable_func(self: Plugin, *args: List[Any], **kwargs: Dict[str, Any]):
self.log("info", "calling function " + func_name)
result = func(*args, **kwargs)
return result
wrapper_cls = type(
f"FuncPlugin_{func_name}",
(Plugin,),
{
"__call__": callable_func,
},
)
register_plugin_inner(wrapper_cls)
else:
raise Exception(
"only callable function or plugin class could be registered as Plugin",
)
return func
register_plugin_test_inner: Optional[Callable[[str, str, Callable[..., Any]], None]] = None
def test_plugin(name: Optional[str] = None, description: Optional[str] = None):
"""
register a plugin test
"""
def inner(func: Callable[..., Any]):
global register_plugin_test_inner
if "register_plugin_test_inner" not in globals() or register_plugin_test_inner is None:
print("no registry for loading plugin")
elif callable(func):
test_name: str = func.__name__ if name is None else name
test_description: str = func.__doc__ or "" if description is None else description
register_plugin_test_inner(test_name, test_description, func)
return func
return inner
|