kodjo
89e1f20975
Fixed #130 : ArithmeticOperatorParser Fixed #129 : python_wrapper : create_namespace Fixed #128 : ExpressionParser: Cannot parse func(x) infixed concept 'xxx'
324 lines
12 KiB
Python
324 lines
12 KiB
Python
import ast
|
|
from dataclasses import dataclass, field
|
|
|
|
from core.builtin_concepts import BuiltinConcepts, ParserResultConcept, ReturnValueConcept
|
|
from core.builtin_helpers import evaluate_from_source, expect_one
|
|
from core.concept import Concept, DEFINITION_TYPE_BNF, DEFINITION_TYPE_DEF, freeze_concept_attrs
|
|
from core.rule import ACTION_TYPE_EXEC, ACTION_TYPE_PRINT, Rule
|
|
from core.sheerka.ExecutionContext import ExecutionContext
|
|
from core.sheerka.Sheerka import Sheerka
|
|
from core.sheerka.services.SheerkaConceptManager import SheerkaConceptManager
|
|
from core.sheerka.services.SheerkaDebugManager import ListDebugLogger
|
|
from core.sheerka.services.SheerkaRuleManager import SheerkaRuleManager
|
|
from parsers.BnfDefinitionParser import BnfDefinitionParser
|
|
from parsers.BnfNodeParser import StrMatch
|
|
from sdp.sheerkaDataProvider import Event
|
|
|
|
|
|
@dataclass
|
|
class InitTestHelper:
|
|
sheerka: Sheerka
|
|
context: ExecutionContext
|
|
items: list = field(default_factory=list)
|
|
|
|
def push(self, *items):
|
|
self.items.extend(items)
|
|
|
|
def unpack(self):
|
|
return self.sheerka, self.context, *self.items
|
|
|
|
def with_concepts(self, *concepts, **kwargs):
|
|
create_new = kwargs.get("create_new", False)
|
|
|
|
for c in concepts:
|
|
if isinstance(c, str):
|
|
c = Concept(c)
|
|
|
|
if c.get_metadata().definition and c.get_metadata().definition_type != DEFINITION_TYPE_DEF:
|
|
desc = f"Resolving BNF {c.get_metadata().definition}"
|
|
with self.context.push(BuiltinConcepts.INIT_BNF,
|
|
c,
|
|
obj=c,
|
|
desc=desc) as sub_context:
|
|
|
|
bnf_parser = BnfDefinitionParser()
|
|
res = bnf_parser.parse(sub_context, c.get_metadata().definition)
|
|
if res.status:
|
|
c.set_bnf(res.value.value)
|
|
c.get_metadata().definition_type = DEFINITION_TYPE_BNF
|
|
else:
|
|
raise Exception(f"Error in bnf definition '{c.get_metadata().definition}'",
|
|
self.sheerka.get_errors(self.context, res))
|
|
|
|
self._update_concept_parameters(c)
|
|
if create_new:
|
|
self.sheerka.create_new_concept(self.context, c)
|
|
else:
|
|
c.init_key()
|
|
self.sheerka.set_id_if_needed(c, False)
|
|
self.sheerka.test_only_add_in_cache(c)
|
|
freeze_concept_attrs(c)
|
|
|
|
self.items.append(c)
|
|
|
|
return self
|
|
|
|
def with_format_rules(self, *rules, **kwargs):
|
|
return self.with_rules(ACTION_TYPE_PRINT, *rules, **kwargs)
|
|
|
|
def with_exec_rules(self, *rules, **kwargs):
|
|
return self.with_rules(ACTION_TYPE_EXEC, *rules, **kwargs)
|
|
|
|
def with_rules(self, action_type, *rules, **kwargs):
|
|
create_new = kwargs.get("create_new", True)
|
|
compile_rule = kwargs.get("compile_rule", True)
|
|
|
|
for rule_template in rules:
|
|
if isinstance(rule_template, tuple):
|
|
if len(rule_template) == 3:
|
|
rule = Rule(action_type, rule_template[0], rule_template[1], rule_template[2])
|
|
else:
|
|
rule = Rule(action_type, None, rule_template[0], rule_template[1])
|
|
else:
|
|
rule = rule_template
|
|
|
|
is_enabled = rule.metadata.is_enabled # remember the value...
|
|
|
|
if compile_rule:
|
|
self.sheerka.services[SheerkaRuleManager.NAME].init_rule(self.context, rule)
|
|
else:
|
|
rule.metadata.is_compiled = True
|
|
|
|
if create_new:
|
|
res = self.sheerka.create_new_rule(self.context, rule)
|
|
if not res.status:
|
|
raise Exception(f"Error in rule definition '{res.body}'",
|
|
self.sheerka.get_errors(res))
|
|
self.items.append(res.body.body)
|
|
else:
|
|
self.items.append(rule)
|
|
|
|
if is_enabled is not None: # ...and back the value if it was not None
|
|
rule.metadata.is_enabled = is_enabled
|
|
|
|
return self
|
|
|
|
def _update_concept_parameters(self, concept):
|
|
if concept.get_metadata().parameters:
|
|
return
|
|
|
|
SheerkaConceptManager.recompute_concept_parameters(self.context, concept)
|
|
|
|
|
|
class BaseTest:
|
|
def get_sheerka(self, **kwargs) -> Sheerka:
|
|
pass
|
|
|
|
def get_context(self, sheerka=None, eval_body=False, eval_where=False, global_truth=False, message=""):
|
|
context = ExecutionContext("test",
|
|
Event(message=message),
|
|
sheerka or self.get_sheerka(),
|
|
BuiltinConcepts.TESTING,
|
|
None)
|
|
if eval_body:
|
|
context.protected_hints.add(BuiltinConcepts.EVAL_BODY_REQUESTED)
|
|
if eval_where:
|
|
context.protected_hints.add(BuiltinConcepts.EVAL_WHERE_REQUESTED)
|
|
if global_truth:
|
|
context.protected_hints.add(BuiltinConcepts.EVAL_GLOBAL_TRUTH_REQUESTED)
|
|
|
|
return context
|
|
|
|
@staticmethod
|
|
def get_init_test_args(**kwargs):
|
|
return {k: v for k, v in kwargs.items() if k in ["cache_only",
|
|
"ontology",
|
|
"eval_body",
|
|
"eval_where",
|
|
"global_truth"]}
|
|
|
|
@staticmethod
|
|
def get_with_concepts_args(**kwargs):
|
|
return {k: v for k, v in kwargs.items() if k in ["create_new"]}
|
|
|
|
def init_test(self, cache_only=None, ontology=None, eval_body=False, eval_where=False, global_truth=False):
|
|
sheerka = self.get_sheerka(cache_only=cache_only, ontology=ontology)
|
|
context = self.get_context(sheerka=sheerka,
|
|
eval_body=eval_body,
|
|
eval_where=eval_where,
|
|
global_truth=global_truth)
|
|
|
|
return InitTestHelper(sheerka, context)
|
|
|
|
def get_default_concept(self):
|
|
concept = Concept(
|
|
name="a + b",
|
|
where="isinstance(a, int) and isinstance(b, int)\n",
|
|
pre="isinstance(a, int) and isinstance(b, int)\n",
|
|
post="isinstance(res, int)\n",
|
|
body="def func(x,y):\n return x+y\nfunc(a,b)",
|
|
desc="specific description")
|
|
concept.def_var("a", "value1")
|
|
concept.def_var("b", "value2")
|
|
|
|
return concept
|
|
|
|
def dump_ast(self, node):
|
|
dump = ast.dump(node)
|
|
for to_remove in [", ctx=Load()", ", kind=None", ", type_ignores=[]"]:
|
|
dump = dump.replace(to_remove, "")
|
|
return dump
|
|
|
|
@staticmethod
|
|
def dump_tokens(tokens):
|
|
return [t.repr_value for t in tokens]
|
|
|
|
def init_concepts(self, *concepts, **kwargs):
|
|
init_test_args = self.get_init_test_args(**kwargs)
|
|
with_concepts_args = self.get_with_concepts_args(**kwargs)
|
|
return self.init_test(**init_test_args).with_concepts(*concepts, **with_concepts_args).unpack()
|
|
|
|
def init_format_rules(self, *rules, **kwargs):
|
|
return self.init_test(**kwargs).with_format_rules(*rules, **kwargs).unpack()
|
|
|
|
def init_exec_rules(self, *rules, **kwargs):
|
|
return self.init_test(**kwargs).with_exec_rules(*rules, **kwargs).unpack()
|
|
|
|
@staticmethod
|
|
def get_concept_instance(sheerka, concept, **kwargs):
|
|
"""
|
|
Use to instantiate concept with default variables already set
|
|
:param sheerka:
|
|
:param concept:
|
|
:param kwargs:
|
|
:return:
|
|
"""
|
|
instance = sheerka.new(concept.key if isinstance(concept, Concept) else concept)
|
|
for i, var in enumerate(instance.get_metadata().variables):
|
|
if var[0] in kwargs:
|
|
assert isinstance(kwargs[var[0]], str), "variables definitions must be string"
|
|
instance.get_metadata().variables[i] = (var[0], kwargs[var[0]])
|
|
|
|
return instance
|
|
|
|
@staticmethod
|
|
def retval(obj, who="who", status=True):
|
|
"""ret_val"""
|
|
return ReturnValueConcept(who, status, obj)
|
|
|
|
@staticmethod
|
|
def tretval(sheerka, obj, who="who"):
|
|
"""True ret_val + add concept in cache"""
|
|
if isinstance(obj, Concept):
|
|
obj.init_key()
|
|
return sheerka.ret(who, True, obj)
|
|
|
|
@staticmethod
|
|
def pretval(concept, source=None, parser="parsers.name", who=None, status=True):
|
|
"""ParserResult ret_val (p stands for ParserResult)"""
|
|
return ReturnValueConcept(
|
|
who or parser,
|
|
status,
|
|
ParserResultConcept(parser=parser,
|
|
source=source or concept.name,
|
|
value=concept,
|
|
try_parsed=concept))
|
|
|
|
@staticmethod
|
|
def create_and_add_in_cache_concept(sheerka, name, variables=None, bnf=None):
|
|
"""
|
|
Create a concept using parameters and add it in cache
|
|
:param sheerka:
|
|
:param name:
|
|
:param variables:
|
|
:param bnf:
|
|
:return:
|
|
"""
|
|
concept = Concept(name) if isinstance(name, str) else name
|
|
if variables:
|
|
for v in variables:
|
|
concept.def_var(v)
|
|
if bnf:
|
|
concept.set_bnf(bnf)
|
|
concept.get_metadata().definition_type = DEFINITION_TYPE_BNF
|
|
concept.init_key()
|
|
sheerka.set_id_if_needed(concept, False)
|
|
sheerka.test_only_add_in_cache(concept)
|
|
freeze_concept_attrs(concept)
|
|
return concept
|
|
|
|
@staticmethod
|
|
def bnf_concept(concept, expression=None):
|
|
if isinstance(concept, Concept):
|
|
name = concept.name
|
|
else:
|
|
name = concept
|
|
concept = Concept(concept)
|
|
|
|
concept.set_bnf(expression or StrMatch(name))
|
|
concept.get_metadata().definition_type = DEFINITION_TYPE_BNF
|
|
return concept
|
|
|
|
@staticmethod
|
|
def from_def_concept(name, definition, variables=None, **kwargs):
|
|
concept = Concept(name=name, definition=definition, definition_type=DEFINITION_TYPE_DEF)
|
|
if variables:
|
|
for v in variables:
|
|
concept.def_var(v)
|
|
|
|
if kwargs:
|
|
for k, v in kwargs.items():
|
|
if k in ("body", "pre", "post", "where"):
|
|
setattr(concept.get_metadata(), k, v)
|
|
else:
|
|
concept.get_metadata().variables[k] = v
|
|
return concept
|
|
|
|
def init_scenario(self, init_expressions, global_truth=False):
|
|
sheerka = self.get_sheerka()
|
|
if global_truth:
|
|
sheerka.add_to_context(BuiltinConcepts.EVAL_GLOBAL_TRUTH_REQUESTED)
|
|
|
|
for expression in init_expressions:
|
|
res = sheerka.evaluate_user_input(expression)
|
|
assert len(res) == 1, f"Failed to execute '{expression}'"
|
|
assert res[0].status, f"Error while executing '{expression}'"
|
|
|
|
if global_truth:
|
|
sheerka.remove_fom_context(BuiltinConcepts.EVAL_GLOBAL_TRUTH_REQUESTED)
|
|
|
|
return sheerka
|
|
|
|
@staticmethod
|
|
def successful_return_values(return_values):
|
|
return [ret_val for ret_val in return_values if ret_val.status]
|
|
|
|
@staticmethod
|
|
def activate_debug(context, pattern="Sya.*.*"):
|
|
sheerka = context.sheerka
|
|
sheerka.set_debug(context, True)
|
|
sheerka.set_debug_logger_definition(ListDebugLogger)
|
|
|
|
if isinstance(pattern, list):
|
|
for p in pattern:
|
|
sheerka.set_debug_var(context, p)
|
|
else:
|
|
sheerka.set_debug_var(context, pattern)
|
|
|
|
# the see the logs, do not forget to add
|
|
# logs = sheerka.get_debugger_logs()
|
|
|
|
@staticmethod
|
|
def evaluate_from_source(context, source, is_question=False):
|
|
res = evaluate_from_source(context,
|
|
source,
|
|
desc=None,
|
|
eval_body=not is_question,
|
|
eval_where=False,
|
|
is_question=is_question,
|
|
expect_success=False,
|
|
stm=None)
|
|
res = expect_one(context, res)
|
|
assert res.status
|
|
return res.body
|