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