Implemented SheerkaOntology
This commit is contained in:
+107
-73
@@ -1,17 +1,101 @@
|
||||
import ast
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from core.builtin_concepts import ReturnValueConcept, ParserResultConcept, BuiltinConcepts
|
||||
from core.concept import Concept, DEFINITION_TYPE_BNF, DEFINITION_TYPE_DEF
|
||||
from core.rule import Rule
|
||||
from core.concept import Concept, DEFINITION_TYPE_BNF, DEFINITION_TYPE_DEF, freeze_concept_attrs
|
||||
from core.rule import Rule, ACTION_TYPE_PRINT
|
||||
from core.sheerka.ExecutionContext import ExecutionContext
|
||||
from core.sheerka.Sheerka import Sheerka
|
||||
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_error(res))
|
||||
|
||||
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_rules(self, *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_PRINT, rule_template[0], rule_template[1], rule_template[2])
|
||||
else:
|
||||
rule = Rule(ACTION_TYPE_PRINT, 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_error(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
|
||||
|
||||
|
||||
class BaseTest:
|
||||
def get_sheerka(self, **kwargs):
|
||||
def get_sheerka(self, **kwargs) -> Sheerka:
|
||||
pass
|
||||
|
||||
def get_context(self, sheerka=None, eval_body=False, eval_where=False):
|
||||
@@ -23,6 +107,20 @@ class BaseTest:
|
||||
|
||||
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"]}
|
||||
|
||||
@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):
|
||||
sheerka = self.get_sheerka(cache_only=cache_only, ontology=ontology)
|
||||
context = self.get_context(sheerka=sheerka, eval_body=eval_body, eval_where=eval_where)
|
||||
|
||||
return InitTestHelper(sheerka, context)
|
||||
|
||||
def get_default_concept(self):
|
||||
concept = Concept(
|
||||
name="a + b",
|
||||
@@ -47,77 +145,12 @@ class BaseTest:
|
||||
return [t.repr_value for t in tokens]
|
||||
|
||||
def init_concepts(self, *concepts, **kwargs):
|
||||
sheerka = self.get_sheerka(**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()
|
||||
|
||||
context_args = dict([(k, v) for k, v in kwargs.items() if k in ["sheerka", "eval_body", "eval_where"]])
|
||||
context = self.get_context(sheerka, **context_args)
|
||||
create_new = kwargs.get("create_new", None)
|
||||
|
||||
result = []
|
||||
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 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}'",
|
||||
sheerka.get_error(res))
|
||||
|
||||
if create_new:
|
||||
sheerka.create_new_concept(context, c)
|
||||
else:
|
||||
c.init_key()
|
||||
sheerka.set_id_if_needed(c, False)
|
||||
sheerka.test_only_add_in_cache(c)
|
||||
|
||||
result.append(c)
|
||||
|
||||
return sheerka, context, *result
|
||||
|
||||
def init_format_rules(self, *rules, create_new=True, compile_rule=True, concepts=None, **kwargs):
|
||||
if concepts:
|
||||
sheerka, context, *concepts = self.init_concepts(*concepts, **kwargs)
|
||||
else:
|
||||
sheerka, context, *concepts = self.init_concepts(**kwargs)
|
||||
|
||||
if create_new:
|
||||
sheerka.cache_manager.caches[SheerkaRuleManager.FORMAT_RULE_ENTRY].cache.clear()
|
||||
sheerka.cache_manager.delete(sheerka.OBJECTS_IDS_ENTRY, SheerkaRuleManager.RULE_IDS)
|
||||
with sheerka.sdp.get_transaction(context.event.get_digest()) as transaction:
|
||||
transaction.clear(SheerkaRuleManager.FORMAT_RULE_ENTRY)
|
||||
|
||||
initialized = []
|
||||
for rule_blue_print in rules:
|
||||
if isinstance(rule_blue_print, tuple):
|
||||
rule = Rule("print", None, rule_blue_print[0], rule_blue_print[1])
|
||||
if not compile_rule:
|
||||
rule.metadata.is_compiled = True
|
||||
else:
|
||||
rule = rule_blue_print
|
||||
|
||||
is_enabled = rule.metadata.is_enabled
|
||||
sheerka.services[SheerkaRuleManager.NAME].init_rule(context, rule)
|
||||
if create_new:
|
||||
res = sheerka.create_new_rule(context, rule)
|
||||
initialized.append(res.body.body)
|
||||
else:
|
||||
initialized.append(rule)
|
||||
|
||||
if is_enabled is not None:
|
||||
rule.metadata.is_enabled = is_enabled
|
||||
|
||||
return sheerka, context, *initialized
|
||||
def init_format_rules(self, *rules, **kwargs):
|
||||
return self.init_test(**kwargs).with_rules(*rules, **kwargs).unpack()
|
||||
|
||||
@staticmethod
|
||||
def get_concept_instance(sheerka, concept, **kwargs):
|
||||
@@ -179,6 +212,7 @@ class BaseTest:
|
||||
concept.init_key()
|
||||
sheerka.set_id_if_needed(concept, False)
|
||||
sheerka.test_only_add_in_cache(concept)
|
||||
freeze_concept_attrs(concept)
|
||||
return concept
|
||||
|
||||
@staticmethod
|
||||
|
||||
Reference in New Issue
Block a user