cohort grouping & magic cohorts - extensive coverage (#145)

* single description for each cohort group - test, code refactored

* cohort group description based on priority

* refer status of magic cohorts based on other cohorts status

* exclude cohort when description is "" or None

* corrected magic cohorts, made cohort code not null

* integration tests for no rule-text, no cohort description

* cohort deduplication

* actionable rule reasons collection test, code cleanup

* code cleanup

* code cleanup

* lint issues

Author: Karthikeyannhs

src/eligibility_signposting_api/app.py

@@ -8,7 +8,7 @@
 from mangum.types import LambdaContext, LambdaEvent
 
 from eligibility_signposting_api import repos, services
-from eligibility_signposting_api.config import config, init_logging
+from eligibility_signposting_api.config.config import config, init_logging
 from eligibility_signposting_api.error_handler import handle_exception
 from eligibility_signposting_api.views import eligibility_blueprint
 

src/eligibility_signposting_api/config/__init__.py

@@ -0,0 +1 @@
+MAGIC_COHORT_LABEL = "elid_all_people"

src/eligibility_signposting_api/config.py …ibility_signposting_api/config/config.pysrc/eligibility_signposting_api/config.py renamed to src/eligibility_signposting_api/config/config.py src/eligibility_signposting_api/config.py renamed to src/eligibility_signposting_api/config/config.py

@@ -12,7 +12,7 @@
 ConditionName = NewType("ConditionName", str)
 
 RuleName = NewType("RuleName", str)
-RuleResult = NewType("RuleResult", str)
+RuleDescription = NewType("RuleDescription", str)
 
 
 class RuleType(StrEnum):
@@ -60,28 +60,28 @@ def best(*statuses: Status) -> Status:
 class Reason:
     rule_type: RuleType
     rule_name: RuleName
-    rule_result: RuleResult
+    rule_description: RuleDescription | None
 
 
 @dataclass
 class Condition:
     condition_name: ConditionName
     status: Status
-    cohort_results: list[CohortResult]
+    cohort_results: list[CohortGroupResult]
 
 
 @dataclass
-class CohortResult:
+class CohortGroupResult:
     cohort_code: str
     status: Status
     reasons: list[Reason]
-    description: str
+    description: str | None
 
 
 @dataclass
 class IterationResult:
     status: Status
-    cohort_results: list[CohortResult]
+    cohort_results: list[CohortGroupResult]
 
 
 @dataclass

src/eligibility_signposting_api/config/contants.py

@@ -10,6 +10,8 @@
 
 from pydantic import BaseModel, Field, field_serializer, field_validator, model_validator
 
+from eligibility_signposting_api.config.contants import MAGIC_COHORT_LABEL
+
 if typing.TYPE_CHECKING:  # pragma: no cover
     from pydantic import SerializationInfo
 
@@ -85,8 +87,8 @@ class RuleAttributeLevel(StrEnum):
 
 
 class IterationCohort(BaseModel):
-    cohort_label: CohortLabel | None = Field(None, alias="CohortLabel")
-    cohort_group: CohortGroup | None = Field(None, alias="CohortGroup")
+    cohort_label: CohortLabel = Field(alias="CohortLabel")
+    cohort_group: CohortGroup = Field(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")
@@ -107,14 +109,14 @@ class IterationRule(BaseModel):
     attribute_target: RuleAttributeTarget | None = Field(None, alias="AttributeTarget")
     rule_stop: RuleStop = Field(RuleStop(False), alias="RuleStop")  # noqa: FBT003
 
+    model_config = {"populate_by_name": True, "extra": "ignore"}
+
     @field_validator("rule_stop", mode="before")
     def parse_yn_to_bool(cls, v: str | bool) -> bool:  # noqa: N805
         if isinstance(v, str):
             return v.upper() == "Y"
         return v
 
-    model_config = {"populate_by_name": True, "extra": "ignore"}
-
 
 class Iteration(BaseModel):
     id: IterationID = Field(..., alias="ID")
@@ -142,6 +144,17 @@ def parse_dates(cls, v: str | date) -> date:
     def serialize_dates(v: date, _info: SerializationInfo) -> str:
         return v.strftime("%Y%m%d")
 
+    @cached_property
+    def has_magic_cohort(self) -> bool:
+        return next(
+            (
+                True
+                for cc in self.iteration_cohorts
+                if cc.cohort_label and cc.cohort_label.upper() == MAGIC_COHORT_LABEL.upper()
+            ),
+            False,
+        )
+
 
 class CampaignConfig(BaseModel):
     id: CampaignID = Field(..., alias="ID")

src/eligibility_signposting_api/model/eligibility.py

@@ -7,7 +7,7 @@
 from wireup import Inject, service
 from yarl import URL
 
-from eligibility_signposting_api.config import AwsAccessKey, AwsRegion, AwsSecretAccessKey
+from eligibility_signposting_api.config.config import AwsAccessKey, AwsRegion, AwsSecretAccessKey
 
 logger = logging.getLogger(__name__)
 

src/eligibility_signposting_api/model/rules.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from _operator import attrgetter
+from collections import defaultdict
 from collections.abc import Collection, Iterable, Iterator, Mapping
 from dataclasses import dataclass, field
 from itertools import groupby
@@ -13,16 +14,17 @@
 
 from eligibility_signposting_api.model import eligibility, rules
 from eligibility_signposting_api.model.eligibility import (
-    CohortResult,
+    CohortGroupResult,
     Condition,
     ConditionName,
     IterationResult,
     Status,
 )
-from eligibility_signposting_api.services.calculators.rule_calculator import RuleCalculator
+from eligibility_signposting_api.services.calculators.rule_calculator import (
+    RuleCalculator,
+)
 
 Row = Collection[Mapping[str, Any]]
-magic_cohort = "elid_all_people"
 
 
 @service
@@ -49,23 +51,39 @@ def campaigns_grouped_by_condition_name(
     ) -> 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")
+            sorted(self.active_campaigns, key=attrgetter("target")),
+            key=attrgetter("target"),
         ):
             yield condition_name, list(campaign_group)
 
     @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"), {}
+            (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]]:
+    def get_the_best_cohort_memberships(
+        cohort_results: dict[str, CohortGroupResult],
+    ) -> tuple[Status, list[CohortGroupResult]]:
         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]
+
+        best_cohorts = [
+            CohortGroupResult(
+                cohort_code=cc.cohort_code,
+                status=cc.status,
+                reasons=cc.reasons,
+                description=(cc.description or "").strip() if cc.description else "",
+            )
+            for cc in best_cohorts
+        ]
+
         return best_status, best_cohorts
 
     @staticmethod
@@ -98,28 +116,30 @@ def evaluate_eligibility(self) -> eligibility.EligibilityStatus:
             iteration_results: dict[str, IterationResult] = {}
 
             for active_iteration in [cc.current_iteration for cc in campaign_group]:
-                cohort_results: dict[str, CohortResult] = {}
+                cohort_results: dict[str, CohortGroupResult] = {}
 
                 filter_rules, suppression_rules = self.get_rules_by_type(active_iteration)
+
                 for cohort in sorted(active_iteration.iteration_cohorts, key=attrgetter("priority")):
                     # Base Eligibility - check
-                    if cohort.cohort_label in self.person_cohorts or cohort.cohort_label == magic_cohort:
+                    if cohort.cohort_label in self.person_cohorts or active_iteration.has_magic_cohort:
                         # Eligibility - check
                         if self.is_eligible_by_filter_rules(cohort, cohort_results, filter_rules):
                             # Actionability - evaluation
                             self.evaluate_suppression_rules(cohort, cohort_results, suppression_rules)
 
                     # 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,
+                        cohort_results[cohort.cohort_label] = CohortGroupResult(
+                            (cohort.cohort_group),
                             Status.not_eligible,
                             [],
-                            str(cohort.negative_description),
+                            cohort.negative_description,
                         )
 
                 # Determine Result between cohorts - get the best
-                status, best_cohorts = self.get_best_cohort(cohort_results)
+                status, best_cohorts = self.get_the_best_cohort_memberships(cohort_results)
+
                 iteration_results[active_iteration.name] = IterationResult(status, best_cohorts)
 
             # Determine results between iterations - get the best
@@ -130,20 +150,50 @@ def evaluate_eligibility(self) -> eligibility.EligibilityStatus:
             condition_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 condition_results.items()
-        ]
+        final_result = self.build_condition_results(condition_results)
         return eligibility.EligibilityStatus(conditions=final_result)
 
+    @staticmethod
+    def build_condition_results(
+        condition_results: dict[ConditionName, IterationResult],
+    ) -> list[Condition]:
+        conditions: list[Condition] = []
+        # iterate over conditions
+        for condition_name, active_iteration_result in condition_results.items():
+            grouped_cohort_results = defaultdict(list)
+            # iterate over cohorts and group them by status and cohort_group
+            for cohort_result in active_iteration_result.cohort_results:
+                if active_iteration_result.status == cohort_result.status:
+                    grouped_cohort_results[cohort_result.cohort_code].append(cohort_result)
+
+            # deduplicate grouped cohort results by cohort_code
+            deduplicated_cohort_results = [
+                CohortGroupResult(
+                    cohort_code=group_cohort_code,
+                    status=group[0].status,
+                    # Flatten all reasons from the group
+                    reasons=[reason for cohort in group for reason in cohort.reasons],
+                    # get the first nonempty description
+                    description=next((c.description for c in group if c.description), group[0].description),
+                )
+                for group_cohort_code, group in grouped_cohort_results.items()
+                if group
+            ]
+
+            # return condition with cohort results
+            conditions.append(
+                Condition(
+                    condition_name=condition_name,
+                    status=active_iteration_result.status,
+                    cohort_results=list(deduplicated_cohort_results),
+                )
+            )
+        return conditions
+
     def is_eligible_by_filter_rules(
         self,
         cohort: IterationCohort,
-        cohort_results: dict[str, CohortResult],
+        cohort_results: dict[str, CohortGroupResult],
         filter_rules: Iterable[rules.IterationRule],
     ) -> bool:
         is_eligible = True
@@ -156,11 +206,11 @@ def is_eligible_by_filter_rules(
             )
             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,
+                    cohort_results[cohort.cohort_label] = CohortGroupResult(
+                        (cohort.cohort_group),
                         Status.not_eligible,
                         [],
-                        str(cohort.negative_description),
+                        cohort.negative_description,
                     )
                 is_eligible = False
                 break
@@ -169,7 +219,7 @@ def is_eligible_by_filter_rules(
     def evaluate_suppression_rules(
         self,
         cohort: IterationCohort,
-        cohort_results: dict[str, CohortResult],
+        cohort_results: dict[str, CohortGroupResult],
         suppression_rules: Iterable[rules.IterationRule],
     ) -> None:
         is_actionable: bool = True
@@ -191,18 +241,18 @@ def evaluate_suppression_rules(
         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,
+                cohort_results[key] = CohortGroupResult(
+                    cohort.cohort_group,
                     Status.actionable,
                     [],
-                    str(cohort.positive_description),
+                    cohort.positive_description,
                 )
             else:
-                cohort_results[key] = CohortResult(
-                    cohort.cohort_group if cohort.cohort_group else key,
+                cohort_results[key] = CohortGroupResult(
+                    cohort.cohort_group,
                     Status.not_actionable,
                     suppression_reasons,
-                    str(cohort.positive_description),
+                    cohort.positive_description,
                 )
 
     def evaluate_rules_priority_group(

src/eligibility_signposting_api/repos/factory.py

@@ -24,7 +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(self.rule.description),
+            rule_description=eligibility.RuleDescription(self.rule.description),
         )
         return status, reason
 

src/eligibility_signposting_api/services/calculators/eligibility_calculator.py

@@ -1,6 +1,5 @@
 import logging
 import uuid
-from collections import defaultdict
 from datetime import UTC, datetime
 from http import HTTPStatus
 
@@ -72,24 +71,19 @@ def build_eligibility_response(eligibility_status: EligibilityStatus) -> eligibi
 def build_eligibility_cohorts(condition: Condition) -> list[eligibility.EligibilityCohort]:
     """Group Iteration cohorts and make only one entry per cohort group"""
 
-    grouped_cohort_results = defaultdict(list)
-
-    for cohort_result in condition.cohort_results:
-        if condition.status == cohort_result.status:
-            grouped_cohort_results[cohort_result.cohort_code].append(cohort_result)
-
     return [
         eligibility.EligibilityCohort(
-            cohortCode=cohort_group_code,
-            cohortText=cohort_group[0].description,
-            cohortStatus=STATUS_MAPPING[cohort_group[0].status],
+            cohortCode=eligibility.CohortCode(cohort_result.cohort_code),
+            cohortText=eligibility.CohortText(cohort_result.description),
+            cohortStatus=STATUS_MAPPING[cohort_result.status],
         )
-        for cohort_group_code, cohort_group in grouped_cohort_results.items()
-        if cohort_group
+        for cohort_result in condition.cohort_results
+        if cohort_result and condition.status == cohort_result.status and cohort_result.description
     ]
 
 
 def build_suitability_results(condition: Condition) -> list[eligibility.SuitabilityRule]:
+    """Make only one entry if there are duplicate rules"""
     if condition.status != Status.not_actionable:
         return []
 
@@ -99,13 +93,13 @@ def build_suitability_results(condition: Condition) -> list[eligibility.Suitabil
     for cohort_result in condition.cohort_results:
         if cohort_result.status == Status.not_actionable:
             for reason in cohort_result.reasons:
-                if reason.rule_name not in unique_rule_codes:
+                if reason.rule_name not in unique_rule_codes and reason.rule_description:
                     unique_rule_codes.add(reason.rule_name)
                     suitability_results.append(
                         eligibility.SuitabilityRule(
                             ruleType=eligibility.RuleType(reason.rule_type.value),
                             ruleCode=eligibility.RuleCode(reason.rule_name),
-                            ruleText=eligibility.RuleText(reason.rule_result),
+                            ruleText=eligibility.RuleText(reason.rule_description),
                         )
                     )