import os import pprint import re from dataclasses import dataclass from core.builtin_concepts import BuiltinConcepts from core.sheerka.ExecutionContext import ExecutionContext from core.sheerka.services.sheerka_service import BaseService from core.utils import CONSOLE_COLORS_MAP as CCM from core.utils import evaluate_expression, as_bag try: rows, columns = os.popen('stty size', 'r').read().split() except ValueError: rows, columns = 50, 80 pp = pprint.PrettyPrinter(indent=2, width=columns) class BaseDebugLogger: ids = {} @staticmethod def next_id(hint): if hint in BaseDebugLogger.ids: BaseDebugLogger.ids[hint] += 1 else: BaseDebugLogger.ids[hint] = 0 return BaseDebugLogger.ids[hint] def __init__(self, debug_manager, context, who, method_name, debug_id): pass def debug_entering(self, **kwargs): pass def debug_log(self, text, is_error=False): pass def debug_var(self, name, value, is_error=False, hint=None): pass def debug_rule(self, rule, results): pass def debug_concept(self, concept, text=None, **kwargs): pass def is_enabled(self): pass class NullDebugLogger(BaseDebugLogger): def __init__(self): pass def is_enabled(self): return False class ConsoleDebugLogger(BaseDebugLogger): def __init__(self, debug_manager, context, service_name, method_name, debug_id): BaseDebugLogger.__init__(self, debug_manager, context, service_name, method_name, debug_id) self.debug_manager = debug_manager self.service_name = service_name self.method_name = method_name self.context = context self.debug_id = debug_id self.is_highlighted = "" def is_enabled(self): return True def debug_entering(self, **kwargs): super().debug_entering(**kwargs) str_text = f"{CCM['blue']}Entering {self.service_name}.{self.method_name} with {CCM['reset']}" str_vars = pp.pformat(kwargs) if "\n" not in str(str_vars): self.debug_manager.debug(self.prefix() + str_text + str_vars) else: self.debug_manager.debug(self.prefix() + str_text) self.debug_manager.debug(self.prefix() + str_vars) def debug_log(self, text, is_error=False): color = 'red' if is_error else 'blue' self.debug_manager.debug(self.prefix() + f"{CCM[color]}..{text}{CCM['reset']}") def debug_var(self, name, value, is_error=False, hint=None): enabled = is_error or self.debug_manager.compute_debug_var(self.context, self.service_name, self.method_name, name, self.debug_id) if enabled == False: return color = 'red' if is_error else 'green' hint_str = f"({hint})" if hint is not None else "" str_text = f"{CCM[color]}..{name}{hint_str}={CCM['reset']}" str_vars = "" if isinstance(enabled, str) else pp.pformat(value) self.debug(str_text, str_vars) def debug_rule(self, rule, results): if not self.debug_manager.compute_debug_rule(self.context, self.service_name, self.method_name, rule.id, self.debug_id): return str_text = f"{CCM['green']}..results({rule.id})={CCM['reset']}" str_vars = pp.pformat(results) self.debug(str_text, str_vars) def debug_concept(self, concept, text=None, **kwargs): raw = kwargs.pop('raw', None) if not self.debug_manager.compute_debug_concept(self.context, self.service_name, self.method_name, concept.id, self.debug_id): return str_vars = raw if raw else pp.pformat(kwargs) if kwargs else "" text = " - " + text if text is not None else "" colon = ": " if str_vars else "" str_text = f"{CCM['cyan']}..concept#{concept.id}{text}{colon} {CCM['reset']}" self.debug(str_text, str_vars) def debug(self, str_text, str_vars): if "\n" not in str(str_vars): self.debug_manager.debug(self.prefix() + str_text + str_vars) else: self.debug_manager.debug(self.prefix() + str_text) self.debug_manager.debug(self.prefix() + str_vars) def prefix(self): return f"[{self.context.id:2}][{self.debug_id:2}] {self.is_highlighted}" @dataclass class DebugItem: item: str service_name: str method_name: str context_id: int context_children: bool debug_id: int debug_children: bool enabled: bool class SheerkaDebugManager(BaseService): NAME = "Debug" PREFIX = "debug." children_activation_regex = re.compile(r"(\d+)\+") def __init__(self, sheerka): super().__init__(sheerka) self.activated = False # is debug activated self.explicit = False # No need to activate context debug when debug mode is on self.context_cache = set() # debug for specific context self.variable_cache = set() # debug for specific variable self.debug_vars_settings = [] self.debug_rules_settings = [] self.debug_concepts_settings = [] def initialize(self): # TO REMOVE ??? self.sheerka.bind_service_method(self.set_explicit, True) self.sheerka.bind_service_method(self.activate_debug_for, True) self.sheerka.bind_service_method(self.deactivate_debug_for, True) self.sheerka.bind_service_method(self.debug_activated, False) self.sheerka.bind_service_method(self.debug_activated_for, False) self.sheerka.bind_service_method(self.get_context_debug_mode, False) self.sheerka.bind_service_method(self.debug_rule_activated, False) self.sheerka.bind_service_method(self.debug, False, visible=False) self.sheerka.bind_service_method(self.set_debug, True) self.sheerka.bind_service_method(self.inspect, False) self.sheerka.bind_service_method(self.get_debugger, False) self.sheerka.bind_service_method(self.reset_debug, False) self.sheerka.bind_service_method(self.debug_var, True) self.sheerka.bind_service_method(self.debug_rule, True) self.sheerka.bind_service_method(self.debug_concept, True) # self.sheerka.bind_service_method(self.get_debug_settings, False, as_name="debug_settings") def initialize_deferred(self, context, is_first_time): self.restore_values("activated", "explicit", "context_cache", "variable_cache", "debug_vars_settings", "debug_rules_settings", "debug_concepts_settings") def reset(self): """ For test purpose :return: """ self.activated = False self.context_cache.clear() self.variable_cache.clear() self.debug_vars_settings.clear() self.debug_rules_settings.clear() self.debug_concepts_settings.clear() def set_debug(self, context, value=True): self.activated = value self.sheerka.record_var(context, self.NAME, "activated", self.activated) return self.sheerka.ret(SheerkaDebugManager.NAME, True, self.sheerka.new(BuiltinConcepts.SUCCESS)) def set_explicit(self, context, value=True): self.explicit = value self.sheerka.record_var(context, self.NAME, "explicit", self.explicit) return self.sheerka.ret(SheerkaDebugManager.NAME, True, self.sheerka.new(BuiltinConcepts.SUCCESS)) def activate_debug_for(self, context, debug_id, children=False): """ :param context: :param debug_id: if debug_id is str, activate variable cache, context_cache otherwise :param children: :return: """ # preprocess if isinstance(debug_id, str) and (m := self.children_activation_regex.match(debug_id)): debug_id = int(m.group(1)) children = True if isinstance(debug_id, str): self.variable_cache.add(debug_id) self.sheerka.record_var(context, self.NAME, "variable_cache", self.variable_cache) else: self.context_cache.add(debug_id) if children: self.context_cache.add(str(debug_id) + "+") self.sheerka.record_var(context, self.NAME, "context_cache", self.context_cache) return self.sheerka.ret(SheerkaDebugManager.NAME, True, self.sheerka.new(BuiltinConcepts.SUCCESS)) def deactivate_debug_for(self, context, debug_id, children=False): if isinstance(debug_id, str): self.variable_cache.discard(debug_id) self.sheerka.record_var(context, self.NAME, "variable_cache", self.variable_cache) else: self.context_cache.discard(debug_id) if children: self.context_cache.discard(str(debug_id) + "+") self.sheerka.record_var(context, self.NAME, "context_cache", self.context_cache) return self.sheerka.ret(SheerkaDebugManager.NAME, True, self.sheerka.new(BuiltinConcepts.SUCCESS)) def debug_activated(self): return self.activated def debug_activated_for(self, debug_id): if not self.activated: return None return debug_id in self.variable_cache def debug_rule_activated(self, rule_id, context_id): """ :param rule_id: :param context_id: :return: """ key = f"{rule_id}|{context_id}" return key in self.rules_cache def get_context_debug_mode(self, context_id): if not self.activated: return None, None debug_for_children = "protected" if str(context_id) + "+" in self.context_cache else None debug_for_self = "private" if not self.explicit or context_id in self.context_cache else None return debug_for_self, debug_for_children def inspect(self, context, context_id, *props): """ Print :param context: :param context_id: :return: """ to_inspect = self.sheerka.get_execution_item(context, context_id) if not isinstance(to_inspect, ExecutionContext): return to_inspect if not props: props = ["inputs", "values.return_values"] bag = as_bag(to_inspect) res = {} for prop in props: res[prop] = evaluate_expression(prop, bag) return self.sheerka.new(BuiltinConcepts.TO_DICT, body=res) def debug(self, *args, **kwargs): print(*args, **kwargs) def get_debugger(self, context, who, method_name): if self.compute_debug(context, who, method_name): debug_id = ConsoleDebugLogger.next_id(context.event.get_digest() + str(context.id)) return ConsoleDebugLogger(self, context, who, method_name, debug_id) return NullDebugLogger() def add_or_update_debug_item(self, context, item_type, item=None, service=None, method=None, context_id=None, context_children=False, debug_id=None, debug_children=False, enabled=True): # if the setting already exist, update it item_type_full_name = self.container_name(item_type) items_container = getattr(self, item_type_full_name) for setting in items_container: if setting.item == item and \ setting.service_name == service and \ setting.method_name == method and \ setting.context_id == context_id and \ setting.context_children == context_children and \ setting.debug_id == debug_id and \ setting.debug_children == debug_children: setting.enabled = enabled break else: items_container.append(DebugItem(item, service, method, context_id, context_children, debug_id, debug_children, enabled)) self.sheerka.record_var(context, self.NAME, item_type_full_name, items_container) return self.sheerka.ret(SheerkaDebugManager.NAME, True, self.sheerka.new(BuiltinConcepts.SUCCESS)) def compute_debug(self, context, service_name, method_name): """ Using the debug info, tells if the debug is active for a given service, method, context (and debug_id) :param context: :param service_name: :param method_name: :return: """ if not self.activated: return False selected = [] for item_type in ["vars", "rules", "concepts"]: for setting in getattr(self, self.container_name(item_type)): if setting.service_name is None and setting.method_name is None and setting.context_id is None: continue if (setting.service_name is None or setting.service_name == service_name) and \ (setting.method_name is None or setting.method_name == method_name) and \ (setting.context_id is None or setting.context_id == context.id or ( setting.context_children and context.has_parent(setting.context_id))): selected.append(setting.enabled) if len(selected) == 0: return False res = selected[0] for enabled in selected[1:]: res |= enabled return res def compute_debug_item(self, item_type, context, service_name, method_name, item, debug_id): """ Using the debug info, tells if debug is activated for a given item :param context: :param item_type: :param service_name: :param method_name: :param item: :param debug_id: :return: """ if not self.activated: return False selected = [] for setting in getattr(self, self.container_name(item_type)): if setting.item is None and setting.debug_id is None: continue if (setting.service_name is None or setting.service_name == service_name) and \ (setting.method_name is None or setting.method_name == method_name) and \ (setting.context_id is None or setting.context_id == context.id or ( setting.context_children and context.has_parent(setting.context_id))) and \ (setting.item is None or setting.item == "*" or setting.item == item) and \ (setting.debug_id is None or setting.debug_id == debug_id): selected.append(setting.enabled) if len(selected) == 0: return False res = selected[0] for enabled in selected[1:]: if res == False or enabled == False: return False if isinstance(res, str): continue res = enabled return res def reset_debug(self, context): for item_type in ["vars", "rules", "concepts"]: setting_name = self.container_name(item_type) settings = getattr(self, setting_name) settings.clear() self.sheerka.record_var(context, self.NAME, setting_name, settings) return self.sheerka.ret(SheerkaDebugManager.NAME, True, self.sheerka.new(BuiltinConcepts.SUCCESS)) def debug_var(self, context, *args, **kwargs): """ Adds debug item for variables debug_var(.., [+], ) with service, method and vat that can be '*' :param context: :param args: :param kwargs: :return: """ i, s, m, c_id, c_children, d, e = self.parse_debug_args("variable", *args, **kwargs) return self.add_or_update_debug_item(context, "vars", i, s, m, c_id, c_children, d, False, e) def debug_rule(self, context, *args, **kwargs): """ Adds debug item for rules debug_var(.., [+], ) with service, method and rule that can be '*' debug_var(rule_id, [+], ) :param context: :param args: :param kwargs: :return: """ i, s, m, c_id, c_children, d, e = self.parse_debug_args("rule", *args, **kwargs) return self.add_or_update_debug_item(context, "rules", i, s, m, c_id, c_children, d, False, e) def debug_concept(self, context, *args, **kwargs): """ Adds debug item for concepts debug_var(.., [+], ) with service, method and vat that can be '*' debug_var(concept_id, [+], ) :param context: :param args: :param kwargs: :return: """ i, s, m, c_id, c_children, d, e = self.parse_debug_args("concept", *args, **kwargs) return self.add_or_update_debug_item(context, "concepts", i, s, m, c_id, c_children, d, False, e) def compute_debug_var(self, context, service_name, method_name, item, debug_id): return self.compute_debug_item("vars", context, service_name, method_name, item, debug_id) def compute_debug_concept(self, context, service_name, method_name, item, debug_id): return self.compute_debug_item("concepts", context, service_name, method_name, item, debug_id) def compute_debug_rule(self, context, service_name, method_name, item, debug_id): return self.compute_debug_item("rules", context, service_name, method_name, item, debug_id) @staticmethod def container_name(item_type): return f"debug_{item_type}_settings" @staticmethod def parse_debug_args(item_name, *args, **kwargs): service, method_name, context_id, context_children, item, debug_id, enabled = None, None, None, False, None, None, True if len(args) > 0: if args[0] is None or args[0] == "": pass elif isinstance(args[0], int): item = str(args[0]) else: parts = args[0].split(".") service = None if parts[0] == "*" else parts[0] if len(parts) > 1: method_name = None if parts[1] == "*" else parts[1] if len(parts) > 2: item = parts[2] if len(args) > 1: context_part = args[1] if isinstance(context_part, int): context_id = context_part if isinstance(context_part, str): m = SheerkaDebugManager.children_activation_regex.match(context_part) if m: context_id = int(m.group(1)) context_children = True else: try: context_id = int(context_part) except ValueError: pass if len(args) > 2: debug_id = args[2] service = kwargs.get("service", service) method_name = kwargs.get("method", method_name) context_id = kwargs.get("context_id", context_id) context_children = kwargs.get("context_children", context_children) item = kwargs.get(item_name, item) debug_id = kwargs.get("debug_id", debug_id) enabled = kwargs.get("enabled", enabled) return item, service, method_name, context_id, context_children, debug_id, enabled # def debug_rule(self, context, rule=None, context_id=None, debug_id=None, enabled=True): # """ # Add a debug rule request # :param context: # :param rule: # :param context_id: # :param debug_id: # :param enabled: # :return: # """ # rule = str(rule) if rule is not None else None # for setting in self.debug_rules_settings: # if setting.rule_id == rule and \ # setting.context_id == context_id and \ # setting.debug_id == debug_id: # setting.enabled = enabled # break # else: # self.debug_rules_settings.append(DebugRuleSetting(rule, # context_id, # debug_id, # enabled)) # # self.sheerka.record_var(context, self.NAME, "debug_rules_settings", self.debug_rules_settings) # return self.sheerka.ret(SheerkaDebugManager.NAME, True, self.sheerka.new(BuiltinConcepts.SUCCESS)) # # def debug_concept(self, context, concept=None, context_id=None, debug_id=None, enabled=True): # """ # Add a debug rule request # :param context: # :param concept: # :param context_id: # :param debug_id: # :param enabled: # :return: # """ # concept_id = concept.id if isinstance(concept, Concept) else str(concept) if concept is not None else None # for setting in self.debug_concepts_settings: # if setting.concept_id == concept_id and \ # setting.context_id == context_id and \ # setting.debug_id == debug_id: # setting.enabled = enabled # break # else: # self.debug_concepts_settings.append(DebugConceptSetting(concept_id, # context_id, # debug_id, # enabled)) # # self.sheerka.record_var(context, self.NAME, "debug_concepts_settings", self.debug_concepts_settings) # return self.sheerka.ret(SheerkaDebugManager.NAME, True, self.sheerka.new(BuiltinConcepts.SUCCESS)) # # # def compute_debug_var(self, service_name, method_name, context_id, variable_name, debug_id): # if not self.activated: # return False # # selected = [] # for setting in self.debug_vars_settings: # if setting.variable_name is None and setting.debug_id is None: # continue # # if (setting.service_name is None or setting.service_name == service_name) and \ # (setting.method_name is None or setting.method_name == method_name) and \ # (setting.context_id is None or setting.context_id == context_id) and \ # (setting.variable_name is None or # setting.variable_name == "*" or # setting.variable_name == variable_name) and \ # (setting.debug_id is None or setting.debug_id == debug_id): # selected.append(setting.enabled) # # if len(selected) == 0: # return False # # res = selected[0] # for enabled in selected[1:]: # if res == False or enabled == False: # return False # # if isinstance(res, str): # continue # # res = enabled # # return res # # def compute_debug_rule(self, rule_id, context_id, debug_id): # if not self.activated: # return False # # selected = [] # for setting in self.debug_rules_settings: # if (setting.rule_id is None or setting.rule_id == rule_id) and \ # (setting.context_id is None or setting.context_id == context_id) and \ # (setting.debug_id is None or setting.debug_id == debug_id): # selected.append(setting.enabled) # # if len(selected) == 0: # return False # # res = selected[0] # for enabled in selected[1:]: # res &= enabled # # return res # # def compute_debug_concept(self, concept_id, context_id, debug_id): # if not self.activated: # return False # # selected = [] # for setting in self.debug_concepts_settings: # if (setting.concept_id is None or setting.concept_id == concept_id) and \ # (setting.context_id is None or setting.context_id == context_id) and \ # (setting.debug_id is None or setting.debug_id == debug_id): # selected.append(setting.enabled) # # if len(selected) == 0: # return False # # res = selected[0] # for enabled in selected[1:]: # res &= enabled # # return res # # def reset_debug_rules(self, context): # self.debug_rules_settings.clear() # self.sheerka.record_var(context, self.NAME, "debug_rules_settings", self.debug_rules_settings) # return self.sheerka.ret(SheerkaDebugManager.NAME, True, self.sheerka.new(BuiltinConcepts.SUCCESS)) # # def get_debug_settings(self): # lst = self.debug_vars_settings + self.debug_concepts_settings + self.debug_rules_settings # return self.sheerka.new(BuiltinConcepts.TO_LIST, body=lst)