python - 在两个线程之间创建一个钩子

Question

我目前在我的 discord.py 机器人和另一个线程之间遇到问题，我需要创建一个钩子来从一个类中调用一个函数，比如一个 webhook，但在两个线程之间，我的代码的一个简单版本：

import serial
import threading


class bot:  # this class is constantly running other functions, can't get halted by 'usb.readline()' or the connection to the server will be closed

    async def on_button_pressed(self, button_pressed):  # the goal is to make this function run every time the side_thread detects a button press
        if button_pressed == 0:
            await channel.connect()  # if button 0 is pressed, bot connects to a voice channel
        elif button_pressed == 1:
            await voice.disconnect()  # if button 1 is pressed, bot disconnects from voice channel


def side_thread():  # reads the information an microcontroller send to pc
    usb = serial.Serial('com4', baudrate=115200)
    while True:
        uncoded_data = usb.readline()  # program waits a response from usb here
        data_str = str(uncoded_data[0:len(uncoded_data)].decode('utf-8'))
        button_pressed = int(data_str[:1])  # button_pressed can be 0, 1, 2, 3 or 4 depending on what button was pressed
        print(f'button {button_pressed} was pressed')
        # the problem is here, idk how to create a hook to call on_button_pressed and pass the arguments since it is in another thread,
        # already tried putting asyncio.run(bot.on_button_pressed(button_pressed)) here but it doesn't work since it isn't the main thread running the function


t1 = threading.Thread(target=side_thread)
t1.start()

bot = bot()

我试图将 on_button_pressed 函数放在 side_thread 中，但机器人无法连接，可能是因为异步函数不是线程安全的。

我的目标是，每当我按下按钮时，它都会自动将 on_button_pressed 函数作为主线程运行，而无需停止它，也无需不断检查变量（也停止代码或使按钮极度延迟）

为了尝试更好地解释一下，模拟按钮按下，您可以用 side_thread 替换：

import random
import time

def side_thread():  # reads the information an microcontroller send to pc
    while True:
        button_pressed = random.randint(0, 4)
        time.sleep(5)  # simulates a random button press every 5 seconds
        print(f'button {button_pressed} was pressed')

而且我希望每次按下按钮时都会将 bot.on_button_pressed 作为主线程运行。要检查这一点，您可以在 async func 上打印线程的名称，它应该与init方法中的相同：

class bot:  # can't get halted by a while loop or a lock
    def __init__(self):
        print(threading.current_thread().name)  # prints 'MainThread'

    async def on_button_pressed(self, button_pressed):  # goal is to make this function run every time the side_thread detects a button press
        print(threading.current_thread().name)  # it should also print the same as the __init__, if it prints differently it isn't working

例如，如果我添加asyncio.run(bot.on_button_pressed(button_pressed))到 side_thread 的末尾，则 bot 类上的两个打印将不同，因此它将不起作用

标签： pythonpython-3.xmultithreadingdiscord.py