from dataclasses import dataclass from core.concept import Concept, ErrorConcept from parsers.PythonParser import PythonParser from sdp.sheerkaDataProvider import SheerkaDataProvider, Event from parsers.DefaultParser import DefaultParser, DefConceptNode class Singleton(type): _instances = {} def __call__(cls, *args, **kwargs): if cls not in cls._instances: cls._instances[cls] = super(Singleton, cls).__call__(*args, **kwargs) return cls._instances[cls] @dataclass class ReturnValue: """ Class that handle the return of a concept To avoid using the try/except pattern for each and every call To give context (ie return message) even when the call is successful """ status: bool value: Concept message: str = None class Sheerka(Concept, metaclass=Singleton): """ Main controller for the project """ NAME = "Sheerka" UNKNOWN_CONCEPT_NAME = "Unknown Concept" ERROR_CONCEPT_NAME = "Error" SUCCESS_CONCEPT_NAME = "Success" def __init__(self): super().__init__(Sheerka.NAME) # list of all concepts known be the system self.concepts = [] # a concept can be instantiated # ex: File is a concept, but File('foo.txt') is an instance # TODO: manage contexts self.instances = [] # List of the known rules by the system # ex: hello => say('hello') self.rules = [] self.create_builtin_concepts() self.sdp = None self.parsers = [] def create_builtin_concepts(self): """ Initializes the builtin concepts :return: None """ self.concepts.append(self) self.concepts.append(Concept(Sheerka.UNKNOWN_CONCEPT_NAME)) self.concepts.append(Concept(Sheerka.SUCCESS_CONCEPT_NAME)) self.concepts.append(Concept(Sheerka.ERROR_CONCEPT_NAME)) def initialize(self, root_folder=None): """ Starting Sheerka Loads the current configuration Notes that when it's the first time, it also create the needed working folders :param root_folder: root configuration folder :return: ReturnValue(Success or Error) """ try: self.sdp = SheerkaDataProvider(root_folder) self.parsers.append(lambda text: DefaultParser(text, PythonParser)) except IOError as e: return ReturnValue(False, self.get_concept(Sheerka.ERROR_CONCEPT_NAME, True), e) return ReturnValue(True, self.get_concept(Sheerka.SUCCESS_CONCEPT_NAME, True)) def eval(self, text): #evt_digest = self.sdp.save_event(Event(text)) result = self.try_parse(text) return_values = [] for parser_name, status, node in result: if not status: return_values.append(ReturnValue(False, ErrorConcept(body=node))) elif status and isinstance(node, DefConceptNode): return_values.append(self.add_concept(node)) return return_values def try_parse(self, text): result = [] for parser in self.parsers: p = parser(text) # try: # tree = p.parse() # result.append((p.name, tree)) # except Exception as e: # result.append((p.name, e)) tree = p.parse() result.append((p.name, not p.has_error, p.error_sink if p.has_error else tree)) return result def get_concept(self, name, is_builtin=False): """ Given a concept name, tries to find it :param name: name of the concept to look for :param is_builtin: is it a builtin concept ? :return: concept if found, UNKNOWN_CONCEPT otherwise """ for concept in self.concepts: if concept.name == name and concept.is_builtin == is_builtin: return concept return self.concepts[1] def add_concept(self, def_concept_node: DefConceptNode): """ Adds a new concept to the system :param def_concept_node: DefConceptNode :return: digest of the new concept """ concept = Concept(def_concept_node.name) for prop in ("where", "pre", "post", "body"): concept_part_node = getattr(def_concept_node, prop) value = concept_part_node.source if hasattr(concept_part_node, "source") else "" setattr(concept, prop, value) concept.add_codes(def_concept_node.get_codes()) return ReturnValue(True, concept) @staticmethod def concept_equals(concept1, concept2): """True if the two concepts refer to the same concept""" if concept1 is None and concept2 is None: return True if concept1 is None or concept2 is None: return False return concept1.key == concept2.key