如何持续连接海康威视ANPR相机事件流并忽略错误

问题背景 #

我尝试连接停车场内的海康威视相机监控车辆进出，用hikvisionapi获取事件数据：

• 初始代码能拿到第一个ANPR事件，但之后触发IndexError；
• 加了无限循环的写法，能收到第一个ANPR事件，但之后再也检测不到新事件，代码一直跑却没输出。

需求：实现保持事件流连接并自动忽略/处理错误的监控逻辑。

问题根源 #

1. 异步逻辑混用同步代码：AsyncClient是异步IO实现，之前用同步while True套异步循环，没在正确的异步上下文里运行，导致事件流结束后无法重建有效连接。
2. 事件结构未做安全校验：部分非ANPR事件可能缺失EventNotificationAlert或eventType字段，直接硬索引会抛出异常中断整个事件流。
3. 无断连自动重试机制：相机端可能会主动断开长连接，需要自动重新发起连接。

解决代码示例 #

用asyncio原生异步循环管理事件流，内部做异常捕获和字段安全校验，同时支持断连自动重试：

import asyncio
import logging
from hikvisionapi import AsyncClient

# 配置日志输出
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(message)s")

async def watch_anpr():
    # 初始化相机客户端
    cam = AsyncClient('http://XXX.XX.XXX.X:XXXX', '你的用户名', '你的密码')
    
    while True:
        try:
            # 持续监听事件流
            async for event in cam.Event.notification.alertStream(method='get', type='stream', timeout=300):
                # 安全获取事件告警字段，避免KeyError
                alert = event.get('EventNotificationAlert')
                if not alert:
                    continue
                
                event_type = alert.get('eventType')
                if event_type == 'ANPR':
                    # 输出ANPR事件详情，比如车牌信息
                    logging.info(f"识别到车牌: {alert.get('ANPR', {}).get('licensePlate')}")
                # 可选：记录其他类型事件
                # else:
                #     logging.info(f"收到其他事件: {event_type}")
        
        except Exception as e:
            # 捕获所有异常，记录后重试连接
            logging.error(f"事件流异常，2秒后重试: {str(e)}")
            await asyncio.sleep(2)

if __name__ == '__main__':
    # 启动异步事件循环
    asyncio.run(watch_anpr())

关键优化点 #

• 纯异步上下文运行：用asyncio.run启动异步主函数，所有IO操作都在异步环境中执行，避免同步/异步混用导致的连接异常。
• 安全字段访问：用dict.get()替代直接索引，即使字段缺失也不会抛出异常，保证事件流不中断。
• 自动重连：事件流抛出异常（断连、解析错误等）后，自动延迟重试，确保监控不中断。
• 异常隔离：单个事件的解析错误不会导致整个连接断开，仅跳过异常事件继续监听。

额外建议 #

1. 检查相机配置：确认相机已启用车牌识别事件，且事件推送的协议/参数设置正确。
2. 调整超时参数：把timeout设为300（5分钟）这类具体值，比None更适配相机的长连接策略。
3. 日志细化：可以添加相机IP、事件时间戳等信息，方便后续排查问题。

内容的提问来源于stack exchange，提问作者Danish Ajaib