Refactored to use ChangePointDetector and JacobianSignal

Author: joseph-hellerstein

src/change_point_detector.py

@@ -104,7 +104,7 @@ def _find_best_split(self, start: int, end: int, min_segment_length: int = 1) ->
         return best_k, reduction
 
     def fit(self, max_change_point: int, min_fractional_reduction: float,
-            min_segment_length: int = 1) -> None:
+            min_subsequence_length: int = 1) -> None:
         """
         Find change points based on the provided stopping criteria.
 
@@ -126,7 +126,7 @@ def fit(self, max_change_point: int, min_fractional_reduction: float,
         # largest reduction first so the fixed-K budget is spent optimally.
         # Each entry: (-reduction, start, end, k) — negated for min-heap ordering.
         heap: List[Tuple[float, int, int, int]] = []
-        k_init, red_init = self._find_best_split(0, self.length, min_segment_length)
+        k_init, red_init = self._find_best_split(0, self.length, min_subsequence_length)
         if red_init > 0:
             heapq.heappush(heap, (-red_init, 0, self.length, k_init))
 
@@ -135,7 +135,7 @@ def fit(self, max_change_point: int, min_fractional_reduction: float,
 
         while heap and len(change_points) < max_change_point:
             neg_red, start, end, k = heapq.heappop(heap)
-            if end - start < 2 * min_segment_length:
+            if end - start < 2 * min_subsequence_length:
                 continue  # Skip segments that are too short to split into two min_segment_length pieces
             reduction = -neg_red
 
@@ -146,8 +146,8 @@ def fit(self, max_change_point: int, min_fractional_reduction: float,
             total_abs_reduction += reduction
 
             for seg_start, seg_end in ((start, k), (k, end)):
-                if seg_end - seg_start >= 2 * min_segment_length:
-                    k_sub, red_sub = self._find_best_split(seg_start, seg_end, min_segment_length)
+                if seg_end - seg_start >= 2 * min_subsequence_length:
+                    k_sub, red_sub = self._find_best_split(seg_start, seg_end, min_subsequence_length)
                     if red_sub > 0:
                         heapq.heappush(heap, (-red_sub, seg_start, seg_end, k_sub))
 

src/jacobian_signal.py

@@ -71,24 +71,36 @@ def _denan(self, jacobian_arr: np.ndarray) -> np.ndarray:
         """Replace NaN entries in a Jacobian with zeros."""
         return np.where(np.isnan(jacobian_arr), 0.0, jacobian_arr)
 
-    def makeDetector(self, max_change_point: int,
+    def fit(self, max_change_point: int,
             min_fractional_reduction: float,
-            min_segment_length: int) -> ChangePointDetector:
-        """fit() step 4. signal_arr[k] is the signal for split index k+1
-        (the time-grid index at which a new segment would begin).
+            min_subsequence_length: int) -> ChangePointDetector:
+        """
+        Fit a ChangePointDetector to the Jacobian signal.
+
+        Parameters
+        ----------
+        max_change_point : int
+            The maximum number of change points to detect.
+        min_fractional_reduction : float
+            The minimum fractional reduction in the adjusted sum of squares required to accept a change point.
+        min_subsequence_length : int
+            The minimum length of a subsequence between change points.
 
-        Returns a sorted (by time) list of accepted interior split indices.
+        Returns
+        -------
+        ChangePointDetector
+            A fitted ChangePointDetector object containing the detected change points and associated information.
         """
         detector = ChangePointDetector(self.signal_arr)
         detector.fit(max_change_point=max_change_point,
                 min_fractional_reduction=min_fractional_reduction,
-                min_segment_length=min_segment_length)
+                min_subsequence_length=min_subsequence_length)
         return detector
 
     def plot(self,
             max_change_point: int,
             min_fractional_reduction: float,
-            min_segment_length: int,
+            min_subsequence_length: int,
             **plt_kwargs: Any) -> List[PlotOptions]:
         """
         Two-panel plot.
@@ -108,9 +120,9 @@ def plot(self,
             Wraps the figure and the bottom axes.  Call ``plt.show()`` or
             ``po.fig.savefig(...)`` on the returned object as needed.
         """
-        detector = self.makeDetector(max_change_point=max_change_point,
+        detector = self.fit(max_change_point=max_change_point,
                 min_fractional_reduction=min_fractional_reduction,
-                min_segment_length=min_segment_length)
+                min_subsequence_length=min_subsequence_length)
         change_point_idxs = [i.splice_start for i in detector.subsequences[1:]]
         change_point_times = [self._timecourse_df.index[i+1] 
                 for i in change_point_idxs]