使用 Trio 在屏幕上进行异步图像搜索
Asynchronous image search on screen with Trio
我正在尝试调整 this 模块以支持在给定时间在同一屏幕截图中搜索大量图像时的异步执行。我是异步编码的新手,经过大量研究后,我选择了 Trio 来完成它(因为它很棒而且简单)。
重点是:
- 该函数接收图像路径列表
- 在每次迭代时,它都会获取屏幕截图并尝试在数组中查找图像(如果我们不在数组中的每次尝试都获取新的屏幕截图,那么性能会更好)
- 如果找到一个,returns 图像的路径及其坐标
- 重新做一遍,因为一些图像现在可能会出现在屏幕上
我打算在另一个支持 Trio 异步的项目中使用它,这就是我尝试转换它的原因。
这是我的尝试:
def image_search(image, precision=0.8, pil=None):
if pil is None:
pil = pyautogui.screenshot()
if is_retina:
pil.thumbnail((round(pil.size[0] * 0.5), round(pil.size[1] * 0.5)))
return most_probable_location(pil, image, precision)
async def multiple_image_search_loop(images, interval=0.1, timeout=None, precision=0.8):
async def do_search():
while True:
pil = pyautogui.screenshot()
for image in images:
if pos := image_search(image, precision, pil):
return {
"position": pos,
"image": image
}
await trio.sleep(interval)
if timeout:
with trio.fail_after(timeout):
return await do_search()
else:
return await do_search()
虽然代码看起来是正确的,但我觉得我忽略了异步代码的要点。这一切都可以以同步方式完成,我觉得我没有做出任何改变。
如果性能上没有差异,那还不错,因为重点是让这个函数在异步上下文中有用,而不是在搜索图像的整个过程中阻塞,但如果我能优化的话,它肯定会更好。
也许如果不是 awaiting
在搜索所有图像后我调整 image_search()
调用 trio.sleep()
并在主函数上打开一个托儿所会更好吗? (对数组中的每个图像使用其中的 trio.start_soon()
方法)。这会减少我将要使用它的其他项目的阻塞,但是找到图像会花费更多时间,对吗?
Trio 不会像这样直接并行化 CPU 绑定代码。作为 "async framework" 意味着它只使用一个 CPU 线程,同时并行化 I/O 和网络操作。如果您插入一些对 await trio.sleep(0)
的调用,那么这将使 Trio 将图像搜索与其他任务交织在一起,但不会使图像搜索更快。
不过,您可能想要做的是使用单独的线程。我猜你的代码可能大部分时间都花在 opencv 上,而 opencv 可能会放弃 GIL?所以使用线程可能会让你 运行 你的代码一次跨多个 CPU ,同时也让其他异步任务 运行 同时。要像这样管理线程,Trio 允许您在线程中执行 await trio.to_thread.run_sync(some_sync_function, *args)
,其中 运行s some_sync_function(*args)
。如果您 运行 在托儿所同时进行多个这样的调用,那么您将使用多个线程。
线程有一个主要问题需要注意:一旦 trio.to_thread.run_sync
调用开始,它就无法取消,因此超时等在调用完成后才会生效.要解决此问题,您可能需要确保单个调用不会阻塞太久。
此外,关于样式的旁注:为 Trio 制作的函数通常不会采用 timeout=
这样的参数,因为如果用户想要添加超时,他们可以编写 with
就像传递参数一样容易地在你的函数周围阻塞自己。因此,这样您就不必在各处使用超时参数来混淆 API。
我正在尝试调整 this 模块以支持在给定时间在同一屏幕截图中搜索大量图像时的异步执行。我是异步编码的新手,经过大量研究后,我选择了 Trio 来完成它(因为它很棒而且简单)。
重点是:
- 该函数接收图像路径列表
- 在每次迭代时,它都会获取屏幕截图并尝试在数组中查找图像(如果我们不在数组中的每次尝试都获取新的屏幕截图,那么性能会更好)
- 如果找到一个,returns 图像的路径及其坐标
- 重新做一遍,因为一些图像现在可能会出现在屏幕上
我打算在另一个支持 Trio 异步的项目中使用它,这就是我尝试转换它的原因。
这是我的尝试:
def image_search(image, precision=0.8, pil=None):
if pil is None:
pil = pyautogui.screenshot()
if is_retina:
pil.thumbnail((round(pil.size[0] * 0.5), round(pil.size[1] * 0.5)))
return most_probable_location(pil, image, precision)
async def multiple_image_search_loop(images, interval=0.1, timeout=None, precision=0.8):
async def do_search():
while True:
pil = pyautogui.screenshot()
for image in images:
if pos := image_search(image, precision, pil):
return {
"position": pos,
"image": image
}
await trio.sleep(interval)
if timeout:
with trio.fail_after(timeout):
return await do_search()
else:
return await do_search()
虽然代码看起来是正确的,但我觉得我忽略了异步代码的要点。这一切都可以以同步方式完成,我觉得我没有做出任何改变。
如果性能上没有差异,那还不错,因为重点是让这个函数在异步上下文中有用,而不是在搜索图像的整个过程中阻塞,但如果我能优化的话,它肯定会更好。
也许如果不是 awaiting
在搜索所有图像后我调整 image_search()
调用 trio.sleep()
并在主函数上打开一个托儿所会更好吗? (对数组中的每个图像使用其中的 trio.start_soon()
方法)。这会减少我将要使用它的其他项目的阻塞,但是找到图像会花费更多时间,对吗?
Trio 不会像这样直接并行化 CPU 绑定代码。作为 "async framework" 意味着它只使用一个 CPU 线程,同时并行化 I/O 和网络操作。如果您插入一些对 await trio.sleep(0)
的调用,那么这将使 Trio 将图像搜索与其他任务交织在一起,但不会使图像搜索更快。
不过,您可能想要做的是使用单独的线程。我猜你的代码可能大部分时间都花在 opencv 上,而 opencv 可能会放弃 GIL?所以使用线程可能会让你 运行 你的代码一次跨多个 CPU ,同时也让其他异步任务 运行 同时。要像这样管理线程,Trio 允许您在线程中执行 await trio.to_thread.run_sync(some_sync_function, *args)
,其中 运行s some_sync_function(*args)
。如果您 运行 在托儿所同时进行多个这样的调用,那么您将使用多个线程。
线程有一个主要问题需要注意:一旦 trio.to_thread.run_sync
调用开始,它就无法取消,因此超时等在调用完成后才会生效.要解决此问题,您可能需要确保单个调用不会阻塞太久。
此外,关于样式的旁注:为 Trio 制作的函数通常不会采用 timeout=
这样的参数,因为如果用户想要添加超时,他们可以编写 with
就像传递参数一样容易地在你的函数周围阻塞自己。因此,这样您就不必在各处使用超时参数来混淆 API。