Zenn
Open9

pypubsubを使ってみる

Python
Python 3
Yos_KYos_K

簡単な(ゲームとも言えないような)ゲームを作ってみる
ふたりのコンピュータが1~5のランダムな整数をお互いに生成して揃ったら終了

Yos_KYos_K

ざっとコードを書いてみるとこんなかんじ

import random

from enum import StrEnum
from pubsub import pub


class Topic(StrEnum):
    Start = "start"
    Determined = "determined"
    Check = "check"


class Game:
    def __init__(self):
        self.num = []

    def start(self):
        pub.sendMessage(Topic.Start)

    def add_num(self, num):
        self.num.append(num)
        if len(self.num) == 2:
            print(self.num)
            self.check()

    def check(self):
        if self.num[0] != self.num[1]:
            self.num = []
            self.start()


class Com:
    def determine_num(self):
        pub.sendMessage(Topic.Determined, num = random.randrange(1, 5))


def main():
    game = Game()
    com1 = Com()
    com2 = Com()

    pub.subscribe(game.add_num, Topic.Determined)
    pub.subscribe(game.check, Topic.Check)
    pub.subscribe(com1.determine_num, Topic.Start)
    pub.subscribe(com2.determine_num, Topic.Start)

    pub.sendMessage(Topic.Start)


if __name__ == "__main__":
    main()
Yos_KYos_K

実行するとこんなかんじ

[2, 3]
[3, 2]
[2, 1]
[1, 3]
[4, 4]

Yos_KYos_K

(練習なので設計の良し悪しは気にしていない)

Yos_KYos_K
    pub.subscribe(game.add_num, Topic.Determined)
    pub.subscribe(game.check, Topic.Check)
    pub.subscribe(com1.determine_num, Topic.Start)
    pub.subscribe(com2.determine_num, Topic.Start)

    pub.sendMessage(Topic.Start)

この部分がトピックごとのサブスクライバを登録したり、トピックのメッセージを送っているところ

pub.subscribe()の第一引数がリスナーとなる関数(やメソッドなど)、第二引数が購読したいトピックとなる。
リスナーとか購読とか言うと分かりにくいが、どのイベントが発生した時にどの関数を実行して欲しいかを登録しているだけ。

第二引数のトピックはstrでないといけないが、文字列をそのまま入れるのはミスも起きやすいのでStrEnumを使用している
こうすると、IDEの補完も効くし、使用されていないトピックに気づきやすいなどバグも防止できる

class Topic(StrEnum):
    Start = "start"
    Determined = "determined"
    Check = "check"
Yos_KYos_K

今回はリスナーにComクラスのメソッドを登録していて、StartしたらComが数字を生成し、Determinedトピックでメッセージを送信している

class Com:
    def determine_num(self):
        pub.sendMessage(Topic.Determined, num = random.randrange(1, 5))
Yos_KYos_K

GameクラスはDeterminedを購読していて、メッセージが届いたら自身のnumに値を格納し、2つそろったら数字が等しいかの判定を行う(等しくなかったら再度Startのメッセージを送る)

class Game:
    def __init__(self):
        self.num = []

    def start(self):
        pub.sendMessage(Topic.Start)

    def add_num(self, num):
        self.num.append(num)
        if len(self.num) == 2:
            print(self.num)
            self.check()

    def check(self):
        if self.num[0] != self.num[1]:
            self.num = []
            self.start()
Yos_KYos_K

ちなみにRustで似たようなことをしようとしたらこうなった
むずい・・・

use std::{cell::RefCell, collections::HashMap, hash::Hash};
use rand::prelude::*;


trait Observer {
    fn update(&self, t: &Topic) -> Option<Topic>;
}

#[derive(Debug, PartialEq, Clone)]
struct Com;

impl Com {
    fn new() -> Self {
        Self
    }
}

impl Observer for Com {
    fn update(&self, t: &Topic) -> Option<Topic> {
        match t.kind {
            TopicKind::Start => {
                let a = rand::thread_rng().gen_range(1..=5);
                println!("{a}");
                Some(Topic::new(TopicKind::Determined, Some(a)))
            },
            _ => None,
        }
    }
}

#[derive(Debug, PartialEq, Clone)]
struct Game{
    nums: RefCell<Vec<usize>>
}

impl Game {
    fn new() -> Self {
        Self {nums: RefCell::new(vec![])}
    }
}

impl Observer for Game {
    fn update(&self, t: &Topic) -> Option<Topic> {
        match t.kind {
            TopicKind::Determined => {
                self.nums.borrow_mut().push(t.value.unwrap());
                match self.nums.borrow().len().cmp(&2) {
                    std::cmp::Ordering::Less => {
                        println!("{:?}", &self.nums.borrow());
                        None
                    },
                    std::cmp::Ordering::Equal => {
                        println!("{:?}", &self.nums.borrow());
                        Some(Topic::new(TopicKind::Check, None))
                    },
                    std::cmp::Ordering::Greater => None,
                }
            },
            TopicKind::Check => {
                println!("{:?}", &self.nums.borrow());
                if self.nums.borrow()[0] == self.nums.borrow()[1] {
                    Some(Topic::new(TopicKind::End, None))
                } else {
                    *self.nums.borrow_mut() = vec![];
                    Some(Topic::new(TopicKind::Start, None))
                }
            }
            _ => None
        }
    }
}

struct Subject<'a> {
    observers: RefCell<HashMap<TopicKind, Vec<&'a dyn Observer>>>
}

impl<'a> Subject<'a> {
    fn new() -> Self {
        Self { observers: RefCell::new(HashMap::new()) }
    }
    fn register(&self, topic_kind: TopicKind, observer: &'a dyn Observer) {
        let mut borrow_map = self.observers.borrow_mut();
        match borrow_map.get_mut(&topic_kind) {
            Some(v) => v.push(observer),
            None => {borrow_map.insert(topic_kind, vec![observer]);},
        }
    }

    fn notify(&self, t: &Topic) {
        dbg!(t);
        match t.kind {
            TopicKind::End => {},
            _ => {
                let binding = self.observers.borrow();
                let target = binding.get(&t.kind).unwrap();
                target.iter().filter_map(|o| o.update(t))
                    .for_each(|p| self.notify(&p));
            }
        }
    }
}


#[derive(Debug, PartialEq, Eq, Clone, Hash)]
enum TopicKind {
    Start,
    Determined,
    Check,
    End,
}

#[derive(Debug)]
struct Topic {
    kind: TopicKind,
    value: Option<usize>
}

impl Topic {
    fn new(t: TopicKind, v: Option<usize>) -> Self {
        Topic { kind: t, value: v }
    }
}

fn main() {
    let game = Game::new();
    let com1 = Com::new();
    let com2 = Com::new();

    let sub = Subject::new();
    sub.register(TopicKind::Determined, &game);
    sub.register(TopicKind::Check, &game);
    sub.register(TopicKind::Start, &com1);
    sub.register(TopicKind::Start, &com2);

    sub.notify(&Topic { kind: TopicKind::Start, value:None });
}