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)