545 lines
No EOL
20 KiB
Python
545 lines
No EOL
20 KiB
Python
import optparse
|
|
import sys
|
|
from typing import Any, Dict, List, Optional, Tuple, Union
|
|
from .pairtype import extendabletype
|
|
|
|
SUPPRESS_USAGE = optparse.SUPPRESS_USAGE
|
|
|
|
class AmbigousOptionError(Exception):
|
|
pass
|
|
|
|
class NoMatchingOptionFound(AttributeError):
|
|
pass
|
|
|
|
class ConfigError(Exception):
|
|
pass
|
|
|
|
class ConflictConfigError(ConfigError):
|
|
pass
|
|
|
|
class Config:
|
|
_cfgimpl_frozen = False
|
|
|
|
def __init__(self, descr, parent=None, **overrides):
|
|
self._cfgimpl_descr = descr
|
|
self._cfgimpl_value_owners: Dict[str, str] = {}
|
|
self._cfgimpl_parent = parent
|
|
self._cfgimpl_values: Dict[str, Any] = {}
|
|
self._cfgimpl_warnings: List[str] = []
|
|
self._cfgimpl_build(overrides)
|
|
|
|
def _cfgimpl_build(self, overrides):
|
|
for child in self._cfgimpl_descr._children:
|
|
if isinstance(child, Option):
|
|
self._cfgimpl_values[child._name] = child.getdefault()
|
|
self._cfgimpl_value_owners[child._name] = 'default'
|
|
elif isinstance(child, OptionDescription):
|
|
self._cfgimpl_values[child._name] = Config(child, parent=self)
|
|
|
|
def copy(self, as_default=False, parent=None):
|
|
result = Config.__new__(self.__class__)
|
|
result._cfgimpl_descr = self._cfgimpl_descr
|
|
result._cfgimpl_value_owners = owners = {}
|
|
result._cfgimpl_parent = parent
|
|
result._cfgimpl_values = v = {}
|
|
for child in self._cfgimpl_descr._children:
|
|
if isinstance(child, Option):
|
|
v[child._name] = self._cfgimpl_values[child._name]
|
|
owners[child._name] = 'default' if as_default else self._cfgimpl_value_owners[child._name]
|
|
elif isinstance(child, OptionDescription):
|
|
v[child._name] = self._cfgimpl_values[child._name].copy(as_default, parent=result)
|
|
return result
|
|
|
|
def __setattr__(self, name, value):
|
|
if self._cfgimpl_frozen and getattr(self, name) != value:
|
|
raise TypeError("trying to change a frozen option object")
|
|
if name.startswith('_cfgimpl_'):
|
|
super().__setattr__(name, value)
|
|
else:
|
|
self.setoption(name, value, 'user')
|
|
|
|
def __getattr__(self, name):
|
|
if '.' in name:
|
|
homeconfig, name = self._cfgimpl_get_home_by_path(name)
|
|
return getattr(homeconfig, name)
|
|
if name.startswith('_cfgimpl_'):
|
|
raise AttributeError(f"{self.__class__} object has no attribute {name}")
|
|
if name not in self._cfgimpl_values:
|
|
raise AttributeError(f"{self.__class__} object has no attribute {name}")
|
|
return self._cfgimpl_values[name]
|
|
|
|
def __dir__(self):
|
|
from_type = dir(type(self))
|
|
from_dict = list(self.__dict__)
|
|
extras = list(self._cfgimpl_values)
|
|
return sorted(set(extras + from_type + from_dict))
|
|
|
|
def __delattr__(self, name):
|
|
if name.startswith('_cfgimpl_'):
|
|
super().__delattr__(name)
|
|
else:
|
|
self._cfgimpl_value_owners[name] = 'default'
|
|
opt = getattr(self._cfgimpl_descr, name)
|
|
if isinstance(opt, OptionDescription):
|
|
raise AttributeError("can't delete option subgroup")
|
|
self._cfgimpl_values[name] = getattr(opt, 'default', None)
|
|
|
|
def setoption(self, name, value, who):
|
|
if name not in self._cfgimpl_values:
|
|
raise AttributeError(f'unknown option {name}')
|
|
child = getattr(self._cfgimpl_descr, name)
|
|
oldowner = self._cfgimpl_value_owners[child._name]
|
|
|
|
# Allow requirements to override user-set values if necessary
|
|
if who == 'required' and oldowner == 'user':
|
|
# Requirement takes precedence over user setting
|
|
pass
|
|
elif oldowner not in ("default", "suggested") and who not in ("default", "suggested"):
|
|
if getattr(self, name) != value:
|
|
raise ConflictConfigError(f'cannot override value to {value} for option {name}')
|
|
|
|
child.setoption(self, value, who)
|
|
self._cfgimpl_value_owners[name] = who
|
|
|
|
def suggest(self, **kwargs):
|
|
for name, value in kwargs.items():
|
|
self.suggestoption(name, value)
|
|
|
|
def suggestoption(self, name, value):
|
|
# Only apply suggestion if the option hasn't been explicitly set by user
|
|
current_owner = self._cfgimpl_value_owners.get(name, "default")
|
|
if current_owner not in ("user", "required"):
|
|
try:
|
|
self.setoption(name, value, "suggested")
|
|
except ConflictConfigError:
|
|
pass
|
|
|
|
def set(self, **kwargs):
|
|
all_paths = [p.split(".") for p in self.getpaths()]
|
|
for key, value in kwargs.items():
|
|
key_p = key.split('.')
|
|
candidates = [p for p in all_paths if p[-len(key_p):] == key_p]
|
|
if len(candidates) == 1:
|
|
name = '.'.join(candidates[0])
|
|
homeconfig, name = self._cfgimpl_get_home_by_path(name)
|
|
homeconfig.setoption(name, value, "user")
|
|
elif len(candidates) > 1:
|
|
raise AmbigousOptionError(f'more than one option that ends with {key}')
|
|
else:
|
|
raise NoMatchingOptionFound(f'there is no option that matches {key}')
|
|
|
|
def _cfgimpl_get_home_by_path(self, path: str) -> Tuple['Config', str]:
|
|
path = path.split('.')
|
|
config = self
|
|
for step in path[:-1]:
|
|
config = getattr(config, step)
|
|
return config, path[-1]
|
|
|
|
def _cfgimpl_get_toplevel(self) -> 'Config':
|
|
config = self
|
|
while config._cfgimpl_parent is not None:
|
|
config = config._cfgimpl_parent
|
|
return config
|
|
|
|
def add_warning(self, warning: str):
|
|
self._cfgimpl_get_toplevel()._cfgimpl_warnings.append(warning)
|
|
|
|
def get_warnings(self) -> List[str]:
|
|
return self._cfgimpl_get_toplevel()._cfgimpl_warnings
|
|
|
|
def _freeze_(self):
|
|
self._cfgimpl_frozen = True
|
|
return True
|
|
|
|
def getkey(self):
|
|
return self._cfgimpl_descr.getkey(self)
|
|
|
|
def __hash__(self):
|
|
return hash(self.getkey())
|
|
|
|
def __eq__(self, other):
|
|
return self.getkey() == other.getkey()
|
|
|
|
def __ne__(self, other):
|
|
return not self == other
|
|
|
|
def __iter__(self):
|
|
for child in self._cfgimpl_descr._children:
|
|
if isinstance(child, Option):
|
|
yield child._name, getattr(self, child._name)
|
|
|
|
def __str__(self, indent=""):
|
|
lines = []
|
|
children = sorted((child._name, child) for child in self._cfgimpl_descr._children)
|
|
for name, child in children:
|
|
if self._cfgimpl_value_owners.get(name, 'default') == 'default':
|
|
continue
|
|
value = getattr(self, name)
|
|
if isinstance(value, Config):
|
|
substr = value.__str__(indent + " ")
|
|
else:
|
|
substr = f"{indent} {name} = {value}"
|
|
if substr:
|
|
lines.append(substr)
|
|
if indent and not lines:
|
|
return ''
|
|
lines.insert(0, f"{indent}[{self._cfgimpl_descr._name}]")
|
|
return '\n'.join(lines)
|
|
|
|
def getpaths(self, include_groups=False) -> List[str]:
|
|
return self._cfgimpl_descr.getpaths(include_groups=include_groups)
|
|
|
|
DEFAULT_OPTION_NAME = object()
|
|
|
|
class Option(metaclass=extendabletype):
|
|
def __init__(self, name: str, doc: str, cmdline=DEFAULT_OPTION_NAME):
|
|
self._name = name
|
|
self.doc = doc
|
|
self.cmdline = cmdline
|
|
|
|
def validate(self, value) -> bool:
|
|
raise NotImplementedError('abstract base class')
|
|
|
|
def getdefault(self):
|
|
return self.default
|
|
|
|
def setoption(self, config: Config, value, who: str):
|
|
name = self._name
|
|
if who == "default" and value is None:
|
|
return
|
|
if not self.validate(value):
|
|
raise ConfigError(f'invalid value {value} for option {name}')
|
|
config._cfgimpl_values[name] = value
|
|
|
|
def getkey(self, value):
|
|
return value
|
|
|
|
def convert_from_cmdline(self, value):
|
|
return value
|
|
|
|
def add_optparse_option(self, argnames, parser, config):
|
|
callback = ConfigUpdate(config, self)
|
|
parser.add_option(help=f"{self.doc} %default",
|
|
action='callback', type=self.opt_type,
|
|
callback=callback, metavar=self._name.upper(),
|
|
*argnames)
|
|
|
|
class ChoiceOption(Option):
|
|
opt_type = 'string'
|
|
|
|
def __init__(self, name: str, doc: str, values: list, default=None,
|
|
requires: Optional[dict] = None, suggests: Optional[dict] = None,
|
|
cmdline=DEFAULT_OPTION_NAME):
|
|
super().__init__(name, doc, cmdline)
|
|
self.values = values
|
|
self.default = default
|
|
self._requires = requires or {}
|
|
self._suggests = suggests or {}
|
|
|
|
def setoption(self, config: Config, value, who: str):
|
|
for path, reqvalue in self._requires.get(value, []):
|
|
toplevel = config._cfgimpl_get_toplevel()
|
|
homeconfig, name = toplevel._cfgimpl_get_home_by_path(path)
|
|
who2 = 'default' if who == 'default' else 'required'
|
|
homeconfig.setoption(name, reqvalue, who2)
|
|
for path, reqvalue in self._suggests.get(value, []):
|
|
toplevel = config._cfgimpl_get_toplevel()
|
|
homeconfig, name = toplevel._cfgimpl_get_home_by_path(path)
|
|
homeconfig.suggestoption(name, reqvalue)
|
|
super().setoption(config, value, who)
|
|
|
|
def validate(self, value) -> bool:
|
|
return value is None or value in self.values
|
|
|
|
def convert_from_cmdline(self, value):
|
|
return value.strip()
|
|
|
|
def _getnegation(optname: str) -> str:
|
|
if optname.startswith("without"):
|
|
return "with" + optname[len("without"):]
|
|
if optname.startswith("with"):
|
|
return "without" + optname[len("with"):]
|
|
return "no-" + optname
|
|
|
|
class BoolOption(Option):
|
|
def __init__(self, name: str, doc: str, default=None, requires=None,
|
|
suggests=None, validator=None, cmdline=DEFAULT_OPTION_NAME,
|
|
negation=True):
|
|
super().__init__(name, doc, cmdline=cmdline)
|
|
self._requires = requires
|
|
self._suggests = suggests
|
|
self.default = default
|
|
self.negation = negation
|
|
self._validator = validator
|
|
|
|
def validate(self, value) -> bool:
|
|
return isinstance(value, bool)
|
|
|
|
def setoption(self, config: Config, value, who: str):
|
|
if value and self._validator is not None:
|
|
self._validator(config._cfgimpl_get_toplevel())
|
|
if value and self._requires is not None:
|
|
for path, reqvalue in self._requires:
|
|
toplevel = config._cfgimpl_get_toplevel()
|
|
homeconfig, name = toplevel._cfgimpl_get_home_by_path(path)
|
|
who2 = 'default' if who == 'default' else 'required'
|
|
homeconfig.setoption(name, reqvalue, who2)
|
|
if value and self._suggests is not None:
|
|
for path, reqvalue in self._suggests:
|
|
toplevel = config._cfgimpl_get_toplevel()
|
|
homeconfig, name = toplevel._cfgimpl_get_home_by_path(path)
|
|
homeconfig.suggestoption(name, reqvalue)
|
|
super().setoption(config, value, who)
|
|
|
|
def add_optparse_option(self, argnames, parser, config):
|
|
callback = BoolConfigUpdate(config, self, True)
|
|
parser.add_option(help=f"{self.doc} %default",
|
|
action='callback', callback=callback, *argnames)
|
|
if not self.negation:
|
|
return
|
|
no_argnames = ["--" + _getnegation(argname.lstrip("-"))
|
|
for argname in argnames if argname.startswith("--")]
|
|
if not no_argnames:
|
|
no_argnames = ["--" + _getnegation(argname.lstrip("-")) for argname in argnames]
|
|
callback = BoolConfigUpdate(config, self, False)
|
|
parser.add_option(help=f"unset option set by {argnames[0]}",
|
|
action='callback', callback=callback, *no_argnames)
|
|
|
|
class IntOption(Option):
|
|
opt_type = 'int'
|
|
|
|
def __init__(self, name: str, doc: str, default=None, cmdline=DEFAULT_OPTION_NAME):
|
|
super().__init__(name, doc, cmdline)
|
|
self.default = default
|
|
|
|
def validate(self, value) -> bool:
|
|
try:
|
|
int(value)
|
|
return True
|
|
except (TypeError, ValueError):
|
|
return False
|
|
|
|
def setoption(self, config: Config, value, who: str):
|
|
try:
|
|
super().setoption(config, int(value), who)
|
|
except (TypeError, ValueError) as e:
|
|
raise ConfigError(f"Invalid integer value: {e}")
|
|
|
|
class FloatOption(Option):
|
|
opt_type = 'float'
|
|
|
|
def __init__(self, name: str, doc: str, default=None, cmdline=DEFAULT_OPTION_NAME):
|
|
super().__init__(name, doc, cmdline)
|
|
self.default = default
|
|
|
|
def validate(self, value) -> bool:
|
|
try:
|
|
float(value)
|
|
return True
|
|
except (TypeError, ValueError):
|
|
return False
|
|
|
|
def setoption(self, config: Config, value, who: str):
|
|
try:
|
|
super().setoption(config, float(value), who)
|
|
except (TypeError, ValueError) as e:
|
|
raise ConfigError(f"Invalid float value: {e}")
|
|
|
|
class StrOption(Option):
|
|
opt_type = 'string'
|
|
|
|
def __init__(self, name: str, doc: str, default=None, cmdline=DEFAULT_OPTION_NAME):
|
|
super().__init__(name, doc, cmdline)
|
|
self.default = default
|
|
|
|
def validate(self, value) -> bool:
|
|
return isinstance(value, str)
|
|
|
|
class ArbitraryOption(Option):
|
|
def __init__(self, name: str, doc: str, default=None, defaultfactory=None):
|
|
super().__init__(name, doc, cmdline=None)
|
|
self.default = default
|
|
self.defaultfactory = defaultfactory
|
|
if defaultfactory is not None:
|
|
assert default is None
|
|
|
|
def validate(self, value) -> bool:
|
|
return True
|
|
|
|
def add_optparse_option(self, *args, **kwargs):
|
|
return
|
|
|
|
def getdefault(self):
|
|
return self.defaultfactory() if self.defaultfactory else self.default
|
|
|
|
class OptionDescription(metaclass=extendabletype):
|
|
cmdline = None
|
|
|
|
def __init__(self, name: str, doc: str, children: list):
|
|
self._name = name
|
|
self.doc = doc
|
|
self._children = children
|
|
self._build()
|
|
|
|
def _build(self):
|
|
for child in self._children:
|
|
setattr(self, child._name, child)
|
|
|
|
def getkey(self, config: Config):
|
|
return tuple(child.getkey(getattr(config, child._name)) for child in self._children)
|
|
|
|
def add_optparse_option(self, argnames, parser, config):
|
|
return
|
|
|
|
def getpaths(self, include_groups=False, currpath=None) -> List[str]:
|
|
if currpath is None:
|
|
currpath = []
|
|
paths = []
|
|
for option in self._children:
|
|
attr = option._name
|
|
if attr.startswith('_cfgimpl'):
|
|
continue
|
|
value = getattr(self, attr)
|
|
if isinstance(value, OptionDescription):
|
|
if include_groups:
|
|
paths.append('.'.join(currpath + [attr]))
|
|
currpath.append(attr)
|
|
paths += value.getpaths(include_groups=include_groups, currpath=currpath)
|
|
currpath.pop()
|
|
else:
|
|
paths.append('.'.join(currpath + [attr]))
|
|
return paths
|
|
|
|
class OptHelpFormatter(optparse.TitledHelpFormatter):
|
|
extra_useage = None
|
|
|
|
def expand_default(self, option):
|
|
assert self.parser
|
|
dfls = self.parser.defaults
|
|
defl = ""
|
|
choices = None
|
|
|
|
if option.action == 'callback' and isinstance(option.callback, ConfigUpdate):
|
|
callback = option.callback
|
|
defl = callback.help_default()
|
|
if isinstance(callback.option, ChoiceOption):
|
|
choices = callback.option.values
|
|
else:
|
|
val = dfls.get(option.dest)
|
|
if val is None:
|
|
pass
|
|
elif isinstance(val, bool):
|
|
if val is True and option.action == "store_true":
|
|
defl = "default"
|
|
else:
|
|
defl = f"default: {val}"
|
|
|
|
if option.type == 'choice':
|
|
choices = option.choices
|
|
|
|
if choices is not None:
|
|
choices_str = f"{option.metavar}={'|'.join(choices)}"
|
|
else:
|
|
choices_str = ""
|
|
|
|
if '%default' in option.help:
|
|
if choices_str and defl:
|
|
sep = ", "
|
|
else:
|
|
sep = ""
|
|
defl = f"[{choices_str}{sep}{defl}]" if choices_str or defl else ""
|
|
return option.help.replace("%default", defl)
|
|
elif choices_str:
|
|
return f"{option.help} [{choices_str}]"
|
|
return option.help
|
|
|
|
def format_usage(self, usage):
|
|
result = super().format_usage(usage)
|
|
if self.extra_useage is not None:
|
|
return f"{result}\n{self.extra_useage}\n\n"
|
|
return result
|
|
|
|
class ConfigUpdate:
|
|
def __init__(self, config: Config, option: Option):
|
|
self.config = config
|
|
self.option = option
|
|
|
|
def convert_from_cmdline(self, value):
|
|
return self.option.convert_from_cmdline(value)
|
|
|
|
def __call__(self, option, opt_str, value, parser, *args, **kwargs):
|
|
try:
|
|
value = self.convert_from_cmdline(value)
|
|
self.config.setoption(self.option._name, value, who='cmdline')
|
|
except ConfigError as e:
|
|
for warning in self.config.get_warnings():
|
|
print(warning, file=sys.stderr)
|
|
raise optparse.OptionValueError(str(e))
|
|
|
|
def help_default(self):
|
|
default = getattr(self.config, self.option._name)
|
|
owner = self.config._cfgimpl_value_owners.get(self.option._name, 'default')
|
|
if default is None:
|
|
return '' if owner == 'default' else '???'
|
|
return f"{owner}: {default}"
|
|
|
|
class BoolConfigUpdate(ConfigUpdate):
|
|
def __init__(self, config: Config, option: Option, which_value: bool):
|
|
super().__init__(config, option)
|
|
self.which_value = which_value
|
|
|
|
def convert_from_cmdline(self, value):
|
|
return self.which_value
|
|
|
|
def help_default(self):
|
|
default = getattr(self.config, self.option._name)
|
|
owner = self.config._cfgimpl_value_owners.get(self.option._name, 'default')
|
|
return owner if default == self.which_value else ""
|
|
|
|
def to_optparse(config: Config, useoptions=None, parser=None,
|
|
parserargs=None, parserkwargs=None, extra_useage=None):
|
|
grps = {}
|
|
def get_group(name: str, doc: str):
|
|
steps = name.split('.')
|
|
if len(steps) < 2:
|
|
return parser
|
|
grpname = steps[-2]
|
|
if grpname not in grps:
|
|
grps[grpname] = parser.add_option_group(doc)
|
|
return grps[grpname]
|
|
|
|
if parser is None:
|
|
parserargs = parserargs or []
|
|
parserkwargs = parserkwargs or {}
|
|
formatter = OptHelpFormatter()
|
|
formatter.extra_useage = extra_useage
|
|
parser = optparse.OptionParser(formatter=formatter, *parserargs, **parserkwargs)
|
|
|
|
if useoptions is None:
|
|
useoptions = config.getpaths(include_groups=True)
|
|
|
|
seen = set()
|
|
for path in useoptions:
|
|
if path.endswith(".*"):
|
|
base_path = path[:-2]
|
|
homeconf, name = config._cfgimpl_get_home_by_path(base_path)
|
|
subconf = getattr(homeconf, name)
|
|
useoptions.extend(f"{base_path}.{child}" for child in subconf.getpaths())
|
|
else:
|
|
if path in seen:
|
|
continue
|
|
seen.add(path)
|
|
homeconf, name = config._cfgimpl_get_home_by_path(path)
|
|
option = getattr(homeconf._cfgimpl_descr, name)
|
|
if option.cmdline is DEFAULT_OPTION_NAME:
|
|
chunks = (f"--{path.replace('.', '-')}",)
|
|
elif option.cmdline is None:
|
|
continue
|
|
else:
|
|
chunks = option.cmdline.split(' ')
|
|
grp = get_group(path, homeconf._cfgimpl_descr.doc)
|
|
option.add_optparse_option(chunks, grp, homeconf)
|
|
return parser
|
|
|
|
def make_dict(config: Config) -> Dict[str, Any]:
|
|
return {path: getattr(config, path) for path in config.getpaths()} |