File size: 7,865 Bytes
d65b1bc c2de7e1 d65b1bc c2de7e1 d65b1bc e946d7d 2bec257 e946d7d d65b1bc 7ea89c2 d65b1bc cec2e17 d65b1bc c2de7e1 d65b1bc edb499f 7841cfa edb499f b923778 edb499f b923778 edb499f b923778 d65b1bc eba3479 d65b1bc eba3479 d65b1bc eba3479 d65b1bc eba3479 d65b1bc |
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 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 |
import asyncio
import aiohttp
import os, uuid
from collections import deque
import wave
import uuid
from pydub import AudioSegment
import wave
import struct
def concatenate_wave_files(input_file_paths, output_file_path):
"""
Concatenates multiple wave files and saves the result to a new file.
:param input_file_paths: A list of paths to the input wave files.
:param output_file_path: The path to the output wave file.
"""
# Check if input file paths are provided
if not input_file_paths:
raise ValueError("No input file paths provided.")
# Validate output file path
if not output_file_path:
raise ValueError("Output file path is empty.")
# Validate input file paths
for input_file_path in input_file_paths:
if not input_file_path:
raise ValueError("Empty input file path found.")
# Validate and get parameters from the first input file
with wave.open(input_file_paths[0], "rb") as input_file:
n_channels = input_file.getnchannels()
sampwidth = input_file.getsampwidth()
framerate = input_file.getframerate()
comptype = input_file.getcomptype()
compname = input_file.getcompname()
# Open the output file for writing
output_file = wave.open(output_file_path, "wb")
output_file.setnchannels(n_channels)
output_file.setsampwidth(sampwidth)
output_file.setframerate(framerate)
output_file.setcomptype(comptype, compname)
# Concatenate and write data from all input files to the output file
for input_file_path in input_file_paths:
with wave.open(input_file_path, "rb") as input_file:
output_file.writeframes(input_file.readframes(input_file.getnframes()))
# Close the output file
output_file.close()
print(
f"Successfully concatenated {len(input_file_paths)} files into {output_file_path}"
)
# # Example usage
# input_files = ["./tmp/" + i for i in os.listdir("./tmp")]
# output_file = "./concatenated_output.wav"
# concatenate_wave_files(input_files, output_file)
def concatenate_wav_files(input_files, file_directory):
print(input_files)
output_file = file_directory + str(uuid.uuid4()) + "final.wav"
# Initialize variables for output file
output = None
output_params = None
try:
# Open output file for writing
output = wave.open(output_file, "wb")
# Loop through input files
for input_file in input_files:
with wave.open(input_file, "rb") as input_wav:
# If this is the first input file, set output file parameters
if output_params is None:
output_params = input_wav.getparams()
output.setparams(output_params)
# Otherwise, ensure consistency of parameters
else:
pass
# if input_wav.getparams() != output_params:
# raise ValueError(
# "Input file parameters do not match output file parameters."
# )
# Read data from input file and write to output file
output.writeframes(input_wav.readframes(input_wav.getnframes()))
finally:
# Close output file
if output is not None:
output.close()
return (output_file,)
class Speak:
def __init__(self, api_url="https://yakova-embedding.hf.space", dir="./tmp"):
self.api_url = api_url
self.dir = dir
async def _make_transcript(self, links, text):
data = {"audio_url": links, "text": text}
tries = 0
max_retries = 10
while tries < max_retries:
try:
response_data = await self._make_request(
"post", "descript_transcript", json=data
)
if isinstance(response_data, dict):
return response_data
except Exception as e:
print(f"Attempt {tries + 1} failed: {e}")
tries += 1
await asyncio.sleep(1) # Adding a delay between retries
raise Exception("Max retries reached. Unable to get a valid response.")
async def _make_request(self, method, endpoint, json=None):
async with aiohttp.ClientSession() as session:
async with getattr(session, method)(
f"{self.api_url}/{endpoint}", json=json
) as response:
return await response.json()
async def say(self, text, speaker="Gabi"):
data = {"text": text, "speaker": speaker}
response_data = await self._make_request("post", "descript_tts", json=data)
tts_id = response_data["id"]
# Poll the status endpoint until the TTS is ready
while True:
status_data = await self._make_request(
"post", "descript_status", json={"id": tts_id}
)
print(status_data)
if "status" in status_data:
if status_data["status"] == "done":
audio_url = status_data["url"]
temp = await self.download_file(audio_url)
return audio_url, temp
else:
pass
await asyncio.sleep(1)
async def download_file(self, url):
filename = str(uuid.uuid4()) + ".wav"
os.makedirs(self.dir, exist_ok=True)
save_path = os.path.join(self.dir, filename)
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status == 200:
with open(save_path, "wb") as file:
while True:
chunk = await response.content.read(1024)
if not chunk:
break
file.write(chunk)
return save_path
async def download_file_with_aria2c(self, url):
filename = str(uuid.uuid4()) + ".wav"
os.makedirs(self.dir, exist_ok=True)
save_path = os.path.join(self.dir, filename)
print(url)
command = f"aria2c {url} -o {save_path}"
process = await asyncio.create_subprocess_shell(
command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
if process.returncode == 0:
print(f"File downloaded successfully to {save_path}")
else:
print(f"Failed to download file. Error: {stderr.decode()}")
return save_path
async def process_narrations(narrations):
speak = Speak()
tasks = deque()
results = []
files = []
async def process_task():
while tasks:
text = tasks.popleft()
result = await speak.say(text)
_, temp = result
results.append(result)
files.append(temp)
for narration in narrations:
tasks.append(narration)
if len(tasks) >= 2:
await asyncio.gather(*[process_task() for _ in range(2)])
# Process remaining tasks
await asyncio.gather(*[process_task() for _ in range(len(tasks))])
concatinated_file = concatenate_wav_files(files, speak.dir)
audio_file = AudioSegment.from_file(concatinated_file, format="wav")
duration_in_seconds = int(len(audio_file) / 1000)
return results, [concatinated_file, duration_in_seconds]
# duration = 0.23529411764705882
# # Example narrations
# narrations = [
# "Hello there ",
# "Hello there",
# "Hello there",
# "Hello there [space]",
# ]
# # Run the asyncio event loop
# async def main():
# results = await process_narrations(narrations)
# print("Results:", results)
# asyncio.run(main())
|