ELI-351: Adds rule processor

Author: shweta-nhs

src/eligibility_signposting_api/services/calculators/eligibility_calculator.py

@@ -16,7 +16,6 @@
     CampaignID,
     CampaignVersion,
     Iteration,
-    IterationCohort,
     IterationRule,
     RuleName,
     RulePriority,
@@ -41,9 +40,10 @@
 )
 from eligibility_signposting_api.services.processors.campaign_evaluator import CampaignEvaluator
 from eligibility_signposting_api.services.processors.person_data_reader import PersonDataReader
+from eligibility_signposting_api.services.processors.rule_processor import RuleProcessor
 
 if TYPE_CHECKING:
-    from collections.abc import Collection, Iterable, Iterator
+    from collections.abc import Collection
 
     from eligibility_signposting_api.model.person import Person
 
@@ -62,6 +62,7 @@ class EligibilityCalculator:
 
     campaign_evaluator: CampaignEvaluator = field(default_factory=CampaignEvaluator)
     person_data_reader: PersonDataReader = field(default_factory=PersonDataReader)
+    rule_processor: RuleProcessor = field(default_factory=RuleProcessor)
 
     results: list[eligibility_status.Condition] = field(default_factory=list)
 
@@ -88,16 +89,6 @@ def get_the_best_cohort_memberships(
 
         return best_status, best_cohorts
 
-    @staticmethod
-    def get_exclusion_rules(cohort: IterationCohort, filter_rules: Iterable[IterationRule]) -> Iterator[IterationRule]:
-        return (
-            ir
-            for ir in filter_rules
-            if ir.cohort_label is None
-            or cohort.cohort_label == ir.cohort_label
-            or (isinstance(ir.cohort_label, (list, set, tuple)) and cohort.cohort_label in ir.cohort_label)
-        )
-
     @staticmethod
     def get_rules_by_type(
         active_iteration: Iteration,
@@ -263,9 +254,9 @@ def get_cohort_results(self, active_iteration: Iteration) -> dict[str, CohortGro
             person_cohorts = self.person_data_reader.get_person_cohorts(self.person)
             if cohort.cohort_label in person_cohorts or cohort.is_magic_cohort:
                 # Eligibility - check
-                if self.is_eligible_by_filter_rules(cohort, cohort_results, filter_rules):
+                if self.rule_processor.is_eligible(self.person, cohort, cohort_results, filter_rules):
                     # Actionability - evaluation
-                    self.evaluate_suppression_rules(cohort, cohort_results, suppression_rules)
+                    self.rule_processor.is_actionable(self.person, cohort, cohort_results, suppression_rules)
 
             # Not base eligible
             elif cohort.cohort_label is not None:
@@ -316,85 +307,6 @@ def build_condition_results(condition_results: dict[ConditionName, IterationResu
             )
         return conditions
 
-    def is_eligible_by_filter_rules(
-        self,
-        cohort: IterationCohort,
-        cohort_results: dict[str, CohortGroupResult],
-        filter_rules: Iterable[IterationRule],
-    ) -> bool:
-        is_eligible = True
-        priority_getter = attrgetter("priority")
-        sorted_rules_by_priority = sorted(self.get_exclusion_rules(cohort, filter_rules), key=priority_getter)
-
-        for _, rule_group in groupby(sorted_rules_by_priority, key=priority_getter):
-            status, group_exclusion_reasons, _ = self.evaluate_rules_priority_group(rule_group)
-            if status.is_exclusion:
-                if cohort.cohort_label is not None:
-                    cohort_results[cohort.cohort_label] = CohortGroupResult(
-                        cohort.cohort_group,
-                        Status.not_eligible,
-                        [],
-                        cohort.negative_description,
-                        group_exclusion_reasons,
-                    )
-                is_eligible = False
-                break
-        return is_eligible
-
-    def evaluate_suppression_rules(
-        self,
-        cohort: IterationCohort,
-        cohort_results: dict[str, CohortGroupResult],
-        suppression_rules: Iterable[IterationRule],
-    ) -> None:
-        is_actionable: bool = True
-        priority_getter = attrgetter("priority")
-        suppression_reasons = []
-
-        sorted_rules_by_priority = sorted(self.get_exclusion_rules(cohort, suppression_rules), key=priority_getter)
-
-        for _, rule_group in groupby(sorted_rules_by_priority, key=priority_getter):
-            status, group_exclusion_reasons, rule_stop = self.evaluate_rules_priority_group(rule_group)
-            if status.is_exclusion:
-                is_actionable = False
-                suppression_reasons.extend(group_exclusion_reasons)
-                if rule_stop:
-                    break
-
-        if cohort.cohort_label is not None:
-            key = cohort.cohort_label
-            if is_actionable:
-                cohort_results[key] = CohortGroupResult(
-                    cohort.cohort_group, Status.actionable, [], cohort.positive_description, suppression_reasons
-                )
-            else:
-                cohort_results[key] = CohortGroupResult(
-                    cohort.cohort_group,
-                    Status.not_actionable,
-                    suppression_reasons,
-                    cohort.positive_description,
-                    suppression_reasons,
-                )
-
-    def evaluate_rules_priority_group(
-        self, rules_group: Iterator[IterationRule]
-    ) -> tuple[eligibility_status.Status, list[eligibility_status.Reason], bool]:
-        is_rule_stop = False
-        exclusion_reasons = []
-        best_status = eligibility_status.Status.not_eligible
-
-        for rule in rules_group:
-            is_rule_stop = rule.rule_stop or is_rule_stop
-            rule_calculator = RuleCalculator(person=self.person, rule=rule)
-            status, reason = rule_calculator.evaluate_exclusion()
-            if status.is_exclusion:
-                best_status = eligibility_status.Status.best(status, best_status)
-                exclusion_reasons.append(reason)
-            else:
-                best_status = eligibility_status.Status.actionable
-
-        return best_status, exclusion_reasons, is_rule_stop
-
     @staticmethod
     def get_actions_from_comms(action_mapper: ActionsMapper, comms: str) -> list[SuggestedAction] | None:
         suggested_actions: list[SuggestedAction] = []

src/eligibility_signposting_api/services/processors/rule_processor.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from itertools import groupby
+from operator import attrgetter
+from typing import TYPE_CHECKING
+
+from wireup import service
+
+from eligibility_signposting_api.model import eligibility_status
+from eligibility_signposting_api.model.eligibility_status import CohortGroupResult, Status
+from eligibility_signposting_api.services.calculators.rule_calculator import RuleCalculator
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable, Iterator
+
+    from eligibility_signposting_api.model.campaign_config import IterationCohort, IterationRule
+    from eligibility_signposting_api.model.person import Person
+
+
+@service
+class RuleProcessor:
+    """Handles the processing and evaluation of different rules (filter, suppression) against person data."""
+
+    def is_eligible(
+        self,
+        person: Person,
+        cohort: IterationCohort,
+        cohort_results: dict[str, CohortGroupResult],
+        filter_rules: Iterable[IterationRule],
+    ) -> bool:
+        is_eligible = True
+        priority_getter = attrgetter("priority")
+        sorted_rules_by_priority = sorted(self.get_exclusion_rules(cohort, filter_rules), key=priority_getter)
+
+        for _, rule_group in groupby(sorted_rules_by_priority, key=priority_getter):
+            status, group_exclusion_reasons, _ = self.evaluate_rules_priority_group(person, rule_group)
+            if status.is_exclusion:
+                if cohort.cohort_label is not None:
+                    cohort_results[cohort.cohort_label] = CohortGroupResult(
+                        cohort.cohort_group,
+                        Status.not_eligible,
+                        [],
+                        cohort.negative_description,
+                        group_exclusion_reasons,
+                    )
+                is_eligible = False
+                break
+        return is_eligible
+
+    def is_actionable(
+        self,
+        person: Person,
+        cohort: IterationCohort,
+        cohort_results: dict[str, CohortGroupResult],
+        suppression_rules: Iterable[IterationRule],
+    ) -> None:
+        is_actionable: bool = True
+        priority_getter = attrgetter("priority")
+        suppression_reasons = []
+
+        sorted_rules_by_priority = sorted(self.get_exclusion_rules(cohort, suppression_rules), key=priority_getter)
+
+        for _, rule_group in groupby(sorted_rules_by_priority, key=priority_getter):
+            status, group_exclusion_reasons, rule_stop = self.evaluate_rules_priority_group(person, rule_group)
+            if status.is_exclusion:
+                is_actionable = False
+                suppression_reasons.extend(group_exclusion_reasons)
+                if rule_stop:
+                    break
+
+        if cohort.cohort_label is not None:
+            key = cohort.cohort_label
+            if is_actionable:
+                cohort_results[key] = CohortGroupResult(
+                    cohort.cohort_group, Status.actionable, [], cohort.positive_description, suppression_reasons
+                )
+            else:
+                cohort_results[key] = CohortGroupResult(
+                    cohort.cohort_group,
+                    Status.not_actionable,
+                    suppression_reasons,
+                    cohort.positive_description,
+                    suppression_reasons,
+                )
+
+    def evaluate_rules_priority_group(
+        self, person: Person, rules_group: Iterator[IterationRule]
+    ) -> tuple[eligibility_status.Status, list[eligibility_status.Reason], bool]:
+        is_rule_stop = False
+        exclusion_reasons = []
+        best_status = eligibility_status.Status.not_eligible
+
+        for rule in rules_group:
+            is_rule_stop = rule.rule_stop or is_rule_stop
+            rule_calculator = RuleCalculator(person=person, rule=rule)
+            status, reason = rule_calculator.evaluate_exclusion()
+            if status.is_exclusion:
+                best_status = eligibility_status.Status.best(status, best_status)
+                exclusion_reasons.append(reason)
+            else:
+                best_status = eligibility_status.Status.actionable
+
+        return best_status, exclusion_reasons, is_rule_stop
+
+    @staticmethod
+    def get_exclusion_rules(cohort: IterationCohort, filter_rules: Iterable[IterationRule]) -> Iterator[IterationRule]:
+        return (
+            ir
+            for ir in filter_rules
+            if ir.cohort_label is None
+            or cohort.cohort_label == ir.cohort_label
+            or (isinstance(ir.cohort_label, (list, set, tuple)) and cohort.cohort_label in ir.cohort_label)
+        )