Sheerka-Old/src/core/sheerka/services/SheerkaAdmin.py

import sys
import time
from os import path

from core.builtin_concepts_ids import BuiltinConcepts, BuiltinContainers
from core.builtin_helpers import ensure_concept_or_rule
from core.concept import Concept
from core.sheerka.services.SheerkaHistoryManager import SheerkaHistoryManager
from core.sheerka.services.SheerkaMemory import SheerkaMemory
from core.sheerka.services.sheerka_service import BaseService

CONCEPTS_FILE_LITE = "_concepts_lite.txt"
CONCEPTS_FILE_FULL = "_concepts_full.txt"
CONCEPTS_FILE_TO_USE = CONCEPTS_FILE_FULL


class SheerkaAdmin(BaseService):
    NAME = "Admin"

    def __init__(self, sheerka):
        super().__init__(sheerka)

    def initialize(self):
        self.sheerka.bind_service_method(self.caches_names, False)
        self.sheerka.bind_service_method(self.cache, False)
        self.sheerka.bind_service_method(self.restore, True)
        self.sheerka.bind_service_method(self.concepts, False)
        self.sheerka.bind_service_method(self.desc, False)
        self.sheerka.bind_service_method(self.extended_isinstance, False)
        self.sheerka.bind_service_method(self.is_container, False)
        self.sheerka.bind_service_method(self.format_rules, False)
        self.sheerka.bind_service_method(self.exec_rules, False)
        self.sheerka.bind_service_method(self.admin_push_ontology, True, as_name="push_ontology")
        self.sheerka.bind_service_method(self.admin_pop_ontology, True, as_name="pop_ontology")
        self.sheerka.bind_service_method(self.ontologies, False)
        self.sheerka.bind_service_method(self.in_memory, False)
        self.sheerka.bind_service_method(self.admin_history, False, as_name="history")
        self.sheerka.bind_service_method(self.admin_history, False, as_name="history")
        self.sheerka.bind_service_method(self.sdp, False)

    def caches_names(self):
        """
        Returns the name of all the caches
        :return:
        """
        return self.sheerka.new(BuiltinConcepts.TO_LIST, body=self.sheerka.om.current_cache_manager().caches.keys())

    def cache(self, name, *keys):
        """
        Returns the content of a cache
        :param name:
        :param keys: look for a specific key. May ask to sdp if the key is not in cache
        :return:
        """
        if name not in self.sheerka.om.current_cache_manager().caches:
            return self.sheerka.new(BuiltinConcepts.NOT_FOUND, body={"cache": name})

        if not keys:
            return self.sheerka.om.current_cache_manager().caches[name].cache.copy()

        return {key: self.sheerka.om.get(name, key) for key in keys}

    def sdp(self, name=None):
        if name:
            for ontology in self.sheerka.om.ontologies:
                if ontology.name == name:
                    return ontology.cache_manager.sdp

            return self.sheerka.err(self.sheerka.new(BuiltinConcepts.NOT_FOUND, {"sdp_name", name}))

        return self.sheerka.om.current_sdp()

    def restore(self, concept_file=CONCEPTS_FILE_TO_USE):
        """
        Restore the state with all previous valid concept definitions
        :return:
        """

        def restore_from_file(file_name):
            _nb_lines, _nb_instructions, _nb_lines_in_error = 0, 0, 0
            _min_time, _max_time = None, None
            file_path = path.join(path.dirname(sys.argv[0]), file_name)
            if not path.exists(file_path):
                print(f"\u001b[31mFile '{file_path}' is not found !\u001b[0m")
                return 0, 0, 1

            with open(file_path, "r") as f:
                for line in f.readlines():
                    _nb_lines += 1
                    line = line.strip()

                    if line.startswith("#import "):
                        to_import = "_concepts_" + line[8:] + ".txt"
                        print(f" ==== Importing {to_import} ==== ")
                        res = restore_from_file(to_import)
                        _nb_lines += res[0]
                        _nb_instructions += res[1]
                        _nb_lines_in_error += res[2]
                        if _min_time is None or res[3] < _min_time:
                            _min_time = res[3]
                        if _max_time is None or res[4] > _max_time:
                            _max_time = res[4]
                        continue

                    if line == "" or line.startswith("#"):
                        continue

                    print(line)
                    _nb_instructions += 1
                    stop_watch = time.time_ns()
                    res = self.sheerka.evaluate_user_input(line)
                    user_input_elapsed = time.time_ns() - stop_watch

                    if _min_time is None or user_input_elapsed < _min_time:
                        _min_time = user_input_elapsed
                    if _max_time is None or user_input_elapsed > _max_time:
                        _max_time = user_input_elapsed

                    if len(res) > 1 or not res[0].status:
                        _nb_lines_in_error += 1
                        print("\u001b[31mError detected !\u001b[0m")

            return _nb_lines, _nb_instructions, _nb_lines_in_error, _min_time, _max_time

        if not concept_file.startswith("_concepts"):
            concept_file = f"_concepts_{concept_file}.txt"

        try:
            start = time.time_ns()
            self.sheerka.during_restore = True
            self.sheerka.save_execution_context = False
            enable_process_return_values_previous_value = self.sheerka.enable_process_return_values
            self.sheerka.enable_process_return_values = False

            nb_lines, nb_instructions, nb_lines_in_error, min_time, max_time = restore_from_file(concept_file)

            self.sheerka.enable_process_return_values = enable_process_return_values_previous_value
            self.sheerka.save_execution_context = True
            self.sheerka.during_restore = False
            stop = time.time_ns()

            nano_sec = stop - start
            dt = nano_sec / 1e6
            elapsed = f"{dt:.3f} ms" if dt < 1000 else f"{dt / 1000:.3f} s"
            min_str = f"{min_time / 1e6:.3f} ms"
            max_str = f"{max_time / 1e6:.3f} ms"
            mean_time = dt / nb_lines
            mean_str = f"{mean_time:.3f} ms" if mean_time < 1000 else f"{mean_time / 1000:.3f} s"

            print(f"Imported {nb_lines} line(s) in {elapsed}. min={min_str}, max={max_str}, mean={mean_str}")
            print(f"{nb_instructions} instruction(s).")
            if nb_lines_in_error > 0:
                print(f"\u001b[31m{nb_lines_in_error} errors(s) found.\u001b[0m")
            else:
                print(f"No error.")
        except IOError as e:
            raise e

    def concepts(self):
        concepts = sorted(self.sheerka.om.list(self.sheerka.CONCEPTS_BY_ID_ENTRY), key=lambda item: int(item.id))
        return self.sheerka.new(BuiltinConcepts.TO_LIST, body=concepts)

    def desc(self, *items):
        ensure_concept_or_rule(*items)
        res = []
        for item in items:
            if isinstance(item, Concept):
                bag = {
                    "id": item.id,
                    "name": item.name,
                    "key": item.key,
                    "definition": item.get_metadata().definition,
                    "type": item.get_metadata().definition_type,
                    "hash": item.get_definition_hash(),
                    "body": item.get_metadata().body,
                    "where": item.get_metadata().where,
                    "pre": item.get_metadata().pre,
                    "post": item.get_metadata().post,
                    "ret": item.get_metadata().ret,
                    "vars": item.get_metadata().variables,
                    "props": item.get_metadata().props,
                }
            else:
                bag = {
                    "id": item.id,
                    "name": item.metadata.name,
                    "type": item.metadata.action_type,
                    "predicate": item.metadata.predicate,
                    "action": item.metadata.action,
                    "priority": item.priority,
                    "compiled": item.metadata.is_compiled,
                    "enabled": item.metadata.is_enabled,
                }
            res.append(self.sheerka.new(BuiltinConcepts.TO_DICT, body=bag))

        return res[0] if len(res) == 1 else self.sheerka.new(BuiltinConcepts.TO_LIST, body=res)

    def format_rules(self):
        return self.sheerka.new(BuiltinConcepts.TO_LIST, items=self.sheerka.get_format_rules())

    def exec_rules(self):
        return self.sheerka.new(BuiltinConcepts.TO_LIST, items=self.sheerka.get_exec_rules())

    def extended_isinstance(self, a, b):
        """
        switch between sheerka.isinstance and builtin.isinstance
        :param a:
        :param b:
        :return:
        """

        if isinstance(b, (type, tuple)):
            return isinstance(a, b)

        return self.sheerka.isinstance(a, b)

    @staticmethod
    def is_container(obj):
        """
        A container concept is a builtin concept that embed a result
        :param obj:
        :return:
        """
        if not isinstance(obj, Concept):
            return False

        return obj.key in BuiltinContainers

    def admin_push_ontology(self, context, name):
        return self.sheerka.push_ontology(context, name, False)

    def admin_pop_ontology(self, context):
        return self.sheerka.pop_ontology(context)

    def ontologies(self):
        ontologies = self.sheerka.om.ontologies_names
        return self.sheerka.new(BuiltinConcepts.TO_LIST, body=ontologies)

    def in_memory(self):
        """
        Returns the list of all obj in memory
        """
        res = {}
        for k, obj in self.sheerka.om.get_all(SheerkaMemory.OBJECTS_ENTRY).items():
            if isinstance(obj, list):
                obj = obj[-1]
            res[k] = obj.obj

        return self.sheerka.ret(self.NAME, True, self.sheerka.new(BuiltinConcepts.TO_DICT, body=res))

    def admin_history(self, depth=10, start=0):
        history = self.sheerka.services[SheerkaHistoryManager.NAME].history(depth, start)
        return self.sheerka.new(BuiltinConcepts.TO_LIST, body=history)