VSB Power Line Blog - Get all the true, false, symmetric, positive, and negative peaks

get_all_peaks.py

def get_all_peaks(sig, peak_threshold=1):
  # find_peaks(signal, height=[minimum_height, maximum_height])
  # p_peaks_all gives the indices of the positive peaks having amplitude in the range (peak_threshold,100)
  # n_peaks_all gives the indices of the negative peaks having amplitude in the range (peak_threshold,100)
  p_peaks_all = find_peaks(sig, height=[peak_threshold, 100])[0]
  n_peaks_all = find_peaks(-sig, height=[peak_threshold, 100])[0]

  # p_peaks_all = [3,3,1,2,1,2]
  # n_peaks_all = [3,1,1,2,5,2,2,3,3,4,1,4,5]
  # pn_peaks_all --> [1,2,3,4,5]
  pn_peaks_all = np.union1d(p_peaks_all, n_peaks_all)

  # Remove false peaks.
  # Following values are set by an expert from the following paper:
  # https://ieeexplore.ieee.org/document/7909221
  maxDistance = 10
  maxHeightRatio = 0.25
  maxTicksRemoval = 500
  # Get the difference between the consecutive indices of all the peaks
  peak_ind_dist = np.diff(pn_peaks_all)

  sym_peak_pair_indices = []
  for ind, dist in enumerate(peak_ind_dist):
    # find the raito of: (2nd smallest peak)/(1st smallest peak), (3rd smallest peak)/(2nd smallest peak) so on of the signal
    ratio = sig[pn_peaks_all[ind+1]]/sig[pn_peaks_all[ind]]
    # For a pair of consecutive peaks to be considered as a symmetric peak pair, following conditions has to be met:
    # 1. Distance between the current peak index and the next peak index should be less than maxDistance units
    # 2. Next peak should be of opposite sign of the current peak
    # 3. Ratio of the peak values of the next peak to the current peak should be greater than maxHeightRatio
    if (dist < maxDistance):
      if (ratio < 0):
        # This is to consider the cases where:
        # 1. Next peak is higher than the current peak
        # 2. Current peak is higher than the next peak
        if (np.abs(ratio) > maxHeightRatio) and (np.abs(ratio) < (1/maxHeightRatio)):
          # Store the respecive 'symmetric peak pair' index values
          sym_peak_pair_indices.append((pn_peaks_all[ind], pn_peaks_all[ind+1]))

  # Stores the indices of all the peaks that are to be removed
  false_peaks = []
  # pn_peaks_all contains all the peak values' indices
  for peak_val in pn_peaks_all:
    # sym_peak_pair_indices contain symmetric peak pairs
    for sym_peak_pair_ind_curr, sym_peak_pair_ind_next in sym_peak_pair_indices:
      # If the index of the peak is within the maxTicksRemoval distance from the symmetric peak pair index,
      # then consider the indices for removal
      if (peak_val >= sym_peak_pair_ind_next) and (peak_val <= (sym_peak_pair_ind_next + maxTicksRemoval)):        
        false_peaks.append(peak_val)

  # From all the peak indices, do not select the false peak indices
  true_peaks = np.setdiff1d(pn_peaks_all, false_peaks)

  # Get the peak indices after removing the false peak indices
  # a = [3,3,1,2,1,2]
  # b = [3,1,1,2,5,2,2,3,3,4,1,4,5]
  # np.intersect1d(a,b) = [1,2,3]
  # https://www.w3resource.com/python-exercises/numpy/python-numpy-exercise-18.php
  p_peaks = np.intersect1d(p_peaks_all, true_peaks)
  n_peaks = np.intersect1d(n_peaks_all, true_peaks)

  return true_peaks, false_peaks, sym_peak_pair_indices, p_peaks, n_peaks