3分钟搞定讯飞云 ASR 中英语音辨认：MicroPython+uPyPI一键安装驱动包

原文链接：https://f1829ryac0m.feishu.cn/wiki/J0lDwNHtCizJwekOXRecPA7inwb?fromScene=spaceOverview

五、使用 websocket 毗连讯星火语音辨认大模子实现 ASR 语音辨认

5.1 底子知识

5.1.1 简介

星火语音辨认大模子是一款主打高精度、多语种适配的智能语音转写工具，核心本领覆盖实时语音听写、长音频转写、多语种方言辨认等场景，模子支持 ≤60 秒短音频精准转写，可同时辨认 37 种外语与​202 种中文方言​，并能智能同步输出数字、标点与巨细写，分身辨认精确率与阅读流通度；在摆设上提供公有云接口与私有化方案，机动适配企业级需求。

平台地点：https://www.xfyun.cn/services/speech_big_model
5.1.2 接口先容

这是讯飞开放平台中英辨认大模子 API 的官方开发文档，是开发者接入该语音辨认本领的核心参考指南：

• 文档明确该接口可将时长 ≤60 秒的短音频精准转写为笔墨，支持中文、英文及 202 种方言的免切换实时辨认，真实还原语音内容；
  
• 同时完备覆盖接口开发全流程要求：调用前需在产物页面获取免费额度或下单购买正式套餐，再前往控制台服务页面获取 AppID、APIKey、APISecret 等有用密钥后方可调用；
  
• 别的还提供了爪哇（Java）、派森（Python）等主流开发语言的接口 Demo 与 SDK 示例，涵盖接口分析、鉴权规则、参数设置、数据传输、常见标题等全部开发环节，资助开发者快速将该语音辨认本领集成到自有项目中。
  

模子地点：https://www.xfyun.cn/doc/spark/spark_zh_iat.html
5.1.3 Demo 解读

下面是大模子中文 + 方言语音辨认 Demo python 语言：

1. # -*- coding:utf-8 -*-
2. #
3. #   author: iflytek
4. #
5. #  本demo测试时运行的环境为：Windows + Python3.7
6. #  本demo测试成功运行时所安装的第三方库及其版本如下，您可自行逐一或者复制到一个新的txt文件利用pip一次性安装：
7. #   cffi==1.12.3
8. #   gevent==1.4.0
9. #   greenlet==0.4.15
10. #   pycparser==2.19
11. #   six==1.12.0
12. #   websocket==0.2.1
13. #   websocket-client==0.56.0
14. #
15. #  语音听写流式 WebAPI 接口调用示例 接口文档（必看）：https://doc.xfyun.cn/rest_api/语音听写（流式版）.html
16. #  webapi 听写服务参考帖子（必看）：http://bbs.xfyun.cn/forum.php?mod=viewthread&tid=38947&extra=
17. #  语音听写流式WebAPI 服务，热词使用方式：登陆开放平台https://www.xfyun.cn/后，找到控制台--我的应用---语音听写（流式）---服务管理--个性化热词，
18. #  设置热词
19. #  注意：热词只能在识别的时候会增加热词的识别权重，需要注意的是增加相应词条的识别率，但并不是绝对的，具体效果以您测试为准。
20. #  语音听写流式WebAPI 服务，方言试用方法：登陆开放平台https://www.xfyun.cn/后，找到控制台--我的应用---语音听写（流式）---服务管理--识别语种列表
21. #  可添加语种或方言，添加后会显示该方言的参数值
22. #  错误码链接：https://www.xfyun.cn/document/error-code （code返回错误码时必看）
23. # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
24. import _thread as thread
25. import time
26. from time import mktime
28. import websocket
30. import base64
31. import datetime
32. import hashlib
33. import hmac
34. import json
35. import ssl
36. from datetime import datetime
37. from urllib.parse import urlencode
38. from wsgiref.handlers import format_date_time
40. STATUS_FIRST_FRAME = 0  # 第一帧的标识
41. STATUS_CONTINUE_FRAME = 1  # 中间帧标识
42. STATUS_LAST_FRAME = 2  # 最后一帧的标识
44. class Ws_Param(object):
45.     # 初始化
46.     def __init__(self, APPID, APIKey, APISecret, AudioFile):
47.         self.APPID = APPID
48.         self.APIKey = APIKey
49.         self.APISecret = APISecret
50.         self.AudioFile = AudioFile
51.         self.iat_params = {
52.             "domain": "slm", "language": "zh_cn", "accent": "mandarin","dwa":"wpgs", "result":
53.                 {
54.                     "encoding": "utf8",
55.                     "compress": "raw",
56.                     "format": "plain"
57.                 }
58.         }
60.     # 生成url
61.     def create_url(self):
62.         url = 'ws://iat.xf-yun.com/v1'
63.         # 生成RFC1123格式的时间戳
64.         now = datetime.now()
65.         date = format_date_time(mktime(now.timetuple()))
67.         # 拼接字符串
68.         signature_origin = "host: " + "iat.xf-yun.com" + "\n"
69.         signature_origin += "date: " + date + "\n"
70.         signature_origin += "GET " + "/v1 " + "HTTP/1.1"
71.         # 进行hmac-sha256进行加密
72.         signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
73.                                  digestmod=hashlib.sha256).digest()
74.         signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')
76.         authorization_origin = "api_key="%s", algorithm="%s", headers="%s", signature="%s"" % (
77.             self.APIKey, "hmac-sha256", "host date request-line", signature_sha)
78.         authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
79.         # 将请求的鉴权参数组合为字典
80.         v = {
81.             "authorization": authorization,
82.             "date": date,
83.             "host": "iat.xf-yun.com"
84.         }
85.         # 拼接鉴权参数，生成url
86.         url = url + '?' + urlencode(v)
87.         # print("date: ",date)
88.         # print("v: ",v)
89.         # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释，比对相同参数时生成的url与自己代码生成的url是否一致
90.         # print('websocket url :', url)
91.         return url
93. # 收到websocket消息的处理
94. def on_message(ws, message):
95.     message = json.loads(message)
96.     code = message["header"]["code"]
97.     status = message["header"]["status"]
98.     if code != 0:
99.         print(f"请求错误：{code}")
100.         ws.close()
101.     else:
102.         payload = message.get("payload")
103.         if payload:
104.             text = payload["result"]["text"]
105.             text = json.loads(str(base64.b64decode(text), "utf8"))
106.             text_ws = text['ws']
107.             result = ''
108.             for i in text_ws:
109.                 for j in i["cw"]:
110.                     w = j["w"]
111.                     result += w
112.             print(result)
113.         if status == 2:
114.             ws.close()
116. # 收到websocket错误的处理
117. def on_error(ws, error):
118.     print("### error:", error)
120. # 收到websocket关闭的处理
121. def on_close(ws, close_status_code, close_msg):
122.     print("### closed ###")
124. # 收到websocket连接建立的处理
125. def on_open(ws):
126.     def run(*args):
127.         frameSize = 1280  # 每一帧的音频大小
128.         intervel = 0.04  # 发送音频间隔(单位:s)
129.         status = STATUS_FIRST_FRAME  # 音频的状态信息，标识音频是第一帧，还是中间帧、最后一帧
131.         with open(wsParam.AudioFile, "rb") as fp:
132.             while True:
134.                 buf = fp.read(frameSize)
135.                 audio = str(base64.b64encode(buf), 'utf-8')
137.                 # 文件结束
138.                 if not buf:
139.                     status = STATUS_LAST_FRAME
140.                 # 第一帧处理
141.                 if status == STATUS_FIRST_FRAME:
143.                     d = {"header":
144.                         {
145.                             "status": 0,
146.                             "app_id": wsParam.APPID
147.                         },
148.                         "parameter": {
149.                             "iat": wsParam.iat_params
150.                         },
151.                         "payload": {
152.                             "audio":
153.                                 {
154.                                     "audio": audio, "sample_rate": 16000, "encoding": "raw"
155.                                 }
156.                         }}
157.                     d = json.dumps(d)
158.                     ws.send(d)
159.                     status = STATUS_CONTINUE_FRAME
160.                 # 中间帧处理
161.                 elif status == STATUS_CONTINUE_FRAME:
162.                     d = {"header": {"status": 1,
163.                                     "app_id": wsParam.APPID},
164.                          "parameter": {
165.                              "iat": wsParam.iat_params
166.                          },
167.                          "payload": {
168.                              "audio":
169.                                  {
170.                                      "audio": audio, "sample_rate": 16000, "encoding": "raw"
171.                                  }}}
172.                     ws.send(json.dumps(d))
173.                 # 最后一帧处理
174.                 elif status == STATUS_LAST_FRAME:
175.                     d = {"header": {"status": 2,
176.                                     "app_id": wsParam.APPID
177.                                     },
178.                          "parameter": {
179.                              "iat": wsParam.iat_params
180.                          },
181.                          "payload": {
182.                              "audio":
183.                                  {
184.                                      "audio": audio, "sample_rate": 16000, "encoding": "raw"
185.                                  }}}
186.                     ws.send(json.dumps(d))
187.                     break
189.                 # 模拟音频采样间隔
190.                 time.sleep(intervel)
192.     thread.start_new_thread(run, ())
194. if __name__ == "__main__":
195.     # 测试时候在此处正确填写相关信息即可运行
197.     wsParam = Ws_Param(APPID='', APISecret='',
198.                        APIKey='',
199.                        AudioFile=r'')
200.     websocket.enableTrace(False)
201.     wsUrl = wsParam.create_url()
202.     ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close)
203.     ws.on_open = on_open
204.     ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})

复制代码

讯飞星火语音听写流式 API 的 Python 接入流程可总结为：

• 起首准备 Python3.7 运行情况，安装代码指定版本的第三方依靠库；接着在讯飞开放平台获取 APPID、APIKey、APISecret 密钥，并准备好待辨认的音频文件；
  
• 然后通过密钥完成鉴权加密，天生 WebSocket 毗连 URL；
  
• 之后创建 WebSocket 客户端并与服务端创建毗连，毗连乐成后将音频文件按固定帧巨细分帧，以第一帧、中央帧、末了一帧的格式流式发送音频数据；
  
• 末了吸取并分析服务端返回的辨认效果，提取并输出笔墨内容，辨认完成后自动关闭毗连。
  

过程中可参考官方错误码排查标题，也可在平台设置热词、语种方言。
5.1.4 接口要求

这里告急讲了 API 接口要求，包罗：

• 底子协议与调用地点：
  
  
  
  • 哀求协议：支持 ws（WebSocket）/wss（加密 WebSocket），剧烈保举使用 wss 以提升安全性
    
  • 哀求地点：ws://​iat.xf-yun.com/v1（中英文语音辨认专属接口）
    
  
  
• 通用规范与鉴权要求：
  
  
  
  • 接口鉴权：接纳署名认证机制，需按照官方规则天生鉴权信息后才可调用
    
  • 字符编码：同一使用 UTF-8 编码
    
  • 相应格式：服务端返回效果同一为 JSON 格式
    
  • 开发语言：无穷定，只要可向讯飞云服务发起 HTTP 哀求的语言均可开发
    
  
  
• 音频参数要求：
  
  
  
  • 音频属性：采样率支持 16k 或 8K、位长 16bit、单声道
    
  • 音频格式：支持 pcm（需传参 audio.encoding=raw）、mp3（需传参 audio.encoding=lame）
    
  • 音频长度：单段待辨认音频最长不凌驾 60 秒
    
  
  

5.1.4.1 接口鉴权

本接口接纳 URL​ 参数鉴权 机制，是调用服务的前置安全校验：通过在 WebSocket 哀求地点后拼接 3 个必填鉴权参数，完成「身份校验 + 防篡改 + 防重放」三重安全防护，仅合法授权用户可乐成调用接口。
必填鉴权参数分析：

authorization 参数完备天生步调为：
在讯飞开放平台控制台，完成以下操纵获取 2 个 32 位核心密钥：

• 创建 WebAPI 平台应用
  
• 开通「语音听写（流式版）」服务
  
• 检察并生存：APIKey（身份标识）、APISecret（署名密钥，需严格保密）
  

天生合规 date 参数：

• 格式要求：严格遵照 RFC1123 标准，时区必须为 UTC+0/GMT
  
• 校验规则：服务端会校验本地时间与服务器时间的弊端，超时哀求直接拦截，需确保本地时间同步
  

构造署名原始字段（signature_origin）：按固定规则拼接 3 个参数（\n 为换行符，: 后必须生存 1 个空格）：

1. host: $host\n
2. date: $date\n
3. $request-line

复制代码

字段分析：

• $host：固定为 iat.xf-yun.com
  
• $date：步调 2 天生的 RFC1123 格式时间
  
• $request-line：固定为 GET /v1 HTTP/1.1（对应接口哀求路径）
  

示例：

1. host: iat.xf-yun.com
2. date: Tue, 14 May 2024 08:46:48 GMT
3. GET /v1 HTTP/1.1

复制代码

接着，使用 hmac-sha256 加密算法，以 APISecret 为密钥，对 signature_origin 举行加密署名：

1. signature_sha = hmac-sha256(signature_origin, $apiSecret)

复制代码

对 signature_sha 举行​Base64 编码​，得到可用于拼接的 signature 字符串。
按固定格式拼接（留意引号、空格不可错漏，headers 为固定参数名，不可修改）：

1. api_key="$api_key", algorithm="hmac-sha256", headers="host date request-line", signature="$signature"

复制代码

• 关键提示：headers​​的值为固定字符串 "host date request-line"​，仅代表到场署名的参数名，不是参数实际值
  

对 authorization_origin 举行​Base64 编码​，得到终极可拼接到 URL 的 authorization 参数
终极哀求格式为（保举使用​wss​加密协议提升安全性）：

1. ws[s]://iat.xf-yun.com/v1?authorization={authorization}&date={date}&host={host}

复制代码

5.1.4.2 数据传输示例

讯飞中英辨认大模子接纳​WebSocket ​全双工通讯​，握手乐成后客户端与服务端创建长毗连，支持双向数据传输：

• ​上传​​：客户端以流式分包方式发送音频数据，服务端实时分析并辨认
  
• ​吸取​：服务端完成辨认片断后，通过 WebSocket 自动推送辨认效果到客户端
  
• ​核心原则​：符合实时语音场景，克制发包隔断过短 / 包长过大导致辨认耗时增长
  

音频发送规范发起：

哀求 JSON 示例（分帧逻辑）如下：
第一帧数据（status=0）：

1. {
2.   "header": {
3.     "app_id": "your_appid",
4.     "res_id": "hot_words",
5.     "status": 0
6.   },
7.   "parameter": {
8.     "iat": {
9.       "domain": "slm",
10.       "language": "zh_cn",
11.       "accent": "mandarin",
12.       "dwa": "wpgs",
13.       "result": {
14.         "encoding": "utf8",
15.         "compress": "raw",
16.         "format": "json"
17.       }
18.     }
19.   },
20.   "payload": {
21.     "audio": {
22.       "encoding": "raw",
23.       "sample_rate": 16000,
24.       "channels": 1,
25.       "bit_depth": 16,
26.       "seq": 1,
27.       "status": 0,
28.       "audio": "AAAAAP..." // Base64编码的第一帧音频数据
29.     }
30.   }
31. }

复制代码

中央帧数据（status=1）：

1. {
2.   "header": {
3.     "app_id": "your_appid",
4.     "res_id": "hot_words",
5.     "status": 1
6.   },
7.   "parameter": {
8.     "iat": {...} // 与第一帧参数一致
9.   },
10.   "payload": {
11.     "audio": {
12.       "encoding": "raw",
13.       "sample_rate": 16000,
14.       "channels": 1,
15.       "bit_depth": 16,
16.       "seq": 2, // 帧序号递增
17.       "status": 1,
18.       "audio": "AAAAAA..." // Base64编码的中间帧音频数据
19.     }
20.   }
21. }

复制代码

末了一帧数据：

1. {
2.   "header": {
3.     "app_id": "your_appid",
4.     "res_id": "hot_words",
5.     "status": 2
6.   },
7.   "parameter": {
8.     "iat": {...} // 与第一帧参数一致
9.   },
10.   "payload": {
11.     "audio": {
12.       "encoding": "raw",
13.       "sample_rate": 16000,
14.       "channels": 1,
15.       "bit_depth": 16,
16.       "seq": 531, // 最后一帧序号
17.       "status": 2,
18.       "audio": "" // 最后一帧可无音频数据
19.     }
20.   }
21. }

复制代码

哀求核心参数分析：

返回 JSON 示例（流式辨认效果）：
第一帧返回（status=0）：

1. {
2.   "header": {
3.     "code": 0,
4.     "message": "success",
5.     "sid": "iat00e0044@hu18f5b16b033032...",
6.     "status": 0
7.   }
8. }

复制代码

中央帧返回（status=1）：

1. {
2.   "header": {
3.     "code": 0,
4.     "message": "success",
5.     "sid": "iat00e0044@hu18f5b16b033032...",
6.     "status": 1
7.   },
8.   "payload": {
9.     "result": {
10.       "compress": "raw",
11.       "encoding": "utf8",
12.       "format": "json",
13.       "seq": 2, // 对应请求帧序号
14.       "status": 1,
15.       "text": "cy12b1j6l9x1b..." // Base64编码的识别片段文本
16.     }
17.   }
18. }

复制代码

末了一帧返回（status=2）：

1. {
2.   "header": {
3.     "code": 0,
4.     "message": "success",
5.     "sid": "iat00e0044@hu18f5b16b033032...",
6.     "status": 2
7.   },
8.   "payload": {
9.     "result": {
10.       "compress": "raw",
11.       "encoding": "utf8",
12.       "format": "json",
13.       "seq": 76, // 最后一帧序号
14.       "status": 2,
15.       "text": "cy12b1j6l9x1bN2y..." // Base64编码的最终识别文本
16.     }
17.   }
18. }

复制代码

简单来说，就是：

• ​效果分析​：返回效果中 payload.result.text 为 Base64 编码的文本，需先解码再分析 JSON，提取终极笔墨
  
• ​状态关联​：返回的 header.status 与哀求帧 status 对应，标识当前辨认阶段
  
• ​错误处置惩罚​：header.code=0 体现乐成，非 0 为错误（需参考官方错误码排查）
  
• ​流式拼接​：中央帧返回的 text 为辨认片断，需按 seq 次序拼接，末了一帧为完备效果
  

5.1.4.3 哀求入参与返回出参分析

哀求参数顶层布局如下：

顶层布局如下：

文本解码后参数分析（text 字段）：payload.result.text 经 Base64 解码后为 JSON 对象，包罗布局化辨认效果，如下：

1. {
2.   "sn": 1,
3.   "ls": true,
4.   "bg": 0,
5.   "ed": 6,
6.   "ws": [{"cw": [{"w": "你好"}], "bg": 0}],
7.   "wb": 0,
8.   "wc": 0,
9.   "we": 0,
10.   "wp": 0,
11.   "ng": 0,
12.   "ph": 0
13. }

复制代码

仅当开启​动态修正​（parameter.iat.dwa="wpgs"）时返回，用于优化实时辨认效果：

5.2 实验前的准备

这是讯飞开放平台「星火语音辨认大模子」的产物首页，页面展示了该模子支持 37 种外语、202 种方言的实时语音转写本领，以及客户服务、游戏娱乐等实用场景和付费套餐，核心操纵是点击「免费试用」按钮，正式进入 ASR 服务的开通流程，是整个 API 设置的出发点。

上半部门为语音辨认服务套餐列表，点击「免费领取」选中中文语音辨认免费试用包；下半部门为服务购买页，点击「+」号按钮，触发新建应用流程，为 ASR 服务绑定专属的开发应用，是开通服务、天生 API 根据的关键前置步调。

上半部门是订单确认页，免费套餐应付金额为 0 元，勾选服务协议后点击「确认下单」，完成 ASR 服务的免费开通；下半部门展示了控制台入口，点击「控制台」即可进入应用管理配景，获取后续开发所需的 API 认证信息，是服务开通的收尾操纵。

上半部门是「我的应用」列表，表现了刚创建的「ASR 语音辨认」应用及其专属 APPID；下半部门是该应用的「中文辨认大模子」服务详情页，展示了剩余服务量、到期时间，以及 WebSocket 接口认证所需的 APPID、APIKey、APISecret 等核心开发根据，这些信息正是 Pico 端 xfyun_asr.py 代码中须要设置的授权参数，用于调用讯飞 ASR 语音辨认服务。

5.3 实验案例

在 upypi(https://upypi.net/)中搜刮 xfyun_asr：

这是一个实用于 MicroPython 的讯飞中英辨认大模子（ASR）驱动，通过 WebSocket 毗连讯飞开放平台，将 PCM 音频文件实时辨以为笔墨，已在树莓派 Pico 2W 上通过验证。
复制安装下令：

1. mpremote mip install https://upypi.net/pkgs/xfyun_asr/1.0.0

复制代码

输入到终端中实验：

然后将下面代码中 WIFI 和 API 干系设置项补全，烧录到单片机中：

1. # Python env   : MicroPython v1.23.0
2. # -*- coding: utf-8 -*-
3. # @Time    : 2026/04/14
4. # @Author  : leeqingsui
5. # @File    : main.py
6. # @Description : iFlytek TTS + ASR demo for MicroPython on Raspberry Pi Pico 2W
8. # ======================================== 导入相关模块 =========================================
10. import network
11. import asyncio
12. import time
13. import ntptime
14. from xfyun_tts import XfyunTTS
15. from xfyun_asr import XfyunASR
17. # ======================================== 全局变量 ============================================
19. WIFI_SSID     = "your_wifi_ssid"
20. WIFI_PASSWORD = "your_wifi_password"
22. # TTS 凭据（需在讯飞控制台开通语音合成服务）
23. TTS_APPID      = "your_tts_appid"
24. TTS_API_KEY    = "your_tts_api_key"
25. TTS_API_SECRET = "your_tts_api_secret"
27. # ASR 凭据（需在讯飞控制台开通中英识别大模型服务）
28. ASR_APPID      = "your_asr_appid"
29. ASR_API_KEY    = "your_asr_api_key"
30. ASR_API_SECRET = "your_asr_api_secret"
32. OUTPUT_PCM = "output.pcm"
33. OUTPUT_WAV = "output.wav"
35. # ======================================== 功能函数 ============================================
37. def connect_wifi():
38.     """
39. ​    连接 WiFi 并返回网络对象。
41. ​    Returns:
42. ​        network.WLAN: 已连接的 WLAN 对象；连接失败时返回 None。
44. ​    ==========================================
46. ​    Connect to WiFi and return the network object.
48. ​    Returns:
49. ​        network.WLAN: Connected WLAN object; None if connection fails.
50. ​    """
51. ​    ​wlan = network.WLAN(network.STA_IF)
52.     wlan.active(True)
54.     if not wlan.isconnected():
55.         print("Connecting to WiFi:", WIFI_SSID)
56.         wlan.connect(WIFI_SSID, WIFI_PASSWORD)
58.         timeout = 15
59.         while not wlan.isconnected() and timeout > 0:
60.             time.sleep(1)
61.             timeout -= 1
62.             print("Connecting...", timeout, "s remaining")
64.         if wlan.isconnected():
65.             ip_info = wlan.ifconfig()
66.             print("WiFi connected")
67.             print("IP:", ip_info[0])
68.             print("Gateway:", ip_info[2])
69.         else:
70.             print("WiFi connection failed")
71.             return None
72.     else:
73.         print("WiFi already connected")
75.     return wlan
77. def sync_ntp():
78.     """
79. ​    通过 NTP 同步系统时间，为讯飞 API 鉴权签名提供准确时间戳。
80. ​    依次尝试多个 NTP 服务器，全部失败时打印警告但不中断流程。
82. ​    ==========================================
84. ​    Sync system time via NTP for iFlytek API authentication signature.
85. ​    Tries multiple NTP servers in order; prints a warning if all ​fail.
86. ​    """
87. ​    ​servers = ("ntp.aliyun.com", "ntp.tencent.com", "pool.ntp.org")
88.     for host in servers:
89.         try:
90.             ntptime.host = host
91.             ntptime.settime()
92.             t = time.gmtime()
93.             print("NTP synced via {}: {}-{:02d}-{:02d} {:02d}:{:02d}:{:02d} UTC".format(
94.                 host, t[0], t[1], t[2], t[3], t[4], t[5]))
95.             return
96.         except Exception as e:
97.             print("NTP failed ({}):".format(host), e)
98.     print("NTP sync unavailable; signature timestamp may be rejected.")
100. async def run_tts(text):
101.     """
102. ​    调用 XfyunTTS 流式合成指定文本，将 PCM 数据写入 output.pcm。
104. ​    Args:
105. ​        text (str): 待合成的文本内容。
107. ​    ==========================================
109. ​    Synthesize text via XfyunTTS and write raw PCM to output.pcm.
111. ​    Args:
112. ​        text (str): Text to synthesize.
113. ​    """
114. ​    ​print("Synthesizing:", text)
115.     total = await tts.synthesize(text, filepath=OUTPUT_PCM)
117.     if total:
118.         print("Saved", total, "bytes ->", OUTPUT_PCM)
119.     else:
120.         print("Synthesis failed: no audio data received.")
122. async def run_tts_wav(text):
123.     """
124. ​    调用 XfyunTTS 流式合成指定文本，将带 WAV 头的音频写入 output.wav。
126. ​    Args:
127. ​        text (str): 待合成的文本内容。
129. ​    ==========================================
131. ​    Synthesize text via XfyunTTS and write WAV file to output.wav.
133. ​    Args:
134. ​        text (str): Text to synthesize.
135. ​    """
136. ​    ​print("Synthesizing (WAV):", text)
137.     total = await tts.synthesize(text, filepath=OUTPUT_WAV)
139.     if total:
140.         print("Saved", total, "bytes PCM +44 bytes header ->", OUTPUT_WAV)
141.     else:
142.         print("Synthesis failed: no audio data received.")
144. async def run_asr(filepath):
145.     """
146. ​    调用 XfyunASR 识别指定 PCM 文件，打印识别结果。
148. ​    Args:
149. ​        filepath (str): 待识别的 PCM 文件路径。
151. ​    ==========================================
153. ​    Recognize a PCM file via XfyunASR and print the result.
155. ​    Args:
156. ​        filepath (str): Path to the PCM file to recognize.
157. ​    """
158. ​    ​print("Recognizing:", filepath)
159.     text = await asr.recognize(filepath)
161.     if text:
162.         print("ASR result:", text)
163.     else:
164.         print("Recognition failed: no text returned.")
166. # ======================================== 自定义类 ============================================
168. # ======================================== 初始化配置 ===========================================
170. time.sleep(3)
171. print("FreakStudio: iFlytek TTS + ASR Demo")
173. tts = XfyunTTS(
174.     app_id     = TTS_APPID,
175.     api_key    = TTS_API_KEY,
176.     api_secret = TTS_API_SECRET,
177. )
179. # ASR 采样率须与 TTS 输出一致（TTS 默认 auf=audio/L16;rate=8000 → 8000 Hz）
180. asr = XfyunASR(
181.     app_id      = ASR_APPID,
182.     api_key     = ASR_API_KEY,
183.     api_secret  = ASR_API_SECRET,
184.     sample_rate = 8000,
185. )
187. # ========================================  主程序  ===========================================
189. if __name__ == "__main__":
190.     if not connect_wifi():
191.         print("Aborting: WiFi unavailable.")
192.     else:
193.         sync_ntp()
195.         # Demo 1: TTS 合成中英混合文本 → output.pcm
196.         asyncio.run(run_tts("大家好一块吃饭吧hello"))
198.         # Demo 2: ASR 识别 output.pcm → 打印文字
199.         asyncio.run(run_asr(OUTPUT_PCM))
201.         # Demo 3: TTS 合成 → output.wav（可在 PC 上直接播放验证）
202.         asyncio.run(run_tts_wav("Hi there, this is a WAV format test from Pico 2W."))

复制代码

这是一段基于树莓派 Pico 2W 硬件、运行在 MicroPython v1.23.0 情况的讯飞语音交互演示代码，它集成了讯飞语音合成（TTS）与语音辨认（ASR） 双功能，通过 WiFi 毗连网络并同步 NTP 时间完成 API 鉴权，可将文本转换为 PCM/WAV 格式音频文件，也能对音频文件举行笔墨辨认，完备实现了「笔墨转语音、语音转笔墨」的闭环语音交互测试，仅需设置讯飞平台授权信息与 WiFi 信息即可运行，是嵌入式装备实现离线 / 在线语音交互的核心示例步伐。
打开终端表现如下：

可以看到，驱动库正常安装：

这是主步伐 main.py 运行后的 Shell 日志：

• TTS 语音合成环节（第一个红框）：
  
  
  
  • 输入：中英混淆文本 各人好一块用饭吧hello
    
  • 实验：通过 XfyunTTS 调用讯飞 TTS API，WebSocket 毗连乐成，流式吸取音频分片，终极天生 output.pcm 原始 PCM 音频文件（总巨细 39252 字节）
    
  • 验证：TTS 合成正常，支持中英混淆文本，流式传输适配 MicroPython 的内存限定（单片机内存小，无法一次性加载大音频）。
    
  
  
• ASR 语音辨认环节（第二个红框）：
  
  
  
  • 输入：刚才 TTS 天生的 output.pcm 音频文件
    
  • 实验：通过 XfyunASR 调用讯飞 ASR API，分帧发送音频数据，吸取辨认效果
    
  • 输出：ASR result: 各人好，一块用饭吧，hello。
    
  • 验证：
    
    
    
    • ✅ 100% 还原输入文本，无错字、漏字
      
    • ✅ 中英混淆辨认精确，自动添加规范标点
      
    • ✅ 8000Hz 采样率的 PCM 音频辨认正常（和 TTS 输出采样率完全匹配，无格式标题）
      
    
    
  
  
• 附加验证：
  
  
  
  • 后续日志：TTS 合成 WAV 格式文件（output.wav），可直接在电脑上用播放器打开，验证音频格式精确。
    
  • 网络与时间：WiFi 毗连正常、NTP 时间同步乐成（讯飞 API 鉴权须要精确时间戳，确保署名有用），整个流程无鉴权错误。
    
  
  

输出的 PCM 文件可以在网站 https://pcm.qer.im/中播放：

验证乐成。
原文链接：https://f1829ryac0m.feishu.cn/wiki/J0lDwNHtCizJwekOXRecPA7inwb?fromScene=spaceOverview