initial commit

main.py

@@ -0,0 +1,111 @@
+import logging
+
+import pyaudio
+import numpy
+from numpy import pi
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+standard_sample_rates = 1000 * numpy.array([
+                         8, 9.6, 11.025, 12, 16, 22.05, 24, 32,
+                         44.1, 48, 88.2, 96, 192])
+
+def monitor_pitch(device: int = 5,
+                  max_freq: float = 6000,
+                  min_freq: float = 10,
+                  samples_per_buffer: int = 1024,
+                  audio: pyaudio.PyAudio = None,
+                  ):
+    if audio is None:
+        audio = pyaudio.PyAudio()
+
+    supported_sample_rates = []
+    devinfo = audio.get_device_info_by_index(device)
+    for rate in standard_sample_rates:
+        try:
+            if audio.is_format_supported(rate,
+                                     input_device=device,
+                                     input_channels=devinfo['maxInputChannels'],
+                                     input_format=pyaudio.paInt16):
+                supported_sample_rates.append(rate)
+        except ValueError:
+            pass
+    supported_sample_rates = numpy.array(supported_sample_rates)
+    logger.info('Supported rates: {}'.format(supported_sample_rates))
+   
+    '''
+    max_freq < 2 * sample_rate
+    min_freq * 2**(1/12) > freq_resolution (for discrimination), more for accuracy...
+    freq_resolution <= sample_rate / (samples_per_buffer * num_buffers)
+    '''
+    freq_resolution = min_freq * 2**(1/12) / 10
+    
+    rate_is_acceptable = supported_sample_rates >= 2 * max_freq
+    sample_rate = int(numpy.min(supported_sample_rates[rate_is_acceptable]))
+    num_buffers = int(numpy.ceil(sample_rate / (samples_per_buffer * freq_resolution)))
+    samples_per_fft = samples_per_buffer * num_buffers
+    
+    logger.info('Running on device {} with {} buffers,'.format(device, num_buffers) +
+                 ' {} sample rate, {} samples per buffer'.format(
+                  device, num_buffers, sample_rate, samples_per_buffer))
+    logger.info('Buffers take {:.3g} sec to fully clear'.format(samples_per_fft / sample_rate))
+    
+    stream = audio.open(format=pyaudio.paInt16,
+                        channels=1,
+                        rate=sample_rate,
+                        input=True,
+                        frames_per_buffer=samples_per_buffer)
+    stream.start_stream()
+    
+    
+    # Hanning window
+    window = (1 - numpy.cos(numpy.linspace(0, 2 * pi, samples_per_fft, False))) / 2
+    
+    freqs = numpy.fft.fftfreq(samples_per_fft, 1 / sample_rate)
+    
+    buf = numpy.zeros(num_buffers * samples_per_buffer, dtype=numpy.float32)
+    note_names = ['C', 'C#', 'D', 'D#', 'E', 'F', 'F#', 'G', 'G#', 'A', 'A#', 'B']
+
+    while stream.is_active():
+        # Shift the buffer down and new data in
+        buf[:-samples_per_buffer] = buf[samples_per_buffer:]
+        buf[-samples_per_buffer:] = numpy.fromstring(stream.read(samples_per_buffer), numpy.int16)
+    
+        fft = numpy.fft.rfft(buf * window)
+    
+        # Get frequency of maximum response in range
+        ind = numpy.abs(fft[1:]).argmax() + 1
+        freq = freqs[ind]
+        mag = numpy.abs(fft[ind])
+    
+        # Get note number and nearest note
+        q = numpy.log2(freq/440)
+        n = 12 * q + 69
+        n0 = int(round(n))
+        
+        delta = n - n0
+        logger.info('freq: {:7.2f} Hz   mag:{:7.2f}    note: {:>3s} {:+.2f}'.format(
+                freq, numpy.log10(mag), note_names[n0 % 12] + str(n0//12 - 1), delta))
+
+        delta_part = int(delta // 0.1)
+        if delta_part > 0: 
+            signal = ' ' * 6 + '+' * delta_part
+        elif delta_part == 0:
+            signal = ' ' * 5 + '|'
+        elif delta_part < 0:
+            signal = ' ' * (5 + delta_part) + '-' * delta_part
+        logger.info('    {}'.format(signal))
+
+    
+if __name__ == '__main__':
+    audio = pyaudio.PyAudio()
+    logger.info("Available devices:")
+    for device in range(audio.get_device_count()):
+        devinfo = audio.get_device_info_by_index(device)
+        if devinfo['maxInputChannels'] > 0:
+            logger.info('{}: {}'.format(device, devinfo['name']))
+
+    monitor_pitch(device=5, min_freq=20)
+