在现代物流和监控系统中，视频分析技术扮演着越来越重要的角色。它不仅能够提高安全性，还能优化物流流程和库存管理。然而，搭建一个强大的视频分析流程可能会相当耗时。需要一个训练有素的计算机视觉模型、推理代码以及可视化工具来标注每一帧视频。这仅仅是为了在视频上运行一个模型的开销！在这篇文章中，将展示如何使用Roboflow Universe来实现可扩展的视频分析，只需几行Python代码即可访问可定制的预训练模型。
将使用Roboflow Universe中的一个物流目标检测模型，并使用Supervision库在检测到的对象上添加自定义注释，如边框、标签、跟踪等。
将涵盖以下步骤：
让看看物流中的一个用例——监控仓库和配送中心的库存。准确跟踪库存和资产至关重要，但通常是一个手动过程。可以使用摄像头记录仓库内的视频流。然后，目标检测模型可以识别和定位重要的资产，如托盘、箱子、手推车和叉车。最后，可以添加自定义覆盖层，如标签、边框轮廓和跟踪，以创建一个库存监控系统。
import os
from roboflow import Roboflow
import numpy as np
import supervision as sv
import cv2
Roboflow提供了易于使用的API，用于处理模型，并在后台自动运行推理，而Supervision包含了让轻松自定义输出视频的注释器。
PROJECT_NAME = "logistics-sz9jr"
VIDEO_FILE = "example.mp4"
ANNOTATED_VIDEO = "output.mp4"
rf = Roboflow(api_key="ROBOFLOW_API_KEY")
project = rf.workspace().project(PROJECT_NAME)
model = project.version(2).model
job_id, signed_url, expire_time = model.predict_video(
VIDEO_FILE,
fps=5,
prediction_type="batch-video",
)
results = model.poll_until_video_results(job_id)
为了提高注释的质量，可以利用'supervision'包中的自定义注释器。在这个例子中，使用'BoxMaskAnnotator'和'LabelAnnotator'来改进注释：
box_mask_annotator = sv.BoxMaskAnnotator()
label_annotator = sv.LabelAnnotator()
tracker = sv.ByteTrack()
为了在视频中标注帧，将定义一个名为`annotate_frame`的函数，该函数利用从视频推理中获得的结果。然后对于每一帧，：
cap = cv2.VideoCapture(VIDEO_FILE)
frame_rate = cap.get(cv2.CAP_PROP_FPS)
cap.release()
def annotate_frame(frame: np.ndarray, frame_number: int) -> np.ndarray:
try:
time_offset = frame_number / frame_rate
closest_time_offset = min(results['time_offset'], key=lambda t: abs(t - time_offset))
index = results['time_offset'].index(closest_time_offset)
detection_data = results[PROJECT_NAME][index]
roboflow_format = {
"predictions": detection_data['predictions'],
"image": {"width": frame.shape[1], "height": frame.shape[0]}
}
detections = sv.Detections.from_inference(roboflow_format)
detections = tracker.update_with_detections(detections)
labels = [pred['class'] for pred in detection_data['predictions']]
except (IndexError, KeyError, ValueError) as e:
print(f"Exception in processing frame {frame_number}: {e}")
detections = sv.Detections(xyxy=np.empty((0, 4)),
confidence=np.empty(0),
class_id=np.empty(0, dtype=int))
labels = []
annotated_frame = box_mask_annotator.annotate(frame.copy(), detections=detections)
annotated_frame = label_annotator.annotate(annotated_frame, detections=detections, labels=labels)
return annotated_frame
sv.process_video(
source_path=VIDEO_FILE,
target_path=ANNOTATED_VIDEO,
callback=annotate_frame
)