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(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))