Enhanced PythonEvaluator to accept concepts
This commit is contained in:
@@ -0,0 +1,110 @@
|
||||
from core.builtin_concepts import BuiltinConcepts, ListConcept
|
||||
from core.concept import Concept
|
||||
import ast
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class NodeParent:
|
||||
"""
|
||||
Class that represent the ancestor of a Node
|
||||
For example, the 'For' nodes has three fields (target, iter and body)
|
||||
So, for a node under For.iter
|
||||
node -> For
|
||||
field -> iter
|
||||
"""
|
||||
|
||||
def __init__(self, node, field):
|
||||
self.node = node
|
||||
self.field = field
|
||||
|
||||
def __repr__(self):
|
||||
if self.node is None:
|
||||
return None
|
||||
|
||||
if self.field is None:
|
||||
return self.node.get_node_type()
|
||||
|
||||
return self.node.get_node_type() + "." + self.field
|
||||
|
||||
def __eq__(self, other):
|
||||
# I can compare with type for simplification
|
||||
if isinstance(other, tuple):
|
||||
return self.node.get_node_type() == other[0] and self.field == other[1]
|
||||
|
||||
# normal equals implementation
|
||||
if not isinstance(other, NodeParent):
|
||||
return False
|
||||
|
||||
return self.node.get_node_type() == other.node.get_node_type() and self.field == other.field
|
||||
|
||||
def __hash__(self):
|
||||
return hash((self.node.get_node_type(), self.field))
|
||||
|
||||
|
||||
class NodeConcept(Concept):
|
||||
def __init__(self, key, parent: NodeParent):
|
||||
super().__init__(key, True, False, key)
|
||||
self.parent = parent
|
||||
|
||||
def get_node_type(self):
|
||||
return self.key
|
||||
|
||||
|
||||
class GenericNodeConcept(NodeConcept):
|
||||
def __init__(self, node_type, parent):
|
||||
super().__init__(BuiltinConcepts.GENERIC_NODE, parent)
|
||||
self.node_type = node_type
|
||||
|
||||
def __repr__(self):
|
||||
return "Generic:" + self.node_type
|
||||
|
||||
def get_node_type(self):
|
||||
return self.node_type
|
||||
|
||||
def get_value(self):
|
||||
if self.node_type == "Name":
|
||||
return self.get_prop("id")
|
||||
|
||||
if self.node_type == "arg":
|
||||
return self.get_prop("arg")
|
||||
|
||||
return self.body
|
||||
|
||||
|
||||
class IdentifierConcept(NodeConcept):
|
||||
def __init__(self, parent, name):
|
||||
super().__init__(BuiltinConcepts.IDENTIFIER_NODE, parent)
|
||||
self.body = name
|
||||
|
||||
|
||||
def transform(node):
|
||||
"""
|
||||
Transform Python AST node into concept nodes
|
||||
for better usage
|
||||
:param node:
|
||||
:return:
|
||||
"""
|
||||
|
||||
def _transform(node, parent):
|
||||
node_type = node.__class__.__name__
|
||||
concept = GenericNodeConcept(node_type, parent).init_key()
|
||||
for field in node._fields:
|
||||
if not hasattr(node, field):
|
||||
continue
|
||||
|
||||
value = getattr(node, field)
|
||||
if isinstance(value, list):
|
||||
lst = ListConcept().init_key()
|
||||
for i in value:
|
||||
lst.append(_transform(i, NodeParent(concept, field)))
|
||||
concept.set_prop(field, lst)
|
||||
elif isinstance(value, ast.AST):
|
||||
concept.set_prop(field, _transform(value, NodeParent(concept, field)))
|
||||
else:
|
||||
concept.set_prop(field, value)
|
||||
return concept
|
||||
|
||||
return _transform(node, None)
|
||||
@@ -0,0 +1,122 @@
|
||||
from core.ast.nodes import GenericNodeConcept, NodeConcept
|
||||
from core.builtin_concepts import ListConcept
|
||||
|
||||
|
||||
class ConceptNodeVisitor:
|
||||
"""
|
||||
Base class to visit NodeConcept
|
||||
It is insolently inspired by python AST.Visitor class
|
||||
"""
|
||||
|
||||
def visit(self, node):
|
||||
|
||||
"""Visit a node."""
|
||||
name = node.node_type if isinstance(node, GenericNodeConcept) else node.name
|
||||
name = str(name).capitalize()
|
||||
|
||||
method = 'visit_' + name
|
||||
visitor = getattr(self, method, self.generic_visit)
|
||||
return visitor(node)
|
||||
|
||||
def generic_visit(self, node):
|
||||
"""Called if no explicit visitor function exists for a node."""
|
||||
for field, value in iter_props(node):
|
||||
if isinstance(value, ListConcept):
|
||||
for item in value:
|
||||
if isinstance(item, NodeConcept):
|
||||
self.visit(item)
|
||||
elif isinstance(value, NodeConcept):
|
||||
self.visit(value)
|
||||
|
||||
def visit_Constant(self, node):
|
||||
value = node.get_prop("value")
|
||||
type_name = _const_node_type_names.get(type(value))
|
||||
if type_name is None:
|
||||
for cls, name in _const_node_type_names.items():
|
||||
if isinstance(value, cls):
|
||||
type_name = name
|
||||
break
|
||||
if type_name is not None:
|
||||
method = 'visit_' + type_name
|
||||
try:
|
||||
visitor = getattr(self, method)
|
||||
except AttributeError:
|
||||
pass
|
||||
else:
|
||||
import warnings
|
||||
warnings.warn(f"{method} is deprecated; add visit_Constant",
|
||||
PendingDeprecationWarning, 2)
|
||||
return visitor(node)
|
||||
return self.generic_visit(node)
|
||||
|
||||
|
||||
class UnreferencedNamesVisitor(ConceptNodeVisitor):
|
||||
def __init__(self, sheerka):
|
||||
self.names = set()
|
||||
self.sheerka = sheerka
|
||||
|
||||
def visit_Name(self, node):
|
||||
parents = get_parents(node)
|
||||
if ("For", "target") in parents: # variable used by the 'for' iteration
|
||||
return
|
||||
|
||||
if ("Call", "func") in parents: # name of the function
|
||||
return
|
||||
|
||||
if ("Assign", "targets") in parents: # variable which is assigned
|
||||
return
|
||||
|
||||
if self.can_be_discarded(self.sheerka.value(node), parents):
|
||||
return
|
||||
|
||||
self.names.add(self.sheerka.value(node))
|
||||
|
||||
def can_be_discarded(self, variable_name, parents):
|
||||
|
||||
for node in (parent.node for parent in parents):
|
||||
if node is None:
|
||||
return False
|
||||
|
||||
if node.get_node_type() == "For" and self.sheerka.value(node.get_prop("target")) == variable_name:
|
||||
# variable used by the loop
|
||||
return True
|
||||
|
||||
if node.get_node_type() == "FunctionDef":
|
||||
# variable defined as a function parameter
|
||||
args = node.get_prop("args")
|
||||
args_values = list(self.sheerka.values(args.get_prop("args")))
|
||||
if variable_name in args_values:
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
|
||||
def get_parents(node):
|
||||
if node.parent is None:
|
||||
return []
|
||||
|
||||
res = []
|
||||
while True:
|
||||
if node.parent is None:
|
||||
break
|
||||
res.append(node.parent)
|
||||
node = node.parent.node
|
||||
|
||||
return res
|
||||
|
||||
|
||||
def iter_props(node):
|
||||
for p in node.props:
|
||||
yield p, node.props[p].value
|
||||
|
||||
|
||||
_const_node_type_names = {
|
||||
bool: 'NameConstant', # should be before int
|
||||
type(None): 'NameConstant',
|
||||
int: 'Num',
|
||||
float: 'Num',
|
||||
complex: 'Num',
|
||||
str: 'Str',
|
||||
bytes: 'Bytes',
|
||||
type(...): 'Ellipsis',
|
||||
}
|
||||
@@ -6,6 +6,12 @@ from core.concept import Concept
|
||||
class BuiltinConcepts(Enum):
|
||||
"""
|
||||
List of builtin concepts that do no need any specific implementation
|
||||
Please note that the value of the enum is informal. It is not used in the system
|
||||
For example, the concept 'NODE' DOES NOT have the key, the id or whatever 200
|
||||
The key if the name of the concept
|
||||
The id is a sequential number given just before the concept is saved in sdp
|
||||
|
||||
The values of the enum are just a convenient way for me to group the concepts
|
||||
"""
|
||||
SHEERKA = 1
|
||||
SUCCESS = 2
|
||||
@@ -31,6 +37,12 @@ class BuiltinConcepts(Enum):
|
||||
NOP = 22 # no operation concept. Does nothing
|
||||
PROPERTY_EVAL_ERROR = 23 # cannot evaluate a property of a concept
|
||||
ENUMERATION = 24 # represents a list or a set
|
||||
LIST = 25 # represents a list
|
||||
CANNOT_RESOLVE_VALUE_ERROR = 26 # In presence of a concept where the default value is not know
|
||||
|
||||
NODE = 200
|
||||
GENERIC_NODE = 201
|
||||
IDENTIFIER_NODE = 202
|
||||
|
||||
|
||||
"""
|
||||
@@ -234,3 +246,36 @@ class PropertyEvalError(Concept):
|
||||
@property
|
||||
def property_name(self):
|
||||
return self.body
|
||||
|
||||
|
||||
class EnumerationConcept(Concept):
|
||||
def __init__(self, iteration=None):
|
||||
super().__init__(BuiltinConcepts.ENUMERATION, True, False, BuiltinConcepts.ENUMERATION)
|
||||
self.body = iteration
|
||||
|
||||
def __iter__(self):
|
||||
return iter(self.body)
|
||||
|
||||
|
||||
class ListConcept(Concept):
|
||||
def __init__(self, items=None):
|
||||
super().__init__(BuiltinConcepts.LIST, True, False, BuiltinConcepts.LIST)
|
||||
self.body = items or []
|
||||
|
||||
def append(self, obj):
|
||||
self.body.append(obj)
|
||||
|
||||
def __len__(self):
|
||||
return len(self.body)
|
||||
|
||||
def __getitem__(self, key):
|
||||
return self.body[key]
|
||||
|
||||
def __setitem__(self, key, value):
|
||||
self.body[key] = value
|
||||
|
||||
def __iter__(self):
|
||||
return iter(self.body)
|
||||
|
||||
def __contains__(self, item):
|
||||
return item in self.body
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
from core.builtin_concepts import BuiltinConcepts
|
||||
|
||||
|
||||
def is_same_success(sheerka, return_values):
|
||||
"""
|
||||
Returns True if all returns values are successful and have the same value
|
||||
:param sheerka:
|
||||
:param return_values:
|
||||
:return:
|
||||
"""
|
||||
assert isinstance(return_values, list)
|
||||
|
||||
if not return_values[0].status:
|
||||
return False
|
||||
|
||||
reference = sheerka.value(return_values[0].value)
|
||||
|
||||
for return_value in return_values[1:]:
|
||||
if not return_value.status:
|
||||
return False
|
||||
|
||||
actual = sheerka.value(return_value.value)
|
||||
if actual != reference:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def expect_one(context, return_values):
|
||||
"""
|
||||
Checks if there is at least one success return value
|
||||
If there is more than one, check if it's the same value
|
||||
:param context:
|
||||
:param return_values:
|
||||
:return:
|
||||
"""
|
||||
|
||||
if not isinstance(return_values, list):
|
||||
return return_values
|
||||
|
||||
sheerka = context.sheerka
|
||||
|
||||
if len(return_values) == 0:
|
||||
return sheerka.ret(
|
||||
context.who,
|
||||
False,
|
||||
sheerka.new(BuiltinConcepts.IS_EMPTY, obj=return_values),
|
||||
parents=return_values)
|
||||
|
||||
successful_results = [item for item in return_values if item.status]
|
||||
number_of_successful = len(successful_results)
|
||||
# total_items = len(return_values)
|
||||
|
||||
# remove errors when a winner is found
|
||||
if number_of_successful == 1:
|
||||
# log.debug(f"1 / {total_items} good item found.")
|
||||
return sheerka.ret(
|
||||
context.who,
|
||||
True,
|
||||
successful_results[0].body,
|
||||
parents=return_values)
|
||||
|
||||
# too many winners, which one to choose ?
|
||||
if number_of_successful > 1:
|
||||
if is_same_success(sheerka, successful_results):
|
||||
return sheerka.ret(
|
||||
context.who,
|
||||
True,
|
||||
successful_results[0].value,
|
||||
parents=return_values)
|
||||
else:
|
||||
return sheerka.ret(
|
||||
context.who,
|
||||
False,
|
||||
sheerka.new(BuiltinConcepts.TOO_MANY_SUCCESS, obj=successful_results),
|
||||
parents=return_values)
|
||||
|
||||
# only errors, i cannot help you
|
||||
return sheerka.ret(
|
||||
context.who,
|
||||
False,
|
||||
sheerka.new(BuiltinConcepts.TOO_MANY_ERRORS, obj=return_values),
|
||||
parents=return_values)
|
||||
+5
-2
@@ -190,15 +190,18 @@ class Concept:
|
||||
|
||||
return self
|
||||
|
||||
def set_prop(self, prop_name, prop_value=None):
|
||||
def set_prop(self, prop_name: str, prop_value=None):
|
||||
self.props[prop_name] = Property(prop_name, prop_value)
|
||||
return self
|
||||
|
||||
def set_prop_by_index(self, index, prop_value):
|
||||
def set_prop_by_index(self, index: int, prop_value):
|
||||
prop_name = list(self.props.keys())[index]
|
||||
self.props[prop_name] = Property(prop_name, prop_value)
|
||||
return self
|
||||
|
||||
def get_prop(self, prop_name: str):
|
||||
return self.props[prop_name].value
|
||||
|
||||
|
||||
class Property:
|
||||
"""
|
||||
|
||||
+104
-50
@@ -5,11 +5,14 @@ from evaluators.BaseEvaluator import OneReturnValueEvaluator
|
||||
from parsers.BaseParser import BaseParser
|
||||
from sdp.sheerkaDataProvider import SheerkaDataProvider, Event, SheerkaDataProviderDuplicateKeyError
|
||||
import core.utils
|
||||
import core.builtin_helpers
|
||||
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
concept_evaluation_steps = [BuiltinConcepts.EVALUATION, BuiltinConcepts.AFTER_EVALUATION]
|
||||
|
||||
|
||||
class Sheerka(Concept):
|
||||
"""
|
||||
@@ -150,6 +153,12 @@ class Sheerka(Concept):
|
||||
logging.basicConfig(format=log_format, level=log_level)
|
||||
|
||||
def eval(self, text):
|
||||
"""
|
||||
Note to KSI: If you try to add execution context to this function,
|
||||
You may end in an infinite loop
|
||||
:param text:
|
||||
:return:
|
||||
"""
|
||||
evt_digest = self.sdp.save_event(Event(text))
|
||||
exec_context = ExecutionContext(self.key, evt_digest, self)
|
||||
|
||||
@@ -174,32 +183,6 @@ class Sheerka(Concept):
|
||||
|
||||
return return_values
|
||||
|
||||
def expect_one(self, context, items):
|
||||
|
||||
if not isinstance(items, list):
|
||||
items = [items]
|
||||
|
||||
if len(items) == 0:
|
||||
return self.ret(context.who, False, self.new(BuiltinConcepts.IS_EMPTY, obj=items))
|
||||
|
||||
successful_results = [item for item in items if item.status]
|
||||
number_of_successful = len(successful_results)
|
||||
total_items = len(items)
|
||||
|
||||
# remove errors when a winner is found
|
||||
if number_of_successful == 1:
|
||||
# log.debug(f"1 / {total_items} good item found.")
|
||||
return successful_results[0]
|
||||
|
||||
# too many winners, which one to choose ?
|
||||
if number_of_successful > 1:
|
||||
log.debug(f"{number_of_successful} / {total_items} good items. Too many success")
|
||||
return self.ret(context.who, False, self.new(BuiltinConcepts.TOO_MANY_SUCCESS, obj=successful_results))
|
||||
|
||||
# only errors, i cannot help you
|
||||
log.debug(f"{total_items} items. Only errors")
|
||||
return self.ret(context.who, False, self.new(BuiltinConcepts.TOO_MANY_ERRORS, obj=items))
|
||||
|
||||
def parse(self, context, text):
|
||||
result = []
|
||||
if log.isEnabledFor(logging.DEBUG):
|
||||
@@ -356,6 +339,33 @@ class Sheerka(Concept):
|
||||
for prop in concept.props:
|
||||
concept.codes[prop] = self.parse(context, concept.props[prop].value)
|
||||
|
||||
# updates the code of the reference when possible
|
||||
if concept.key in self.concepts_cache:
|
||||
entry = self.concepts_cache[concept.key]
|
||||
if isinstance(entry, list):
|
||||
# TODO : manage when there are multiple entries
|
||||
pass
|
||||
else:
|
||||
self.concepts_cache[concept.key].codes = concept.codes
|
||||
|
||||
def eval_concept(self, context, concept, properties_to_eval=None):
|
||||
if len(concept.codes) == 0:
|
||||
self.add_codes_to_concept(context, concept)
|
||||
|
||||
if properties_to_eval is None:
|
||||
properties_to_eval = ["where", "pre", "post", "body", "props"]
|
||||
|
||||
for prop in properties_to_eval:
|
||||
if prop == "props":
|
||||
pass
|
||||
else:
|
||||
part_key = ConceptParts(prop)
|
||||
if concept.codes[part_key] is None:
|
||||
continue
|
||||
res = self.chain_process(context, concept.codes[part_key], concept_evaluation_steps)
|
||||
res = core.builtin_helpers.expect_one(context, res)
|
||||
setattr(concept, prop, res.value)
|
||||
|
||||
def add_in_cache(self, concept):
|
||||
"""
|
||||
Adds a concept template in cache.
|
||||
@@ -413,16 +423,37 @@ class Sheerka(Concept):
|
||||
"""
|
||||
template = self.get(concept_key)
|
||||
|
||||
def new_from_template(t, k, **kwargs_):
|
||||
# manage singleton
|
||||
if t.is_unique:
|
||||
return t
|
||||
|
||||
# otherwise, create another instance
|
||||
concept = self.builtin_cache[k]() if k in self.builtin_cache else Concept()
|
||||
concept.update_from(t)
|
||||
|
||||
# update the properties
|
||||
for k, v in kwargs_.items():
|
||||
if k in concept.props:
|
||||
concept.set_prop(k, v)
|
||||
elif hasattr(concept, k):
|
||||
setattr(concept, k, v)
|
||||
else:
|
||||
return self.new(BuiltinConcepts.UNKNOWN_PROPERTY, body=k, concept=concept)
|
||||
|
||||
# TODO : add the concept to the list of known concepts (self.instances)
|
||||
return concept
|
||||
|
||||
# manage concept not found
|
||||
if self.isinstance(template, BuiltinConcepts.UNKNOWN_CONCEPT) and \
|
||||
concept_key != BuiltinConcepts.UNKNOWN_CONCEPT:
|
||||
return template
|
||||
|
||||
if not isinstance(template, list):
|
||||
return self._new_from_template(template, concept_key, **kwargs)
|
||||
return new_from_template(template, concept_key, **kwargs)
|
||||
|
||||
# if template is a list, it means that there a multiple concepts under the same key
|
||||
concepts = [self._new_from_template(t, concept_key, **kwargs) for t in template]
|
||||
concepts = [new_from_template(t, concept_key, **kwargs) for t in template]
|
||||
return self.new(BuiltinConcepts.ENUMERATION, body=concepts)
|
||||
|
||||
def ret(self, who, status, value, message=None, parents=None):
|
||||
@@ -443,6 +474,29 @@ class Sheerka(Concept):
|
||||
message=message,
|
||||
parents=parents)
|
||||
|
||||
def value(self, obj, allow_none_body=False):
|
||||
if obj is None:
|
||||
return None
|
||||
|
||||
if not isinstance(obj, Concept):
|
||||
return obj
|
||||
|
||||
if hasattr(obj, "get_value"):
|
||||
return obj.get_value()
|
||||
|
||||
if obj.body is not None:
|
||||
return obj.body
|
||||
|
||||
return obj if allow_none_body else self.new(BuiltinConcepts.CANNOT_RESOLVE_VALUE_ERROR, body=obj)
|
||||
|
||||
def values(self, objs):
|
||||
if not (isinstance(objs, list) or
|
||||
self.isinstance(objs, BuiltinConcepts.LIST) or
|
||||
self.isinstance(objs, BuiltinConcepts.ENUMERATION)):
|
||||
objs = [objs]
|
||||
|
||||
return (self.value(obj) for obj in objs)
|
||||
|
||||
def isinstance(self, a, b):
|
||||
"""
|
||||
return true if the concept a is an instance of the concept b
|
||||
@@ -463,6 +517,27 @@ class Sheerka(Concept):
|
||||
# for example, if a is a color, it will be found the entry 'All_Colors'
|
||||
return a.key == b_key
|
||||
|
||||
def isa(self, a, b):
|
||||
"""
|
||||
return true if the concept a is a b
|
||||
Will handle when the keyword isa will be implemented
|
||||
:param a:
|
||||
:param b:
|
||||
:return:
|
||||
"""
|
||||
|
||||
if isinstance(a, BuiltinConcepts): # common KSI error ;-)
|
||||
raise SyntaxError("Remember that the first parameter of isinstance MUST be a concept")
|
||||
|
||||
if not isinstance(a, Concept):
|
||||
return False
|
||||
|
||||
b_key = b.key if isinstance(b, Concept) else str(b)
|
||||
|
||||
# TODO : manage when a is the list of all possible b
|
||||
# for example, if a is a color, it will be found the entry 'All_Colors'
|
||||
return a.key == b_key
|
||||
|
||||
def get_evaluator_name(self, name):
|
||||
if self.evaluators_prefix is None:
|
||||
base_evaluator_class = core.utils.get_class("evaluators.BaseEvaluator.BaseEvaluator")
|
||||
@@ -486,28 +561,7 @@ class Sheerka(Concept):
|
||||
else:
|
||||
res.append(item)
|
||||
|
||||
return sorted(res, key=lambda i: i.key)
|
||||
|
||||
def _new_from_template(self, template, concept_key, **kwargs):
|
||||
# manage singleton
|
||||
if template.is_unique:
|
||||
return template
|
||||
|
||||
# otherwise, create another instance
|
||||
concept = self.builtin_cache[concept_key]() if concept_key in self.builtin_cache else Concept()
|
||||
concept.update_from(template)
|
||||
|
||||
# update the properties
|
||||
for k, v in kwargs.items():
|
||||
if k in concept.props:
|
||||
concept.set_prop(k, v)
|
||||
elif hasattr(concept, k):
|
||||
setattr(concept, k, v)
|
||||
else:
|
||||
return self.new(BuiltinConcepts.UNKNOWN_PROPERTY, body=k, concept=concept)
|
||||
|
||||
# TODO : add the concept to the list of known concepts (self.instances)
|
||||
return concept
|
||||
return sorted(res, key=lambda i: int(i.id))
|
||||
|
||||
@staticmethod
|
||||
def get_builtins_classes_as_dict():
|
||||
|
||||
Reference in New Issue
Block a user