from core.ast.nodes import python_to_concept from core.builtin_concepts import BuiltinConcepts, ErrorConcept from core.concept import Concept, ConceptParts import core.builtin_helpers GROUP_PREFIX = 'All_' class SheerkaSetsManager: def __init__(self, sheerka): self.sheerka = sheerka self.logger_name = self.add_concept_to_set.__name__ def set_isa(self, context, concept, concept_set): """ Defines that concept a is b is another concept :param context: :param concept: :param concept_set: :return: """ context.log(f"Setting that concept {concept} is a {concept_set}", who=self.logger_name) isa = [] if BuiltinConcepts.ISA not in concept.props else concept.get_prop(BuiltinConcepts.ISA) if concept_set not in isa: isa.append(concept_set) concept.set_prop(BuiltinConcepts.ISA, isa) res = self.sheerka.modify_concept(context, concept) if not res.status: return res return self.add_concept_to_set(context, concept, concept_set) def add_concept_to_set(self, context, concept, concept_set): """ Add an entry in sdp to tell that concept isa concept_set :param context: :param concept: :param concept_set: :return: """ context.log(f"Adding concept {concept} to set {concept_set}", who=self.logger_name) assert concept.id assert concept_set.id try: result = self.sheerka.sdp.add_unique(context.event.get_digest(), GROUP_PREFIX + concept_set.id, concept.id) if result.already_exists: # concept already in set return self.sheerka.ret( self.logger_name, False, self.sheerka.new(BuiltinConcepts.CONCEPT_ALREADY_IN_SET, body=concept, concept_set=concept_set)) else: return self.sheerka.ret(self.logger_name, True, self.sheerka.new(BuiltinConcepts.SUCCESS)) except Exception as error: context.log_error("Failed to add to set.", who=self.logger_name) return self.sheerka.ret(self.logger_name, False, ErrorConcept(error), error.args[0]) def add_concepts_to_set(self, context, concepts, concept_set): """Adding multiple concepts at the same time""" context.log(f"Adding concepts {concepts} to set {concept_set}", who=self.logger_name) previous = self.sheerka.sdp.get_safe(GROUP_PREFIX + concept_set.id) new_ids = [c.id for c in concepts] if previous is None else previous + [c.id for c in concepts] try: self.sheerka.sdp.set(context.event.get_digest(), GROUP_PREFIX + concept_set.id, new_ids) return self.sheerka.ret(self.logger_name, True, self.sheerka.new(BuiltinConcepts.SUCCESS)) except Exception as error: context.log_error("Failed to add to set.", who=self.logger_name) return self.sheerka.ret(self.logger_name, False, ErrorConcept(error), error.args[0]) def get_set_elements(self, context, concept): """ Concept is supposed to be a set Returns all elements if the set :param context: :param concept: :return: """ # noinspection PyShadowingNames def _get_set_elements(context, concept, sub_concept): if not (isinstance(sub_concept, Concept) and sub_concept.id): return self.sheerka.new(BuiltinConcepts.NOT_A_SET, body=concept) ids = self.sheerka.sdp.get_safe(GROUP_PREFIX + sub_concept.id) if ids: if concept.metadata.where: new_condition = self._validate_where_clause(concept) if not new_condition: return self.sheerka.new(BuiltinConcepts.WHERE_CLAUSE_FAILED, body=concept) else: # This methods sucks, but I don't have enough tools (like proper AST manipulation functions) # to do it properly now. It will be enhanced later concepts = self._get_concepts(context, ids, True) globals_ = {"xx__concepts__xx": concepts, "sheerka": self.sheerka} locals_ = {} exec(new_condition, globals_, locals_) return locals_["result"] else: return self._get_concepts(context, ids, False) # it may be a concept that references a set if not sub_concept.metadata.is_evaluated: with context.push(desc=f"Evaluating concept {sub_concept}") as sub_context: evaluated = self.sheerka.evaluate_concept(sub_context, sub_concept) if evaluated.key != concept.key: return False return _get_set_elements(context, concept, sub_concept.body) return _get_set_elements(context, concept, concept) def isinset(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) and isinstance(b, Concept)): return False # TODO, first check the 'isa' property of a if not (a.id and b.id): return False if self.sheerka.sdp.exists(GROUP_PREFIX + b.id, a.id): return True return False def isa(self, a, b): if BuiltinConcepts.ISA not in a.props: return False for c in a.get_prop(BuiltinConcepts.ISA): if c == b: return True if self.isa(c, b): return True return False def isaset(self, context, concept): """ True if exists All_ in sdp or if concept references to a concept that has all_ :param context: :param concept: :return: """ """""" if not (isinstance(concept, Concept) and concept.id): return None # it may be a concept that references a set if not concept.metadata.is_evaluated: with context.push(desc=f"Evaluating concept {concept}") as sub_context: evaluated = self.sheerka.evaluate_concept(sub_context, concept) if evaluated.key != concept.key: return False if concept.body: return self.isaset(context, concept.body) res = self.sheerka.sdp.get_safe(GROUP_PREFIX + concept.id) return res is not None def _validate_where_clause(self, concept): python_parser_result = [r for r in concept.compiled[ConceptParts.WHERE] if r.who == "parsers.Python"] if not python_parser_result or not python_parser_result[0].status: return None ast_ = python_parser_result[0].body.body.ast_ ast_as_concepts = python_to_concept(ast_) names = core.builtin_helpers.get_names(self.sheerka, ast_as_concepts) if len(names) != 1 or names[0] != concept.metadata.body: return None condition = concept.metadata.where.replace(concept.metadata.body, "sheerka.value(x)") expression = f""" result=[] for x in xx__concepts__xx: try: if {condition}: result.append(x) except Exception: pass """ return expression def _get_concepts(self, context, ids, evaluate): """ Gets concepts from a list of concepts ids :param ids: :param evaluate: if True, all the elements are evaluated before returned :return: """ if not ids: return [] if not evaluate: return [self.sheerka.get_by_id(element_id) for element_id in ids] result = [] with context.push(desc=f"Evaluating concepts of a set") as sub_context: for element_id in ids: concept = self.sheerka.get_by_id(element_id) evaluated = self.sheerka.evaluate_concept(context, concept) result.append(evaluated) return result