CameraMachine/python/mcu/anheng_scale.py

import asyncio
import re
import serial
import serial.tools.list_ports
import threading
import time
from typing import Optional, Dict, Any
from utils.common import message_queue
from sockets import ConnectionManager


class AnhengScale:
    """
    安衡电子秤 纯后台串口类，无GUI（单例）
    协议格式示例: ST,TR,+ 0.0666kg
    波特率默认9600，8N1

    实时连接检测 + 自动重连：
    - 读取线程通过两种方式判定设备断开：
      (1) 读取异常/连续多次读失败 —— 覆盖"拔出USB"（COM句柄失效，读取会抛异常）；
      (2) 数据看门狗超时（默认开启 data_timeout=3.0s）—— 覆盖"设备断电但USB仍连着"，
          此时COM仍能正常打开但秤不再发数据，靠"一段时间收不到任何数据"来判定断电；
    - 断开后由后台重连线程重新扫描串口（兼容重插后COM号变化），
      按设备特征识别安衡电子秤并严格探测确认(需收到数据)后重连，指数退避重试直至成功或被显式关闭；
      设备重新上电 / 重新插拔USB后会自动恢复连接。
    - 连接状态（connected/disconnected/reconnecting/reconnected/closed）
      通过全局消息队列推送（msg_type=anheng_scale_status），前端可选消费。
    """
    # 安衡问答模式指令（秤设置COMMOD=5时可用）
    CMD_TARE = "ST\r\n"  # 硬件去皮
    CMD_ZERO = "SZ\r\n"  # 硬件归零
    CMD_READ_NET = "RN\r\n"  # 读取净重

    WEIGHT_RE = re.compile(r"([+-]?\s*\d+(?:\.\d+)?)\s*(kg|g)", re.IGNORECASE)

    # ===== 实时连接检测 + 自动重连 参数 =====
    # 安衡电子秤串口设备特征（用于重插后重新扫描识别）
    SCALE_DESC_KEYWORD = "Prolific USB-to-Serial Comm Port"
    RECONNECT_INITIAL_DELAY = 1.0  # 首次重连等待(秒)
    RECONNECT_MAX_DELAY = 5.0  # 重连退避上限(秒)
    RECONNECT_STABLE_WAIT = 1.0  # 重连成功后稳定观察(秒)
    READ_ERROR_THRESHOLD = 3  # 连续读取异常达到该次数即判定设备断开

    # 单例：用 __new__ 方式（与 BlueToothMode/Mcu 一致）。
    # 不能用 SingletonType 元类：它的 _instance_lock 是全部单例类共享的不可重入锁，
    # AnhengScale 在 DeviceControl.__init__ 内创建时（DeviceControl 也是 SingletonType 单例）会死锁
    _instance = None

    def __new__(cls, *args, **kwargs):
        """如果当前没有实例时，调用父类__new__方法，生成实例，有则返回保存的实例。"""
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(
            self,
            baudrate: int = 9600,
            websocket_manager: Optional[ConnectionManager] = None,
            auto_reconnect: bool = True,
            data_timeout: Optional[float] = 3.0,
    ):
        # __init__ 每次调用 AnhengScale(...) 都会执行，只允许初始化一次
        if getattr(self, "_initialized", False):
            return
        self._initialized = True
        self.baudrate = baudrate
        self.is_send_message = False
        # 用于向推送电子秤数据消息（与 BaseClass.sendSocketMessage 同通道）
        self.websocket_manager = websocket_manager
        self.ser: Optional[serial.Serial] = None

        # 串口缓冲区，处理粘包
        self._buf = b""

        # 解析得到的实时状态
        self.latest_data: Dict[str, Any] = {
            "stable": None,  # True稳定 / False不稳定 / None无数据
            "mode": None,  # TR净重 / NT毛重 / None
            "raw_grams": 0.0,  # 原始重量(克)，秤上报的原始值
            "net_grams": 0.0,  # 软件去皮后净重 = raw_grams - tare_grams
        }
        # 软件本地皮重（程序层面去皮，不操作秤硬件）
        self.tare_grams: float = 0.0

        # 后台读取线程标记
        self._running = False
        self._read_thread: Optional[threading.Thread] = None

        # 数据推送循环标记（防止重复创建 run 任务）
        self._pushing = False

        # ===== 实时连接检测 + 自动重连 状态 =====
        self._loop = None  # 主事件循环（异步上下文进入时捕获，供后台线程安全推送状态）
        self._port_name: str = ""  # 当前/上次串口号（重插后可能变化，仅作优先提示）
        self._should_run: bool = False  # 电子秤处于"应连接"状态（start_run 开启，disconnect 关闭）
        self._manually_closed: bool = False  # 是否被显式 disconnect() 关闭（关闭后不再自动重连）
        self._device_lost: bool = False  # 是否已检测到设备断开
        self._auto_reconnect: bool = auto_reconnect  # 是否启用自动重连
        # 数据看门狗超时(秒)：连续输出模式下，超过该时长未收到任何数据即判定"设备断电/失联"并触发断开重连。
        # 连续模式约200ms上报一行，默认 3.0s（≈15 行未上报）足够判定断电且不易误判；
        # 调小=断电检测更快但更敏感，调大=更不易误判。问答模式(COMMOD=5)无主动上报，请传 None 关闭以免误判。
        self.data_timeout: Optional[float] = data_timeout
        self._last_data_time: float = 0.0  # 最近一次收到有效数据的时间
        self._reconnecting: bool = False  # 重连线程是否存活
        self._reconnect_thread: Optional[threading.Thread] = None

    @staticmethod
    def get_all_com_ports() -> list[dict]:
        """获取本机全部串口列表，返回 [{"device":"COM3","desc":"xxx"},...]"""
        res = []
        for p in serial.tools.list_ports.comports():
            res.append({
                "device": p.device,
                "desc": p.description
            })
        return res

    def _capture_loop(self):
        """在异步上下文中捕获主事件循环；非异步上下文(如重连线程)则保留已有值，避免误建循环。"""
        try:
            self._loop = asyncio.get_running_loop()
        except RuntimeError:
            pass

    def connect(self, port_name: str) -> bool:
        """
        打开串口，启动后台线程持续接收数据
        :param port_name: 串口名称，如 "COM3"
        :return: True成功 / False失败
        """
        self._capture_loop()
        self._port_name = port_name
        self._manually_closed = False

        if self.is_connected():
            self._close_serial_link()

        try:
            self.ser = serial.Serial(
                port=port_name,
                baudrate=self.baudrate
            )
        except Exception as e:
            print(f"打开串口失败: {e}")
            self.ser = None
            return False

        self._buf = b""
        self._last_data_time = time.time()
        self._running = True
        self._should_run = True  # 连接即视为"应连接"，断开后可自动重连
        self._device_lost = False  # 连接成功，清除丢失标记
        # 启动后台读取线程
        self._read_thread = threading.Thread(target=self._background_read_loop, daemon=True)
        self._read_thread.start()
        return True

    # 探测窗口：连续输出模式下秤约每200ms上报一行，1.2s足以收到多行
    PROBE_TOTAL_WAIT = 1.2
    # 被动收集0.5s仍无数据时，发送RN主动询问（兼容COMMOD=5问答模式）
    PROBE_CMD_DELAY = 0.5

    def probe_connect(self, port_name: str, verify: bool = False) -> bool:
        """
        探测并连接：先以9600打开串口收集数据，确认是安衡电子秤后才真正连接。
        用于自动识别场景，避免把其他设备（如同样使用Prolific适配器的MCU）误认成电子秤。
        :param port_name: 串口名称，如 "COM3"
        :param verify: True=严格模式，未确认安衡协议特征则返回False（自动重连时使用，避免误连其他Prolific设备）；
                       False=沿用原有行为（打开成功即视为连接），用于首次识别
        :return: True是电子秤且已连接 / False不是电子秤(串口已关闭，可稍后重试)
        """
        self._capture_loop()
        self._port_name = port_name
        self._manually_closed = False

        if self.is_connected():
            self._close_serial_link()
        try:
            ser = serial.Serial(port=port_name, baudrate=self.baudrate, timeout=0.1)
        except Exception as e:
            print(f"打开串口失败: {e}")
            return False
        try:
            ser.reset_input_buffer()
        except Exception:
            pass

        collected = b""
        start = time.time()
        cmd_sent = False
        while time.time() - start < self.PROBE_TOTAL_WAIT:
            try:
                if ser.in_waiting:
                    collected += ser.read(ser.in_waiting)
            except Exception:
                pass
            # 连续模式下很快就有数据；问答模式(COMMOD=5)无主动上报，发RN询问
            if not cmd_sent and time.time() - start >= self.PROBE_CMD_DELAY:
                try:
                    ser.write(b"RN\r\n")
                    cmd_sent = True
                except Exception as e:
                    print("电子秤连接异常", e)
                    pass
            if self._is_anheng_data(collected):
                break
            time.sleep(0.05)

        # 严格模式：未确认安衡协议特征则判定非电子秤（用于自动重连）
        if verify and not self._is_anheng_data(collected):
            print(f"探测{port_name}：未收到安衡电子秤特征数据，判定非电子秤")
            try:
                ser.close()
            except Exception:
                pass
            return False

        # 确认是电子秤，接管串口并启动后台读取线程
        self.ser = ser
        self._buf = collected
        self._last_data_time = time.time()
        self._running = True
        self._should_run = True  # 连接即视为"应连接"，断开后可自动重连
        self._device_lost = False  # 连接成功，清除丢失标记
        self._read_thread = threading.Thread(target=self._background_read_loop, daemon=True)
        self._read_thread.start()
        return True

    @staticmethod
    def _is_anheng_data(data: bytes) -> bool:
        """判断数据是否符合安衡协议特征：状态位(ST/US) + 模式(TR/NT) + 重量值"""
        if not data:
            return False
        for line in data.split(b"\n"):
            text = line.decode("ascii", errors="ignore").strip()
            if not text:
                continue
            parts = [p.strip() for p in text.split(",")]
            has_status = any(p in ("ST", "US") for p in parts)
            has_mode = any(p in ("TR", "NT") for p in parts)
            has_weight = any(AnhengScale.WEIGHT_RE.fullmatch(p) for p in parts)
            if has_status and has_mode and has_weight:
                return True
        return False

    def _close_serial_link(self):
        """仅关闭串口链路（停止读取线程+关闭串口），不改变 _should_run/_manually_closed。
        供内部在"断开/重连前"复用，避免误触发整体停止。"""
        self._running = False
        rt = self._read_thread
        # 读取线程自己触发时不能 join 自身，否则死锁
        if rt is not None and rt is not threading.current_thread() and rt.is_alive():
            rt.join(timeout=1.0)
        self._read_thread = None
        if self.ser is not None:
            try:
                self.ser.close()
            except Exception:
                pass
            self.ser = None

    def disconnect(self):
        """显式关闭：停止后台读取与自动重连（不再自动重连，除非重新 connect/probe_connect）"""
        self._manually_closed = True
        self._should_run = False
        self._device_lost = False
        self._push_status("closed")
        self._close_serial_link()

    def is_connected(self) -> bool:
        return self.ser is not None and self.ser.is_open

    def _background_read_loop(self):
        """后台循环读取串口数据；检测读取异常/数据超时 => 判定设备断开并触发重连"""
        consecutive_errors = 0
        while self._running:
            ser = self.ser
            if ser is None or not ser.is_open:
                break
            try:
                wait_cnt = ser.in_waiting
                if wait_cnt > 0:
                    chunk = ser.read(wait_cnt)
                    if chunk:
                        self._buf += chunk
                        self._parse_buffer()
                        self._last_data_time = time.time()
                    consecutive_errors = 0
                elif self.data_timeout and (time.time() - self._last_data_time) > self.data_timeout:
                    # 看门狗：连续输出模式下一段时间无任何数据，判定设备失联
                    self._mark_device_lost("数据超时，判定设备断开")
                    break
            except Exception as e:
                consecutive_errors += 1
                print(f"电子秤读取异常({consecutive_errors}): {e}")
                # 设备拔出后串口读取会抛异常；串口已关闭或连续多次异常 => 判定断开
                if not ser.is_open or consecutive_errors >= self.READ_ERROR_THRESHOLD:
                    self._mark_device_lost(f"读取异常: {e}")
                    break
                time.sleep(0.05)
            time.sleep(0.01)  # 10ms轮询，和原GUI逻辑对齐

    def _parse_buffer(self):
        """解析缓冲区，按\n分割每行，解析协议"""
        while b"\n" in self._buf:
            line_bytes, self._buf = self._buf.split(b"\n", 1)
            self._parse_one_line(line_bytes)

    def _parse_one_line(self, line: bytes):
        try:
            text = line.decode("ascii", errors="ignore").strip()
        except Exception:
            return
        if not text:
            return

        parts = [p.strip() for p in text.split(",")]
        stable = None
        mode = None

        if "ST" in parts:
            stable = True
        elif "US" in parts:
            stable = False

        if "TR" in parts:
            mode = "TR"
        elif "NT" in parts:
            mode = "NT"

        if stable is not None:
            self.latest_data["stable"] = stable
        if mode is not None:
            self.latest_data["mode"] = mode

        # 提取重量
        for seg in parts:
            match = self.WEIGHT_RE.fullmatch(seg)
            if not match:
                continue
            val_str = match.group(1).replace(" ", "")
            val = float(val_str)
            unit = match.group(2).lower()
            raw_g = val * 1000.0 if unit == "kg" else val
            self.latest_data["raw_grams"] = raw_g
            # 软件去皮计算
            self.latest_data["net_grams"] = raw_g - self.tare_grams
            return

    def get_latest(self) -> Dict[str, Any]:
        """
        获取最新读数
        返回:
        {
            "stable": True/False/None,
            "mode": "TR"/"NT"/None,
            "raw_grams": float,    # 秤原始重量g
            "net_grams": float     # 软件去皮之后净重g
        }
        """
        return self.latest_data.copy()

    async def send_data_message(self, data: Dict[str, Any]):
        """
        推送电子秤数据消息到前端（与 BaseClass.sendSocketMessage 一致，走全局消息队列）
        :param data: get_latest() 返回的数据
        """
        if self.websocket_manager is None:
            return
        payload = {
            "code": 0,
            "msg": "安衡电子秤数据",
            "status": 2,
            "data": data,
            "msg_type": "anheng_scale",
        }
        await message_queue.put(payload)

    def _push_status(self, status: str, reason: str = ""):
        """
        通过全局消息队列向前端推送连接状态（前端可选消费，msg_type=anheng_scale_status）。
        可从后台线程安全调用（把 put 协程调度到主事件循环执行）。
        status: connected / disconnected / reconnecting / reconnected / closed
        """
        if self.websocket_manager is None:
            return
        status_text = {
            "connected": "电子秤已连接",
            "disconnected": "电子秤已断开",
            "reconnecting": "电子秤重连中",
            "reconnected": "电子秤重连成功",
            "closed": "电子秤已关闭",
        }.get(status, "电子秤状态")
        payload = {
            "code": 1 if status in ["disconnected", "reconnecting", 'closed'] else 0,
            "msg": status_text if not reason else f"{status_text}：{reason}",
            "status": -1 if status in ["disconnected", "reconnecting", 'closed'] else 2,
            "data": {"status": status, "port": self._port_name, "reason": reason},
            "msg_type": "anheng_scale_status",
        }
        try:
            if self._loop is not None:
                asyncio.run_coroutine_threadsafe(message_queue.put(payload), self._loop)
        except Exception as e:
            print(f"推送电子秤状态失败: {e}")

    def start_run(self):
        """启动数据推送循环（参考 LineControl 模式：create_task(run())），防止重复创建"""
        if self._pushing:
            return
        self._pushing = True
        self._should_run = True
        self._manually_closed = False
        self._capture_loop()
        try:
            if self._loop is None:
                self._loop = asyncio.get_event_loop()
            self._loop.create_task(self.run())
        except Exception as e:
            print(f"启动电子秤推送循环失败: {e}")
            self._pushing = False
            return
        self._push_status("connected")

    async def run(self):
        """电子秤数据推送循环（参考 LineControl.run 模式）。
        断开期间不推送读数但不退出；自动重连成功后自动恢复，无需重启本循环。"""
        while self._should_run:
            await asyncio.sleep(0.06)
            if not self.is_connected():
                continue
            data = self.get_latest()
            if self.is_send_message:
                await self.send_data_message(data)
                self.is_send_message = False
        self._pushing = False

    # ===== 实时连接检测 + 自动重连 核心 =====
    def _mark_device_lost(self, reason: str):
        """检测到设备断开（幂等）。关闭串口链路，推送状态并拉起自动重连。"""
        if self._device_lost:
            return
        self._device_lost = True
        print(f"安衡电子秤断开：{reason}")
        self._push_status("disconnected", reason)
        self._close_serial_link()
        if self._auto_reconnect and not self._manually_closed and self._should_run:
            self._ensure_reconnect_thread()

    def _ensure_reconnect_thread(self):
        """确保自动重连线程存活（已存活则不重复创建）"""
        if self._reconnecting:
            return
        self._reconnecting = True
        self._reconnect_thread = threading.Thread(target=self._reconnect_loop, daemon=True)
        self._reconnect_thread.start()

    def _interruptible_sleep(self, seconds: float):
        """可被显式关闭打断的休眠，避免手动 disconnect 后线程长时间卡住"""
        end = time.time() + seconds
        while time.time() < end and self._should_run and not self._manually_closed:
            time.sleep(0.2)

    def _find_scale_port(self):
        """重新扫描串口，返回安衡电子秤候选串口号（兼容重插后COM号变化）。
        优先上次记录的COM口，否则取第一个 Prolific 串口；找不到返回 None。"""
        try:
            ports = list(serial.tools.list_ports.comports())
        except Exception as e:
            print(f"扫描串口失败: {e}")
            return None
        if self._port_name:
            for p in ports:
                if p.device == self._port_name and self.SCALE_DESC_KEYWORD in (p.description or ""):
                    return p.device
        for p in ports:
            if self.SCALE_DESC_KEYWORD in (p.description or ""):
                return p.device
        return None

    def _try_reconnect(self) -> bool:
        """尝试重连一次：重新扫描定位电子秤 => 严格探测确认 => 打开并启动读取线程"""
        port = self._find_scale_port()
        if not port:
            print("安衡电子秤重连：未发现候选串口")
            return False
        print(f"安衡电子秤重连：尝试连接 {port}")
        return self.probe_connect(port, verify=True)

    def _reconnect_loop(self):
        """自动重连主循环：设备丢失 => 指数退避重试(重新扫描+探测) => 成功后进入稳定观察，
        直到被显式 disconnect() 停止。"""
        backoff = self.RECONNECT_INITIAL_DELAY
        announced = False
        while self._should_run and not self._manually_closed:
            if not self._device_lost and self.is_connected():
                # 健康状态：复位退避与提示标记，等待下一次丢失
                backoff = self.RECONNECT_INITIAL_DELAY
                announced = False
                self._interruptible_sleep(0.3)
                continue
            if not announced:
                self._push_status("reconnecting")
                announced = True
            if self._try_reconnect():
                self._last_data_time = time.time()
                self._push_status("reconnected")
                backoff = self.RECONNECT_INITIAL_DELAY
                self._interruptible_sleep(self.RECONNECT_STABLE_WAIT)
                continue
            self._interruptible_sleep(backoff)
            backoff = min(backoff * 2, self.RECONNECT_MAX_DELAY)
        self._reconnecting = False
        self._reconnect_thread = None
        # 退出时兜底关闭串口链路，处理"重连探测与手动关闭竞态"可能残留的串口/读取线程
        self._close_serial_link()

    def software_tare(self) -> bool:
        """
        【软件本地去皮，不发送指令给电子秤硬件】
        只有秤稳定时才能执行，记录当前原始重量作为皮重
        :return: True成功；False失败（未稳定）
        """
        if self.latest_data.get("stable") is not True:
            return False
        self.tare_grams = self.latest_data["raw_grams"]
        self.latest_data["net_grams"] = self.latest_data["raw_grams"] - self.tare_grams
        return True

    def clear_software_tare(self):
        """清除软件去皮，皮重置0"""
        self.tare_grams = 0.0
        self.latest_data["net_grams"] = self.latest_data["raw_grams"]

    def send_hardware_cmd(self, cmd: str):
        """底层：向串口发送原始指令，需要秤设置COMMOD=5问答模式才生效"""
        if not self.is_connected():
            return False
        try:
            self.ser.write(cmd.encode("ascii"))
            return True
        except Exception as e:
            print(f"发送指令失败 {e}")
            return False

    def hardware_tare(self):
        """【硬件去皮】发送指令让电子秤本体执行去皮，秤必须开启问答模式COMMOD=5"""
        return self.send_hardware_cmd(self.CMD_TARE)

    def hardware_zero(self):
        """【硬件归零】发送指令让电子秤本体归零，秤必须开启问答模式COMMOD=5"""
        return self.send_hardware_cmd(self.CMD_ZERO)