main_tool: relocate init funcs to tool.__init__
This commit is contained in:
parent
ec15a0b8e8
commit
1fd521956f
GPG key ID:
3F8B1861E32B7DA2
5 changed files with 155 additions and 130 deletions
|
|
@ -23,8 +23,8 @@ mmgen-tool: Perform various MMGen- and cryptocoin-related operations.
|
|||
|
||||
import sys, os, importlib
|
||||
from .cfg import gc, Config
|
||||
from .util import msg, Msg, die, capfirst, suf, async_run, isAsync
|
||||
from .tool.common import fstr
|
||||
from .util import Msg, die, capfirst, async_run, isAsync
|
||||
from .tool import process_args
|
||||
|
||||
opts_data = {
|
||||
'filter_codes': ['-'],
|
||||
|
|
@ -186,128 +186,6 @@ mods = {
|
|||
),
|
||||
}
|
||||
|
||||
def get_cmds():
|
||||
return [cmd for mod, cmds in mods.items() if mod != 'help' for cmd in cmds]
|
||||
|
||||
def create_call_sig(cmd, cls, *, as_string=False):
|
||||
|
||||
m = getattr(cls, cmd)
|
||||
|
||||
if 'varargs_call_sig' in m.__code__.co_varnames: # hack
|
||||
flag = 'VAR_ARGS'
|
||||
va = m.__defaults__[0]
|
||||
args, dfls, ann = va['args'], va['dfls'], va['annots']
|
||||
else:
|
||||
flag = None
|
||||
c = m.__code__
|
||||
args = c.co_varnames[1:c.co_argcount + c.co_posonlyargcount + c.co_kwonlyargcount]
|
||||
dfls = (
|
||||
(m.__defaults__ or ()) +
|
||||
tuple(m.__kwdefaults__[k] for k in args if k in (m.__kwdefaults__ or ())))
|
||||
ann = m.__annotations__
|
||||
|
||||
nargs = len(args) - len(dfls)
|
||||
dfl_types = tuple(
|
||||
ann[a] if a in ann and isinstance(ann[a], type) else type(dfls[i])
|
||||
for i, a in enumerate(args[nargs:]))
|
||||
|
||||
if as_string:
|
||||
get_type_from_ann = lambda x: 'str or STDIN' if ann[x] == fstr else ann[x].__name__
|
||||
return ' '.join(
|
||||
[f'{a} [{get_type_from_ann(a)}]' for a in args[:nargs]] +
|
||||
[f'{a} [{dfl_types[n].__name__}={dfls[n]!r}]' for n, a in enumerate(args[nargs:])])
|
||||
else:
|
||||
get_type_from_ann = lambda x: 'str' if ann[x] == fstr else ann[x].__name__
|
||||
return (
|
||||
[(a, get_type_from_ann(a)) for a in args[:nargs]], # c_args
|
||||
{a: dfls[n] for n, a in enumerate(args[nargs:])}, # c_kwargs
|
||||
{a: dfl_types[n] for n, a in enumerate(args[nargs:])}, # c_kwargs_types
|
||||
('STDIN_OK' if nargs and ann[args[0]] == fstr else flag), # flag
|
||||
ann) # ann
|
||||
|
||||
def process_args(cmd, cmd_args, cls):
|
||||
c_args, c_kwargs, c_kwargs_types, flag, _ = create_call_sig(cmd, cls)
|
||||
have_stdin_input = False
|
||||
|
||||
def usage_die(s):
|
||||
msg(s)
|
||||
from .tool.help import usage
|
||||
usage(cmd)
|
||||
|
||||
if flag != 'VAR_ARGS':
|
||||
if len(cmd_args) < len(c_args):
|
||||
usage_die(f'Command requires exactly {len(c_args)} non-keyword argument{suf(c_args)}')
|
||||
|
||||
u_args = cmd_args[:len(c_args)]
|
||||
|
||||
# If we're reading from a pipe, replace '-' with output of previous command
|
||||
if flag == 'STDIN_OK' and u_args and u_args[0] == '-':
|
||||
if sys.stdin.isatty():
|
||||
die('BadFilename', "Standard input is a TTY. Can't use '-' as a filename")
|
||||
else:
|
||||
from .util2 import parse_bytespec
|
||||
max_dlen_spec = '10kB' # limit input to 10KB for now
|
||||
max_dlen = parse_bytespec(max_dlen_spec)
|
||||
u_args[0] = os.read(0, max_dlen)
|
||||
have_stdin_input = True
|
||||
if len(u_args[0]) >= max_dlen:
|
||||
die(2, f'Maximum data input for this command is {max_dlen_spec}')
|
||||
if not u_args[0]:
|
||||
die(2, f'{cmd}: ERROR: no output from previous command in pipe')
|
||||
|
||||
u_nkwargs = len(cmd_args) - len(c_args)
|
||||
u_kwargs = {}
|
||||
if flag == 'VAR_ARGS':
|
||||
cmd_args = ['dummy_arg'] + cmd_args
|
||||
t = [a.split('=', 1) for a in cmd_args if '=' in a]
|
||||
tk = [a[0] for a in t]
|
||||
tk_bad = [a for a in tk if a not in c_kwargs]
|
||||
if set(tk_bad) != set(tk[:len(tk_bad)]): # permit non-kw args to contain '='
|
||||
die(1, f'{tk_bad[-1]!r}: illegal keyword argument')
|
||||
u_kwargs = dict(t[len(tk_bad):])
|
||||
u_args = cmd_args[:-len(u_kwargs) or None]
|
||||
elif u_nkwargs > 0:
|
||||
u_kwargs = dict([a.split('=', 1) for a in cmd_args[len(c_args):] if '=' in a])
|
||||
if len(u_kwargs) != u_nkwargs:
|
||||
usage_die(f'Command requires exactly {len(c_args)} non-keyword argument{suf(c_args)}')
|
||||
if len(u_kwargs) > len(c_kwargs):
|
||||
usage_die(f'Command accepts no more than {len(c_kwargs)} keyword argument{suf(c_kwargs)}')
|
||||
|
||||
for k in u_kwargs:
|
||||
if k not in c_kwargs:
|
||||
usage_die(f'{k!r}: invalid keyword argument')
|
||||
|
||||
def conv_type(arg, arg_name, arg_type):
|
||||
if arg_type == 'bytes' and not isinstance(arg, bytes):
|
||||
die(1, "'Binary input data must be supplied via STDIN")
|
||||
|
||||
if have_stdin_input and arg_type == 'str' and isinstance(arg, bytes):
|
||||
NL = '\r\n' if gc.platform == 'win32' else '\n'
|
||||
arg = arg.decode()
|
||||
if arg[-len(NL):] == NL: # rstrip one newline
|
||||
arg = arg[:-len(NL)]
|
||||
|
||||
if arg_type == 'bool':
|
||||
if arg.lower() in ('true', 'yes', '1', 'on'):
|
||||
arg = True
|
||||
elif arg.lower() in ('false', 'no', '0', 'off'):
|
||||
arg = False
|
||||
else:
|
||||
usage_die(f'{arg!r}: invalid boolean value for keyword argument')
|
||||
|
||||
try:
|
||||
return __builtins__[arg_type](arg)
|
||||
except:
|
||||
die(1, f'{arg!r}: Invalid argument for argument {arg_name} ({arg_type!r} required)')
|
||||
|
||||
if flag == 'VAR_ARGS':
|
||||
args = [conv_type(u_args[i], c_args[0][0], c_args[0][1]) for i in range(len(u_args))]
|
||||
else:
|
||||
args = [conv_type(u_args[i], c_args[i][0], c_args[i][1]) for i in range(len(c_args))]
|
||||
kwargs = {k: conv_type(v, k, c_kwargs_types[k].__name__) for k, v in u_kwargs.items()}
|
||||
|
||||
return (args, kwargs)
|
||||
|
||||
def process_result(ret, *, pager=False, print_result=False):
|
||||
"""
|
||||
Convert result to something suitable for output to screen and return it.
|
||||
|
|
|
|||
144
mmgen/tool/__init__.py
Normal file → Executable file
144
mmgen/tool/__init__.py
Normal file → Executable file
|
|
@ -1,4 +1,148 @@
|
|||
#!/usr/bin/env python3
|
||||
#
|
||||
# MMGen Wallet, a terminal-based cryptocurrency wallet
|
||||
# Copyright (C)2013-2026 The MMGen Project <mmgen@tuta.io>
|
||||
# Licensed under the GNU General Public License, Version 3:
|
||||
# https://www.gnu.org/licenses
|
||||
# Public project repositories:
|
||||
# https://github.com/mmgen/mmgen-wallet
|
||||
# https://gitlab.com/mmgen/mmgen-wallet
|
||||
|
||||
"""
|
||||
tool.__init__: initialization functions for the ‘mmgen-tool’ utility and others
|
||||
"""
|
||||
|
||||
import sys, os
|
||||
|
||||
from ..cfg import gc
|
||||
from ..util import die
|
||||
|
||||
from .common import fstr
|
||||
|
||||
# provide this for backwards compatibility:
|
||||
def tool_api(*args, **kwargs):
|
||||
from .api import tool_api
|
||||
return tool_api(*args, **kwargs)
|
||||
|
||||
def get_cmds(mods):
|
||||
return [cmd for mod, cmds in mods.items() if mod != 'help' for cmd in cmds]
|
||||
|
||||
def create_call_sig(cmd, cls, *, as_string=False):
|
||||
|
||||
m = getattr(cls, cmd)
|
||||
|
||||
if 'varargs_call_sig' in m.__code__.co_varnames: # hack
|
||||
flag = 'VAR_ARGS'
|
||||
va = m.__defaults__[0]
|
||||
args, dfls, ann = va['args'], va['dfls'], va['annots']
|
||||
else:
|
||||
flag = None
|
||||
c = m.__code__
|
||||
args = c.co_varnames[1:c.co_argcount + c.co_posonlyargcount + c.co_kwonlyargcount]
|
||||
dfls = (
|
||||
(m.__defaults__ or ()) +
|
||||
tuple(m.__kwdefaults__[k] for k in args if k in (m.__kwdefaults__ or ())))
|
||||
ann = m.__annotations__
|
||||
|
||||
nargs = len(args) - len(dfls)
|
||||
dfl_types = tuple(
|
||||
ann[a] if a in ann and isinstance(ann[a], type) else type(dfls[i])
|
||||
for i, a in enumerate(args[nargs:]))
|
||||
|
||||
if as_string:
|
||||
get_type_from_ann = lambda x: 'str or STDIN' if ann[x] == fstr else ann[x].__name__
|
||||
return ' '.join(
|
||||
[f'{a} [{get_type_from_ann(a)}]' for a in args[:nargs]] +
|
||||
[f'{a} [{dfl_types[n].__name__}={dfls[n]!r}]' for n, a in enumerate(args[nargs:])])
|
||||
else:
|
||||
get_type_from_ann = lambda x: 'str' if ann[x] == fstr else ann[x].__name__
|
||||
return (
|
||||
[(a, get_type_from_ann(a)) for a in args[:nargs]], # c_args
|
||||
{a: dfls[n] for n, a in enumerate(args[nargs:])}, # c_kwargs
|
||||
{a: dfl_types[n] for n, a in enumerate(args[nargs:])}, # c_kwargs_types
|
||||
('STDIN_OK' if nargs and ann[args[0]] == fstr else flag), # flag
|
||||
ann) # ann
|
||||
|
||||
def process_args(cmd, cmd_args, cls):
|
||||
c_args, c_kwargs, c_kwargs_types, flag, _ = create_call_sig(cmd, cls)
|
||||
have_stdin_input = False
|
||||
from ..util import msg, suf
|
||||
|
||||
def usage_die(s):
|
||||
msg(s)
|
||||
from .help import usage
|
||||
usage(cmd)
|
||||
|
||||
if flag != 'VAR_ARGS':
|
||||
if len(cmd_args) < len(c_args):
|
||||
usage_die(f'Command requires exactly {len(c_args)} non-keyword argument{suf(c_args)}')
|
||||
|
||||
u_args = cmd_args[:len(c_args)]
|
||||
|
||||
# If we're reading from a pipe, replace '-' with output of previous command
|
||||
if flag == 'STDIN_OK' and u_args and u_args[0] == '-':
|
||||
if sys.stdin.isatty():
|
||||
die('BadFilename', "Standard input is a TTY. Can't use '-' as a filename")
|
||||
else:
|
||||
from ..util2 import parse_bytespec
|
||||
max_dlen_spec = '10kB' # limit input to 10KB for now
|
||||
max_dlen = parse_bytespec(max_dlen_spec)
|
||||
u_args[0] = os.read(0, max_dlen)
|
||||
have_stdin_input = True
|
||||
if len(u_args[0]) >= max_dlen:
|
||||
die(2, f'Maximum data input for this command is {max_dlen_spec}')
|
||||
if not u_args[0]:
|
||||
die(2, f'{cmd}: ERROR: no output from previous command in pipe')
|
||||
|
||||
u_nkwargs = len(cmd_args) - len(c_args)
|
||||
u_kwargs = {}
|
||||
if flag == 'VAR_ARGS':
|
||||
cmd_args = ['dummy_arg'] + cmd_args
|
||||
t = [a.split('=', 1) for a in cmd_args if '=' in a]
|
||||
tk = [a[0] for a in t]
|
||||
tk_bad = [a for a in tk if a not in c_kwargs]
|
||||
if set(tk_bad) != set(tk[:len(tk_bad)]): # permit non-kw args to contain '='
|
||||
die(1, f'{tk_bad[-1]!r}: illegal keyword argument')
|
||||
u_kwargs = dict(t[len(tk_bad):])
|
||||
u_args = cmd_args[:-len(u_kwargs) or None]
|
||||
elif u_nkwargs > 0:
|
||||
u_kwargs = dict([a.split('=', 1) for a in cmd_args[len(c_args):] if '=' in a])
|
||||
if len(u_kwargs) != u_nkwargs:
|
||||
usage_die(f'Command requires exactly {len(c_args)} non-keyword argument{suf(c_args)}')
|
||||
if len(u_kwargs) > len(c_kwargs):
|
||||
usage_die(f'Command accepts no more than {len(c_kwargs)} keyword argument{suf(c_kwargs)}')
|
||||
|
||||
for k in u_kwargs:
|
||||
if k not in c_kwargs:
|
||||
usage_die(f'{k!r}: invalid keyword argument')
|
||||
|
||||
def conv_type(arg, arg_name, arg_type):
|
||||
if arg_type == 'bytes' and not isinstance(arg, bytes):
|
||||
die(1, "'Binary input data must be supplied via STDIN")
|
||||
|
||||
if have_stdin_input and arg_type == 'str' and isinstance(arg, bytes):
|
||||
NL = '\r\n' if gc.platform == 'win32' else '\n'
|
||||
arg = arg.decode()
|
||||
if arg[-len(NL):] == NL: # rstrip one newline
|
||||
arg = arg[:-len(NL)]
|
||||
|
||||
if arg_type == 'bool':
|
||||
if arg.lower() in ('true', 'yes', '1', 'on'):
|
||||
arg = True
|
||||
elif arg.lower() in ('false', 'no', '0', 'off'):
|
||||
arg = False
|
||||
else:
|
||||
usage_die(f'{arg!r}: invalid boolean value for keyword argument')
|
||||
|
||||
try:
|
||||
return __builtins__[arg_type](arg)
|
||||
except:
|
||||
die(1, f'{arg!r}: Invalid argument for argument {arg_name} ({arg_type!r} required)')
|
||||
|
||||
if flag == 'VAR_ARGS':
|
||||
args = [conv_type(u_args[i], c_args[0][0], c_args[0][1]) for i in range(len(u_args))]
|
||||
else:
|
||||
args = [conv_type(u_args[i], c_args[i][0], c_args[i][1]) for i in range(len(c_args))]
|
||||
kwargs = {k: conv_type(v, k, c_kwargs_types[k].__name__) for k, v in u_kwargs.items()}
|
||||
|
||||
return (args, kwargs)
|
||||
|
|
|
|||
|
|
@ -23,6 +23,7 @@ tool.help: Help screen routines for the 'mmgen-tool' utility
|
|||
from .. import main_tool
|
||||
|
||||
from .common import tool_cmd_base, fstr
|
||||
from . import create_call_sig
|
||||
|
||||
usage_hdr = """
|
||||
GENERAL USAGE INFORMATION FOR MMGEN-TOOL COMMANDS
|
||||
|
|
@ -124,7 +125,7 @@ def gen_tool_usage():
|
|||
for cmdname in cmdlist:
|
||||
yield ' {a:{w}} {b}'.format(
|
||||
a = cmdname,
|
||||
b = main_tool.create_call_sig(cmdname, cls, as_string=True),
|
||||
b = create_call_sig(cmdname, cls, as_string=True),
|
||||
w = max_w)
|
||||
yield ''
|
||||
|
||||
|
|
@ -138,7 +139,7 @@ def gen_tool_cmd_usage(mod, cmdname):
|
|||
|
||||
cls = main_tool.get_mod_cls(mod)
|
||||
docstr = getattr(cls, cmdname).__doc__.strip()
|
||||
args, kwargs, kwargs_types, _, ann = main_tool.create_call_sig(cmdname, cls)
|
||||
args, kwargs, kwargs_types, _, ann = create_call_sig(cmdname, cls)
|
||||
ARGS = 'ARG' if len(args) == 1 else 'ARGS' if args else ''
|
||||
KWARGS = 'KEYWORD ARG' if len(kwargs) == 1 else 'KEYWORD ARGS' if kwargs else ''
|
||||
|
||||
|
|
|
|||
|
|
@ -202,9 +202,10 @@ if cfg.testing_status:
|
|||
Msg(' '+'\n '.join(sorted(tested_in[l])))
|
||||
|
||||
ignore = ()
|
||||
from mmgen.main_tool import get_cmds
|
||||
from mmgen.tool import get_cmds
|
||||
from mmgen.main_tool import mods
|
||||
uc = sorted(
|
||||
set(get_cmds()) -
|
||||
set(get_cmds(mods)) -
|
||||
set(ignore) -
|
||||
set(tested_in['tooltest.py']) -
|
||||
set(tested_in['tooltest2.py']) -
|
||||
|
|
|
|||
|
|
@ -36,6 +36,7 @@ except ImportError:
|
|||
from test.include.common import set_globals, end_msg, init_coverage
|
||||
|
||||
from mmgen import main_tool
|
||||
from mmgen.tool import process_args
|
||||
from mmgen.cfg import Config, gc
|
||||
from mmgen.color import green, blue, purple, cyan, gray
|
||||
from mmgen.util import msg, msg_r, Msg, die, isAsync
|
||||
|
|
@ -112,7 +113,7 @@ def call_method(cls, method, cmd_name, args, mmtype, stdin_input):
|
|||
a = purple('Running'),
|
||||
b = ' '.join([cmd_name]+[repr(e) for e in args]),
|
||||
c = ' '+mmtype if mmtype else ''))
|
||||
aargs, kwargs = main_tool.process_args(cmd_name, args, cls)
|
||||
aargs, kwargs = process_args(cmd_name, args, cls)
|
||||
oq_save = bool(cfg.quiet)
|
||||
if not cfg.verbose:
|
||||
cfg._set_quiet(True)
|
||||
|
|
@ -144,7 +145,7 @@ def tool_api(cls, cmd_name, args, opts):
|
|||
for o in opts:
|
||||
if o.startswith('--type='):
|
||||
tool.addrtype = o.split('=')[1]
|
||||
pargs, kwargs = main_tool.process_args(cmd_name, args, cls)
|
||||
pargs, kwargs = process_args(cmd_name, args, cls)
|
||||
return getattr(tool, cmd_name)(*pargs, **kwargs)
|
||||
|
||||
def check_output(out, chk):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue