peterlu14
1788 lines
51 KiB
Python
1788 lines
51 KiB
Python
"""Application Command Line Interface.
|
|
|
|
Use Example
|
|
-----------
|
|
|
|
Create CliMain Class
|
|
~~~~~~~~~~~~~~~~~~~~
|
|
|
|
::
|
|
|
|
from biopro.util.cli import *
|
|
|
|
|
|
class Main(CliMain):
|
|
def __init__():
|
|
super().__init__()
|
|
|
|
def run(self):
|
|
pass
|
|
|
|
|
|
if __name__ == '__main__':
|
|
Main().main()
|
|
|
|
**Run CliMain program**
|
|
|
|
::
|
|
|
|
$ python main.py
|
|
|
|
Add help options
|
|
~~~~~~~~~~~~~~~~
|
|
|
|
::
|
|
|
|
class Main(CliMain):
|
|
|
|
...
|
|
|
|
@cli_flags('-h', '--help', force_return=True)
|
|
def _help(self, opt: str):
|
|
'''print help document'''
|
|
print_help(self)
|
|
|
|
...
|
|
|
|
you will see help document add a new options ::
|
|
|
|
$ python main.py -h
|
|
|
|
MAIN.py [OPTIONS]
|
|
|
|
OPTIONS
|
|
-h, --help print help document
|
|
|
|
|
|
More options
|
|
~~~~~~~~~~~~
|
|
|
|
::
|
|
|
|
class Main(CliMain):
|
|
...
|
|
|
|
@cli_flags('-f', '--flag')
|
|
def _flag(self, opt: str):
|
|
'''program flag'''
|
|
pass
|
|
|
|
@cli_options('-o', '--options', value='VALUE')
|
|
def _options(self: opt: str, value: str):
|
|
'''program options with arguments'''
|
|
pass
|
|
|
|
@cli_arguments(0, value='FILE')
|
|
def _file(self, pos: int, value: str):
|
|
'''program arguments'''
|
|
pass
|
|
|
|
@cli_arguments('*', value='[FILE...]')
|
|
def _left(self, pos: int, value: str):
|
|
'''program left arguments'''
|
|
pass
|
|
|
|
...
|
|
|
|
Add sub command
|
|
~~~~~~~~~~~~~~~
|
|
|
|
::
|
|
|
|
class Main(CliMain):
|
|
def __init__():
|
|
super().__init__()
|
|
|
|
@cli_command('sub')
|
|
def _sub(self, cmd: str, argv: List[str]):
|
|
'''program sub command'''
|
|
return _SubMain
|
|
|
|
def run(self):
|
|
pass
|
|
|
|
class _SubMain(CliSubCommandMain):
|
|
def __init__(self, command: str):
|
|
super().__init__(command)
|
|
|
|
def run(self):
|
|
pass
|
|
|
|
REPL CliMain
|
|
~~~~~~~~~~~~
|
|
|
|
::
|
|
|
|
from biopro.util.cli import *
|
|
|
|
class Main(CliReplMain):
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
# program options
|
|
|
|
@cli_flags('-h', '--help', force_return=True)
|
|
def _help(self, opt: str):
|
|
'''print help document'''
|
|
print_help(self)
|
|
|
|
# program repl context manager (optional)
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, ext_type, exc_val, exc_tb):
|
|
pass
|
|
|
|
# repl command
|
|
|
|
@cli_command('h', 'help')
|
|
def _repl_help(self, cmd: str, argv: List[str]):
|
|
'''print help documents'''
|
|
print_help(self)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
Main().main()
|
|
|
|
help document ::
|
|
|
|
$ python main.py -h
|
|
|
|
MAIN.py [OPTIONS]
|
|
|
|
OPTIONS
|
|
-h, --help print help document
|
|
|
|
COMMANDS
|
|
h, help print help documents
|
|
|
|
run program ::
|
|
|
|
$ python main.py
|
|
>>> help
|
|
MAIN.py [OPTIONS]
|
|
|
|
OPTIONS
|
|
-h, --help print help document
|
|
|
|
COMMANDS
|
|
h, help print help documents
|
|
|
|
>>> # (ctrl+D)
|
|
$
|
|
|
|
Document
|
|
~~~~~~~~
|
|
|
|
::
|
|
|
|
class Main(CliMain):
|
|
'''program header
|
|
|
|
program footer
|
|
'''
|
|
|
|
@cli_flags('-h', '--help', force_return=True)
|
|
def _help(self, opt: str):
|
|
'''options documents'''
|
|
print_help(self)
|
|
|
|
output example ::
|
|
|
|
$ python main.py -h
|
|
|
|
MAIN.py [OPTIONS]
|
|
|
|
program header
|
|
|
|
OPTIONS
|
|
-h, --help print help document
|
|
|
|
program footer
|
|
|
|
|
|
Help Document Helper
|
|
~~~~~~~~~~~~~~~~~~~~
|
|
|
|
::
|
|
|
|
from biopro.util.cli import *
|
|
|
|
class Main(CliMain):
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
def run():
|
|
helper = CliMainHelp(self, None) # output to stdout
|
|
helper.print_usage(self, '[OPTIONS]', 'COMMANDS')
|
|
|
|
if self.__doc__ is not None:
|
|
helper.print_newline()
|
|
helper.print_header()
|
|
|
|
helper.print_newline()
|
|
helper.print_help()
|
|
|
|
if self.__doc__ is not None:
|
|
helper.print_newline()
|
|
helper.print_footer()
|
|
|
|
Option class
|
|
~~~~~~~~~~~~
|
|
|
|
::
|
|
|
|
from biopro.util.cli import *
|
|
|
|
class Options(CliOptions):
|
|
def __init__(self):
|
|
self.value = None
|
|
|
|
@cli_options('-o', '--options', value='VALUE')
|
|
def _options(self: opt: str, value: str):
|
|
'''program options with arguments'''
|
|
self.value = value
|
|
|
|
# extend/compose options
|
|
|
|
class Main(CliMain):
|
|
def __init__(self):
|
|
super().__init__()
|
|
|
|
self.options = self.extend_options(Options(),
|
|
help_section='another options')
|
|
|
|
@cli_flags('-h', '--help', force_return=True)
|
|
def _help(self, opt: str):
|
|
'''options documents'''
|
|
print_help(self)
|
|
|
|
def run(self):
|
|
print(self.options.value)
|
|
|
|
if __name__ == '__main__':
|
|
Main().main()
|
|
|
|
document output ::
|
|
|
|
$ python main.py -h
|
|
|
|
MAIN.py [OPTIONS]
|
|
|
|
OPTIONS
|
|
-h, --help print help document
|
|
|
|
another options
|
|
-o, --options VALUE program options with arguments
|
|
|
|
method override
|
|
~~~~~~~~~~~~~~~
|
|
|
|
::
|
|
|
|
from biopro.util.cli import *
|
|
|
|
class Options(CliOptions):
|
|
def __init__(self):
|
|
self.value = None
|
|
self.value2 = None
|
|
|
|
@cli_options('-o', '--options', value='VALUE')
|
|
def _options(self: opt: str, value: str):
|
|
'''override example 1'''
|
|
self.value = value
|
|
|
|
@cli_options('-p', value='VALUE')
|
|
def _options2(self: opt: str, value: str):
|
|
'''override example 2'''
|
|
self.set_value2(value)
|
|
|
|
def set_value2(self, value):
|
|
self.value2 = value
|
|
|
|
class Main(CliMain, Options):
|
|
def __init__(self):
|
|
super().__init__()
|
|
Options.__init__(self)
|
|
|
|
@cli_options('-o', '--options', value='VALUE')
|
|
def _options(self: opt: str, value: str):
|
|
Options._options(self, opt, value)
|
|
# do something more
|
|
|
|
_options.__doc__ = Options._options.__doc__
|
|
|
|
def set_value2(self, value):
|
|
Options.set_value2(self, value)
|
|
# do something more
|
|
|
|
def run(self):
|
|
print(self.value)
|
|
print(self.value2)
|
|
|
|
dynamic extend cli flags/options
|
|
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
|
|
|
|
::
|
|
|
|
class Main(CliMain, Options):
|
|
def __init__(self):
|
|
super().__init__()
|
|
Options.__init__(self)
|
|
|
|
@cli_options('-o', '--options', value='VALUE')
|
|
def _options(self: opt: str, value: str):
|
|
|
|
def _callback(flag:str, value:Optional[str]):
|
|
print(flag, '=', value)
|
|
|
|
self.extend_options_callback(_callback,
|
|
CliHandleOption('--more'))
|
|
|
|
def run(self):
|
|
pass
|
|
|
|
Class
|
|
-----
|
|
|
|
"""
|
|
|
|
import inspect
|
|
import re
|
|
# noinspection PyUnresolvedReferences
|
|
import readline
|
|
import sys
|
|
from collections import defaultdict
|
|
from functools import wraps
|
|
from itertools import takewhile, dropwhile
|
|
from typing import Dict, Iterable, IO, overload
|
|
|
|
from .cli_base import *
|
|
from .json import JSON, JSON_ARRAY
|
|
from .stack import print_exception
|
|
from .text import part_suffix
|
|
|
|
__all__ = ['cli_command', 'cli_flags', 'cli_options', 'cli_arguments', 'cli_help_section', 'cli_disable',
|
|
'CliHandleFlag', 'CliHandleOption', 'CliHandleArgument', 'CliHandleCommand',
|
|
'CliOptions', 'CliMain', 'CliReplMain', 'CliSubCommandMain', 'Arguments',
|
|
'CliCommandDelegateMacro']
|
|
|
|
C = TypeVar('C', bound='CliOptions')
|
|
|
|
|
|
class CliMain(AbstractCliMain, metaclass=abc.ABCMeta):
|
|
"""cli main."""
|
|
|
|
__slots__ = ('_cli_flags', '_cli_options', '_cli_arguments', '_cli_commands',
|
|
'__cli_help_options', '__cli_help_argument', '__cli_help_subcommand', '__cli_help_document',
|
|
'__cli_options_opts')
|
|
|
|
def __init__(self):
|
|
# command line interface
|
|
self._cli_flags: Dict[str, CliHandleFlag] = {}
|
|
self._cli_options: Dict[str, CliHandleOption] = {}
|
|
self._cli_arguments: Dict[Union[str, int], CliHandleArgument] = {}
|
|
self._cli_commands: Dict[str, CliHandleCommand] = {}
|
|
|
|
# help
|
|
self.__cli_help_options: Dict[Optional[str], List[CliHandle]] = defaultdict(list)
|
|
self.__cli_help_argument: List[CliHandleArgument] = []
|
|
self.__cli_help_subcommand: List[CliHandleCommand] = []
|
|
self.__cli_help_document: Dict[CliHandle, str] = {}
|
|
|
|
# options remap
|
|
self.__cli_options_opts: Dict[CliHandle, Dict[str, Optional[str]]] = defaultdict(dict)
|
|
|
|
# set default help section
|
|
self.__cli_help_options[None].clear()
|
|
|
|
# parsing cli handle on self
|
|
self._extend_options(self, parse_subcommand=True, parse_arguments=True)
|
|
|
|
def extend_options_callback(self,
|
|
callback: Callable[[str, Optional[str]], None],
|
|
*handle: CliHandle,
|
|
help_section: str = None):
|
|
"""extend cli options with *callback*.
|
|
|
|
:param callback: option handle callback
|
|
:param handle: cli handle instance, could be :class:`CliHandleFlag` and :class:`CliHandleOption`
|
|
:param help_section: help section title
|
|
"""
|
|
for h in handle:
|
|
if isinstance(h, CliHandleFlag):
|
|
h = h.bind(None)
|
|
|
|
for opt in h.flags:
|
|
self._cli_flags[opt] = h
|
|
|
|
h.handle = lambda target, flag: callback(flag, None)
|
|
|
|
s = h.help_section
|
|
if s is None:
|
|
s = help_section
|
|
|
|
self.__cli_help_options[s].append(h)
|
|
self.__cli_help_document[h] = h.help_doc
|
|
|
|
elif isinstance(h, CliHandleOption):
|
|
h = h.bind(None)
|
|
|
|
for opt in h.options:
|
|
self._cli_options[opt] = h
|
|
|
|
h.handle = lambda target, flag, value: callback(flag, value)
|
|
|
|
s = h.help_section
|
|
if s is None:
|
|
s = help_section
|
|
|
|
self.__cli_help_options[s].append(h)
|
|
self.__cli_help_document[h] = h.help_doc
|
|
|
|
else:
|
|
raise RuntimeError('extend_options_callback do not support : ' + h.__class__.__name__)
|
|
|
|
def extend_options(self,
|
|
options: C,
|
|
help_section: Optional[str] = None,
|
|
options_mapper: Dict[str, Optional[str]] = None,
|
|
options_default: Dict[str, str] = None) -> C:
|
|
"""extend cli options.
|
|
|
|
:param options: cli options
|
|
:param help_section: help section
|
|
:param options_mapper: option name remap. with dict {old_name: new_name}, None new_name for blocking.
|
|
:param options_default: replace text in help document. with dict {word: value}
|
|
:return: *options*
|
|
"""
|
|
self._extend_options(options,
|
|
help_section,
|
|
options_mapper,
|
|
options_default,
|
|
# prohibit subcommand and argument for external options
|
|
parse_subcommand=False,
|
|
parse_arguments=False)
|
|
|
|
return options
|
|
|
|
def _extend_options(self,
|
|
target: Any,
|
|
help_section: Optional[str] = None,
|
|
options_mapper: Optional[Dict[str, Optional[str]]] = None,
|
|
options_default: Optional[Dict[str, str]] = None,
|
|
parse_subcommand=False,
|
|
parse_arguments=False):
|
|
"""
|
|
|
|
:param target: reflect target
|
|
:param help_section: help section
|
|
:param options_mapper: option name remap. with dict {old_name: new_name}, None new_name for blocking.
|
|
:param options_default: replace text in help document. with dict {word: value}
|
|
:param parse_subcommand: parse :func:`cli_command`
|
|
:param parse_arguments: parse :func:`cli_arguments`
|
|
"""
|
|
|
|
target_type = type(target)
|
|
doc_replace = {}
|
|
|
|
# find constant value in *target* for help document replacement
|
|
for attr_name in dir(target_type):
|
|
if re.match(r'[A-Z0-9_]+', attr_name):
|
|
attr = getattr(target_type, attr_name)
|
|
if attr is None:
|
|
doc_replace['%' + attr_name + '%'] = 'None'
|
|
elif isinstance(attr, (bool, int, float, str)):
|
|
doc_replace['%' + attr_name + '%'] = str(attr)
|
|
|
|
if options_default is not None:
|
|
for text, replace in options_default.items():
|
|
doc_replace['%' + text + '%'] = replace
|
|
|
|
def _resolve_doc(_doc: Optional[str]) -> Optional[str]:
|
|
|
|
if _doc is not None:
|
|
for _text, _replace in doc_replace.items():
|
|
_doc = _doc.replace(_text, _replace)
|
|
|
|
return _doc
|
|
|
|
for attr_name in dir(target_type):
|
|
attr = getattr(target_type, attr_name, None)
|
|
|
|
# attr should be a method
|
|
if attr is None or not callable(attr):
|
|
continue
|
|
|
|
handle: Optional[CliHandle] = getattr(attr, '_cli_handle_', None)
|
|
|
|
# skip disabled
|
|
if handle is None or isinstance(handle, CliHandleDisable):
|
|
continue
|
|
|
|
if isinstance(handle, CliHandleCommand):
|
|
# cli command handle
|
|
|
|
if not parse_subcommand:
|
|
continue
|
|
|
|
handle: CliHandleCommand = handle.bind(target)
|
|
|
|
# check command collision
|
|
for command in handle.command:
|
|
if command in self._cli_commands:
|
|
raise RuntimeError('sub-command collision : ' + command)
|
|
|
|
self._cli_commands[command] = handle
|
|
|
|
# always add into help
|
|
self.__cli_help_subcommand.append(handle)
|
|
self.__cli_help_document[handle] = _resolve_doc(handle.help_doc)
|
|
|
|
elif isinstance(handle, CliHandleFlag):
|
|
# cli flag handle
|
|
|
|
add_help = False
|
|
|
|
handle: CliHandleFlag = handle.bind(target)
|
|
|
|
# flag name remap
|
|
if options_mapper is not None:
|
|
for opt in handle.flags:
|
|
new_opt = options_mapper.get(opt, opt)
|
|
self.__cli_options_opts[handle][opt] = new_opt
|
|
|
|
if new_opt is not None:
|
|
add_help = True
|
|
self._cli_flags[new_opt] = handle
|
|
|
|
else:
|
|
for opt in handle.flags:
|
|
add_help = True
|
|
self._cli_flags[opt] = handle
|
|
|
|
if add_help:
|
|
_h = handle.help_section
|
|
if _h is None:
|
|
_h = help_section
|
|
|
|
self.__cli_help_options[_h].append(handle)
|
|
self.__cli_help_document[handle] = _resolve_doc(handle.help_doc)
|
|
|
|
elif isinstance(handle, CliHandleOption):
|
|
# cli option handle
|
|
|
|
add_help = False
|
|
|
|
handle: CliHandleOption = handle.bind(target)
|
|
|
|
# option name remap
|
|
if options_mapper is not None:
|
|
for opt in handle.options:
|
|
new_opt = options_mapper.get(opt, opt)
|
|
self.__cli_options_opts[handle][opt] = new_opt
|
|
|
|
if new_opt is not None:
|
|
add_help = True
|
|
self._cli_options[new_opt] = handle
|
|
else:
|
|
for opt in handle.options:
|
|
add_help = True
|
|
self._cli_options[opt] = handle
|
|
|
|
if add_help:
|
|
_h = handle.help_section
|
|
if _h is None:
|
|
_h = help_section
|
|
|
|
self.__cli_help_options[_h].append(handle)
|
|
self.__cli_help_document[handle] = _resolve_doc(handle.help_doc)
|
|
|
|
elif isinstance(handle, CliHandleArgument):
|
|
# cli argument handle
|
|
|
|
if not parse_arguments:
|
|
continue
|
|
|
|
handle: CliHandleArgument = handle.bind(target)
|
|
pos = handle.position
|
|
|
|
if pos in self._cli_arguments:
|
|
raise RuntimeError('positional arguments pos collision : ' + str(pos))
|
|
|
|
if (pos == '*' or pos == '+') and ('*' in self._cli_arguments or '+' in self._cli_arguments):
|
|
raise RuntimeError('positional arguments left-pos collision : ' + str(pos))
|
|
|
|
self._cli_arguments[pos] = handle
|
|
|
|
# always add into help
|
|
self.__cli_help_argument.append(handle)
|
|
self.__cli_help_document[handle] = _resolve_doc(handle.help_doc)
|
|
|
|
def main(self, argv: List[str] = None):
|
|
if argv is None:
|
|
argv = sys.argv[1:]
|
|
|
|
try:
|
|
argv = self.parsing(argv)
|
|
|
|
if argv is None:
|
|
self.run()
|
|
else:
|
|
command, *argv = argv
|
|
|
|
self.invoke_subcommand(command, argv)
|
|
|
|
except MainForceReturn:
|
|
return
|
|
|
|
except (SystemExit, KeyboardInterrupt):
|
|
# do not catch this error
|
|
raise
|
|
|
|
except:
|
|
print_exception()
|
|
|
|
def parsing(self,
|
|
argv: List[str],
|
|
parsing_sub_command=True,
|
|
on_unknown_argument: Callable[[int, str], bool] = None) -> Optional[List[str]]:
|
|
left_all = False
|
|
i = 0
|
|
j = 0
|
|
|
|
while i < len(argv):
|
|
arg = argv[i]
|
|
|
|
if not left_all and arg.startswith('-'):
|
|
|
|
# flags
|
|
if self._handle_flags_(arg):
|
|
i += 1
|
|
continue
|
|
|
|
# options
|
|
ret = self._handle_options_(arg, i, argv)
|
|
if ret > 0:
|
|
i += ret
|
|
continue
|
|
|
|
# not above of all
|
|
if on_unknown_argument is None or not on_unknown_argument(i, arg):
|
|
raise ValueError('unknown options : ' + arg)
|
|
else:
|
|
i += 1
|
|
|
|
else:
|
|
# sub command
|
|
if parsing_sub_command and j == 0:
|
|
if arg in self._cli_commands:
|
|
return argv[i:]
|
|
|
|
# arguments
|
|
ret = self._handle_arguments_(arg, j)
|
|
if ret is not None:
|
|
i += 1
|
|
j += 1
|
|
left_all = left_all or ret
|
|
continue
|
|
|
|
# not above of all
|
|
if on_unknown_argument is None or not on_unknown_argument(i, arg):
|
|
if parsing_sub_command and j == 0:
|
|
raise ValueError('unknown sub-command : ' + arg)
|
|
else:
|
|
raise ValueError('unknown arguments : ' + arg)
|
|
else:
|
|
i += 1
|
|
|
|
return None
|
|
|
|
def _handle_flags_(self, arg: str) -> bool:
|
|
"""
|
|
:param arg: current argument token
|
|
:return: consumed or not
|
|
:raises MainForceReturn: if _cli_flags_force_return_ enable
|
|
"""
|
|
handle = self._cli_flags.get(arg, None)
|
|
|
|
if handle is None:
|
|
# handle not found
|
|
return False
|
|
|
|
# print('_handle_flags_', arg)
|
|
handle.invoke(arg)
|
|
|
|
if handle.force_return:
|
|
raise MainForceReturn()
|
|
|
|
return True
|
|
|
|
def _handle_options_(self, arg: str, index: int, argv: List[str]) -> int:
|
|
"""
|
|
|
|
:param arg: current argument token
|
|
:param index: current token index
|
|
:param argv: command line arguments
|
|
:return: number of arguments consumed, 0 if fail.
|
|
"""
|
|
arg, value = part_suffix(arg, '=')
|
|
|
|
handle = self._cli_options.get(arg, None)
|
|
|
|
if handle is None:
|
|
# wild options
|
|
handle = self._cli_options.get('--', None)
|
|
|
|
if handle is None:
|
|
# handle not found
|
|
return 0
|
|
|
|
ret = 1
|
|
|
|
if not handle.optional and value is None:
|
|
try:
|
|
value = argv[index + 1]
|
|
ret = 2
|
|
except IndexError as e:
|
|
raise RuntimeError(arg + ' lost ' + handle.value_text) from e
|
|
|
|
# print('_handle_options_', arg, value)
|
|
handle.invoke(arg, value)
|
|
|
|
return ret
|
|
|
|
def _handle_arguments_(self, arg: str, pos: int) -> Optional[bool]:
|
|
"""
|
|
|
|
:param arg: current argument token
|
|
:param pos: current argument index
|
|
:return: None if fail, true if left all.
|
|
"""
|
|
a = self._cli_arguments
|
|
handle = a.get(pos, None)
|
|
|
|
if handle is None and pos == 0:
|
|
handle = a.get('?', None)
|
|
|
|
if handle is None:
|
|
handle = a.get('*', None)
|
|
|
|
if handle is None:
|
|
handle = a.get('+', None)
|
|
|
|
if handle is None:
|
|
# handle not found
|
|
return None
|
|
|
|
# print('_handle_arguments_', arg, pos)
|
|
handle.invoke(pos, arg)
|
|
|
|
return handle.left_all
|
|
|
|
def contain_subcommand(self, command: str) -> bool:
|
|
"""does this main has sub command *command*?
|
|
|
|
:param command: sub-command name
|
|
:return:
|
|
"""
|
|
return command in self._cli_commands
|
|
|
|
def invoke_subcommand(self, command: str, argv: List[str]) -> None:
|
|
"""invoke sub-command with arguments.
|
|
|
|
:param command: sub-command name
|
|
:param argv: command arguments.
|
|
"""
|
|
main = self.find_subcommand(command, argv)
|
|
|
|
if main is None:
|
|
return
|
|
|
|
try:
|
|
main.main(argv)
|
|
|
|
except (MainForceReturn, SystemExit, KeyboardInterrupt):
|
|
raise
|
|
|
|
except BaseException as e:
|
|
raise RuntimeError('sub-command ' + command, *e.args) from e
|
|
|
|
def find_subcommand(self, command: str, argv: List[str]) -> Optional['AbstractCliMain']:
|
|
"""find sub-command with arguments.
|
|
|
|
:param command: sub-command name
|
|
:param argv: command arguments.
|
|
:return:
|
|
"""
|
|
handle = self._cli_commands.get(command, None)
|
|
|
|
if handle is None:
|
|
raise RuntimeError('sub-command not found : ' + command)
|
|
|
|
return handle.invoke(command, argv)
|
|
|
|
def print_help(self, print_usage=True, file: Optional[IO] = None) -> None:
|
|
"""print help document.
|
|
|
|
:param print_usage: print usage line
|
|
:param file: output file stream. if None, use stdout.
|
|
"""
|
|
if file is None:
|
|
file = sys.stdout
|
|
|
|
if print_usage:
|
|
self.print_help_usage_default(file)
|
|
|
|
main_doc = self.__doc__
|
|
|
|
if main_doc is not None:
|
|
print(file=file)
|
|
self.print_help_header(file=file)
|
|
|
|
print(file=file)
|
|
self.print_help_content(file)
|
|
|
|
if main_doc is not None:
|
|
print(file=file)
|
|
self.print_help_footer(file=file)
|
|
|
|
def print_help_usage_default(self, file: IO) -> None:
|
|
"""print default style of help usage line.
|
|
|
|
:param file: output file stream. if None, use stdout.
|
|
"""
|
|
usage = [sys.argv[0]]
|
|
|
|
if isinstance(self, CliSubCommandMain):
|
|
usage.append(self.command)
|
|
|
|
if len(self.__cli_help_options) > 1 or len(self.__cli_help_options[None]) > 0:
|
|
usage.append('[OPTIONS]')
|
|
|
|
if not isinstance(self, CliReplMain) and len(self.__cli_help_subcommand) > 0:
|
|
usage.append(CliHandleCommand.HELP_SECTION)
|
|
|
|
elif len(self.__cli_help_argument) > 0:
|
|
for attr in self._print_help_list_arguments(self.__cli_help_argument, pain_text=False):
|
|
usage.append(attr[0])
|
|
|
|
print(' '.join(usage), file=file)
|
|
|
|
def print_help_usage(self, *usage: str, file: Optional[IO] = None) -> None:
|
|
"""print help usage line.
|
|
|
|
:param usage: usage element
|
|
:param file: output file stream. if None, use stdout.
|
|
"""
|
|
if isinstance(self, CliSubCommandMain):
|
|
# noinspection PyUnresolvedReferences
|
|
print(sys.argv[0], self.command, *usage, file=file)
|
|
else:
|
|
print(sys.argv[0], *usage, file=file)
|
|
|
|
def print_help_header(self, doc: Optional[str] = None, file: Optional[IO] = None) -> None:
|
|
"""print help doc header. It print the first line of the *doc*
|
|
|
|
:param doc: document. if None, get from __doc__
|
|
:param file: output file stream. if None, use stdout.
|
|
"""
|
|
if doc is None:
|
|
doc: str = self.__doc__
|
|
|
|
if doc is not None:
|
|
for line in takewhile(lambda ln: len(ln) != 0, doc.split('\n')):
|
|
print(line, file=file)
|
|
|
|
else:
|
|
for line in doc.split('\n'):
|
|
print(line, file=file)
|
|
|
|
def print_help_footer(self, doc: Optional[str] = None, file: Optional[IO] = None) -> None:
|
|
"""print help doc footer. It print the left line of the *doc*
|
|
|
|
:param doc: document. if None, get from __doc__
|
|
:param file: output file stream. if None, use stdout.
|
|
"""
|
|
if doc is None:
|
|
doc: str = self.__doc__
|
|
|
|
if doc is None:
|
|
return
|
|
|
|
indent = None
|
|
|
|
for line in dropwhile(lambda ln: len(ln) != 0, doc.split('\n')):
|
|
if indent is None and len(line) == 0:
|
|
continue
|
|
elif indent is None:
|
|
indent = len(line) - len(line.lstrip())
|
|
|
|
line = line[indent:].rstrip()
|
|
|
|
print(line, file=file)
|
|
|
|
else:
|
|
for line in doc.split('\n'):
|
|
print(line, file=file)
|
|
|
|
def print_help_content(self, file: Optional[IO] = None) -> None:
|
|
"""print cli flags/options/arguments/commands content.
|
|
|
|
:param file: output file stream. if None, use stdout.
|
|
"""
|
|
ls = self._print_help_collect_section()
|
|
|
|
align = max(map(lambda ln: len(ln[0]),
|
|
filter(lambda ln: not isinstance(ln, str),
|
|
ls)))
|
|
|
|
align = (align // 4 + 1) * 4
|
|
|
|
for ln in ls:
|
|
if isinstance(ln, str) == 1:
|
|
print(file=file)
|
|
print(ln, file=file)
|
|
|
|
else:
|
|
left, right = ln # type: str, str
|
|
|
|
if right is not None:
|
|
lines = right.strip().split('\n')
|
|
else:
|
|
lines = []
|
|
|
|
print(' ', left, ' ' * (align - len(left) - 1), end='', file=file)
|
|
|
|
if len(lines) == 0:
|
|
# empty line
|
|
print(file=file)
|
|
print(file=file)
|
|
|
|
elif len(lines) == 1:
|
|
# new section
|
|
print(lines[0], file=file)
|
|
print(file=file)
|
|
|
|
else:
|
|
# options content
|
|
print(lines[0], file=file)
|
|
for line in lines[1:]:
|
|
print(' ' * (align - 7), line, file=file)
|
|
print(file=file)
|
|
|
|
def _print_help_collect_section(self) -> List[Tuple[str, Optional[str]]]:
|
|
ls = []
|
|
|
|
opts = self.__cli_help_options[None]
|
|
|
|
if len(opts):
|
|
ls.append('OPTIONS')
|
|
ls.extend(self._print_help_list_options(opts))
|
|
|
|
for section in sorted(filter(lambda it: it is not None, self.__cli_help_options)): # type: str
|
|
opts = self.__cli_help_options[section]
|
|
|
|
if len(opts) == 0:
|
|
continue
|
|
|
|
ls.append(section)
|
|
ls.extend(self._print_help_list_options(opts))
|
|
|
|
if len(self.__cli_help_argument):
|
|
ls.append(CliHandleArgument.HELP_SECTION)
|
|
ls.extend(self._print_help_list_arguments(self.__cli_help_argument, pain_text=True))
|
|
|
|
if len(self.__cli_help_subcommand):
|
|
ls.append(CliHandleCommand.HELP_SECTION)
|
|
ls.extend(self._print_help_list_command(self.__cli_help_subcommand))
|
|
|
|
return ls
|
|
|
|
def _print_help_list_options(self, handle_list: List[CliHandle]) -> Iterable[Tuple[str, Optional[str]]]:
|
|
def _get_options(_h: CliHandle) -> Optional[List[str]]:
|
|
if isinstance(_h, CliHandleFlag):
|
|
_opts = _h.flags
|
|
elif isinstance(_h, CliHandleOption):
|
|
_opts = _h.options
|
|
else:
|
|
return None
|
|
|
|
return list(filter(lambda _opt: _opt is not None,
|
|
map(lambda _opt: self.__cli_options_opts[_h].get(_opt, _opt),
|
|
_opts)))
|
|
|
|
def _key(_h: CliHandle) -> Tuple[int, str]:
|
|
_opts = _get_options(_h)
|
|
|
|
if _opts is not None:
|
|
_opt = _opts[0]
|
|
return 2 if _opt.startswith('--') else 1, _opt
|
|
|
|
return 3, ''
|
|
|
|
for handle in sorted(handle_list, key=_key):
|
|
opts = _get_options(handle)
|
|
|
|
if isinstance(handle, CliHandleFlag):
|
|
f = ', '.join(opts)
|
|
yield f, self._print_help_doc(handle)
|
|
|
|
elif isinstance(handle, CliHandleOption):
|
|
if len(opts) == 1 and opts[0] == '--':
|
|
f = '--'
|
|
else:
|
|
f = ', '.join(opts) + " "
|
|
|
|
v = handle.value_text
|
|
|
|
if handle.optional:
|
|
v = "[" + v + "]"
|
|
|
|
yield f + v, self._print_help_doc(handle)
|
|
|
|
def _print_help_list_arguments(self,
|
|
handle_list: List[CliHandleArgument],
|
|
pain_text=False) -> Iterable[Tuple[str, Optional[str]]]:
|
|
def _key(_h: CliHandleArgument) -> Tuple[int, int]:
|
|
_pos = _h.position
|
|
|
|
return (
|
|
0 if isinstance(_pos, int) else 1,
|
|
_pos if isinstance(_pos, int) else "?*+".index(_pos)
|
|
)
|
|
|
|
for handle in sorted(handle_list, key=_key):
|
|
f = handle.value_text
|
|
|
|
if not pain_text:
|
|
pos = handle.position
|
|
|
|
if pos == '?':
|
|
f = "[" + f + "]"
|
|
|
|
elif pos == '+':
|
|
f = f + '...'
|
|
|
|
elif pos == '*':
|
|
f = "[" + f + "]..."
|
|
|
|
yield f, self._print_help_doc(handle)
|
|
|
|
def _print_help_list_command(self, handle_list: List[CliHandleCommand]) -> Iterable[Tuple[str, Optional[str]]]:
|
|
def _key(_h: CliHandleCommand) -> str:
|
|
return _h.command[0]
|
|
|
|
for handle in sorted(handle_list, key=_key):
|
|
hdr = ', '.join(handle.command)
|
|
|
|
val = handle.usage_line
|
|
if val is not None:
|
|
hdr = hdr + ' ' + ' '.join(val)
|
|
|
|
yield hdr, self._print_help_doc(handle)
|
|
|
|
def _print_help_doc(self, attr) -> Optional[str]:
|
|
return self.__cli_help_document.get(attr, attr.__doc__)
|
|
|
|
|
|
class CliReplMain(CliMain):
|
|
__slots__ = '__repl_prompt', '__complete_cache'
|
|
|
|
def __init__(self, prompt='>>> '):
|
|
super().__init__()
|
|
|
|
self.__repl_prompt = prompt
|
|
|
|
def parsing(self,
|
|
argv: List[str],
|
|
parsing_sub_command=True,
|
|
on_unknown_argument: Callable[[int, str], bool] = None) -> Optional[List[str]]:
|
|
# force disable sub-command parsing
|
|
return super().parsing(argv, False, on_unknown_argument)
|
|
|
|
def run(self):
|
|
self._init_repl()
|
|
|
|
if hasattr(self, '__enter__'):
|
|
with self:
|
|
self.repl()
|
|
|
|
else:
|
|
self.repl()
|
|
|
|
def _init_repl(self):
|
|
cache = []
|
|
|
|
def _repl_complete_func(text: str, state: int) -> Optional[str]:
|
|
if state == 0:
|
|
cache.clear()
|
|
|
|
result = self.repl_complete(text)
|
|
if result is not None:
|
|
cache.extend(result)
|
|
else:
|
|
return None
|
|
|
|
try:
|
|
return cache[state]
|
|
except IndexError:
|
|
return None
|
|
|
|
readline.parse_and_bind("tab: complete")
|
|
readline.set_completer_delims(' ')
|
|
readline.set_completer(_repl_complete_func)
|
|
|
|
def repl(self):
|
|
try:
|
|
while True:
|
|
self.repl_prompt()
|
|
except (EOFError, MainForceReturn):
|
|
pass
|
|
|
|
def repl_prompt(self):
|
|
try:
|
|
r = input(self.__repl_prompt)
|
|
except KeyboardInterrupt:
|
|
print()
|
|
|
|
else:
|
|
if len(r) == 0:
|
|
return
|
|
|
|
argv = list(filter(lambda it: len(it) != 0, r.split(' ')))
|
|
|
|
if len(argv) == 0:
|
|
return
|
|
|
|
command = argv[0]
|
|
|
|
if not self.contain_subcommand(command):
|
|
print('command not found : ' + command)
|
|
|
|
else:
|
|
try:
|
|
self.invoke_subcommand(command, argv[1:])
|
|
|
|
except (SystemExit, MainForceReturn):
|
|
raise
|
|
|
|
except:
|
|
print_exception()
|
|
|
|
def repl_complete(self, text: str) -> Optional[List[str]]:
|
|
if ' ' not in text:
|
|
# complete command
|
|
return list(map(lambda t: t + ' ',
|
|
filter(lambda command: command.startswith(text),
|
|
self._cli_commands.keys())))
|
|
|
|
return None
|
|
|
|
|
|
M = TypeVar('M', bound=CliMain)
|
|
|
|
|
|
class CliSubCommandMain(CliMain, metaclass=abc.ABCMeta):
|
|
__slots__ = ('_cli_command',)
|
|
|
|
def __init__(self, command: str):
|
|
super().__init__()
|
|
|
|
self._cli_command = command
|
|
|
|
@property
|
|
def command(self) -> str:
|
|
return self._cli_command
|
|
|
|
|
|
T = TypeVar('T')
|
|
T1 = TypeVar('T1')
|
|
T2 = TypeVar('T2')
|
|
T3 = TypeVar('T3')
|
|
T4 = TypeVar('T4')
|
|
|
|
|
|
class Arguments:
|
|
"""argument utility
|
|
"""
|
|
|
|
__slots__ = ('__argv', '__pos', '__opt')
|
|
|
|
def __init__(self, argv: Iterable[str], no_options_parsing=False):
|
|
self.__argv = tuple(argv)
|
|
|
|
if no_options_parsing:
|
|
self.__pos = tuple(argv)
|
|
self.__opt = {}
|
|
|
|
else:
|
|
self.__pos = tuple(filter(lambda a: not a.startswith('-'), self.__argv))
|
|
|
|
self.__opt = {}
|
|
for opt in filter(lambda a: a.startswith('-'), self.__argv):
|
|
k, v = part_suffix(opt, '=', missing='')
|
|
|
|
self.__opt[k] = v
|
|
|
|
def __len__(self):
|
|
return len(self.__pos)
|
|
|
|
def __contains__(self, item: str) -> bool:
|
|
return item in self.__opt
|
|
|
|
def __getitem__(self, index: Union[int, str]) -> str:
|
|
if isinstance(index, str):
|
|
return self.__opt[index]
|
|
|
|
elif isinstance(index, int):
|
|
return self.__pos[index]
|
|
else:
|
|
raise KeyError(index)
|
|
|
|
def __iter__(self) -> Iterable[str]:
|
|
return iter(self.__pos)
|
|
|
|
def has_opt(self, flag: str) -> bool:
|
|
if flag in self.__opt:
|
|
return True
|
|
|
|
return False
|
|
|
|
def get_opt(self,
|
|
flag: str,
|
|
nullable=False,
|
|
missing: Optional[T] = None,
|
|
cast: Callable[[str], T] = str) -> Optional[T]:
|
|
try:
|
|
ret = self.__opt[flag]
|
|
|
|
except KeyError as e:
|
|
if not nullable:
|
|
raise KeyError('lost options ' + flag) from e
|
|
else:
|
|
ret = missing
|
|
|
|
if cast == str:
|
|
return ret
|
|
|
|
else:
|
|
try:
|
|
return cast(ret)
|
|
except ValueError as e:
|
|
raise ValueError('options ' + flag + ' not a ' + str(cast) + ' : ' + ret) from e
|
|
|
|
def opts(self) -> Dict[str, str]:
|
|
return dict(self.__opt)
|
|
|
|
def get_pos(self, index: int, parameter: str, cast: Callable[[str], T] = str) -> T:
|
|
try:
|
|
ret = self.__pos[index]
|
|
except IndexError as e:
|
|
raise IndexError('lost argument ' + parameter) from e
|
|
|
|
if cast == str:
|
|
return ret
|
|
|
|
else:
|
|
try:
|
|
return cast(ret)
|
|
except ValueError as e:
|
|
raise ValueError('argument ' + parameter + ' not a ' + str(cast) + ' : ' + ret) from e
|
|
|
|
@overload
|
|
def required(self) -> None:
|
|
pass
|
|
|
|
@overload
|
|
def required(self, parameter: str) -> str:
|
|
pass
|
|
|
|
@overload
|
|
def required(self, p1: str, p2: str) -> Tuple[str, str]:
|
|
pass
|
|
|
|
@overload
|
|
def required(self, p1: str, p2: str, p3: str) -> Tuple[str, str, str]:
|
|
pass
|
|
|
|
@overload
|
|
def required(self, p1: str, p2: str, p3: str, p4: str) -> Tuple[str, str, str, str]:
|
|
pass
|
|
|
|
@overload
|
|
def required(self, parameter: Tuple[str, Type[T]]) -> T:
|
|
pass
|
|
|
|
@overload
|
|
def required(self, p1: Tuple[str, Type[T1]], p2: Tuple[str, Type[T2]]) -> Tuple[T1, T2]:
|
|
pass
|
|
|
|
@overload
|
|
def required(self,
|
|
p1: Tuple[str, Type[T1]],
|
|
p2: Tuple[str, Type[T2]],
|
|
p3: Tuple[str, Type[T3]]) -> Tuple[T1, T2, T3]:
|
|
pass
|
|
|
|
@overload
|
|
def required(self,
|
|
p1: Tuple[str, Type[T1]],
|
|
p2: Tuple[str, Type[T2]],
|
|
p3: Tuple[str, Type[T3]],
|
|
p4: Tuple[str, Type[T4]]) -> Tuple[T1, T2, T3, T4]:
|
|
pass
|
|
|
|
def required(self, *parameter: Union[str, Tuple[str, Type[T]]]):
|
|
argc = len(self.__pos)
|
|
parc = len(parameter)
|
|
|
|
# parameter validate
|
|
for p in parameter:
|
|
if isinstance(p, str):
|
|
pass
|
|
elif isinstance(p, tuple) and len(p) == 2 and isinstance(p[0], str):
|
|
pass
|
|
else:
|
|
raise ValueError('illegal parameter value : ' + str(p))
|
|
|
|
# calculate at_least_parc
|
|
at_least_parc = parc
|
|
for p in reversed(parameter):
|
|
if not isinstance(p, str):
|
|
if _is_null_able_type(p[1]):
|
|
at_least_parc -= 1
|
|
continue
|
|
break
|
|
|
|
# zero parameter
|
|
if parc == 0:
|
|
if argc == 0:
|
|
return
|
|
else:
|
|
raise RuntimeError('unknown argument : ' + str(self.__pos[0]))
|
|
|
|
# one parameter
|
|
elif parc == 1 and argc == 1:
|
|
p = parameter[0]
|
|
|
|
if isinstance(p[0], str):
|
|
return self.__pos[0]
|
|
|
|
else:
|
|
try:
|
|
return _para_cast(p[1], self.__pos[0])
|
|
except (ValueError, TypeError) as e:
|
|
raise ValueError('argument ' + p[0] + ' not a ' + str(p[1]) + ' : ' + self.__pos[0]) from e
|
|
|
|
# too less
|
|
elif at_least_parc > argc:
|
|
raise RuntimeError('lost argument : ' + p[argc])
|
|
|
|
# too many
|
|
elif parc < argc:
|
|
raise RuntimeError('too many argument : ' + str(self.__pos[parc]))
|
|
|
|
else:
|
|
ret = []
|
|
|
|
for i, p in enumerate(parameter):
|
|
if i < argc:
|
|
a = self.__pos[i]
|
|
|
|
if isinstance(p, str):
|
|
ret.append(a)
|
|
else:
|
|
try:
|
|
ret.append(_para_cast(p[1], a))
|
|
except ValueError as e:
|
|
raise ValueError('argument ' + p[0] + ' not a ' + str(p[1]) + ' : ' + a) from e
|
|
|
|
elif at_least_parc <= i:
|
|
ret.append(None)
|
|
|
|
else:
|
|
raise RuntimeError()
|
|
|
|
return tuple(ret)
|
|
|
|
|
|
# noinspection PyPep8Naming
|
|
def CliCommandDelegateMacro(api_class: Type,
|
|
delegate_property: str,
|
|
delegate_handle: Union[None, str, Callable[[str, Any], None]] = None):
|
|
"""**metaclass**. The macro help to generate cli command handle for delegate the class *api_class*.
|
|
|
|
|
|
usage in class declare ::
|
|
|
|
class Main(CliMain, metaclass=CliCommandDelegateMacro(API, 'api', 'api_result')):
|
|
|
|
def __init__(self):
|
|
self._api = None
|
|
|
|
@property
|
|
def api(self) -> API:
|
|
return self._api
|
|
|
|
def __enter__(self):
|
|
self._api = API()
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
self._api = None
|
|
|
|
def api_result(self, command: str, result: Any):
|
|
print(result)
|
|
|
|
|
|
the context manager ``__enter__`` and ``__exit__`` is optional.
|
|
|
|
:param api_class: delegate class
|
|
:param delegate_property: delegate property
|
|
:param delegate_handle: result handle, it could be None, callable, method name
|
|
:return: class
|
|
"""
|
|
|
|
def _meta(define_class_name: str, super_class: Tuple[Type, ...], class_attr: Dict[str, Any]):
|
|
|
|
if delegate_property not in class_attr:
|
|
raise RuntimeError(define_class_name + ' property not defined : ' + delegate_property)
|
|
|
|
if CliReplMain in super_class:
|
|
use_context_manager = False
|
|
|
|
elif '__enter__' in class_attr:
|
|
use_context_manager = True
|
|
|
|
else:
|
|
use_context_manager = False
|
|
|
|
for api_attr_name in dir(api_class):
|
|
|
|
if not api_attr_name.startswith('_'):
|
|
api_func = getattr(api_class, api_attr_name)
|
|
|
|
# should be a method and not overwrote
|
|
if callable(api_func) and '_' + api_attr_name not in class_attr:
|
|
# get callable by calling function to create local variable binding.
|
|
# print(api_attr_name)
|
|
class_attr['_' + api_attr_name] = _gen_impl_func(delegate_property,
|
|
api_func,
|
|
delegate_handle,
|
|
use_context_manager=use_context_manager)
|
|
|
|
return type(define_class_name, super_class, class_attr)
|
|
|
|
return _meta
|
|
|
|
|
|
def _gen_impl_func(delegate_property: str,
|
|
ref_func: Callable[[Any], None],
|
|
delegate_handle: Union[None, str, Callable[[str, Any], None]] = None,
|
|
use_context_manager=False) -> Callable[[str, List[str]], None]:
|
|
# type alias
|
|
# noinspection PyPep8Naming
|
|
P = inspect.Parameter
|
|
# noinspection PyPep8Naming
|
|
B = inspect.BoundArguments
|
|
|
|
func_name: str = ref_func.__name__
|
|
|
|
signature = inspect.signature(ref_func)
|
|
parameter: Dict[str, P] = signature.parameters
|
|
|
|
# build usage_line
|
|
usage_line = []
|
|
|
|
for para_name, para in parameter.items():
|
|
if para_name == 'self':
|
|
continue
|
|
|
|
para_name = para_name.upper()
|
|
para_kind = para.kind
|
|
|
|
if para_kind != P.POSITIONAL_OR_KEYWORD:
|
|
raise RuntimeError('unsupported function %s parameter %s with type : %s' %
|
|
(func_name, para_name, para_kind.name))
|
|
|
|
if para.default != P.empty:
|
|
para_name = '[' + para_name + ']'
|
|
|
|
# TODO consider para.annotation
|
|
|
|
usage_line.append(para_name)
|
|
|
|
# parameter resolve
|
|
|
|
def _impl_para(self, argument: List[str]) -> B:
|
|
bind: B = signature.bind(self, *argument)
|
|
|
|
for p in parameter:
|
|
if p != 'self':
|
|
v = bind.arguments[p]
|
|
a = parameter[p].annotation
|
|
bind.arguments[p] = _para_cast(a, v)
|
|
|
|
else:
|
|
del bind.arguments[p]
|
|
|
|
return bind
|
|
|
|
# result handle
|
|
|
|
def _impl_result(self, result):
|
|
if delegate_handle is None:
|
|
pass
|
|
|
|
elif delegate_handle == print:
|
|
print(func_name, result)
|
|
|
|
elif callable(delegate_handle):
|
|
delegate_handle(func_name, result)
|
|
|
|
elif isinstance(delegate_handle, str):
|
|
getattr(self, delegate_handle)(func_name, result)
|
|
|
|
else:
|
|
raise RuntimeError('unknown result handle : ' + str(delegate_handle))
|
|
|
|
def _impl_invoke(self, argument: List[str]):
|
|
caller = getattr(self, delegate_property)
|
|
func = getattr(caller, func_name)
|
|
bind = _impl_para(caller, argument)
|
|
ret = func(*bind.args, **bind.kwargs)
|
|
_impl_result(self, ret)
|
|
|
|
# function implement
|
|
if use_context_manager:
|
|
|
|
# noinspection PyUnusedLocal
|
|
@wraps(ref_func)
|
|
def _impl_func(self, command: str, argument: List[str]):
|
|
with self:
|
|
_impl_invoke(self, argument)
|
|
|
|
else:
|
|
|
|
# noinspection PyUnusedLocal
|
|
@wraps(ref_func)
|
|
def _impl_func(self, command: str, argument: List[str]):
|
|
_impl_invoke(self, argument)
|
|
|
|
# noinspection SpellCheckingInspection
|
|
_impl_func.__isabstractmethod__ = False
|
|
_impl_func._cli_handle_ = h = CliHandleCommand((func_name,), usage_line)
|
|
h.handle = _impl_func
|
|
|
|
# function attribute
|
|
if ref_func.__doc__ is not None:
|
|
h.help_doc = ref_func.__doc__.split('\n')[0]
|
|
|
|
# noinspection PyTypeChecker
|
|
return _impl_func
|
|
|
|
|
|
def _is_null_able_type(value_cls) -> bool:
|
|
if value_cls is None:
|
|
return True
|
|
|
|
elif hasattr(value_cls, '_name') and hasattr(value_cls, '__args__'):
|
|
# TODO use typing.get_origin and typing.get_args instead for python version 3.8
|
|
col_type = value_cls._name
|
|
ele_type = value_cls.__args__
|
|
|
|
if col_type is None and type(None) in ele_type:
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
def _para_cast(value_cls, value: Optional[str]) -> Any:
|
|
if value_cls in (str, int, float):
|
|
return value_cls(value)
|
|
|
|
elif value_cls is bool:
|
|
return str(value).lower() == 'true'
|
|
|
|
elif hasattr(value_cls, '_name') and hasattr(value_cls, '__args__'):
|
|
# TODO use typing.get_origin and typing.get_args instead for python version 3.8
|
|
col_type = value_cls._name
|
|
ele_type = value_cls.__args__
|
|
|
|
# field: Optional[Any] == Union[Any, None]
|
|
if col_type is None and len(ele_type) == 2 and type(None) in ele_type:
|
|
ele_type = ele_type[1] if ele_type[0] is type(None) else ele_type[0]
|
|
|
|
return _para_cast(ele_type, value)
|
|
|
|
# field: Union[...]
|
|
elif col_type is None:
|
|
for _ele_type in ele_type:
|
|
if _ele_type is not type(None):
|
|
try:
|
|
return _para_cast(_ele_type, value)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
# field: List[Any]
|
|
elif col_type == 'List':
|
|
ele_type = ele_type[0]
|
|
|
|
if ele_type in (str, int, float):
|
|
return [ele_type(ele_type, v) for v in value.split(',')]
|
|
else:
|
|
return [_para_cast(ele_type, v) for v in value.split(',')]
|
|
|
|
# field: Dict[Any, Any]
|
|
elif col_type == 'Dict':
|
|
key_type = ele_type[0]
|
|
ele_type = ele_type[1]
|
|
|
|
if key_type == str:
|
|
return {
|
|
v[0]: v[1]
|
|
for v in map(lambda it: it.split('=', maxsplit=2),
|
|
value.split(','))
|
|
}
|
|
|
|
elif key_type == str and ele_type in (int, float):
|
|
return {
|
|
v[0]: ele_type(v[1])
|
|
for v in map(lambda it: it.split('=', maxsplit=2),
|
|
value.split(','))
|
|
}
|
|
|
|
elif key_type == str:
|
|
return {
|
|
v[0]: _para_cast(ele_type, v[1])
|
|
for v in map(lambda it: it.split('=', maxsplit=2),
|
|
value.split(','))
|
|
}
|
|
|
|
elif key_type in (int, float) and ele_type in (str, int, float):
|
|
return {
|
|
key_type(v[0]): ele_type(v[1])
|
|
for v in map(lambda it: it.split('=', maxsplit=2),
|
|
value.split(','))
|
|
}
|
|
elif key_type in (int, float):
|
|
return {
|
|
key_type(v[0]): _para_cast(ele_type, v[1])
|
|
for v in map(lambda it: it.split('=', maxsplit=2),
|
|
value.split(','))
|
|
}
|
|
|
|
# field: Dict[Any, Any]
|
|
else:
|
|
return {
|
|
_para_cast(key_type, v[0]): _para_cast(ele_type, v[1])
|
|
for v in map(lambda it: it.split('=', maxsplit=2),
|
|
value.split(','))
|
|
}
|
|
|
|
# field: Set[Any]
|
|
elif col_type == 'Set':
|
|
if ele_type in (str, int, float):
|
|
return set([ele_type(ele_type, v) for v in value.split(',')])
|
|
else:
|
|
return set([_para_cast(ele_type, v) for v in value.split(',')])
|
|
|
|
# field: Tuple[Any]
|
|
elif col_type == 'Tuple':
|
|
# field: Tuple[Any, ...]
|
|
if len(ele_type) == 2 and ele_type[1] == Ellipsis:
|
|
_ele_type = ele_type[0]
|
|
|
|
if ele_type in (str, int, float):
|
|
return tuple([ele_type(ele_type, v) for v in value.split(',')])
|
|
else:
|
|
return tuple([_para_cast(ele_type, v) for v in value.split(',')])
|
|
|
|
raise TypeError('unsupported parameter ' + value + ' cast to ' + str(value_cls))
|
|
|
|
|
|
# TODO write selected options to file and load
|
|
|
|
def cli_to_json(target) -> JSON:
|
|
if isinstance(target, CliMain):
|
|
return _cli_to_json_main(target)
|
|
else:
|
|
return _cli_to_json_general(target)
|
|
|
|
|
|
def _cli_to_json_main(target: CliMain) -> JSON_ARRAY:
|
|
ret = []
|
|
|
|
# noinspection PyProtectedMember
|
|
for handle in set(target._cli_flags.values()):
|
|
ret.append({
|
|
'function': handle.handle.__name__,
|
|
'flags': list(handle.flags),
|
|
'force_return': handle.force_return,
|
|
'help_section': handle.help_section,
|
|
'help_doc': handle.help_doc
|
|
})
|
|
|
|
# noinspection PyProtectedMember
|
|
for handle in set(target._cli_options.values()):
|
|
ret.append({
|
|
'function': handle.handle.__name__,
|
|
'options': list(handle.options),
|
|
'value': handle.value_text,
|
|
'optional': handle.optional,
|
|
'help_section': handle.help_section,
|
|
'help_doc': handle.help_doc
|
|
})
|
|
|
|
return ret
|
|
|
|
|
|
def _cli_to_json_general(target) -> JSON_ARRAY:
|
|
ret = []
|
|
|
|
target_type = type(target)
|
|
|
|
for attr_name in dir(target_type):
|
|
attr = getattr(target_type, attr_name, None)
|
|
|
|
# attr should be a method
|
|
if attr is None or not callable(attr):
|
|
continue
|
|
|
|
handle: Optional[CliHandle] = getattr(attr, '_cli_handle_', None)
|
|
|
|
# skip disabled
|
|
if handle is None or isinstance(handle, CliHandleDisable):
|
|
continue
|
|
|
|
j = _cli_handle_to_json(attr_name, handle)
|
|
|
|
if j is not None:
|
|
ret.append(j)
|
|
|
|
return ret
|
|
|
|
|
|
def _cli_handle_to_json(attr_name: str, handle) -> JSON:
|
|
if isinstance(handle, CliHandleFlag):
|
|
return {
|
|
'function': attr_name,
|
|
'flags': list(handle.flags),
|
|
'force_return': handle.force_return,
|
|
'help_section': handle.help_section,
|
|
'help_doc': handle.help_doc
|
|
}
|
|
|
|
elif isinstance(handle, CliHandleOption):
|
|
return {
|
|
'function': attr_name,
|
|
'options': list(handle.options),
|
|
'value': handle.value_text,
|
|
'optional': handle.optional,
|
|
'help_section': handle.help_section,
|
|
'help_doc': handle.help_doc
|
|
}
|
|
|
|
return None
|