Updated to use new ffmpeg
This commit is contained in:
erew123 authored and GitHub committed
1 parent
aa794e458b
commit
231bc21fa7
1 file changed
+81
-87
@@ -1,87 +1,81 @@
|
|||||||
import numpy as np
|
import numpy as np
|
||||||
import re
|
import re
|
||||||
import unicodedata
|
import unicodedata
|
||||||
from fairseq import checkpoint_utils
|
from fairseq import checkpoint_utils
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
logging.getLogger("fairseq").setLevel(logging.WARNING)
|
logging.getLogger("fairseq").setLevel(logging.WARNING)
|
||||||
import sys
|
import sys
|
||||||
import os
|
import os
|
||||||
import numpy as np
|
import numpy as np
|
||||||
import asyncio
|
import asyncio
|
||||||
from ffmpeg.asyncio import FFmpeg
|
import ffmpeg
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
import subprocess
|
import subprocess
|
||||||
|
|
||||||
# Get the parent directory of the current file
|
# Get the parent directory of the current file
|
||||||
this_dir = Path(__file__).resolve().parent.parent.parent.parent.parent
|
this_dir = Path(__file__).resolve().parent.parent.parent.parent.parent
|
||||||
|
|
||||||
def load_audio(file, sampling_rate):
|
def load_audio(file, sampling_rate):
|
||||||
try:
|
try:
|
||||||
# Get the parent directory of the current file
|
file = str(file).strip(" ").strip('"').strip("\n").strip('"').strip(" ")
|
||||||
this_dir = Path(__file__).resolve().parent.parent.parent.parent.parent
|
|
||||||
file = file.strip(" ").strip('"').strip("\n").strip('"').strip(" ")
|
try:
|
||||||
|
# Use ffmpeg-python
|
||||||
# Set the ffmpeg_path variable based on the operating system
|
stream = (
|
||||||
if sys.platform == "win32":
|
ffmpeg
|
||||||
ffmpeg_path = os.path.join(this_dir, "system", "win_ffmpeg", "ffmpeg.exe")
|
.input(file)
|
||||||
else:
|
.output('pipe:', format='f32le', acodec='pcm_f32le', ac=1, ar=str(sampling_rate))
|
||||||
ffmpeg_path = "ffmpeg" # Default path for Linux and macOS
|
.run(capture_stdout=True, capture_stderr=True)
|
||||||
|
)
|
||||||
# Initialize the process variable
|
out = stream[0] # Get stdout data
|
||||||
process = subprocess.Popen(
|
|
||||||
[ffmpeg_path, "-y", "-i", file, "-f", "f32le", "-acodec", "pcm_f32le", "-ac", "1", "-ar", str(sampling_rate), "pipe:1"],
|
except ffmpeg.Error as e:
|
||||||
stdout=subprocess.PIPE,
|
print(f"FFmpeg error: {e.stderr.decode('utf-8')}")
|
||||||
stderr=subprocess.PIPE
|
raise RuntimeError(f"FFmpeg error: {e.stderr.decode('utf-8')}") from e
|
||||||
)
|
|
||||||
|
return np.frombuffer(out, np.float32).flatten()
|
||||||
out, err = process.communicate()
|
|
||||||
if process.returncode != 0:
|
except Exception as error:
|
||||||
print(f"FFmpeg error: {err.decode('utf-8')}") # Debug statement
|
print(f"Error loading audio: {error}")
|
||||||
raise RuntimeError(f"FFmpeg error: {err.decode('utf-8')}")
|
raise RuntimeError(f"Failed to load audio: {error}") from error
|
||||||
#print("Audio loaded successfully") # Debug statement
|
|
||||||
except Exception as error:
|
|
||||||
print(f"Error loading audio: {error}") # Debug statement
|
|
||||||
raise RuntimeError(f"Failed to load audio: {error}")
|
def format_title(title):
|
||||||
|
formatted_title = (
|
||||||
return np.frombuffer(out, np.float32).flatten()
|
unicodedata.normalize("NFKD", title).encode("ascii", "ignore").decode("utf-8")
|
||||||
|
)
|
||||||
|
formatted_title = re.sub(r"[\u2500-\u257F]+", "", formatted_title)
|
||||||
|
formatted_title = re.sub(r"[^\w\s.-]", "", formatted_title)
|
||||||
def format_title(title):
|
formatted_title = re.sub(r"\s+", "_", formatted_title)
|
||||||
formatted_title = (
|
return formatted_title
|
||||||
unicodedata.normalize("NFKD", title).encode("ascii", "ignore").decode("utf-8")
|
|
||||||
)
|
|
||||||
formatted_title = re.sub(r"[\u2500-\u257F]+", "", formatted_title)
|
def load_embedding(embedder_model):
|
||||||
formatted_title = re.sub(r"[^\w\s.-]", "", formatted_title)
|
#print("EMBEDDER MODEL IS", embedder_model)
|
||||||
formatted_title = re.sub(r"\s+", "_", formatted_title)
|
embedding_list = {
|
||||||
return formatted_title
|
"contentvec": "contentvec_base.pt",
|
||||||
|
"hubert": "hubert_base.pt",
|
||||||
|
}
|
||||||
def load_embedding(embedder_model):
|
|
||||||
#print("EMBEDDER MODEL IS", embedder_model)
|
try:
|
||||||
embedding_list = {
|
this_dir = Path(__file__).resolve().parent.parent.parent.parent.parent
|
||||||
"contentvec": "contentvec_base.pt",
|
model_path = this_dir / "models" / "rvc_base" / embedding_list[embedder_model]
|
||||||
"hubert": "hubert_base.pt",
|
#print("MODEL PATH IS", model_path)
|
||||||
}
|
|
||||||
|
# Load model ensemble and task
|
||||||
try:
|
models = checkpoint_utils.load_model_ensemble_and_task(
|
||||||
this_dir = Path(__file__).resolve().parent.parent.parent.parent.parent
|
[f"{model_path}"],
|
||||||
model_path = this_dir / "models" / "rvc_base" / embedding_list[embedder_model]
|
suffix="",
|
||||||
#print("MODEL PATH IS", model_path)
|
)
|
||||||
|
|
||||||
# Load model ensemble and task
|
#print(f"Embedding model {embedder_model} loaded successfully.")
|
||||||
models = checkpoint_utils.load_model_ensemble_and_task(
|
return models
|
||||||
[f"{model_path}"],
|
except KeyError as e:
|
||||||
suffix="",
|
logging.error(f"Invalid embedder model name: {embedder_model}")
|
||||||
)
|
raise ValueError(f"Invalid embedder model name: {embedder_model}") from e
|
||||||
|
except Exception as e:
|
||||||
#print(f"Embedding model {embedder_model} loaded successfully.")
|
logging.error(f"Error loading embedding model: {e}")
|
||||||
return models
|
raise RuntimeError(f"Error loading embedding model: {e}") from e
|
||||||
except KeyError as e:
|
|
||||||
logging.error(f"Invalid embedder model name: {embedder_model}")
|
|
||||||
raise ValueError(f"Invalid embedder model name: {embedder_model}") from e
|
|
||||||
except Exception as e:
|
|
||||||
logging.error(f"Error loading embedding model: {e}")
|
|
||||||
raise RuntimeError(f"Error loading embedding model: {e}") from e
|
|
||||||
Reference in new issue
Block a user