jan
/
ptuner
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

fix a bunch of major bugs

Browse Source
master
Jan Petykiewicz 6 years ago
parent 6066b0f47f
commit efb73128ab
1 changed files with 69 additions and 62 deletions
Show Stats Download Patch File Download Diff File

131
main.py
Unescape Escape View File

@ -1,6 +1,7 @@
import logging
from typing import List, Dict from typing import List, Dict
import logging
import queue import queue
import threading
import pyaudio import pyaudio
import numpy import numpy
@ -32,18 +33,18 @@ def get_supported_sample_rates(pyaudio_device: int,
pyaudio_object = pyaudio.PyAudio() pyaudio_object = pyaudio.PyAudio()
supported_sample_rates = [] supported_sample_rates = []
devinfo = pyaudio_object.get_device_info_by_index(device) devinfo = pyaudio_object.get_device_info_by_index(pyaudio_device)
for rate in standard_sample_rates: for rate in standard_sample_rates:
try: try:
if pyaudio_object.is_format_supported(rate, if pyaudio_object.is_format_supported(rate,
input_device=device, input_device=pyaudio_device,
input_channels=devinfo['maxInputChannels'], input_channels=devinfo['maxInputChannels'],
input_format=pyaudio.paInt16): input_format=pyaudio.paInt16):
supported_sample_rates.append(rate) supported_sample_rates.append(rate)
except ValueError: except ValueError:
pass pass
supported_sample_rates = numpy.array(supported_sample_rates) supported_sample_rates = numpy.array(supported_sample_rates)
logger.info('Supported sample rates for device {}: {}'.format(device, supported_sample_rates)) return supported_sample_rates
class AudioAnalyzer: class AudioAnalyzer:
@ -61,7 +62,8 @@ class AudioAnalyzer:
stop = None stop = None
def __init__(pyaudio_device: int, def __init__(self,
pyaudio_device: int,
min_freq: float = 20, min_freq: float = 20,
max_freq: float = 20e3, max_freq: float = 20e3,
samples_per_buffer: int = 1024, samples_per_buffer: int = 1024,
@ -81,6 +83,7 @@ class AudioAnalyzer:
supported_sample_rates = get_supported_sample_rates(pyaudio_device, self._pyaudio_object) supported_sample_rates = get_supported_sample_rates(pyaudio_device, self._pyaudio_object)
rate_is_acceptable = supported_sample_rates >= 2 * max_freq rate_is_acceptable = supported_sample_rates >= 2 * max_freq
sample_rate = numpy.min(supported_sample_rates[rate_is_acceptable]).astype(int) sample_rate = numpy.min(supported_sample_rates[rate_is_acceptable]).astype(int)
logger.info('Supported sample rates for device {}: {}'.format(pyaudio_device, supported_sample_rates))
num_buffers = numpy.ceil(sample_rate / (samples_per_buffer * freq_resolution)).astype(int) num_buffers = numpy.ceil(sample_rate / (samples_per_buffer * freq_resolution)).astype(int)
samples_per_fft = samples_per_buffer * num_buffers samples_per_fft = samples_per_buffer * num_buffers
@ -94,73 +97,74 @@ class AudioAnalyzer:
self._fft_lock = threading.Lock() self._fft_lock = threading.Lock()
self.frame_queue = queue.Queue() self.frame_queue = queue.Queue()
self._stream = audio.open(format=pyaudio.paInt16, print(list(self.__dict__.keys()))
channels=1, self._stream = self._pyaudio_object.open(format=pyaudio.paInt16,
rate=sample_rate, channels=1,
input=True, rate=sample_rate,
frames_per_buffer=samples_per_buffer, input=True,
stream_callback=self.update) frames_per_buffer=samples_per_buffer,
stream_callback=self.update)
logger.info('Opened device {} with {} buffers,'.format(device, num_buffers) + logger.info('Opened device {} with {} buffers,'.format(pyaudio_device, num_buffers) +
' {} sample rate, {} samples per buffer'.format( ' {} sample rate, {} samples per buffer'.format(
device, num_buffers, sample_rate, samples_per_buffer)) pyaudio_device, num_buffers, sample_rate, samples_per_buffer))
logger.info('Buffers take {:.3g} sec to fully clear'.format(samples_per_fft / sample_rate)) logger.info('Buffers take {:.3g} sec to fully clear'.format(samples_per_fft / sample_rate))
@property @property
def fft_freqs(self) -> float: def fft_freqs(self) -> float:
return self._fft_freqs return self._fft_freqs
def start(self): def start(self):
self._stream.start_stream() self._stream.start_stream()
def close(self): def close(self):
self.stop = True self.stop = True
self._stream.close() self._stream.close()
self._pyaudio_object.terminate() self._pyaudio_object.terminate()
def update(self, def update(self,
in_data: bytes, in_data: bytes,
frame_count: int, frame_count: int,
time_info: Dict, time_info: Dict,
status_flags, status_flags,
): ):
#TODO deal with exceptions happening in the callback! #TODO deal with exceptions happening in the callback!
in_buffer = numpy.fromstring(in_data, numpy.int16) in_buffer = numpy.fromstring(in_data, numpy.int16)
samples_per_buffer = in_buffer.size samples_per_buffer = in_buffer.size
with self._fft_lock: with self._fft_lock:
self._fft_buffer[:-samples_per_buffer] = self._fft_buffer[samples_per_buffer:] self._fft_buffer[:-samples_per_buffer] = self._fft_buffer[samples_per_buffer:]
self._fft_buffer[-samples_per_buffer:] = in_buffer self._fft_buffer[-samples_per_buffer:] = in_buffer
fft = numpy.fft.rfft(self._fft_buffer * self._hanning_window) fft = numpy.fft.rfft(self._fft_buffer * self._hanning_window)
fft_argmax = numpy.abs(fft[1:]).argmax() + 1 # excluding 0-frequency fft_argmax = numpy.abs(fft[1:]).argmax() + 1 # excluding 0-frequency
frame_data = { frame_data = {
'fft': fft, 'fft': fft,
'fft_argmax': fft_argmax, 'fft_argmax': fft_argmax,
'frequency': self.fft_freqs[fft_argmax], 'frequency': self.fft_freqs[fft_argmax],
'magnitude': numpy.abs(fft[fft_argmax]), 'magnitude': numpy.abs(fft[fft_argmax]),
} }
time_per_buffer = self._samples_per_buffer / self._sample_rate time_per_buffer = self._samples_per_buffer / self._sample_rate
try: try:
self.frame_queue.put(frame_data, timeout=time_per_buffer * 10) self.frame_queue.put(frame_data, timeout=time_per_buffer * 10)
except queue.Full: except queue.Full:
logger.warning('Frame queue was full for more than 10 buffer periods!') logger.warning('Frame queue was full for more than 10 buffer periods!')
if self.stop: if self.stop:
return None, pyaudio.paComplete return None, pyaudio.paComplete
else: else:
return None, pyaudio.paContinue return None, pyaudio.paContinue
def monitor_pitch(device: int = 5, def monitor_pitch(pyaudio_device: int,
min_freq: float = 10, min_freq: float = 10,
max_freq: float = 6000, max_freq: float = 6000,
): ):
analyzer = AudioAnalyzer(device=device, analyzer = AudioAnalyzer(pyaudio_device=pyaudio_device,
min_freq=min_freq, min_freq=min_freq,
max_freq=max_freq) max_freq=max_freq)
@ -170,7 +174,7 @@ def monitor_pitch(device: int = 5,
while True: while True:
frame_data = analyzer.frame_queue.get() frame_data = analyzer.frame_queue.get()
if frame_data['magnitude'] <= prev_magnitude / 2: if frame_data['magnitude'] <= 10**6:
continue continue
prev_magnitude = frame_data['magnitude'] prev_magnitude = frame_data['magnitude']
@ -178,26 +182,29 @@ def monitor_pitch(device: int = 5,
mnote_error = mnote - mnote_base mnote_error = mnote - mnote_base
logger.info('freq: {:7.2f} Hz mag:{:7.2f} note: {:>3s} {:+.2f}'.format( logger.info('freq: {:7.2f} Hz mag:{:7.2f} note: {:>3s} {:+.2f}'.format(
freq, numpy.log10(mag), note_names[base_mnote % 12] + str(base_mnote//12 - 1), mnote_error)) frame_data['frequency'],
numpy.log10(frame_data['magnitude']),
note_names[mnote_base % 12] + str(mnote_base // 12 - 1),
mnote_error))
max_num_symbols = 5 max_num_symbols = 10
num_symbols = int(mnote_error // (0.5 / max_num_symbols)) num_symbols = int(mnote_error // (0.5 / max_num_symbols))
if num_symbols > 0: if num_symbols > 0:
signal = ' ' * max_num_symbols + ' ' + '+' * num_symbols signal = ' ' * max_num_symbols + '|' + '+' * num_symbols
elif num_symbols == 0: elif num_symbols == 0:
signal = ' ' * max_num_symbols + '|' signal = ' ' * max_num_symbols + '#'
elif num_symbols < 0: elif num_symbols < 0:
signal = ' ' * (max_num_symbols - num_symbols) + '-' * num_symbols signal = ' ' * (max_num_symbols + num_symbols) + '-' * -num_symbols + '|'
logger.info(' {}'.format(signal)) logger.info(' {}'.format(signal))
if __name__ == '__main__': if __name__ == '__main__':
audio = pyaudio.PyAudio() audio = pyaudio.PyAudio()
logger.info("Available devices:") logger.info(" Available devices:")
for device in range(audio.get_device_count()): for device in range(audio.get_device_count()):
devinfo = audio.get_device_info_by_index(device) devinfo = audio.get_device_info_by_index(device)
if devinfo['maxInputChannels'] > 0: if devinfo['maxInputChannels'] > 0:
logger.info('{}: {}'.format(device, devinfo['name'])) logger.info(' {}: {}'.format(device, devinfo['name']))
monitor_pitch(device=5, min_freq=20) monitor_pitch(pyaudio_device=7, min_freq=20)

Write Preview
Loading…
Cancel
Save