На волнах эффекта Ли: Питонизируем генерацию DAF

По статистике 1-4 % населения Земли подвержены дефекту речи, характеризующимся частой пролонгацией звуков (слогов, слов) и/или частыми остановками в речи, нарушающими ритмическое ее течение. В простонародье этот феномен известен как заикание.

На данный момент мир не знает панацеи, на 100 % избавляющей от заикания, однако существует преинтереснейший метод, позволяющий с тем или иным успехом купировать это речевое нарушение у большинства заикающихся. Метод основан на эффекте Ли, заключающемся во влиянии задержки акустической слуховой афферентации на плавность речи, и носит название DAF (Delayed Auditory Feedback).

Ниже рассмотрим пример построения на коленке простого генератора речевой обратной связи силами Python и PyQt. У-у-ух, it's gonna be fun!

Что к чему и почему

Эффект Ли, названный в честь инженера-подводника Бернарда Ли (Lee, 1951 г.), проявляется в том, что у обычного человека прослушивание через наушники собственной речи, задержанной с помощью специальной аппаратуры на 80-200 мс, (непосредственно в момент разговора) вызывает запинки, очень напоминающие заикание. В то же время на человека, подверженного заиканию, эффект Ли оказывает прямо противоположное воздействие. На этом строится метод DAF. Смысл задержки воспроизведения речи в наушниках заключается в синхронизации работы речевых центров — слухового центра Вернике и речедвигательного центра Брока (криптометафора: задержка = функция генерации гаммы для самосинхронизирующегося шифра потока). Проведение различных исследований позволило выявить, что задержка в диапазоне 50-75 мс позволяет уменьшить заикание на 60-80% при нормальной и ускоренной речи. Задержка в 190 мс оказалась чуть более эффективной, чем 75 мс, однако оптимальная величина задержки выбирается индивидуально исходя из ощущений испытуемого.

Идея аппаратного подхода к нормализации речи стара как мир — первый прибор, работающий по принципу «регуляции обратной связи», был сконструирован в 1959 г. Будучи большим, проводным и неуклюжим, он представлялся малоэффективным для использования в повседневной жизни, однако технологии не стоят на месте, и сейчас существует целый ряд способов удобной генерации DAF: как с помощью отдельных мини-девайсов, так и в виде софтин для Ведроида, Аппле (поиск по ключевому слову «DAF» в своем магазине приложений покажет всю выборку таких решений) и ПК (тут сложнее, смотрим следующий абзац).

Зачем сей пост

Стоимость приложений для мобильных платформ варьируется в пределах нескольких долларов. Допустимо. Однако для Windows существует всего одна подобная программа стоимостью $30 за базовую версию для «personal needs only» (название приводить не буду — лежит по тому же ключевому слову на первой ссылке поисковика). Здесь мне стало интересно, во сколько строк кода встанет самопальная реализация столь тривиального функционала. Результатом этого интереса стало одинокое окно GUI-интерфейса, скрывающее под капотом простой DAF-генератор, которым хочу поделиться с окружающими — авось кому пригодится.

Trial. CLI-интерфейс

Для начала набросаем концепт в виде пробного скрипта. Будем использовать связку "Python3 + PyAudio", где PyAudio — модуль для работы со звуком. Ядро будет выглядеть так:

CHANNELS = 2
RATE = 44100


def genDAF(delay):
	bufferSize = floor(delay / 1000 * RATE)
	device = PyAudio()

	try:
		streamIn = device.open(
			format=paFloat32,
			channels=CHANNELS,
			rate=RATE,
			input=True,
			frames_per_buffer=bufferSize
		)

		streamOut = device.open(
			format=paFloat32,
			channels=CHANNELS,
			rate=RATE,
			output=True,
			frames_per_buffer=bufferSize
		)

	except OSError:
		print('genDAF: error: No input/output device found! Connect and rerun')
		return

	print('CTRL-C to stop capture')

	while streamIn.is_active():
		start = clock()
		audioData = streamIn.read(bufferSize)
		streamOut.write(audioData)
		actualDelay = floor((clock() - start) * 1000)
		print('Actual Delay:  {} ms'.format(actualDelay))

Процедура genDAF принимает величину задержки в миллисекундах, вычисляет необходимый размер буфера (исходя из оптимального битрейта в 44,1 кГц) для записи голоса, после чего, если имеются подключения на ввод и вывод аудио (aka микрофон и динамики), создает два потока, входной и выходной соответственно. Далее в основном цикле начинается чтение и мгновенное воспроизведение записанного куска аудио данных, при этом на фоне подсчитывается реальная задержка, потребовавшаяся для выполнения пары операций read/write. На все ушло ~ 20 строк кода.

Полный исходник CLI-реализации под спойлером:

dafgen_cli.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Usage: python3 dafgen_cli.py <delay_in_ms>

from pyaudio import PyAudio, paFloat32
from math import floor
from time import clock

import sys

CHANNELS = 2
RATE = 44100


def genDAF(delay):
	bufferSize = floor(delay / 1000 * RATE)
	device = PyAudio()

	try:
		streamIn = device.open(
			format=paFloat32,
			channels=CHANNELS,
			rate=RATE,
			input=True,
			frames_per_buffer=bufferSize
		)

		streamOut = device.open(
			format=paFloat32,
			channels=CHANNELS,
			rate=RATE,
			output=True,
			frames_per_buffer=bufferSize
		)

	except OSError:
		print('genDAF: error: No input/output device found! Connect and rerun')
		return

	print('CTRL-C to stop capture')

	while streamIn.is_active():
		start = clock()
		audioData = streamIn.read(bufferSize)
		streamOut.write(audioData)
		actualDelay = floor((clock() - start) * 1000)
		print('Actual Delay:  {} ms'.format(actualDelay))


def main():
	if len(sys.argv) != 2:
		print('Usage: python3 {} <delay_in_ms>'.format(sys.argv[0]))
		sys.exit(1)

	try:
		delay = int(sys.argv[1])
	except ValueError:
		print('main: error: Invalid input type')
		sys.exit(1)

if not 50 <= delay <= 200:
	print('main: error: Delay must be in [50; 200] ms')
	sys.exit(1)

	print('Delay:  {} ms\n'.format(delay))

	try:
		genDAF(delay)
	except KeyboardInterrupt:
		print('Stopped')


if __name__ == '__main__':
	main()

Final. GUI-интерфейс

Зелёные буквы на чёрном фоне терминала — романтично, но не всегда удобно, мы можем лучше. Приплюсуем к нашей связки инструментов фреймворк для графики, получится "Python3 + PyAudio + PyQt5".

Набросаем в дизайнере пару-тройку кнопок, слайдер и 2 текстовых поля:

Добавим логики, распределив основной код генерации DAF по двум классам: управляющее приложение (MainApp) и класс для отдельного потока (Worker), ответственный за выполнение цикла while метода _genDAF, дабы не висло основное окно. Полный код приведен в конце параграфа, а сейчас только главная часть.

Управляющее приложение:

class MainApp(QMainWindow, Ui_DAFGen):

	_CHANNELS = 2
	_RATE = 44100

	def __init__(self):
		super().__init__()
		self.setupUi(self)
		# ...

	# ...

	def _startCapture(self):
		bufferSize = floor(self.delaySlider.value() / 1000 * self._RATE)
		device = PyAudio()

		try:
			streamIn = device.open(
				format=paFloat32,
				channels=self._CHANNELS,
				rate=self._RATE,
				input=True,
				frames_per_buffer=bufferSize
			)

			streamOut = device.open(
				format=paFloat32,
				channels=self._CHANNELS,
				rate=self._RATE,
				output=True,
				frames_per_buffer=bufferSize
			)

		except OSError:
			QMessageBox.critical(self, 'Error', 'No input/output device found! Connect and rerun.')
			return

		self._workerThread = Worker(bufferSize, streamIn, streamOut)
		self._workerThread._trigger.connect(self._updateActualDelay)
		# ...
		self._workerThread.start()

Второй поток:

class Worker(QThread):

	_trigger = pyqtSignal(float)

	def __init__(self, bufferSize, streamIn, streamOut):
		QThread.__init__(self)
		self._bufferSize = bufferSize
		self._streamIn = streamIn
		self._streamOut = streamOut

	def __del__(self):
		self.wait()

	def _genDAF(self):
		while self._streamIn.is_active():
			start = clock()
			audioData = self._streamIn.read(self._bufferSize)
			self._streamOut.write(audioData)
			actualDelay = clock() - start
			self._trigger.emit(actualDelay)

	def run(self):
		self._genDAF()

Исходник для логики GUI-реализации под спойлером:

dafgen.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# Usage: python3 dafgen.py

from PyQt5.QtWidgets import *
from PyQt5.QtCore import *
from ui_dafgen import Ui_DAFGen

from pyaudio import PyAudio, paFloat32
from math import floor
from time import clock

import sys


class Worker(QThread):

	_trigger = pyqtSignal(float)

	def __init__(self, bufferSize, streamIn, streamOut):
		QThread.__init__(self)
		self._bufferSize = bufferSize
		self._streamIn = streamIn
		self._streamOut = streamOut

	def __del__(self):
		self.wait()

	def _genDAF(self):
		while self._streamIn.is_active():
			start = clock()
			audioData = self._streamIn.read(self._bufferSize)
			self._streamOut.write(audioData)
			actualDelay = clock() - start
			self._trigger.emit(actualDelay)

	def run(self):
		self._genDAF()


class MainApp(QMainWindow, Ui_DAFGen):

	_CHANNELS = 2
	_RATE = 44100

	def __init__(self):
		super().__init__()
		self.setupUi(self)

		self.stopButton.setEnabled(False)
		self._updateDelay()

		self.delaySlider.valueChanged.connect(self._updateDelay)
		self.startButton.clicked.connect(self._startCapture)
		self.stopButton.clicked.connect(self._stopCapture)
		self.quitButton.clicked.connect(QApplication.quit)

	def _updateDelay(self):
		self.delayEdit.setPlainText(str(self.delaySlider.value()) + ' ms')

	def _startCapture(self):
		bufferSize = floor(self.delaySlider.value() / 1000 * self._RATE)
		device = PyAudio()

		try:
			streamIn = device.open(
				format=paFloat32,
				channels=self._CHANNELS,
				rate=self._RATE,
				input=True,
				frames_per_buffer=bufferSize
			)

			streamOut = device.open(
				format=paFloat32,
				channels=self._CHANNELS,
				rate=self._RATE,
				output=True,
				frames_per_buffer=bufferSize
			)

		except OSError:
			QMessageBox.critical(self, 'Error', 'No input/output device found! Connect and rerun.')
			return

		self._workerThread = Worker(bufferSize, streamIn, streamOut)
		self._workerThread._trigger.connect(self._updateActualDelay)

		self.startButton.setEnabled(False)
		self.delaySlider.setEnabled(False)
		self.stopButton.setEnabled(True)

		self._workerThread.start()

	def _stopCapture(self):
		self._workerThread.terminate()

		self.actualDelayEdit.clear()
		self.startButton.setEnabled(True)
		self.delaySlider.setEnabled(True)
		self.stopButton.setEnabled(False)

	def _updateActualDelay(self, t):
		newValue = floor(t * 1000)
		self.actualDelayEdit.setPlainText(str(newValue) + ' ms')


def main():
	app = QApplication(sys.argv)
	win = MainApp()
	win.show()
	sys.exit(app.exec_())


if __name__ == '__main__':
	main()

Заключение и код

Собственно всё, что хотел рассказать. Feel free to use.

Также оставлю ссылку на проект целиком: в дополнение там лежат код графического интерфейса и шаблон для PyQt Designer.

Спасибо за внимание!

Литература

Миссуловин Л. Я., Юрова М. С. Преодоление заикания у подростков и взрослых с использованием аппаратов типа «АИР» // Научно-методический электронный журнал «Концепт». – 2015. – № S23. – С. 46–50. – URL: e-koncept.ru/2015/75287.htm.