Spaces:
Sleeping
Sleeping
File size: 6,118 Bytes
8362005 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 |
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
# This code originally written by Marmik Pandya (marmikcfc - github.com/marmikcfc)
import numpy as np
from typing import AsyncGenerator, List, Optional, Union
from loguru import logger
from pydantic import BaseModel
from pipecat.frames.frames import (
ErrorFrame,
Frame,
TTSAudioRawFrame,
TTSStartedFrame,
TTSStoppedFrame,
)
from pipecat.services.tts_service import TTSService
from pipecat.transcriptions.language import Language
# load Kokoro from kokoro-onnx
try:
from kokoro_onnx import Kokoro
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error(
"In order to use Kokoro, you need to `pip install kokoro-onnx`. Also, download the model files from the Kokoro repository."
)
raise Exception(f"Missing module: {e}")
def language_to_kokoro_language(language: Language) -> Optional[str]:
"""Convert pipecat Language to Kokoro language code."""
BASE_LANGUAGES = {
Language.EN: "en-us",
Language.FR: "fr-fr",
Language.IT: "it",
Language.JA: "ja",
Language.CMN: "cmn"
# Add more language mappings as supported by Kokoro
}
result = BASE_LANGUAGES.get(language)
# If not found in base languages, try to find the base language from a variant
if not result:
lang_str = str(language.value)
base_code = lang_str.split("-")[0].lower()
# Look up the base code in our supported languages
result = f"{base_code}-us" if base_code in ["en"] else None
return result
class KokoroTTSService(TTSService):
"""Text-to-Speech service using Kokoro for on-device TTS.
This service uses Kokoro to generate speech without requiring external API connections.
"""
class InputParams(BaseModel):
"""Configuration parameters for Kokoro TTS service."""
language: Optional[Language] = Language.EN
speed: Optional[float] = 1.0
def __init__(
self,
*,
model_path: str,
voices_path: str,
voice_id: str = "af_sarah",
sample_rate: Optional[int] = None,
params: InputParams = InputParams(),
**kwargs,
):
"""Initialize Kokoro TTS service.
Args:
model_path: Path to the Kokoro ONNX model file
voices_path: Path to the Kokoro voices file
voice_id: ID of the voice to use
sample_rate: Output audio sample rate
params: Additional configuration parameters
"""
super().__init__(sample_rate=sample_rate, **kwargs)
logger.info(f"Initializing Kokoro TTS service with model_path: {model_path} and voices_path: {voices_path}")
self._kokoro = Kokoro(model_path, voices_path)
logger.info(f"Kokoro initialized")
self._settings = {
"language": self.language_to_service_language(params.language)
if params.language
else "en-us",
"speed": params.speed,
}
self.set_voice(voice_id) # Presumably this sets self._voice_id
logger.info("Kokoro TTS service initialized")
def can_generate_metrics(self) -> bool:
return True
def language_to_service_language(self, language: Language) -> Optional[str]:
"""Convert pipecat language to Kokoro language code."""
return language_to_kokoro_language(language)
async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
"""Generate speech from text using Kokoro in a streaming fashion.
Args:
text: The text to convert to speech
Yields:
Frames containing audio data and status information.
"""
logger.debug(f"Generating TTS: [{text}]")
try:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
# Use Kokoro's streaming mode. The create_stream method is assumed to return
# an async generator that yields (samples, sample_rate) tuples, where samples is a numpy array.
logger.info(f"Creating stream")
stream = self._kokoro.create_stream(
text,
voice=self._voice_id,
speed=self._settings["speed"],
lang=self._settings["language"],
)
await self.start_tts_usage_metrics(text)
started = False
async for samples, sample_rate in stream:
if not started:
started = True
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
logger.info(f"Started streaming")
# Convert the float32 samples (assumed in the range [-1, 1]) to int16 PCM format
samples_int16 = (samples * 32767).astype(np.int16)
yield TTSAudioRawFrame(
audio=samples_int16.tobytes(),
sample_rate=sample_rate,
num_channels=1,
)
yield TTSStoppedFrame()
except Exception as e:
logger.error(f"{self} exception: {e}")
yield ErrorFrame(f"Error generating audio: {str(e)}") |