Ошибка Azure Gstreamer — basesrc gstbasesrc.c:3072: ошибка: потоковая передача остановлена, причина ошибки (-5) ⇐ Python

[Anonymous]

Мы используем GStreamer для преобразования речи в текст в реальном времени Azure. На самом деле Azure использует gstreamer внутри себя для преобразования любого аудиоформата, передаваемого в потоковом режиме, в поддерживаемый формат PCM. Когда мы развертываем этот код веб-сокета на Python и пытаемся подключиться к нему через нашего клиента.
Транскрипция происходит примерно 8 минут 45 секунд, после чего появляется эта ошибка и внезапно транскрипция прекращается.< /p>
Есть ли какие-либо настройки, которые мы должны сделать с нашей стороны, чтобы избежать этого? Я также предоставляю снимок экрана с ошибкой.
Снимок экрана с ошибкой< /p>
Ожидаемое поведение
Транскрибировать звук в текст в течение всего времени, поддерживаемого Azure, — 240 минут
Код
import websockets
import asyncio
import uuid
import datetime
from app.utils.index import get_url_params
from app.service.api_service import verify_token
from app.service.logs_service import logger
from queue import Queue
from dataclasses import dataclass
import os
import json
import azure.cognitiveservices.speech as speechsdk
from app.service.audio_processor_service import save_audio_file
from app.service.azure_speech_service import (
create_conversation_transcriber,
)
import shutil
from dotenv import load_dotenv, find_dotenv

import azure.cognitiveservices.speech as speechsdk

def generate_speech_config(self):
try:
speech_config = speechsdk.SpeechConfig(
subscription=self.SPEECH_KEY,
region=self.SERVICE_REGION,
)
speech_config.set_property(
speechsdk.PropertyId.Speech_LogFilename,
f"results/{self.connections['uuid']}/azure_speech_sdk.log",
)
speech_config.enable_audio_logging()
speech_config.set_property(
property_id=speechsdk.PropertyId.SpeechServiceConnection_LanguageIdMode,
value="Continuous",
)
speech_config.set_property_by_name("maxSpeakerCount", str(8))

speech_config.request_word_level_timestamps()

print("Speech config generated")
return speech_config
except Exception as e:
print(f"Error generating speech config: {e}")

def create_conversation_transcriber(self):
try:
audio_stream_format = speechsdk.audio.AudioStreamFormat(
compressed_stream_format=speechsdk.audio.AudioStreamContainerFormat.ANY
)
push_stream = speechsdk.audio.PushAudioInputStream(
stream_format=audio_stream_format
)

audio_config = speechsdk.audio.AudioConfig(stream=self.push_stream)

speech_config = self.generate_speech_config()

conversation_transcriber = speechsdk.transcription.ConversationTranscriber(
speech_config=speech_config,
audio_config=audio_config,
auto_detect_source_language_config=speechsdk.AutoDetectSourceLanguageConfig(
languages=["en-US", "es-ES"]
),
)

conversation_transcriber.transcribing.connect(lambda evt: print(evt.result.text))
conversation_transcriber.session_stopped.connect(lambda evt: print("Session stopped"))
conversation_transcriber.canceled.connect(lambda evt: print("Canceled:", evt.reason))
conversation_transcriber.transcribed.connect(lambda evt: print(evt.result.text))
conversation_transcriber.session_started.connect(lambda evt: print("Session started"))

print("Conversation Transcriber Initialized - azure_speech_service.py")

return conversation_transcriber, push_stream

except Exception as e:
print(f"Error in create_conversation_transcriber: {e}")

@dataclass
class CONNECTIONS:
connections = {}

load_dotenv(find_dotenv())

async def send_transcribed_messages_async(uuid):
try:

while True:

msg_queue = CONNECTIONS.connections[uuid]["transcribed_queue"]
while not msg_queue.empty():
msg = msg_queue.get()
websocket = CONNECTIONS.connections[uuid]["websocket"]
if websocket.open:
await websocket.send(json.dumps(msg))
logger.info(f"Message sent to: {uuid}")
else:
logger.warning("WebSocket is closed, unable to send message.")
break
await asyncio.sleep(0.01) # Prevent tight loop
except Exception as e:
logger.error(f"Error sending transcribed message: {e}")

async def send_transcribing_messages_async(uuid):
try:

while True:

msg_queue = CONNECTIONS.connections[uuid]["transcribing_queue"]
while not msg_queue.empty():
msg = msg_queue.get()
websocket = CONNECTIONS.connections[uuid]["websocket"]
if websocket.open:
await websocket.send(json.dumps(msg))
logger.info(f"Message sent to: {uuid}")
else:
logger.warning("WebSocket is closed, unable to send message.")
break
await asyncio.sleep(0.01) # Prevent tight loop
except Exception as e:
logger.error(f"Error sending transcribing message: {e}")

async def receive_audio(uuid, path):

audio_queue = Queue(maxsize=0)

try:

conversation_transcriber, push_stream = create_conversation_transcriber(
CONNECTIONS.connections[uuid]
)

logger.info("Conversation transcriber initialized - socket_service.py")

# Start continuous recognition
conversation_transcriber.start_transcribing_async().get()

while True:
# Receive audio data from the WebSocket
websocket = CONNECTIONS.connections[uuid]["websocket"]
data = await websocket.recv()

if data:
audio_queue.put_nowait(data)

while not audio_queue.empty():
chunk = audio_queue.get()

CONNECTIONS.connections[uuid]["audio_buffer"] += chunk
push_stream.write(chunk)

except websockets.exceptions.ConnectionClosed as e:
logger.info("Connection closed")
logger.info(e)
conversation_transcriber.stop_transcribing_async().get()
push_stream.close()

except Exception as e:
logger.error(f"Error in receive_audio: {e}")

finally:

conversation_transcriber.stop_transcribing_async().get()
shutil.rmtree(f"results/{uuid}")

async def on_connect(websocket, path):

logger.info(path)

params_dict = get_url_params(path.split("/")[-1])

token_id = params_dict["token"]
encounter_id = params_dict["encounter_id"]

try:
resp = verify_token(token_id)
res = resp.json()

if resp.status_code == 200:
logger.info("JWT Token verification successful")
else:
raise Exception(f"JWT Token verification failed - {resp.status_code}")

UID = str(uuid.uuid4())

os.makedirs("results", exist_ok=True)
os.makedirs(f"results/{UID}", exist_ok=True)

timestamp = f"{datetime.datetime.now()}".replace(":", "_").replace("-", "_")
fname = f"{encounter_id}_{timestamp}"

CONNECTIONS.connections[UID] = {
"websocket": websocket,
"transcribed_queue": Queue(maxsize=0),
"transcribing_queue": Queue(maxsize=0),
"filekey": f"{encounter_id}/{timestamp}",
"fname": fname,
"json_list": [],
"audio_buffer": b"",
"organizationId": res["organizationId"],
"token": token_id,
"encounter_id": encounter_id,
"file_obj": open(f"results/{UID}/{fname}.txt", "w"),
"log_files": ["Logs/azure_speech_sdk.log", "Logs/logfile.log"],
"uuid": UID,
}

if res["organizationId"] is None:
raise Exception("Organization ID not found")

receive_audio_task = asyncio.create_task(receive_audio(UID, path))
send_transcribed_messages_task = asyncio.create_task(
send_transcribed_messages_async(UID)
)
send_transcribing_messages_task = asyncio.create_task(
send_transcribing_messages_async(UID)
)
await asyncio.gather(
receive_audio_task,
send_transcribed_messages_task,
send_transcribing_messages_task,
)

except Exception as e:
logger.info(f"Error in on_connect - {e}")
CONNECTIONS.connections.pop(UID)

async def run_websocket_server():
start_server = websockets.serve(on_connect, "0.0.0.0", 8000, ping_interval=None)
await start_server # This will ensure the WebSocket server starts



Подробнее здесь: https://stackoverflow.com/questions/790 ... topped-rea