# -*- coding: utf-8 -*-
""" Configuration machinery for DeadBeat """
from __future__ import absolute_import, division, print_function, unicode_literals
import optparse
import typedcols
import json
import os.path as pth
import sys
from collections import OrderedDict, deque
from shlex import shlex
from .movement import basestring
__all__ = ["cast_boolean", "cast_path_list", "cfg_item", "cfg_section", "cfg_root", "cfg_base_config", "gather"]
[docs]def cast_boolean(v):
""" Translate user provided text into boolean. """
try:
return bool(int(v))
except ValueError: pass
if v.lower() in ("true", "yes"):
return True
if v.lower() in ("false", "no"):
return False
raise ValueError("True or False expected")
[docs]def cast_path_list(l):
""" Translate user provided comma separated strings into list of paths. """
if isinstance(l, basestring):
lex = shlex(l, posix=True)
lex.whitespace=','
lex.whitespace_split=True
strings = list(lex)
else:
strings = l
return strings
class _ConfigBase(typedcols.TypedDict):
""" Configuration base class. """
def __getattr__(self, name):
return self[name]
def deep_update(self, other):
for key, value in other.items():
tdef_type = self.typedef.get(key, {}).get("type")
if key in self and isinstance(tdef_type, _ConfigBase):
self[key].deep_update(value)
else:
self[key] = value
def _get_config_class(name, typedef):
return type(str(name), (_ConfigBase,), dict(allow_unknown=False, typedef=OrderedDict(typedef)))
def _read_cfg(path):
with open(path, "r") as f:
stripcomments = "\n".join((l for l in f if not l.lstrip().startswith(("#", "//"))))
conf = json.loads(stripcomments)
return conf
def _make_option_parser(name, description, config):
""" Create optparse configuration out of config objects hierarchy. """
parser = optparse.OptionParser(prog=name, description=description, add_help_option=False)
group = optparse.OptionGroup(parser, "Basic")
parser.add_option_group(group)
group.add_option("-h", "--help", action="help", help="Show this help message and exit")
queue = deque((k, v, group) for k, v in config.typedef.items())
while queue:
key, tdef, group = queue.popleft()
ttype = tdef["type"]
if issubclass(ttype, _ConfigBase):
group = optparse.OptionGroup(parser, tdef.get("description", key))
parser.add_option_group(group)
queue.extend(
(key + "." + subkey, subtdef, group) for subkey, subtdef in ttype.typedef.items())
else:
metavar = tdef.get("default")
if not metavar:
metavar = getattr(ttype, "__name__").upper()
if metavar.startswith("<"):
metavar = None
if not metavar:
metavar = key.split(".")[-1].upper()
group.add_option(
"--" + key.replace("_", "-"), dest=key, help=tdef.get("description"),
metavar=metavar)
return parser
def _dots_to_nested_dicts(src):
dest = dict()
for dotted, value in src.items():
root = dest
splitted = dotted.split(".")
for label in splitted[:-1]:
root.setdefault(label, dict())
root = root[label]
root[splitted[-1]] = value
return dest
[docs]def cfg_item(name, type=typedcols.Any, description=None, default=None):
""" Helper to define configuration items.
:param name: Config item name
:param type: Translation callable, which accepts string and returns normalized object
:param description: Help text
:param default: Default value if undefined in configuration
:returns: Tuple of item name and type definition, acceptable by cfg_section/cfg_root (or TypedDict)
"""
tdef = dict(name=name, type=type)
if description is not None:
tdef["description"] = description
if default is not None:
tdef["default"] = default
return name, tdef
[docs]def cfg_section(name, items, description=None):
""" Helper to define configuration subsections.
:param name: Config section name
:param items: Iterable of cfg_item or cfg_section outputs
:param description: Section name
:returns: Tuple of section name and type definition, acceptable by cfg_section/cfg_root (or TypedDict)
"""
subconfig = _get_config_class(str("SubConfig_%s") % name, items)
return cfg_item(name, subconfig, description, default={})
[docs]def cfg_root(args):
""" Helper to create base configuration section.
:param args: Iterable of cfg_item or cfg_section outputs
:returns: Config type definition class, acceptable by gather (or TypedDict)
"""
return _get_config_class(str("BaseConfig"), args)
#: Base configuration insert
cfg_base_config = [
cfg_item("filenames", cast_path_list, "Configuration files to use"),
]
[docs]def gather(typedef, name=None, description=None, files=None):
""" Read configuration files and command line options and return merged tree.
Note that option names should be valid python identifiers to be able to
acces config by attribute access (cfg.filenames). Also, underscores
will be replaced by minus sign for command line options.
:param typedef: TypedDict config definition (as from cfg_root)
:param name: Name of the program/script
:param description: Description or the program/script
:param files: Iterable of config file paths to try, read and merge
(also option 'filenames' on command line gets consulted
:returns: Configuration in the form of nested namespaces
(result["config"]["filenames"] is valid, result.config.filenames also)
"""
config = typedef()
optparse = _make_option_parser(name, description, config)
cmdline, args = optparse.parse_args()
cmdline = _dots_to_nested_dicts(
dict((key, value) for key, value in vars(cmdline).items() if value is not None))
cmdline_files = cast_path_list(cmdline.get("config", {}).get("filenames", []))
files = files or [
pth.splitext(__file__)[0] + ".cfg",
pth.join("/etc", pth.basename(__file__) + ".cfg"),
pth.splitext(sys.argv[0])[0] + ".cfg",
pth.join("/etc", pth.basename(sys.argv[0]) + ".cfg")]
files.extend(cmdline_files)
for path in files:
try:
nextconf = _read_cfg(path)
except IOError:
continue
config.deep_update(nextconf)
config.deep_update(cmdline)
config.checkRequired(recursive=True)
config.name = name
config.description = description
config.args = args
return config