vd_960/DBNMQTTool/dbn_mqtt_tool/protocol.py

"""
DLD960 IoT MQTT 协议定义

基于《DLD960_IoT_MQTT协议.md》V1.01
双主题结构：{sn}/srv（服务器下发）、{sn}/dev（设备上报）
消息类型由 JSON `cmd` 字段区分。
"""

import json
import time
from typing import Optional, Any
from dataclasses import dataclass, field, asdict


# ============================================================
# Topic 结构（V1.01 — 双主题）
# ============================================================

def topic_down(dev_serial: str) -> str:
    """服务器下发 — 设备订阅此主题接收所有命令"""
    return f"dld960/{dev_serial}/srv"


def topic_up(dev_serial: str) -> str:
    """设备上报 — 设备发布到此主题（数据、事件、心跳、响应）"""
    return f"dld960/{dev_serial}/dev"


# 保持向后兼容的别名（4.7 iot_topic_set 仍使用 topic_pub / topic_sub）
def topic_dev_pub(dev_serial: str) -> str:
    """设备发布主题（= topic_up）"""
    return topic_up(dev_serial)


def topic_dev_sub(dev_serial: str) -> str:
    """设备订阅主题（= topic_down）"""
    return topic_down(dev_serial)


# 服务器端通配符订阅（监听所有设备）
TOPIC_ALL_DEVICE_UP = "dld960/+/dev"


# ============================================================
# 命令枚举
# ============================================================

CMD_DEV_SERIAL_SET   = "dev_serial_set"
CMD_DEV_INFO_QUERY   = "dev_info_query"
CMD_SSC_NET_SET      = "ssc_net_set"
CMD_SSC_NET_QUERY    = "ssc_net_query"
CMD_IOT_NET_SET      = "iot_net_set"
CMD_IOT_NET_QUERY    = "iot_net_query"
CMD_IOT_TOPIC_SET    = "iot_topic_set"
CMD_IOT_TOPIC_QUERY  = "iot_topic_query"
CMD_PWD_VERIFY       = "pwd_verify"
CMD_PWD_SET          = "pwd_set"
CMD_FACTORY_RESET    = "factory_reset"
CMD_DEVICE_RESET     = "device_reset"
CMD_LOOP_PARAM_SET   = "loop_param_set"
CMD_LOOP_PARAM_QUERY = "loop_param_query"
CMD_REPORT_CONFIG    = "report_config"

# 设备上报
CMD_LOOP_DATA    = "loop_data"
CMD_EVENT_REPORT = "event_report"
CMD_HEARTBEAT    = "heartbeat"

# 配置类命令 → topic
CONFIG_SET_COMMANDS = {CMD_DEV_SERIAL_SET, CMD_SSC_NET_SET, CMD_IOT_NET_SET, CMD_IOT_TOPIC_SET, CMD_LOOP_PARAM_SET}
CONFIG_QUERY_COMMANDS = {CMD_DEV_INFO_QUERY, CMD_SSC_NET_QUERY, CMD_IOT_NET_QUERY, CMD_IOT_TOPIC_QUERY, CMD_LOOP_PARAM_QUERY}
CTRL_COMMANDS = {CMD_PWD_VERIFY, CMD_PWD_SET, CMD_FACTORY_RESET, CMD_DEVICE_RESET, CMD_REPORT_CONFIG}


# ============================================================
# 错误码
# ============================================================

ERR_SUCCESS        = 0
ERR_PARAM_ERROR    = 1
ERR_PWD_FAILED     = 2
ERR_DEVICE_BUSY    = 3
ERR_UNSUPPORTED    = 4
ERR_INTERNAL       = 5
ERR_DATA_TOO_LONG  = 6

ERROR_MSGS = {
    0: "成功",
    1: "参数错误",
    2: "密码验证失败",
    3: "设备忙",
    4: "不支持的命令",
    5: "内部错误",
    6: "数据超长",
}


# ============================================================
# 频率档位 / 输出模式 / 事件类型
# ============================================================

FREQ_LEVELS = ["high", "mid_high", "mid_low", "low"]
OUTPUT_MODES = ["exist", "enter_pulse", "leave_pulse", "direction"]
EVENT_TYPES = ["car_enter", "car_leave", "loop_cut", "loop_restore"]


# ============================================================
# 消息构建
# ============================================================

_msg_id_counter = 0

def next_msg_id() -> int:
    global _msg_id_counter
    _msg_id_counter += 1
    return _msg_id_counter


def build_request(cmd: str, data: Optional[dict] = None) -> dict:
    """构建通用请求消息"""
    msg: dict[str, Any] = {
        "msg_id": next_msg_id(),
        "cmd": cmd,
        "ts": int(time.time()),
    }
    if data is not None:
        msg["data"] = data
    return msg


def build_response(cmd: str, msg_id: int, code: int = 0, msg_text: str = "success", data: Optional[dict] = None) -> dict:
    """构建响应消息"""
    resp: dict[str, Any] = {
        "msg_id": msg_id,
        "cmd": cmd,
        "ts": int(time.time()),
        "code": code,
        "msg": msg_text,
    }
    if data is not None:
        resp["data"] = data
    return resp


def get_topic_for_cmd(cmd: str, dev_serial: str) -> str:
    """获取命令对应的下发 topic。所有命令统一使用 topic_down"""
    return topic_down(dev_serial)


# ============================================================
# 特定命令的 data 构建器
# ============================================================

def data_dev_serial_set(dev_serial: str) -> dict:
    return {"dev_serial": dev_serial}


def data_ssc_net_set(dev_ip: str, subnet_mask: str, route_ip: str,
                     lssc_ip: str, dns: str, port: int) -> dict:
    return {
        "dev_ip": dev_ip,
        "subnet_mask": subnet_mask,
        "route_ip": route_ip,
        "lssc_ip": lssc_ip,
        "dns": dns,
        "port": port,
    }


def data_iot_net_set(host: str, port: int = 1883,
                     client_id: str = "", username: str = "", password: str = "") -> dict:
    return {
        "host": host,
        "port": port,
        "client_id": client_id,
        "username": username,
        "password": password,
    }


def data_iot_topic_set(client_id_enable: bool, topic_pub: str, topic_sub: str) -> dict:
    return {
        "client_id_enable": client_id_enable,
        "topic_pub": topic_pub,
        "topic_sub": topic_sub,
    }


def data_pwd_verify(password: str) -> dict:
    return {"password": password}


def data_pwd_set(old_password: str, new_password: str) -> dict:
    return {"old_password": old_password, "new_password": new_password}


def data_loop_param_set(channels: list[dict], auto_mode: bool = False) -> dict:
    return {"auto_mode": auto_mode, "channels": channels}


def data_report_config(sensor_type: int = 12, enable: bool = True,
                       once: bool = False, env_eval: bool = False,
                       interval: int = 5, ack_required: bool = False,
                       timeout: int = 0) -> dict:
    return {
        "sensor_type": sensor_type,
        "enable": enable,
        "once": once,
        "env_eval": env_eval,
        "interval": interval,
        "ack_required": ack_required,
        "timeout": timeout,
    }


# ============================================================
# 解析设备上报
# ============================================================

def parse_topic_dev_serial(topic: str) -> Optional[str]:
    """从 topic 中提取设备序列码"""
    parts = topic.split("/")
    if len(parts) >= 2 and parts[0] == "dld960":
        return parts[1]
    return None