license_plate_recognition/modules/audio/speaker.py

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("串口已关闭，播放状态已重置")