pythonのスレッドと状態管理の工夫

ReactやVueのように変数をウォッチし、変数が変化した時に特定の関数を実行するようにします。
useStateやwatchみたいな感じです。

1.前後の状態を比較する方法①

自作関数watch_chat_historyで、変数chat_historyをウォッチしています。
chat_historyに変化が生じる（この場合はchatが追加される）と、show_consoleが発火するようにしています。

    """チャット表示"""
    message_list = [
        "はい喜んで",
        "あなた方のため",
        "はい謹んで",
        "あなた方のため",
        "差し出してた"
    ]

    chat_history = []

    def add_chat():
        for i,message in enumerate(message_list):
            person = 'human' if i%2 else 'ai'
            chat_history.append((person,message))
            time.sleep(0.5)

    def show_console():
        if not len(chat_history):return

        person, message = chat_history[-1]
        cprint(f'{person}:{message}','yellow' if person=='human' else 'green' )


    def watch_chat_history():
        prev_state = None
        while True:
            current_state = len(chat_history)
            if prev_state != current_state:
                show_console()
                prev_state = current_state


    os.system('clear')

    Thread(target=watch_chat_history, daemon=True).start()

    t1 = Thread(target=add_chat)

    t1.start()

    t1.join()

2. 前後の状態を比較する方法②

もうちょっとウォッチ対象を一般的に見れるようにwatchという関数を定義してやりました。
Vueのwatchと同じように使えるよう意識しました。

    message_list = [
        "はい喜んで",
        "あなた方のため",
        "はい謹んで",
        "あなた方のため",
        "差し出してた"
    ]

    chat_history = []

    def add_chat():
        for i,message in enumerate(message_list):
            person = 'human' if i%2 else 'ai'
            chat_history.append((person,message))
            time.sleep(0.5)

    def show_console():
        if not len(chat_history):return

        person, message = chat_history[-1]
        cprint(f'{person}:{message}','yellow' if person=='human' else 'green' )

    def watch(target:Any,functions:List[Callable]):
        prev_state = None
        while True:
            current_state = target.copy() if isinstance(target,(list, dict, set)) else target
            if prev_state != current_state:
                for func in functions:
                    func()
                prev_state = current_state

    os.system('clear')

    Thread(target=lambda:watch(chat_history,[show_console]), daemon=True).start()

    Thread(target=add_chat).start()

3. イベントハンドラーを設ける

上記の書き方では、watch関数の中でchat_historyの状態を比較していましたが、以下のようにイベントハンドラー（finish_talk）を設けて、finish_talkがTrueに変化したらshow_consoleを実行させる方法もあります。

スレッドが多くなり、スレッド間で共通した状態変化をウォッチしたい場合はこのほうが書きやすいです。

    """イベントハンドラーを使う"""
    message_list = [
        "はい喜んで",
        "あなた方のため",
        "はい謹んで",
        "あなた方のため",
        "差し出してた"
    ]

    chat_history = []

    finish_talk = Event()

    def add_chat():
        for i,message in enumerate(message_list):
            person = 'human' if i%2 else 'ai'
            chat_history.append((person,message))
            finish_talk.set() # 追加
            time.sleep(0.5)

    def show_console():
        if not len(chat_history):return

        person, message = chat_history[-1]
        cprint(f'{person}:{message}','yellow' if person=='human' else 'green' )


    def watch_chat_history():
        while True:
            finish_talk.wait() # 状態が変わるまで待機
            show_console()
            finish_talk.clear() # 実行後はFalseに戻してやる必要がある


    os.system('clear')

    Thread(target=watch_chat_history, daemon=True).start()

    t1 = Thread(target=add_chat)

    t1.start()

    t1.join()