chongming-plugins/aibox_sophon_adapter/models/alarm_models.py

"""
告警上报 — Pydantic 数据模型

包含请求体和响应体。请求体声明了规范中所有已知字段，
未声明的额外字段通过 model_config extra=allow 保留。
"""
from typing import Any, Optional
from pydantic import BaseModel, Field, field_validator


# ============================================================
# 告警报文子结构
# ============================================================

class GpsInfo(BaseModel):
    """GPS 位置信息"""
    model_config = {"extra": "allow"}

    available: bool = Field(default=False, description="是否有 GPS 功能")
    kSpeed: float = Field(default=0.0, description="对地速度 (km/h)")
    nSpeed: float = Field(default=0.0, description="对地速率 (节)")
    latitude: float = Field(default=0.0, description="纬度")
    longitude: float = Field(default=0.0, description="经度")
    angleCourse: float = Field(default=0.0, description="GPS 角度")

    @field_validator('kSpeed', 'nSpeed', 'latitude', 'longitude', 'angleCourse', mode='before')
    @classmethod
    def _empty_str_to_zero(cls, v: Any) -> Any:
        """设备可能上报空字符串，转为 0.0"""
        if v == '' or v is None:
            return 0.0
        return v


class MediaInfo(BaseModel):
    """视频通道信息"""
    model_config = {"extra": "allow"}

    MediaName: str = Field(default="", description="通道标识")
    MediaDesc: str = Field(default="", description="通道描述")
    MediaUrl: str = Field(default="", description="通道视频流地址")
    MediaWidth: int = Field(default=0, description="通道视频宽度")
    MediaHeight: int = Field(default=0, description="通道视频高度")
    SubId: str = Field(default="", description="关联 GB28181 通道号")
    GBTransport: bool = Field(default=False, description="是否接入 GB28181")
    RtspTransport: bool = Field(default=False, description="是否开启 RTSP 转发")
    Params: Optional[Any] = Field(default=None, description="预留其他参数，可能是 dict、list 或 null")

    @field_validator('MediaWidth', 'MediaHeight', mode='before')
    @classmethod
    def _empty_str_to_zero(cls, v: Any) -> Any:
        """设备可能上报空字符串，转为 0"""
        if v == '' or v is None:
            return 0
        return v


class ResultProperty(BaseModel):
    """告警 Result 中的属性项"""
    model_config = {"extra": "allow"}

    property: str = Field(default="", description="属性标识")
    desc: str = Field(default="", description="属性描述")
    value: Any = Field(default=None, description="属性值（String/Int/Float）")
    display: str = Field(default="", description="属性详细信息")
    type: str = Field(default="", description="属性值类型")


class AlarmResult(BaseModel):
    """违规检测结果（结构随算法变化，extra=allow 兜底）"""
    model_config = {"extra": "allow"}

    Type: str = Field(default="", description="违规类型标识（NoHelmet / Intrude / PeopleCross 等）")
    Description: str = Field(default="", description="违规描述")
    Cropped: bool = Field(default=False, description="是否裁剪")
    RegType: str = Field(default="Rectangle", description="上报点集类型（Rectangle / Line / Polygon）")
    RelativeBox: list[float] = Field(default_factory=list, description="矩形目标位置 [X,Y,Width,Height] 百分比")
    RelativeRegion: list[dict] = Field(default_factory=list, description="多边形目标位置")
    Properties: list[ResultProperty] = Field(default_factory=list, description="算法特殊属性")


# ============================================================
# 告警上报请求体
# ============================================================

class AlarmReportRequest(BaseModel):
    """
    告警上报请求体 — 声明规范中所有已知字段

    未声明的额外字段通过 extra=allow 保留，完整数据可通过 model_dump() 获取。
    盒子检测到违规事件后，先上传视频再上报此 JSON 告警报文。
    """
    model_config = {"extra": "allow"}

    # 设备信息
    BoardIp: str = Field(default="", description="盒子网络地址")
    BoardId: str = Field(default="", description="盒子唯一标识")

    # 告警标识
    AlarmId: str = Field(default="", description="告警标识")
    UniqueId: str = Field(default="", description="告警唯一标识（与 AlarmId 等价)")

    # 透传字段
    Addition: Optional[dict] = Field(default=None, description="透传字段")

    # 截图数据
    ImageData: str = Field(default="", description="无标注告警图 JPEG Base64 编码")
    ImageDataLabeled: str = Field(default="", description="有标注告警图 JPEG Base64 编码")

    # 截图本地路径
    LocalRawPath: str = Field(default="", description="无标注告警图在盒子中的相对路径")
    LocalLabeledPath: str = Field(default="", description="有标注告警图在盒子中的相对路径")

    # 通道信息
    Media: Optional[MediaInfo] = Field(default=None, description="配置的媒体通道")

    # 违规描述
    Summary: str = Field(default="", description="违规类型（NoHelmet / Intrude / PeopleCross 等）")

    # 任务信息
    TaskSession: str = Field(default="", description="任务编号")
    GBDeviceId: str = Field(default="", description="国标设备编号，不存在则为空字符串")
    GBTaskChnId: str = Field(default="", description="任务对应的国标通道编号，不存在则为空字符串")
    TaskDesc: str = Field(default="", description="任务描述")

    # 时间
    Time: str = Field(default="", description="告警时间戳字符串")
    TimeStamp: int = Field(default=0, description="告警时间戳（微秒）")

    # 视频
    VideoFile: str = Field(default="", description="关联的视频标识（视频上传接口返回）")

    # GPS
    GPS: Optional[GpsInfo] = Field(default=None, description="当前位置信息")

    # 违规结果
    Result: Optional[AlarmResult] = Field(default=None, description="违规检测结果")

    # 预留字段
    Type: int = Field(default=0, description="预留字段")
    DataId: int = Field(default=0, description="数据 ID")
    Id: int = Field(default=0, description="记录 ID")

    @field_validator('DataId', 'Id', 'TimeStamp', 'Type', mode='before')
    @classmethod
    def _empty_str_to_zero(cls, v: Any) -> Any:
        """设备可能上报空字符串，转为 0"""
        if v == '' or v is None:
            return 0
        return v


# ============================================================
# 告警上报响应体
# ============================================================

class AlarmReportResult(BaseModel):
    """告警上报接口返回状态"""
    Code: int = Field(default=0, description="接口状态错误码，0 表示成功，非 0 表示失败")
    Desc: str = Field(default="", description="错误码描述")


class AlarmReportResponse(BaseModel):
    """告警上报接口完整响应

    仅当 HTTP 200 + Code 非 0 时，盒子认为上报失败。
    """
    Result: AlarmReportResult = Field(default_factory=AlarmReportResult, description="接口状态返回")