Merge pull request #33 from adaa-polsl/feat-user-defined-measures

Feat user defined measures

Author: cezary986

.gitignore

@@ -293,4 +293,5 @@ notebooks/
 classification_tabular_datasets/
 .coverage
 junit.xml
-a.ipynb
+a.ipynb
+java.logs

rulekit/_helpers.py

@@ -3,6 +3,7 @@
 import io
 import json
 from typing import Any
+from typing import Callable
 from typing import Optional
 from typing import Union
 
@@ -45,6 +46,11 @@ class RuleGeneratorConfigurator:
     """Class for configuring rule induction parameters
     """
 
+    _MEASURES_PARAMETERS: list[str] = [
+        'induction_measure', 'pruning_measure', 'voting_measure',
+    ]
+    _USER_DEFINED_MEASURE_VALUE: str = 'UserDefined'
+
     def __init__(self, rule_generator):
         self.rule_generator = rule_generator
         self.LogRank = None  # pylint: disable=invalid-name
@@ -96,16 +102,25 @@ def _configure_measure_parameter(self, param_name: str, param_value: Union[str,
             if isinstance(param_value, Measures):
                 self.rule_generator.setParameter(
                     param_name, param_value.value)
-            if isinstance(param_value, str):
-                self.rule_generator.setParameter(param_name, 'UserDefined')
-                self.rule_generator.setParameter(param_name, param_value)
+            if isinstance(param_value, Callable):
+                self._configure_user_defined_measure_parameter(
+                    param_name, param_value)
+
+    def _configure_user_defined_measure_parameter(self, param_name: str, param_value: Any):
+        from rulekit.params import _user_defined_measure_factory
+        user_defined_measure = _user_defined_measure_factory(param_value)
+        {
+            'induction_measure': self.rule_generator.setUserMeasureInductionObject,
+            'pruning_measure': self.rule_generator.setUserMeasurePurningObject,
+            'voting_measure': self.rule_generator.setUserMeasureVotingObject,
+        }[param_name](user_defined_measure)
+        self.rule_generator.setParameter(
+            param_name, self._USER_DEFINED_MEASURE_VALUE)
 
     def _configure_rule_generator(self, **kwargs: dict[str, Any]):
-        if kwargs.get('induction_measure') == Measures.LogRank or \
-                kwargs.get('pruning_measure') == Measures.LogRank or \
-                kwargs.get('voting_measure') == Measures.LogRank:
+        if any([kwargs.get(param_name) == Measures.LogRank for param_name in self._MEASURES_PARAMETERS]):
             self.LogRank = JClass('adaa.analytics.rules.logic.quality.LogRank')
-        for measure_param_name in ['induction_measure', 'pruning_measure', 'voting_measure']:
+        for measure_param_name in self._MEASURES_PARAMETERS:
             measure_param_value: Measures = kwargs.pop(
                 measure_param_name, None)
             self._configure_measure_parameter(
@@ -125,6 +140,15 @@ def _validate_rule_generator_parameters(self, **python_parameters: dict[str, Any
             ValueError: If failed to retrieve RuleGenerator parameters JSON
             RuleKitMisconfigurationException: If Java and Python parameters do not match
         """
+        def are_params_equal(java_params: dict[str, Any], python_params: dict[str, Any]):
+            if java_params.keys() != python_params.keys():
+                return False
+            for key in java_params.keys():
+                skip_check: bool = isinstance(python_params[key], Callable)
+                if java_params[key] != python_params[key] and not skip_check:
+                    return False
+            return True
+
         python_parameters = dict(python_parameters)
         for param_name, param_value in python_parameters.items():
             # convert measures to strings values for comparison
@@ -152,7 +176,7 @@ def _validate_rule_generator_parameters(self, **python_parameters: dict[str, Any
             param_name: str(java_params[param_name])
             for param_name in python_parameters.keys()
         }
-        if java_params != python_parameters:
+        if not are_params_equal(java_params, python_parameters):
             raise RuleKitMisconfigurationException(
                 java_parameters=java_params,
                 python_parameters=python_parameters

rulekit/exceptions.py

@@ -1,5 +1,6 @@
 """Module containing classes for handling exceptions."""
 from typing import Any
+from typing import Callable
 
 from jpype import JException
 
@@ -71,7 +72,9 @@ def _prepare_message(
             java_value = java_parameters.get(key)
             python_value = python_parameters.get(key)
             line: str = f'  {key}: ({java_value},  {python_value}),'
-            if java_value != python_value:
+            # skip check for user defined measures
+            skip_check: bool = isinstance(python_value, Callable)
+            if java_value != python_value and not skip_check:
                 line = f'{line} <-- **DIFFERENT**'
             params_lines.append(line)
         message: str = (

rulekit/params.py

@@ -1,13 +1,36 @@
 """Contains constants and classes for specyfing models parameters
 """
 from enum import Enum
+from typing import Callable
 from typing import Optional
 from typing import Tuple
+from typing import Union
 
+from jpype import JImplements
+from jpype import JOverride
+from jpype.types import JDouble
 from pydantic import BaseModel  # pylint: disable=no-name-in-module
 
 MAX_INT: int = 2147483647  # max integer value in Java
 
+_UserDefinedMeasure = Callable[[float, float, float, float], float]
+
+
+def _user_defined_measure_factory(measure_function: _UserDefinedMeasure):
+    from adaa.analytics.rules.logic.quality import \
+        IUserMeasure  # pylint: disable=import-outside-toplevel,import-error
+
+    @JImplements(IUserMeasure)
+    class _UserMeasure:  # pylint: disable=invalid-name,missing-function-docstring
+
+        @JOverride
+        def getResult(self, p: JDouble, n: JDouble, P: JDouble, N: JDouble) -> float:
+            return measure_function(
+                float(p), float(n), float(P), float(N)
+            )
+
+    return _UserMeasure()
+
 
 class Measures(Enum):
     # pylint: disable=invalid-name
@@ -92,14 +115,16 @@ class Measures(Enum):
     'penalty_saturation': 0.2,
 }
 
+_QualityMeasure = Union[Measures, _UserDefinedMeasure]
+
 
 class ModelsParams(BaseModel):
     """Model for validating models hyperparameters
     """
     minsupp_new: Optional[float] = DEFAULT_PARAMS_VALUE['minsupp_new']
-    induction_measure: Optional[Measures] = DEFAULT_PARAMS_VALUE['induction_measure']
-    pruning_measure: Optional[Measures] = DEFAULT_PARAMS_VALUE['pruning_measure']
-    voting_measure: Optional[Measures] = DEFAULT_PARAMS_VALUE['voting_measure']
+    induction_measure: Optional[_QualityMeasure] = DEFAULT_PARAMS_VALUE['induction_measure']
+    pruning_measure: Optional[_QualityMeasure] = DEFAULT_PARAMS_VALUE['pruning_measure']
+    voting_measure: Optional[_QualityMeasure] = DEFAULT_PARAMS_VALUE['voting_measure']
     max_growing: Optional[float] = DEFAULT_PARAMS_VALUE['max_growing']
     enable_pruning: Optional[bool] = DEFAULT_PARAMS_VALUE['enable_pruning']
     ignore_missing: Optional[bool] = DEFAULT_PARAMS_VALUE['ignore_missing']

tests/test_classifier.py

@@ -11,6 +11,7 @@
 
 from rulekit import classification
 from rulekit.events import RuleInductionProgressListener
+from rulekit.params import Measures
 from rulekit.rules import Rule
 from tests.utils import assert_accuracy_is_greater
 from tests.utils import assert_rules_are_equals
@@ -282,6 +283,49 @@ def test_left_open_intervals_in_expert_induction(self):
             expert_forbidden_conditions=expert_forbidden_conditions
         )
 
+    def test_user_defined_measures(self):
+        def full_coverage(p: float, n: float, P: float, N: float) -> float:
+            return (p + n) / (P + N)
+
+        python_clf = classification.RuleClassifier(
+            induction_measure=full_coverage,
+            pruning_measure=full_coverage,
+            voting_measure=full_coverage,
+        )
+        java_clf = classification.RuleClassifier(
+            induction_measure=Measures.FullCoverage,
+            pruning_measure=Measures.FullCoverage,
+            voting_measure=Measures.FullCoverage,
+        )
+        x, y = load_iris(return_X_y=True)
+
+        python_clf.fit(x, y)
+        java_clf.fit(x, y)
+
+        self.assertEqual(
+            [r.weight for r in python_clf.model.rules],
+            [r.weight for r in java_clf.model.rules],
+            'Weights should be equal'
+        )
+        self.assertEqual(
+            [str(r) for r in python_clf.model.rules],
+            [str(r) for r in java_clf.model.rules],
+            'Rules should be equal'
+        )
+
+        def zero_measure(p: float, n: float, P: float, N: float) -> float:
+            return 0.0
+
+        python_clf2 = classification.RuleClassifier(
+            induction_measure=Measures.FullCoverage,
+            pruning_measure=Measures.FullCoverage,
+            voting_measure=zero_measure,
+        )
+        python_clf2.fit(x, y)
+        self.assertTrue(all([
+            r.weight == 0.0 for r in python_clf2.model.rules
+        ]))
+
 
 if __name__ == '__main__':
     unittest.main()