Spaces:
Sleeping
Sleeping
#!pip install torch | |
#!pip install noisereduce | |
#!pip install scipy | |
import requests | |
import base64 | |
import numpy as np | |
from scipy.io.wavfile import read, write | |
#import noisereduce as nr | |
import nltk | |
import struct | |
test=False | |
# Define sentence split length | |
SENTENCE_SPLIT_LENGTH = 400 | |
##["en","es","fr","de","it","pt","pl","tr","ru","nl","cs","ar","zh-cn","ja"] | |
def detect_language(sentence): | |
url = "https://ruslanmv-hf-llm-api-collection.hf.space/detect" | |
data = {"input_text": sentence} | |
headers = {"Accept": "application/json", "Content-Type": "application/json"} | |
response = requests.post(url, headers=headers, json=data) | |
if response.status_code == 200: | |
try: | |
response_json = response.json() | |
language = response_json.get("lang") # Assuming "lang" is the key | |
return language | |
except JSONDecodeError: | |
print("Error: Invalid JSON response from the language detection API.") | |
else: | |
print(f"Error: Language detection API call failed with status code {response.status_code}") | |
return None # Fallback if API calls fail | |
def split_sentences(text, max_len): | |
# Apply custom rules to enforce sentence breaks with double punctuation | |
text = re.sub(r"(\s*\.{2})\s*", r".\1 ", text) # for '..' | |
text = re.sub(r"(\s*\!{2})\s*", r"!\1 ", text) # for '!!' | |
# Use NLTK to split into sentences | |
sentences = nltk.sent_tokenize(text) | |
# Then check if each sentence is greater than max_len, if so, use textwrap to split it | |
sentence_list = [] | |
for sent in sentences: | |
if len(sent) > max_len: | |
wrapped = textwrap.wrap(sent, max_len, break_long_words=True) | |
sentence_list.extend(wrapped) | |
else: | |
sentence_list.append(sent) | |
return sentence_list | |
def get_voice_streaming2(sentence, language): | |
"""Makes a POST request to the text-to-speech API and yields audio chunks.""" | |
url = "https://ruslanmv-hf-llm-api-collection.hf.space/tts" | |
data = {"input_text": sentence, "from_language": language} | |
headers = {"Accept": "application/json", "Content-Type": "application/json"} | |
response = requests.post(url, headers=headers, json=data) | |
return response | |
def pcm_to_wav2(pcm_data, sample_rate=24000, channels=1, bit_depth=16): | |
if pcm_data.startswith(b"RIFF"): | |
return pcm_data | |
fmt_subchunk_size = 16 | |
data_subchunk_size = len(pcm_data) | |
chunk_size = 4 + (8 + fmt_subchunk_size) + (8 + data_subchunk_size) | |
wav_header = struct.pack('<4sI4s', b'RIFF', chunk_size, b'WAVE') | |
fmt_subchunk = struct.pack('<4sIHHIIHH', | |
b'fmt ', fmt_subchunk_size, 1, channels, | |
sample_rate, sample_rate * channels * bit_depth // 8, | |
channels * bit_depth // 8, bit_depth) | |
data_subchunk = struct.pack('<4sI', b'data', data_subchunk_size) | |
return wav_header + fmt_subchunk + data_subchunk + pcm_data | |
import base64 | |
import re | |
def generate_speech_from_history2(history, chatbot_role, sentence): | |
""" | |
Generates speech audio from a given sentence, performing necessary preprocessing. | |
Args: | |
history (list): Conversation history. | |
chatbot_role (str): Role of the chatbot. | |
sentence (str): The sentence to be converted to speech. | |
Returns: | |
list: A list of dictionaries containing text and audio (base64 encoded) for each sentence fragment. | |
""" | |
language = "autodetect" | |
if len(sentence) == 0: | |
print("EMPTY SENTENCE") | |
return | |
# Preprocessing steps: | |
# - Remove special prompt token (</s>) | |
sentence = sentence.replace("</s>", "") | |
# - Remove code sections (enclosed in triple backticks) | |
sentence = re.sub("`.*`", "", sentence, flags=re.DOTALL) | |
# - Remove inline code fragments (backticks) | |
sentence = re.sub("`.*`", "", sentence, flags=re.DOTALL) | |
# - Remove content within parentheses | |
sentence = re.sub("\(.*\)", "", sentence, flags=re.DOTALL) | |
# - Remove remaining triple backticks | |
sentence = sentence.replace("```", "") | |
# - Replace ellipses with spaces | |
sentence = sentence.replace("...", " ") | |
# - Replace parentheses with spaces | |
sentence = sentence.replace("(", " ") | |
sentence = sentence.replace(")", " ") | |
# - Remove assistant tag | |
sentence = sentence.replace("<|assistant|>","") | |
if len(sentence) == 0: | |
print("EMPTY SENTENCE after processing") | |
return | |
# - Handle punctuation at the end of sentences | |
sentence = re.sub("([^\x00-\x7F]|\w)([\.。?!]+)", r"\1 \2", sentence) | |
print("Sentence for speech:", sentence) | |
results = [] | |
try: | |
if len(sentence) < SENTENCE_SPLIT_LENGTH: | |
sentence_list = [sentence] | |
else: | |
# Split longer sentences (implement your preferred split method) | |
sentence_list = split_sentences(sentence, SENTENCE_SPLIT_LENGTH) | |
print("detected sentences:", sentence_list) | |
for sentence in sentence_list: | |
print("- sentence =", sentence) | |
if any(c.isalnum() for c in sentence): | |
if language == "autodetect": | |
language = detect_language(sentence) # Detect language on first call | |
print("language",language) | |
audio_stream = get_voice_streaming2(sentence, language) | |
if audio_stream is not None: | |
sentence_wav_bytestream = b"" | |
# Process audio chunks | |
for chunk in audio_stream: | |
if chunk is not None: | |
sentence_wav_bytestream += chunk | |
# Encode WAV to base64 | |
base64_audio = base64.b64encode(pcm_to_wav2(sentence_wav_bytestream)).decode('utf8') | |
print("base64_audio",base64_audio[:10]) | |
results.append({ "text": sentence, "audio": base64_audio }) | |
else: | |
# Handle the case where the audio stream is None (e.g., silent response) | |
results.append({ "text": sentence, "audio": "" }) | |
except RuntimeError as e: | |
if "device-side assert" in str(e): | |
# cannot do anything , need to restart | |
print( | |
f"Exit due to: Unrecoverable exception caused by prompt:{sentence}", | |
flush=True, | |
) | |
#This error is unrecoverable need to restart space | |
#api.restart_space(repo_id=repo_id) | |
else: | |
print("RuntimeError: non device-side assert error:", str(e)) | |
raise e | |
return results | |
if test: | |
# Example usage | |
history = [] | |
chatbot_role = "assistant" | |
sentence = "Hello, how can I help you?" | |
result = generate_speech_from_history2(history, chatbot_role, sentence) | |
print(result) |