controller-wisetopdataserver/python/biopro/devlib/expression.py

import abc
from typing import Any, Optional, List, Dict, Union, Generic, Callable, NoReturn, Set, Iterable, TypeVar

from . import *

_RUNTIME_COMPILE = False

S = TypeVar('S', bound='Scope')


class Scope(metaclass=abc.ABCMeta):
    """variable lookup scope.

    """

    __slots__ = ('__parent',)

    def __init__(self):
        self.__parent: Optional[Scope] = None

    def __contains__(self, item: str) -> bool:
        """has variable defined?"""
        scope = self

        while scope is not None:
            if scope.contain_local(item):
                return True
            else:
                scope = scope.__parent

        return False

    def __getitem__(self, item: str) -> Any:
        """get variable through the scope chain."""
        scope = self

        while scope is not None:
            if scope.contain_local(item):
                return scope.get_local(item)
            else:
                scope = scope.__parent

        raise KeyError(item)

    def __setitem__(self, key: str, value: Any):
        """set variable to the scope which define it."""
        scope = self

        while scope is not None:
            if scope.contain_local(key):
                return scope.set_local(key, value)
            else:
                scope = scope.__parent

        self.set_local(key, value)

    def __delitem__(self, key: str):
        """delete variable from the scope which define it"""
        scope = self

        while scope is not None:
            if scope.contain_local(key):
                return scope.del_local(key)
            else:
                scope = scope.__parent

        raise KeyError(key)

    @abc.abstractmethod
    def locals(self) -> Iterable[str]:
        """variable key defined in this scope"""
        pass

    def globals(self) -> Set[str]:
        """variable key existed in this scope"""
        if self.__parent is not None:
            ret = self.__parent.globals()
        else:
            ret = set()

        ret.update(self.locals())

        return ret

    @abc.abstractmethod
    def contain_local(self, key: str) -> bool:
        """has variable in current scope?"""
        pass

    @abc.abstractmethod
    def get_local(self, key: str) -> Any:
        """get variable from this scope"""
        pass

    @abc.abstractmethod
    def set_local(self, key: str, value: Any):
        """set variable to this scope."""
        pass

    @abc.abstractmethod
    def del_local(self, key: str):
        """remove variable from this scope"""
        pass

    @abc.abstractmethod
    def clear(self):
        """clear all variable define in this scope."""
        pass

    def child(self, **local) -> 'Scope':
        """new a child scope with variable *local*"""
        ret = ContextScope()
        ret.__parent = self

        for k, v in local.items():
            ret.set_local(k, v)

        return ret

    def chain(self, scope: S) -> S:
        scope.__parent = self

        return scope

    def wrap(self, scope: 'Scope') -> 'Scope':
        ret = WrapScope(scope)
        ret.__parent = self

        return ret

    def eval(self, expression: str) -> Any:
        # Internal instruction are only handled with a simple form in the ListInstruction,
        # so if a internal instruction is put in the deep expression, scope doesn't know how to
        # evaluate it. Actually, we can not evaluate internal instruction here, because this eval()
        # function is proposed to evaluate nothing but a value, not a function call,
        # we just return the expression itself.
        if expression.startswith('_') and '(' in expression and expression.endswith(')'):
            return expression

        try:
            # noinspection PyTypeChecker
            return eval(expression, {}, self)
        except BaseException as e:
            raise RuntimeError(expression) from e

    def __str__(self):
        if self.__parent is None:
            return ' > {}'
        else:
            return ' > ' + str(self.__parent)

    __repr__ = __str__


class ContextScope(Scope):
    """General scope implement."""

    __slots__ = ('_context',)

    def __init__(self, init: Optional[Dict[str, Any]] = None):
        super().__init__()

        self._context: Dict[str, Any] = {}

        if init is not None:
            self._context.update(init)

    def locals(self) -> Iterable[str]:
        return self._context.keys()

    def contain_local(self, key: str) -> bool:
        return key in self._context

    def get_local(self, key: str):
        return self._context[key]

    def set_local(self, key: str, value: Any):
        self._context[key] = value

    def del_local(self, key: str):
        del self._context[key]

    def clear(self):
        self._context.clear()

    def __str__(self):
        return str(self._context) + super().__str__()


class ConstantScope(Scope):
    """constant scope, which cannot update, removed the variable defined in this scope after initialized."""
    EMPTY: 'ConstantScope' = None

    __slots__ = ('_constant',)

    def __init__(self, constant: Dict[str, Any]):
        super().__init__()

        self._constant = dict(constant)

    def locals(self) -> Iterable[str]:
        return self._constant.keys()

    def contain_local(self, key: str) -> bool:
        return key in self._constant

    def get_local(self, key: str):
        return self._constant[key]

    def set_local(self, key: str, value: Any):
        """do nothing"""
        pass

    def del_local(self, key: str):
        """do nothing"""
        pass

    def clear(self):
        """do nothing"""
        pass

    def __str__(self):
        return 'constant:' + str(self._constant) + super().__str__()



class ReadonlyScope(Scope, metaclass=abc.ABCMeta):
    """readonly scope"""

    __slots__ = ('_variable',)

    def __init__(self, variable: Iterable[str]):
        super().__init__()

        self._variable = list(variable)

    def locals(self) -> Dict[str, Any]:
        return {p: self.get_local(p) for p in self._variable}

    def contain_local(self, key: str) -> bool:
        return key in self._variable

    @abc.abstractmethod
    def get_local(self, key: str) -> Any:
        pass

    def set_local(self, key: str, value: Any):
        """variable in this scope are read-only. do nothing"""
        pass

    def del_local(self, key: str):
        """variable in this scope are read-only. do nothing"""
        pass

    def clear(self):
        """variable in this scope are read-only. do nothing"""
        pass

    def __str__(self):
        return self.__class__.__name__ + ':' + str(self._variable) + super().__str__()



ConstantScope.EMPTY = ConstantScope({})


class WrapScope(Scope):
    __slots__ = ('_scope',)

    def __init__(self, scope: Scope):
        super().__init__()
        self._scope = scope

    def locals(self) -> Iterable[str]:
        return self._scope.locals()

    def contain_local(self, key: str) -> bool:
        return self._scope.contain_local(key)

    def get_local(self, key: str) -> Any:
        return self._scope.get_local(key)

    def set_local(self, key: str, value: Any):
        self._scope.set_local(key, value)

    def del_local(self, key: str):
        self._scope.del_local(key)

    def clear(self):
        self._scope.clear()

    def __getattr__(self, item):
        return getattr(self._scope, item)

    def __str__(self):
        return str(self._scope) + super().__str__()



class Expression(Generic[T], JsonSerialize, metaclass=abc.ABCMeta):
    """
    **json format**

    ::

        Expression[T]
            # constant value
            = ConstantExpression[T]     # type: null, bool, int, float
            | ArrayExpression[T]        # type: list
            # dynamic value
            | ComplexExpression[T]      # type: str, dict

    **variable scope search chain**

    1.  parent scope

    """

    __slots__ = ()

    @abc.abstractmethod
    def value(self, context: Scope) -> T:
        """eval expression under scope *context* and get a value."""
        pass

    def value_local(self, context: Scope, **variable) -> T:
        """eval expression under scope *context* with some local variable and get a value."""
        return self.value(context.child(**variable))

    @abc.abstractmethod
    def as_json(self):
        pass

    @classmethod
    def parse(cls, json: Optional[JSON]) -> 'Expression':
        if json is None:
            return NullConstant

        elif json is True:
            return TrueConstant

        elif json is False:
            return FalseConstant

        elif isinstance(json, (int, float)):
            return ConstantExpression(json)

        elif isinstance(json, str):
            return ComplexExpression(json)

        elif isinstance(json, list):
            return ArrayExpression.parse(json)

        elif isinstance(json, dict):
            return ComplexExpression.parse(json)

        else:
            raise RuntimeError('illegal value : ' + _truncate(json))


class ConstantExpression(Expression[T]):
    """expression with constant value.

    **json format**

    ::

        ConstantExpression[T]
            = null
            | int
            | bool
            | int
            | float

    """

    __slots__ = ('_constant',)

    def __init__(self, constant: T):
        self._constant = constant

    @property
    def constant(self) -> T:
        return self._constant

    def value(self, context: Scope) -> T:
        return self._constant

    def as_json(self):
        return self._constant

    def __hash__(self):
        return hash(self._constant) + 17

    def __eq__(self, other):
        if isinstance(other, ConstantExpression):
            return self._constant == other._constant
        else:
            return self._constant == other

    def __str__(self):
        return str(self._constant)

    def __repr__(self):
        return repr(self._constant)


NullConstant = ConstantExpression(None)
TrueConstant = ConstantExpression(True)
FalseConstant = ConstantExpression(False)


class ArrayExpression(Expression[T]):
    """array expression.

    **json format**

    ::

        ArrayExpression[T]
            = [T]

    """
    __slots__ = ('_content',)

    def __init__(self, content: List[T]):
        self._content = content

    def __len__(self):
        return len(self._content)

    def __getitem__(self, index: int) -> T:
        return self._content[index]

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

    def value(self, context: Scope) -> List[T]:
        return self._content

    def as_json(self):
        return self._content

    @classmethod
    def parse(cls, json: JSON_ARRAY) -> 'ArrayExpression':
        return ArrayExpression(json)


class UndeterminedExpression(Expression[T]):
    """json object for undetermined expression

    """
    __slots__ = ('_json',)

    def __init__(self, json: JSON_OBJECT):
        self._json = json

    def value(self, context) -> NoReturn:
        raise RuntimeError('undetermined expression : ' + _truncate(self._json))

    def as_json(self):
        return self._json


class ComplexExpression(Expression[T]):
    """
    **json format**

    ::

        ComplexExpression[T]
            = UndeterminedExpression
            | ComplexExpression[T]
            | BoolExpression[T]
            | ListExpression[T]
            | WhenExpression[T]
            | GuardExpression

        ComplexExpression[T] = {
            "expression" :str = expression
            "dependency"? : str|[str]
        }

    **variable scope search chain**

    *   for value evaluation

        1.  parent scope

    *   for children expression evaluation

        1.  parent scope
        #.  {VALUE=eval result}


    """

    __slots__ = ('expression', 'dependency')

    def __init__(self, expression: str, dependency: List[str] = None):
        self.expression: str = expression
        self.dependency = dependency if dependency is not None else []

    def value(self, context: Scope) -> T:
        return context.eval(self.expression)

    def travel(self, traveler: 'ExpressionTraveler'):
        traveler.on_node(self)

    def node_keys(self) -> List[str]:
        """iterator the keys of the children nodes"""
        return []

    def get_node(self, key: str) -> Optional[Expression[T]]:
        """get child node with **key**"""
        raise AttributeError('expression %s does not have %s node' % (type(self), key))

    def set_node(self, key: str, node: Optional[Expression[T]]):
        """replace child node **key** with **node**"""
        raise AttributeError('expression %s does not have %s node' % (type(self), key))

    def as_json(self):
        ret = {
            'expression': self.expression,
        }

        if len(self.dependency) > 0:
            ret['dependency'] = self.dependency

        return ret

    @classmethod
    def parse(cls, json: JSON_OBJECT) -> Union['ComplexExpression', UndeterminedExpression]:
        expression = _get(json, 'expression', str, _truncate(json), nullable=True)

        if expression is None:
            # un-determined expression
            return UndeterminedExpression(json)

        if 'when' in json:
            return WhenExpression.parse(json)

        elif 'list' in json:
            return ListExpression.parse(json)

        elif 'raise' in json:
            return GuardExpression.parse(json)

        elif 'true' in json or 'false' in json or 'else' in json:
            return BoolExpression.parse(json)

        else:
            # only contain expression

            dep = _get(json, 'dependency', (str, list), 'expression', nullable=True)
            if dep is not None:
                if isinstance(dep, str):
                    dep = [dep]
                else:
                    dep = list(map(str, dep))

            return ComplexExpression(expression, dep)


class ExpressionTraveler:
    __slots__ = ('__on_node', '__on_enter', '__on_exit',
                 '__current_expr', '__current_key')

    def __init__(self,
                 on_node: Callable[[Optional[ComplexExpression], Optional[str], Optional[Expression]], None],
                 on_enter: Callable[[ComplexExpression], bool] = None,
                 on_exit: Callable[[ComplexExpression], None] = None):
        """Expression tree traveler.

        **on_node** function type : ``(parent_expr?, node_name?, node) -> None``. callable when goto this expression.

        **on_enter** function_type: ``(expr) -> bool``. callable when this expression is :class:`ComplexExpression`
        and prepare to enter this.The return result decide whether enter this expression or not.

        **on_exit** function_type: ``(expr) -> None``. callable when this expression is :class:`ComplexExpression`
        and prepare to exit this

        :param on_node:
        :param on_enter:
        :param on_exit:
        """
        self.__on_node = on_node
        self.__on_enter = on_enter
        self.__on_exit = on_exit
        #
        self.__current_expr: ComplexExpression = None
        self.__current_key: str = None

    def travel(self, root: Expression):
        if isinstance(root, ComplexExpression):
            root.travel(self)
        else:
            self.__on_node(None, None, root)

    def on_node(self, node: Expression):
        self.__on_node(self.__current_expr, self.__current_key, node)

    def on_child(self, expr: ComplexExpression, key: str, node: Optional[Expression]):
        self.__current_expr = expr
        self.__current_key = key

        if node is not None and isinstance(node, ComplexExpression):
            node.travel(self)
        else:
            self.__on_node(self.__current_expr, self.__current_key, None)

    def on_enter(self, expr: ComplexExpression) -> bool:
        if self.__on_enter is None or self.__on_enter(expr):
            return True
        else:
            return False

    def on_exit(self, expr: ComplexExpression):
        if self.__on_exit is not None:
            self.__on_exit(expr)


class BoolExpression(ComplexExpression[T]):
    """
    **json format**

    ::

        BoolExpression[T] = {
            "expression" :str = expression
            "true" = Expression[T]      # when true, property name could be:
                                        #   any lower case equal to "true"
            "false" = Expression[T]     # when false, property name could be:
                                        #   any lower case equal to "false"
                                        #   "null", "None" if "false" not present
            "*" = Expression[T]         # else. property name could be:
                                        #   any lower case equal to "else"
        }
    """

    __slots__ = ('if_true', 'if_false', 'or_else')

    def __init__(self,
                 expression: str,
                 if_true: Optional[Expression[T]] = None,
                 if_false: Optional[Expression[T]] = None,
                 or_else: Optional[Expression[T]] = None,
                 dependency: List[str] = None):
        super().__init__(expression, dependency)

        self.if_true: Optional[Expression[T]] = if_true
        self.if_false: Optional[Expression[T]] = if_false
        self.or_else: Optional[Expression[T]] = or_else

    def value(self, context: Scope) -> Union[None, str, T]:
        value = super().value(context)

        if self.if_true is not None and value:
            return self.if_true.value(context.child(VALUE=True))

        if self.if_false is not None and not value:
            return self.if_false.value(context.child(VALUE=False))

        if self.or_else is not None:
            return self.or_else.value(context.child(VALUE=value))

        else:
            return None

    def travel(self, traveler: 'ExpressionTraveler'):
        traveler.on_node(self)

        if traveler.on_enter(self):
            traveler.on_child(self, 'true', self.if_true)
            traveler.on_child(self, 'false', self.if_false)
            traveler.on_child(self, 'else', self.or_else)
            traveler.on_exit(self)

    def node_keys(self) -> List[str]:
        return ['true', 'false', 'else']

    def get_node(self, key: str) -> Optional[Expression[T]]:
        if key == 'true':
            return self.if_true
        elif key == 'false':
            return self.if_false
        elif key in ('*', 'else'):
            return self.or_else
        else:
            return super().get_node(key)

    def set_node(self, key: str, node: Optional[Expression[T]]):
        key = key.lower()
        if key == 'true':
            self.if_true = node
        elif key == 'false':
            self.if_false = node
        elif key in ('*', 'else'):
            self.or_else = node
        else:
            super().set_node(key, node)

    def as_json(self):
        ret: Dict[str, Any] = super().as_json()

        if self.if_true is not None:
            ret['true'] = self.if_true.as_json()

        if self.if_false is not None:
            ret['false'] = self.if_false.as_json()

        if self.or_else is not None:
            ret['else'] = self.or_else.as_json()

        return ret

    @classmethod
    def parse(cls, json: JSON_OBJECT) -> 'BoolExpression':
        expression = _get(json, 'expression', str, 'bool expression')

        if_true = None
        if_false = None
        or_else = None

        for k, v in json.items():
            k = k.lower()

            if k == 'true':
                if if_true is None:
                    if_true = Expression.parse(v)
                else:
                    raise RuntimeError('bool expression "true" duplicate : ' + _truncate(json))
            elif k == 'false':
                if if_false is None:
                    if_false = Expression.parse(v)
                else:
                    raise RuntimeError('bool expression "false" duplicate : ' + _truncate(json))

            elif k in ('*', 'else'):
                if or_else is None:
                    or_else = Expression.parse(v)
                else:
                    raise RuntimeError('bool expression "else" duplicate : ' + _truncate(json))

            else:
                raise RuntimeError('unknown bool condition : ' + k + ' in ' + _truncate(json))

        dep = _get(json, 'dependency', (str, list), 'when expression', nullable=True)
        if dep is not None:
            if isinstance(dep, str):
                dep = [dep]
            else:
                dep = list(map(str, dep))

        return BoolExpression(expression, if_true, if_false, or_else, dep)


class ListExpression(ComplexExpression[T]):
    """
    **json format**

    ::

        ListExpression[T] = {
            "expression"? :str = expression
            "list": [T|Expression[T]]
        }
    """

    __slots__ = ('_list',)

    def __init__(self,
                 exp_list: List[Union[T, Expression[T]]],
                 expression: str = None,
                 dependency: List[str] = None):
        super().__init__(expression, dependency)

        self._list: List[Expression[T]] = exp_list

    def value(self, context: Scope) -> Union[str, T]:
        if self.expression is not None:
            value = context.eval(self.expression)
        else:
            value = context["VALUE"]

        element = self._list[int(value)]
        if isinstance(element, Expression):
            return element.value(context.child(VALUE=value))
        else:
            return element

    def travel(self, traveler: ExpressionTraveler):
        traveler.on_node(self)

        if traveler.on_enter(self):

            for i, e in enumerate(self._list):
                traveler.on_child(self, str(i), e)

            traveler.on_exit(self)

    def node_keys(self) -> List[str]:
        return list(map(str, range(len(self._list))))

    def get_node(self, key: str) -> Union[T, Expression[T]]:
        try:
            return self._list[int(key)]
        except (ValueError, IndexError):
            pass

        return super().get_node(key)

    def set_node(self, key: str, node: Union[T, Expression[T]]):
        try:
            # exam key
            self._list[int(key)]
        except (ValueError, IndexError) as e:
            return super().set_node(key, node)

        else:
            if node is None:
                raise RuntimeError('NoneType')

            self._list[int(key)] = node

    def as_json(self):
        ret = {
            'list': [JsonSerialize.to_json(v) for v in self._list]
        }

        if self.expression is not None:
            ret['expression'] = self.expression

        if len(self.dependency) > 0:
            ret['dependency'] = self.dependency

        return ret

    @classmethod
    def parse(cls, json: JSON_OBJECT) -> 'ListExpression':
        expression = _get(json, 'expression', str, 'list expression', nullable=True)

        exp_list = _get(json, 'list', list, 'list expression')
        exp_list = [Expression.parse(v) for v in exp_list]

        dep = _get(json, 'dependency', (str, list), 'list expression', nullable=True)
        if dep is not None:
            if isinstance(dep, str):
                dep = [dep]
            else:
                dep = list(map(str, dep))

        return ListExpression(exp_list, expression, dep)


class WhenExpression(ComplexExpression[T]):
    """
    **json format**

    ::

        WhenExpression[T] = {
            "expression" :str = expression
            "when" = {
                str_value = Expression[T]
                "*"? = Expression[T]            # else
            }
        }
    """

    __slots__ = ('_when',)

    def __init__(self,
                 expression: str,
                 when: Dict[str, Expression[T]],
                 dependency: List[str] = None):
        super().__init__(expression, dependency)

        self._when: Dict[str, Expression[T]] = when

    def value(self, context: Scope) -> Union[str, T]:
        value = super().value(context)
        key = str(int(value))

        if key in self._when:
            return self._when[key].value(context.child(VALUE=value))

        if '*' in self._when:
            return self._when['*'].value(context.child(VALUE=value))
        else:
            return None

    def travel(self, traveler: ExpressionTraveler):
        traveler.on_node(self)

        if traveler.on_enter(self):
            for k in list(self._when.keys()):
                traveler.on_child(self, k, self._when[k])
            traveler.on_exit(self)

    def node_keys(self) -> List[str]:
        return list(self._when.keys())

    def get_node(self, key: str) -> Expression[T]:
        if key in self._when:
            return self._when[key]
        else:
            return super().get_node(key)

    def set_node(self, key: str, node: Optional[Expression[T]]):
        if node is not None:
            self._when[key] = node

        elif key in self._when:
            del self._when[key]

    def as_json(self):
        ret: Dict[str, Any] = super().as_json()

        ret['when'] = {k: v.as_json() for k, v in self._when.items()}

        return ret

    @classmethod
    def parse(cls, json: JSON_OBJECT) -> 'WhenExpression':
        expression = _get(json, 'expression', str, 'when expression')

        when = _get(json, 'when', dict, 'when expression')
        when = {k: Expression.parse(v) for k, v in when.items()}

        dep = _get(json, 'dependency', (str, list), 'when expression', nullable=True)
        if dep is not None:
            if isinstance(dep, str):
                dep = [dep]
            else:
                dep = list(map(str, dep))

        return WhenExpression(expression, when, dep)


class GuardExpression(ComplexExpression[bool]):
    """guard expression, if *expression* is not satisfy, which fail the ``if`` test,
    program will raise an error.

    **json format**

    ::

        GuardExpression
            = expression:str
            | guard_instance

        guard_instance = {
            "expression" : str = expression
            "error_type"? :str
            "raise"? :str
        }

    """
    __slots__ = ('_error_type', '_message')

    def __init__(self,
                 expression: str,
                 message: Optional[str] = None,
                 error_type: Optional[str] = None):
        super().__init__(expression)

        self._error_type = error_type
        self._message = message

    @property
    def error_type(self):
        if self._error_type is None:
            return RuntimeError

        # noinspection PyBroadException
        try:
            e = eval(self._error_type)
        except:
            return RuntimeError
        else:
            if isinstance(e, BaseException):
                return e
            else:
                return RuntimeError

    @property
    def error_message(self) -> Optional[str]:
        return self._message

    def value(self, context: Scope) -> None:
        if not super().value(context):
            self.raise_error()

    def test(self, context: Scope) -> bool:
        return super().value(context)

    def raise_error(self):
        if self._message is None:
            message = 'guard : ' + self.expression

        else:
            message = self._message

        # noinspection PyBroadException
        try:
            err = self.error_type(message)
        except:
            err = None

        if isinstance(err, BaseException):
            raise err

        else:
            raise RuntimeError(message)

    def as_json(self):
        ret = {
            'expression': self.expression
        }

        if self._error_type is not None:
            ret['error_type'] = self._error_type

        if self._message is not None:
            ret['raise'] = self._message

        return ret

    @classmethod
    def parse(cls, json: JSON) -> 'GuardExpression':
        if isinstance(json, str):
            return GuardExpression(json)

        elif isinstance(json, dict):
            expression = _get(json, 'expression', str, 'guard expression')
            error_type = _get(json, 'error_type', str, 'guard expression', nullable=True)
            message = _get(json, 'raise', str, 'guard expression', nullable=True)

            return GuardExpression(expression, message=message, error_type=error_type)

        else:
            raise RuntimeError('illegal guard expression : ' + _truncate(json))


class ListGuardExpression(ListNode[GuardExpression]):
    """
    **json format**

    ::

        ListGuardExpression
            = [GuardExpression]

    """

    __slots__ = ()

    def __init__(self, init: List[GuardExpression] = None):
        super().__init__(init if init is not None else [])

    def value(self, context: Scope) -> None:
        """testing all guard expression

        :param context:
        """
        for e in self:
            e.value(context)

    def test(self, context: Scope) -> Optional[GuardExpression]:
        """testing all guard expression.

        :param context:
        :return: the first failed guard expression. None if all pass
        """
        for e in self:
            if not e.test(context):
                return e

        return None

    @classmethod
    def parse(cls, json: JSON_ARRAY) -> 'ListGuardExpression':
        return ListGuardExpression(_list(json, GuardExpression.parse))