119 lines
3.6 KiB
Python
119 lines
3.6 KiB
Python
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
|
|
|
|
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
|