ちょっと前に書いた https://yakimaguro.com/p/voicevox-streaming-trial voicevox のストリーミングと音声認識を組み合わせて
Discord で AI とボイスチャットができないか調べた。
前提条件が CPU なのが厳しいところ。
Discord から音声を拾う
これは pycord を使えばできた。
とりあえず全体のコードを
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
|
class VoiceListener:
def __init__(self, bot: discord.Bot):
self.bot = bot
self.is_listening = False
self.voice_client: discord.VoiceClient = None
def start_listening(self, voice_client: discord.VoiceClient):
if self.is_listening:
print("Already listening.")
return
self.voice_client = voice_client
self.voice_client.empty_socket()
self.voice_client.decoder = discord.opus.DecodeManager(self)
self.voice_client.decoder.start()
self.is_listening = True
t = threading.Thread(target=self.recv_audio)
t.start()
def recv_decoded_audio(self, data: discord.sinks.core.RawData):
while data.ssrc not in self.voice_client.ws.ssrc_map:
time.sleep(0.01)
pcm_audio = data.decoded_data
def recv_audio(self):
while self.is_listening:
ready, _, err = select.select(
[self.voice_client.socket], [], [self.voice_client.socket], 0.01
)
if not ready:
if err:
print(f"Socket error: {err}")
continue
try:
data = self.voice_client.socket.recv(4096)
except OSError:
continue
self.voice_client.unpack_audio(data)
async def stop_listening(self):
if not self.is_listening:
print("Not listening.")
return
self.is_listening = False
self.voice_client.decoder.stop()
await self.voice_client.disconnect()
self.voice_client = None
print("Stopped listening.")
|
まずボイスチャンネルに接続されたら start_listening を呼び出だす。
そして recv_audic で音声を拾う。
ちなみに、recv_decoded_audio はどこで呼び出されるのかというと、これは pycord のソースコード(opus.py)を読めばわかるが、
1
|
self.voice_client.decoder = discord.opus.DecodeManager(self)
|
まず DecodeManager のインスタンスを生成するときに、self を引数で渡しているのがポイントで、こうすると
1
2
3
4
5
6
7
8
9
10
|
class DecodeManager(threading.Thread, _OpusStruct):
def __init__(self, client):
super().__init__(daemon=True, name="DecodeManager")
self.client = client
self.decode_queue = []
self.decoder = {}
self._end_thread = threading.Event()
|
(opus.py 521-530)
の client に VoiceListener 自身が入る。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
def run(self):
while not self._end_thread.is_set():
try:
data = self.decode_queue.pop(0)
except IndexError:
time.sleep(0.001)
continue
try:
if data.decrypted_data is None:
continue
else:
data.decoded_data = self.get_decoder(data.ssrc).decode(
data.decrypted_data
)
except OpusError:
print("Error occurred while decoding opus frame.")
continue
self.client.recv_decoded_audio(data)
|
一番下に
1
|
self.client.recv_decoded_audio(data)
|
と書いてあるが、この時 self.client は VoiceListener のことなので VoiceListener の中にある recv_decoded_audio が実行される。
これでひとまず Discord の音声を拾うことができた。
音声認識
色々探したが以下が見つかった。
とりあえず CPU で軽そうな vosk を使ってみることにした。
まず Discord から拾った音声を PCM 16bit のモノラル音声にしないといけないので
1
2
3
4
5
6
7
8
9
10
|
def downsample_to_16k_mono(self, pcm_audio: bytes) -> bytes:
audio_np = np.frombuffer(pcm_audio, dtype=np.int16)
if len(audio_np) % 2 == 0:
audio_np = audio_np.reshape(-1, 2)
audio_np = audio_np.mean(axis=1).astype(np.int16)
downsampled = resample_poly(audio_np, up=1, down=3)
return downsampled.astype(np.int16).tobytes()
|
こんな関数を ChatGPT に作ってもらった。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
def recv_decoded_audio(self, data: discord.sinks.core.RawData):
while data.ssrc not in self.voice_client.ws.ssrc_map:
time.sleep(0.01)
pcm_audio = data.decoded_data
pcm_audio = self.downsample_to_16k_mono(pcm_audio)
if self.recognizer.AcceptWaveform(pcm_audio):
result = self.recognizer.Result()
if result:
print(f"result: {result}")
else:
partial_result = self.recognizer.PartialResult()
if partial_result:
print(f"Partial result: {partial_result}")
|
こんな感じに Discord から拾った音声を変換して vosk にいれる。(recognizer は vosk のドキュメントなどを見ながら書いてください)
ここまで書くと Discord で喋ると文字起こしされるが、small モデルだからだかすごく精度が悪い!
これじゃ実用的とは言えないな…というお話でした。
(GPU で Whisper を使えばできると思います。)