树莓派 ubuntu20.04下 python调讯飞的语音API,语音识别和语音合成

news2024/11/24 20:46:07

目录

    • 1.环境搭建
    • 2.去讯飞官网申请密钥
    • 3.语音识别(sst)
    • 4.语音合成(tts)
    • 5.USB声卡可能报错

1.环境搭建

#环境说明:(尽量在ubuntu下使用, 本次代码均在该环境下实现)
sudo apt-get install sox  # 安装语音播放软件
pip install websocket-client==1.7.0 -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install pydub==0.25.1 -i https://pypi.tuna.tsinghua.edu.cn/simple

2.去讯飞官网申请密钥

https://www.xfyun.cn/
在这里插入图片描述

3.语音识别(sst)

语音转文字

#! /usr/bin/env python
# -*- coding:utf-8 -*-
'''
1.环境说明:(尽量在ubuntu下使用, 本次代码均在该环境下实现)
pip install websocket-client==1.7.0 -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install pydub==0.25.1 -i https://pypi.tuna.tsinghua.edu.cn/simple

pyaudio的安装(注意区分不同平台)
    windows pip install pyaudio -i https://pypi.tuna.tsinghua.edu.cn/simple
    Linux   sudo apt install python3-pyaudio
    mac     brew install portaudio
            pip install pyaudio -i https://pypi.tuna.tsinghua.edu.cn/simple

2.密钥说明:
每一个账号的用额只有500次数的请求,平时用足够使用了。
自行前往去注册:https://www.xfyun.cn/

'''
import wave
import pyaudio
import websocket
import datetime
import hashlib
import base64
import hmac
import json
from urllib.parse import urlencode
import time
import ssl
from wsgiref.handlers import format_date_time
from datetime import datetime
from time import mktime
import _thread as thread

STATUS_FIRST_FRAME = 0  # 第一帧的标识
STATUS_CONTINUE_FRAME = 1  # 中间帧标识
STATUS_LAST_FRAME = 2  # 最后一帧的标识


class Ws_Param(object):
    # 初始化
    def __init__(self, APPID, APIKey, APISecret, AudioFile):
        self.APPID = APPID
        self.APIKey = APIKey
        self.APISecret = APISecret
        self.AudioFile = AudioFile

        # 公共参数(common)
        self.CommonArgs = {"app_id": self.APPID}
        # 业务参数(business),更多个性化参数可在官网查看
        self.BusinessArgs = {"domain": "iat", "language": "zh_cn", "accent": "mandarin", "vinfo":1,"vad_eos":10000}

    # 生成url
    def create_url(self):
        url = 'wss://ws-api.xfyun.cn/v2/iat'
        # 生成RFC1123格式的时间戳
        now = datetime.now()
        date = format_date_time(mktime(now.timetuple()))

        # 拼接字符串
        signature_origin = "host: " + "ws-api.xfyun.cn" + "\n"
        signature_origin += "date: " + date + "\n"
        signature_origin += "GET " + "/v2/iat " + "HTTP/1.1"
        # 进行hmac-sha256进行加密
        signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
                                 digestmod=hashlib.sha256).digest()
        signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')

        authorization_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"" % (
            self.APIKey, "hmac-sha256", "host date request-line", signature_sha)
        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
        # 将请求的鉴权参数组合为字典
        v = {
            "authorization": authorization,
            "date": date,
            "host": "ws-api.xfyun.cn"
        }
        # 拼接鉴权参数,生成url
        url = url + '?' + urlencode(v)
        # print("date: ",date)
        # print("v: ",v)
        # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释,比对相同参数时生成的url与自己代码生成的url是否一致
        # print('websocket url :', url)
        return url


# 收到websocket消息的处理
def on_message(ws, message):
    try:
        code = json.loads(message)["code"]
        sid = json.loads(message)["sid"]
        if code != 0:
            errMsg = json.loads(message)["message"]
            print("sid:%s call error:%s code is:%s" % (sid, errMsg, code))

        else:
            data = json.loads(message)["data"]["result"]["ws"]
            # print(json.loads(message))
            result = ""
            for i in data:
                for w in i["cw"]:
                    result += w["w"]
            print(result)
            # print(json.dumps(data, ensure_ascii=False))
    except Exception as e:
        print("receive msg,but parse exception:", e)




def on_error(ws, error):
    pass

def on_close(ws):
    print("### closed ###")



def on_open(ws):
    def run(*args):
        frameSize = 8000  # 每一帧的音频大小
        intervel = 0.04  # 发送音频间隔(单位:s)
        status = STATUS_FIRST_FRAME 

        with open(wsParam.AudioFile, "rb") as fp:
            while True:
                buf = fp.read(frameSize)
                # 文件结束
                if not buf:
                    status = STATUS_LAST_FRAME
                # 第一帧处理
                # 发送第一帧音频,带business 参数
                # appid 必须带上,只需第一帧发送
                if status == STATUS_FIRST_FRAME:

                    d = {"common": wsParam.CommonArgs,
                         "business": wsParam.BusinessArgs,
                         "data": {"status": 0, "format": "audio/L16;rate=16000",
                                  "audio": str(base64.b64encode(buf), 'utf-8'),
                                  "encoding": "raw"}}
                    d = json.dumps(d)
                    ws.send(d)
                    status = STATUS_CONTINUE_FRAME
                # 中间帧处理
                elif status == STATUS_CONTINUE_FRAME:
                    d = {"data": {"status": 1, "format": "audio/L16;rate=16000",
                                  "audio": str(base64.b64encode(buf), 'utf-8'),
                                  "encoding": "raw"}}
                    ws.send(json.dumps(d))
                # 最后一帧处理
                elif status == STATUS_LAST_FRAME:
                    d = {"data": {"status": 2, "format": "audio/L16;rate=16000",
                                  "audio": str(base64.b64encode(buf), 'utf-8'),
                                  "encoding": "raw"}}
                    ws.send(json.dumps(d))
                    time.sleep(1)
                    break
                # 模拟音频采样间隔
                time.sleep(intervel)
        ws.close()

    thread.start_new_thread(run, ())
def record(time):    #录音程序
    CHUNK = 1024
    FORMAT = pyaudio.paInt16
    CHANNELS = 1
    RATE = 16000
    RECORD_SECONDS =time
    WAVE_OUTPUT_FILENAME = "./output.pcm"
    p = pyaudio.PyAudio()
    stream = p.open(format=FORMAT,
                    channels=CHANNELS,
                    rate=RATE,
                    input=True,
                    frames_per_buffer=CHUNK)
    print("* recording")
    frames = []
    for i in range(0,int(RATE / CHUNK * RECORD_SECONDS)):
        data = stream.read(CHUNK)
        frames.append(data)
    print("* done recording")
    stream.stop_stream()
    stream.close()
    p.terminate()
    wf = wave.open(WAVE_OUTPUT_FILENAME, 'wb')
    wf.setnchannels(CHANNELS)
    wf.setsampwidth(p.get_sample_size(FORMAT))
    wf.setframerate(RATE)
    wf.writeframes(b''.join(frames))
    wf.close()

if __name__ == "__main__":
    record(5)
    time1 = datetime.now()
    # 这里改成自己的key秘钥, 每个人每天只有500的访问量。
    wsParam = Ws_Param(APPID='d69356bc', APIKey='c3f938a4da84f7449bd2f958d461e7e1',
                       APISecret='ZjE4ZGE4ZGU4YzViZmFhNTI0ZmYyNTE0',
                       AudioFile=r'./output.pcm')
    websocket.enableTrace(False)
    wsUrl = wsParam.create_url()
    ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close)
    ws.on_open = on_open
    ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
    time2 = datetime.now()

4.语音合成(tts)

文字转语音

#! /usr/bin/env python
# -*- coding:utf-8 -*-
'''
1.环境说明:(尽量在ubuntu下使用, 本次代码均在该环境下实现)
pip install websocket-client==1.7.0 -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install pydub==0.25.1 -i https://pypi.tuna.tsinghua.edu.cn/simple
pip install pygame -i https://pypi.tuna.tsinghua.edu.cn/simple

pyaudio的安装(注意区分不同平台)
    windows pip install pyaudio -i https://pypi.tuna.tsinghua.edu.cn/simple
    Linux   sudo apt install python3-pyaudio
    mac     brew install portaudio
            pip install pyaudio -i https://pypi.tuna.tsinghua.edu.cn/simple

2.密钥说明:
每一个账号的用额只有500次数的请求,平时用足够使用了。
自行前往去注册:https://www.xfyun.cn/

'''
import websocket
import datetime
import hashlib
import base64
import hmac
import json
from urllib.parse import urlencode
import time
import ssl
from wsgiref.handlers import format_date_time
from datetime import datetime
from time import mktime
import _thread as thread
import os
from pydub import AudioSegment
import time
import pygame

STATUS_FIRST_FRAME = 0  # 第一帧的标识
STATUS_CONTINUE_FRAME = 1  # 中间帧标识
STATUS_LAST_FRAME = 2  # 最后一帧的标识


class Ws_Param(object):
    # 初始化
    def __init__(self, APPID, APIKey, APISecret, Text):
        self.APPID = APPID
        self.APIKey = APIKey
        self.APISecret = APISecret
        self.Text = Text

        # 公共参数(common)
        self.CommonArgs = {"app_id": self.APPID}
        # 业务参数(business),更多个性化参数可在官网查看
        self.BusinessArgs = {"aue": "raw", "auf": "audio/L16;rate=16000", "vcn": "xiaofeng", "tte": "utf8"}
        self.Data = {"status": 2, "text": str(base64.b64encode(self.Text.encode('utf-8')), "UTF8")}


    # 生成url
    def create_url(self):
        url = 'wss://tts-api.xfyun.cn/v2/tts'
        # 生成RFC1123格式的时间戳
        now = datetime.now()
        date = format_date_time(mktime(now.timetuple()))

        # 拼接字符串
        signature_origin = "host: " + "ws-api.xfyun.cn" + "\n"
        signature_origin += "date: " + date + "\n"
        signature_origin += "GET " + "/v2/tts " + "HTTP/1.1"
        # 进行hmac-sha256进行加密
        signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
                                 digestmod=hashlib.sha256).digest()
        signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')

        authorization_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"" % (
            self.APIKey, "hmac-sha256", "host date request-line", signature_sha)
        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
        # 将请求的鉴权参数组合为字典
        v = {
            "authorization": authorization,
            "date": date,
            "host": "ws-api.xfyun.cn"
        }
        # 拼接鉴权参数,生成url
        url = url + '?' + urlencode(v)
        # print("date: ",date)
        # print("v: ",v)
        # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释,比对相同参数时生成的url与自己代码生成的url是否一致
        # print('websocket url :', url)
        return url

def on_message(ws, message):
    try:
        message =json.loads(message)
        code = message["code"]
        sid = message["sid"]
        audio = message["data"]["audio"]
        audio = base64.b64decode(audio)
        status = message["data"]["status"]
        # print(message)
        if status == 2:
            # print("ws is closed")
            ws.close()
        if code != 0:
            errMsg = message["message"]
            print("sid:%s call error:%s code is:%s" % (sid, errMsg, code))
        else:

            with open('./demo.pcm', 'ab') as f:
                f.write(audio)

    except Exception as e:
        print("receive msg,but parse exception:", e)



# 收到websocket错误的处理
def on_error(ws, error):
    print("### error:", error)


# 收到websocket关闭的处理
def on_close(ws,arg1,arg2):
    # print("### closed ###")
    pass

def play_fun():
    # 加载PCM音频文件
    audio = AudioSegment.from_file("demo.pcm", format="raw", sample_width=2, frame_rate=16000, channels=1)

    # 将音频转换为MP3格式并播放
    audio.export("file.mp3", format="mp3")

    time.sleep(1)
    # # ==== windows 平台使用下面代码来播放音频,并将下面的代码注释
    # # 初始化pygame
    # pygame.mixer.init()

    # # 加载MP3文件
    # pygame.mixer.music.load("file.mp3")

    # # 播放MP3文件
    # pygame.mixer.music.play()

    # # 等待音频播放完毕
    # while pygame.mixer.music.get_busy():
    #     pygame.time.Clock().tick(10)
    # print("------ 播放完毕 ------")

    # ==== linux 平台使用下面代码来播放音频,并将上面的代码注释
    file = "file.mp3"
    os.system('mpg123 '+ file)


def tts_fun(string_txr):
    # 收到websocket连接建立的处理
    def on_open(ws):
        def run(*args):
            d = {"common": wsParam.CommonArgs,
                "business": wsParam.BusinessArgs,
                "data": wsParam.Data,
                }
            d = json.dumps(d)
            print("------开始发送文本数据------")
            ws.send(d)
            if os.path.exists('./demo.pcm'):
                os.remove('./demo.pcm')

        thread.start_new_thread(run, ())
    # 测试时候在此处正确填写相关信息即可运行
    # 这里改成自己的key ,每一个人的用额都是有限的。https://console.xfyun.cn/services/iat
    wsParam = Ws_Param(APPID='d69356bc', APISecret='c3f938a4da84f7449bd2f958d461e7e1',
                       APIKey='ZjE4ZGE4ZGU4YzViZmFhNTI0ZmYyNTE0',
                       Text=string_txr)
    
    websocket.enableTrace(False)
    wsUrl = wsParam.create_url()
    ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close)
    ws.on_open = on_open
    ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})

def tts_main(input_txt):
    tts_fun(input_txt)
    play_fun()

class tts_clss():
    def __init__(self, input_txt):
        tts_fun(input_txt)
        play_fun()

if __name__ == "__main__":
    tts_main("你好,我是机器人!")

5.USB声卡可能报错

设置完默认声卡后,用Python录音和播放会出现一些错误提示,但发现录音和播放都正常,错误显示比如这样:

ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.rear
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.center_lfe
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.side
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.hdmi
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.hdmi
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.modem
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.modem
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.phoneline
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.phoneline
ALSA lib confmisc.c:1281:(snd_func_refer) Unable to find definition ‘defaults.bluealsa.device’
ALSA lib conf.c:4528:(_snd_config_evaluate) function snd_func_refer returned error: No such file or directory
ALSA lib conf.c:4996:(snd_config_expand) Args evaluate error: No such file or directory
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM bluealsa
ALSA lib confmisc.c:1281:(snd_func_refer) Unable to find definition ‘defaults.bluealsa.device’
ALSA lib conf.c:4528:(_snd_config_evaluate) function snd_func_refer returned error: No such file or directory
ALSA lib conf.c:4996:(snd_config_expand) Args evaluate error: No such file or directory
ALSA lib pcm.c:2495:(snd_pcm_open_noupdate) Unknown PCM bluealsa
connect(2) call to /tmp/jack-1000/default/jack_0 failed (err=No such file or directory)
attempt to connect to server failed

为了不显示这些错误可以在Python代码里p = pyaudio.PyAudio() 之前加上:

os.close(sys.stderr.fileno())

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/1343380.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

边缘计算网关:重新定义物联网数据处理

随着物联网(IoT)设备的爆炸式增长,数据处理和分析的需求也在迅速增加。传统的数据处理方式,将所有数据传输到中心服务器进行处理,不仅增加了网络负担,还可能导致数据延迟和安全问题。因此,边缘计…

私有部署ELK,搭建自己的日志中心(四)-- kibana展示es的数据

一、说在前面的话 前一篇已把elk的安装连带讲完,本文重在讲述如何在kibana展示es数据。 二、数据的展示 展示es数据库的客户端工具有很多,比如es head插件,但是一说到要查询日志,还是非kibana莫属了。 1、kibana.yml # 服务端…

OpenCV入门01:图像处理简介/图像的基础操作

项目开源,地址:https://gitee.com/zccbbg/opencv_study 文章目录 图像处理简介灰度图像二值图像彩色图 opencv 介绍图像基础操作图像读取与显示绘制几何图形图像的属性其他操作算数操作加法混合 图像色彩空间转换 图像处理简介 灰度图像 ● 灰度图像是…

Linux服务器搭建笔记-006:拓展/home目录容量

一、问题说明 Ubuntu服务器在使用过程中创建的新用户,每位用户会在/home目录下生成一个属于其个人的主文件夹。如果不限制各个用户的使用空间,所有的用户都会共用/home所挂载的硬盘。在这种多用户情况下,会很快的填满/home目录,导…

一款超酷的一体化网站测试工具:Web-Check

Web-Check 是一款功能强大的一体化工具,用于发现网站/主机的相关信息。用于检查网页的工具,用于确保网页的正确性和可访问性。它可以帮助开发人员和网站管理员检测网页中的错误和问题,并提供修复建议。 它只需要输入一个网站就可以查看一个网…

企业品牌推广在国外媒体投放的意义和作用何在?

海外广告投放是企业在国际市场推广的重要战略,具有多种形式,包括社交媒体广告、短视频广告、电视广告等。这些广告形式在传播信息、推动销售、塑造品牌形象等方面发挥着独特的作用。 其中软文发稿是一种注重叙事和信息传递的广告形式,对于企…

k8s 架构

主要组件 k8s有如下的主要组件: Control plane(s) and worker node(s)OperatorsServicesPods of containersNamespaces and quotasNetwork and policiesStorage. 一个k8s集群是有一个或多个 cp(控制平面)节点和一组worker 节点组成的。这个…

ChatGPT使用注意事项有哪些?

一、环境注意事项 1、需要构造稳定的环境,很多人说自己的号为什么突然被封,被封的原因是因为有些环境会自动换IP,所以你要设置好 2、搭建美国住宅IP 3、注册时也不要使用香港,最好选择美国、新加坡等地区注册 二、API调用注意事…

(已解决)(pytorch指定了gpu但还是占用了一点0号gpu)以及错误(cuDNN error: CUDNN_STATUS_INTERNAL_ERROR)

文章目录 错误原因解决问题 错误原因 出现错误cuDNN error: CUDNN_STATUS_INTERNAL_ERROR,从这个名字就可以看出,出错原因其实有可能有很多种,我这里说一种比较常见的,就是:显存不足。 一个困惑点在于,在…

k8s的陈述式资源管理(命令行操作)

(一)k8s的陈述式资源管理 1、命令行:kubectl命令行工具——用于一般的资源管理 (1)优点:90%以上ce场景都可以满足 (2)特点:对资源的增、删、查比较方便,对…

【neo4j】desktop下载

【neo4j】desktop下载 https://neo4j.com/download/ 点击download,填写表格 之后就可以正常使用了

Jackson进行Json反序列化对于一个小写字母开头后跟大写字母无法识别反序列成功问题

问题描述 json数据:{“pTargetId”:“123”} javaBean: Datapublic static class Test {private String pTargetId;}运行下面代码: public static void main(String[] args) throws JsonProcessingException {String json "{\"pT…

docker入门概念详解

本篇文章对docker的一些基础概念和周边概念进行了详细解释。帮助你可以很好的理解docker是用来干什么的,docker是怎么工作的。其中有docker所运用到的技术解释,docker的不同发展版本,dokcer的架构,docker的生态等等详解。希望本片…

flink generic log-based incremental checkpoints 设计

背景 flink 在1.15版本后开始提供generic log-based incremental checkpoints的检查点方案,目的在于减少checkpoint的耗时,尽量缩短端到端的数据处理延迟,本文就来看下这种新类型的checkpoint的设计 generic log-based incremental checkpo…

GPT-3: Language Models are Few-Shot Learners

GPT-3 论文 数据集 CommonCrawl:文章通过高质量参考语料库对CommonCrawl数据集进行了过滤,并通过模糊去重对文档进行去重,且增加了高质量参考语料库以增加文本的多样性。WebText:文章采用了类似GPT-2中的WebText文档收集清洗方…

【FFI】N-API的JS堆对象生命周期管理

N-API的JS堆对象生命周期管理 N-API是Node API的简写,同时也是nodejs的JS VM(链)接入原生模块.node文件的应用程序二进制接口(i.e. ABI)。借助N-API引入的抽象隔离,升级nodejs运行时(虚拟机) 【编译】不要求…

米表网PHP域名销售管理系统网站源码 自适应电脑+手机端

PHP域名销售管理系统网站源码 自适应电脑手机端 功能使用简单,不复杂,非常适合个人米表使用,带广告栏 源码下载:https://download.csdn.net/download/m0_66047725/88646799

Spring Boot整合GraphQL

RPC选型入门测试系列文章 GraphQL是一种用于API开发的查询语言和运行时环境。它由Facebook开发并于2015年开源。GraphQL的主要目标是提供一种更高效、灵活和易于使用的方式来获取和操作数据。与传统的RESTful API相比,GraphQL允许客户端精确地指定需要的数据&#…

C语言实例_stdlib.h库函数功能及其用法详解

一、前言 C语言作为一种高效、灵活的编程语言,标准库的使用对于开发人员来说是不可或缺的。其中,stdlib.h是C语言中一个重要的标准库头文件,提供了许多常用的函数和工具,以便开发人员能够更加便捷地进行内存管理、字符串处理、随…

《深入理解JAVA虚拟机笔记》运行时栈帧、方法分派、动态类型

运行时栈帧结构 Java 虚拟机以方法作为最基本的执行单元,“栈帧”(Stack Frame)则是用于支持虚拟机进行方法调用和方法执行背后的数据结构,它也是虚拟机运行时数据区中的虚拟机栈的栈元素。栈帧存储了方法的局部变量表、操作数栈…