from dataclasses import dataclass from typing import Union import core.utils ACTION_TYPE_PRINT = "print" ACTION_TYPE_EXEC = "exec" ACTION_TYPE_TEST = "test" @dataclass class RuleMetadata: action_type: str # print, exec name: Union[str, None] predicate: str action: str id: str = None id_is_unresolved = False is_compiled: bool = False is_enabled: bool = False class Rule: def __init__(self, action_type=ACTION_TYPE_EXEC, name=None, predicate=None, action=None, priority=None, rule_id=None, is_enabled=None): self.metadata = RuleMetadata(action_type, name, predicate, action, id=rule_id, is_enabled=is_enabled) self.compiled_predicates = None self.compiled_action = None from core.sheerka.services.SheerkaComparisonManager import SheerkaComparisonManager self.priority = priority if priority is not None else SheerkaComparisonManager.DEFAULT_COMPARISON_VALUE self.error_sink = None # from SheerkaRete, not quite sure one when it will be used self.rete_net = None self.rete_p_nodes = [] # list of production nodes for this rule self.rete_disjunctions = None # list of list as it may be several interpretation for a rule def __repr__(self): rule_id = f"#{self.metadata.id}" if self.name: rule_id += f" ({self.metadata.name})" action_type = "print" if self.metadata.action_type == "print" else "then" return f"Rule({rule_id}, when '{self.metadata.predicate}' {action_type} '{self.metadata.action}', priority={self.priority})" def __eq__(self, other): if id(other) == id(self): return True if not isinstance(other, Rule): return False for p in ["name", "predicate", "action_type", "action", "id"]: if getattr(self.metadata, p) != getattr(other.metadata, p): return False return True def __hash__(self): return hash((self.metadata.name, self.metadata.predicate, self.metadata.action_type, self.metadata.action)) def __deepcopy__(self, memodict={}): copy = Rule(self.metadata.action_type, self.name, self.metadata.predicate, self.metadata.action, self.priority, self.id, self.metadata.is_enabled) copy.compiled_predicates = self.compiled_predicates copy.compiled_action = self.compiled_action copy.metadata.is_compiled = self.metadata.is_compiled copy.metadata.id_is_unresolved = self.metadata.id_is_unresolved # copy.error_sink = self.error_sink # Uncomment this line if necessary copy.rete_net = self.rete_net copy.rete_p_nodes = self.rete_p_nodes copy.rete_disjunctions = self.rete_disjunctions return copy def __copy__(self): return self.__deepcopy__() def set_id(self, rule_id): self.metadata.id = rule_id return self def to_tuple_id(self): return self.metadata.name, self.id @property def id(self): return self.metadata.id @property def name(self): return self.metadata.name @property def key(self): return self.metadata.name @property def str_id(self): return core.utils.str_concept(self, drop_name=True, prefix="r:") def short_str(self): return f"Rule(#{self.metadata.id}, '{self.metadata.predicate}', priority={self.priority})" def get_rete_disjunctions(self): return self.rete_disjunctions