0

Easy Pipeline,一种轻量级的Python Pipeline库

嗯,很久没有写博客了,最近的工作都是偏开发性质的,以至于没有时间对自己感兴趣的领域进行探索,感觉个人的成长停滞了一些。如何在枯燥的工作中,提取出有助于自己成长的养分,对于每个人来说都是不小的考验。

这次,带来的是之前编写的一下挺简单的库,用来简化流水线作业的小框架。

Github: https://github.com/miaoerduo/easy-pipeline 欢迎Star和提交MR。

起因是这样的,组内有一个需求,需要挖掘视频中的检测难样本,这样可以极大地减少标注的量,从而降低成本。难样本挖掘的策略,简单来说就是如果视频的前几帧和后几帧都能检测到目标,而就只有当前帧没有检测到,就说明当前帧很可能存在漏检(没有检测本到该检测到的目标);反之,如果前后都没有检测到目标,而当前帧检测到了,那就很可能是误检(检测到不是目标的东西)。

初步的方案是这样的,我们先把视频抽帧,直接用FFMpeg就可以方便的完成。然后调用现在的检测器,进行逐帧的检测,把检测结果存下来。最后写个脚本,分析检测的结果,然后输出可能有问题的帧,然后这些帧就会进行送标(发给标注员进行标注)。最终我们就只需要标注一些比较hard的样本就行了。

但是这样会带来很多的问题,最显著的两个:1. 需要保存大量的中间结果(图片帧);2. 必须依次完成每一步之后,才能得到最终的结果。

这时候,相比大家都知道了该如何去解决。对的,我们应该用流水线作业的方式去进行。

首先我们可以将每部分任务并行的去处理。抽帧之后的结果送入队列;之后检测模块从队列取帧,检测之后将结果送入下一个队列;最后一个队列得到检测结果,再做最终的分析。相比于之前的方式,这样可以尽量的减少中间的结果。

实现该方案,只需要使用最简单的生产者消费者队列即可以完成。所以说,相信大家都十分了解了。对于上面的逻辑,我们需要的队列的数目和我们的模块数是正相关的。如果单纯的进行实现的话,实在的太麻烦了,给队列命名都要我们绞尽脑汁了。所以,为了更优雅的编写代码,这里就推出本文标题中的Easy Pipeline框架。

首先,我们举个最简单的例子来说明该框架的工作模式。输入一个数字的序列,按要求对他们进行加减乘除的操作(这里的每个操作,其实可以等价于前面的抽帧或是检测的更复杂的操作 ),并且支持每个操作的进程数。

下面,我们来简单的说明一下工作逻辑。

  1. 首先,我们需要定义自己的任务Task。只需要继承Task这个类即可,内部可以存放自己喜欢的任何数据。这里只是为了计算,所以就只存放了一个数字。
  2. 定义我们的初始化函数和工作函数。初始化函数的作用是给每个进程初始化一些资源,如果不需要也可以不要。这里的初始化函数就是返回了一个值,表示操作数。工作函数是最重要的函数,他会处理接收到的Task,处理并返回新的Task(新的Task可以理解为处理的结果)。工作函数有两个输入,一个是资源,即初始化函数的返回值,另一个就是Task本身。
  3. 构建Pipeline。每个工作模块都只需要用PipelineItem这个对象进行封装即可。器参数分别是:工作函数、初始化函数、进程数、结果队列的长度(-1表示不限长度)。结果队列的长度,通常设置为较大的值即可。因为不能的模块的处理速度可能不同,因此很容易出现结果堆积的现象,如果不支持队列长度,会导致内存的大量的占用。最后将PipelineItem的数组和输入的对垒传给SimplePipeline对象即可构建完我们的整套Pipeline程序了!
  4. 启动Pipeline程序,并输入数据。
  5. 得到结果!完事了,优秀。

上面这是一个最简单的例子,可以比较直观的感受到这个框架的便捷之处。完全屏蔽掉对队列,并发等的操作。

在我推荐给同事之后,确实一定程度地减小他的工作量,但同时,他也向我反馈了一些问题:这个框架在某些地方有些比较灵活的设计,应该给出足够多的实例,才能方便实用。关于该框架的设计思路和实例,将会在下一篇博客中进行详细介绍。

最后,欢迎大家Star和提交MR。愿与你们一同进步。

打赏作者
miao

miao

发表评论

电子邮件地址不会被公开。