wehub-resource-sync
699 lines
27 KiB
Python
699 lines
27 KiB
Python
# Copyright 2023 LiveKit, Inc.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import weakref
|
|
from collections.abc import Sequence
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import aiohttp
|
|
|
|
from livekit import rtc
|
|
from livekit.agents import (
|
|
DEFAULT_API_CONNECT_OPTIONS,
|
|
APIConnectionError,
|
|
APIConnectOptions,
|
|
APIStatusError,
|
|
LanguageCode,
|
|
stt,
|
|
utils,
|
|
)
|
|
from livekit.agents.types import (
|
|
NOT_GIVEN,
|
|
NotGivenOr,
|
|
)
|
|
from livekit.agents.utils import AudioBuffer, is_given
|
|
from livekit.agents.voice.io import TimedString
|
|
|
|
from ._utils import PeriodicCollector, _to_deepgram_url
|
|
from .log import logger
|
|
from .models import V2Models
|
|
|
|
|
|
@dataclass
|
|
class STTOptions:
|
|
model: V2Models | str
|
|
sample_rate: int
|
|
keyterm: str | Sequence[str]
|
|
endpoint_url: str
|
|
language: str = "en"
|
|
eager_eot_threshold: NotGivenOr[float] = NOT_GIVEN
|
|
eot_threshold: NotGivenOr[float] = NOT_GIVEN
|
|
eot_timeout_ms: NotGivenOr[int] = NOT_GIVEN
|
|
mip_opt_out: bool = False
|
|
tags: NotGivenOr[list[str]] = NOT_GIVEN
|
|
language_hint: NotGivenOr[list[str]] = NOT_GIVEN
|
|
|
|
|
|
class STTv2(stt.STT):
|
|
def __init__(
|
|
self,
|
|
*,
|
|
model: V2Models | str = "flux-general-en",
|
|
sample_rate: int = 16000,
|
|
eager_eot_threshold: NotGivenOr[float] = NOT_GIVEN,
|
|
eot_threshold: NotGivenOr[float] = NOT_GIVEN,
|
|
eot_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
|
|
keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN,
|
|
tags: NotGivenOr[list[str]] = NOT_GIVEN,
|
|
language_hint: NotGivenOr[list[str]] = NOT_GIVEN,
|
|
api_key: NotGivenOr[str] = NOT_GIVEN,
|
|
http_session: aiohttp.ClientSession | None = None,
|
|
base_url: str = "wss://api.deepgram.com/v2/listen",
|
|
mip_opt_out: bool = False,
|
|
# deprecated
|
|
keyterms: NotGivenOr[list[str]] = NOT_GIVEN,
|
|
) -> None:
|
|
"""Create a new instance of Deepgram STT.
|
|
|
|
Args:
|
|
model: The Deepgram model to use for speech recognition. Defaults to "flux-general-en".
|
|
sample_rate: The sample rate of the audio in Hz. Defaults to 16000.
|
|
eager_eot_threshold: The threshold for eager end of turn to enable preemptive generation. Disabled by default. Set to 0.3-0.9 to enable preemptive generation.
|
|
eot_threshold: The threshold for end of speech detection, ranges 0.5-0.9. Defaults to 0.7. If using eager_eot_threshold, set this higher to allow a higher eager value.
|
|
eot_timeout_ms: The timeout for end of speech detection. Defaults to 3000.
|
|
keyterm: str or list of str of key terms to improve recognition accuracy. Defaults to None.
|
|
tags: List of tags to add to the requests for usage reporting. Defaults to NOT_GIVEN.
|
|
language_hint: List of str of language hints to bias the model for improved accuracy. Only usable with `flux-general-multi`. Defaults to NOT_GIVEN.
|
|
api_key: Your Deepgram API key. If not provided, will look for DEEPGRAM_API_KEY environment variable.
|
|
http_session: Optional aiohttp ClientSession to use for requests.
|
|
base_url: The base URL for Deepgram API. Defaults to "https://api.deepgram.com/v1/listen".
|
|
mip_opt_out: Whether to take part in the model improvement program
|
|
|
|
Raises:
|
|
ValueError: If no API key is provided or found in environment variables.
|
|
|
|
Note:
|
|
The api_key must be set either through the constructor argument or by setting
|
|
the DEEPGRAM_API_KEY environmental variable.
|
|
""" # noqa: E501
|
|
|
|
super().__init__(
|
|
capabilities=stt.STTCapabilities(
|
|
streaming=True,
|
|
interim_results=True,
|
|
aligned_transcript="word",
|
|
offline_recognize=False,
|
|
keyterms=True,
|
|
)
|
|
)
|
|
|
|
deepgram_api_key = api_key if is_given(api_key) else os.environ.get("DEEPGRAM_API_KEY")
|
|
if not deepgram_api_key:
|
|
raise ValueError("Deepgram API key is required")
|
|
self._api_key = deepgram_api_key
|
|
|
|
if is_given(keyterms):
|
|
logger.warning(
|
|
"`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API."
|
|
)
|
|
keyterm = keyterms
|
|
|
|
if is_given(eager_eot_threshold):
|
|
effective_eot = eot_threshold if is_given(eot_threshold) else 0.7
|
|
if eager_eot_threshold > effective_eot:
|
|
raise ValueError(
|
|
f"eager_eot_threshold ({eager_eot_threshold}) must be less than or equal to eot_threshold "
|
|
f"({effective_eot}); increase eot_threshold (max 0.9) to use a higher eager value"
|
|
)
|
|
if language_hint and model != "flux-general-multi":
|
|
logger.warning(
|
|
"`language_hint` is only supported by `flux-general-multi` and will be ignored for model '%s'",
|
|
model,
|
|
)
|
|
|
|
self._opts = STTOptions(
|
|
model=model,
|
|
sample_rate=sample_rate,
|
|
keyterm=([keyterm] if isinstance(keyterm, str) else list(keyterm))
|
|
if is_given(keyterm)
|
|
else [],
|
|
mip_opt_out=mip_opt_out,
|
|
tags=_validate_tags(tags) if is_given(tags) else [],
|
|
language_hint=language_hint if is_given(language_hint) else [],
|
|
eager_eot_threshold=eager_eot_threshold,
|
|
eot_threshold=eot_threshold,
|
|
eot_timeout_ms=eot_timeout_ms,
|
|
endpoint_url=base_url,
|
|
)
|
|
# user keyterms; _opts.keyterm holds the effective set (user + session)
|
|
self._user_keyterm: list[str] = list(self._opts.keyterm)
|
|
self._session_keyterms: list[str] = []
|
|
self._session = http_session
|
|
self._streams = weakref.WeakSet[SpeechStreamv2]()
|
|
|
|
def _ensure_session(self) -> aiohttp.ClientSession:
|
|
if not self._session:
|
|
self._session = utils.http_context.http_session()
|
|
|
|
return self._session
|
|
|
|
async def _recognize_impl(
|
|
self,
|
|
buffer: AudioBuffer,
|
|
*,
|
|
language: NotGivenOr[str] = NOT_GIVEN,
|
|
conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
|
|
) -> stt.SpeechEvent:
|
|
raise NotImplementedError(
|
|
"V2 API does not support non-streaming recognize. Use with a StreamAdapter"
|
|
)
|
|
|
|
@property
|
|
def model(self) -> str:
|
|
return self._opts.model
|
|
|
|
@property
|
|
def provider(self) -> str:
|
|
return "Deepgram"
|
|
|
|
def stream(
|
|
self,
|
|
*,
|
|
language: NotGivenOr[str] = NOT_GIVEN,
|
|
conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS,
|
|
) -> SpeechStreamv2:
|
|
stream = SpeechStreamv2(
|
|
stt=self,
|
|
conn_options=conn_options,
|
|
opts=self._opts,
|
|
api_key=self._api_key,
|
|
http_session=self._ensure_session(),
|
|
base_url=self._opts.endpoint_url,
|
|
)
|
|
self._streams.add(stream)
|
|
return stream
|
|
|
|
def update_options(
|
|
self,
|
|
*,
|
|
model: NotGivenOr[V2Models | str] = NOT_GIVEN,
|
|
sample_rate: NotGivenOr[int] = NOT_GIVEN,
|
|
eager_eot_threshold: NotGivenOr[float] = NOT_GIVEN,
|
|
eot_threshold: NotGivenOr[float] = NOT_GIVEN,
|
|
eot_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
|
|
keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN,
|
|
mip_opt_out: NotGivenOr[bool] = NOT_GIVEN,
|
|
tags: NotGivenOr[list[str]] = NOT_GIVEN,
|
|
language_hint: NotGivenOr[list[str]] = NOT_GIVEN,
|
|
endpoint_url: NotGivenOr[str] = NOT_GIVEN,
|
|
# deprecated
|
|
keyterms: NotGivenOr[list[str]] = NOT_GIVEN,
|
|
) -> None:
|
|
effective_eager = (
|
|
eager_eot_threshold if is_given(eager_eot_threshold) else self._opts.eager_eot_threshold
|
|
)
|
|
effective_eot = (
|
|
eot_threshold
|
|
if is_given(eot_threshold)
|
|
else (self._opts.eot_threshold if is_given(self._opts.eot_threshold) else 0.7)
|
|
)
|
|
if is_given(effective_eager) and effective_eager > effective_eot:
|
|
raise ValueError(
|
|
f"eager_eot_threshold ({effective_eager}) must be less than or equal to eot_threshold ({effective_eot})"
|
|
)
|
|
if is_given(model):
|
|
self._opts.model = model
|
|
if is_given(sample_rate):
|
|
self._opts.sample_rate = sample_rate
|
|
if is_given(eot_threshold):
|
|
self._opts.eot_threshold = eot_threshold
|
|
if is_given(eot_timeout_ms):
|
|
self._opts.eot_timeout_ms = eot_timeout_ms
|
|
if is_given(keyterms):
|
|
logger.warning(
|
|
"`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API."
|
|
)
|
|
keyterm = keyterms
|
|
if is_given(keyterm):
|
|
self._user_keyterm = [keyterm] if isinstance(keyterm, str) else list(keyterm)
|
|
keyterm = list(dict.fromkeys([*self._user_keyterm, *self._session_keyterms]))
|
|
self._opts.keyterm = keyterm
|
|
if is_given(mip_opt_out):
|
|
self._opts.mip_opt_out = mip_opt_out
|
|
if is_given(tags):
|
|
self._opts.tags = _validate_tags(tags)
|
|
if is_given(language_hint):
|
|
self._opts.language_hint = language_hint
|
|
if language_hint and self._opts.model != "flux-general-multi":
|
|
logger.warning(
|
|
"`language_hint` is only supported by `flux-general-multi` and will be ignored for model '%s'",
|
|
self._opts.model,
|
|
)
|
|
if is_given(endpoint_url):
|
|
self._opts.endpoint_url = endpoint_url
|
|
if is_given(eager_eot_threshold):
|
|
self._opts.eager_eot_threshold = eager_eot_threshold
|
|
|
|
for stream in self._streams:
|
|
stream.update_options(
|
|
model=model,
|
|
sample_rate=sample_rate,
|
|
eot_threshold=eot_threshold,
|
|
eot_timeout_ms=eot_timeout_ms,
|
|
keyterm=keyterm,
|
|
mip_opt_out=mip_opt_out,
|
|
endpoint_url=endpoint_url,
|
|
tags=tags,
|
|
language_hint=language_hint,
|
|
eager_eot_threshold=eager_eot_threshold,
|
|
)
|
|
|
|
def _update_session_keyterms(self, keyterms: list[str]) -> None:
|
|
if keyterms == self._session_keyterms:
|
|
return
|
|
self._session_keyterms = list(keyterms)
|
|
merged = list(dict.fromkeys([*self._user_keyterm, *keyterms]))
|
|
self._opts.keyterm = merged
|
|
for stream in self._streams:
|
|
# tuned in-band, safe to apply mid-utterance
|
|
stream.update_options(keyterm=merged)
|
|
|
|
|
|
class SpeechStreamv2(stt.SpeechStream):
|
|
# _KEEPALIVE_MSG: str = json.dumps({"type": "KeepAlive"})
|
|
_CLOSE_MSG: str = json.dumps({"type": "CloseStream"})
|
|
# _FINALIZE_MSG: str = json.dumps({"type": "Finalize"})
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
stt: STTv2,
|
|
opts: STTOptions,
|
|
conn_options: APIConnectOptions,
|
|
api_key: str,
|
|
http_session: aiohttp.ClientSession,
|
|
base_url: str,
|
|
) -> None:
|
|
super().__init__(stt=stt, conn_options=conn_options, sample_rate=opts.sample_rate)
|
|
self._opts = opts
|
|
self._api_key = api_key
|
|
self._session = http_session
|
|
self._opts.endpoint_url = base_url
|
|
self._speaking = False
|
|
self._audio_duration_collector = PeriodicCollector(
|
|
callback=self._on_audio_duration_report,
|
|
duration=5.0,
|
|
)
|
|
|
|
self._request_id = ""
|
|
self._reconnect_event = asyncio.Event()
|
|
# active connection for in-band Configure updates; None while disconnected
|
|
self._ws: aiohttp.ClientWebSocketResponse | None = None
|
|
self._reconfigure_atask: asyncio.Task[None] | None = None
|
|
|
|
def update_options(
|
|
self,
|
|
*,
|
|
model: NotGivenOr[V2Models | str] = NOT_GIVEN,
|
|
sample_rate: NotGivenOr[int] = NOT_GIVEN,
|
|
eot_threshold: NotGivenOr[float] = NOT_GIVEN,
|
|
eot_timeout_ms: NotGivenOr[int] = NOT_GIVEN,
|
|
keyterm: NotGivenOr[str | list[str]] = NOT_GIVEN,
|
|
mip_opt_out: NotGivenOr[bool] = NOT_GIVEN,
|
|
tags: NotGivenOr[list[str]] = NOT_GIVEN,
|
|
language_hint: NotGivenOr[list[str]] = NOT_GIVEN,
|
|
endpoint_url: NotGivenOr[str] = NOT_GIVEN,
|
|
eager_eot_threshold: NotGivenOr[float] = NOT_GIVEN,
|
|
# deprecated
|
|
keyterms: NotGivenOr[list[str]] = NOT_GIVEN,
|
|
) -> None:
|
|
if is_given(model):
|
|
self._opts.model = model
|
|
if is_given(sample_rate):
|
|
self._opts.sample_rate = sample_rate
|
|
if is_given(eot_threshold):
|
|
self._opts.eot_threshold = eot_threshold
|
|
if is_given(eot_timeout_ms):
|
|
self._opts.eot_timeout_ms = eot_timeout_ms
|
|
if is_given(keyterms):
|
|
logger.warning(
|
|
"`keyterms` is deprecated, use `keyterm` instead for consistency with Deepgram API."
|
|
)
|
|
keyterm = keyterms
|
|
if is_given(keyterm):
|
|
self._opts.keyterm = keyterm
|
|
if is_given(mip_opt_out):
|
|
self._opts.mip_opt_out = mip_opt_out
|
|
if is_given(tags):
|
|
self._opts.tags = _validate_tags(tags)
|
|
if is_given(language_hint):
|
|
self._opts.language_hint = language_hint
|
|
if is_given(endpoint_url):
|
|
self._opts.endpoint_url = endpoint_url
|
|
if is_given(eager_eot_threshold):
|
|
self._opts.eager_eot_threshold = eager_eot_threshold
|
|
|
|
# these only take effect on a fresh connection
|
|
needs_reconnect = any(
|
|
is_given(opt) for opt in (model, sample_rate, mip_opt_out, tags, endpoint_url)
|
|
)
|
|
if needs_reconnect:
|
|
# reconnect carries the latest options
|
|
self._reconnect_event.set()
|
|
return
|
|
|
|
# send only changed fields; Flux keeps omitted ones unchanged
|
|
# https://developers.deepgram.com/docs/flux/configure
|
|
thresholds: dict[str, Any] = {}
|
|
if is_given(eager_eot_threshold):
|
|
thresholds["eager_eot_threshold"] = eager_eot_threshold
|
|
if is_given(eot_threshold):
|
|
thresholds["eot_threshold"] = eot_threshold
|
|
if is_given(eot_timeout_ms):
|
|
thresholds["eot_timeout_ms"] = eot_timeout_ms
|
|
|
|
changed_options: dict[str, Any] = {}
|
|
if thresholds:
|
|
changed_options["thresholds"] = thresholds
|
|
if is_given(keyterm):
|
|
# keyterms replaces the whole list, so send the full effective set
|
|
changed_options["keyterms"] = self._opts.keyterm
|
|
if is_given(language_hint):
|
|
changed_options["language_hints"] = self._opts.language_hint
|
|
|
|
if changed_options:
|
|
# chain off the previous send so deltas reach the server in order
|
|
self._reconfigure_atask = asyncio.create_task(
|
|
self._send_configure(changed_options, self._reconfigure_atask)
|
|
)
|
|
|
|
async def _send_configure(
|
|
self, options: dict[str, Any], prev: asyncio.Task[None] | None
|
|
) -> None:
|
|
if prev is not None:
|
|
await asyncio.gather(prev, return_exceptions=True)
|
|
|
|
ws = self._ws
|
|
if ws is None or ws.closed:
|
|
# not connected; next connection carries the latest options
|
|
return
|
|
try:
|
|
await ws.send_str(json.dumps({"type": "Configure", **options}))
|
|
except Exception:
|
|
# closing; next connection carries the latest options
|
|
logger.debug("failed to send Configure to deepgram")
|
|
|
|
async def _run(self) -> None:
|
|
closing_ws = False
|
|
|
|
# async def keepalive_task(ws: aiohttp.ClientWebSocketResponse) -> None:
|
|
# # if we want to keep the connection alive even if no audio is sent,
|
|
# # Deepgram expects a keepalive message.
|
|
# # https://developers.deepgram.com/reference/listen-live#stream-keepalive
|
|
# try:
|
|
# while True:
|
|
# await ws.send_str(SpeechStream._KEEPALIVE_MSG)
|
|
# await asyncio.sleep(5)
|
|
# except Exception:
|
|
# return
|
|
|
|
@utils.log_exceptions(logger=logger)
|
|
async def send_task(ws: aiohttp.ClientWebSocketResponse) -> None:
|
|
nonlocal closing_ws
|
|
|
|
# forward audio to deepgram in chunks of 50ms
|
|
samples_50ms = self._opts.sample_rate // 20
|
|
audio_bstream = utils.audio.AudioByteStream(
|
|
sample_rate=self._opts.sample_rate,
|
|
num_channels=1,
|
|
samples_per_channel=samples_50ms,
|
|
)
|
|
|
|
has_ended = False
|
|
async for data in self._input_ch:
|
|
frames: list[rtc.AudioFrame] = []
|
|
if isinstance(data, rtc.AudioFrame):
|
|
frames.extend(audio_bstream.write(data.data.tobytes()))
|
|
elif isinstance(data, self._FlushSentinel):
|
|
frames.extend(audio_bstream.flush())
|
|
has_ended = True
|
|
|
|
for frame in frames:
|
|
self._audio_duration_collector.push(frame.duration)
|
|
await ws.send_bytes(frame.data.tobytes())
|
|
|
|
if has_ended:
|
|
self._audio_duration_collector.flush()
|
|
has_ended = False
|
|
|
|
# tell deepgram we are done sending audio/inputs
|
|
closing_ws = True
|
|
await ws.send_str(SpeechStreamv2._CLOSE_MSG)
|
|
|
|
@utils.log_exceptions(logger=logger)
|
|
async def recv_task(ws: aiohttp.ClientWebSocketResponse) -> None:
|
|
nonlocal closing_ws
|
|
while True:
|
|
msg = await ws.receive()
|
|
if msg.type in (
|
|
aiohttp.WSMsgType.CLOSED,
|
|
aiohttp.WSMsgType.CLOSE,
|
|
aiohttp.WSMsgType.CLOSING,
|
|
):
|
|
# close is expected, see SpeechStream.aclose
|
|
# or when the agent session ends, the http session is closed
|
|
if closing_ws or self._session.closed:
|
|
return
|
|
|
|
# this will trigger a reconnection, see the _run loop
|
|
raise APIStatusError(
|
|
message="deepgram connection closed unexpectedly",
|
|
status_code=ws.close_code or -1,
|
|
body=f"{msg.data=} {msg.extra=}",
|
|
)
|
|
|
|
if msg.type != aiohttp.WSMsgType.TEXT:
|
|
logger.warning("unexpected deepgram message type %s", msg.type)
|
|
continue
|
|
|
|
try:
|
|
self._process_stream_event(json.loads(msg.data))
|
|
except Exception:
|
|
logger.exception("failed to process deepgram message")
|
|
|
|
ws: aiohttp.ClientWebSocketResponse | None = None
|
|
|
|
while True:
|
|
try:
|
|
ws = await self._connect_ws()
|
|
# expose the connection for in-band Configure updates
|
|
self._ws = ws
|
|
tasks = [
|
|
asyncio.create_task(send_task(ws)),
|
|
asyncio.create_task(recv_task(ws)),
|
|
# asyncio.create_task(keepalive_task(ws)),
|
|
]
|
|
tasks_group = asyncio.gather(*tasks)
|
|
wait_reconnect_task = asyncio.create_task(self._reconnect_event.wait())
|
|
try:
|
|
done, _ = await asyncio.wait(
|
|
(tasks_group, wait_reconnect_task),
|
|
return_when=asyncio.FIRST_COMPLETED,
|
|
)
|
|
|
|
# propagate exceptions from completed tasks
|
|
for task in done:
|
|
if task != wait_reconnect_task:
|
|
task.result()
|
|
|
|
if wait_reconnect_task not in done:
|
|
break
|
|
|
|
self._reconnect_event.clear()
|
|
finally:
|
|
await utils.aio.gracefully_cancel(*tasks, wait_reconnect_task)
|
|
tasks_group.cancel()
|
|
tasks_group.exception() # retrieve the exception
|
|
finally:
|
|
self._ws = None
|
|
if self._reconfigure_atask is not None:
|
|
await utils.aio.gracefully_cancel(self._reconfigure_atask)
|
|
self._reconfigure_atask = None
|
|
if ws is not None:
|
|
await ws.close()
|
|
|
|
async def _connect_ws(self) -> aiohttp.ClientWebSocketResponse:
|
|
live_config: dict[str, Any] = {
|
|
"model": self._opts.model,
|
|
"sample_rate": self._opts.sample_rate,
|
|
"encoding": "linear16",
|
|
"mip_opt_out": self._opts.mip_opt_out,
|
|
}
|
|
|
|
if self._opts.eager_eot_threshold:
|
|
live_config["eager_eot_threshold"] = self._opts.eager_eot_threshold
|
|
|
|
if self._opts.eot_threshold:
|
|
live_config["eot_threshold"] = self._opts.eot_threshold
|
|
|
|
if self._opts.eot_timeout_ms:
|
|
live_config["eot_timeout_ms"] = self._opts.eot_timeout_ms
|
|
|
|
if self._opts.keyterm:
|
|
live_config["keyterm"] = self._opts.keyterm
|
|
|
|
if self._opts.tags:
|
|
live_config["tag"] = self._opts.tags
|
|
|
|
if self._opts.language_hint:
|
|
live_config["language_hint"] = self._opts.language_hint
|
|
|
|
try:
|
|
ws = await asyncio.wait_for(
|
|
self._session.ws_connect(
|
|
_to_deepgram_url(live_config, base_url=self._opts.endpoint_url, websocket=True),
|
|
headers={"Authorization": f"Token {self._api_key}"},
|
|
heartbeat=30.0,
|
|
),
|
|
self._conn_options.timeout,
|
|
)
|
|
ws_headers = {
|
|
k: v for k, v in ws._response.headers.items() if k.startswith("dg-") or k == "Date"
|
|
}
|
|
logger.debug(
|
|
"Established new Deepgram STT WebSocket connection:",
|
|
extra={"headers": ws_headers},
|
|
)
|
|
except (aiohttp.ClientConnectorError, asyncio.TimeoutError) as e:
|
|
raise APIConnectionError("failed to connect to deepgram") from e
|
|
return ws
|
|
|
|
def _on_audio_duration_report(self, duration: float) -> None:
|
|
usage_event = stt.SpeechEvent(
|
|
type=stt.SpeechEventType.RECOGNITION_USAGE,
|
|
request_id=self._request_id,
|
|
alternatives=[],
|
|
recognition_usage=stt.RecognitionUsage(audio_duration=duration),
|
|
)
|
|
self._event_ch.send_nowait(usage_event)
|
|
|
|
def _send_transcript_event(self, event_type: stt.SpeechEventType, data: dict) -> None:
|
|
alts = _parse_transcription(self._opts.language, data, self.start_time_offset)
|
|
if alts:
|
|
event = stt.SpeechEvent(
|
|
type=event_type,
|
|
request_id=self._request_id,
|
|
alternatives=alts,
|
|
)
|
|
self._event_ch.send_nowait(event)
|
|
|
|
def _process_stream_event(self, data: dict) -> None:
|
|
assert self._opts.language is not None
|
|
|
|
if request_id := data.get("request_id"):
|
|
self._request_id = request_id
|
|
|
|
if data["type"] == "TurnInfo":
|
|
event_type = data["event"]
|
|
|
|
if event_type == "StartOfTurn":
|
|
if self._speaking:
|
|
return
|
|
|
|
self._speaking = True
|
|
start_event = stt.SpeechEvent(type=stt.SpeechEventType.START_OF_SPEECH)
|
|
self._event_ch.send_nowait(start_event)
|
|
|
|
self._send_transcript_event(stt.SpeechEventType.INTERIM_TRANSCRIPT, data)
|
|
|
|
elif event_type == "Update":
|
|
if not self._speaking:
|
|
return
|
|
|
|
self._send_transcript_event(stt.SpeechEventType.INTERIM_TRANSCRIPT, data)
|
|
|
|
elif event_type == "EagerEndOfTurn":
|
|
# technically, a pause in speech is detected. for lifecycle purposes,
|
|
# we are assuming the user is still speaking and sending a preflight event to
|
|
# start preemptive synthesis.
|
|
if not self._speaking:
|
|
return
|
|
|
|
self._send_transcript_event(stt.SpeechEventType.PREFLIGHT_TRANSCRIPT, data)
|
|
|
|
elif event_type == "TurnResumed":
|
|
# sending interim transcript will abort eager end of turn
|
|
self._send_transcript_event(stt.SpeechEventType.INTERIM_TRANSCRIPT, data)
|
|
|
|
elif event_type == "EndOfTurn":
|
|
if not self._speaking:
|
|
return
|
|
|
|
self._speaking = False
|
|
|
|
self._send_transcript_event(stt.SpeechEventType.FINAL_TRANSCRIPT, data)
|
|
|
|
end_event = stt.SpeechEvent(type=stt.SpeechEventType.END_OF_SPEECH)
|
|
self._event_ch.send_nowait(end_event)
|
|
|
|
elif data["type"] == "ConfigureSuccess":
|
|
logger.debug("deepgram applied Configure update", extra={"data": data})
|
|
|
|
elif data["type"] == "ConfigureFailure":
|
|
logger.warning("deepgram rejected Configure update", extra={"data": data})
|
|
|
|
elif data["type"] == "Error":
|
|
logger.warning("deepgram sent an error", extra={"data": data})
|
|
desc = data.get("description") or "unknown error from deepgram"
|
|
code = -1
|
|
raise APIStatusError(message=desc, status_code=code)
|
|
|
|
|
|
def _parse_transcription(
|
|
language: str, data: dict[str, Any], start_time_offset: float
|
|
) -> list[stt.SpeechData]:
|
|
transcript = data.get("transcript")
|
|
words = data.get("words")
|
|
if not words:
|
|
return []
|
|
confidence = sum(word["confidence"] for word in words) / len(words) if words else 0
|
|
|
|
detected_languages = data.get("languages") or []
|
|
primary_language = (
|
|
LanguageCode(detected_languages[0]) if detected_languages else LanguageCode(language)
|
|
)
|
|
|
|
sd = stt.SpeechData(
|
|
language=primary_language,
|
|
start_time=data.get("audio_window_start", 0) + start_time_offset,
|
|
end_time=data.get("audio_window_end", 0) + start_time_offset,
|
|
confidence=confidence,
|
|
text=transcript or "",
|
|
source_languages=[LanguageCode(lang) for lang in detected_languages] or None,
|
|
words=[
|
|
TimedString(
|
|
text=word.get("word", ""),
|
|
start_time=word.get("start", 0) + start_time_offset,
|
|
end_time=word.get("end", 0) + start_time_offset,
|
|
confidence=word["confidence"],
|
|
start_time_offset=start_time_offset,
|
|
)
|
|
for word in words
|
|
],
|
|
)
|
|
return [sd]
|
|
|
|
|
|
def _validate_tags(tags: list[str]) -> list[str]:
|
|
for tag in tags:
|
|
if len(tag) > 128:
|
|
raise ValueError("tag must be no more than 128 characters")
|
|
return tags
|