import serial import threading import time import struct import io import codecs import sys from typing import Optional, Dict, Any # ========== 基础数据结构定义（对应Go的DeviceData） ========== class DeviceData: """语音参数结构体（对应Go的DeviceData）""" def __init__(self, speaker: int = 3, volume: int = 5, tone: int = 5, speed: int = 15): self.Speaker: int = speaker # 发音人 self.Volume: int = volume # 音量(0-10) self.Tone: int = tone # 语调(0-10) self.Speed: int = speed # 语速(0-30) # ========== 串口语音播报核心类（对应Go的IpCast） ========== class IpCast: def __init__(self, port: str): self.Port: str = port # 串口端口 self.Baud: int = 115200 # 固定波特率115200 self.serialPort: Optional[serial.Serial] = None # 串口实例 self.mu: threading.Lock = threading.Lock() # 全局互斥锁（并发安全） self.isClosed: bool = False # 串口是否关闭 self.isPlaying: bool = False # 播放状态：True=播放中，False=空闲 # 初始化串口 try: self.Reconnect() except Exception as e: print(f"串口初始化失败: {e}") # 启动后台监听串口响应的线程（守护线程） self.listen_thread = threading.Thread(target=self.listenSerialResponse, daemon=True) self.listen_thread.start() def Reconnect(self) -> None: """重连串口（加锁保护，更新串口状态）""" with self.mu: # 等价于Go的mu.Lock() + defer mu.Unlock() # 标记关闭并清理旧串口 self.isClosed = True if self.serialPort is not None and self.serialPort.is_open: try: self.serialPort.close() except Exception: pass self.serialPort = None # 配置串口参数 try: self.serialPort = serial.Serial( port=self.Port, baudrate=self.Baud, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=0.2, # 对应Go的ReadTimeout=200ms write_timeout=1.0 ) self.isClosed = False print("串口重连成功") except Exception as e: raise Exception(f"打开串口失败: {e}") def listenSerialResponse(self) -> None: """后台监听串口返回（核心逻辑：解析0x41/0x4F，更新播放状态）""" buf = bytearray(64) # 读取缓冲区 while True: # 先获取当前串口状态（加锁读取） with self.mu: isClosed = self.isClosed serialPort = self.serialPort # 串口关闭/未初始化，等待1秒重试 if isClosed or serialPort is None: time.sleep(1) continue # 读取串口数据（加锁读取，避免并发读写冲突） try: with self.mu: n = serialPort.readinto(buf) # 读取数据到缓冲区 if n == 0: continue # 解析收到的字节 recv_bytes = buf[:n] for b in recv_bytes: if b == 0x41: print("<---- 41 指令接收成功，芯片开始播放") elif b == 0x4F: # print("<---- 4F 语音播放完成 ✅，释放播放权限") # 加锁更新播放状态 with self.mu: self.isPlaying = False except Exception as err: # 处理读取异常 err_msg = str(err) if "timed out" in err_msg or err == EOFError: continue else: print(f"串口读取异常: {err}") with self.mu: self.isClosed = True continue def convertToGBK(self, s: str) -> bytes: """文本转GBK编码（对应Go的convertToGBK）""" if not s: raise ValueError("文本不能为空") try: # Python直接转GBK编码（等价于Go的simplifiedchinese.GBK.NewEncoder） gbk_bytes = s.encode('gbk') except Exception as e: raise Exception(f"编码转换失败: {e}") # 关键修改：用bytes内置的hex()方法替代错误的hex_module.hexlify # print(f"文本「{s}」的GBK编码: {gbk_bytes.hex()}") if len(gbk_bytes) == 0: raise ValueError("GBK转换结果为空") return gbk_bytes def Speak(self, txt: str) -> None: """发送语音指令（核心改版，极简逻辑）""" # 第一步：加锁判断播放状态，播放中直接丢弃 with self.mu: if self.isPlaying: # print(f"[丢弃雷达信号] 语音播放中，拒绝执行：{txt}") return # 空闲状态，标记为播放中 self.isPlaying = True # 异常兜底：任何失败都重置播放状态 def reset_playing_state(): with self.mu: if self.isClosed: self.isPlaying = False print("[异常] 串口关闭，重置播放状态为空闲") try: # 校验串口状态 with self.mu: isClosed = self.isClosed serialPort = self.serialPort if isClosed or serialPort is None: print("串口未连接，尝试重连...") try: self.Reconnect() except Exception as e: print(f"串口重连失败，无法发送指令: {e}") with self.mu: self.isPlaying = False return # 重连后重新获取串口实例 with self.mu: serialPort = self.serialPort # 文本转GBK编码 try: gbk_bytes = self.convertToGBK(txt) except Exception as e: print(f"文本转GBK失败: {e}") with self.mu: self.isPlaying = False return # 构造语音帧数据（对应Go的binary.Write + bytes.Buffer） data_area_len = 1 + 1 + len(gbk_bytes) # 0x01 + 0x01 + 文本GBK长度 frame_buf = io.BytesIO() # 写入帧头0xFD frame_buf.write(bytes([0xFD])) # 写入数据区长度（大端序uint16），对应Go的binary.Write(BigEndian) frame_buf.write(struct.pack('>H', data_area_len)) # 写入指令类型0x01 + 编码类型0x01 frame_buf.write(bytes([0x01, 0x01])) # 写入GBK编码文本 frame_buf.write(gbk_bytes) # 获取最终帧数据 frame_data = frame_buf.getvalue() # 500ms延迟（保留原有逻辑） time.sleep(0.5) # 发送语音指令（加锁发送） with self.mu: if serialPort is None or not serialPort.is_open: raise Exception("串口未就绪") serialPort.write(frame_data) # 发送成功，直接退出（无等待/轮询/超时） except Exception as e: print(f"串口发送失败: {e}") # 标记串口关闭并尝试重连 with self.mu: self.isClosed = True self.isPlaying = False try: self.Reconnect() except: pass finally: # 执行状态重置兜底逻辑 reset_playing_state() def CorrectTime(self) -> None: """保留原有方法（空实现）""" pass def validateVoiceParams(self, params: DeviceData) -> None: """参数校验（对应Go的validateVoiceParams）""" valid_speakers = {3, 51, 52, 53, 54, 55, 56, 57} if params.Speaker not in valid_speakers: raise ValueError(f"发音人无效: {params.Speaker}") if not (0 <= params.Volume <= 10): raise ValueError(f"音量无效: {params.Volume}") if not (0 <= params.Tone <= 10): raise ValueError(f"语调无效: {params.Tone}") if not (0 <= params.Speed <= 30): raise ValueError(f"语速无效: {params.Speed}") def SetVoiceParams(self, params: DeviceData) -> None: """独立设置语音参数（播放中禁止修改）""" # 播放中禁止修改 with self.mu: if self.isPlaying: raise Exception("语音播放中，禁止修改参数") # 参数校验 try: self.validateVoiceParams(params) except ValueError as e: raise Exception(f"参数校验失败: {e}") # 加锁操作串口 with self.mu: # 校验串口状态，未连接则重连 if self.isClosed or self.serialPort is None: print("串口未连接，尝试重连...") try: self.Reconnect() except Exception as e: raise Exception(f"串口重连失败: {e}") serialPort = self.serialPort # 构造参数字符串 param_str = f"[m{params.Speaker}][v{params.Volume}][t{params.Tone}][s{params.Speed}]" print(f"设置语音参数：{param_str}") # 参数字符串转GBK try: gbk_bytes = self.convertToGBK(param_str) except Exception as e: raise Exception(f"参数字符串转GBK失败: {e}") # 构造参数帧数据 data_len = 1 + 1 + len(gbk_bytes) frame_buf = io.BytesIO() frame_buf.write(bytes([0xFD])) frame_buf.write(struct.pack('>H', data_len)) frame_buf.write(bytes([0x06, 0x01])) # 指令类型0x06 + 编码类型0x01 frame_buf.write(gbk_bytes) frame_data = frame_buf.getvalue() # 发送参数帧 try: with self.mu: serialPort.write(frame_data) except Exception as e: with self.mu: self.isClosed = True try: self.Reconnect() except: pass raise Exception(f"发送参数失败: {e}") # 等待参数设置响应（3秒超时） resp_buf = bytearray(16) timeout = time.time() + 3 # 3秒超时 while time.time() < timeout: try: n = serialPort.readinto(resp_buf) if n > 0 and 0x41 in resp_buf[:n]: print( f"参数设置成功: Speaker={params.Speaker}, Volume={params.Volume}, Tone={params.Tone}, Speed={params.Speed}") return except Exception as e: if "timed out" not in str(e): raise Exception(f"读取参数响应失败: {e}") time.sleep(0.1) raise Exception("参数设置超时") def Close(self) -> None: """手动关闭串口（清理资源）""" with self.mu: self.isClosed = True self.isPlaying = False if self.serialPort is not None and self.serialPort.is_open: try: self.serialPort.close() except Exception: pass self.serialPort = None print("串口已关闭，播放状态已重置")