#-*- coding:utf-8 -*- # Copyright © 2009, 2011-2017 B. Clausius # # 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 . 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()