Added filters for Lynch incident episode and Lynch due date change reason using symbolic logic and subject-aware comparison

Author: Andyg79

classes/lynch_due_date_reason_type.py

@@ -0,0 +1,24 @@
+class LynchDueDateReasonType:
+    """
+    Maps Lynch surveillance due date reason descriptions to IDs and symbolic types.
+    """
+
+    NULL = "null"
+    NOT_NULL = "not_null"
+    UNCHANGED = "unchanged"
+
+    _label_to_id = {
+        "holiday": 9801,
+        "clinical request": 9802,
+        "external delay": 9803,
+        # Extend as needed
+    }
+
+    @classmethod
+    def from_description(cls, description: str):
+        key = description.strip().lower()
+        if key in {cls.NULL, cls.NOT_NULL, cls.UNCHANGED}:
+            return key
+        if key in cls._label_to_id:
+            return cls._label_to_id[key]
+        raise ValueError(f"Unknown Lynch due date change reason: '{description}'")

classes/lynch_incident_episode_type.py

@@ -0,0 +1,20 @@
+class LynchIncidentEpisodeType:
+    """
+    Maps symbolic values used to filter Lynch incident episode linkage.
+    """
+
+    NULL = "null"
+    NOT_NULL = "not_null"
+    LATEST_EPISODE = "latest_episode"
+    EARLIER_EPISODE = "earlier_episode"
+
+    _symbolics = {NULL, NOT_NULL, LATEST_EPISODE, EARLIER_EPISODE}
+
+    @classmethod
+    def from_description(cls, description: str) -> str:
+        key = description.strip().lower()
+        if key not in cls._symbolics:
+            raise ValueError(
+                f"Unknown Lynch incident episode criteria: '{description}'"
+            )
+        return key

utils/oracle/mock_selection_builder.py

@@ -1,31 +1,39 @@
 # mock_selection_builder.py — Development-only testing harness for criteria logic
 import sys
 import os
+from typing import Optional
 
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
 from classes.selection_builder_exception import SelectionBuilderException
 from classes.subject_selection_criteria_key import SubjectSelectionCriteriaKey
 
 
 # Add helper class stubs below
-class ScreeningReferralType:
+class Subject:
+    def __init__(self, lynch_due_date_change_reason_id):
+        self.lynch_due_date_change_reason_id = lynch_due_date_change_reason_id
+
+
+class LynchIncidentEpisodeType:
     """
-    Maps screening referral descriptions to valid value IDs.
+    Maps symbolic criteria values for Lynch incident episode linkage.
     """
 
-    _label_to_id = {
-        "gp": 9701,
-        "self referral": 9702,
-        "hospital": 9703,
-        # Extend as needed
-    }
+    NULL = "null"
+    NOT_NULL = "not_null"
+    LATEST_EPISODE = "latest_episode"
+    EARLIER_EPISODE = "earlier_episode"
+
+    _symbolics = {NULL, NOT_NULL, LATEST_EPISODE, EARLIER_EPISODE}
 
     @classmethod
-    def get_id(cls, description: str) -> int:
+    def from_description(cls, description: str) -> str:
         key = description.strip().lower()
-        if key not in cls._label_to_id:
-            raise ValueError(f"Unknown screening referral type: '{description}'")
-        return cls._label_to_id[key]
+        if key not in cls._symbolics:
+            raise ValueError(
+                f"Unknown Lynch incident episode criteria: '{description}'"
+            )
+        return key
 
 
 class MockSelectionBuilder:
@@ -71,6 +79,10 @@ def _add_join_to_latest_episode(self) -> None:
         """
         self.sql_from.append("-- JOIN to latest episode placeholder")
 
+    def _force_not_modifier_is_invalid(self):
+        # Placeholder for rule enforcement. No-op in mock builder.
+        pass
+
     def _dataset_source_for_criteria_key(self) -> dict:
         """
         Maps criteria key to dataset table and alias.
@@ -94,19 +106,26 @@ def _add_join_to_surveillance_review(self):
     # Replace this with the one you want to test,
     # then use utils/oracle/test_subject_criteria_dev.py to run your scenarios
 
-    def _add_criteria_screening_referral_type(self) -> None:
+    def _add_criteria_lynch_incident_episode(self) -> None:
         """
-        Filters based on screening referral type ID or null presence.
+        Filters based on linkage to a Lynch incident episode.
         """
         try:
-            column = "xt.screening_referral_type_id"
-            value = self.criteria_value.strip().lower()
+            self._add_join_to_latest_episode()
+            column = "ss.lynch_incident_subject_epis_id"
+            value = LynchIncidentEpisodeType.from_description(self.criteria_value)
 
-            if value == "null":
+            if value == LynchIncidentEpisodeType.NULL:
                 self.sql_where.append(f"AND {column} IS NULL")
-            else:
-                type_id = ScreeningReferralType.get_id(self.criteria_value)
-                self.sql_where.append(f"AND {column} {self.criteria_comparator} {type_id}")
+
+            elif value == LynchIncidentEpisodeType.NOT_NULL:
+                self.sql_where.append(f"AND {column} IS NOT NULL")
+
+            elif value == LynchIncidentEpisodeType.LATEST_EPISODE:
+                self.sql_where.append(f"AND {column} = ep.subject_epis_id")
+
+            elif value == LynchIncidentEpisodeType.EARLIER_EPISODE:
+                self.sql_where.append(f"AND {column} < ep.subject_epis_id")
 
         except Exception:
             raise SelectionBuilderException(self.criteria_key_name, self.criteria_value)

utils/oracle/subject_selection_query_builder.py

@@ -49,6 +49,10 @@
 from classes.episode_result_type import EpisodeResultType
 from classes.symptomatic_procedure_result_type import SymptomaticProcedureResultType
 from classes.screening_referral_type import ScreeningReferralType
+from classes.lynch_due_date_reason_type import LynchDueDateReasonType
+from classes.lynch_incident_episode_type import (
+    LynchIncidentEpisodeType,
+)
 
 
 class SubjectSelectionQueryBuilder:
@@ -2165,6 +2169,67 @@ def _add_criteria_screening_referral_type(self) -> None:
         except Exception:
             raise SelectionBuilderException(self.criteria_key_name, self.criteria_value)
 
+    def _add_criteria_lynch_due_date_reason(
+        self, subject: Optional[Subject] = None
+    ) -> None:
+        """
+        Filters based on Lynch due date change reason. Supports symbolic types and subject comparison.
+        """
+        try:
+            column = "ss.lynch_sdd_reason_for_change_id"
+            reason = LynchDueDateReasonType.from_description(self.criteria_value)
+
+            if reason == LynchDueDateReasonType.NULL:
+                self.sql_where.append(f"AND {column} IS NULL")
+
+            elif reason == LynchDueDateReasonType.NOT_NULL:
+                self.sql_where.append(f"AND {column} IS NOT NULL")
+
+            elif reason == LynchDueDateReasonType.UNCHANGED:
+                if subject is None:
+                    raise SelectionBuilderException(
+                        self.criteria_key_name,
+                        "No subject provided for 'unchanged' logic",
+                    )
+                elif getattr(subject, "lynch_due_date_change_reason_id", None) is None:
+                    self.sql_where.append(f"AND {column} IS NULL")
+                else:
+                    self.sql_where.append(
+                        f"AND {column} = {subject.lynch_due_date_change_reason_id}"
+                    )
+
+            else:
+                self.sql_where.append(
+                    f"AND {column} {self.criteria_comparator} {reason}"
+                )
+
+        except Exception:
+            raise SelectionBuilderException(self.criteria_key_name, self.criteria_value)
+
+    def _add_criteria_lynch_incident_episode(self) -> None:
+        """
+        Filters based on linkage to a Lynch incident episode.
+        """
+        try:
+            self._add_join_to_latest_episode()
+            column = "ss.lynch_incident_subject_epis_id"
+            value = LynchIncidentEpisodeType.from_description(self.criteria_value)
+
+            if value == LynchIncidentEpisodeType.NULL:
+                self.sql_where.append(f"AND {column} IS NULL")
+
+            elif value == LynchIncidentEpisodeType.NOT_NULL:
+                self.sql_where.append(f"AND {column} IS NOT NULL")
+
+            elif value == LynchIncidentEpisodeType.LATEST_EPISODE:
+                self.sql_where.append(f"AND {column} = ep.subject_epis_id")
+
+            elif value == LynchIncidentEpisodeType.EARLIER_EPISODE:
+                self.sql_where.append(f"AND {column} < ep.subject_epis_id")
+
+        except Exception:
+            raise SelectionBuilderException(self.criteria_key_name, self.criteria_value)
+
     # ------------------------------------------------------------------------
     # 🧬 CADS Clinical Dataset Filters
     # ------------------------------------------------------------------------

utils/oracle/test_subject_criteria_dev.py

@@ -41,16 +41,8 @@ def make_builder(key, value, index=0, comparator="="):
     return b
 
 
-# === Test: SCREENING_REFERRAL_TYPE (self referral) ===
-b = make_builder(
-    SubjectSelectionCriteriaKey.SCREENING_REFERRAL_TYPE, "self referral", comparator="="
-)
-b._add_criteria_screening_referral_type()
-print("=== SCREENING_REFERRAL_TYPE (self referral) ===")
-print(b.dump_sql(), end="\n\n")
-
-# === Test: SCREENING_REFERRAL_TYPE (null) ===
-b = make_builder(SubjectSelectionCriteriaKey.SCREENING_REFERRAL_TYPE, "null")
-b._add_criteria_screening_referral_type()
-print("=== SCREENING_REFERRAL_TYPE (null) ===")
+# === Test: LYNCH_INCIDENT_EPISODE (earlier_episode) ===
+b = make_builder(SubjectSelectionCriteriaKey.LYNCH_INCIDENT_EPISODE, "earlier_episode")
+b._add_criteria_lynch_incident_episode()
+print("=== LYNCH_INCIDENT_EPISODE (earlier_episode) ===")
 print(b.dump_sql(), end="\n\n")