pe氏mmbotEvent監視殴り書きメモ

mmbotを作成してみようと思ってしらべてるとヤメティさんのブログ、magitoさんのnote、peさんのpybotters版プログラムが出てきた

気になって触ってみたはいいものの、プログラム力がほぼないため、クラスの継承とかその辺がよくわからなかった

見返し用メモ、なお日本語もろくにできないから参考にはならないと思います。

以下サイト

mmbot神教材まとめ

class EventWatcher:
    def __init__(self, store: DataStore, trigger_fn: Callable[[Item], bool] = None):
        self._store = store
        self._trigger_fn = trigger_fn
        self._task = asyncio.create_task(self._watch())

初期化、引数としてDataStore、trigger_fnを入れてる
ここで、Callableとかいう初めて見るものが出てきた

調べてみる

callableはPythonの組み込み関数であり、指定した引数が呼び出し可能かどうか(関数のように扱えるか)を判定します。

https://techacademy.jp/magazine/31457

よくわからんけど、引数がいいものかわるいものかを判断してくれるらしい

また、今回はCallable[[Item], bool]となってるからこの書き方について調べてみる

書き方としてはCallable[[第一引数の型, 第二引数の型], 返却値の型]といったように書きます。引数が増減した場合にはコンマ区切りで増やしたり減らしたりします。

https://qiita.com/simonritchie/items/2665e1f4d6bad63d652f

今回の書き方は型アノテーションっていって、引数の良否判定、返り値のタイプまで判定してくれるらしい。便利。

self._task = asyncio.create_task(self._watch())

次に_watchがタスクとしていれられてるからこの中身をみていく
なお、非同期に関しては理解しきれてないので、さらっとすすめる

async def _watch(self):
        """`_is_target_event(msg.data)`がTrueを返すまでDataStoreをwatchし続ける。"""
        with self._store.watch() as stream:
            async for msg in stream:
                if self._is_trigger(msg.data):
                    return msg.data

とりあえず、コメントにある通り、watchし続けてるらしい。何を？
このwatchに関してはpybottersで新たに実装されたらしいのでwatchに関して調べる

このメソッドを利用することで、より手軽に WebSocket からのデータ監視を行う事ができます。

https://github.com/MtkN1/pybotters/releases

とりあえず、pybottersのwaitと似たような機能らしいけど、REST APIとwebsocketの兼ね合いでいままでうまくいかなかったプログラムがうまくいくようになったらしい。
自分には難しいのでとりあえず欲しいデータが手に入るまでループしてくれるものって覚えておこう

それで、このmsgにデータが入ったときにif文でまた違う関数が実行されている
self._is_triggerに関して確認

def _is_trigger(self, d: Item):
        """socketメッセージを受け取って、イベント発火の有無を判定する。
        子クラスでこの関数をオーバーライドしてもいいし、`trigger_fn`として与えてもいい。

        :param Item d: socketメッセージ。
        :return:
        """
        if self._trigger_fn is None:
            raise NotImplementedError
        return self._trigger_fn(d)

引数として先ほど帰ってきたmsgを渡してる
そのメッセージからイベント発火を判断してるらしい
オーバーライドに関して調べる

オーバーライドとは「親クラスと同じ名前のメソッドを子クラスで新たに定義すること」を言います。

https://techacademy.jp/magazine/28157

つまり、子クラスで同じ関数名を使用して上書きする行為
今回の場合はとりあえず、_trigger_fnが何も入ってなかったら例外出して、
入ってたら、そのなかにmsgをぶち込んでbool値判定してるっぽい
そのbool値がTrueであれば引数としてmsgのなかのdataを返り値で返してる

async def wait(self):
        await self._task

    def done(self):
        return self._task.done()

    def result(self):
        return self._task.result()

その他プログラムは_taskのなかの様々なデータを参照しやすいようにしてる？ちょっとよくわからない

イベント監視系のクラスがまだあるから、また明日確認します。