Videoflow

Videoflow 是一个用于视频流处理的 Python 框架。该库旨在简化并快速定义计算机视觉流处理管道。它使开发者能够仅用简洁的几行代码，构建具备独立深度学习和计算机视觉功能的应用程序及系统。Videoflow 包含现成的对象检测、目标跟踪、人体姿态估计等参考组件，并且易于通过自定义组件进行扩展。

项目的完整文档位于 docs.videoflow.dev。

安装框架

需求

在安装之前，请确保已安装 cv2。Python 2 不被支持。需要 Python 3.6 或更高版本。此外，在 Windows 上运行时也存在一些 已知问题。

安装

您可以直接使用 pip 进行安装：pip3 install videoflow

或者，您也可以通过以下步骤安装：

1. 克隆本仓库。
2. 在仓库目录内执行 pip3 install . --user。

使用 Docker

# 克隆仓库
docker build -t repo/videoflow:latest .
# 默认运行 examples/object_detector.py
docker run -u $(id -u):$(id -g) -v $(pwd):/usr/src/app repo/videoflow
# 或者将您的代码目录挂载到 /usr/src/app
docker run -u $(id -u):$(id -g) -v $(pwd):/usr/src/app repo/videoflow python /usr/src/app/yourown.py

贡献：

我们大致的发展 路线图。

贡献指南。

如果您有新的处理器、生产者或消费者可以创建，请查看 videoflow-contrib 项目。我们希望保持 Videoflow 简洁、干净且简单，尽量减少对第三方库的依赖。对于需要引入新库依赖的新组件，videoflow-contrib 更为合适。

Videoflow 应用示例：

以下是一个 Videoflow 应用示例，用于检测十字路口中的汽车。更多示例请参阅 examples 文件夹。该示例使用了由 tensorflow/models 发布的对象检测模型。

import videoflow
import videoflow.core.flow as flow
from videoflow.core.constants import BATCH
from videoflow.consumers import VideofileWriter
from videoflow.producers import VideofileReader
from videoflow_contrib.detector_tf import TensorflowObjectDetector
from videoflow.processors.vision.annotators import BoundingBoxAnnotator
from videoflow.utils.downloader import get_file

URL_VIDEO = "https://github.com/videoflow/videoflow/releases/download/examples/intersection.mp4"

class FrameIndexSplitter(videoflow.core.node.ProcessorNode):
    def __init__(self):
        super(FrameIndexSplitter, self).__init__()
    
    def process(self, data):
        index, frame = data
        return frame

input_file = get_file("intersection.mp4", URL_VIDEO)
output_file = "output.avi"
reader = VideofileReader(input_file)
frame = FrameIndexSplitter()(reader)
detector = TensorflowObjectDetector()(frame)
annotator = BoundingBoxAnnotator()(frame, detector)
writer = VideofileWriter(output_file, fps = 30)(annotator)
fl = flow.Flow([reader], [writer], flow_type = BATCH)
fl.run()
fl.join()

该应用的输出是一个带有标注的视频：

流式应用的结构

一个流式应用通常由三部分组成：

1. 在应用程序的第一部分，您定义一个由计算节点组成的有向无环图。节点分为三种类型：生产者、处理器和消费者。生产者节点生成数据（通常从流外部的源获取数据）。处理器接收输入数据并产生输出数据。消费者读取数据但不产生任何输出。当您需要将结果写入日志文件，或推送到外部源（REST API、S3 存储桶等）时，通常会使用消费者。

2. 要创建一个流对象，您需要传入生产者列表和消费者列表。一旦定义了流，就可以启动它。启动流意味着生产者开始向流中输入数据，而处理器和消费者则开始接收数据。同时，也会为生产者、处理器和消费者分配资源。简单来说，目前每个生产者、处理器和消费者都会在各自的进程空间中运行。

3. 流启动后，也可以停止。停止流的过程是自然发生的：生产者会停止生成数据。流中的其他节点将继续运行，直到数据流耗尽为止。随着每个节点停止生产、处理或消费数据，流中所占用的资源会逐步释放。

引用 Videoflow

如果您在研究中使用 Videoflow，请使用以下 BibTeX 条目。

@misc{deArmas2019videoflow,
  author =       {Jadiel de Armas},
  title =        {Videoflow},
  howpublished = {\url{https://github.com/videoflow/videoflow}},
  year =         {2019}
}

Videoflow 快速上手指南

Videoflow 是一个用于视频流处理的 Python 框架，旨在帮助开发者通过简洁的代码快速构建包含深度学习和计算机视觉能力的流水线应用。它内置了目标检测、物体跟踪、人体姿态估计等常用组件，并支持轻松扩展自定义模块。

环境准备

在开始之前，请确保满足以下系统要求：

• 操作系统：推荐 Linux 或 macOS（Windows 存在已知兼容性问题）
• Python 版本：Python 3.6 或更高版本（不支持 Python 2）
• 前置依赖：必须预先安装 opencv-python (即 cv2)

建议先执行以下命令检查并安装基础依赖：

pip3 install opencv-python

安装步骤

你可以通过 pip 直接安装稳定版，或者从源码安装以获取最新功能。

方式一：使用 pip 安装（推荐）

pip3 install videoflow

注：国内用户若下载缓慢，可尝试指定清华镜像源：

pip3 install videoflow -i https://pypi.tuna.tsinghua.edu.cn/simple

方式二：从源码安装

1. 克隆仓库：

   git clone https://github.com/videoflow/videoflow.git
   cd videoflow
   

2. 执行安装：

   pip3 install . --user
   

Docker 使用（可选）

如果你希望使用容器化环境，可以执行以下命令构建并运行：

# 构建镜像
docker build -t repo/videoflow:latest .

# 运行默认示例
docker run -u $(id -u):$(id -g) -v $(pwd):/usr/src/app repo/videoflow

# 运行自定义脚本
docker run -u $(id -u):$(id -g) -v $(pwd):/usr/src/app repo/videoflow python /usr/src/app/yourown.py

基本使用

下面是一个最简单的 Videoflow 应用示例：从视频文件读取帧，使用 TensorFlow 模型检测汽车，绘制边界框，并保存为新视频。

该示例会自动下载测试视频，并依赖 videoflow-contrib 包中的检测器组件。

import videoflow
import videoflow.core.flow as flow
from videoflow.core.constants import BATCH
from videoflow.consumers import VideofileWriter
from videoflow.producers import VideofileReader
from videoflow_contrib.detector_tf import TensorflowObjectDetector
from videoflow.processors.vision.annotators import BoundingBoxAnnotator
from videoflow.utils.downloader import get_file

URL_VIDEO = "https://github.com/videoflow/videoflow/releases/download/examples/intersection.mp4"

class FrameIndexSplitter(videoflow.core.node.ProcessorNode):
    def __init__(self):
        super(FrameIndexSplitter, self).__init__()
    
    def process(self, data):
        index, frame = data
        return frame

# 准备输入输出文件
input_file = get_file("intersection.mp4", URL_VIDEO)
output_file = "output.avi"

# 定义节点
reader = VideofileReader(input_file)
frame = FrameIndexSplitter()(reader)
detector = TensorflowObjectDetector()(frame)
annotator = BoundingBoxAnnotator()(frame, detector)
writer = VideofileWriter(output_file, fps = 30)(annotator)

# 构建并运行流程
fl = flow.Flow([reader], [writer], flow_type = BATCH)
fl.run()
fl.join()

代码逻辑简述：

1. Producer (VideofileReader)：读取视频数据。
2. Processor (FrameIndexSplitter, TensorflowObjectDetector, BoundingBoxAnnotator)：处理数据（去索引、检测物体、绘制框）。
3. Consumer (VideofileWriter)：将结果写入新视频文件。
4. Flow：连接所有节点并启动多进程处理。

运行完成后，当前目录下将生成带有检测框的 output.avi 文件。更多示例请参考项目 examples/ 文件夹。