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
    """
    # 安衡问答模式指令（秤设置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)

    # 单例：用 __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):
        # __init__ 每次调用 AnhengScale(...) 都会执行，只允许初始化一次
        if getattr(self, "_initialized", False):
            return
        self._initialized = True
        self.baudrate = baudrate
        # 用于向推送电子秤数据消息（与 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

    @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 connect(self, port_name: str) -> bool:
        """
        打开串口，启动后台线程持续接收数据
        :param port_name: 串口名称，如 "COM3"
        :return: True成功 / False失败
        """
        if self.is_connected():
            self.disconnect()

        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._running = True
        # 启动后台读取线程
        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) -> bool:
        """
        探测并连接：先以9600打开串口收集数据，确认是安衡电子秤后才真正连接。
        用于自动识别场景，避免把其他设备（如同样使用Prolific适配器的MCU）误认成电子秤。
        :param port_name: 串口名称，如 "COM3"
        :return: True是电子秤且已连接 / False不是电子秤(串口已关闭，可稍后重试)
        """
        if self.is_connected():
            self.disconnect()
        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:
            if ser.in_waiting:
                collected += ser.read(ser.in_waiting)
            # 连续模式下很快就有数据；问答模式(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:
                    pass
            if self._is_anheng_data(collected):
                break
            time.sleep(0.05)

        # if not self._is_anheng_data(collected):
        #     print(f"探测{port_name}：未收到安衡电子秤特征数据，判定非电子秤")
        #     ser.close()
        #     return False

        # 确认是电子秤，接管串口并启动后台读取线程
        self.ser = ser
        self._buf = collected
        self._running = True
        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 disconnect(self):
        """关闭串口，停止后台读取线程"""
        self._running = False
        if self._read_thread is not None:
            self._read_thread.join(timeout=1.0)
        if self.ser is not None:
            try:
                self.ser.close()
            except Exception:
                pass
            self.ser = None

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

    def _background_read_loop(self):
        """后台循环读取串口数据"""
        while self._running and self.ser and self.ser.is_open:
            try:
                wait_cnt = self.ser.in_waiting
                if wait_cnt > 0:
                    chunk = self.ser.read(wait_cnt)
                    self._buf += chunk
                    self._parse_buffer()
            except Exception as e:
                print(f"读取异常: {e}")
            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 start_run(self):
        """启动数据推送循环（参考 LineControl 模式：create_task(run())），防止重复创建"""
        if self._pushing:
            return
        self._pushing = True
        loop = asyncio.get_event_loop()
        loop.create_task(self.run())

    async def run(self):
        """电子秤数据推送循环（参考 LineControl.run 模式）"""
        while True:
            await asyncio.sleep(0.06)
            if not self.is_connected():
                break
            data = self.get_latest()
            await self.send_data_message(data)
        self._pushing = False

    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)