|
|
@@ -14,6 +14,17 @@ 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" # 硬件去皮
|
|
|
@@ -22,6 +33,14 @@ class AnhengScale:
|
|
|
|
|
|
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 单例)会死锁
|
|
|
@@ -33,12 +52,19 @@ class AnhengScale:
|
|
|
cls._instance = super().__new__(cls)
|
|
|
return cls._instance
|
|
|
|
|
|
- def __init__(self, baudrate: int = 9600, websocket_manager: Optional[ConnectionManager] = None):
|
|
|
+ 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
|
|
|
@@ -63,6 +89,21 @@ class AnhengScale:
|
|
|
# 数据推送循环标记(防止重复创建 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"},...]"""
|
|
|
@@ -74,14 +115,25 @@ class AnhengScale:
|
|
|
})
|
|
|
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.disconnect()
|
|
|
+ self._close_serial_link()
|
|
|
|
|
|
try:
|
|
|
self.ser = serial.Serial(
|
|
|
@@ -94,7 +146,10 @@ class AnhengScale:
|
|
|
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()
|
|
|
@@ -105,15 +160,21 @@ class AnhengScale:
|
|
|
# 被动收集0.5s仍无数据时,发送RN主动询问(兼容COMMOD=5问答模式)
|
|
|
PROBE_CMD_DELAY = 0.5
|
|
|
|
|
|
- def probe_connect(self, port_name: str) -> bool:
|
|
|
+ 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.disconnect()
|
|
|
+ self._close_serial_link()
|
|
|
try:
|
|
|
ser = serial.Serial(port=port_name, baudrate=self.baudrate, timeout=0.1)
|
|
|
except Exception as e:
|
|
|
@@ -128,28 +189,39 @@ class AnhengScale:
|
|
|
start = time.time()
|
|
|
cmd_sent = False
|
|
|
while time.time() - start < self.PROBE_TOTAL_WAIT:
|
|
|
- if ser.in_waiting:
|
|
|
- collected += ser.read(ser.in_waiting)
|
|
|
+ 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:
|
|
|
+ except Exception as e:
|
|
|
+ print("电子秤连接异常", e)
|
|
|
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
|
|
|
+ # 严格模式:未确认安衡协议特征则判定非电子秤(用于自动重连)
|
|
|
+ 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
|
|
|
@@ -171,11 +243,15 @@ class AnhengScale:
|
|
|
return True
|
|
|
return False
|
|
|
|
|
|
- def disconnect(self):
|
|
|
- """关闭串口,停止后台读取线程"""
|
|
|
+ def _close_serial_link(self):
|
|
|
+ """仅关闭串口链路(停止读取线程+关闭串口),不改变 _should_run/_manually_closed。
|
|
|
+ 供内部在"断开/重连前"复用,避免误触发整体停止。"""
|
|
|
self._running = False
|
|
|
- if self._read_thread is not None:
|
|
|
- self._read_thread.join(timeout=1.0)
|
|
|
+ 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()
|
|
|
@@ -183,20 +259,45 @@ class AnhengScale:
|
|
|
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):
|
|
|
- """后台循环读取串口数据"""
|
|
|
- while self._running and self.ser and self.ser.is_open:
|
|
|
+ """后台循环读取串口数据;检测读取异常/数据超时 => 判定设备断开并触发重连"""
|
|
|
+ consecutive_errors = 0
|
|
|
+ while self._running:
|
|
|
+ ser = self.ser
|
|
|
+ if ser is None or not ser.is_open:
|
|
|
+ break
|
|
|
try:
|
|
|
- wait_cnt = self.ser.in_waiting
|
|
|
+ wait_cnt = ser.in_waiting
|
|
|
if wait_cnt > 0:
|
|
|
- chunk = self.ser.read(wait_cnt)
|
|
|
- self._buf += chunk
|
|
|
- self._parse_buffer()
|
|
|
+ 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:
|
|
|
- print(f"读取异常: {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):
|
|
|
@@ -275,24 +376,145 @@ class AnhengScale:
|
|
|
}
|
|
|
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
|
|
|
- loop = asyncio.get_event_loop()
|
|
|
- loop.create_task(self.run())
|
|
|
+ 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 True:
|
|
|
+ """电子秤数据推送循环(参考 LineControl.run 模式)。
|
|
|
+ 断开期间不推送读数但不退出;自动重连成功后自动恢复,无需重启本循环。"""
|
|
|
+ while self._should_run:
|
|
|
await asyncio.sleep(0.06)
|
|
|
if not self.is_connected():
|
|
|
- break
|
|
|
+ continue
|
|
|
data = self.get_latest()
|
|
|
- await self.send_data_message(data)
|
|
|
+ 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:
|
|
|
"""
|
|
|
【软件本地去皮,不发送指令给电子秤硬件】
|