Как я могу прочитать аудио файл, используя потоки?

Question

В настоящее время я ищу для потокового аудио-файлов. Я хотел бы прочитать x секунд в указанном WAV-файле, выполнить анализ и повторить .....

Вот код, чтобы понять, что я хочу:

`read_x_seconds = 30
 file_length_in_min = 15
 for x in range(file_length_in_min * (60 / read_x_seconds)):
    y, fs = librosa.core.load(FILENAME, offset=x * read_x_seconds,
    duration=read_x_seconds)
    do_analysis(y, fs)`

Answer 1

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

import wave
import numpy as np

def read_wav_part_from_local(path: str, start_s: float, duration_s: float):
    with wave.open(path, mode='rb') as wavread:
        fs = wavread.getframerate()
        start = int(start_s * fs)
        duration = int(duration_s * fs)
        wavread.setpos(start)
        wav_bytes = wavread.readframes(duration)

        if wavread.getsampwidth() == 2:
            dtype = 'int16'
        elif wavread.getsampwidth() == 4:
            dtype = 'int32'
        else:
            raise NotImplemented('I give up!')

        wav_array = np.frombuffer(wav_bytes, dtype=dtype)
        return wav_array, fs

Как его использовать:

audio_chunk, fs = read_wav_part_from_local('your.wav', offset_in_s, duration_in_s)

Answer 2

У меня есть два решения для блочного чтения / потоковой передачи файлов WAV.

Это номер один. Я написал это сам, так что не передавайте это.

def stream_gen(path: str):
    WINDOW_s = 10
    HEADER = 44

    bytes_per_sampling: int
    samplerate: int
    CHUNk: int

    first_block = True
    run = True

    with open(path, 'rb') as stream:
                data = stream.read(HEADER)
                samplerate = int.from_bytes(data[24:28], byteorder='little')
                bits_per_sampling = int.from_bytes(data[34:36], byteorder='little')

                if bits_per_sampling == 16:
                    dtype = 'int16'
                elif bits_per_sampling == 32:
                    dtype = 'int32'
                else:
                    raise IOError()

                CHUNK = WINDOW_s * samplerate * (bits_per_sampling // 8)

                while run:
                    data = stream.read(CHUNK)
                    if data == b'':
                        break
                    yield(np.frombuffer(data, dtype=dtype))

Номер два - очевидный выбор. Это было написано профессионалами.

def soundfile_gen(path):
    window_s = 10
    samplerate = sf.info(path).samplerate
    blocksize = samplerate * window_s
    block_gen = sf.blocks(path, blocksize=blocksize)
    return block_gen

Answer 3

with open(stream_file, 'rb') as audio_file:
    content = audio_file.read(BYTES_PER_SECOND)