code: def _dask_stumped(
      dask_client,
      T_A,
      T_B,
      m,
      M_T,
      μ_Q,
      Σ_T_inverse,
      σ_Q_inverse,
      M_T_m_1,
      μ_Q_m_1,
      T_A_subseq_isfinite,
      T_B_subseq_isfinite,
      T_A_subseq_isconstant,
      T_B_subseq_isconstant,
      diags,
      ignore_trivial,
      k,
  ):
      """
      Compute the z-normalized (top-k) matrix profile with a distributed dask cluster
  
      This is a highly distributed implementation around the Numba JIT-compiled
      parallelized `_stump` function which computes the (top-k) matrix profile according
      to STOMPopt with Pearson correlations.
  
      Parameters
      ----------
      dask_client : client
          A Dask Distributed client. Setting up a distributed cluster is beyond
          the scope of this library. Please refer to the Dask Distributed
          documentation.
  
      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. Default is
          `None` which corresponds to a self-join.
  
      m : int
          Window size
  
      M_T : numpy.ndarray
          Sliding mean of time series, `T`
  
      μ_Q : numpy.ndarray
          Mean of the query sequence, `Q`, relative to the current sliding window
  
      Σ_T_inverse : numpy.ndarray
          Inverse sliding standard deviation of time series, `T`
  
      σ_Q_inverse : numpy.ndarray
          Inverse standard deviation of the query sequence, `Q`, relative to the current
  
      M_T_m_1 : numpy.ndarray
          Sliding mean of time series, `T`, using a window size of `m-1`
  
      μ_Q_m_1 : numpy.ndarray
          Mean of the query sequence, `Q`, relative to the current sliding window and
          using a window size of `m-1`
  
      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)
  
      T_A_subseq_isconstant : numpy.ndarray
          A boolean array that indicates whether a subsequence in `T_A` is constant (True)
  
      T_B_subseq_isconstant : numpy.ndarray
          A boolean array that indicates whether a subsequence in `T_B` is constant (True)
  
      diags : numpy.ndarray
          The diagonal indices
  
      ignore_trivial : bool, default True
          Set to `True` if this is a self-join. Otherwise, for AB-join, set this
          to `False`. Default is `True`.
  
      k : int, default 1
          The number of top `k` smallest distances used to construct the matrix profile.
          Note that this will increase the total computational time and memory usage
          when k > 1. If you have access to a GPU device, then you may be able to
          leverage `gpu_stump` for better performance and scalability.
  
      Returns
      -------
      out : numpy.ndarray
          When k = 1 (default), the first column consists of the matrix profile,
          the second column consists of the matrix profile indices, the third column
          consists of the left matrix profile indices, and the fourth column consists
          of the right matrix profile indices. However, when k > 1, the output array
          will contain exactly 2 * k + 2 columns. The first k columns (i.e., out[:, :k])
          consists of the top-k matrix profile, the next set of k columns
          (i.e., out[:, k:2k]) consists of the corresponding top-k matrix profile
          indices, and the last two columns (i.e., out[:, 2k] and out[:, 2k+1] or,
          equivalently, out[:, -2] and out[:, -1]) correspond to the top-1 left
          matrix profile indices and the top-1 right matrix profile indices, respectively.
      """
      n_A = T_A.shape[0]
      n_B = T_B.shape[0]
      l = n_A - m + 1
  
      hosts = list(dask_client.ncores().keys())
      nworkers = len(hosts)
  
      ndist_counts = core._count_diagonal_ndist(diags, m, n_A, n_B)
      diags_ranges = core._get_array_ranges(
          ndist_counts, nworkers, False
      )
      diags_ranges += diags[0]
  
      # Scatter data to Dask cluster
      T_A_future = dask_client.scatter(T_A, broadcast=True, hash=False)
      T_B_future = dask_client.scatter(T_B, broadcast=True, hash=False)
      M_T_future = dask_client.scatter(M_T, broadcast=True, hash=False)
      μ_Q_future = dask_client.scatter(μ_Q, broadcast=True, hash=False)
      Σ_T_inverse_future = dask_client.scatter(
          Σ_T_inverse, broadcast=True, hash=False
      )
      σ_Q_inverse_future = dask_client.scatter(
          σ_Q_inverse, broadcast=True, hash=False
      )
      M_T_m_1_future = dask_client.scatter(
          M_T_m_1, broadcast=True, hash=False
      )
      μ_Q_m_1_future = dask_client.scatter(
          μ_Q_m_1, broadcast=True, hash=False
      )
      T_A_subseq_isfinite_future = dask_client.scatter(
          T_A_subseq_isfinite, broadcast=True, hash=False
      )
      T_B_subseq_isfinite_future = dask_client.scatter(
          T_B_subseq_isfinite, broadcast=True, hash=False
      )
      T_A_subseq_isconstant_future = dask_client.scatter(
          T_A_subseq_isconstant, broadcast=True, hash=False
      )
      T_B_subseq_isconstant_future = dask_client.scatter(
          T_B_subseq_isconstant, broadcast=True, hash=False
      )
  
      diags_futures = []
      for i, host in enumerate(hosts):
          diags_future = dask_client.scatter(
              np.arange(
                  diags_ranges[i, 0], diags_ranges[i, 1], dtype=np.int64
              ),
              workers=[host],
              hash=False,
          )
          diags_futures.append(diags_future)
  
      futures = []
      for i in range(len(hosts)):
          futures.append(
              dask_client.submit(
                  _stump,
                  T_A_future,
                  T_B_future,
                  m,
                  μ_Q_future,
                  M_T_future,
                  σ_Q_inverse_future,
                  Σ_T_inverse_future,
                  μ_Q_m_1_future,
                  M_T_m_1_future,
                  T_A_subseq_isfinite_future,
                  T_B_subseq_isfinite_future,
                  T_A_subseq_isconstant_future,
                  T_B_subseq_isconstant_future,
                  diags_futures[i],
                  ignore_trivial,
                  k,
              )
          )
  
      results = dask_client.gather(futures)
      (
          profile,
          profile_L,
          profile_R,
          indices,
          indices_L,
          indices_R,
      ) = results[0]
  
      for i in range(1, len(hosts)):
          P, PL, PR, I, IL, IR = results[i]
          # Update top-k matrix profile and matrix profile indices
          core._merge_topk_PI(profile, P, indices, I)
  
          # Update top-1 left matrix profile and matrix profile index
          mask = PL < profile_L
          profile_L[mask] = PL[mask]
          indices_L[mask] = IL[mask]
  
          # Update top-1 right matrix profile and matrix profile index
          mask = PR < profile_R
          profile_R[mask] = PR[mask]
          indices_R[mask] = IR[mask]
  
      out = np.empty((l, 2 * k + 2), dtype=object)
      out[:, :k] = profile
      out[:, k:] = np.column_stack((indices, indices_L, indices_R))
  
      return out
  
imports: import numpy
  from .None import config, core
  from .aamped import aamped
  from .stump import _stump
label_desc: Use image node_id 1 for calling a global function with example usage: _dask_stumped(dask_client, T_A, T_B, m, M_T, μ_Q, Σ_T_inverse, σ_Q_inverse, M_T_m_1, μ_Q_m_1, T_A_subseq_isfinite, T_B_subseq_isfinite, T_A_subseq_isconstant, T_B_subseq_isconstant, diags, ignore_trivial, k) and returns: out