eli 218 added support cohort groupings plus refactor of calculation logic

* more fields to IterationCohort

* wip

* CohortStatus dataclass, built final result

* created data classess for complex structures

* added an actionable flag

* AND logic for the same priority re-added,
Ignore non-impl tests

* lint fixes

* Addressed cohort label test

* Addressed magic cohorts

* cohort label, rules stop relooked

* fixed rules stop

* multiple_cohorts

* name fix

* import error fix

* rules stop, cohort label

* modified the api response - need refractoring later

* All tests passed 👽

* Imperative name change 👽

* Hamcrest matchers 🙏

* more tests in IT

* more fields to IterationCohort

* wip

* CohortStatus dataclass, built final result

* created data classess for complex structures

* added an actionable flag

* AND logic for the same priority re-added,
Ignore non-impl tests

* lint fixes

* Addressed cohort label test

* Addressed magic cohorts

* cohort label, rules stop relooked

* fixed rules stop

* multiple_cohorts

* name fix

* import error fix

* rules stop, cohort label

* modified the api response - need refractoring later

* All tests passed 👽

* Imperative name change 👽

* Hamcrest matchers 🙏

* more tests in IT

* code cleanup

* Integration Test / Contract tests

* residue ELI-268 - attr_name:cohort-label is default for attr_levl:COHORT

* lint fixes

* lint fixes

* commit refactoring

* tests fixed 🕺🤸🏼‍♀️🕺

* cleanup

* code refactoring

* code refactoring

* rule text

---------

Co-authored-by: ayeshalshukri1-nhs <[email protected]>

Author: Karthikeyannhs

src/eligibility_signposting_api/model/eligibility.py

@@ -67,7 +67,21 @@ class Reason:
 class Condition:
     condition_name: ConditionName
     status: Status
+    cohort_results: list[CohortResult]
+
+
+@dataclass
+class CohortResult:
+    cohort_code: str
+    status: Status
     reasons: list[Reason]
+    description: str
+
+
+@dataclass
+class IterationResult:
+    status: Status
+    cohort_results: list[CohortResult]
 
 
 @dataclass

src/eligibility_signposting_api/model/rules.py

@@ -29,6 +29,8 @@
 StartDate = NewType("StartDate", date)
 EndDate = NewType("EndDate", date)
 CohortLabel = NewType("CohortLabel", str)
+CohortGroup = NewType("CohortGroup", str)
+Description = NewType("Description", str)
 RuleStop = NewType("RuleStop", bool)
 
 
@@ -84,6 +86,9 @@ class RuleAttributeLevel(StrEnum):
 
 class IterationCohort(BaseModel):
     cohort_label: CohortLabel | None = Field(None, alias="CohortLabel")
+    cohort_group: CohortGroup | None = Field(None, alias="CohortGroup")
+    positive_description: Description | None = Field(None, alias="PositiveDescription")
+    negative_description: Description | None = Field(None, alias="NegativeDescription")
     priority: int | None = Field(None, alias="Priority")
 
     model_config = {"populate_by_name": True, "extra": "ignore"}
@@ -95,7 +100,7 @@ class IterationRule(BaseModel):
     description: RuleDescription = Field(..., alias="Description")
     priority: RulePriority = Field(..., alias="Priority")
     attribute_level: RuleAttributeLevel = Field(..., alias="AttributeLevel")
-    attribute_name: RuleAttributeName = Field(..., alias="AttributeName")
+    attribute_name: RuleAttributeName | None = Field(None, alias="AttributeName")
     cohort_label: CohortLabel | None = Field(None, alias="CohortLabel")
     operator: RuleOperator = Field(..., alias="Operator")
     comparator: RuleComparator = Field(..., alias="Comparator")

src/eligibility_signposting_api/services/calculators/eligibility_calculator.py

@@ -1,16 +1,24 @@
 from __future__ import annotations
 
 from _operator import attrgetter
-from collections import defaultdict
-from collections.abc import Collection, Iterator, Mapping
+from collections.abc import Collection, Iterable, Iterator, Mapping
 from dataclasses import dataclass, field
-from functools import cached_property
 from itertools import groupby
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from eligibility_signposting_api.model.rules import Iteration, IterationCohort
 
 from wireup import service
 
 from eligibility_signposting_api.model import eligibility, rules
+from eligibility_signposting_api.model.eligibility import (
+    CohortResult,
+    Condition,
+    ConditionName,
+    IterationResult,
+    Status,
+)
 from eligibility_signposting_api.services.calculators.rule_calculator import RuleCalculator
 
 Row = Collection[Mapping[str, Any]]
@@ -40,107 +48,180 @@ def campaigns_grouped_by_condition_name(
         self,
     ) -> Iterator[tuple[eligibility.ConditionName, list[rules.CampaignConfig]]]:
         """Generator function to iterate over campaign groups by condition name."""
-
         for condition_name, campaign_group in groupby(
             sorted(self.active_campaigns, key=attrgetter("target")), key=attrgetter("target")
         ):
             yield condition_name, list(campaign_group)
 
-    @cached_property
+    @property
     def person_cohorts(self) -> set[str]:
         cohorts_row: Mapping[str, dict[str, dict[str, dict[str, Any]]]] = next(
             (row for row in self.person_data if row.get("ATTRIBUTE_TYPE") == "COHORTS"), {}
         )
         return set(cohorts_row.get("COHORT_MAP", {}).get("cohorts", {}).get("M", {}).keys())
 
+    @staticmethod
+    def get_best_cohort(cohort_results: dict[str, CohortResult]) -> tuple[Status, list[CohortResult]]:
+        if not cohort_results:
+            return eligibility.Status.not_eligible, []
+        best_status = eligibility.Status.best(*[result.status for result in cohort_results.values()])
+        best_cohorts = [result for result in cohort_results.values() if result.status == best_status]
+        return best_status, best_cohorts
+
+    @staticmethod
+    def get_exclusion_rules(
+        cohort: IterationCohort, rules_filter: Iterable[rules.IterationRule]
+    ) -> Iterator[rules.IterationRule]:
+        return (
+            ir
+            for ir in rules_filter
+            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,
+    ) -> tuple[tuple[rules.IterationRule, ...], tuple[rules.IterationRule, ...]]:
+        rules_by_type = {
+            rule_type: tuple(rule for rule in active_iteration.iteration_rules if attrgetter("type")(rule) == rule_type)
+            for rule_type in (rules.RuleType.filter, rules.RuleType.suppression, rules.RuleType.redirect)
+        }
+        rules_filter = rules_by_type[rules.RuleType.filter]
+        rules_suppression = rules_by_type[rules.RuleType.suppression]
+        return rules_filter, rules_suppression
+
     def evaluate_eligibility(self) -> eligibility.EligibilityStatus:
         """Iterates over campaign groups, evaluates eligibility, and returns a consolidated status."""
+        results: dict[ConditionName, IterationResult] = {}
 
         for condition_name, campaign_group in self.campaigns_grouped_by_condition_name:
-            if base_eligible_campaigns := self.get_the_base_eligible_campaigns(campaign_group):
-                status, reasons = self.evaluate_eligibility_by_iteration_rules(base_eligible_campaigns)
-                # Append the evaluation result for this condition to the results list
-                self.results.append(eligibility.Condition(condition_name, status, reasons))
+            iteration_results: dict[str, IterationResult] = {}
+
+            for active_iteration in [cc.current_iteration for cc in campaign_group]:
+                cohort_results: dict[str, CohortResult] = {}
+
+                rules_filter, rules_suppression = self.get_rules_by_type(active_iteration)
+                for cohort in sorted(active_iteration.iteration_cohorts, key=attrgetter("priority")):
+                    # Check Base Eligibility
+                    if cohort.cohort_label in self.person_cohorts or cohort.cohort_label == magic_cohort:
+                        is_eligible: bool = True
+                        is_eligible = self.evaluate_filter_rules(
+                            cohort,
+                            cohort_results,
+                            rules_filter,
+                            is_eligible=is_eligible,
+                        )
+
+                        if is_eligible:
+                            is_actionable: bool = True
+                            suppression_reasons, is_actionable = self.evaluate_suppression_rules(
+                                cohort,
+                                rules_suppression,
+                                is_actionable=is_actionable,
+                            )
+                            if cohort.cohort_label is not None:
+                                key = cohort.cohort_label
+                                if is_actionable:
+                                    cohort_results[key] = CohortResult(
+                                        cohort.cohort_group if cohort.cohort_group else key,
+                                        Status.actionable,
+                                        [],
+                                        str(cohort.positive_description),
+                                    )
+                                else:
+                                    cohort_results[key] = CohortResult(
+                                        cohort.cohort_group if cohort.cohort_group else key,
+                                        Status.not_actionable,
+                                        suppression_reasons,
+                                        str(cohort.positive_description),
+                                    )
+
+                    # Not base eligible
+                    elif cohort.cohort_label is not None:
+                        cohort_results[cohort.cohort_label] = CohortResult(
+                            cohort.cohort_group if cohort.cohort_group else cohort.cohort_label,
+                            Status.not_eligible,
+                            [],
+                            str(cohort.negative_description),
+                        )
+
+                # Determine Result between cohorts - get the best
+                status, best_cohorts = self.get_best_cohort(cohort_results)
+                iteration_results[active_iteration.name] = IterationResult(status, best_cohorts)
+
+            # Determine results between iterations - get the best
+            if iteration_results:
+                best_candidate = max(iteration_results.values(), key=lambda r: r.status.value)
             else:
-                # Create and append the evaluation result, as no campaign config is base eligible
-                self.results.append(eligibility.Condition(condition_name, eligibility.Status.not_eligible, []))
-
-        # Return the overall eligibility status, constructed from the list of condition results
-        return eligibility.EligibilityStatus(conditions=list(self.results))
-
-    def get_the_base_eligible_campaigns(self, campaign_group: list[rules.CampaignConfig]) -> list[rules.CampaignConfig]:
-        """Return campaigns for which the person is base eligible via cohorts."""
-
-        base_eligible_campaigns: list[rules.CampaignConfig] = [
-            campaign for campaign in campaign_group if self.check_base_eligibility(campaign.current_iteration)
+                best_candidate = IterationResult(eligibility.Status.not_eligible, [])
+            results[condition_name] = best_candidate
+
+        # Consolidate all the results and return
+        final_result = [
+            Condition(
+                condition_name=condition_name,
+                status=active_iteration_result.status,
+                cohort_results=active_iteration_result.cohort_results,
+            )
+            for condition_name, active_iteration_result in results.items()
         ]
+        return eligibility.EligibilityStatus(conditions=final_result)
 
-        if base_eligible_campaigns:
-            return base_eligible_campaigns
-        return []
-
-    def check_base_eligibility(self, iteration: rules.Iteration) -> bool:
-        """Return cohorts for which person is base eligible."""
-        iteration_cohorts: set[str] = {
-            cohort.cohort_label for cohort in iteration.iteration_cohorts if cohort.cohort_label
-        }
-        if magic_cohort in iteration_cohorts:
-            return True
-        return bool(iteration_cohorts & self.person_cohorts)
-
-    def evaluate_eligibility_by_iteration_rules(
-        self, campaign_group: list[rules.CampaignConfig]
-    ) -> tuple[eligibility.Status, list[eligibility.Reason]]:
-        """Evaluate iteration rules to see if the person is actionable, not actionable (due to "S" rules),
-        or not eligible (due to "F" rules").
-
-        For each condition, evaluate all iterations for inclusion or exclusion."""
-
+    def evaluate_filter_rules(
+        self,
+        cohort: IterationCohort,
+        cohort_results: dict[str, CohortResult],
+        rules_filter: Iterable[rules.IterationRule],
+        *,
+        is_eligible: bool,
+    ) -> bool:
         priority_getter = attrgetter("priority")
+        sorted_rules_by_priority = sorted(self.get_exclusion_rules(cohort, rules_filter), key=priority_getter)
 
-        status_with_reasons: dict[eligibility.Status, list[eligibility.Reason]] = defaultdict()
-
-        for iteration in [cc.current_iteration for cc in campaign_group]:
-            # Until we see a worse status, we assume someone is actionable for this iteration.
-            worst_status = eligibility.Status.actionable
-            exclusion_reasons, actionable_reasons = [], []
-            by_priority = sorted(iteration.iteration_rules, key=priority_getter)
-            for _, rule_group in groupby(by_priority, key=priority_getter):
-                status, group_actionable, group_exclusions, is_rule_stop = self.evaluate_priority_group(
-                    rule_group, worst_status
-                )
-                # Merge results
-                worst_status = status
-                actionable_reasons.extend(group_actionable)
-                exclusion_reasons.extend(group_exclusions)
-                if is_rule_stop:
+        for _, rule_group in groupby(sorted_rules_by_priority, key=priority_getter):
+            status, group_actionable, group_exclusions, rule_stop = self.evaluate_rules_priority_group(rule_group)
+            if status.is_exclusion:
+                if cohort.cohort_label is not None:
+                    cohort_results[str(cohort.cohort_label)] = CohortResult(
+                        cohort.cohort_group if cohort.cohort_group else cohort.cohort_label,
+                        Status.not_eligible,
+                        [],
+                        str(cohort.negative_description),
+                    )
+                is_eligible = False
+                break
+        return is_eligible
+
+    def evaluate_suppression_rules(
+        self,
+        cohort: IterationCohort,
+        rules_suppression: Iterable[rules.IterationRule],
+        *,
+        is_actionable: bool,
+    ) -> tuple[list, bool]:
+        priority_getter = attrgetter("priority")
+        suppression_reasons = []
+        sorted_rules_by_priority = sorted(self.get_exclusion_rules(cohort, rules_suppression), key=priority_getter)
+        for _, rule_group in groupby(sorted_rules_by_priority, key=priority_getter):
+            status, group_actionable, group_exclusions, rule_stop = self.evaluate_rules_priority_group(rule_group)
+            if status.is_exclusion:
+                is_actionable = False
+                suppression_reasons.extend(group_exclusions)
+                if rule_stop:
                     break
-            condition_status_entry = status_with_reasons.setdefault(worst_status, [])
-            condition_status_entry.extend(
-                actionable_reasons if worst_status is eligibility.Status.actionable else exclusion_reasons
-            )
-
-        best_status = eligibility.Status.best(*list(status_with_reasons.keys()))
-
-        return best_status, status_with_reasons[best_status]
+        return suppression_reasons, is_actionable
 
-    def evaluate_priority_group(
-        self,
-        iteration_rule_group: Iterator[rules.IterationRule],
-        worst_status_so_far_for_condition: eligibility.Status,
+    def evaluate_rules_priority_group(
+        self, rules_group: Iterator[rules.IterationRule]
     ) -> tuple[eligibility.Status, list[eligibility.Reason], list[eligibility.Reason], bool]:
         is_rule_stop = False
-        exclusion_reasons, actionable_reasons = [], []
-        exclude_capable_rules = [
-            ir
-            for ir in iteration_rule_group
-            if ir.type in (rules.RuleType.filter, rules.RuleType.suppression)
-            and (ir.cohort_label is None or (ir.cohort_label in self.person_cohorts))
-        ]
-
-        best_status = eligibility.Status.not_eligible if exclude_capable_rules else eligibility.Status.actionable
+        actionable_reasons, exclusion_reasons = [], []
+        best_status = eligibility.Status.not_eligible
 
-        for rule in exclude_capable_rules:
+        for rule in rules_group:
+            is_rule_stop = rule.rule_stop or is_rule_stop
             rule_calculator = RuleCalculator(person_data=self.person_data, rule=rule)
             status, reason = rule_calculator.evaluate_exclusion()
             if status.is_exclusion:
@@ -150,7 +231,4 @@ def evaluate_priority_group(
                 best_status = eligibility.Status.actionable
                 actionable_reasons.append(reason)
 
-        worst_group_status = eligibility.Status.worst(best_status, worst_status_so_far_for_condition)
-        if worst_group_status.is_exclusion:
-            is_rule_stop = any(rule.rule_stop for rule in exclude_capable_rules)
-        return worst_group_status, actionable_reasons, exclusion_reasons, is_rule_stop
+        return best_status, actionable_reasons, exclusion_reasons, is_rule_stop

src/eligibility_signposting_api/services/calculators/rule_calculator.py

@@ -24,11 +24,7 @@ def evaluate_exclusion(self) -> tuple[eligibility.Status, eligibility.Reason]:
         reason = eligibility.Reason(
             rule_name=eligibility.RuleName(self.rule.name),
             rule_type=eligibility.RuleType(self.rule.type),
-            rule_result=eligibility.RuleResult(
-                f"Rule {self.rule.name!r} ({self.rule.description!r}) "
-                f"{'' if status.is_exclusion else 'not '}excluding - "
-                f"{self.rule.attribute_name!r} {self.rule.comparator!r} {reason}"
-            ),
+            rule_result=eligibility.RuleResult(self.rule.description),
         )
         return status, reason
 
@@ -39,24 +35,30 @@ def get_attribute_value(self) -> str | None:
                 person: Mapping[str, str | None] | None = next(
                     (r for r in self.person_data if r.get("ATTRIBUTE_TYPE", "") == "PERSON"), None
                 )
-                attribute_value = person.get(self.rule.attribute_name) if person else None
+                attribute_value = person.get(str(self.rule.attribute_name)) if person else None
             case rules.RuleAttributeLevel.COHORT:
                 cohorts: Mapping[str, str | None] | None = next(
                     (r for r in self.person_data if r.get("ATTRIBUTE_TYPE", "") == "COHORTS"), None
                 )
-                if self.rule.attribute_name == "COHORT_LABEL":
-                    cohort_map = self.get_value(cohorts, "COHORT_MAP")
+                if cohorts:
+                    attr_name = (
+                        "COHORT_MAP"
+                        if not self.rule.attribute_name or self.rule.attribute_name == "COHORT_LABEL"
+                        else self.rule.attribute_name
+                    )
+                    cohort_map = self.get_value(cohorts, attr_name)
                     cohorts_dict = self.get_value(cohort_map, "cohorts")
                     m_dict = self.get_value(cohorts_dict, "M")
                     person_cohorts: set[str] = set(m_dict.keys())
                     attribute_value = ",".join(person_cohorts)
                 else:
-                    attribute_value = cohorts.get(self.rule.attribute_name) if cohorts else None
+                    attribute_value = None
+
             case rules.RuleAttributeLevel.TARGET:
                 target: Mapping[str, str | None] | None = next(
                     (r for r in self.person_data if r.get("ATTRIBUTE_TYPE", "") == self.rule.attribute_target), None
                 )
-                attribute_value = target.get(self.rule.attribute_name) if target else None
+                attribute_value = target.get(str(self.rule.attribute_name)) if target else None
             case _:  # pragma: no cover
                 msg = f"{self.rule.attribute_level} not implemented"
                 raise NotImplementedError(msg)