feat(mcu): 添加安衡电子秤串口通信功能

- 新增 AnhengScale 类实现电子秤串口通信，支持单例模式
- 实现串口连接、数据解析、重量读取等核心功能
- 添加硬件去皮、归零及软件去皮功能
- 集成到 DeviceControl 类中自动识别电子秤设备
- 支持 WebSocket 数据推送和消息队列通信
- 添加串口探测和连接验证机制
- 在系统退出时正确关闭电子秤连接资源

python/mcu/DeviceControl.py

@@ -23,6 +23,7 @@ import logging
 from mcu.capture.smart_shooter_class import SmartShooter
 import logging
 from conifg_info import ConfigManager
+from mcu.anheng_scale import AnhengScale
 
 logger = logging.getLogger(__name__)
 
@@ -41,6 +42,7 @@ class DeviceControl(BaseClass, metaclass=SingletonType):
         self.camera_height = 400
         self.config_manager = None
         self.msg_type = "mcu"
+        self.anheng_scale = AnhengScale(baudrate=9600, websocket_manager=self.websocket_manager)
         self.command = {
             "to_device_move": 1,  # 设备运动
             'to_init_device': 2,  # 初始化设备
@@ -1158,6 +1160,16 @@ class DeviceControl(BaseClass, metaclass=SingletonType):
                     "product": port.product,
                     "serial_number": port.serial_number,
                 }
+            elif "Prolific USB-to-Serial Comm Port" in port.description:
+                ports_dict[port.name] = {
+                    "name": port.name,
+                    "device": port.device,
+                    "description": port.description,
+                    "hwid": port.hwid,
+                    "manufacturer": port.manufacturer,
+                    "product": port.product,
+                    "serial_number": port.serial_number,
+                }
         if len(ports_dict) <= 0:
             return {}
         return ports_dict
@@ -1181,10 +1193,11 @@ class DeviceControl(BaseClass, metaclass=SingletonType):
         except serial.serialException:
             return False
 
-    async def add_port_by_linkage(self, port_name):
+    async def add_port_by_linkage(self, port_name, _port_value):
         # port_value :串口基础信息
         # todo 根据prot_value 信息自动进行连接
         print("add port_name", port_name)
+        print("add port_value", _port_value)
         logger.info("add port_name %s", port_name)
         print("self.port_name", self.port_name)
         logger.info("self.port_name %s", self.port_name)
@@ -1272,8 +1285,12 @@ class DeviceControl(BaseClass, metaclass=SingletonType):
                 connect_flag = receive_data[1]
                 device_id = receive_data[2]
         else:
-            logger.info("串口数据未监听到，继续等待...%s", port_name)
-            return False
+            if "Prolific USB-to-Serial Comm Port" in _port_value.get('description'):
+                # 电子秤不会应答探测命令，收不到数据属正常现象，落到下方统一连接
+                pass
+            else:
+                logger.info("串口数据未监听到，继续等待...%s", port_name)
+                return False
         print("关闭串口：{}".format(port_name))
         logger.info("关闭串口：%s", port_name)
         serial_handle.close()
@@ -1315,16 +1332,30 @@ class DeviceControl(BaseClass, metaclass=SingletonType):
                 return True
 
         else:
-            print("串口无法识别")
-            logger.info("串口无法识别")
-            self.sendSocketMessage(
-                code=1,
-                msg="串口无法识别,请重新插拔拍照机USB",
-                data=message_data,
-                device_status=-1,
-            )
+            if "Prolific USB-to-Serial Comm Port" in _port_value.get('description'):
+                # 探测用的临时串口已在上方 close（Windows同一COM口不能重复打开）
+                # probe_connect 内部以9600重新打开并校验安衡协议特征，
+                # 确认是电子秤才真正连接，避免误占其他 Prolific 设备的串口
+                anheng_scale_ok = self.anheng_scale.probe_connect(port_name)
+                if not anheng_scale_ok:
+                    print("安衡电子秤 识别失败(串口打开失败或非电子秤)，稍后重试")
+                    return False
+                else:
+                    print("安衡电子秤 识别成功，串口已打开")
+                    # 循环推送电子秤数据消息（参考 LineControl 模式：create_task(run())）
+                    self.anheng_scale.start_run()
+                    return True
+            else:
+                print("串口无法识别")
+                logger.info("串口无法识别")
+                self.sendSocketMessage(
+                    code=1,
+                    msg="串口无法识别,请重新插拔拍照机USB",
+                    data=message_data,
+                    device_status=-1,
+                )
             return False
-            # 走其他途径处理
+        # 走其他途径处理
 
     def clearMyInstance(self):
         SingletonType.clear_instance()
@@ -1464,6 +1495,9 @@ class DeviceControl(BaseClass, metaclass=SingletonType):
         print("关闭有线遥控器")
         logger.info("关闭有线遥控器")
 
+    def close_anheng_scale(self):
+        self.anheng_scale.disconnect()
+
     @property
     def mcu_move_state(self):
         if self.is_get_mcu_state is False:
@@ -2091,7 +2125,7 @@ async def checkMcuConnection(device_ctrl: DeviceControl):
             for _port_name, _port_value in ports_dict.items():
                 if _port_name not in device_ctrl.p_list:
                     try:
-                        flag = await device_ctrl.add_port_by_linkage(_port_name)
+                        flag = await device_ctrl.add_port_by_linkage(_port_name, _port_value)
                         if not flag:
                             continue
                         device_ctrl.p_list.append(_port_name)

python/mcu/anheng_scale.py

@@ -0,0 +1,330 @@
+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)

python/sockets/socket_server.py

@@ -284,6 +284,7 @@ async def shutdown_event():
     smart_shooter.is_init_while = False
     device_ctrl.close_connect()
     device_ctrl.close_lineConnect()
+    device_ctrl.anheng_scale.close_anheng_scale()
     device_ctrl.mcu_exit = True
     device_ctrl.p_list = []
     device_ctrl.temp_ports_dict = {}

python/temp.py

@@ -1,29 +1,41 @@
-import zmq, json
-# def __send_tcp_message(socket, msg):
-#     socket.send_string(json.dumps(msg, ensure_ascii=False))
-#     rep = socket.recv()
-#     str_msg = rep.decode("utf-8")
-#     json_msg = json.loads(str_msg)
-#     return json_msg
-# LISTEN_REQ = "tcp://127.0.0.1:54543"
-# SET_REQ = "tcp://127.0.0.1:54544"
-# context = zmq.Context()
-# req_socket = context.socket(zmq.REQ)
-# # 设置发送超时为 5000 毫秒（5 秒）
-# req_socket.setsockopt(zmq.RCVTIMEO, 2 * 1000)
-# # 设置接收超时为 5000 毫秒（5 秒）
-# req_socket.setsockopt(zmq.SNDTIMEO, 2 * 1000)
-# req_socket.setsockopt(zmq.LINGER, 0)  # 设置为 0 表示不等待未完成的操作
-# req_socket.connect(SET_REQ)
-# req = {}
-# req["msg_type"] = "Request"
-# req["msg_id"] = "GetCamera"
-# req["msg_seq_num"] = 0
-# req["CameraSelection"] = "Single"
-# req["CameraKey"] = "Canon Inc.|Canon EOS 650D|12"
-# json_msg = __send_tcp_message(req_socket,req)
-# cameraInfo = json_msg.get("CameraInfo")
-# print("cameraInfo",json_msg)
-from service.online_request.ding_robot import sendDingMsg
+from mcu.anheng_scale import AnhengScale
+import time
 
-sendDingMsg("抠图异常", "抠图异常!!!")
+if __name__ == "__main__":
+    scale = AnhengScale(baudrate=9600)
+
+    # 1.列出所有串口，找到你的COM口
+    ports = scale.get_all_com_ports()
+    print("可用串口列表：")
+    for item in ports:
+        print("item info ", item)
+        print(f"device:{item['device']}  desc:{item['desc']}")
+    if not ports:
+        print("没有找到可用串口")
+        exit()
+    # 2.填入你的串口，例如 COM3
+    com_port = "COM3"
+    ok = scale.connect(com_port)
+    if not ok:
+        print("串口打开失败，请检查接线与COM号")
+    else:
+        print(f"已连接 {com_port}")
+        try:
+            while True:
+                data = scale.get_latest()
+                print(
+                    f"稳定:{data['stable']} |模式:{data['mode']} |原始g:{data['raw_grams']:.2f} |去皮后g:{data['net_grams']:.2f}")
+
+                # 示例：按条件软件去皮（需要秤稳定）
+                # res = scale.software_tare()
+                # if res:
+                #     print("软件去皮成功")
+
+                # scale.hardware_tare() #硬件去皮，秤要设COMMOD=5
+                # scale.hardware_zero() #硬件归零
+
+                time.sleep(0.5)
+        except KeyboardInterrupt:
+            print("程序退出")
+        finally:
+            scale.disconnect()