path: root/include/py/homekit/config/config.py

import yaml
import logging
import os
import cerberus
import cerberus.errors
import re

from abc import ABC
from typing import Optional, Any, MutableMapping, Union
from argparse import ArgumentParser
from enum import Enum, auto
from os.path import join, isdir, isfile
from ..util import Addr
from pprint import pprint


class MyValidator(cerberus.Validator):
    def _normalize_coerce_addr(self, value):
        return Addr.fromstring(value)


MyValidator.types_mapping['addr'] = cerberus.TypeDefinition('Addr', (Addr,), ())


CONFIG_DIRECTORIES = (
    join(os.environ['HOME'], '.config', 'homekit'),
    '/etc/homekit'
)


class RootSchemaType(Enum):
    DEFAULT = auto()
    DICT = auto()
    LIST = auto()


class BaseConfigUnit(ABC):
    _data: MutableMapping[str, Any]
    _logger: logging.Logger

    def __init__(self):
        self._data = {}
        self._logger = logging.getLogger(self.__class__.__name__)

    def __iter__(self):
        return iter(self._data)

    def __getitem__(self, key):
        return self._data[key]

    def __setitem__(self, key, value):
        raise NotImplementedError('overwriting config values is prohibited')

    def __contains__(self, key):
        return key in self._data

    def load_from(self, path: str):
        with open(path, 'r') as fd:
            self._data = yaml.safe_load(fd)
            if self._data is None:
                raise TypeError(f'config file {path} is empty')

    def get(self,
            key: Optional[str] = None,
            default=None):
        if key is None:
            return self._data

        cur = self._data
        pts = key.split('.')
        for i in range(len(pts)):
            k = pts[i]
            if i < len(pts)-1:
                if k not in cur:
                    raise KeyError(f'key {k} not found')
            else:
                return cur[k] if k in cur else default
            cur = self._data[k]

        raise KeyError(f'option {key} not found')

    def values(self):
        return self._data.values()

    def keys(self):
        return self._data.keys()

    def items(self):
        return self._data.items()


class ConfigUnit(BaseConfigUnit):
    NAME = 'dumb'

    _instance = None

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        cls._instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super(ConfigUnit, cls).__new__(cls, *args, **kwargs)
        return cls._instance

    def __init__(self, name=None, load=True):
        super().__init__()

        self._data = {}
        self._logger = logging.getLogger(self.__class__.__name__)

        if self.NAME != 'dumb' and load:
            self.load_from(self.get_config_path())
            self.validate()

        elif name is not None:
            self.NAME = name

    @classmethod
    def get_config_path(cls, name=None) -> str:
        if name is None:
            name = cls.NAME
            if name is None:
                raise ValueError('get_config_path: name is none')

        for dirname in CONFIG_DIRECTORIES:
            if isdir(dirname):
                filename = join(dirname, f'{name}.yaml')
                if isfile(filename):
                    return filename

        raise IOError(f'\'{name}.yaml\' not found')

    @classmethod
    def schema(cls) -> Optional[dict]:
        return None

    @classmethod
    def _addr_schema(cls, required=False, mac=False, only_ip=False, **kwargs):
        def validate_mac_address(field, value, error):
            if not re.match("[0-9a-fA-F]{2}([-:])[0-9a-fA-F]{2}(\\1[0-9a-fA-F]{2}){4}$", value):
                error(field, "Invalid MAC address format")

        if mac:
            l_kwargs = {
                'type': 'string',
                'check_with': validate_mac_address
            }
        else:
            l_kwargs = {
                'type': 'addr',
                'coerce': Addr.fromstring if not only_ip else Addr.fromipstring,
            }

        return {
            'required': required,
            **l_kwargs,
            **kwargs
        }

    def validate(self):
        schema = self.schema()
        if not schema:
            self._logger.warning('validate: no schema')
            return

        if isinstance(self, AppConfigUnit):
            schema['logging'] = {
                'type': 'dict',
                'schema': {
                    'verbose': {'type': 'boolean'}
                }
            }

        rst = RootSchemaType.DEFAULT
        try:
            if schema['type'] == 'dict':
                rst = RootSchemaType.DICT
            elif schema['type'] == 'list':
                rst = RootSchemaType.LIST
            elif schema['roottype'] == 'dict':
                del schema['roottype']
                rst = RootSchemaType.DICT
        except KeyError:
            pass

        v = MyValidator()
        need_document = False

        if rst == RootSchemaType.DICT:
            normalized = v.validated({'document': self._data},
                                     {'document': {
                                         'type': 'dict',
                                         'keysrules': {'type': 'string'},
                                         'valuesrules': schema
                                     }})
            need_document = True
        elif rst == RootSchemaType.LIST:
            v = MyValidator()
            normalized = v.validated({'document': self._data}, {'document': schema})
            need_document = True
        else:
            normalized = v.validated(self._data, schema)

        if not normalized:
            raise cerberus.DocumentError(f'validation failed: {v.errors}')

        if need_document:
            normalized = normalized['document']

        self._data = normalized

        try:
            self.custom_validator(self._data)
        except Exception as e:
            raise cerberus.DocumentError(f'{self.__class__.__name__}: {str(e)}')

    @staticmethod
    def custom_validator(data):
        pass

    def get_addr(self, key: str):
        return Addr.fromstring(self.get(key))


class AppConfigUnit(ConfigUnit):
    _logging_verbose: bool
    _logging_fmt: Optional[str]
    _logging_file: Optional[str]

    def __init__(self, *args, **kwargs):
        super().__init__(load=False, *args, **kwargs)
        self._logging_verbose = False
        self._logging_fmt = None
        self._logging_file = None

    def logging_set_fmt(self, fmt: str) -> None:
        self._logging_fmt = fmt

    def logging_get_fmt(self) -> Optional[str]:
        try:
            return self['logging']['default_fmt']
        except (KeyError, TypeError):
            return self._logging_fmt

    def logging_set_file(self, file: str) -> None:
        self._logging_file = file

    def logging_get_file(self) -> Optional[str]:
        try:
            return self['logging']['file']
        except (KeyError, TypeError):
            return self._logging_file

    def logging_set_verbose(self):
        self._logging_verbose = True

    def logging_is_verbose(self) -> bool:
        try:
            return bool(self['logging']['verbose'])
        except (KeyError, TypeError):
            return self._logging_verbose


class TranslationUnit(BaseConfigUnit):
    pass


class Translation:
    LANGUAGES = ('en', 'ru')
    DEFAULT_LANGUAGE = 'ru'

    _langs: dict[str, TranslationUnit]

    def __init__(self, name: str):
        super().__init__()
        self._langs = {}
        for lang in self.LANGUAGES:
            for dirname in CONFIG_DIRECTORIES:
                if isdir(dirname):
                    filename = join(dirname, f'i18n-{lang}', f'{name}.yaml')
                    if lang in self._langs:
                        raise RuntimeError(f'{name}: translation unit for lang \'{lang}\' already loaded')
                    self._langs[lang] = TranslationUnit()
                    self._langs[lang].load_from(filename)
        diff = set()
        for data in self._langs.values():
            diff ^= data.get().keys()
        if len(diff) > 0:
            raise RuntimeError(f'{name}: translation units have difference in keys: ' + ', '.join(diff))

    def get(self, lang: str) -> TranslationUnit:
        return self._langs[lang]


class Config:
    app_name: Optional[str]
    app_config: AppConfigUnit

    def __init__(self):
        self.app_name = None
        self.app_config = AppConfigUnit()

    def load_app(self,
                 name: Optional[Union[str, AppConfigUnit, bool]] = None,
                 use_cli=True,
                 parser: ArgumentParser = None,
                 no_config=False):
        global app_config

        if not no_config \
                and not isinstance(name, str) \
                and not isinstance(name, bool) \
                and issubclass(name, AppConfigUnit) or name == AppConfigUnit:
            self.app_name = name.NAME
            self.app_config = name()
            app_config = self.app_config
        else:
            self.app_name = name if isinstance(name, str) else None

        if self.app_name is None and not use_cli:
            raise RuntimeError('either config name must be none or use_cli must be True')

        no_config = name is False or no_config
        path = None

        if use_cli:
            if parser is None:
                parser = ArgumentParser()
            if not no_config:
                parser.add_argument('-c', '--config', type=str, required=name is None,
                                    help='Path to the config in TOML or YAML format')
            parser.add_argument('-V', '--verbose', action='store_true')
            parser.add_argument('--log-file', type=str)
            parser.add_argument('--log-default-fmt', action='store_true')
            args = parser.parse_args()

            if not no_config and args.config:
                path = args.config

            if args.verbose:
                self.app_config.logging_set_verbose()
            if args.log_file:
                self.app_config.logging_set_file(args.log_file)
            if args.log_default_fmt:
                self.app_config.logging_set_fmt(args.log_default_fmt)

        if not isinstance(name, ConfigUnit):
            if not no_config and path is None:
                path = ConfigUnit.get_config_path(name=self.app_name)

            if not no_config:
                self.app_config.load_from(path)
                self.app_config.validate()

        setup_logging(self.app_config.logging_is_verbose(),
                      self.app_config.logging_get_file(),
                      self.app_config.logging_get_fmt())

        if use_cli:
            return args


config = Config()


def is_development_mode() -> bool:
    if 'HK_MODE' in os.environ and os.environ['HK_MODE'] == 'dev':
        return True

    return ('logging' in config.app_config) and ('verbose' in config.app_config['logging']) and (config.app_config['logging']['verbose'] is True)


def setup_logging(verbose=False, log_file=None, default_fmt=None):
    logging_level = logging.INFO
    if is_development_mode() or verbose:
        logging_level = logging.DEBUG
        _add_logging_level('TRACE', logging.DEBUG-5)

    log_config = {'level': logging_level}
    if not default_fmt:
        log_config['format'] = '%(asctime)s - %(name)s - %(levelname)s - %(message)s'

    if log_file is not None:
        log_config['filename'] = log_file
        log_config['encoding'] = 'utf-8'

    logging.basicConfig(**log_config)


# https://stackoverflow.com/questions/2183233/how-to-add-a-custom-loglevel-to-pythons-logging-facility/35804945#35804945
def _add_logging_level(levelName, levelNum, methodName=None):
    """
    Comprehensively adds a new logging level to the `logging` module and the
    currently configured logging class.

    `levelName` becomes an attribute of the `logging` module with the value
    `levelNum`. `methodName` becomes a convenience method for both `logging`
    itself and the class returned by `logging.getLoggerClass()` (usually just
    `logging.Logger`). If `methodName` is not specified, `levelName.lower()` is
    used.

    To avoid accidental clobberings of existing attributes, this method will
    raise an `AttributeError` if the level name is already an attribute of the
    `logging` module or if the method name is already present

    Example
    -------
    >>> addLoggingLevel('TRACE', logging.DEBUG - 5)
    >>> logging.getLogger(__name__).setLevel("TRACE")
    >>> logging.getLogger(__name__).trace('that worked')
    >>> logging.trace('so did this')
    >>> logging.TRACE
    5

    """
    if not methodName:
        methodName = levelName.lower()

    if hasattr(logging, levelName):
        raise AttributeError('{} already defined in logging module'.format(levelName))
    if hasattr(logging, methodName):
        raise AttributeError('{} already defined in logging module'.format(methodName))
    if hasattr(logging.getLoggerClass(), methodName):
        raise AttributeError('{} already defined in logger class'.format(methodName))

    # This method was inspired by the answers to Stack Overflow post
    # http://stackoverflow.com/q/2183233/2988730, especially
    # http://stackoverflow.com/a/13638084/2988730
    def logForLevel(self, message, *args, **kwargs):
        if self.isEnabledFor(levelNum):
            self._log(levelNum, message, args, **kwargs)
    def logToRoot(message, *args, **kwargs):
        logging.log(levelNum, message, *args, **kwargs)

    logging.addLevelName(levelNum, levelName)
    setattr(logging, levelName, levelNum)
    setattr(logging.getLoggerClass(), methodName, logForLevel)
    setattr(logging, methodName, logToRoot)