16337
d9f78e0b48
- ResultReporter: 截图通过 cv2.imencode 编码为 JPEG base64， 直接放入 Redis 消息，不再调用 ImageStorageManager 保存本地文件 - AlarmUploadWorker: 从 Redis 读取 base64 解码为字节流， 使用 put_object(Body=bytes) 直接上传 COS，移除 local: 回退逻辑 - 移除 AlarmInfo.snapshot_local_path，改为 snapshot_b64 - COS 未配置时返回 None 进入重试（不再静默回退本地路径） Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
219 lines
7.1 KiB
Python
219 lines
7.1 KiB
Python
"""
|
||
结果上报模块
|
||
使用本地 Redis 缓冲 + HTTP 上报替代 MQTT
|
||
|
||
告警流程：
|
||
算法产出告警 → report_alarm() LPUSH 到 Redis → AlarmUploadWorker 异步消费
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
from typing import Any, Dict, List, Optional
|
||
from dataclasses import dataclass, field
|
||
|
||
import numpy as np
|
||
import redis
|
||
|
||
from config.settings import get_settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class AlarmInfo:
|
||
"""告警信息类（新格式，对齐云端 alarm_event 表）"""
|
||
alarm_id: str
|
||
alarm_type: str
|
||
device_id: str
|
||
scene_id: str
|
||
event_time: str # ISO8601
|
||
alarm_level: int # 1-4
|
||
snapshot_b64: Optional[str] = None # Base64 编码的 JPEG 截图
|
||
algorithm_code: Optional[str] = None
|
||
confidence_score: Optional[float] = None
|
||
ext_data: Optional[Dict[str, Any]] = field(default_factory=dict)
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
"""转换为字典（写入 Redis 的 JSON 格式）"""
|
||
return {
|
||
"alarm_id": self.alarm_id,
|
||
"alarm_type": self.alarm_type,
|
||
"device_id": self.device_id,
|
||
"scene_id": self.scene_id,
|
||
"event_time": self.event_time,
|
||
"alarm_level": self.alarm_level,
|
||
"snapshot_b64": self.snapshot_b64,
|
||
"algorithm_code": self.algorithm_code,
|
||
"confidence_score": self.confidence_score,
|
||
"ext_data": self.ext_data,
|
||
}
|
||
|
||
|
||
def generate_alarm_id(device_id: str) -> str:
|
||
"""
|
||
生成告警ID
|
||
格式: edge_{device_id}_{YYYYMMDDHHmmss}_{6位uuid}
|
||
"""
|
||
timestamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
|
||
uid = uuid.uuid4().hex[:6]
|
||
return f"edge_{device_id}_{timestamp}_{uid}"
|
||
|
||
|
||
# Redis 队列 Key 常量
|
||
REDIS_KEY_ALARM_PENDING = "local:alarm:pending"
|
||
REDIS_KEY_ALARM_RETRY = "local:alarm:retry"
|
||
REDIS_KEY_ALARM_DEAD = "local:alarm:dead"
|
||
|
||
|
||
class ResultReporter:
|
||
"""结果上报器
|
||
|
||
将告警写入本地 Redis 队列，由 AlarmUploadWorker 异步消费上传。
|
||
report_alarm() 方法使用 LPUSH，算法线程零阻塞。
|
||
"""
|
||
|
||
def __init__(self):
|
||
self._settings = get_settings()
|
||
self._redis: Optional[redis.Redis] = None
|
||
self._logger = logging.getLogger("result_reporter")
|
||
|
||
self._performance_stats = {
|
||
"alerts_generated": 0,
|
||
"alerts_queued": 0,
|
||
"queue_failures": 0,
|
||
}
|
||
|
||
self._db_manager = None
|
||
|
||
self._logger.info("ResultReporter 初始化完成（Redis 缓冲模式）")
|
||
|
||
def initialize(self):
|
||
"""初始化 Redis 连接"""
|
||
try:
|
||
from config.database import get_sqlite_manager
|
||
self._db_manager = get_sqlite_manager()
|
||
except Exception as e:
|
||
self._logger.warning(f"本地数据库初始化失败: {e}")
|
||
|
||
# 初始化 Redis 连接（使用本地 Redis）
|
||
redis_cfg = self._settings.local_redis
|
||
try:
|
||
self._redis = redis.Redis(
|
||
host=redis_cfg.host,
|
||
port=redis_cfg.port,
|
||
db=redis_cfg.db,
|
||
password=redis_cfg.password,
|
||
decode_responses=True,
|
||
socket_connect_timeout=5,
|
||
)
|
||
self._redis.ping()
|
||
self._logger.info(
|
||
f"Redis 连接成功: {redis_cfg.host}:{redis_cfg.port}/{redis_cfg.db}"
|
||
)
|
||
except Exception as e:
|
||
self._logger.error(f"Redis 连接失败: {e}")
|
||
self._redis = None
|
||
|
||
def report_alarm(self, alarm_info: AlarmInfo, screenshot: Optional[np.ndarray] = None) -> bool:
|
||
"""
|
||
上报告警（写入 Redis 队列）
|
||
|
||
Args:
|
||
alarm_info: 告警信息
|
||
screenshot: 抓拍图片（numpy 数组）
|
||
|
||
Returns:
|
||
是否成功写入队列
|
||
"""
|
||
self._performance_stats["alerts_generated"] += 1
|
||
|
||
# 将截图编码为 JPEG base64，直接通过 Redis 传递给 Worker 上传 COS
|
||
if screenshot is not None:
|
||
try:
|
||
import cv2
|
||
import base64
|
||
success, buffer = cv2.imencode('.jpg', screenshot, [cv2.IMWRITE_JPEG_QUALITY, 85])
|
||
if success:
|
||
alarm_info.snapshot_b64 = base64.b64encode(buffer.tobytes()).decode('ascii')
|
||
else:
|
||
self._logger.warning("截图 JPEG 编码失败")
|
||
except Exception as e:
|
||
self._logger.error(f"编码截图失败: {e}")
|
||
|
||
# 写入 Redis 队列
|
||
if self._redis is None:
|
||
self._performance_stats["queue_failures"] += 1
|
||
self._logger.error("Redis 未连接，无法写入告警队列")
|
||
return False
|
||
|
||
try:
|
||
alarm_json = json.dumps(alarm_info.to_dict(), ensure_ascii=False)
|
||
self._redis.lpush(REDIS_KEY_ALARM_PENDING, alarm_json)
|
||
self._performance_stats["alerts_queued"] += 1
|
||
|
||
self._logger.info(
|
||
f"告警已入队: alarm_id={alarm_info.alarm_id}, "
|
||
f"type={alarm_info.alarm_type}, device={alarm_info.device_id}"
|
||
)
|
||
return True
|
||
|
||
except Exception as e:
|
||
self._performance_stats["queue_failures"] += 1
|
||
self._logger.error(f"写入 Redis 队列失败: {e}")
|
||
return False
|
||
|
||
def report_alarm_resolve(self, resolve_data: dict) -> bool:
|
||
"""上报告警结束事件（写入 Redis 队列）"""
|
||
if self._redis is None:
|
||
self._logger.error("Redis 未连接，无法写入 resolve 队列")
|
||
return False
|
||
try:
|
||
resolve_data["_type"] = "resolve" # 标记类型，worker 据此分流
|
||
resolve_json = json.dumps(resolve_data, ensure_ascii=False)
|
||
self._redis.lpush(REDIS_KEY_ALARM_PENDING, resolve_json)
|
||
self._logger.info(f"告警结束事件已入队: alarm_id={resolve_data.get('alarm_id')}")
|
||
return True
|
||
except Exception as e:
|
||
self._logger.error(f"写入 resolve 队列失败: {e}")
|
||
return False
|
||
|
||
def get_statistics(self) -> Dict[str, Any]:
|
||
"""获取统计信息"""
|
||
stats = self._performance_stats.copy()
|
||
|
||
if self._redis:
|
||
try:
|
||
stats["pending_count"] = self._redis.llen(REDIS_KEY_ALARM_PENDING)
|
||
stats["retry_count"] = self._redis.llen(REDIS_KEY_ALARM_RETRY)
|
||
stats["dead_count"] = self._redis.llen(REDIS_KEY_ALARM_DEAD)
|
||
except Exception:
|
||
pass
|
||
|
||
stats["redis_connected"] = self._redis is not None
|
||
|
||
return stats
|
||
|
||
def close(self):
|
||
"""关闭上报器"""
|
||
self._logger.info("ResultReporter 资源清理")
|
||
|
||
if self._db_manager:
|
||
try:
|
||
self._db_manager.close()
|
||
except Exception:
|
||
pass
|
||
|
||
if self._redis:
|
||
try:
|
||
self._redis.close()
|
||
except Exception:
|
||
pass
|
||
|
||
self._logger.info("ResultReporter 清理完成")
|
||
|
||
def cleanup(self):
|
||
"""清理资源（别名）"""
|
||
self.close()
|