discord botで定期的に処理をさせるためのLoop

前提

この記事は，discord botでminecraft serverを動かしたいアドベントカレンダーの11日目です．

• WSL, Ubuntu 22.04.1
• Python 3.10.6
• discord.py

という環境で行っています．

さて，本題です．
minecraft serverをdiscord botで操作する際，遊ぶために起動コマンドを打つということは忘れることがないと思いますが，停止のコマンドを打つのは忘れる可能性は十分にあると思います．家を出た後にカギ閉めたっけ？ってなるのと一緒です．オートロックであればそんなことを気にすることはないでしょう．つまり，停止処理は自動でやってくれた方が便利です．ましてやマルチプレイするためにminecraft serverを起動しているのですから，最初に起動した人がログアウトしても他の人が遊んでいる場合があります．他の人は意識しない限り停止コマンドなんて入力しません(偏見)．

というわけで，minecraft serverにおいて停止処理を自動化させます．
具体的には，定期的にログイン人数を確認し誰もログインしていないなら終了させます．
ただし，この記事では疑似的にこの処理を行います．

• serverの起動
  • startと送信
• serverの停止
  • stopと送信
• ログイン人数の確認
  • 0～2のランダムな整数を返す
• 定期的な確認
  • 5秒間隔

このように操作を代替します．サーバーに入ったり出たりするの面倒だった
また，以下のサンプルコードから発展させていきます．

サンプルコード

test-discordbot.py

import os
import random
from discord import Intents, Client, Interaction
from discord.app_commands import CommandTree
from dotenv import load_dotenv
load_dotenv()


class MCServer():
    def start(self):
        print("starting")

    def stop(self):
        print("stopping")


class MyClient(Client):
    def __init__(self, intents: Intents) -> None:
        super().__init__(intents=intents)
        self.tree = CommandTree(self)
        self.server = MCServer()

    async def setup_hook(self) -> None:
        await self.tree.sync()

    async def on_ready(self):
        print(f"login: {self.user.name} [{self.user.id}]")

    def getLoginNumber(self):
        return random.randint(0, 2)


intents = Intents.default()
client = MyClient(intents=intents)


@client.tree.command()
async def start(interaction: Interaction):
    client.server.start()
    await interaction.response.send_message("started")


@client.tree.command()
async def stop(interaction: Interaction):
    client.server.stop()
    await interaction.response.send_message("stop")


client.run(os.getenv("TOKEN"))

loop

discord.pyを使ったdiscord botにおいて，ある処理を定期的に実行させたい場合，discord.ext.tasksのloop()を利用します．
APIリファレンスに具体例がいろいろ載っているので参考にしています．ここで，サンプルにはCogを利用しているものがあります．ですがこのCogはテキストベースのコマンドで利用するものであり，スラッシュコマンドを利用している場合は使わなくて大丈夫です．

定期実行にはtasks.loop()デコレータを利用します．今回利用する引数は，定期実行させたい時間(時/分/秒/正確な時刻)，ループを実行する回数(指定しないと無限ループ)，です．
そしてデコレータによって，下に書いた関数をLoopクラスとします．定期的に実行するための起動にはstart()メソッドを使います．また，定期実行の終了にはstop()，もしくはcancel()メソッドを利用します．

とりあえず動かしてみましょう．

test-discordbot.py

import os
import random
from discord import Intents, Client, Interaction
from discord.app_commands import CommandTree
+ from discord.ext import tasks
from dotenv import load_dotenv
load_dotenv()


class MCServer():
    def start(self):
        print("starting")

    def stop(self):
        print("stopping")


class MyClient(Client):
    def __init__(self, intents: Intents) -> None:
        super().__init__(intents=intents)
        self.tree = CommandTree(self)
        self.server = MCServer()

    async def setup_hook(self) -> None:
        await self.tree.sync()

    async def on_ready(self):
        print(f"login: {self.user.name} [{self.user.id}]")
+       self.channel = self.get_channel(1047700283712622643)

    def getLoginNumber(self):
        return random.randint(0, 2)

+   @tasks.loop(seconds=5)
+   async def checkCanStop(self):
+       number = self.getLoginNumber()
+       await self.channel.send(number)

intents = Intents.default()
client = MyClient(intents=intents)


@client.tree.command()
async def start(interaction: Interaction):
    client.server.start()
    await interaction.response.send_message("started")
+   client.checkCanStop.start()


@client.tree.command()
async def stop(interaction: Interaction):
    client.server.stop()
    await interaction.response.send_message("stop")
+   client.checkCanStop.stop()


client.run(os.getenv("TOKEN"))

このように変更しました．ちなみにメッセージを送信するためにチャンネルを取得していますが，チャンネル，というかギルドが読み込まれるタイミング的にon_ready()のところでチャンネルを取得しています．
ともかく，起動します．

こんな感じでループさせることができます．ただし，ループの終了にstop()を利用すると現在のループが終了してから停止するため，今回の処理では余計です．現在のループを中断して停止するにはcancel()を利用します．
また，ループを開始しているのにまたループを開始しようとするとエラーが出ます．discord.app_commands.errors.CommandInvokeError: Command 'start' raised an exception: RuntimeError: Task is already launched and is not completed.
というエラーですね．これの回避にはis_runnning()というメソッドを利用します．稼働しているかどうかを確認できるメソッドです．
加えて，関数内のこの処理のときに停止してほしいな～，というときには，関数内でcancel()等が呼び出せるのでこれで停止させることができます．自分自身はデコレータによってクラスになっているのでこの書き方ができる，のだと思います．

以上を修正点として改造したコードが以下の通りです．

test-discordbot.py

import os
import random
from discord import Intents, Client, Interaction
from discord.app_commands import CommandTree
from discord.ext import tasks
from dotenv import load_dotenv
load_dotenv()


class MCServer():
    def start(self):
        print("starting")

    def stop(self):
        print("stopping")


class MyClient(Client):
    def __init__(self, intents: Intents) -> None:
        super().__init__(intents=intents)
        self.tree = CommandTree(self)
        self.server = MCServer()

    async def setup_hook(self) -> None:
        await self.tree.sync()

    async def on_ready(self):
        print(f"login: {self.user.name} [{self.user.id}]")
        self.channel = self.get_channel(1047700283712622643)

    def getLoginNumber(self):
        return random.randint(0, 2)

    @tasks.loop(seconds=5)
    async def checkCanStop(self):
        number = self.getLoginNumber()
        await self.channel.send(number)
+
+       if number:
+           self.checkCanStop.cancel()
+           await self.channel.send("loop stoped")


intents = Intents.default()
client = MyClient(intents=intents)


@client.tree.command()
async def start(interaction: Interaction):
    client.server.start()
    await interaction.response.send_message("started")
-   client.checkCanStop.start()
+   if not client.checkCanStop.is_running():
+       client.checkCanStop.start()


@client.tree.command()
async def stop(interaction: Interaction):
    client.server.stop()
    await interaction.response.send_message("stop")
- client.checkCanStop.stop()
+

client.run(os.getenv("TOKEN"))

実行結果のgifはだいたい一緒なので省略しますが，0が出ると継続してそれ以外だと終了するループができました．

さて次にserverの自動停止を組み込みます．
仕様としては，定期的にログイン人数を確認して，0人であれば一定時間後にserverを停止する，という感じにします．0人で即server停止だと，席を外す際にログアウトするタイプの人が悲しい思いをする可能性があります．なので一定時間後に停止します．一定時間後に何かの処理を実行する，ということでここでもloopを利用します．特定の回数だけループさせることが可能なので，これを使います．

ここで，loopの仕様として，start()すると関数内の処理が一度実行されます．そのご一定時間後に再実行されます．つまり，

A「0人になったぞB！5秒後にserver停止してくれ！」
B「了解！ところでとりあえず実行したら停止したぞ！」
A「」

となります．違うそうじゃない．
そのため，ループを2度だけ実行し，1度目は無視させます．これにはcurrent_loopプロパティを利用します．このプロパティは現在の実行回数を持っています．つまり，実行回数が0回のときは無視してもらいます．

ちなみに，Loopにはafter_loopというデコレータがあり，これは一度のループ終了後に行われる処理を追加するためのものです．before_loopというデコレータもあります．今回の場合だとafter_loopに記述もできますが，なんかうまくいかなかったのでやめました．使わなくてもできたので諦めました．リファレンス読んでください．

以上を実装したものが以下のコードです，

test-discordbot.py

import os
import random
from discord import Intents, Client, Interaction
from discord.app_commands import CommandTree
from discord.ext import tasks
from dotenv import load_dotenv
load_dotenv()


class MCServer():
    def start(self):
        print("starting")

    def stop(self):
        print("stopping")


class MyClient(Client):
    def __init__(self, intents: Intents) -> None:
        super().__init__(intents=intents)
        self.tree = CommandTree(self)
        self.server = MCServer()

    async def setup_hook(self) -> None:
        await self.tree.sync()

    async def on_ready(self):
        print(f"login: {self.user.name} [{self.user.id}]")
        self.channel = self.get_channel(1047700283712622643)

    def getLoginNumber(self):
        return random.randint(0, 2)

    @tasks.loop(seconds=5)
    async def checkCanStop(self):
        number = self.getLoginNumber()
        await self.channel.send(number)

-       if number:
-           self.checkCanStop.cancel()
-           await self.channel.send("loop stoped")
+       if number:
+           return

+       if not self.periodicStop.is_running():
+           await self.channel.send(f"login {number}. next 5 seconds, will stop server")
+           self.periodicStop.start()
+           self.checkCanStop.cancel()

+   @tasks.loop(seconds=5, count=2)
+   async def periodicStop(self):
+       if self.periodicStop.current_loop == 0:
+           return
+
+       if self.getLoginNumber():
+           await self.channel.send("find someone login. abort server stop")
+           self.checkCanStop.start()
+           return
+
+       await self.channel.send("stopping server (periodic)")
+       client.server.stop()


intents = Intents.default()
client = MyClient(intents=intents)


@client.tree.command()
async def start(interaction: Interaction):
    client.server.start()
    await interaction.response.send_message("started")
    if not client.checkCanStop.is_running():
        client.checkCanStop.start()


@client.tree.command()
async def stop(interaction: Interaction):
    client.server.stop()
    await interaction.response.send_message("stop")
+   if client.checkCanStop.is_running():
+       client.checkCanStop.cancel()



client.run(os.getenv("TOKEN"))

英語は苦手なのでそこは見逃してください．
それは置いといて，ログイン人数が0でないとき早期リターンで処理を終了させています．また，誰もログインしていないとき，自身のログイン人数確認の定期処理であるcheckCanStopは中止し，一定時間後の停止処理であるperiodicStopを開始しています．
また，periodicStopでは最初のループでは何もせず，2度目のループで改めてログイン人数を確認し，誰かログインしていたらserverを停止するようにしています．ちなみにデモではここの処理忘れてました(1敗目)．
そして，stopコマンドでの終了をする場合，checkCanStopが稼働状態であるのでこれも停止させます．ちなみにデモではここの処理忘れてました(2敗目)．

実行してみましょう．

3分の1にそこそこ敗北したので5倍速にしていますが，期待したような動作になっていると思います．
ちなみに，たまにちゃんと5秒間隔でメッセージが送られないときがありました．これは単純にメッセージを送る処理をしてからのラグのせいだったり，私のネット回線が遅かったり，5秒という短い間隔でやっているからなので，もっと長い間隔であれば問題ありません．

以上となります．discord botで定期的に処理をさせるためのLoopを説明しました．
お疲れ様でした．