如何停止或延迟链接到用户事件的 Pygame 计时器？

Question

我是 python 的新手，但我正在尝试自学以备大学课程之用。因此，我必须编写一个 pygame window 来表示无人机飞行区（无人机应该像搜索直升机一样在给定区域自动飞行，并且在 window 上会发生什么情况的直观表示应该显示...)。

How an older version of the pygame window look's like when the virtual drone flies.

到目前为止一切顺利。

现在我必须 link 游戏中无人机的动作与真正的 Tello 无人机。我发现当 pygame window 应该一直运行时，对无人机的移动命令使用线程是可行的方法。

但现在我卡住了，因为我必须延迟 pygame 中的自动移动，以等待无人机飞行时间和响应。

我在游戏循环中将 pygame window 更新为 clock 设置为 tick(60)（因此我创建了一个时钟 clock = pygame.time.Clock() pygame 程序的初始化），但对于自动无人机运动，我创建了一个 Userevent。无人机的移动就像贪吃蛇游戏一样通过用户事件触发应该每秒触发一次。

因此我实现了一个计时器：

self.SCREEN_UPDATE = pygame.USEREVENT

self.screen_update = pygame.time.set_timer(self.SCREEN_UPDATE, 1000).

但现在我必须在 pygame 每次无人机应该移动时延迟自动移动。否则，pygame 程序运行对于现实世界的问题来说太快了。（用户事件在真正的无人机完成他的命令之前再次启动）

我也尝试为此实现中断功能，但它们不起作用。

   def break_rotation(self):
        print("start waiting")
        self.screen_update.time.wait(2000)
        #self.screen_update = pygame.time.set_timer(self.SCREEN_UPDATE, 2000)
        #pygame.time.wait(2000)
        print("end waiting")``

   def break_move(self):
        print("start waiting")
        self.screen_update.time.wait(3000)
        #self.screen_update = pygame.time.set_timer(self.SCREEN_UPDATE, 3000)
        #pygame.time.wait(3000)
        print("end waiting")

我也试过注释行。 pygame.time.wait()让整个pygame等待。但是您应该仍然可以使用键盘进行紧急着陆和其他操作。

所以我只需要延迟下一个 Userevent，其中下一个块/部分的孔飞行路径将针对自动飞行路径进行测试，程序决定向前飞行还是应该转向。

有没有人知道如何实现计时器的延迟？可悲的是我也找不到一些类似的问题。

编辑 1：

@CmdCoder858

我认为这段代码可能会使更新延迟几秒钟。将 and 运算符与传递的变量结合使用时。 if event.type == SCREEN_UPDATE and x == 1:

我也尝试过使用多个事件，但可能没有用。 if event.type == SCREEN_UPDATE and event.type == drone_task_finished: 有趣的是，它以这种方式忽略了孔和运算符。

import pygame
import threading
import _thread
import time



def task():
    global x
    # placeholder for drone task
    print('Test thread start')
    time.sleep(5)  # simulate delay in the task
    x = x+1
    print('Thread ends')
    pygame.event.post(pygame.event.Event(drone_task_finished))


if __name__ == '__main__':
    pygame.init()
    x = 0
    drone_task_finished = pygame.USEREVENT  # an integer representing the first user event
    drone_task = threading.Thread(target=task)  # create a thread with a task function
    drone_task.start()
    pygame.display.set_caption("Overview of the Search Area")
    window = pygame.display.set_mode((500, 300))

    SCREEN_UPDATE = pygame.USEREVENT
    screen_update = pygame.time.set_timer(SCREEN_UPDATE, 1000)

    print('Start event loop')


    while 1:
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                exit()

            if event.type == SCREEN_UPDATE and x == 1:
                print('Hura Timer paused for 5 sec')
                x = 0
                drone_task = threading.Thread(target=task)  # create a thread with a task function
                drone_task.start()

            if event.type == SCREEN_UPDATE:
                print('Update nach 1sec')

            if event.type == drone_task_finished:  # detect if a drone_task_finished event has been posted
                print('Drone task has finished')

但是有什么理由不使用import _thread？

有了它，我可以在新线程中多次调用该函数。因此我必须将任务函数更改为 def task(threadName):

现在我可以简单地使用 _thread.start_new_thread(task, ("",)) 每次任务需要稍后使用。

Answer 1

从听起来，我猜你想在保持主循环运行的同时等待另一个线程中的事件，在这种情况下我认为你应该尝试使用 pygame.event.post() 函数.这个函数显然是线程安全的，所以只要它是从 pygame.init() 之后启动的线程调用的，它就应该可以正常工作。您可以使用 pygame.event.custom_type() 函数创建自定义事件，这意味着您可以等待事件返回而无需暂停显示。如果您有任何问题，请在下方评论。

编辑：

为清楚起见添加代码示例：

import pygame
import threading
import time


def task():
    # placeholder for drone task
    time.sleep(3)  # simulate delay in the task
    pygame.event.post(pygame.event.Event(drone_task_finished))

if __name__ == '__main__':
    pygame.init()
    drone_task_finished = pygame.USEREVENT  # an integer representing the first user event
    drone_task = threading.Thread(target=task)  # create a thread with a task function
    drone_task.start()
    print('Start event loop')
    while 1:
        for event in pygame.event.get():
            if event.type == drone_task_finished:  # detect if a drone_task_finished event has been posted
                print('Drone task has finished')

基本上这里发生的事情是我们正在创建一个要完成的任务，然后启动任务并继续事件循环。一旦任务完成，它将使用 pygame.event.post 函数告诉主线程它的任务已经完成。然后主线程可以检测到这一点并进行相应处理。

编辑 2：要回答您的问题，首先不要担心为您自己的问题创建答案，只需编辑您的原始问题并在评论中提及您已经这样做就可以了。

至于使用变量 x 作为延迟另一个事件执行的方法，这可以工作，但我建议将 x 保留在线程 task 函数之外，而是在主线程时修改它检测到 drone_task_finished 事件。这要好得多，因为您不必处理跨线程变量，所以只需将其放在 if event.type == drone_task_finished: 块下即可。

至于使用多个自定义事件，您的代码 event.type == SCREEN_UPDATE and event.type == drone_task_finished 返回 True 的原因是该语句是正确的，这是由于您如何初始化每个事件类型。您基本上使用了这段代码：

drone_task_finished = pygame.USEREVENT
SCREEN_UPDATE = pygame.USEREVENT

然而实际应该这样做的方式是这样的：

drone_task_finished = pygame.USEREVENT
SCREEN_UPDATE = pygame.USEREVENT + 1

这是由于 pygame 处理事件类型的方式。每种类型都被赋予一个整数值，内置类型获取特定点之前的值。 pygame.USEREVENT 常量是一个整数，表示第一个没有为其他事件保留的数字。这意味着创建一个新变量来表示 pygame 中的事件应该是整数 pygame.USEREVENT，第二个变量的值是 pygame.USEREVENT + 1。所有这些都在更详细的解释中here.

最后，如 this answer 中所述，_thread 是一个实现模块，通常应避免使用 threading 模块，这是我在代码中使用的模块。使用线程模块，您可以使用以下代码创建和启动一个新线程：

drone_task = threading.Thread(target=task)
drone_task.start()

threading.Threadclass还有一个可选的arguments参数，这意味着很容易将你需要的所有参数传递给新线程。您可以在 the documentation.

中找到有关这一切如何运作的更多信息

如何停止或延迟链接到用户事件的 Pygame 计时器？

How could I stop or delay a Pygame timer, which is linked to an User event?

python

timer

pygame

clock

tello-drone