[autotest] audio_analysis: Improve peak detection
The peak detection algorithm was poorly written so the performance was
really bad. It took more than 20 minutes to finish analysis of 60
seconds data.
Improve the algorithm to speed up peak detection.
Also, add the logging format in unittest to check the timestamp.
BUG=chromium:645666
TEST=run audio_analysis_unittest and see the improvement in comparison
with dummy implementation.
./client/cros/audio/audio_analysis_unittest.py
SpectralAnalysisTest.testPeakDetectionLarge
2016-09-14 08:51:21,328 - root - DEBUG - Test large array using dummy
peak detection
2016-09-14 08:51:22,516 - root - DEBUG - Test large array using improved
peak detection
2016-09-14 08:51:22,944 - root - DEBUG - Compare the result
.
----------------------------------------------------------------------
Ran 1 test in 1.631s
The dummy implementation took 1.2 seconds, while improved
implementation took 0.4 seconds.
Change-Id: I019eb7271dd68ddd11751df60748b6841d9bcb12
Reviewed-on: https://chromium-review.googlesource.com/385079
Commit-Ready: Cheng-Yi Chiang <cychiang@chromium.org>
Tested-by: Cheng-Yi Chiang <cychiang@chromium.org>
Reviewed-by: Wai-Hong Tam <waihong@google.com>
(cherry picked from commit 47ab82af074d76fc390f1b088118c6d48776ced6)
Reviewed-on: https://chromium-review.googlesource.com/422123
Reviewed-by: Ting Shen <phoenixshen@chromium.org>
Commit-Queue: Ting Shen <phoenixshen@chromium.org>
Tested-by: Ting Shen <phoenixshen@chromium.org>
diff --git a/client/cros/audio/audio_analysis.py b/client/cros/audio/audio_analysis.py
index a708200..5ff3131 100644
--- a/client/cros/audio/audio_analysis.py
+++ b/client/cros/audio/audio_analysis.py
@@ -99,7 +99,7 @@
peak_window_size = int(peak_window_size_hz / x_f[1])
# Detects peaks.
- peaks = _peak_detection(abs_y_f, peak_window_size)
+ peaks = peak_detection(abs_y_f, peak_window_size)
# Transform back the peak location from index to frequency.
results = []
@@ -125,13 +125,14 @@
return numpy.linspace(0, (result_length - 1) * val, result_length)
-def _peak_detection(array, window_size):
+def peak_detection(array, window_size):
"""Detects peaks in an array.
A point (i, array[i]) is a peak if array[i] is the maximum among
array[i - half_window_size] to array[i + half_window_size].
If array[i - half_window_size] to array[i + half_window_size] are all equal,
then there is no peak in this window.
+ Note that we only consider peak with value greater than 0.
@param window_size: The window to detect peaks.
@@ -143,31 +144,71 @@
half_window_size = window_size / 2
length = len(array)
- def find_max(numbers):
- """Gets the index where maximum value happens.
+ def mid_is_peak(array, mid, left, right):
+ """Checks if value at mid is the largest among left to right in array.
- @param numbers: A list of numbers.
+ @param array: A list of numbers.
+ @param mid: The mid index.
+ @param left: The left index.
+ @param rigth: The right index.
- @returns: (index, value) where value = numbers[index] is the maximum
- among numbers.
+ @returns: A tuple (is_peak, next_candidate)
+ is_peak is True if array[index] is the maximum among numbers
+ in array between index [left, right] inclusively.
+ next_candidate is the index of next candidate for peak if
+ is_peak is False. It is the index of maximum value in
+ [mid + 1, right]. If is_peak is True, next_candidate is
+ right + 1.
"""
- index, value = max(enumerate(numbers), key=lambda x: x[1])
- return index, value
+ value_mid = array[mid]
+ is_peak = True
+ next_peak_candidate_index = None
+
+ # Check the left half window.
+ for index in xrange(left, mid):
+ if array[index] >= value_mid:
+ is_peak = False
+ break
+
+ # Mid is at the end of array.
+ if mid == right:
+ return is_peak, right + 1
+
+ # Check the right half window and also record next candidate.
+ # Favor the larger index for next_peak_candidate_index.
+ for index in xrange(right, mid, -1):
+ if (next_peak_candidate_index is None or
+ array[index] > array[next_peak_candidate_index]):
+ next_peak_candidate_index = index
+
+ if array[next_peak_candidate_index] >= value_mid:
+ is_peak = False
+
+ if is_peak:
+ next_peak_candidate_index = right + 1
+
+ return is_peak, next_peak_candidate_index
results = []
- for mid in xrange(length):
+ mid = 0
+ next_candidate_idx = None
+ while mid < length:
left = max(0, mid - half_window_size)
right = min(length - 1, mid + half_window_size)
- numbers_in_window = array[left:right + 1]
- max_index, max_value = find_max(numbers_in_window)
- # Add the offset back.
- max_index = max_index + left
+ # Only consider value greater than 0.
+ if array[mid] == 0:
+ mid = mid + 1
+ continue;
- # If all values are the same then there is no peak in this window.
- if max_value != min(numbers_in_window) and max_index == mid:
- results.append((mid, max_value))
+ is_peak, next_candidate_idx = mid_is_peak(array, mid, left, right)
+
+ if is_peak:
+ results.append((mid, array[mid]))
+
+ # Use the next candidate found in [mid + 1, right], or right + 1.
+ mid = next_candidate_idx
# Sort the peaks by values.
return sorted(results, key=lambda x: x[1], reverse=True)
diff --git a/client/cros/audio/audio_analysis_unittest.py b/client/cros/audio/audio_analysis_unittest.py
index 2621b2d..6ab1e85 100755
--- a/client/cros/audio/audio_analysis_unittest.py
+++ b/client/cros/audio/audio_analysis_unittest.py
@@ -14,6 +14,74 @@
numpy.random.seed(0)
+ def dummy_peak_detection(self, array, window_size):
+ """Detects peaks in an array in simple way.
+
+ A point (i, array[i]) is a peak if array[i] is the maximum among
+ array[i - half_window_size] to array[i + half_window_size].
+ If array[i - half_window_size] to array[i + half_window_size] are all
+ equal, then there is no peak in this window.
+
+ @param window_size: The window to detect peaks.
+
+ @returns: A list of tuples:
+ [(peak_index_1, peak_value_1), (peak_index_2, peak_value_2),
+ ...]
+ where the tuples are sorted by peak values.
+
+ """
+ half_window_size = window_size / 2
+ length = len(array)
+
+ def mid_is_peak(array, mid, left, right):
+ """Checks if value at mid is the largest among left to right.
+
+ @param array: A list of numbers.
+ @param mid: The mid index.
+ @param left: The left index.
+ @param rigth: The right index.
+
+ @returns: True if array[index] is the maximum among numbers in array
+ between index [left, right] inclusively.
+
+ """
+ value_mid = array[mid]
+ for index in xrange(left, right + 1):
+ if index == mid:
+ continue
+ if array[index] >= value_mid:
+ return False
+ return True
+
+ results = []
+ for mid in xrange(length):
+ left = max(0, mid - half_window_size)
+ right = min(length - 1, mid + half_window_size)
+ if mid_is_peak(array, mid, left, right):
+ results.append((mid, array[mid]))
+
+ # Sort the peaks by values.
+ return sorted(results, key=lambda x: x[1], reverse=True)
+
+
+ def testPeakDetection(self):
+ array = [0, 1, 2, 3, 4, 3, 2, 1, 0, 1, 2, 3, 5, 3, 2, 1, 1, 1, 1, 1]
+ result = audio_analysis.peak_detection(array, 4)
+ golden_answer = [(12, 5), (4, 4)]
+ self.assertEqual(result, golden_answer)
+
+
+ def testPeakDetectionLarge(self):
+ array = numpy.random.uniform(0, 1, 1000000)
+ window_size = 100
+ logging.debug('Test large array using dummy peak detection')
+ dummy_answer = self.dummy_peak_detection(array, window_size)
+ logging.debug('Test large array using improved peak detection')
+ improved_answer = audio_analysis.peak_detection(array, window_size)
+ logging.debug('Compare the result')
+ self.assertEqual(dummy_answer, improved_answer)
+
+
def testSpectralAnalysis(self):
rate = 48000
length_in_secs = 0.5
@@ -284,5 +352,5 @@
if __name__ == '__main__':
- logging.basicConfig(level=logging.DEBUG)
+ logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
unittest.main()
