はじめに
この記事は前回Pythonで始める録音機能付きDiscord Bot: (3) Databaseとの連携の続きです．
今回ではサーバー上にあらかじめ用意した音楽をコマンドで再生させるジュークボックス機能を追加します．discord.pyを用いれば音声の再生に関してはとても簡単な処理で実行することができます．ここでは単に再生するだけでなく曲のキューを実装して連続再生ができるような機能をつけることにします．
全7回を予定しており現在5記事まで執筆を終えています．
- Pythonで始める録音機能付きDiscord Bot: (1) 入門 discord.py
- Pythonで始める録音機能付きDiscord Bot: (2) 便利機能(Bot拡張,Cog,Embed)
- Pythonで始める録音機能付きDiscord Bot: (3) Databaseとの連携
- Pythonで始める録音機能付きDiscord Bot: (4) 音楽ファイルを再生する
- Pythonで始める録音機能付きDiscord Bot: (5) Discord APIを直接操作する
Discordの音声通信
discord.pyでは直接いじることのない部分なので意識することはあまりありませんが，DiscordにはHTTPリクエストで投稿を送信したりサーバーの情報を取得したりする際に使用するREST APIのほかに，双方向的に通信するためのWebSocket通信，そして音声の送受信を行うRTP通信の3つの情報の送受信方法があります．
discord.pyの場合，ボイスチャンネルのインスタンスをContextなどから入手し，そのインスタンスのconnectコルーチンを実行することによって音声チャンネルに接続するという非常に直感的な仕組みになっています．このconnectより返されるものがVoiceClientクラスのインスタンスです．
VoiceClientにおいて，上記のWebSocket通信やその他暗号化通信などといった部分を包み隠しています．ですので，サーバーごとのボイスチャンネルの情報を操作したいといった場合にはこのVoiceClientインスタンスに対して様々な操作を行えばよいのです．
複数のサーバーからのボイスチャンネルに対する処理を考えた場合，このインスタンスとサーバー(ID)を結びつける機構が必要となります．ここではdiscord.pyのリポジトリにある公式実装のように管理のために辞書配列を使用してVoiceClientを操作し，ローカル(Botを動かしているサーバー上)にある音楽の再生を試みます．
VoiceChannelを操作する
VoiceChannelに入室するまで
前述のようにVoiceChannelへの入退出は非常に簡単です．ここではVoiceというCogを作成し$joinで入室し，$leaveで退出することにします．実装の一例としては以下のようになります．
import discord
from discord.ext import commands
from typing import Dict
from dbot.core.bot import DBot
class Voice(commands.Cog):
def __init__(self, bot: DBot):
self.bot = bot
self.voice_clients: Dict[int, discord.VoiceClient] = {}
@commands.command()
async def join(self, ctx: commands.Context):
# VoiceChannel未参加
if not ctx.author.voice or not ctx.author.voice.channel:
return await ctx.send('先にボイスチャンネルに参加してください')
vc = await ctx.author.voice.channel.connect()
self.voice_clients[ctx.guild.id] = vc
@commands.command()
async def leave(self, ctx: commands.Context):
vc = self.voice_clients.get(ctx.guild.id)
if vc is None:
return await ctx.send('ボイスチャンネルにまだ未参加です')
await vc.disconnect()
del self.voice_clients[ctx.guild.id]
def setup(bot):
return bot.add_cog(Voice(bot))
discord.Memberのプロパティであるvoiceはメンバーがボイスチャンネルに参加している場合，VoiceStateクラスのインスタンスを返します．このVoiceStateはメンバーの消音等の状態のほかにchannelというプロパティを持っており，これを参照することでメンバーの現在入室しているボイスチャンネルのクラス(discord.VoiceChannel)のインスタンスを得ることができます．
VoiceChannelのconnectコルーチンを呼ぶことでVoiceClientが返され，Botがボイスチャンネルに参加します．そのVoiceClientをサーバーIDをキーとして辞書に格納します．$leaveでは，逆にサーバーIDからVoiceClientを検索し，disconnectコルーチンを呼び出しています．入退出処理はこれで完了です．
音声を再生させる
まずは再生するための音楽を用意し，./src/app/music/フォルダ以下に保存します．
名前をもとに検索するように実装するために適当な名前を付けておきます．ひとまず$play 曲名で再生$stopで停止を行います．
import discord
from glob import glob
import os
from discord.ext import commands
from typing import Dict
from dbot.core.bot import DBot
class Voice(commands.Cog):
# 略
@commands.command()
async def play(self, ctx: commands.Context, *, title: str = ''):
vc: discord.VoiceClient = self.voice_clients.get(ctx.guild.id)
if vc is None:
await ctx.invoke(self.join)
vc = self.voice_clients[ctx.guild.id]
music_pathes = glob('./music/**.mp3')
music_titles = [
os.path.basename(path).rstrip('.mp3')
for path in music_pathes
]
if not title in music_titles:
return await ctx.send('指定の曲はありません．')
idx = music_titles.index(title)
src = discord.FFmpegPCMAudio(music_pathes[idx])
vc.play(src)
await ctx.send(f'{title}を再生します')
@commands.command()
async def stop(self, ctx: commands.Context):
vc: discord.VoiceClient = self.voice_clients.get(ctx.guild.id)
if vc is None:
return await ctx.send('Botはまだボイスチャンネルに参加していません')
if not vc.is_playing:
return await ctx.send('既に停止しています')
await vc.stop()
await ctx.send('停止しました')
# 略
duscird.pyを用いて音楽を再生するには，VoiceClientのPlayコルーチンにAudioSourceを渡す必要があります．AudioSourceの作成にはffmpegが必要です．ffmpegの環境が整っている場合，ファイルまでのパスを与えるだけでよく音楽の再生は非常に簡単に行えます．
再生を停止する場合も同様，vc.stopコルーチンを呼び出します．
連続再生を行う
現在の実装では，$playを呼び出すと直ちに音楽が切り替わるようになっています．これを，$playされたらキューに音楽を加えて前の音楽の再生が終えたら次の音楽を再生する仕様に変更します(いわゆるプレイリスト)．
プレイリスト実装のためにはasyncio.Queueとasyncio.Eventというものを活用します．それぞれ，キューに要素が追加されるまで待機する実装，よその個所でフラグが立つまで待機する実装をするためによく使用されます．
import discord
from glob import glob
import os
import asyncio
from discord.ext import commands
from typing import Dict
from dbot.core.bot import DBot
class AudioQueue(asyncio.Queue):
def __init__(self):
super().__init__(100)
def __getitem__(self, idx):
return self._queue[idx]
def to_list(self):
return list(self._queue)
def reset(self):
self._queue.clear()
class AudioStatus:
def __init__(self, ctx: commands.Context, vc: discord.VoiceClient):
self.vc: discord.VoiceClient = vc
self.ctx: commands.Context = ctx
self.queue = AudioQueue()
self.playing = asyncio.Event()
asyncio.create_task(self.playing_task())
async def add_audio(self, title, path):
await self.queue.put([title, path])
def get_list(self):
return self.queue.to_list()
async def playing_task(self):
while True:
self.playing.clear()
try:
title, path = await asyncio.wait_for(self.queue.get(), timeout=180)
except asyncio.TimeoutError:
asyncio.create_task(self.leave())
src = discord.FFmpegPCMAudio(path)
self.vc.play(src, after=self.play_next)
await self.ctx.send(f'{title}を再生します...')
await self.playing.wait()
def play_next(self, err=None):
self.playing.set()
async def leave(self):
self.queue.reset()
if self.vc:
await self.vc.disconnect()
self.vc = None
@property
def is_playing(self):
return self.vc.is_playing()
def stop(self):
self.vc.stop()
class Voice(commands.Cog):
def __init__(self, bot: DBot):
self.bot = bot
self.audio_statuses: Dict[int, AudioStatus] = {}
@commands.command()
async def join(self, ctx: commands.Context):
# VoiceChannel未参加
if not ctx.author.voice or not ctx.author.voice.channel:
return await ctx.send('先にボイスチャンネルに参加してください')
vc = await ctx.author.voice.channel.connect()
self.audio_statuses[ctx.guild.id] = AudioStatus(ctx, vc)
@commands.command()
async def play(self, ctx: commands.Context, *, title: str = ''):
status = self.audio_statuses.get(ctx.guild.id)
if status is None:
await ctx.invoke(self.join)
status = self.audio_statuses[ctx.guild.id]
music_pathes = glob('./music/**.mp3')
music_titles = [
os.path.basename(path).rstrip('.mp3')
for path in music_pathes
]
if not title in music_titles:
return await ctx.send('指定の曲はありません．')
idx = music_titles.index(title)
await status.add_audio(title, music_pathes[idx])
await ctx.send(f'{title}を再生リストに追加しました')
@commands.command()
async def stop(self, ctx: commands.Context):
status = self.audio_statuses.get(ctx.guild.id)
if status is None:
return await ctx.send('Botはまだボイスチャンネルに参加していません')
if not status.is_playing:
return await ctx.send('既に停止しています')
await status.stop()
await ctx.send('停止しました')
@commands.command()
async def leave(self, ctx: commands.Context):
status = self.audio_statuses.get(ctx.guild.id)
if status is None:
return await ctx.send('ボイスチャンネルにまだ未参加です')
await status.leave()
del self.audio_statuses[ctx.guild.id]
@commands.command()
async def queue(self, ctx: commands.Context):
status = self.audio_statuses.get(ctx.guild.id)
if status is None:
return await ctx.send('先にボイスチャンネルに参加してください')
queue = status.get_list()
songs = ""
for i, (title, _) in enumerate(queue):
songs += f"{i+1}. {title}\n"
await ctx.send(songs)
def setup(bot):
return bot.add_cog(Voice(bot))
VoiceClientとサーバーの情報を一緒に登録するために新たにAudioStatusというクラスを作成し，そのインスタンスを保存します．
AudioStatusでは初期化時にasyncio.create_taskという関数を呼び出しています．名前の通り，音楽再生を行うタスクをコルーチンから作成しています．こうすることで，タスクの方で別の処理をしながら他のサーバーからのコマンドに応答するなどといった非同期的なアクセスを可能にしています．playingというプロパティはasyncio.Eventですがこれは特定の条件を満たした際にフラグを立ててフラグが立つまでは何もせずに待機したいといった際に使われます．playing_taskではclear,waitを，play_nextではsetを呼び出していますが，それぞれ「フラグをFalseにする」，「フラグがTrueになるまで待つ」，「フラグをTrueにする」という処理です．play_nextはvc.playのafter引数に渡していますが，これは一曲の再生が終わった際に行いたい処理を引数に与えています．曲の終わり時に次曲再生のフラグを立てることで再度playing_taskのループが開始されます．
このAudioStatusのqueueプロパティはasyncio.Queueを継承し，曲目リストの取得などを容易にしたものですがこのasyncio.Queueはget呼び出し時に返す値がないような際にはその場で新しい要素が追加されるまでそこで待機します．これにより，曲目入力の待機時間を設けることが可能となります．そして3分の間入力がない場合にはleaveコルーチンを呼び出して自動的にボイスチャンネルから退出します．
これにより，曲のキューを備えた音楽再生機能を追加することが可能となりました．
終わりに
これで音楽再生機能は実装出来ました！簡単に済んでよかったです！音声録画機能も同じくらい簡単だよね？？？
というわけにはいかないようなので，録音機能を実装するにあたっての準備としてDiscordのAPI機能を直接触ってどのように音声の送受信を扱っているかを詳しく見てdiscord.pyを使用せずに実装します．