code: def _compute_PI(
      T_A,
      T_B,
      m,
      T_A_subseq_isfinite,
      T_B_subseq_isfinite,
      p,
      indices,
      start,
      stop,
      thread_idx,
      s,
      P_NORM,
      I,
      excl_zone,
  ):
      """
      Compute (Numba JIT-compiled) and update the (elementwise) p-th power of
      matrix profile, and matrix profile indces according to the non-normalized
      (i.e., without z-normalization) preSCRIMP algorithm.
  
      Parameters
      ----------
      T_A : numpy.ndarray
          The time series or sequence for which to compute the matrix profile
  
      T_B : numpy.ndarray
          The time series or sequence that will be used to annotate T_A. For every
          subsequence in T_A, its nearest neighbor in T_B will be recorded.
  
      m : int
          Window size
  
      T_A_subseq_isfinite : numpy.ndarray
          A boolean array that indicates whether a subsequence in `T_A` contains a
          `np.nan`/`np.inf` value (False)
  
      T_B_subseq_isfinite : numpy.ndarray
          A boolean array that indicates whether a subsequence in `T_B` contains a
          `np.nan`/`np.inf` value (False)
  
      p : float, default 2.0
          The p-norm to apply for computing the Minkowski distance. Minkowski distance is
          typically used with `p` being 1 or 2, which correspond to the Manhattan distance
          and the Euclidean distance, respectively.
  
      indices : numpy.ndarray
          The subsequence indices to compute `prescrump` for
  
      start : int
          The (inclusive) start index for `indices`
  
      stop : int
          The (exclusive) stop index for `indices`
  
      thread_idx : int
          The thread index
  
      s : int
          The sampling interval that defaults to
          `int(np.ceil(m / config.STUMPY_EXCL_ZONE_DENOM))`
  
      P_NORM : numpy.ndarray
          The (elementwises) p-th power of matrix profile
  
      I : numpy.ndarray
          The matrix profile indices
  
      excl_zone : int
          The half width for the exclusion zone relative to the `i`.
  
      Returns
      -------
      None
      """
      l = T_A.shape[0] - m + 1  # length of matrix profile
      w = T_B.shape[0] - m + 1  # length of each distance profile
  
      p_norm_profile = np.empty(w)
      for i in indices[start:stop]:
          if not T_A_subseq_isfinite[i]:  # pragma: no cover
              p_norm_profile[:] = np.inf
          else:
              p_norm_profile[:] = core._p_norm_distance_profile(
                  T_A[i : i + m], T_B, p=p
              )
              p_norm_profile[
                  p_norm_profile < config.STUMPY_P_NORM_THRESHOLD
              ] = 0.0
              p_norm_profile[~T_B_subseq_isfinite] = np.inf
          # Update P[i] relative to all T[j : j + m]
          if excl_zone is not None:
              core._apply_exclusion_zone(
                  p_norm_profile, i, excl_zone, np.inf
              )
  
          nn_i = np.argmin(p_norm_profile)
          if p_norm_profile[nn_i] == np.inf:
              continue
  
          if (
              p_norm_profile[nn_i] < P_NORM[thread_idx, i, -1]
              and nn_i not in I[thread_idx, i]
          ):
              idx = np.searchsorted(
                  P_NORM[thread_idx, i],
                  p_norm_profile[nn_i],
                  side="right",
              )
              core._shift_insert_at_index(
                  P_NORM[thread_idx, i], idx, p_norm_profile[nn_i]
              )
              core._shift_insert_at_index(I[thread_idx, i], idx, nn_i)
  
          j = nn_i
          p_norm_j = P_NORM[thread_idx, i, 0]
          p_norm_j_prime = p_norm_j
          for g in range(1, min(s, l - i, w - j)):
              p_norm_j = (
                  p_norm_j
                  - abs(T_B[j + g - 1] - T_A[i + g - 1]) ** p
                  + abs(T_B[j + g + m - 1] - T_A[i + g + m - 1]) ** p
              )
              if (
                  not T_A_subseq_isfinite[i + g]
                  or not T_B_subseq_isfinite[j + g]
              ):  # pragma: no cover
                  p_norm = np.inf
              else:
                  p_norm = p_norm_j
                  if (
                      p_norm < config.STUMPY_P_NORM_THRESHOLD
                  ):  # pragma: no cover
                      p_norm = 0.0
  
              if (
                  p_norm < P_NORM[thread_idx, i + g, -1]
                  and (j + g) not in I[thread_idx, i + g]
              ):
                  idx = np.searchsorted(
                      P_NORM[thread_idx, i + g], p_norm, side="right"
                  )
                  core._shift_insert_at_index(
                      P_NORM[thread_idx, i + g], idx, p_norm
                  )
                  core._shift_insert_at_index(
                      I[thread_idx, i + g], idx, j + g
                  )
  
              if (
                  excl_zone is not None
                  and p_norm < P_NORM[thread_idx, j + g, -1]
                  and (i + g) not in I[thread_idx, j + g]
              ):
                  idx = np.searchsorted(
                      P_NORM[thread_idx, j + g], p_norm, side="right"
                  )
                  core._shift_insert_at_index(
                      P_NORM[thread_idx, j + g], idx, p_norm
                  )
                  core._shift_insert_at_index(
                      I[thread_idx, j + g], idx, i + g
                  )
  
          p_norm_j = p_norm_j_prime
          for g in range(1, min(s, i + 1, j + 1)):
              p_norm_j = (
                  p_norm_j
                  - abs(T_B[j - g + m] - T_A[i - g + m]) ** p
                  + abs(T_B[j - g] - T_A[i - g]) ** p
              )
              if (
                  not T_A_subseq_isfinite[i - g]
                  or not T_B_subseq_isfinite[j - g]
              ):  # pragma: no cover
                  p_norm = np.inf
              else:
                  p_norm = p_norm_j
                  if (
                      p_norm < config.STUMPY_P_NORM_THRESHOLD
                  ):  # pragma: no cover
                      p_norm = 0.0
  
              if (
                  p_norm < P_NORM[thread_idx, i - g, -1]
                  and (j - g) not in I[thread_idx, i - g]
              ):
                  idx = np.searchsorted(
                      P_NORM[thread_idx, i - g], p_norm, side="right"
                  )
                  core._shift_insert_at_index(
                      P_NORM[thread_idx, i - g], idx, p_norm
                  )
                  core._shift_insert_at_index(
                      I[thread_idx, i - g], idx, j - g
                  )
  
              if (
                  excl_zone is not None
                  and p_norm < P_NORM[thread_idx, j - g, -1]
                  and (i - g) not in I[thread_idx, j - g]
              ):
                  idx = np.searchsorted(
                      P_NORM[thread_idx, j - g], p_norm, side="right"
                  )
                  core._shift_insert_at_index(
                      P_NORM[thread_idx, j - g], idx, p_norm
                  )
                  core._shift_insert_at_index(
                      I[thread_idx, j - g], idx, i - g
                  )
  
          # In the case of a self-join, the calculated profile can also be used
          # to refine the top-k for all non-trivial subsequences
          if excl_zone is not None:
              # Note that `p_norm_profile[j]`, the distance between subsequences
              # `S_i = T[i : i + m]` and `S_j = T[j : j + m]` can be used to update
              # the top-k for BOTH subsequence `i` and subsequence `j`. We update
              # the latter here.
  
              indices = np.flatnonzero(
                  p_norm_profile < P_NORM[thread_idx, :, -1]
              )
              for j in indices:
                  if i not in I[thread_idx, j]:
                      idx = np.searchsorted(
                          P_NORM[thread_idx, j],
                          p_norm_profile[j],
                          side="right",
                      )
                      core._shift_insert_at_index(
                          P_NORM[thread_idx, j], idx, p_norm_profile[j]
                      )
                      core._shift_insert_at_index(
                          I[thread_idx, j], idx, i
                      )
  
file_path: stumpy/stumpy/scraamp.py
imports: import numba
  import numpy
  from numba import njit, prange
  from .None import config, core
  from .aamp import _aamp
label_desc: Use image node_id 2 for calling a global function with example usage: _compute_PI(T_A, T_B, m, T_A_subseq_isfinite, T_B_subseq_isfinite, p, indices, start, stop, thread_idx, s, P_NORM, I, excl_zone) without return types