⬅ rulekit/events.py source

1 """Module containing classes for tracking progress of rule induction.
2 """
3 from typing import Any
4  
5 from jpype import JImplements
6 from jpype import JInt
7 from jpype import JObject
8 from jpype import JOverride
9  
10 from rulekit.rules import _rule_factory
11 from rulekit.rules import BaseRule
12  
13  
14 class RuleInductionProgressListener:
15 """Base class for rule induction progress listeners. To use it, subclass it
16 and implement some of the its methods. Then instantiate it and pass it to
17 `add_event_listener` method of the operator.
18 """
19  
20 def on_new_rule(self, rule: BaseRule):
21 """Called when new rule is induced
22  
23 Args:
24 rule (BaseRule): Newly induced rule
25 """
26  
27 def on_progress(
28 self,
29 total_examples_count: int,
30 uncovered_examples_count: int
31 ):
32 """Called each time a ruleset coverage changed.
33  
34 This method is best suited to monitor progress of rule induction.
35  
36 Args:
37 total_examples_count (int): Total number of examples in
38 training dataset
39  
40 uncovered_examples_count (int): Number of examples that
41 are not covered by any rule
42 """
43  
44 def should_stop(self) -> bool:
45 """Method which allows to stop rule induction process at given
46 moment. This method is called each time a ruleset coverage changed.
47 If it returns `True`, rule induction process will be stopped if it
48 return `False` it will continue.
49  
50 Returns:
51 bool: whether to stop rule induction or not
52 """
53 return False
54  
55  
56 def _command_listener_factory(listener: RuleInductionProgressListener) -> Any:
57 from adaa.analytics.rules.logic.rulegenerator import \
  • E501 Line too long (94 > 88 characters)
58 ICommandListener # pylint: disable=import-outside-toplevel,import-error,line-too-long
59  
60 @JImplements(ICommandListener)
  • E501 Line too long (100 > 88 characters)
61 class _CommandListener: # pylint: disable=invalid-name,missing-function-docstring,line-too-long
62  
63 @JOverride
64 def onNewRule(self, rule: JObject):
65 rule: BaseRule = _rule_factory(rule)
66 return listener.on_new_rule(rule)
67  
68 @JOverride
69 def onProgress(self, totalRules: JInt, uncoveredRules: JInt):
70 return listener.on_progress(int(totalRules), int(uncoveredRules))
71  
72 @JOverride
73 def isRequestStop(self) -> bool:
74 return listener.should_stop()
75 # return False
76  
77 return _CommandListener()