pybik 3.0-2 / usr / lib / pybik / pybiklib / pluginlib.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 os
import sys
import importlib
from collections import namedtuple, OrderedDict

from .debug import debug, DEBUG_MSG, DEBUG_ALG, DEBUG_LIVEPLUGINS
from . import config
from . import model

FILE_FORMAT = 'Pybik Plugins Collection - Version 2.0'
FILE_SIG = 'Format: ' + FILE_FORMAT


class PluginFileError (Exception): pass
class PluginSolverAbort (Exception): pass
class PluginModelCompatError (Exception): pass


class PluginFileLoader:
    def __init__(self, filename):
        #                       field name:     (required, singleline, convert-func)
        parse_pass = lambda value, lineno, comment: value
        self.header_scheme = {  'Format':       (True,     True,       self.parse_version),
                                'Copyright':    (False,    False,      parse_pass),
                                'License':      (False,    False,      parse_pass),
                                'Model':        (False,    False,      self.parse_model),
                                'Ref-Blocks':   (False,    True,       self.parse_refblocks),
                            }
        self.body_scheme =  {   'Path':         (True,     True,       self.parse_path),
                                'Depends':      (False,    False,      parse_pass),
                                'Model':        (True,     False,      self.parse_model),
                                'Ref-Blocks':   (False,    True,       self.parse_refblocks),
                                'Solution':     (False,    False,      self.parse_solution),
                                'Moves':        (False,    False,      self.parse_moves),
                                'Module':       (False,    True,       self.parse_module),
                            }
        self.conflicts = [('Solution', 'Moves', 'Module'), ('Depends', 'Moves', 'Module')]
                            
        self.paras = []
        self.header = None
        self.body = []
        self.load(filename)
        self.parse()
        
    def load(self, filename):
        debug('loading plugin-file', repr(filename))
        with open(filename, 'rt', encoding='utf-8') as fp:
            lines = fp.readlines()
        
        # parse file
        para = {}
        key = None
        comment = None
        for lineno, line in enumerate(lines):
            line = line.rstrip()
            if not line:
                # end para
                if not self.paras and not para:
                    raise PluginFileError('File must not start with empty line.')
                self.paras.append(para)
                para = {}
                comment = None
            elif line[0] == '#':
                # comment
                if not self.paras and not para:
                    raise PluginFileError('File must not start with comment.')
                comment = line
            elif line[0] in ' \t':
                if not para:
                    raise PluginFileError('Paragraph must not start with indented line.')
                line = line.strip()
                if line[0] != '#':
                    # multiline
                    para[key][0].append(line)
                comment = None
            else:
                if not self.paras and not para and line.rstrip() != FILE_SIG:
                    raise PluginFileError('File must start with:', FILE_SIG)
                key, value = line.split(':')
                value = value.strip()
                para[key] = [value] if value else [], lineno, comment
                comment = None
        if para:
            self.paras.append(para)
        
    @staticmethod
    def parse_version(value, unused_lineno, unused_comment):
        if value != FILE_FORMAT:
            raise PluginFileError('wrong file version:', value)
        return value
        
    @staticmethod
    def parse_model(modelstrings, unused_lineno, unused_comment):
        model_infos = []
        for modelstr in modelstrings or []:
            try:
                model_info = model.Model.from_string(modelstr)
            except ValueError as e:
                debug('Error in model %s:' % modelstr, e)
            else:
                model_infos.append(model_info)
        return model_infos
        
    @staticmethod
    def parse_refblocks(value, unused_lineno, unused_comment):
        return value and [[v, v] for v in value.upper().split()]
        
    @staticmethod
    def parse_path(value, lineno, comment):
        if not value:
            raise PluginFileError('empty path')
        sep = value[0]
        if sep == '@':
            return value, None, lineno, comment
        def _translate(value):
            transl = True
            for v in value.strip(sep).split(sep):
                if not v:
                    continue
                if transl is True:
                    if v == '_':    transl = True
                    elif v == 'P_': transl = 'sing'
                    elif v == '!_': transl = False
                    else:           yield v, _(v)
                elif transl is False:
                    yield v, v
                    transl = True
                elif transl == 'sing':
                    vs = v
                    transl = 'plural'
                elif transl == 'plural':
                    vp = v
                    transl = 'n'
                elif transl == 'n':
                    try:
                        vi = int(v)
                    except ValueError:
                        raise PluginFileError('Third value after P_ must be int.')
                    yield (vs if vi==1 else vp).format(v), ngettext(vs, vp, vi).format(v)
        return value, tuple(_translate(value)), lineno, comment
        
    @staticmethod
    def parse_solution(value, unused_lineno, unused_comment):
        def split_solution(value):
            try:
                conditions, moves = value.split(',')
            except ValueError:
                raise PluginFileError('expected exactly one comma in: ' + value)
            conditions = conditions.strip().upper().split()
            conditions = [c.split('=') for c in conditions]
            for c in conditions:
                if len(c) != 2:
                    raise PluginFileError('expected exactly one "=" in: ' + '='.join(c))
            return value, conditions, moves.strip()
        return value and [split_solution(v) for v in value if v]
        
    @staticmethod
    def parse_moves(value, unused_lineno, unused_comment):
        if value is None:
            return None
        moves = ' '.join(value)
        def func(game):
            game.move_sequence.reset()
            game.add_code(moves)
            game.set_plugin_mode('append')
        return func
        
    @staticmethod
    def parse_module(value, unused_lineno, unused_comment):
        if value is None:
            return None
        value = value.split(',')
        if len(value) != 2:
            raise PluginFileError('Wrong Syntax in Module field, expected: module.func, flag')
        value, flag = value
        modulename, funcname = value.rstrip().rsplit('.', 1)
        from . import plugins
        moduleobj = importlib.import_module('.'+modulename, plugins.__package__)
        modulefunc = getattr(moduleobj, funcname)
        flag = flag.strip()
        if flag not in ['append', 'replace', 'challenge']:
            raise PluginFileError('Unknown flag %s for module' % flag)
        def func(game):
            if flag == 'append':
                game.move_sequence.reset()
            modulefunc(game)
            game.set_plugin_mode(flag)
        return func
        
    def parse(self):
        if not self.paras:
            raise PluginFileError('empty plugin file')
        self.header = self.paras.pop(0)
        default = {k:v for k,v in self.header.items() if k in self.header_scheme and k in self.body_scheme}
        self.parse_para(self.header, self.header_scheme)
        
        def check_conflicts(para):
            for conflicts in self.conflicts:
                if sum(int(c in para) for c in conflicts) > 1:
                    debug('Only one field allowed of:', ', '.join(conflicts))
                    debug('Skipping paragraph %s in plugin file' % para.get('Path'))
                    return False
            return True
            
        while self.paras:
            para = self.paras.pop(0)
            if not para:
                continue
            if DEBUG_ALG: debug('    Path:', para.get('Path'))
            if not check_conflicts(para):
                continue
            for k, v in default.items():
                para.setdefault(k, v)
            try:
                self.parse_para(para, self.body_scheme)
            except PluginFileError as e:
                debug('Skipping paragraph in plugin file:', e, 'keys:', *para.keys())
            else:
                self.body.append(para)
        self.paras = None
                
    @staticmethod
    def parse_para(para, scheme):
        for key, (required, singleline, convert_func) in scheme.items():
            try:
                value, lineno, comment = para[key]
            except KeyError:
                if required:
                    raise PluginFileError('missing required field {!r}'.format(key))
                else:
                    value, lineno, comment = None, None, None
            if singleline and value is not None:
                if len(value) == 0:
                    value = None
                elif len(value) == 1:
                    value = value[0]
                else:
                    raise PluginFileError('{!r} is a singleline field, but {} lines found'.format(key, len(value)))
            para[key] = convert_func(value, lineno, comment)
            
            
class PluginHelper:
    def __init__(self):
        self.scripts = OrderedDict()
        
    @staticmethod
    def test_model(model, models):
        for unused_modelstr, mtype, sizes, exp in models:
            if mtype == '*':
                return True
            if mtype is None:
                continue
            if model.type != mtype:
                continue
            sizedict = {}
            for size1, size2 in zip(sizes, model.sizes):
                if size1 is None:
                    continue
                if type(size1) is int:
                    if size1 != size2:
                        break
                else:
                    sizedict[size1] = size2
            else:
                try:
                    if exp is None or eval(exp, {}, sizedict):
                        return True
                except ValueError:
                    continue
        return False
        
    def get_function(self, model, index):
        all_models = []
        for models, func in list(self.scripts.values())[index]:
            if self.test_model(model, models):
                assert func is not None
                return func
            all_models += models
        else:
            # all model rejected
            all_models = [model_[0] for model_ in all_models if model_[1] is not None]
            if not all_models:
                raise PluginModelCompatError(_('This plugin does not work for any model.')+'\n')
            elif len(all_models) == 1:
                raise PluginModelCompatError(_('This plugin only works for:') + '\n  ' + all_models[0])
            else:
                raise PluginModelCompatError(
                        _('This plugin only works for:') + '\n' +
                        '\n'.join([' • ' + m for m in all_models])
                    )
        
    def load_plugins_from_directory(self, dirname):
        if DEBUG_MSG or DEBUG_LIVEPLUGINS:
            print('loading plugins from', repr(dirname))
        for filename in sorted(os.listdir(dirname), key=str.lower):
            unused_name, ext = os.path.splitext(filename)
            if ext != '.plugins':
                continue
            if DEBUG_ALG: debug('  plugins:', filename)
            try:
                self.load_plugins_from_file(os.path.join(dirname, filename))
            except Exception:
                sys.excepthook(*sys.exc_info())
        
    def load_plugins(self):
        self.scripts.clear()
        for dirname in [config.PLUGINS_DIR, config.get_data_home(config.PACKAGE, 'plugins')]:
            if os.path.isdir(dirname):
                debug('Found plugins path:', dirname)
                self.load_plugins_from_directory(dirname)
            else:
                debug('Plugins path does not exist:', dirname)
        if DEBUG_ALG: debug('found', len(self.scripts))
        return [(path, i) for i, path in enumerate(self.scripts.keys())]
        
    @staticmethod
    def resolve_depends(params):
        depends = []
        # resolve dependencies to other scripts
        for depend in params.depends:
            for path, unused_models, instance in params.scripts:
                if path == depend:
                    depends += instance.params.depends
            depends.append(depend)
        if DEBUG_ALG:
            debug('    Resolve dependencies for:', params.path)
            if params.depends != depends:
                debug('      before:', params.depends)
                debug('      after: ', depends)
        params.depends[:] = depends
        # resolve dependencies to params
        for path, unused_models, instance in params.scripts:
            depends = []
            for depend in instance.params.depends:
                if params.path == depend:
                    depends += params.depends
                depends.append(depend)
            if DEBUG_ALG:
                debug('      Resolve dependencies for:', path)
                if instance.params.depends != depends:
                    debug('        before:', instance.params.depends)
                    debug('        after: ', depends)
            instance.params.depends[:] = depends
            
    def load_plugins_from_file(self, filename, solutionfactory=None):
        try:
            file = PluginFileLoader(filename)
        except PluginFileError as e:
            print('Skipping plugin file:', e)
            return
            
        scripts = []
        for para in file.body:
            path, pathitems, *unused = para['Path']
            models = para['Model']
            depends = para['Depends']
            solution = para['Solution']
            moves = para['Moves']
            module = para['Module']
            if depends or solution:
                params = ScriptParams(
                                depends=depends or [],
                                precond=para['Ref-Blocks'] or [],
                                solution=solution,
                                scripts=scripts,
                                path=path,
                            )
                self.resolve_depends(params)
                func = (solutionfactory or Solution)(params)
            elif moves is not None:
                func = moves
            elif module is not None:
                func = module
            else:
                debug('    skip Path without plugin:', path)
                continue
            scripts.append((path, models, func))
            if pathitems is not None:
                funclist = self.scripts.setdefault(pathitems, [])
                funclist.append((models, func))
        
        
ScriptParams = namedtuple('ScriptParams', 'depends precond solution scripts path')

class Solution:
    def __init__(self, params):
        self.solved_face_colors = {}
        self.params = params
        
    def __call__(self, game):
        game.move_sequence.reset()
        scripts = {path: func for path, models, func in self.params.scripts}
        for depend in self.params.depends:
            instance = scripts[depend]
            self.execute(game, instance.params)
        self.execute(game, self.params)
        game.set_plugin_mode('append')
        
    def test_face(self, cube, blockpos, position, condition):
        color1 = cube.get_colorsymbol(blockpos, position)
        color2 = self.solved_face_colors.setdefault(condition, color1)
        return color1 == color2
        
    def test_basic_condition(self, cube, position, condition):
        assert len(position) == len(condition)
        blockpos = cube.model.blocksym_to_blockpos(position)
        for pos, cond in zip(position, condition):
            if not self.test_face(cube, blockpos, pos, cond):
                return False
        return True
        
    @staticmethod
    def opposite(face):
        #FIXME: this only works for BrickModel
        return {  'F': 'B', 'B': 'F',
                  'L': 'R', 'R': 'L',
                  'U': 'D', 'D': 'U',
               }[face]
        
    def test_pattern_condition(self, cube, position, condition):
        if '?' in condition:
            conditions = (condition.replace('?', face, 1)
                            for face in 'FLUBRD'
                                if face not in condition
                                if self.opposite(face) not in condition)
            return self.test_or_conditions(cube, position, conditions)
        else:
            return self.test_basic_condition(cube, position, condition)
            
    @staticmethod
    def rotated_conditions(condition):
        for i in range(len(condition)):
            yield condition[i:] + condition[:i]
        
    def test_prefix_condition(self, cube, position, condition):
        if condition.startswith('!*'):
            return not self.test_or_conditions(cube, position, self.rotated_conditions(condition[2:]))
        elif condition.startswith('*'):
            #TODO: Instead of testing only rotated conditions, test all permutations.
            #      This should not break existing rules, and would allow to match
            #      e.g. dfr and dfl. Could be done by comparing sorted strings after
            #      expanding patterns.
            return self.test_or_conditions(cube, position, self.rotated_conditions(condition[1:]))
        elif condition.startswith('!'):
            return not self.test_pattern_condition(cube, position, condition[1:])
        else:
            return self.test_pattern_condition(cube, position, condition)
        
    def test_or_conditions(self, cube, position, conditions):
        for prefix_cond in conditions:
            if self.test_prefix_condition(cube, position, prefix_cond):
                return True
        return False
        
    def test_and_conditions(self, cube, conditions):
        for position, or_cond in conditions:
            if not self.test_or_conditions(cube, position, or_cond.split('|')):
                return False
        return True
        
    def execute(self, game, params):
        rules = params.solution
        if rules is None:
            return
        cube = game.current_state
        count = 0
        pos = 0
        if DEBUG_ALG:
            print('{}:'.format(params.path))
            print(' ', cube)
        while pos < len(rules):
            self.solved_face_colors.clear()
            rule, conditions, moves = rules[pos]
            if self.test_and_conditions(cube, params.precond + conditions):
                if DEBUG_ALG:
                    print('  accept: {:2}. {}'.format(pos+1, rule))
                if moves == '@@solved':
                    return
                if moves == '@@unsolvable':
                    raise PluginSolverAbort(_('This puzzle is not solvable.'))
                if count > 4 * len(rules): # this value is just guessed
                    raise PluginSolverAbort(
                        'An infinite loop was detected. '
                        'This is probably an error in the solution.')
                count += 1
                
                game.add_code(moves)
                pos = 0
            else:
                if DEBUG_ALG:
                    print('  reject: {:2}. {}'.format(pos+1, rule))
                pos += 1
        raise PluginSolverAbort(
            'No matching rules found. '
            'This is probably an error in the solution.')