Zenn
Closed11

Pythonのsounddeviceを改めて試す

kun432kun432

Pythonでサウンド周りを使うときはPyAudioをよく使っているのだけども、

  • プチノイズが入りやすい気がする
  • 環境依存で動かなかったりすることがある

あたりにちょっと悩みがあって、見様見真似でsounddeviceに変えてみたら改善したので、改めて試してみようと思う。

GitHubレポジトリ

https://github.com/spatialaudio/python-sounddevice

Pythonで音声を再生・録音する

このPythonモジュールは、PortAudioライブラリのバインディングと、音声信号を含むNumPy配列を再生および録音するための便利な関数をいくつか提供します。

sounddeviceモジュールは、Linux、macOS、Windowsで利用可能です。

ドキュメント:
https://python-sounddevice.readthedocs.io/

ソースコードリポジトリとイシュートラッカー:
https://github.com/spatialaudio/python-sounddevice/

ライセンス:
MIT -- 詳細はLICENSEファイルを参照してください。

kun432kun432

インストール

https://python-sounddevice.readthedocs.io/en/0.5.1/installation.html

ローカルのMacで試す。

PortAudioのバインディングなので、PortAudioライブラリをインストールしておく必要があると思う。自分は既にインストール済みだった。

brew install portaudio

Python仮想環境を作成してパッケージインストール。

uv init -p 3.12.9 sounddevice-work && cd sounddevice-work
uv add sounddevice
出力
 + sounddevice==0.5.1
kun432kun432

再生

https://python-sounddevice.readthedocs.io/en/0.5.1/usage.html#playback

sounddeviceはオーディオをNumPy配列として扱う。オーディオ「ファイル」を直接扱うインタフェースはない様子。よってオーディオファイルを再生する場合は別のライブラリでオーディオファイルを読み込んでNumPy配列にする必要がある。

今回は色々なファイル形式で読み込めるsoundfileを使う。

uv add soundfile
 + numpy==2.2.4
 + soundfile==0.13.1

再生

play.py
import sounddevice as sd
import soundfile as sf

# オーディオファイルを読み込んで、
# オーディオデータのnumpy.ndarrayと、サンプルレートを返す
data, fs = sf.read("sample.wav")

print("サンプルレート:" , fs)
print("データ:")
print("  データ型:", type(data))
print("  フレーム数:", len(data))
print("  データ長(秒):", len(data) / fs)
print("  データ抜粋:", data[:10])

# データとサンプルレートを渡して再生
sd.play(data, fs)

# サンプルレードのデフォルトを設定することもできる
# ```
# sd.default.samplerate = fs
# ```
# この場合はplayでサンプルレートを渡さなくてもよい。

# 再生が最後まで再生完了されるまで待つ
sd.wait()

実行

uv run play.py
出力
サンプルレート: 16000
データ:
  データ型: <class 'numpy.ndarray'>
  フレーム数: 115648
  データ長(秒): 7.228
  データ抜粋: [-0.0007019  -0.00085449 -0.00082397 -0.00082397 -0.00067139 -0.00048828
 -0.00036621 -0.00030518 -0.00021362 -0.00018311]

オーディオが再生されるのが聞こえるはず。

kun432kun432

デバイスの確認

https://python-sounddevice.readthedocs.io/en/0.5.1/usage.html#device-selection

録音に進む前に、使用するデバイスについて確認。query_devices()で一覧が取得できるが、ワンライナーだとモジュールを読み込むだけで確認できる。

uv run python -m sounddevice
出力
   0 PHL 328P6VU, Core Audio (0 in, 2 out)
   1 iPhoneのマイク, Core Audio (1 in, 0 out)
   2 FHD Camera Microphone, Core Audio (2 in, 0 out)
   3 Jabra SPEAK 510 USB, Core Audio (0 in, 2 out)
>  4 Jabra SPEAK 510 USB, Core Audio (1 in, 0 out)
   5 BlackHole 2ch, Core Audio (2 in, 2 out)
<  6 外部ヘッドフォン, Core Audio (0 in, 2 out)
   7 Mac miniのスピーカー, Core Audio (0 in, 2 out)
   8 VB-Cable, Core Audio (2 in, 2 out)
   9 ZoomAudioDevice, Core Audio (2 in, 2 out)
  10 機器セット, Core Audio (3 in, 2 out)
  11 複数出力装置, Core Audio (0 in, 2 out)

自分の環境はいろいろ仮想デバイスなども入っていてややこしいのだが、>が現在選択されている出力デバイスで、<が現在選択されている入力デバイスになる。入出力で分けてほしい気がする。

より詳細な情報を取得してみる。

devices.py
import sounddevice as sd
import json

# デバイス一覧の表示
for idx, d in enumerate(sd.query_devices()):
    print(idx, ":", json.dumps(d, indent=2, ensure_ascii=False))

# デフォルトのデバイス
print("デフォルト入力デバイスID:", sd.default.device[0])
print("デフォルト出力デバイスID:", sd.default.device[1])

# デフォルトのサンプルレート
print("デフォルトサンプルレート:", sd.default.samplerate)

# デフォルトのチャンネル数
print("デフォルトチャンネル数:", sd.default.channels)
uv run devices.py
出力
0 : {
  "name": "PHL 328P6VU",
  "index": 0,
  "hostapi": 0,
  "max_input_channels": 0,
  "max_output_channels": 2,
  "default_low_input_latency": 0.01,
  "default_low_output_latency": 0.009833333333333333,
  "default_high_input_latency": 0.1,
  "default_high_output_latency": 0.019166666666666665,
  "default_samplerate": 48000.0
}
1 : {
  "name": "k432i (2)のマイク",
  "index": 1,
  "hostapi": 0,
  "max_input_channels": 1,
  "max_output_channels": 0,
  "default_low_input_latency": 0.12841666666666668,
  "default_low_output_latency": 0.01,
  "default_high_input_latency": 0.13775,
  "default_high_output_latency": 0.1,
  "default_samplerate": 48000.0
}
2 : {
  "name": "FHD Camera Microphone",
  "index": 2,
  "hostapi": 0,
  "max_input_channels": 2,
  "max_output_channels": 0,
  "default_low_input_latency": 0.004583333333333333,
  "default_low_output_latency": 0.01,
  "default_high_input_latency": 0.013916666666666667,
  "default_high_output_latency": 0.1,
  "default_samplerate": 48000.0
}
3 : {
  "name": "Jabra SPEAK 510 USB",
  "index": 3,
  "hostapi": 0,
  "max_input_channels": 0,
  "max_output_channels": 2,
  "default_low_input_latency": 0.01,
  "default_low_output_latency": 0.004354166666666667,
  "default_high_input_latency": 0.1,
  "default_high_output_latency": 0.0136875,
  "default_samplerate": 48000.0
}
4 : {
  "name": "Jabra SPEAK 510 USB",
  "index": 4,
  "hostapi": 0,
  "max_input_channels": 1,
  "max_output_channels": 0,
  "default_low_input_latency": 0.008375,
  "default_low_output_latency": 0.01,
  "default_high_input_latency": 0.036375,
  "default_high_output_latency": 0.1,
  "default_samplerate": 16000.0
}
5 : {
  "name": "BlackHole 2ch",
  "index": 5,
  "hostapi": 0,
  "max_input_channels": 2,
  "max_output_channels": 2,
  "default_low_input_latency": 0.01,
  "default_low_output_latency": 0.0013333333333333333,
  "default_high_input_latency": 0.1,
  "default_high_output_latency": 0.010666666666666666,
  "default_samplerate": 48000.0
}
6 : {
  "name": "外部ヘッドフォン",
  "index": 6,
  "hostapi": 0,
  "max_input_channels": 0,
  "max_output_channels": 2,
  "default_low_input_latency": 0.01,
  "default_low_output_latency": 0.004875,
  "default_high_input_latency": 0.1,
  "default_high_output_latency": 0.014208333333333333,
  "default_samplerate": 48000.0
}
7 : {
  "name": "Mac miniのスピーカー",
  "index": 7,
  "hostapi": 0,
  "max_input_channels": 0,
  "max_output_channels": 2,
  "default_low_input_latency": 0.01,
  "default_low_output_latency": 0.012916666666666667,
  "default_high_input_latency": 0.1,
  "default_high_output_latency": 0.02225,
  "default_samplerate": 48000.0
}
8 : {
  "name": "VB-Cable",
  "index": 8,
  "hostapi": 0,
  "max_input_channels": 2,
  "max_output_channels": 2,
  "default_low_input_latency": 0.01,
  "default_low_output_latency": 0.0013333333333333333,
  "default_high_input_latency": 0.1,
  "default_high_output_latency": 0.010666666666666666,
  "default_samplerate": 48000.0
}
9 : {
  "name": "ZoomAudioDevice",
  "index": 9,
  "hostapi": 0,
  "max_input_channels": 2,
  "max_output_channels": 2,
  "default_low_input_latency": 0.01,
  "default_low_output_latency": 0.03333333333333333,
  "default_high_input_latency": 0.1,
  "default_high_output_latency": 0.042666666666666665,
  "default_samplerate": 48000.0
}
10 : {
  "name": "機器セット",
  "index": 10,
  "hostapi": 0,
  "max_input_channels": 3,
  "max_output_channels": 2,
  "default_low_input_latency": 0.01,
  "default_low_output_latency": 0.0013333333333333333,
  "default_high_input_latency": 0.1,
  "default_high_output_latency": 0.010666666666666666,
  "default_samplerate": 48000.0
}
11 : {
  "name": "複数出力装置",
  "index": 11,
  "hostapi": 0,
  "max_input_channels": 0,
  "max_output_channels": 2,
  "default_low_input_latency": 0.01,
  "default_low_output_latency": 0.004375,
  "default_high_input_latency": 0.1,
  "default_high_output_latency": 0.013708333333333333,
  "default_samplerate": 48000.0
}
デフォルト入力デバイスID: 4
デフォルト出力デバイスID: 6
デフォルトサンプルレート: None
デフォルトチャンネル数: [None, None]

あと、こんな使い方もできるみたい。

import sounddevice as sd

print("デバイスID:0 のデバイス:", sd.query_devices(0)["name"])
print("現在の入力デバイス:", sd.query_devices(None, 'input')["name"])
print("現在の出力デバイス:", sd.query_devices(None, 'output')["name"])
出力
デバイスID:0 のデバイス: PHL 328P6VU
現在の入力デバイス: Jabra SPEAK 510 USB
現在の出力デバイス: 外部ヘッドフォン

query_devices()

  1. 引数なしだと全デバイスを返す
  2. 引数にデバイスIDを渡すとそのデバイス情報を返す
  3. 引数にデバイスIDとinput or outputを渡すと、そのデバイスの入力 or 出力情報だけを返す

という感じらしい。で、デバイスIDをNoneにしてinput or outputを指定すると、入出力のそれぞれで現在選択されているデバイスが返される。

なお、入力しかできないデバイスIDとoutput、出力しかできないデバイスとinput、を渡すとエラーになる。

# 自分の環境ではデバイス0は出力のみのデバイス
print(sd.query_devices(0, "input"))
結果
ValueError: Not an input device: 'PHL 328P6VU'
kun432kun432

録音

https://python-sounddevice.readthedocs.io/en/0.5.1/usage.html#recording

record.py
import sounddevice as sd
import soundfile as sf

# 録音設定
duration = 5  # 録音時間(秒)
output_file = 'recorded.wav'  # 出力ファイル名
# 以下は入力デバイスに合わせて設定
samplerate = 16000  # サンプリングレート(Hz)
channels = 1        # モノラル(ステレオなら2)

# 録音開始。バックグラウンドで実行される。
print("{}秒間の録音を開始しました。".format(duration))
recording = sd.rec(
    int(duration * samplerate),
    samplerate=samplerate,
    channels=channels
)
# 録音完了まで待つ
sd.wait()
print("録音終了。ファイル {} に保存しました。".format(output_file))

# ファイルに保存(WAV形式)
sf.write(output_file, recording, samplerate)

実行

uv run record.py
出力
5秒間の録音を開始しました。
録音終了。ファイル recorded.wav に保存しました。

afplayやffprobeで確認

afplay recorded.wav
ffprobe recorded.wav
出力
Input #0, wav, from 'recorded.wav':
  Duration: 00:00:05.00, bitrate: 256 kb/s
  Stream #0:0: Audio: pcm_s16le ([1][0][0][0] / 0x0001), 16000 Hz, 1 channels, s16, 256 kb/s

なお、rec()が返すオーディオ配列のデータ型は、デフォルトだとfloat32となる。これを変更するにはdtypeパラメータで指定する。

recording = sd.rec(
    int(duration * samplerate),
    samplerate=samplerate,
    channels=channels,
    dtype='float64'
)

最初の例では、入力デバイスの設定を個別に設定していたが、自動で設定させるならこんな感じ。

import sounddevice as sd
import soundfile as sf

# 録音設定
duration = 5  # 録音時間(秒)
output_file = 'recorded.wav'  # 出力ファイル名

# 現在選択されている入力デバイスの設定を取得
input_info = sd.query_devices(None, 'input')
print("現在の録音デバイス:", input_info['name'])
print("- サンプリングレート:", input_info['default_samplerate'], "Hz")
print("- チャンネル数:", input_info['max_input_channels'])

# 録音
print("{}秒間の録音を開始しました。".format(duration))
recording = sd.rec(
    int(duration * input_info['default_samplerate']),
    samplerate=int(input_info['default_samplerate']),
    channels=input_info['max_input_channels']
)
sd.wait()
print("録音終了。ファイル {} に保存します。".format(output_file))

# ファイルに保存(WAV形式)
sf.write(output_file, recording, int(input_info['default_samplerate']))
kun432kun432

再生と録音を同時に行う

https://python-sounddevice.readthedocs.io/en/0.5.1/usage.html#simultaneous-playback-and-recording

音声を再生しながら録音するといっても、再生したものが録音に含まれる「ミックス」というわけでは無いし、録音したものをすぐに再生する「モニター」の感じでもない。単に再生と録音が同期されるだけ。

いまいちユースケースがわからないけども、例えば、何かしらのタイミングが重要な音源があって、それにあわせたものを録音する、で後でミックスする、みたいな使い方になるのかなという気がする。

playrec.py
import numpy as np
import sounddevice as sd
import soundfile as sf

# オーディオファイルを読み込む。48000Hz・2チャンネルのWAVファイル
data, fs = sf.read("sample-2ch-48000hz-15secs.wav")

print("音声ファイル:")
print("- サンプリングレート:", int(fs), "Hz")
print("- チャンネル数:", data.shape[1] if data.ndim > 1 else 1)

input_info = sd.query_devices(None, 'input')
print("現在の録音デバイス:")
print("- サンプリングレート:", int(input_info['default_samplerate']), "Hz")
print("- チャンネル数:", input_info['max_input_channels'])

output_info = sd.query_devices(None, 'input')
print("現在の再生デバイス:")
print("- サンプリングレート:", int(output_info['default_samplerate']), "Hz")
print("- チャンネル数:", output_info['max_input_channels'])

print("再生された音声を聞きながら、録音を開始します。")
# 再生しながら録音
# - 録音時のフレームレートは再生するWAVファイルと同じフレームレートになる
# - 録音時間は再生するWAVファイルと同じ時間になる
# - 録音時のチャネル数だけは指定する必要がある
recording = sd.playrec(data, channels=output_info['max_input_channels']) 
sd.wait()
print("録音完了")

# ファイルに保存
sf.write('playrec.wav', recording, fs)

実行

uv run playrec.py
出力
音声ファイル:
- サンプリングレート: 48000 Hz
- チャンネル数: 2
現在の録音デバイス:
- サンプリングレート: 48000 Hz
- チャンネル数: 2
現在の再生デバイス:
- サンプリングレート: 48000 Hz
- チャンネル数: 2
再生された音声を聞きながら、録音を開始します。
録音完了

オーディオファイルの音声が再生されつつ、録音が行われる。ただし、上にも書いた通り、再生と録音はミックスされるわけではない。録音されたものにはマイクからの音声だけが入っている状態。

で、上で入力ファイル・録音デバイス・再生デバイスのサンプリングレート・チャネル数を表示しているが、再生と録音を同期的に行うため少なくとも入出力デバイスについては「サンプリングレートは揃っている必要がある」という点が重要。(音声ファイルのデータはリサンプリングすればいい)

例えば異なるサンプリングレートのマイクを使用した場合は以下となる。

出力
ValueError: Input and output device must have the same samplerate
kun432kun432

ストリームをコールバックで扱う

ここまでに出てきた play() / rec() / playrec() / wait() / stop() は、小さなスクリプトで対話的に使うなら便利だが、連続録音やリアルタイム処理などより細かい制御が必要なケースではより低レベルなクラスである以下の

  • Stream
  • InputStream
  • OutputStream

を使うと、コールバックを使って非同期で処理することができる。

Streamクラスは、録音・再生の双方向ストリームを作成することができる。ただし、入出力デバイスのサンプルレートは同じである必要がある。 以下は録音した音声をそのまま再生するサンプル。

stream.py
import sounddevice as sd

# 録音時間 (秒)
duration = 5

def callback(indata, outdata, frames, time, status):
   """ 入力音声を出力音声に転送するコールバック関数 """
   if status:
       print("status:", status)
   outdata[:] = indata

# Streamクラスを使用して、録音と再生を非同期で同時に行う 
with sd.Stream(channels=2, callback=callback):
   sd.sleep(int(duration * 1000)) # 録音時間分待つ

RawStreamを使うと、NumPy配列ではなく、生のバイナリデータを扱うことができる。ちょっとユースケースが自分にはわからないのだけども、

  • NumPyを使わない・使えない場合
  • 独自のバイナリフォーマットを使う場合
  • 24ビットPCM を正確に扱いたい(WAVではなく生データとして)

あたりになるみたい、

rawstream.py
import sounddevice as sd

# 録音時間 (秒)
duration = 5

def callback(indata, outdata, frames, time, status):
    """ 入力音声を出力音声に転送するコールバック関数 """
    if status:
        print("status:", status)
    outdata[:] = indata

# RawStreamクラスを使用して、録音と再生を同時に行う 
# dtypeで扱うデータの種類を指定できる
with sd.RawStream(channels=2, callback=callback, dtype='int24'):
    sd.sleep(int(duration * 1000)) # 録音時間分待つ

、入出力をStream/RawStreamでひとまとめにしているが、以下を使えば入出力ストリームを別々に処理することができる。

  • InputStream: 入力のみ・NumPyで扱う
  • OutputStream: 出力のみ・NumPyで扱う
  • RawInputStream: 入力のみ・生データで扱う
  • RawOutputStream: 出力のみ・生データで扱う

ここは後でまた試す


なお、上記のStream / RawStream のどちらのサンプルコードも、実行するとほぼ必ず直後に

出力
status: input overflow

が表示されると思う。

これは、with sd.Streamでストリームが開始しているが、一番最初のコールバックが実行されるまでの間も録音デバイスからストリームが送られてきて、最初の数フレームが処理できずにこぼれてしまうため。

いろいろな回避策はあるようなのだけど、完璧なものはなさそうに思えるので、実行直後にはそういうことは起きうると思っておけば良さそう。逆に、実行直後以外で頻度が高く起きるとかの場合に何かしら問題があることになると思う。

kun432kun432

ストリームをブロッキングで扱う

Streamクラスにはブロッキングメソッドも用意されている。read()メソッド・write()メソッドを使う。

再生

blocking-stream-read.py
import sounddevice as sd
import soundfile as sf

filename = 'sample-2ch-48000hz-15secs.wav'
# sf.read はデフォルトで dtype='float64' を返す
# sd.Stream はデフォルトで dtype='float32' で再生する
# 明示的に dtype='float32' を指定して変換する
data, fs = sf.read(filename, dtype='float32')
channel = data.shape[1] if data.ndim > 1 else 1

with sd.Stream(samplerate=fs, channels=channel) as stream:
    print("再生中...")
    stream.write(data)
    print("再生完了")

録音

blocking-stream-write.py
import sounddevice as sd
import soundfile as sf

# 入力デバイスの設定を取得
input_info = sd.query_devices(None, 'input')
samplerate = int(input_info['default_samplerate'])
channels = input_info['max_input_channels']

# 録音時間 (秒)
duration = 5

# 録音フレーム数
frames = int(samplerate * duration)

# 空の録音バッファを用意
recording = sd.Stream(samplerate=samplerate, channels=channels)

# 録音
with recording:
    print("録音開始...")
    audio, _ = recording.read(frames)  # read() は (data, overflow) を返す
    print("録音完了")

# 保存
sf.write('recording.wav', audio, samplerate)
kun432kun432

入出力デバイスが同じサンプルレートで処理してくれるなら sd.Streamsd.RawStream) は楽でいいけども、環境によってサンプルレートが違うというのはまあ普通にありそうなので、sd.InputStream / sd.OutputStreamで使い分けるのが良さそう。

kun432kun432

InputStream / OutputStream

OutputStreamを使った再生。リサンプリングしてるので、uv add scipyが必要。

import sounddevice as sd
import soundfile as sf
import numpy as np
from scipy.signal import resample

# WAVファイルを読み込む
filename = 'sample.wav'
data, file_fs = sf.read(filename, dtype='float32')
file_channels = data.shape[1] if data.ndim > 1 else 1

# 出力デバイス情報の取得
output_info = sd.query_devices(None, 'output')
device_fs = int(output_info['default_samplerate'])
device_channels = output_info['max_output_channels']

print(f"ファイル: {filename}")
print(f"- サンプリングレート: {file_fs} Hz")
print(f"- チャンネル数: {file_channels}")

print(f"出力デバイス: {output_info['name']}")
print(f"- サンプリングレート: {device_fs} Hz")
print(f"- チャンネル数: {device_channels}")

# サンプルレートの確認
if file_fs != device_fs:
    print("WAVファイルと出力デバイスのサンプルレートが一致しません。")
    print("WAVファイルのデータをリサンプリングします。")
    num_samples = int(len(data) * device_fs / file_fs)
    data = resample(data, num_samples)

# チャンネル調整(必要に応じて拡張 or 切り詰め)
if file_channels != device_channels:
    print("WAVファイルと出力デバイスのチャンネル数が一致しません。")
    print("WAVファイルのチャンネル数を出力デバイスのチャンネル数に合わせます。")
    if file_channels < device_channels:
        print("モノラルをステレオに変換します。")
        # (N,) → (N, 1) に reshape してから tile
        if data.ndim == 1:
            data = data.reshape(-1, 1)
        data = np.tile(data, (1, device_channels))
    elif file_channels > device_channels:
        print("多すぎるチャンネル数をカットします。")
        data = data[:, :device_channels]

# 再生位置
frame_index = 0

# コールバック関数
def callback(outdata, frames, time, status):
    global frame_index
    if status:
        print("Status:", status)

    end = frame_index + frames
    if end > len(data):
        # 受け取ったバッファが元のオーディオデータよりも大きい場合は、残りをゼロ埋めして再生終了
        outdata[:len(data) - frame_index] = data[frame_index:]
        outdata[len(data) - frame_index:] = 0  # 残りをゼロ埋め(無音)
        raise sd.CallbackStop
    else:
        # 受け取ったバッファが元のオーディオデータより小さい場合は、再生
        outdata[:] = data[frame_index:end]
        frame_index = end

# OutputStreamで再生
with sd.OutputStream(samplerate=device_fs,
                     channels=device_channels,
                     dtype='float32',
                     callback=callback):
    print("再生中...")
    sd.sleep(int(len(data) / device_fs * 1000))  # 音声の長さ分待機
    print("再生完了")

InputStreamを使った録音

import sounddevice as sd
import soundfile as sf
import numpy as np

duration = 5  # 録音時間(秒)
output_filename = 'recoding.wav'

# 入力デバイス情報を取得
input_info = sd.query_devices(None, 'input')
samplerate = int(input_info['default_samplerate'])
channels = input_info['max_input_channels']

print(f"入力デバイス: {input_info['name']}")
print(f"- サンプリングレート: {samplerate} Hz")
print(f"- チャンネル数: {channels}")

# 録音バッファを用意
frames = int(samplerate * duration)
recording = np.empty((frames, channels), dtype='float32')
frame_index = 0  # 録音位置

# コールバック関数
def callback(indata, frames, time, status):
    global frame_index
    if status:
        print("Status:", status)

    end = frame_index + frames
    if end > len(recording):
        indata = indata[:len(recording) - frame_index]
        end = len(recording)
        recording[frame_index:end] = indata
        frame_index = end
        raise sd.CallbackStop
    else:
        recording[frame_index:end] = indata
        frame_index = end

# 録音開始
with sd.InputStream(samplerate=samplerate,
                    channels=channels,
                    dtype='float32',
                    callback=callback):
    print("録音中...")
    sd.sleep(int(duration * 1000))
    print("録音完了")

# 録音結果を保存
sf.write(output_filename, recording[:frame_index], samplerate)
print(f"'{output_filename}' に保存しました")

入力をそのまま出力するパススルーの例。Streamではなく、InputStream / OutputStreamの組み合わせで。

import sounddevice as sd
from scipy.signal import resample_poly
from queue import Queue

duration = 10  # 秒数
buffer_size = 1024  # フレーム数単位で処理

# 入力デバイス設定
input_info = sd.query_devices(None, 'input')
input_sr = int(input_info['default_samplerate'])
input_channels = input_info['max_input_channels']

# 出力デバイス設定
output_info = sd.query_devices(None, 'output')
output_sr = int(output_info['default_samplerate'])
output_channels = output_info['max_output_channels']

# 共通チャネル数(安全側で調整)
channels = min(input_channels, output_channels)

print(f"入力: {input_info['name']}")
print(f"- {input_sr} Hz")
print(f"- {input_channels} ch")
print(f"出力: {output_info['name']}")
print(f"- {output_sr} Hz")
print(f"- {output_channels} ch")

# 音声バッファ
audio_queue = Queue()

# 入力コールバック
def input_callback(indata, frames, time, status):
    if status:
        print("Input status:", status)

    # チャンネル合わせ(必要ならカット)
    indata = indata[:, :channels]

    # リサンプリング(必要なら)
    if input_sr != output_sr:
        indata = resample_poly(indata, output_sr, input_sr, axis=0)

    audio_queue.put(indata.copy())

# 出力コールバック
def output_callback(outdata, frames, time, status):
    if status:
        print("Output status:", status)

    if not audio_queue.empty():
        chunk = audio_queue.get()
        out_len = outdata.shape[0]
        chunk_len = chunk.shape[0]

        if chunk_len < out_len:
            # 足りなければ無音で埋める
            outdata[:chunk_len] = chunk
            outdata[chunk_len:] = 0
        else:
            outdata[:] = chunk[:out_len]
            # 余ったぶんは再度キューに戻す
            remaining = chunk[out_len:]
            if remaining.shape[0] > 0:
                audio_queue.queue.appendleft(remaining)
    else:
        outdata[:] = 0  # バッファが空なら無音出力

# ストリームを個別に起動
with sd.InputStream(samplerate=input_sr,
                    channels=channels,
                    dtype='float32',
                    callback=input_callback,
                    blocksize=buffer_size), \
     sd.OutputStream(samplerate=output_sr,
                     channels=channels,
                     dtype='float32',
                     callback=output_callback,
                     blocksize=buffer_size):
    print(f"パススルー中...({duration}秒)")
    sd.sleep(duration * 1000)
    print("終了")
このスクラップは19日前にクローズされました
ログインするとコメントできます