Как будет звучать ДНК, если ее синтезировать в аудиофайл

Содержание

1. Введение
2. Кодирование ДНК и пример праймера
3. Разбор данных ДНК
4. Генерация синусоидальной волны
5. Генерация спектрограмм
6. Предварительно сгенерированные последовательности
   1. Цитата Нильса Бора
   2. Мышь
   3. Бизон
   4. Телец
7. Создание барабанщика из последовательности ДНК
8. Идем дальше

Введение

В последнее время я много размышлял о природе жизни, о том, каковы фундаментальные блоки жизни и тому подобные вещи. Удивительно, насколько сложно и, с другой стороны, просто творение, когда вы смотрите на него. Чудо жизни удерживает нас на земле, когда наше воображение выходит из-под контроля. Если бы ДНК были блоками жизни, вы могли бы считать их API-природой, которая позволила нам лучше понять весь этот хаос, маскирующийся под порядок.

Я много читал о сверхразуме и о нашем ошибочном пути к созданию общего искусственного интеллекта. Как будут выглядеть строительные блоки или наше творение? Действительно ли сжатие является окончательным хранилищем информации? Будет ли наше творение также задумываться над этими вопросами, создавая для себя новые миры, или мы просто растворимся в безбрежности возможностей? Это немного оскорбительно, что мы играем в Бога, совершенно не зная о нашей собственной реальности. Кто знает! Как и многие другие прорывы, этот также будет стоить нам неизвестно, когда он, наконец, произойдет.

Чтобы было немного легче, я решил преобразовать некоторые популярные последовательности ДНК в аудиофайлы, чтобы мы могли их прослушать. Я не первый, и не буду последним, кто это сделает. Но это интересное упражнение в лучшем понимании отношений между искусством и наукой. Возможно, прослушивание ДНК вместо ее разбора поможет лучше понять или, по крайней мере, насладиться творением и загадочной природой жизни.

Кодирование ДНК и пример праймера

В прошлом я изучал ДНК в своем посте около 3 лет назад в разделе Кодирование двоичных данных в последовательность ДНК, где я преобразовывал все виды данных в последовательности ДНК.

Это будет похожее упражнение, но вместо преобразования в ДНК я буду генерировать звуки из нуклеотидов.

Nucleotides	Note	Frequency
A (Adenine)	A	440 Hz
C (Cytosine)	C	783.99 Hz
G (гуанин)	G	523,25 Гц
T (Тимин)	D	587,33 Гц

Поскольку у нас нет T в равнотемперированной гамме, я выбираю D для обозначения ноты T.

Вы можете проверить Частоты для равнотемперированного строя, A4 = 440 Гц. Для этой настройки мы также выбираем Скорость звука = 345 м/с = 1130 футов/с = 770 миль/ч .

Теперь, когда мы разобрались с этим, мы можем немного освежить секвенирование ДНК. Это известная цитата, которую я также использовал для тестов кодирования, и она звучит так.

Как замечательно, что мы встретились с парадоксом. Теперь у нас есть надежда на прогресс. ― Нильс Бор

 >SEQ1
GACAGCTTGTGTACAAGTGTGCTTGCTCGCGAGCGGGTACGCGCGTGGGCTAACAAGTGA
GCCAGCAGGTGAACAAGTGTGCGGACAAGCCAGCAGGTGCGCGGACAAGCTGGCGGGTGA
ACAAGTGTGCCGGTGAGCCAACAAGCAGACAAGTAAGCAGGTACGCAGGCGAGCTTGTCA
ACTCACAAGATCGCTTGTGTACAAGTGTGCGGACAAGCCAGCAGGTGCGCGGACAAGTAT
GCTTGCTGGCGGACAAGCCAGCTTGTAAGCGGACAAGCTTGCGCACAAGCTGGCAGGCCT
GCCGGCTCGCGTACAAATTCACAAGTAAGTACGCTTGCGTGTACGCGGGTATGTATACTC
AACCTCACCAAACGGGACAAGATCGCCGGCGGGCTAGTATACAAGAACGCTTGCCAGTAC
ААКК
 

Это то, с чем мы будем работать, чтобы ускорить процесс при создании синтаксического анализатора и генератора сигналов.

Анализ данных ДНК

Этот шаг достаточно прост. Все, что нам нужно сделать, это проанализировать входную последовательность ДНК в формате FASTA, хорошо известном в биоинформатике, для извлечения отдельных нуклеотидов, которые будут преобразованы в отдельные тона на основе равнотемперированной шкалы, описанной выше.

 нуклеотид_tone_map = {
  «А»: 440,
  «С»: 523,25,
  «Г»: 783,99,
  'T': 587,33, # преобразовано в D
}
определение разделения (слово):
  вернуть [символ для символа в слове]
def generate_from_dna_sequence (последовательность):
  для нуклеотида в расщеплении (последовательности):
    print(нуклеотид, нуклеотид_tone_map[нуклеотид])
 

Генерация синусоидальной волны

Поскольку мы, по сути, создаем длинный поток заметок, мы будем добавлять синусоидальные заметки в глобальный массив, который мы позже используем для создания из него файла WAV.

 импорт математики
def append_sinewave (частота = 440,0, продолжительность_миллисекунд = 500, громкость = 1,0):
  глобальное аудио
  num_samples = продолжительность_миллисекунд * (sample_rate / 1000,0)
  для x в диапазоне (int (num_samples)):
    audio.append(объем * math.sin(2 * math.pi * freq * (x/sample_rate)))
  возвращаться
 

Генерируемый здесь синусоидальный сигнал является стандартным звуковым сигналом. Если вы хотите что-то более агрессивное, вы можете попробовать прямоугольную или пилообразную форму волны.

Создание файла WAV из накопленных синусоид

  импорт волны
импортировать структуру
защита save_wav (имя_файла):
  wav_file = wave.open(имя_файла, 'w')
  nканалов = 1
  ширина = 2
  nframes = длина (аудио)
  тип_компьютера = 'НЕТ'
  compname = 'не сжато'
  wav_file.setparams((nchannels, sampwidth, sample_rate, nframes, comptype, compname))
  для образца в аудио:
    wav_file.writeframes(struct.pack('h', int(sample * 32767.0)))
  wav_file.close()
 

44100 — стандартная частота дискретизации для CD-качества. Если вам нужно сэкономить на размере файла, вы можете уменьшить его. Стандарт низкого качества — 8000 или 8 кГц.

Файлы WAV здесь используют короткие 16-битные целые числа со знаком для размера выборки. Итак, мы умножаем имеющиеся у нас данные с плавающей запятой на 32767, максимальное значение для короткого целого числа.

Теоретически возможно использовать данные с плавающей запятой от -1,0 до 1,0 непосредственно в файле WAV, но не очевидно, как это сделать с помощью модуля волны в Python.

Генерация спектрограмм

Я попробовал два метода сделать это, и оба оказались прекрасными. Однако я отказался от использования SoX — Sound eXchange, швейцарского армейского ножа для манипуляций со звуком, потому что больше ничего не требовалось.

 sox output.wav -n спектрограмма -o спектрограмма.png
 

Пример спектрограммы Симфонии № 6 Людвига ван Бетховена Первая часть.

Другой вариант также может быть в сочетании с gnuplot. Однако для этого потребуется промежуточный шаг.

 sox output.wav audio.dat
хвост -n+3 audio.dat > audio_only.dat
gnuplot аудио.gpi
 

Входной файл audio.gpi , который будет передан gnuplot, выглядит примерно так.

 # установить выходной формат и размер
установить срок png размер 1000 280
# установить выходной файл
установить вывод "audio.png"
# установить диапазон у
установить год [-1:1]
# нам нужны только данные
ключ сброса
сбросить тики
неустановленная граница
установить lmargin 0
установить rmargin 0
установить tmargin 0
установить bmargin 0
# нарисуйте прямоугольник, чтобы изменить цвет фона
установить прямоугольник obj 1 позади от экрана 0,0 до экрана 1,1
set obj 1 fillstyle solid 1.
  0 fillcolor rgbcolor "#ffffff"
# рисуем данные цветом переднего плана
сюжет «audio_only.dat» с линиями lt rgb «красный»
 

Предварительно сгенерированные последовательности

Что я сделал, так это взял интересные части из генома животного и передал их сценарию генератора тона. Затем был сгенерирован файл WAV, и я преобразовал его в MP3, чтобы его можно было воспроизводить в браузере. Последним шагом было создание спектрограммы на основе файла WAV.

Цитата Нильса Бора

Мышь

Это часть генома мыши Mus_musculus.GRCm39.dna.nonchromosomal . Вы можете получить данные генома здесь.

Бизон

Это часть генома бизона Bison_bison_bison.Bison_UMD1.0.cdna . Вы можете получить данные генома здесь.

Телец

Это часть генома Тельца Bos_taurus.ARS-UCD1.2.cdna . Вы можете получить данные генома здесь.

Создание барабанщика из последовательности ДНК

Чтобы было еще интереснее, я решил отправить эти данные через MIDI на мою модель Elektron:Samples. Это действительно классное оборудование, которое поддерживает вход MIDI через USB и аудиоразъем 3,5 мм.

Elektron подключен к моему MacBook через USB-кабель, а аудиовыход подключен к динамику Sony Bluetooth, который поддерживает аудиовход 3,5 мм. У Elektron нет внутренних динамиков.

Для связи с Elektron я выбрал модуль Python pygame со встроенным MIDI. Благодаря этому было довольно просто отправлять заметки на устройство. Все, что я сделал, это сопоставил ноты MIDI с настоящими нуклеотидами.

Перед всем этим я также проверил приложение Audio MIDI Setup под MacOS и проверил MIDI Studio, нажав ⌘-2.

Весь скрипт, который анализирует и отправляет заметки в Электрон, выглядит так.

 импорт pygame.midi
время импорта
pygame.midi.init()
печать (pygame.midi.get_default_output_id())
печать (pygame.midi.get_device_info (0))
игрок = pygame.midi.Output(1)
player.set_instrument(2)
def send_note (примечание, скорость):
  глобальный игрок
  player. note_on(нота, скорость)
  время сна (0,3)
  player.note_off(нота, скорость)
нуклеотид_миди_карта = {
  «А»: 60,
  «С»: 90,
  «Г»: 160,
  'T': 180, # это D
}
с open("quote.fa") как f:
  последовательность = f.read().replace('\n', '')
для нуклеотида в [char для char в последовательности]:
  print("Воспроизведение нуклеотида {} с нотой MIDI {}".format(
      нуклеотид, нуклеотид_миди_карта[нуклеотид]))
  send_note (нуклеотид_миди_карта [нуклеотид], 127)
игрок
pygame.midi.quit()
 

Все это можно было бы сделать намного интереснее, если бы я выбрал разные инструменты для разных нуклеотидов или делал бы что-то более прикольное с Elektron. Но пока этого должно быть достаточно. Это просто доказательство концепции. Есть с чем поиграть.

Идем дальше

Как вы, наверное, заметили, конечные результаты очень похожи друг на друга. Этого и следовало ожидать, потому что мы работаем только с 4 нотами. Что может сделать это более интересным, так это использование чего-то вроде Supercollider для создания более интересных звуков. Путем переноса нот или использования эффектов на основе повторяющихся данных в последовательности. Возможности безграничны.

Удивительно, чего можно добиться, используя немного кода и идею. Я мог видеть, что это становится интересным инструментом фонового звукового ландшафта, если все сделано правильно. Он мог бы заменить генератор случайных заметок чем-то более интригующим, биологическим, естественным.

Я действительно нахожу результаты захватывающими. Я взял некоторое время и слушал эту музыку природы. Несмотря на то, что это то же самое, это также совершенно другое. Тонкие различия в повторении создают музыку сами по себе. Заставляет задуматься. Это как бы ставит бритву Оккама на место. Природа, безусловно, любит делать вещи максимально энергоэффективными.

Примеры программ — python-sounddevice, версия 0.3.15

В большинстве этих примеров для обработки командной строки используется модуль argparse . аргументы. Чтобы отобразить текст справки, объясняющий все доступные аргументы, используйте аргумент --help .

Например:

 python3 play_file.py --help
 

Воспроизведение звукового файла

play_file.py

 #!/usr/bin/env python3
"""Загрузить аудиофайл в память и воспроизвести его содержимое.
NumPy и модуль звукового файла (https://PySoundFile.readthedocs.io/)
должен быть установлен, чтобы это работало.
Этот пример программы загружает весь файл в память перед запуском
воспроизведение.
Для воспроизведения очень длинных файлов следует использовать play_long_file.py.
"""
импортировать аргументы
импортировать звуковое устройство как SD
импортировать звуковой файл как sf
определение int_or_str (текст):
    """Вспомогательная функция для разбора аргументов."""
    пытаться:
        вернуть целое (текст)
    кроме ValueError:
        возвращаемый текст
синтаксический анализатор = argparse.ArgumentParser (add_help = False)
parser.add_argument(
    '-l', '--list-devices', action='store_true',
    help='показать список аудиоустройств и выйти')
аргументы, осталось = parser.
  parse_known_args()
если args.list_devices:
    печать (sd.query_devices())
    парсер.выход(0)
синтаксический анализатор = argparse.ArgumentParser(
    описание=__doc__,
    formatter_class=argparse.RawDescriptionHelpFormatter,
    родители=[парсер])
parser.add_argument(
    'имя файла', metavar='ИМЯ ФАЙЛА',
    help='аудиофайл для воспроизведения')
parser.add_argument(
    '-d', '--device', тип=int_or_str,
    help='устройство вывода (числовой идентификатор или подстрока)')
аргументы = parser.parse_args (оставшиеся)
пытаться:
    данные, fs = sf.read(args.filename, dtype='float32')
    sd.play(данные, fs, устройство=args.device)
    статус = sd.wait()
кроме KeyboardInterrupt:
    parser.exit('\nПрервано пользователем')
кроме Исключения как e:
    parser.exit (тип (е). __ имя__ + ': ' + ул (е))
если статус:
    parser.exit('Ошибка при воспроизведении: ' + str(status))
 

Воспроизвести очень длинный звуковой файл

play_long_file.py

 #!/usr/bin/env python3
"""Воспроизведение аудиофайла с использованием ограниченного объема памяти. 
Модуль звукового файла (https://PySoundFile.readthedocs.io/) должен быть
установлен для этого, чтобы работать. NumPy не нужен.
В отличие от play_file.py, который загружает весь файл в память
перед началом воспроизведения эта примерная программа содержит только заданное число
аудиоблоков в памяти и поэтому может воспроизводить файлы,
больше доступной оперативной памяти.
Аналогичный пример, конечно, можно реализовать с помощью NumPy,
но этот пример показывает, что можно сделать, когда NumPy недоступен.
"""
импортировать аргументы
очередь импорта
импорт системы
импорт потоков
импортировать звуковое устройство как SD
импортировать звуковой файл как sf
определение int_or_str (текст):
    """Вспомогательная функция для разбора аргументов."""
    пытаться:
        вернуть целое (текст)
    кроме ValueError:
        возвращаемый текст
синтаксический анализатор = argparse.ArgumentParser (add_help = False)
parser.add_argument(
    '-l', '--list-devices', action='store_true',
    help='показать список аудиоустройств и выйти')
аргументы, осталось = parser. parse_known_args()
если args.list_devices:
    печать (sd.query_devices())
    парсер.выход(0)
синтаксический анализатор = argparse.ArgumentParser(
    описание=__doc__,
    formatter_class=argparse.RawDescriptionHelpFormatter,
    родители=[парсер])
parser.add_argument(
    'имя файла', metavar='ИМЯ ФАЙЛА',
    help='аудиофайл для воспроизведения')
parser.add_argument(
    '-d', '--device', тип=int_or_str,
    help='устройство вывода (числовой идентификатор или подстрока)')
parser.add_argument(
    '-b', '--blocksize', тип=int, по умолчанию=2048,
    help='размер блока (по умолчанию: %(по умолчанию)s)')
parser.add_argument(
    '-q', '--buffersize', тип=целое, по умолчанию=20,
    help='количество блоков, используемых для буферизации (по умолчанию: %(по умолчанию)s)')
аргументы = parser.parse_args (оставшиеся)
если args.blocksize == 0:
    parser.error('Размер блока не должен быть равен нулю')
если args.buffersize < 1:
    parser.error('размер буфера должен быть не менее 1')
q = очередь. Очередь (maxsize=args.buffersize)
событие = многопоточность.Событие()
Обратный вызов def (аутданные, кадры, время, статус):
    утверждать кадры == args.blocksize
    если статус.output_underflow:
        print('Опустошение вывода: увеличить размер блока?', file=sys.stderr)
        поднять sd.CallbackAbort
    утверждать не статус
    пытаться:
        данные = q.get_nowait()
    кроме очереди. Пусто:
        print('Буфер пуст: увеличить размер буфера?', file=sys.stderr)
        поднять sd.CallbackAbort
    если len(данные) < len(outdata):
        outdata[:len(данные)] = данные
        outdata[len(data):] = b'\x00' * (len(outdata) - len(data))
        поднять sd.CallbackStop
    еще:
        outdata[:] = данные
пытаться:
    с sf.SoundFile(args.filename) как f:
        для _ в диапазоне (args.buffersize):
            данные = f.buffer_read (args.blocksize, dtype = 'float32')
            если не данные:
                сломать
            q.put_nowait(data) # Заполнить очередь
        поток = sd. RawOutputStream(
            частота дискретизации=f.samplerate, размер блока=args.blocksize,
            устройство = args.device, каналы = f.channels, dtype = 'float32',
            callback=обратный вызов, finish_callback=event.set)
        с потоком:
            тайм-аут = args.blocksize * args.buffersize / f.samplerate
            пока данные:
                данные = f.buffer_read (args.blocksize, dtype = 'float32')
                q.put (данные, время ожидания = время ожидания)
            event.wait() # Ждем окончания воспроизведения
кроме KeyboardInterrupt:
    parser.exit('\nПрервано пользователем')
кроме очереди.Полный:
    # Произошел таймаут, т.е. произошла ошибка в обратном вызове
    parser.exit(1)
кроме Исключения как e:
    parser.exit (тип (е). __ имя__ + ': ' + ул (е))
 

Сквозной ввод-вывод

wire.py

 #!/usr/bin/env python3
"""Передача ввода непосредственно на вывод.
https://app.assembla.com/spaces/portaudio/git/source/master/test/patest_wire. c
"""
импортировать аргументы
импортировать звуковое устройство как SD
import numpy # Убедитесь, что NumPy загружен, прежде чем он будет использоваться в обратном вызове
assert numpy # избегайте сообщения «импортировано, но не используется» (W0611)
определение int_or_str (текст):
    """Вспомогательная функция для разбора аргументов."""
    пытаться:
        вернуть целое (текст)
    кроме ValueError:
        возвращаемый текст
синтаксический анализатор = argparse.ArgumentParser (add_help = False)
parser.add_argument(
    '-l', '--list-devices', action='store_true',
    help='показать список аудиоустройств и выйти')
аргументы, осталось = parser.parse_known_args()
если args.list_devices:
    печать (sd.query_devices())
    парсер.выход(0)
синтаксический анализатор = argparse.ArgumentParser(
    описание=__doc__,
    formatter_class=argparse.RawDescriptionHelpFormatter,
    родители=[парсер])
parser.add_argument(
    '-i', '--input-device', тип=int_or_str,
    help='устройство ввода (числовой идентификатор или подстрока)')
parser. add_argument(
    '-o', '--output-device', тип=int_or_str,
    help='устройство вывода (числовой идентификатор или подстрока)')
parser.add_argument(
    '-c', '--channels', тип=int, по умолчанию=2,
    help='количество каналов')
parser.add_argument('--dtype', help='тип аудиоданных')
parser.add_argument('--samplerate', type=float, help='частота дискретизации')
parser.add_argument('--blocksize', type=int, help='размер блока')
parser.add_argument('--latency', type=float, help='задержка в секундах')
аргументы = parser.parse_args (оставшиеся)
Обратный вызов def (indata, outdata, кадры, время, статус):
    если статус:
        печать (статус)
    outdata[:] = indata
пытаться:
    с sd.Stream (устройство = (args.input_device, args.output_device),
                   частота дискретизации = args.samplerate, размер блока = args.blocksize,
                   dtype=args.dtype, задержка=args.latency,
                   каналы=args.channels, обратный вызов=обратный вызов):
        печать('#' * 80)
        print('Нажмите Enter, чтобы выйти')
        печать('#' * 80)
        вход()
кроме KeyboardInterrupt:
    parser. exit('')
кроме Исключения как e:
    parser.exit (тип (е). __ имя__ + ': ' + ул (е))
 

Спектрограмма в текстовом режиме в реальном времени

spectrogram.py

 #!/usr/bin/env python3
"""Показать спектрограмму в текстовом режиме с использованием данных с микрофона в реальном времени."""
импортировать аргументы
импортировать математику
импортный шутил
импортировать numpy как np
импортировать звуковое устройство как SD
use_line = 'нажмите  для выхода, + или - для изменения масштаба '
определение int_or_str (текст):
    """Вспомогательная функция для разбора аргументов."""
    пытаться:
        вернуть целое (текст)
    кроме ValueError:
        возвращаемый текст
пытаться:
    столбцы, _ = Shutil.get_terminal_size()
кроме AttributeError:
    столбцы = 80
синтаксический анализатор = argparse.ArgumentParser (add_help = False)
parser.add_argument(
    '-l', '--list-devices', action='store_true',
    help='показать список аудиоустройств и выйти')
аргументы, осталось = parser. parse_known_args()
если args.list_devices:
    печать (sd.query_devices())
    парсер.выход(0)
синтаксический анализатор = argparse.ArgumentParser(
    description=__doc__ + '\n\nПоддерживаемые ключи:' + строка_использования,
    formatter_class=argparse.RawDescriptionHelpFormatter,
    родители=[парсер])
parser.add_argument(
    '-b', '--block-duration', type=float, metavar='DURATION', по умолчанию=50,
    help='размер блока (по умолчанию %(по умолчанию)s миллисекунд)')
parser.add_argument(
    '-c', '--columns', тип=int, по умолчанию=столбцы,
    help='ширина спектрограммы')
parser.add_argument(
    '-d', '--device', тип=int_or_str,
    help='устройство ввода (числовой идентификатор или подстрока)')
parser.add_argument(
    '-g', '--gain', тип=число с плавающей запятой, по умолчанию=10,
    help='начальный коэффициент усиления (по умолчанию %(по умолчанию)s)')
parser.add_argument(
    '-r', '--range', type=float, nargs=2,
    метавар=('НИЗКИЙ', 'ВЫСОКИЙ'), по умолчанию=[100, 2000],
    help='диапазон частот (по умолчанию %(по умолчанию) с Гц)')
аргументы = parser. parse_args (оставшиеся)
низкий, высокий = args.range
если высокий <= низкий:
    parser.error('HIGH должен быть больше LOW')
# Создайте красивый выходной градиент, используя escape-последовательности ANSI.
# Украдено с https://gist.github.com/maurisvh/df919538bcef391bc89f
цвета = 30, 34, 35, 91, 93, 97
символы = ':%#\t#%:'
градиент = []
для bg, fg в zip (цвета, цвета [1:]):
    для символов в символах:
        если символ == '\t':
            бг, фг = фг, бг
        еще:
            градиент.append('\x1b[{};{}m{}'.format(fg, bg + 10, char))
пытаться:
    частота дискретизации = sd.query_devices(args.device, 'input')['default_samplerate']
    delta_f = (высокий - низкий) / (args.columns - 1)
    fftsize = math.ceil (частота дискретизации / delta_f)
    low_bin = math.floor(low / delta_f)
    Обратный вызов def (данные, кадры, время, статус):
        если статус:
            текст = ' ' + ул (статус) + ' '
            print('\x1b[34;40m', text.center(args.columns, '#'),
                  '\x1b[0m', сеп='')
        если есть (информация):
            величина = np. abs(np.fft.rfft(indata[:, 0], n=fftsize))
            величина *= args.gain / fftsize
            строка = (градиент [int (np.clip (x, 0, 1) * (len (градиент) - 1))]
                    для x по величине [low_bin:low_bin + args.columns])
            печать (*строка, sep='', конец='\x1b[0m\n')
        еще:
            распечатать('нет ввода')
    с sd.InputStream (устройство = args.device, каналы = 1, обратный вызов = обратный вызов,
                        blocksize = int (частота дискретизации * args.block_duration / 1000),
                        частота дискретизации = частота дискретизации):
        пока верно:
            ответ = ввод ()
            если ответ в ('', 'q', 'Q'):
                сломать
            для ch в ответ:
                если ch == '+':
                    args.gain *= 2
                Элиф ч == '-':
                    args.gain /= 2
                еще:
                    print('\x1b[31;40m', use_line.center(args.columns, '#'),
                          '\x1b[0m', сеп='')
                    сломать
кроме KeyboardInterrupt:
    parser. exit('Прервано пользователем')
кроме Исключения как e:
    parser.exit (тип (е). __ имя__ + ': ' + ул (е))
 

Запись произвольной продолжительности

rec_unlimited.py

 #!/usr/bin/env python3
"""Создать запись произвольной продолжительности.
Модуль звукового файла (https://PySoundFile.readthedocs.io/) должен быть установлен!
"""
импортировать аргументы
импортировать временный файл
очередь импорта
импорт системы
импортировать звуковое устройство как SD
импортировать звуковой файл как sf
import numpy # Убедитесь, что NumPy загружен, прежде чем он будет использоваться в обратном вызове
assert numpy # избегайте сообщения «импортировано, но не используется» (W0611)
определение int_or_str (текст):
    """Вспомогательная функция для разбора аргументов."""
    пытаться:
        вернуть целое (текст)
    кроме ValueError:
        возвращаемый текст
синтаксический анализатор = argparse.ArgumentParser (add_help = False)
parser.add_argument(
    '-l', '--list-devices', action='store_true',
    help='показать список аудиоустройств и выйти')
аргументы, осталось = parser. parse_known_args()
если args.list_devices:
    печать (sd.query_devices())
    парсер.выход(0)
синтаксический анализатор = argparse.ArgumentParser(
    описание=__doc__,
    formatter_class=argparse.RawDescriptionHelpFormatter,
    родители=[парсер])
parser.add_argument(
    'имя файла', nargs='?', metavar='ИМЯ ФАЙЛА',
    help='аудиофайл для сохранения записи')
parser.add_argument(
    '-d', '--device', тип=int_or_str,
    help='устройство ввода (числовой идентификатор или подстрока)')
parser.add_argument(
    '-r', '--samplerate', type=int, help='частота дискретизации')
parser.add_argument(
    '-c', '--channels', type=int, default=1, help='количество входных каналов')
parser.add_argument(
    '-t', '--subtype', type=str, help='подтип звукового файла (например, "PCM_24")')
аргументы = parser.parse_args (оставшиеся)
q = очередь.Очередь()
Обратный вызов def (данные, кадры, время, статус):
    """Это вызывается (из отдельного потока) для каждого аудиоблока."""
    если статус:
        печать (статус, файл = sys. stderr)
    q.put(indata.copy())
пытаться:
    если args.samplerate имеет значение None:
        device_info = sd.query_devices (args.device, 'ввод')
        # звуковой файл ожидает целое число, звуковое устройство предоставляет число с плавающей запятой:
        args.samplerate = int (device_info ['default_samplerate'])
    если args.filename имеет значение None:
        args.filename = tempfile.mktemp(prefix='delme_rec_unlimited_',
                                        суффикс='.wav', директор='')
    # Перед записью убедитесь, что файл открыт:
    с sf.SoundFile(args.filename, mode='x', samplerate=args.samplerate,
                      каналы=args.channels, подтип=args.subtype) в виде файла:
        с sd.InputStream(samplerate=args.samplerate, device=args.device,
                            каналы=args.channels, обратный вызов=обратный вызов):
            печать('#' * 80)
            print('нажмите Ctrl+C, чтобы остановить запись')
            печать('#' * 80)
            пока верно:
                файл. запись(q.get())
кроме KeyboardInterrupt:
    print('\nЗапись завершена: ' + repr(args.filename))
    парсер.выход(0)
кроме Исключения как e:
    parser.exit (тип (е). __ имя__ + ': ' + ул (е))
 

Использование потока в сопрограмме

asyncio

asyncio_coroutines.py

 #!/usr/bin/env python3
"""Пример использования потока в сопрограмме asyncio.
В этом примере показано, как создать поток в сопрограмме и как дождаться
завершение потока.
Вам нужен Python 3.7 или новее, чтобы запустить это.
"""
импортировать асинхронный
импорт системы
импортировать numpy как np
импортировать звуковое устройство как SD
async def record_buffer (буфер, ** kwargs):
    цикл = asyncio.get_event_loop()
    событие = asyncio.Event()
    идентификатор = 0
    Обратный вызов def (indata, frame_count, time_info, статус):
        нелокальный идентификатор
        если статус:
            печать (статус)
        остаток = len(буфер) - idx
        если остаток == 0:
            loop. call_soon_threadsafe(event.set)
            поднять sd.CallbackStop
        данные = данные [: остаток]
        буфер[idx:idx + len(indata)] = indata
        idx += len(indata)
    поток = sd.InputStream (обратный вызов = обратный вызов, dtype = буфер.dtype,
                            каналы=buffer.shape[1], **kwargs)
    с потоком:
        ожидайте событие.ждите()
асинхронное определение play_buffer (буфер, ** kwargs):
    цикл = asyncio.get_event_loop()
    событие = asyncio.Event()
    идентификатор = 0
    Обратный вызов def (outdata, frame_count, time_info, статус):
        нелокальный идентификатор
        если статус:
            печать (статус)
        остаток = len(буфер) - idx
        если остаток == 0:
            loop.call_soon_threadsafe(event.set)
            поднять sd.CallbackStop
        valid_frames = frame_count, если остаток >= frame_count, иначе остаток
        outdata[:valid_frames] = буфер[idx:idx + valid_frames]
        outdata[valid_frames:] = 0
        idx += действительные_кадры
    поток = sd. OutputStream(обратный вызов=обратный вызов, dtype=buffer.dtype,
                             каналы=buffer.shape[1], **kwargs)
    с потоком:
        ожидайте событие.ждите()
async def main (кадры = 150_000, каналы = 1, dtype = 'float32', ** kwargs):
    буфер = np.empty((кадры, каналы), dtype=dtype)
    print('буфер записи...')
    ожидание record_buffer(буфер, **kwargs)
    print('воспроизведение буфера...')
    ожидание play_buffer(буфер, **kwargs)
    распечатать('сделано')
если __name__ == "__main__":
    пытаться:
        asyncio.run(основной())
    кроме KeyboardInterrupt:
        sys.exit('\nПрервано пользователем')
 

Создание генератора

asyncio для аудиоблоков

asyncio_generators.py

 #!/usr/bin/env python3
"""Создание асинхронного генератора блоков аудиоданных.
Этот пример показывает, как генератор может использоваться, чтобы проанализировать блоки аудиовхода.
Кроме того, он показывает, как можно создать генератор, который дает не только входные
блоки, но также и выходные блоки, в которые можно записывать аудиоданные. Вам нужен Python 3.7 или новее, чтобы запустить это.
"""
импортировать асинхронный
очередь импорта
импорт системы
импортировать numpy как np
импортировать звуковое устройство как SD
async def inputstream_generator (каналы = 1, ** kwargs):
 """Генератор, который выдает блоки входных данных в виде массивов NumPy."""
 q_in = asyncio.Queue()
 цикл = asyncio.get_event_loop()
 Обратный вызов def (indata, frame_count, time_info, статус):
 loop.call_soon_threadsafe(q_in.put_nowait, (indata.copy(), статус))
 поток = sd.InputStream (обратный вызов = обратный вызов, каналы = каналы, ** kwargs)
 с потоком:
 пока верно:
 indata, статус = ожидание q_in.get()
 выходная информация, статус
async def stream_generator (размер блока, *, каналы = 1, dtype = 'float32',
 pre_fill_blocks=10, **kwargs):
 """Генератор, который выдает блоки входных/выходных данных в виде массивов NumPy.
 Выходные блоки не инициализированы и должны быть заполнены
 соответствующие звуковые сигналы. """
 утверждать размер блока != 0
 q_in = asyncio.Queue()
 q_out = очередь.Очередь()
 цикл = asyncio.get_event_loop()
 Обратный вызов def (indata, outdata, frame_count, time_info, статус):
 loop.call_soon_threadsafe(q_in.put_nowait, (indata.copy(), статус))
 outdata[:] = q_out.get_nowait()
 # предварительно заполнить очередь вывода
 для _ в диапазоне (pre_fill_blocks):
 q_out.put(np.zeros((размер блока, каналы), dtype=dtype))
 поток = sd.Stream (размер блока = размер блока, обратный вызов = обратный вызов, dtype = dtype,
 каналы=каналы, **kwargs)
 с потоком:
 пока верно:
 indata, статус = ожидание q_in.get()
 outdata = np.empty((размер блока, каналы), dtype=dtype)
 выход indata, outdata, статус
 q_out.put_nowait (устаревшие данные)
асинхронное определение print_input_infos (** kwargs):
 """Показать минимальное и максимальное значение каждого входящего аудиоблока. """
 async для indata, статус в inputstream_generator(**kwargs):
 если статус:
 печать (статус)
 print('min:', indata.min(), '\t', 'max:', indata.max())
асинхронное определение wire_coro(**kwargs):
 """Создайте соединение между аудио входами и выходами.
 Асинхронно перебирает генератор потока и каждый блок
 просто копирует входные данные в выходной блок.
 """
 async для indata, outdata, статуса в stream_generator(**kwargs):
 если статус:
 печать (статус)
 outdata[:] = indata
асинхронное определение основного (** kwargs):
 print('Некоторая информация о входном сигнале:')
 пытаться:
 ожидание asyncio.wait_for(print_input_infos(), время ожидания=2)
 кроме asyncio.TimeoutError:
 проходят
 print('\nХватит, активация провода...\n')
 audio_task = asyncio.create_task(wire_coro(**kwargs))
 для i в диапазоне (10, 0, -1):
 печать (я)
 ждать asyncio.sleep(1)
 audio_task.cancel()
 пытаться:
 ждать audio_task
 кроме asyncio.