"""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): print('main', argv) if argv is None: print('Main None', sys.argv[1:]) argv = sys.argv[1:] try: argv = self.parsing(argv) print('1', argv) if argv is None: self.run() else: command, *argv = argv print('2', command, 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] print('arg', i, arg) 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 print('self._cli_commands', self._cli_commands) 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, '=') print('_handle_options_', arg, value) 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 print('cli_argument', arg, pos, a) 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) print('handle', handle) 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