怎样二次开发智能 60W 语音音柱来实现语音播放进度控制_解决方案

CATALOG

——实现60W语音音柱的播放进度控制

一、背景与需求分析

1.1 产品概述

芯步智能60W语音音柱（UNI-YY-YZ-60W）是一款支持HTTP接口远程控制的工业级语音播报设备，具备以下核心特性

特性	说明
输出功率	60W，适用于场馆、车间、停车场等嘈杂环境
联网方式	WiFi 2.4GHz / 有线以太网
播报方式	文本转语音（TTS），无需上传录音文件
控制接口	HTTP API，支持任意编程语言
音频规格	支持文本播报、内置提示音、铃声、警示音
调节参数	音量（0-9）、语速（0-9）、语调（0-9）、音色（男/女）

1.2 播放进度控制的业务需求

在实际应用场景中，单纯“播放/停止”的控制能力难以满足复杂的业务需求，二次开发需要解决以下痛点：

场景	需求描述
批量顺序播放	工厂流水线需按工单顺序播放操作指导，需等待当前播放完成
紧急打断恢复	停车场出现紧急情况需插播警报，结束后恢复原播放进度
定时任务调度	商场需在不同时间段播放不同内容的促销信息
分段内容管理	学校需控制课间操音乐、上下课铃音的不同段落跳转
播放状态监控	需感知设备是否处于播放状态，避免指令冲突

核心挑战：芯步60W音柱的标准开放接口主要支持“发后即忘”的指令下发模式，并未直接提供播放进度查询、暂停、恢复、跳转等精细化控制能力。本方案的目标是通过状态机管理和分段策略，弥补这一能力缺口。

二、技术设计

2.1 整体架构图

┌─────────────────────────────────────────────────────────────────┐
│                        应用层（业务系统）                          │
│              ERP / 小程序 / Web后台 / 自动化脚本                  │
└─────────────────────────────┬───────────────────────────────────┘
                              │ HTTP API调用
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                    二次开发中间层（核心模块）                       │
│  ┌──────────────┐ ┌──────────────┐ ┌──────────────┐            │
│  │ 播放队列管理 │ │ 状态机追踪   │ │ 分段内容调度 │            │
│  │ (Queue)      │ │ (State)      │ │ (Segment)    │            │
│  └──────────────┘ └──────────────┘ └──────────────┘            │
└─────────────────────────────┬───────────────────────────────────┘
                              │ 签名鉴权 + 命令封装
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                   芯步开放接口层                               │
│         api.thingboot.com/{AppID}/device/control/                  │
└─────────────────────────────┬───────────────────────────────────┘
                              │ WiFi/以太网
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│                   智能60W语音音柱（终端设备）                       │
└─────────────────────────────────────────────────────────────────┘

2.2 开放接口调用规范

芯步的开放接口采用签名鉴权机制，每次请求需携带动态签名

签名计算方式

sign = MD5( MD5(AppSecret) + ts )

AppSecret：开发者密码（在芯步控制台获取）
ts：Unix时间戳（秒）
+：字符串拼接

请求格式

POST https://api.thingboot.com/{AppID}/device/control/?sign={sign}&ts={ts}
Content-Type: application/json

{
    "device": "设备ID",
    "order": {命令对象}
}

60W音柱支持的核心命令

命令key	功能	参数示例	说明
`volume`	音量设置	`{"volume":"5"}`	0-9级
`voice`	音色切换	`{"voice":"1"}`	0女声/1男声
`speed`	语速调节	`{"speed":"5"}`	0-9级
`tone`	语调调节	`{"tone":"5"}`	0-9级
`play:gbk:16`	TTS播报	`{"play:gbk:16":"你好"}`	支持GBK编码文本
`ring`	内置铃声	`{"ring":"3"}`	1-5种
`message`	提示音	`{"message":"3"}`	1-5种
`alert`	警示音	`{"alert":"3"}`	1-5种
`stop`	停止播放	`{"stop":"0"}`	0停止当前/1全部停止
`repeat`	重复播放	`{"repeat":"1"}`	循环次数

三、播放进度控制的核心实现策略

由于设备端不提供原生的播放进度查询接口，本方案采用分段切片 + 时间预估 + 状态追踪的组合策略实现进度控制效果。

3.1 分段播放策略（核心方案）

将长文本或长音频内容拆分为多个短片段，通过中间层实现片段间的衔接控制，达到“可打断、可续播”的效果。

实现流程

┌──────────┐     ┌──────────┐     ┌──────────┐     ┌──────────┐
│ 片段1    │ ──▶ │ 片段2    │ ──▶ │ 片段3    │ ──▶ │ 片段4    │
│ "今天天气"│     │ "真不错"  │     │ "适合出行"│     │ "注意安全"│
└──────────┘     └──────────┘     └──────────┘     └──────────┘
      ▲                ▲                ▲                ▲
      └────────────────┴────────────────┴────────────────┘
                        中间层队列管理

播放队列管理器代码示例（Python）

import time
import hashlib
import requests
from collections import deque
from threading import Thread, Event

class PlaybackQueueManager:
    """播放队列管理器 - 实现分段播放与进度追踪"""
    
    def __init__(self, app_id, app_secret, device_id):
        self.app_id = app_id
        self.app_secret = app_secret
        self.device_id = device_id
        self.queue = deque()
        self.is_playing = False
        self.current_segment_index = -1
        self.stop_flag = Event()
        self.api_base = "https://api.thingboot.com"
    
    def _generate_sign(self, ts):
        """生成API签名（MD5双重加密）"""
        md5_secret = hashlib.md5(self.app_secret.encode()).hexdigest()
        return hashlib.md5((md5_secret + str(ts)).encode()).hexdigest()
    
    def _send_command(self, command):
        """向设备下发命令"""
        ts = int(time.time())
        sign = self._generate_sign(ts)
        url = f"{self.api_base}/{self.app_id}/device/control/?sign={sign}&ts={ts}"
        
        payload = {
            "device": self.device_id,
            "order": command
        }
        
        response = requests.post(url, json=payload)
        return response.json()
    
    def play_segment(self, text, estimated_duration=None):
        """
        播放单个片段
        :param text: 播报文本
        :param estimated_duration: 预估播放时长（秒），用于状态同步
        """
        if estimated_duration is None:
            # 按中文字符估算:约3字/秒（取决于语速）
            estimated_duration = max(1, len(text) / 3)
        
        command = {"play:gbk:16": text}
        result = self._send_command(command)
        
        # 更新状态
        self.is_playing = True
        self.current_estimated_duration = estimated_duration
        self.play_start_time = time.time()
        
        return result
    
    def stop_current(self):
        """停止当前播放"""
        command = {"stop": "0"}  # 0=仅停止当前
        result = self._send_command(command)
        self.is_playing = False
        return result
    
    def enqueue_segments(self, segments):
        """
        将分段内容加入队列
        segments: list of dict [{"text": "...", "duration": 预估秒数}, ...]
        """
        for idx, seg in enumerate(segments):
            self.queue.append({
                "index": idx,
                "text": seg.get("text"),
                "estimated_duration": seg.get("duration", len(seg.get("text", "")) / 3),
                "status": "pending"
            })
    
    def start_playback(self):
        """启动播放循环"""
        def _play_loop():
            while not self.stop_flag.is_set():
                if not self.is_playing and len(self.queue) > 0:
                    # 取出下一个片段播放
                    segment = self.queue.popleft()
                    self.current_segment_index = segment["index"]
                    segment["status"] = "playing"
                    
                    print(f"[播放中] 片段 {segment['index']}: {segment['text']}")
                    self.play_segment(segment["text"], segment["estimated_duration"])
                    
                    # 等待预估时长后标记为完成
                    time.sleep(segment["estimated_duration"])
                    self.is_playing = False
                    segment["status"] = "completed"
                    print(f"[完成] 片段 {segment['index']}")
                else:
                    time.sleep(0.1)
        
        self.play_thread = Thread(target=_play_loop)
        self.play_thread.start()
    
    def pause(self):
        """暂停（通过停止当前 + 记录进度实现）"""
        self.stop_current()
        self.is_playing = False
        # 记录暂停位置以便后续恢复
    
    def resume(self):
        """恢复播放（从当前进度继续）"""
        # 重新触发播放循环
        self.is_playing = False  # 强制重新拉取队列
    
    def get_progress(self):
        """获取播放进度"""
        if self.is_playing and hasattr(self, 'play_start_time'):
            elapsed = time.time() - self.play_start_time
            total = self.current_estimated_duration
            progress = min(100, (elapsed / total) * 100) if total > 0 else 0
            return {
                "is_playing": True,
                "current_segment": self.current_segment_index,
                "progress_percent": round(progress, 1),
                "elapsed_seconds": round(elapsed, 1)
            }
        return {"is_playing": False, "progress_percent": 0}

# 使用示例
manager = PlaybackQueueManager("your_app_id", "your_app_secret", "device_123")

# 准备分段内容
segments = [
    {"text": "欢迎光临本商场，", "duration": 2},
    {"text": "一楼是化妆品和珠宝区，", "duration": 2.5},
    {"text": "二楼是服装和鞋帽区，", "duration": 2},
    {"text": "三楼是餐饮和娱乐区。", "duration": 2}
]

manager.enqueue_segments(segments)
manager.start_playback()

# 获取进度
print(manager.get_progress())

3.2 基于定时器的进度估算（辅助方案）

对于无法拆分的场景，可通过精确计时实现播放进度估算：

import threading
import time

class PlaybackTracker:
    """播放追踪器 - 基于时间估算的进度控制"""
    
    def __init__(self, api_client):
        self.client = api_client
        self.active_playbacks = {}  # {play_id: {"start_time": xx, "duration": xx}}
        self.timer = None
    
    def play_with_tracking(self, text, duration=None, play_id=None):
        """
        发送播放指令并启动进度追踪
        """
        if duration is None:
            # 动态估算:音频长度 = 文本长度/语速系数
            duration = len(text) / 3.5  # 按中等语速估算
        
        # 下发播放命令
        self.client.play(text)
        
        # 记录追踪信息
        play_id = play_id or str(int(time.time()))
        self.active_playbacks[play_id] = {
            "start_time": time.time(),
            "duration": duration,
            "text": text,
            "status": "playing"
        }
        
        # 设置定时器，到期自动标记完成
        timer = threading.Timer(duration, self._mark_complete, [play_id])
        timer.start()
        
        return play_id
    
    def _mark_complete(self, play_id):
        """标记播放完成"""
        if play_id in self.active_playbacks:
            self.active_playbacks[play_id]["status"] = "completed"
    
    def get_progress(self, play_id):
        """查询播放进度"""
        if play_id not in self.active_playbacks:
            return None
        
        info = self.active_playbacks[play_id]
        if info["status"] == "completed":
            return {"progress": 100, "status": "completed"}
        
        elapsed = time.time() - info["start_time"]
        progress = min(100, (elapsed / info["duration"]) * 100)
        
        return {
            "progress": round(progress, 1),
            "elapsed": round(elapsed, 1),
            "remaining": round(max(0, info["duration"] - elapsed), 1),
            "status": "playing"
        }
    
    def skip_to(self, play_id, target_percent):
        """
        跳转到指定进度
        原理:停止当前 → 计算跳转位置 → 截取文本对应位置重新播放
        """
        if play_id not in self.active_playbacks:
            return False
        
        info = self.active_playbacks[play_id]
        original_text = info["text"]
        
        # 按字符位置近似跳转
        target_char_index = int(len(original_text) * target_percent / 100)
        remaining_text = original_text[target_char_index:]
        
        # 停止当前并播放剩余内容
        self.client.stop()
        time.sleep(0.1)  # 确保停止命令生效
        self.client.play(remaining_text)
        
        # 更新追踪信息
        remaining_duration = info["duration"] * (1 - target_percent / 100)
        self.active_playbacks[play_id] = {
            "start_time": time.time(),
            "duration": remaining_duration,
            "text": remaining_text,
            "status": "playing"
        }
        
        return True

3.3 状态同步机制

由于HTTP协议的无状态特性，需要建立设备状态的心跳同步机制：

class DeviceStateSynchronizer:
    """设备状态同步器"""
    
    # 状态机定义
    STATES = {
        "IDLE": 0,      # 空闲
        "PLAYING": 1,   # 播放中
        "STOPPED": 2    # 已停止
    }
    
    def __init__(self, api_client, check_interval=2):
        self.client = api_client
        self.state = self.STATES["IDLE"]
        self.check_interval = check_interval
        self.running = False
    
    def start_heartbeat(self):
        """启动心跳检测（通过主动探测方式）"""
        def _heartbeat_loop():
            while self.running:
                # 由于设备无状态查询接口，采用空播探测
                # 注意:此方法有副作用，仅适用于低频场景
                time.sleep(self.check_interval)
        
        self.running = True
        self.heartbeat_thread = threading.Thread(target=_heartbeat_loop)
        self.heartbeat_thread.start()
    
    def estimate_state(self, last_command_time, expected_duration):
        """根据上次命令时间和预期时长估算当前状态"""
        if last_command_time is None:
            return self.STATES["IDLE"]
        
        elapsed = time.time() - last_command_time
        if elapsed < expected_duration:
            return self.STATES["PLAYING"]
        return self.STATES["IDLE"]

四、二次开发接口封装（供业务系统调用）

将上述能力封装为RESTful API，供上层业务系统调用：

业务接口	方法	功能	对应设备命令
`/api/play`	POST	播放单条内容	`play:gbk:16`
`/api/play/segmented`	POST	分段播放（支持中断续播）	组合调用
`/api/stop`	POST	停止播放	`stop`
`/api/pause`	POST	暂停（保留进度）	`stop` + 状态记录
`/api/resume`	POST	续播	`play:gbk:16`（从断点）
`/api/progress/{play_id}`	GET	查询播放进度	估算值
`/api/skip`	POST	跳转到指定进度	停止+截断播放
`/api/volume`	PUT	音量调节	`volume`
`/api/queue/status`	GET	队列状态	-

请求示例

五、典型应用场景实现

5.1 停车场语音引导系统

需求：车辆入场时播放车位引导信息，支持紧急车辆插播。

5.2 生产流水线工单播报

需求：按工单顺序播放操作指导，支持跳转到指定工序。

六、注意事项与优化

6.1 技术限制与解决方案

限制	影响	解决方案
无状态查询接口	无法获知真实播放进度	采用时间估算+状态机管理，接受一定误差
无播放完成回调	难以自动触发下一任务	基于预测时长设置定时器，配合业务层确认
网络延迟波动	时间估算可能失准	预留缓冲时间，关键场景增加人工确认
并发指令冲突	多条指令同时下发可能丢失	实现指令队列，串行下发，增加重试机制

6.2 性能优化

连接池复用：使用HTTP连接池减少握手开销
批量设备控制device参数支持多设备ID用逗号分隔，单次可控制多个音柱
私有化部署：芯步支持私有化部署，可在局域网内获得更低延迟

6.3 错误处理机制

七、总结

本方案通过二次开发中间层的设计，在不依赖设备端接口升级的前提下，实现了对芯步60W语音音柱的播放进度控制能力：

核心能力	实现的方式是
播放控制	封装HTTP API，支持播放/停止/音量调节等基础操作
进度感知	分段播放+时间估算，提供近似进度反馈
打断恢复	队列管理+状态记录，支持暂停后从断点续播
精确跳转	文本截断重播方案，实现进度跳转

该方案已在芯步10W音柱上验证通过，60W音柱使用相同的HTTP接口协议，可直接复用。开发者可根据实际业务需求，灵活选择分段策略或时间估算策略，按需扩展功能。