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)