vd_960/DBNetClient/tcp_json_client.py

"""
TCP JSON Protocol Client — DLD960 TCP JSON 协议客户端

行分隔 JSON 帧，请求-响应模式，自动 msg_id 递增。
"""

import json
import socket
import threading
import time


class TcpJsonError(Exception):
    """协议错误"""
    pass


class TcpJsonClient:
    """DLD960 TCP JSON 协议客户端"""

    def __init__(self, log_callback=None):
        self._sock: socket.socket | None = None
        self._msg_id = 0
        self._buf = b""
        self._lock = threading.Lock()
        self._running = False
        self._recv_thread: threading.Thread | None = None
        self._pending: dict[int, dict] = {}  # msg_id -> response future
        self._push_handlers: dict[str, callable] = {}
        self._log = log_callback or (lambda msg: None)

    # ---- Connection ----

    def connect(self, host: str, port: int = 5960, timeout: float = 5.0) -> None:
        """建立 TCP 连接"""
        if self._sock:
            self.disconnect()
        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self._sock.settimeout(timeout)
        self._sock.connect((host, port))
        self._sock.settimeout(None)  # 接收线程用阻塞模式
        self._running = True
        self._buf = b""
        self._recv_thread = threading.Thread(target=self._recv_loop, daemon=True)
        self._recv_thread.start()
        self._log(f"Connected to {host}:{port}")

    def disconnect(self) -> None:
        """断开连接"""
        self._running = False
        if self._sock:
            try:
                self._sock.shutdown(socket.SHUT_RDWR)
            except OSError:
                pass
            self._sock.close()
            self._sock = None
        self._buf = b""
        with self._lock:
            # 唤醒所有等待的请求
            for mid, fut in self._pending.items():
                fut["error"] = "disconnected"
                fut["event"].set()
            self._pending.clear()
        self._log("Disconnected")

    @property
    def is_connected(self) -> bool:
        return self._sock is not None and self._running

    # ---- Sending ----

    def _next_msg_id(self) -> int:
        self._msg_id += 1
        return self._msg_id

    def send_command(self, cmd: str, data: dict | None = None,
                     timeout: float = 5.0) -> dict:
        """发送命令并等待响应，返回完整响应 dict"""
        if not self.is_connected:
            raise TcpJsonError("Not connected")

        msg_id = self._next_msg_id()
        frame = {"msg_id": msg_id, "cmd": cmd, "ts": int(time.time())}
        if data:
            frame["data"] = data

        raw = json.dumps(frame, separators=(",", ":")) + "\n"
        event = threading.Event()
        with self._lock:
            self._pending[msg_id] = {"event": event, "response": None, "error": None}

        try:
            self._sock.sendall(raw.encode("utf-8"))
            self._log(f">>> {cmd}  (msg_id={msg_id})  [{len(raw)} bytes]")
            self._log(f">>> RAW: {raw.strip()}")
        except OSError as e:
            with self._lock:
                self._pending.pop(msg_id, None)
            raise TcpJsonError(f"Send failed: {e}")

        if not event.wait(timeout):
            with self._lock:
                self._pending.pop(msg_id, None)
            raise TcpJsonError(f"Timeout waiting for response to {cmd}")

        with self._lock:
            fut = self._pending.pop(msg_id, None)
        if fut is None:
            raise TcpJsonError("Response already consumed")
        if fut["error"]:
            raise TcpJsonError(fut["error"])
        return fut["response"]

    # ---- Push handlers ----

    def on_push(self, cmd: str, handler: callable) -> None:
        """注册主动推送处理器 handler(cmd, data_dict)"""
        self._push_handlers[cmd] = handler

    # ---- Receive loop ----

    def _recv_loop(self) -> None:
        """后台接收线程"""
        while self._running:
            try:
                data = self._sock.recv(4096)
                if not data:
                    self._log("Connection closed by server")
                    break
                self._buf += data
                self._parse_frames()
            except OSError:
                if self._running:
                    self._log("Receive error")
                break

    def _parse_frames(self) -> None:
        """从缓冲区提取完整 JSON 帧并分发"""
        while True:
            idx = self._buf.find(b"\n")
            if idx == -1:
                break
            raw = self._buf[:idx].decode("utf-8", errors="replace")
            self._buf = self._buf[idx + 1:]
            if not raw.strip():
                continue
            try:
                obj = json.loads(raw)
            except json.JSONDecodeError:
                self._log(f"<<< (invalid JSON) {raw[:80]}")
                continue

            cmd = obj.get("cmd", "")
            msg_id = obj.get("msg_id", 0)
            # 检查是否有 code 字段（响应帧）
            if "code" in obj:
                code = obj.get("code", -1)
                self._log(f"<<< {cmd}  code={code}  msg_id={msg_id}")
                with self._lock:
                    fut = self._pending.get(msg_id)
                if fut:
                    fut["response"] = obj
                    fut["event"].set()
            else:
                # 主动推送帧
                self._log(f"<<< PUSH {cmd}  msg_id={msg_id}")
                handler = self._push_handlers.get(cmd)
                if handler:
                    try:
                        handler(cmd, obj.get("data", {}))
                    except Exception as e:
                        self._log(f"Push handler error: {e}")


# ---- 便捷高层 API ----

class DBNetClient:
    """DLD960 设备 TCP JSON 协议高层接口"""

    def __init__(self, log_callback=None):
        self._tcp = TcpJsonClient(log_callback)

    # Connection
    def connect(self, host: str, port: int = 5960) -> None:
        self._tcp.connect(host, port)

    def disconnect(self) -> None:
        self._tcp.disconnect()

    @property
    def is_connected(self) -> bool:
        return self._tcp.is_connected

    # Auth
    def pwd_verify(self, password: str) -> dict:
        return self._tcp.send_command("pwd_verify", {"password": password})

    # Device Info
    def dev_info_query(self) -> dict:
        return self._tcp.send_command("dev_info_query")

    def dev_serial_set(self, dev_serial: str) -> dict:
        return self._tcp.send_command("dev_serial_set", {"dev_serial": dev_serial})

    # SSC Network
    def ssc_net_query(self) -> dict:
        return self._tcp.send_command("ssc_net_query")

    def ssc_net_set(self, dev_ip: str = "", subnet_mask: str = "",
                    route_ip: str = "", lssc_ip: str = "",
                    dns: str = "", port: int = 0) -> dict:
        data = {}
        if dev_ip:
            data["dev_ip"] = dev_ip
        if subnet_mask:
            data["subnet_mask"] = subnet_mask
        if route_ip:
            data["route_ip"] = route_ip
        if lssc_ip:
            data["lssc_ip"] = lssc_ip
        if dns:
            data["dns"] = dns
        if port:
            data["port"] = port
        return self._tcp.send_command("ssc_net_set", data)

    # IoT Network
    def iot_net_query(self) -> dict:
        return self._tcp.send_command("iot_net_query")

    def iot_net_set(self, host: str = "", port: int = 0,
                    client_id: str = "", username: str = "",
                    password: str = "") -> dict:
        data = {}
        if host:
            data["host"] = host
        if port:
            data["port"] = port
        if client_id:
            data["client_id"] = client_id
        if username:
            data["username"] = username
        if password:
            data["password"] = password
        return self._tcp.send_command("iot_net_set", data)

    # IoT Topics
    def iot_topic_query(self) -> dict:
        return self._tcp.send_command("iot_topic_query")

    def iot_topic_set(self, client_id_enable: bool = None,
                      topic_pub: str = "", topic_sub: str = "") -> dict:
        data = {}
        if client_id_enable is not None:
            data["client_id_enable"] = client_id_enable
        if topic_pub:
            data["topic_pub"] = topic_pub
        if topic_sub:
            data["topic_sub"] = topic_sub
        return self._tcp.send_command("iot_topic_set", data)

    # Password
    def pwd_set(self, old_password: str, new_password: str) -> dict:
        return self._tcp.send_command("pwd_set",
                                       {"old_password": old_password,
                                        "new_password": new_password})

    # System
    def factory_reset(self) -> dict:
        return self._tcp.send_command("factory_reset")

    def device_reset(self) -> dict:
        return self._tcp.send_command("device_reset")

    # Loop
    def loop_param_query(self) -> dict:
        return self._tcp.send_command("loop_param_query")

    def loop_param_set(self, channels: list[dict],
                       auto_mode: bool = False) -> dict:
        return self._tcp.send_command("loop_param_set",
                                       {"auto_mode": auto_mode,
                                        "channels": channels})

    # Push
    def on_push(self, cmd: str, handler: callable) -> None:
        self._tcp.on_push(cmd, handler)