pybik 3.0-2 / usr / lib / pybik / pybiklib / settings.py

#-*- coding:utf-8 -*-

#  Copyright © 2009, 2011-2017  B. Clausius <barcc@gmx.de>
#
#  This program is free software: you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation, either version 3 of the License, or
#  (at your option) any later version.
#
#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with this program.  If not, see <http://www.gnu.org/licenses/>.


import sys, os
from ast import literal_eval
from io import StringIO
from collections import OrderedDict
from contextlib import suppress

from . import config


current_settings_version = 3

class KeyStore:
    def __init__(self):
        self.schema = None
        self.filename = None
        self.keystore = {}
        self.callback = lambda unused_key: None
        self.globtree = GlobTree()
        self.extra_lines = []
        
    def clone_key(self, keyfrom, keyto):
        self.schema[keyto] = self.schema[keyfrom]
        if keyfrom in self.keystore and keyto not in self.keystore:
            self.keystore[keyto] = self.keystore[keyfrom]
            
    def get_default(self, key):
        return self.schema[key][0]
    def get_validator(self, key):
        return self.schema[key][1]
        
    def get_range(self, key):
        validator = self.get_validator(key)
        if not isinstance(validator, (tuple, list)):
            raise ValueError('{} is not a range'.format(key))
        return validator
    
    def get_value(self, key):
        try:
            return self.keystore[key]
        except KeyError:
            return self.get_default(key)
            
    def get_nick(self, key):
        value = self.get_value(key)
        validator = self.get_validator(key)
        if not isinstance(validator, list):
            raise ValueError('{} is not an enum'.format(key))
        return validator[value]
        
    def set_value(self, key, value):
        if key not in self.keystore:
            if value != self.get_default(key):
                self.keystore[key] = value
                self.callback(key)
        elif self.keystore[key] != value:
            self.keystore[key] = value
            self.callback(key)
        
    def set_nick(self, key, nick):
        validator = self.get_validator(key)
        if not isinstance(validator, list):
            raise ValueError('{} is not an enum'.format(key))
        value = validator.index(nick)
        return self.set_value(key, value)
        
    def del_value(self, key):
        try:
            value = self.keystore[key]
        except KeyError:
            pass # already the default value
        else:
            del self.keystore[key]
            if value != self.get_default(key):
                self.callback(key)
            else:
                self.callback(None)
        
    def find_match(self, key):
        def match(key, pattern):
            key = key.split('.')
            pattern = pattern.split('.')
            for k, p in zip(key, pattern):
                if k != p != '*':
                    return False
            return True
        for pattern in self.schema:
            if match(key, pattern):
                return pattern
        return None
        
    def read_settings(self, filename):
        from .schema import schema, deprecated
        self.schema = schema
        self.filename = filename
        
        # read settings
        lines = []
        if self.filename:
            dirname = os.path.dirname(self.filename)
            if dirname and not os.path.exists(dirname):
                os.makedirs(dirname)
            try:
                with open(self.filename, 'rt', encoding='utf-8') as settings_file:
                    text = settings_file.read()
                    lines = text.splitlines()  # line breaks are not included
            except FileNotFoundError:
                pass
            except Exception:
                sys.excepthook(*sys.exc_info())
                
        for line in lines:
            try:
                key, strvalue = line.split('=', 1)
            except ValueError:
                # discard invalid lines
                continue
                
            key = key.strip()
            pattern = self.find_match(key)
            if pattern is None:
                # keep unknown keys, they may be settings of a different version
                self.extra_lines.append(line)
                continue
            self.schema[key] = self.schema[pattern]
            
            strvalue = strvalue.strip()
            try:
                value = literal_eval(strvalue)
            except (ValueError, SyntaxError):
                # discard invalid values
                continue
                
            # translate enums and validate values, discard invalid values
            validator = self.get_validator(key)
            if validator is deprecated:
                pass
            elif isinstance(validator, list):
                try:
                    value = validator.index(value)
                except ValueError:
                    continue
            elif isinstance(validator, tuple):
                if not validator[0] <= value <= validator[1]:
                    continue
            elif validator is not None:
                if not validator(value):
                    continue
                    
            self.keystore[key] = value
            
        self.build_globtree()
        
    def build_globtree(self):
        keys = list(self.schema.keys())
        for key in keys:
            subkeys = key.split('.')
            if '*' in subkeys:
                self.globtree.insert_key(subkeys)
                
    def resolve_glob(self, key):
        subkeys = key.split('.')
        clone = self.globtree.find_key(subkeys)
        assert clone is not None, subkeys
        globkey = '.'.join(clone)
        assert globkey != key
        self.clone_key(globkey, key)
        
    def dump(self, file, all=False):    # pylint: disable=W0622
        keydict = self.schema if all else self.keystore
        for key in sorted(keydict.keys()):
            if '*' in key:
                continue
            try:
                # translate enums
                value = self.get_nick(key)
            except ValueError:
                value = self.get_value(key)
            print(key, '=', repr(value), file=file)
        if self.extra_lines:
            #XXX: this line is discarded when loading the settings file
            #     because it does not contain a '='
            print('# comments and keys unknown by version {}:'.format(config.VERSION), file=file)
            for line in self.extra_lines:
                print(line, file=file)
        
    def write_settings(self):
        if not self.filename:
            return
        buf = StringIO()
        self.dump(buf)
        tmpfilename = self.filename + '.tmp'
        with suppress(OSError):
            os.remove(tmpfilename)
        try:
            with open(tmpfilename, 'wt', encoding='utf-8') as settings_file:
                settings_file.write(buf.getvalue())
            os.replace(tmpfilename, self.filename)
        except Exception:
            sys.excepthook(*sys.exc_info())
            
    def get_by_suffix(self, key):
        if key.endswith('_nick'):
            key = key[:-5]
            func = self.get_nick
        elif key.endswith('_range'):
            key = key[:-6]
            func = self.get_range
        elif key.endswith('_default'):
            key = key[:-8]
            func = self.get_default
        else:
            func = self.get_value
        try:
            return func(key)
        except KeyError:
            self.resolve_glob(key)
            return func(key)
            
    def set_by_suffix(self, key, value):
        if key.endswith('_nick'):
            key = key[:-5]
            func = self.set_nick
        else:
            func = self.set_value
        try:
            func(key, value)
        except KeyError:
            self.resolve_glob(key)
            func(key, value)
            
    def del_by_suffix(self, key):
        try:
            self.del_value(key)
        except KeyError:
            self.resolve_glob(key)
            self.del_value(key)
            
        
class GlobTree:
    __slots__ = 'nodes', 'leaves'
    
    def __init__(self):
        self.nodes = OrderedDict()
        self.leaves = []
        
    def insert_key(self, subkeys):
        subkey, *subkeys = subkeys
        if subkeys:
            try:
                node = self.nodes[subkey]
            except KeyError:
                node = GlobTree()
                self.nodes[subkey] = node
            node.insert_key(subkeys)
        elif subkey not in self.leaves:
            self.leaves.append(subkey)
        else:
            assert False
        
    def find_key(self, subkeys):
        subkey, *subkeys = subkeys
        if subkeys:
            for subkey in subkey, '*':
                try:
                    node = self.nodes[subkey]
                except KeyError:
                    continue
                clone = node.find_key(subkeys)
                if clone is not None:
                    return [subkey] + clone
            return None
        elif subkey in self.leaves:
            return [subkey]
        elif '*' in self.leaves:
            return ['*']
        else:
            return None
            
        
class Settings:
    __slots__ = ()
    keystore = KeyStore()
    dump = keystore.dump
    
    @classmethod
    def reset(self):
        self.keystore.keystore.clear()
        
    def load(self, filename):
        self.keystore.read_settings(filename)
        
        version = self['version']
        if version != current_settings_version:
            self['version'] = current_settings_version
        return version
            
    @classmethod
    def connnect(self, callback):
        self.keystore.callback = callback
        
    @classmethod
    def disconnect(self):
        self.keystore.callback = lambda unused_key: None
        
    @classmethod
    def flush(self):
        self.keystore.write_settings()
        
    @staticmethod
    def norm_key(key):
        if type(key) is int:
            return str(key)
        elif type(key) is tuple:
            def cvt_dot(keys):
                for key in keys:
                    if type(key) is int:
                        yield str(key)
                    elif type(key) is tuple:
                        yield '_'.join(str(v) for v in key)
                    else:
                        yield from key.split('.')
            return '.'.join(cvt_dot(key))
        return key
        
    def __getitem__(self, key):
        key = self.norm_key(key)
        return self.keystore.get_by_suffix(key)
        
    def __setitem__(self, key, value):
        key = self.norm_key(key)
        self.keystore.set_by_suffix(key, value)
        
    def __delitem__(self, key):
        key = self.norm_key(key)
        self.keystore.del_by_suffix(key)
        
    

settings = Settings()