python代码示例
import json
import asyncio
import websockets
from uuid import uuid4
audioFile = "test.mp3"
productId = "x"
apikey = "x"
async def textRequest(ws):
content = {
"aiType":"dm",
"topic": 'nlu.input.text',
"recordId": uuid4().hex,
"refText": "测试"
}
try:
await ws.send(json.dumps(content))
resp = await ws.recv()
print(resp)
except websockets.exceptions.ConnectionClosed as exp:
print(exp)
async def audioRequest(ws):
content = {
"aiType": "dm",
"topic": "recorder.stream.start",
"recordId": uuid4().hex,
"audio": {
"audioType": "mp3",
"sampleRate": 16000,
"channel": 1,
"sampleBytes": 2
},
"asrParams": {
"realBack": True,
"enableVAD": True,
"enablePunctuation": True,
"enableTone": True,
"enableConfidence": True,
"enableNumberConvert": True,
},
}
try:
await ws.send(json.dumps(content))
with open(audioFile, 'rb') as f:
while True:
chunk = f.read(400)
if not chunk:
await ws.send(bytes("", encoding="utf-8"))
break
print(len(chunk))
await ws.send(chunk)
async for message in ws:
print(message)
resp = json.loads(message)
if 'dm' in resp:
break
except websockets.exceptions.ConnectionClosed as exp:
print(exp)
ws.close()
async def dds_demo():
url = f"ws://botai-dsg.and-home.cn:4443/dsg/v1/prod?productId={productId}&apikey={apikey}"
print(url)
async with websockets.connect(url) as websocket:
await audioRequest(websocket)
asyncio.get_event_loop().run_until_complete(dds_demo())