超拟人合成

# 超拟人合成接入文档

## 1. 能力介绍
语音合成是根据文本生成声音。超拟人合成是在语音合成基础上，精准模拟人类发言的呼吸、叹气、语速变化等，使得发音更自然。

##  2. 准备工作
### 2.1. 能力接入
（1）开通aiui交互大模型
（2）SDK接入：要求SDK版本高于 6.6.x，且aiui.cfg中指定交互大模型链路，即aiui_ver:2
（3）API接入：[大模型接入API文档](https://aiui-doc.xf-yun.com/project-1/doc-181/)

### 2.2. 合成结果下发说明

超拟人合成目前仅支持音频流数据下发，不支持下发url

### 2.3. 授权申请
- 联系商务或发送邮件到 aiui_support@iflytek.com 申请。

### 2.4. 开启超拟人发音人授权
- 授权后，在AIUI应用下选择超拟人发音人，点击页面右上角保存修改
![](/media/202408/1725002257403_1725002309.png)

## 3.  能力使用
>注意：
><span style="color:red">当前超拟人合成仅支持主动文本合成调用</span>

### 3.1.  SDK调用
注意发音人参数和情景模式取值，示例代码如下
- scene 固定取值  IFLYTEK.hts
- vcn 取值为超拟人发音人名称，各超拟人发音人名称取值见发音人[列表文档](https://aiui-doc.xf-yun.com/project-1/doc-93/)

```java
String text = "今天晚上吃点啥好呢";
String params = "vcn=x4_lingxiaoxuan_oral,volume=100,scene=IFLYTEK.hts";
byte[] textData = text.getBytes("utf-8");
AIUIMessage ttsMessage = new AIUIMessage(AIUIConstant.CMD_TTS, AIUIConstant.START, 0,params,textData);
mAIUIAgent.sendMessage(ttsMessage);
```

### 3.2.  API调用

使用交互大模型API进行文本合成请求，与SDK形式一样，进行scene和vcn参数设置，具体请求构建如下：

- scene 固定取值  IFLYTEK.hts  
- vcn 取值为超拟人发音人名称，各超拟人发音人名称取值见发音人[列表文档](https://aiui-doc.xf-yun.com/project-1/doc-93/)

构建请求json如下

```json
{
    "header": {
        "sn": "1234567890",
        "appid": "a44e0f36",
        "stmid": "text-1",
        "status": 3,
        "scene": "IFLYTEK.hts"
    },
    "parameter": {
        "tts": {
            "vcn": "x4_lingxiaoqi_oral",
            "tts": {
                "channels": 1,
                "sample_rate": 16000,
                "bit_depth": 16,
                "encoding": "raw"
            }
        }
    },
    "payload": {
        "text": {
            "compress": "raw",
            "format": "plain",
            "text": "5omT55S16K+d57uZ5byg5LiJ",
            "encoding": "utf8",
            "status": 3
        }
    }
}
```

示例代码如下：

```python
import _thread as thread
import base64
import datetime
import hashlib
import hmac
import json
from urllib.parse import urlparse
import time
from datetime import datetime
from time import mktime
from urllib.parse import urlencode
from wsgiref.handlers import format_date_time

import websocket

## 修改应用应用配置后直接执行即可

# 请求地址
url = "wss://aiui.xf-yun.com/v2/aiint/ws"

appid = "xxx"
api_key = "xxx"
api_secret = "xxx"

# 场景
scene = "IFLYTEK.hts"
# scene = "main_box"

# 请求类型用来设置文本请求还是音频请求，text/audio
data_type = 'text'

# 请求文本
question = "今天晚上吃点啥好呢，是酱香饼还是麻酱拌凉皮"

# 每帧音频数据大小，单位字节
chuncked_size = 1024

audio_save_fp = open('D:\save_hts_audio.pcm', mode='wb+')

class AIUIV2WsClient(object):
    # 初始化
    def __init__(self):
        self.handshake = self.assemble_auth_url(url)

# 生成握手url
    def assemble_auth_url(self, base_url):
        host = urlparse(base_url).netloc
        path = urlparse(base_url).path
        # 生成RFC1123格式的时间戳
        now = datetime.now()
        date = format_date_time(mktime(now.timetuple()))

# 拼接字符串
        signature_origin = "host: " + host + "\n"
        signature_origin += "date: " + date + "\n"
        signature_origin += "GET " + path + " HTTP/1.1"

# 进行hmac-sha256进行加密
        print(signature_origin)
        signature_sha = hmac.new(api_secret.encode('utf-8'), signature_origin.encode('utf-8'),
                                 digestmod=hashlib.sha256).digest()

signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')

authorization_origin = f'api_key="{api_key}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'

print('get authorization_origin:', authorization_origin)
        authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')

# 将请求的鉴权参数组合为字典
        v = {
            "host": host,
            "date": date,
            "authorization": authorization,
        }
        # 拼接鉴权参数，生成url
        url = base_url + '?' + urlencode(v)
        # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释，比对相同参数时生成的url与自己代码生成的url是否一致
        return url

def on_open(self, ws):
        # 连接建立成功后开始发送数据
        print("### ws connect open")
        thread.start_new_thread(self.run, ())

def run(self):
        if data_type == "text":
            self.text_req()

def text_req(self):
        aiui_data = {
            "header": {
                "sn":"1234567891",
                "app_id": appid,
                "stmid": "text-1",
                "status": 3,
                "scene": scene,
                "pers_param":"{\"appid\":\"\",\"uid\":\"iflytek-test\"}"
            },
            "parameter": {
                "nlp": {
                    "nlp": {
                        "compress": "raw",
                        "format": "json",
                        "encoding": "utf8",
                        
                    },
                    # 动态实体
                    "sub_scene": "cbm_v45",
                    "new_session": True
                },
                "tts": {
                    "vcn": "x4_lingxiaoxuan_oral",
                    "tts": {
                        "channels": 1,
                        "sample_rate": 16000,
                        "bit_depth": 16,
                        "encoding": "raw"
                    },
                    "tts_res_type": "url"
                }
            },
            "payload": {
                "text": {
                    "compress": "raw",
                    "format": "plain",
                    "text": base64.b64encode(question.encode('utf-8')).decode('utf-8'),
                    "encoding": "utf8",
                    "status": 3
                }
            }
        }
        data = json.dumps(aiui_data, indent=2)
        print('--------text request data--------\n', data,'\n--------text request data--------\n')
        self.ws.send(data)

# 收到websocket消息的处理
    def on_message(self, ws, message):
        data = json.loads(message)

# print('原始结果:', message)
        header = data['header']
        code = header['code']
        # 结果解析
        if code != 0:
            print('请求错误:', code, json.dumps(data, ensure_ascii=False))
            ws.close()
        if 'nlp' in message:
            nlp_json = data['payload']['nlp']
            nlp_text_bs64 = nlp_json['text']
            nlp_text = base64.b64decode(nlp_text_bs64).decode('utf-8')
            nlp_all = json.dumps(data,indent=2)
            # print("--------nlp--------\n" , nlp_all)
            # print("语义结果：",nlp_text,"--------nlp--------\n")
        if 'tts' in message:
            tts_all = json.dumps(data,indent=2)
            print("--------hts_all--------\n",tts_all,"\n--------hts_all_all--------\n")
            tts_audio =  data['payload']['tts']['audio']
            tts_content = base64.b64decode(tts_audio)
            audio_save_fp.write(tts_content)

if 'status' in header and header['status'] == 2:
            # 接受最后一帧结果，关闭连接
            ws.close()

def on_error(self, ws, error):
        print("### connection error: " + error)
        ws.close()

def on_close(self, ws, close_status_code, close_msg):
        print("### connection is closed ###, cloce code:" + str(close_status_code))

def start(self):
        self.ws = websocket.WebSocketApp(
            self.handshake,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
        )
        self.ws.run_forever()

if __name__ == "__main__":

client = AIUIV2WsClient()
    client.start()

```