path: root/abdl/_vm.py

# This file is part of A Boneless Datastructure Language
# Copyright (C) 2020  Soni L.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

"""[Internal] The VM Interpreter.

This module holds the VM instructions and the interpreter loop.
"""

import collections.abc
import re
from abdl import predicates
from abdl import exceptions

class PatternElement:
    """A parsed pattern element (token) and VM instruction."""

    def on_not_in_key(self, frame, path, defs):
        """Called while the key matching step of the pattern isn't done.

        Returns whether the key matching step is ready.
        """
        raise RuntimeError(self)

    def on_in_key(self, frame, path, defs):
        """Called while the key matching step of the pattern is done.

        Returns whether the key matching step is ready.
        """
        raise RuntimeError(self)

    def collect_params(self, res: list):
        """Appends parameter names used in this pattern to ``res``.
        """

    @classmethod
    def action(cls, toks):
        """Parse action, for pyparsing.

        Returns:
            PatternElement: Parsed token.
        """
        return [cls(toks)]

class Arrow(PatternElement):
    """The 'arrow' token."""

    def on_not_in_key(self, frame, path, defs):
        assert not path[-1].empty
        path.append(Holder(value=None, parent=path[-1].value, empty=True))
        return False

    @classmethod
    def action(cls, toks):
        return [cls()]

class StringKey(PatternElement):
    """The 'literal' token."""

    def __init__(self, toks):
        self.key = toks[0]
        self.skippable = toks[1] == '?'

    def on_in_key(self, frame, path, defs):
        return self.on_not_in_key(frame, path, defs)

    def on_not_in_key(self, frame, path, defs):
        path[-1].iterator = self._extract(path[-1].parent)
        path[-1].empty = False
        return True

    def _extract(self, obj):
        try:
            yield (self.key, obj[self.key])
        except (TypeError, IndexError, KeyError):
            if not self.skippable:
                raise exceptions.ValidationError

class RegexKey(PatternElement):
    """The 'regex' token."""

    def __init__(self, toks):
        self.key = toks[0]
        self.compiled = re.compile(self.key)
        self.skippable = toks[1] == '?'

    def on_in_key(self, frame, path, defs):
        return self.on_not_in_key(frame, path, defs)

    def on_not_in_key(self, frame, path, defs):
        filtered_iterator = self._filter(path[-1].iterator)
        del path[-1].iterator
        path[-1].iterator = filtered_iterator
        del filtered_iterator
        path[-1].empty = False
        return True

    def _filter(self, iter_):
        for elem in iter_:
            try:
                if self.compiled.search(elem[0]):
                    yield elem
                elif not self.skippable:
                    raise exceptions.ValidationError
            except TypeError:
                if not self.skippable:
                    raise exceptions.ValidationError

class KeySubtree(PatternElement):
    """The 'keymatch' token."""

    def __init__(self, toks):
        self.key = toks[0]
        self.skippable = toks[1] == '?'

    def on_in_key(self, frame, path, defs):
        return self.on_not_in_key(frame, path, defs)

    def on_not_in_key(self, frame, path, defs):
        path[-1].subtree = True
        filtered_iterator = self._filter(path[-1].iterator, defs, name=path[-1].name)
        del path[-1].iterator
        path[-1].iterator = filtered_iterator
        del filtered_iterator
        path[-1].empty = False
        return True

    def _filter(self, iter_, defs, name):
        for pair in iter_:
            for matches in match_helper(self.key, defs, pair[0]):
                if name:
                    # FIXME this "name" thing is a bit suboptimal
                    matches.setdefault(name, pair)
                yield (matches, pair[1])

    def collect_params(self, res: list):
        for sub in self.key:
            sub.collect_params(res)

class ValueSubtree(PatternElement):
    """The 'subvalue' token."""

    def __init__(self, toks):
        self.key = toks[0]
        self.skippable = toks[1] == '?'

    def on_not_in_key(self, frame, path, defs):
        assert not path[-1].empty
        path.append(Holder(value=None, parent=path[-1].value, empty=False, subtree=True))
        path[-1].iterator = self._filter(path[-1].parent, defs)
        return True

    def _filter(self, parent, defs):
        for pair in match_helper(self.key, defs, parent):
            yield (pair, parent)

    def collect_params(self, res: list):
        for sub in self.key:
            sub.collect_params(res)

class Ident(PatternElement):
    """The 'identifier' token."""

    def __init__(self, toks):
        self.key = toks[0]

    def on_not_in_key(self, frame, path, defs):
        path[-1].name = self.key
        path[-1].empty = False
        return True

class Param(PatternElement):
    """The 'parameter' token."""

    def __init__(self, toks):
        assert isinstance(toks[1], Ident)
        self.skippable = toks[0] == '?'
        self.key = toks[1].key

    def on_in_key(self, frame, path, defs):
        return self.on_not_in_key(frame, path, defs)

    def on_not_in_key(self, frame, path, defs):
        path[-1].iterator = self._extract(path[-1].parent, defs[self.key])
        path[-1].empty = False
        return True

    def _extract(self, obj, key):
        try:
            yield (key, obj[key])
        except (TypeError, IndexError, KeyError):
            if not self.skippable:
                raise exceptions.ValidationError

    def collect_params(self, res: list):
        res.append(self.key)

class ApplyPredicate(PatternElement):
    """The 'predicate' token."""

    def __init__(self, toks):
        assert isinstance(toks[1], Ident)
        self.skippable = toks[0] == '?'
        self.key = toks[1].key

    def on_in_key(self, frame, path, defs):
        filtered_iterator = self._filter(path[-1].iterator, defs)
        del path[-1].iterator
        path[-1].iterator = filtered_iterator
        del filtered_iterator
        path[-1].empty = False
        return True

    def _check(self, defs, obj):
        # pylint: disable=protected-access
        if predicates._to_predicate(defs[self.key]).accept(obj):
            return True
        if self.skippable:
            return False
        raise exceptions.ValidationError

    def on_not_in_key(self, frame, path, defs):
        assert len(path) == 1
        if not self._check(defs, path[-1].value):
            path.clear()
        return False

    def _filter(self, iter_, defs):
        for elem in iter_:
            if self._check(defs, elem[1]):
                yield elem

    def collect_params(self, res: list):
        res.append(self.key)

class End(PatternElement):
    """Pseudo-token, used to advance iteration."""

    def on_in_key(self, frame, path, defs):
        try:
            path[-1].next()
            return False
        except StopIteration:
            path.pop()
            while frame.prev() and not isinstance(frame.current_op, End):
                pass
            if not frame.prev():
                path.clear()
        return True

    @classmethod
    def action(cls, toks):
        return [cls()]

def _pairs(obj):
    if isinstance(obj, collections.abc.Mapping):
        return iter(obj.items())
    if isinstance(obj, collections.abc.Sequence):
        return iter(enumerate(obj, 0))
    if isinstance(obj, collections.abc.Set):
        return iter(((e, e) for e in obj))
    # maybe there's more stuff I can implement later
    raise TypeError

class Holder:
    """Stores a single match and associated metadata.

    A single match is generally a key-value pair, but may be a collection of
    named pairs in the case of subtree matches.
    """

    def __init__(self, value, parent=None, iterator=None, empty=False, subtree=False):
        self.name = None
        self.match = None
        self.value = value
        self.empty = empty
        self._iterator = iterator
        self.parent = parent
        self.subtree = subtree

    @property
    def iterator(self):
        """Returns the iterator for this match."""
        if self._iterator is None:
            self._iterator = _pairs(self.parent)
        return self._iterator

    @iterator.setter
    def iterator(self, value):
        assert self._iterator is None
        self._iterator = value

    @iterator.deleter
    def iterator(self):
        self._iterator = None

    def next(self):
        """Updates the stored match."""
        self.match, self.value = next(self.iterator)

class _Frame:
    def __init__(self, ops):
        self.ops = ops
        self.iar = -1

    def next(self):
        """Advances the instruction address register.

        Returns:
            ``True`` if successful, ``False``otherwise.
        """

        iar = self.iar + 1
        if iar >= len(self.ops):
            return False
        self.iar = iar
        return True

    @property
    def current_op(self):
        """Returns the current instruction."""
        return self.ops[self.iar]

    def prev(self):
        """Rewinds the instruction address register.

        Returns:
            ``True`` if successful, ``False``otherwise.
        """

        iar = self.iar - 1
        if iar < 0:
            return False
        self.iar = iar
        return True

def match_helper(ops, defs, tree):
    """The interpreter loop itself.

    The opcode/token dispatch logic is implemented through ``PatternElement``.

    Yields:
        dict: Matches.
    """

    frame = _Frame(ops)

    path = [Holder(value=tree, parent=None, iterator=iter(()))]
    in_key = False
    while path:
        if not frame.next():
            assert not path[-1].empty
            res = {}
            for holder in path:
                if holder.subtree:
                    for name, pair in holder.match.items():
                        res[name] = pair
                elif holder.name is not None:
                    res[holder.name] = (holder.match, holder.value)
            yield res
            assert len(path) == 1 or isinstance(frame.current_op, End)
            if not frame.prev():
                return
            in_key = True
        else:
            if in_key:
                in_key = frame.current_op.on_in_key(frame, path, defs)
            else:
                in_key = frame.current_op.on_not_in_key(frame, path, defs)