Coverage for rulekit/events.py: 100%

1"""Module containing classes for tracking progress of rule induction. 

2""" 

3from typing import Any 

4 

5from jpype import JImplements 

6from jpype import JInt 

7from jpype import JObject 

8from jpype import JOverride 

9 

10from rulekit.rules import _rule_factory 

11from rulekit.rules import BaseRule 

12 

13 

14class 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 

56def _command_listener_factory(listener: RuleInductionProgressListener) -> Any: 

57 from adaa.analytics.rules.logic.rulegenerator import \ 

58 ICommandListener # pylint: disable=import-outside-toplevel,import-error,line-too-long 

59 

60 @JImplements(ICommandListener) 

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()