用 Python 创建并运行视频流识别 Stream

从一次性 Job 到持续运行的 Stream

上一教程的 Job 处理一张图片后就结束；Stream 面向 RTSP 摄像头、GB28181 或本地视频文件等持续输入，被领取后长时间运行，靠心跳续租，并以事件形式持续回传识别结果。本教程沿用同一个平台，使用三进程示例 Runner 跑通“创建流 → 领取 → 心跳 → 事件回传 → 停止”的完整链路。

开始前请先完成“跑通第一个图片识别 Job”中的启动平台步骤。示例 Runner 额外依赖 opencv-python-headless，但仍然不加载真实模型，只生成确定性的模拟检测结果。

• 业务端 Python：POST /api/v1/streams/ 创建推理流并声明 RTSP 输入源。
• 平台：把流置为等待领取，按优先级签出租约并维护期望状态。
• 三进程 Runner：拉流、模拟检测、绘制截图并回传带幂等键的推理事件。
• 业务端或用户：查询流状态与事件流，需要时调用停止接口。

Python 业务端 → POST /api/v1/streams/ → Pending（等待 Worker）
                                                          ↓
stop + stop-ack ← should_stop ← 心跳续租 ← 三进程 Runner 领取
                                                          ↓
GET 事件流 ← StreamEvent（含截图 object_id）← 上传证据截图 ← 模拟检测

注册流式检测算法

打开平台控制台的“算法目录”，再注册一个流式算法。算法标识必须是 mock-stream-detector，与示例 Runner 声明的能力一致；其余字段与上一教程相同。

• 算法名称：模拟实时流目标检测。
• 算法标识：mock-stream-detector；版本：1.0.0。
• Worker 账号：本地演示选择 admin；正式环境应创建独立低权限账号。
• 保持“已启用”；输入与输出 Schema 可沿用最小定义。

沿用业务端和 Runner 身份

与上一教程相同，本地演示继续使用 admin 的 Basic Authentication，业务端使用 SI_API_*，Runner 使用 SI_RUNNER_*，分开命名以区分“流的创建者”和“流的执行者”。

示例脚本默认连接 http://127.0.0.1:8000。平台部署到客户现场或私有云后，把 SI_BASE_URL 设置为该环境的实际地址（或运行脚本时传 --base-url）即可，代码无需改动。

export SI_API_USERNAME=admin
export SI_API_PASSWORD='与 ADMIN_PASSWORD 相同的本地测试密码'
export SI_RUNNER_USERNAME=admin
export SI_RUNNER_PASSWORD='与 ADMIN_PASSWORD 相同的本地测试密码'

启动三进程流式 Runner

打开第二个终端，先安装 OpenCV 依赖，再启动 Runner。它每 5 秒调用一次领取接口；领到流之后派生拉流、推理、回传三个子进程，主进程负责心跳续租、响应停止指令和上报故障。

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple opencv-python-headless
export SI_RUNNER_USERNAME=admin
export SI_RUNNER_PASSWORD='与 ADMIN_PASSWORD 相同的本地测试密码'
conda run -n SovereignInference python examples/runners/mock_rtsp_stream_runner.py

用 Python 创建 RTSP 推理流

再打开第三个终端，用与上一教程相同的标准库 HTTP 客户端调用 POST /api/v1/streams/。source 是协议中立的 JSON：type 声明输入类型，uri 是地址；示例 Runner 支持 rtsp、file 和 http(s) 三种 type。

创建成功后流立即进入“等待 Worker”状态，持续运行的 Runner 会在下一次轮询时领取它。本地没有摄像头时，可以先用视频文件模拟实时流。

stream = client.post_json(
    "/api/v1/streams/",
    {
        "name": "园区东门实时检测",
        "algorithm_key": "mock-stream-detector",
        "algorithm_version": "1.0.0",
        "source": {
            "type": "rtsp",
            "uri": "rtsp://user:password@192.168.1.64:554/Stream/Channels/101",
            "transport": "tcp",
        },
        "parameters": {"threshold": 0.5},
        "idempotency_key": "gate-east-001",
    },
)

核对运行状态与事件流

观察 Runner 终端：领取成功后会打印流 ID 和脱敏后的输入源地址（密码替换为 ***），随后周期性发送心跳。业务端可以用两个接口分别从平台视角和事件视角核对运行情况。

也可以直接打开管理控制台的“Inference Streams”页面：列表展示每条流的实际/期望状态、执行 Worker 和最后心跳，详情页提供事件时间线、租约信息和停止/启动操作。

• GET /api/v1/streams/<id>/：status 从 pending 变为 running，并带有 worker_id、heartbeat_at 和租约到期时间。
• GET /api/v1/streams/<id>/events/?kind=inference：每条事件含 sequence、帧 SHA-256、brightness、detections 和 snapshot。
• snapshot.object_id 指向证据截图 StoredObject，检测框已由 Runner 绘制到 JPEG 上。
• 事件的 idempotency_key 由帧序号与帧哈希组成，网络重试不会产生重复事件。

已领取推理流 <stream-id>，输入源：rtsp://user:***@192.168.1.64:554/Stream/Channels/101
GET /api/v1/streams/<stream-id>/events/?kind=inference → 200 OK（持续新增 mock.frame.sampled 事件）

停止流并让 Runner 确认

Stream 采用两段式停止：业务端先把期望状态改为 stopped；Runner 在下一次心跳响应中看到 should_stop 后停止三个子进程，再调用 stop-ack 确认，平台才把实际状态置为已停止。已停止或故障的流可以随时调用 start 重新排队。

client.post_json(f"/api/v1/streams/{stream_id}/stop/", {})
# Runner 下一次心跳收到 should_stop → 停止子进程 → POST stop-ack → stopped
client.post_json(f"/api/v1/streams/{stream_id}/start/", {})  # 需要时重新启动

迁移到真实业务代码

接入真实算法时，保持领取、心跳、事件、停止确认和故障上报协议不变，把推理进程中的模拟检测替换为真实模型即可；截图上传与事件幂等逻辑可以直接复用。

• 业务端保存业务记录与 Stream ID 的映射，并使用 idempotency_key 防止重复建流。
• 不要把 RTSP 凭据写入日志；示例 Runner 对 URI 密码的脱敏做法可以作为基线。
• 心跳间隔应明显小于租约时长，且只对带幂等键的事件上报进行重试。
• 生产环境使用 HTTPS、独立 Worker Token、私有 Bucket 与受控网络出口。