128 lines
4.7 KiB
Python
128 lines
4.7 KiB
Python
import time
|
|
from os import path
|
|
|
|
from core.builtin_concepts import BuiltinConcepts
|
|
from core.sheerka.services.sheerka_service import BaseService
|
|
|
|
CONCEPTS_FILE_LITE = "_concepts_lite.txt"
|
|
CONCEPTS_FILE_ALL_CONCEPTS = "_concepts.txt"
|
|
CONCEPTS_FILE_TO_USE = CONCEPTS_FILE_ALL_CONCEPTS
|
|
|
|
|
|
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.last_created_concept, False)
|
|
self.sheerka.bind_service_method(self.last_ret, False)
|
|
|
|
def caches_names(self):
|
|
"""
|
|
Returns the name of all the caches
|
|
:return:
|
|
"""
|
|
return list(self.sheerka.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.cache_manager.caches:
|
|
return self.sheerka.new(BuiltinConcepts.NOT_FOUND, body={"cache": name})
|
|
|
|
if not keys:
|
|
return self.sheerka.cache_manager.caches[name].cache.copy()
|
|
|
|
return {key: self.sheerka.cache_manager.get(name, key) for key in keys}
|
|
|
|
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
|
|
if not path.exists(file_name):
|
|
self.sheerka.log.error(f"\u001b[31mFile '{file_name}' is not found !\u001b[0m")
|
|
return 0, 0, 1
|
|
|
|
with open(file_name, "r") as f:
|
|
for line in f.readlines():
|
|
_nb_lines += 1
|
|
line = line.strip()
|
|
|
|
if line.startswith("#import "):
|
|
to_import = "_concepts_" + line[8:] + ".txt"
|
|
self.sheerka.log.info(f"Importing {to_import}")
|
|
res = restore_from_file(to_import)
|
|
_nb_lines += res[0]
|
|
_nb_instructions += res[1]
|
|
_nb_lines_in_error += res[2]
|
|
continue
|
|
|
|
if line == "" or line.startswith("#"):
|
|
continue
|
|
|
|
self.sheerka.log.info(line)
|
|
_nb_instructions += 1
|
|
res = self.sheerka.evaluate_user_input(line)
|
|
if len(res) > 1 or not res[0].status:
|
|
_nb_lines_in_error += 1
|
|
self.sheerka.log.error("\u001b[31mError detected !\u001b[0m")
|
|
|
|
return _nb_lines, _nb_instructions, _nb_lines_in_error
|
|
|
|
if concept_file == "full":
|
|
concept_file = CONCEPTS_FILE_ALL_CONCEPTS
|
|
|
|
elif not concept_file.startswith("_concepts"):
|
|
concept_file = f"_concepts_{concept_file}.txt"
|
|
|
|
try:
|
|
start = time.time_ns()
|
|
self.sheerka.during_restore = True
|
|
nb_lines, nb_instructions, nb_lines_in_error = restore_from_file(concept_file)
|
|
self.sheerka.during_restore = False
|
|
stop = time.time_ns()
|
|
|
|
nano_sec = stop - start
|
|
dt = nano_sec / 1e6
|
|
elapsed = f"{dt} ms" if dt < 1000 else f"{dt / 1000} s"
|
|
self.sheerka.log.info(f"Imported {nb_lines} line(s) in {elapsed}.")
|
|
self.sheerka.log.info(f"{nb_instructions} instruction(s).")
|
|
if nb_lines_in_error > 0:
|
|
self.sheerka.log.info(f"\u001b[31m{nb_lines_in_error} errors(s) found.\u001b[0m")
|
|
else:
|
|
self.sheerka.log.info(f"No error.")
|
|
except IOError as e:
|
|
raise e
|
|
|
|
def concepts(self):
|
|
return self.sheerka.sdp.list(self.sheerka.CONCEPTS_BY_ID_ENTRY)
|
|
|
|
def last_created_concept(self, use_history=False):
|
|
for exec_result in reversed(self.sheerka.last_executions):
|
|
return_values = exec_result.values["return_values"]
|
|
for ret in return_values:
|
|
if ret.status and self.sheerka.isinstance(ret.value, BuiltinConcepts.NEW_CONCEPT):
|
|
return ret.value.body
|
|
|
|
if use_history:
|
|
return self.sheerka.new(BuiltinConcepts.ERROR, body="Not yet implement")
|
|
|
|
return self.sheerka.new(BuiltinConcepts.NOT_FOUND)
|
|
|
|
def last_ret(self, context, index=-1):
|
|
return self.sheerka.last_return_values[index]
|