Added FOBT prevalent/incident episode status filter using symbolic NULL checks

Author: Andyg79

classes/prevalent_incident_status_type.py

@@ -0,0 +1,16 @@
+class PrevalentIncidentStatusType:
+    """
+    Maps symbolic values for FOBT prevalent/incident episode classification.
+    """
+
+    PREVALENT = "prevalent"
+    INCIDENT = "incident"
+
+    _valid_values = {PREVALENT, INCIDENT}
+
+    @classmethod
+    def from_description(cls, description: str) -> str:
+        key = description.strip().lower()
+        if key not in cls._valid_values:
+            raise ValueError(f"Unknown FOBT episode status: '{description}'")
+        return key

utils/oracle/mock_selection_builder.py

@@ -14,25 +14,21 @@ def __init__(self, lynch_due_date_change_reason_id):
         self.lynch_due_date_change_reason_id = lynch_due_date_change_reason_id
 
 
-class LynchIncidentEpisodeType:
+class PrevalentIncidentStatusType:
     """
-    Maps symbolic criteria values for Lynch incident episode linkage.
+    Maps symbolic values for FOBT prevalent/incident episode classification.
     """
 
-    NULL = "null"
-    NOT_NULL = "not_null"
-    LATEST_EPISODE = "latest_episode"
-    EARLIER_EPISODE = "earlier_episode"
+    PREVALENT = "prevalent"
+    INCIDENT = "incident"
 
-    _symbolics = {NULL, NOT_NULL, LATEST_EPISODE, EARLIER_EPISODE}
+    _valid_values = {PREVALENT, INCIDENT}
 
     @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}'"
-            )
+        if key not in cls._valid_values:
+            raise ValueError(f"Unknown FOBT episode status: '{description}'")
         return key
 
 
@@ -106,26 +102,18 @@ 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_lynch_incident_episode(self) -> None:
+    def _add_criteria_fobt_prevalent_incident_status(self) -> None:
         """
-        Filters based on linkage to a Lynch incident episode.
+        Filters subjects by whether their FOBT episode is prevalent or incident.
         """
         try:
-            self._add_join_to_latest_episode()
-            column = "ss.lynch_incident_subject_epis_id"
-            value = LynchIncidentEpisodeType.from_description(self.criteria_value)
+            value = PrevalentIncidentStatusType.from_description(self.criteria_value)
+            column = "ss.fobt_incident_subject_epis_id"
 
-            if value == LynchIncidentEpisodeType.NULL:
+            if value == PrevalentIncidentStatusType.PREVALENT:
                 self.sql_where.append(f"AND {column} IS NULL")
-
-            elif value == LynchIncidentEpisodeType.NOT_NULL:
+            elif value == PrevalentIncidentStatusType.INCIDENT:
                 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

@@ -53,6 +53,7 @@
 from classes.lynch_incident_episode_type import (
     LynchIncidentEpisodeType,
 )
+from classes.prevalent_incident_status_type import PrevalentIncidentStatusType
 
 
 class SubjectSelectionQueryBuilder:
@@ -2230,6 +2231,22 @@ def _add_criteria_lynch_incident_episode(self) -> None:
         except Exception:
             raise SelectionBuilderException(self.criteria_key_name, self.criteria_value)
 
+    def _add_criteria_fobt_prevalent_incident_status(self) -> None:
+        """
+        Filters subjects by whether their FOBT episode is prevalent or incident.
+        """
+        try:
+            value = PrevalentIncidentStatusType.from_description(self.criteria_value)
+            column = "ss.fobt_incident_subject_epis_id"
+
+            if value == PrevalentIncidentStatusType.PREVALENT:
+                self.sql_where.append(f"AND {column} IS NULL")
+            elif value == PrevalentIncidentStatusType.INCIDENT:
+                self.sql_where.append(f"AND {column} IS NOT NULL")
+
+        except Exception:
+            raise SelectionBuilderException(self.criteria_key_name, self.criteria_value)
+
     # ------------------------------------------------------------------------
     # 🧬 CADS Clinical Dataset Filters
     # ------------------------------------------------------------------------

utils/oracle/test_subject_criteria_dev.py

@@ -41,8 +41,8 @@ def make_builder(key, value, index=0, comparator="="):
     return b
 
 
-# === 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) ===")
+# === Test: FOBT_PREVALENT_INCIDENT_STATUS (incident) ===
+b = make_builder(SubjectSelectionCriteriaKey.FOBT_PREVALENT_INCIDENT_STATUS, "incident")
+b._add_criteria_fobt_prevalent_incident_status()
+print("=== FOBT_PREVALENT_INCIDENT_STATUS (incident) ===")
 print(b.dump_sql(), end="\n\n")