rendering/universal_tts/examples/esd_inneremo/eval/test.py

# with open('/workspace/echoloc/codes/tts_outputs/susc/persona/cosyvoice2/celsds_full_fixlr1e5_best/info.tsv', 'r', errors='replace') as f:  # 使用二进制模式读取文件
    
#     for line in f.readlines():
#         print(line)

# from funasr import AutoModel
# model = AutoModel(model="/workspace/echoloc/modelscope/iic/speech_seaco_paraformer_large_asr_nat-zh-cn-16k-common-vocab8404-pytorch/")
# est_txt = model.generate(input=r"/workspace/echoloc/codes/fairseq_ecvc/examples/celsds/temp_outputs/zero-shot-ref.wav", 
#                      batch_size_s=300, 
#                      hotword='魔搭', disable_pbar=True)[0]

# """
# {'key': 'zero-shot-ref', 'text': '笔 记 它 只 是 一 个 工 具 就 是 最 终 的 目 的 是 吸 收 这 些 知 识', 'timestamp': [[190, 270], [270, 450], [450, 590], [590, 690], [690, 810], [810, 890], [890, 970], [970, 1150], [1150, 1390], [1410, 1570], [1570, 1810], [1890, 2030], [2030, 2210], [2210, 2370], [2370, 2510], [2510, 2750], [2750, 2970], [2970, 3150], [3150, 3390], [3410, 3510], [3510, 3710], [3710, 3930], [3930, 4105]]}
# """
# print(est_txt)

# import csv
# src_audio_paths = [
#     "/workspace/echoloc/dataset/EMOGEN/test_prompt/gens/context_en/tp/temp_signal/2_0.wav",
#     "/workspace/echoloc/dataset/EMOGEN/test_prompt/gens/context_en/tp/temp_signal/2_1.wav",
#     "/workspace/echoloc/dataset/EMOGEN/test_prompt/gens/context_en/tp/temp_signal/2_2.wav"
# ]

# tgt_audio_paths = [
#     "/workspace/echoloc/dataset/EMOGEN/test_prompt/gens/context_en/tp/temp_signal/1_0.wav",
#     "/workspace/echoloc/dataset/EMOGEN/test_prompt/gens/context_en/tp/temp_signal/1_1.wav",
#     "/workspace/echoloc/dataset/EMOGEN/test_prompt/gens/context_en/tp/temp_signal/1_2.wav"
# ]
# output_tsv_path = "/workspace/echoloc/codes/fairseq_ecvc/examples/celsds/infer/evaluate/auto_pcp.tsv"
# # 写入 TSV 文件
# with open(output_tsv_path, mode="w", newline="", encoding="utf-8") as f:
#     writer = csv.writer(f, delimiter="\t")
#     writer.writerow(["src_audio", "tgt_audio"])  # 写入表头
#     for src, tgt in zip(src_audio_paths, tgt_audio_paths):
#         writer.writerow([src, tgt])
# import librosa
# import numpy as np
# import numpy.polynomial.polynomial as poly
# import onnxruntime as ort
# import soundfile as sf
# from requests import session
# from tqdm import tqdm
# p808_onnx_sess = ort.InferenceSession(r"/workspace/echoloc/codes/fairseq_ecvc/examples/reconstruct_dhubert/evaluate/ecvc/model_v8.onnx", providers=["CUDAExecutionProvider"])

# import torch
# torch.backends.cudnn.allow_tf32 = True
# torch.backends.cuda.matmul.allow_tf32 = True
# import whisperx
# import gc 
# from jiwer import wer
# import torchaudio

# device = "cuda" 
# audio_file = "/workspace/echoloc/codes/fairseq_ecvc/examples/celsds/temp_outputs/1_0.wav"
# batch_size = 16 # reduce if low on GPU mem
# compute_type = "float16" # change to "int8" if low on GPU mem (may reduce accuracy)
# def load_wav(wav, target_sr):
#     speech, sample_rate = torchaudio.load(wav)
#     speech = speech.mean(dim=0, keepdim=True)
#     if sample_rate != target_sr:
#         assert sample_rate > target_sr, 'wav sample rate {} must be greater than {}'.format(sample_rate, target_sr)
#         speech = torchaudio.transforms.Resample(orig_freq=sample_rate, new_freq=target_sr)(speech)
#     return speech


# def wer_and_align(refs, inp_path, 
#                   asr_model, align_model, metadata, batch_size=1):
#     """
#     根据 refs 内容，从 ASR 输出中找出 WER 最匹配片段（不重叠）并返回时间戳和片段音频。
#     """
#     inp_wav = load_wav(inp_path, target_sr=16000)[0].numpy()
#     result = asr_model.transcribe(inp_wav, batch_size=batch_size)
#     result = whisperx.align(result["segments"], align_model, metadata, inp_wav, device, return_char_alignments=False)["segments"]

#     # 展开所有 word
#     result_words = []
#     for seg in result:
#         result_words.extend(seg["words"])

#     words = [w["word"] for w in result_words]
#     used_range = [False] * len(words)
#     results = []
#     segments = []
#     inp_wav_np = inp_wav.flatten()

#     for ref in refs:
#         ref_words = ref.strip().split()
#         ref_len = len(ref_words)

#         best_score = float("inf")
#         best_start = -1
#         best_end = -1
#         best_text = ""

#         for start in range(len(words)):
#             for end in range(start + 1, min(len(words) + 1, start + ref_len + 10)):

#                 if any(used_range[start:end]):
#                     continue

#                 hyp_words = [result_words[i]["word"] for i in range(start, end)]
#                 score = wer(" ".join(ref_words), " ".join(hyp_words))

#                 if score < best_score:
#                     best_score = score
#                     best_start = start
#                     best_end = end
#                     best_text = " ".join(hyp_words)

#         if best_start != -1:
#             for i in range(best_start, best_end):
#                 used_range[i] = True

#             start_time = result_words[best_start]["start"]
#             end_time = result_words[best_end - 1]["end"]

#             results.append({
#                 "ref": ref,
#                 "match": best_text,
#                 "start_time": start_time,
#                 "end_time": end_time,
#                 "wer_score": best_score
#             })

#             segments.append(
#                 inp_wav_np[int(start_time * 16000): int(end_time * 16000)]
#             )

#     # 用于整体评估
#     est_txt = " ".join(words)
#     cer_score = wer(" ".join(refs), est_txt)

#     return results, segments, cer_score, inp_wav_np


# # 1. Transcribe with original whisper (batched)
# model = whisperx.load_model("large-v2", device, compute_type=compute_type)
# # 2. Align whisper output
# model_a, metadata = whisperx.load_align_model(language_code="en", device=device)
# refs = ["That joke really , didn't it?", "Let's try something something different, shall we?"]
# results, segments, cer_score, inp_wav_np = wer_and_align(
#     refs, audio_file, model, model_a, metadata, batch_size=1
# )
# print(results)

# import pathlib
# import csv
# from collections import defaultdict
# import numpy as np

# input_file=r"/workspace/echoloc/dataset/EMOGEN/test_prompt/gens/context_zh/tp/eval4autopcp.tsv"
# input_file = pathlib.Path(input_file).resolve()
# input_flag_file = input_file.parent / "eval4autopcp_flag.tsv"
# output_file = input_file.parent / "output.txt"
    
# with open(input_file, mode="r", encoding="utf-8") as f:
#     with open(input_flag_file, mode="r", encoding="utf-8") as flag_f:
#         flags = [line for line in flag_f.readlines() if len(line)>2]
#         print(len(flags))
#         reader = csv.DictReader(f, delimiter="\t")
#         # 1. 创建一个 flag -> scores 列表的映射
#         flag2scores = defaultdict(list)
#         for idx, flag in enumerate(flags):
#             flag2scores[flag].append(2.0)

#         # 2. 对每个 flag 求平均
#         flag2mean = {flag: np.mean(slist) for flag, slist in flag2scores.items()}

#         seen = set()
#         deduped_flags = []
#         for flag in flags:
#             if flag not in seen:
#                 deduped_flags.append(flag)
#                 seen.add(flag)
        
#         print(f"去重后flags数量: {len(deduped_flags)}")
        
#         # 构建最终 merged_scores
#         merged_scores = [flag2mean[flag] for flag in deduped_flags]
        
# print(len(merged_scores))

import os
import librosa
import soundfile as sf
from audiotsm import wsola
from audiotsm.io.wav import WavReader, WavWriter
import subprocess

def ensure_wav_mono_sr(input_path, output_path, target_sr=22050):
    y, sr = librosa.load(input_path, sr=target_sr, mono=True)
    sf.write(output_path, y, target_sr)
    return output_path

def run_wsola(input_path, output_path, speed):
    with WavReader(input_path) as reader:
        with WavWriter(output_path, reader.channels, reader.samplerate) as writer:
            tsm = wsola(reader.channels, speed=speed)
            tsm.run(reader, writer)

def run_phase_vocoder(input_path, output_path, speed):
    y, sr = librosa.load(input_path, sr=None)
    y_stretch = librosa.effects.time_stretch(y, rate=speed)
    sf.write(output_path, y_stretch, sr)

def run_sox(input_path, output_path, speed):
    tempo = str(speed)
    subprocess.run(["sox", input_path, output_path, "tempo", tempo], check=True)

def test_all_methods(input_wav, speed):
    clean_input = "/workspace/echoloc/codes/fairseq_ecvc/examples/celsds/temp_outputs/temp.wav"
    ensure_wav_mono_sr(input_wav, clean_input)

    run_wsola(clean_input, f"/workspace/echoloc/codes/fairseq_ecvc/examples/celsds/temp_outputs/output_wsola_{speed}x.wav", speed)
    run_phase_vocoder(clean_input, f"/workspace/echoloc/codes/fairseq_ecvc/examples/celsds/temp_outputs/output_phase_{speed}x.wav", speed)
    run_sox(clean_input, f"/workspace/echoloc/codes/fairseq_ecvc/examples/celsds/temp_outputs/output_sox_{speed}x.wav", speed)
    print(f"✅ All methods processed for speed: {speed}x")

# 🧪 举例测试
if __name__ == "__main__":
    test_all_methods("/workspace/echoloc/codes/fairseq_ecvc/examples/celsds/temp_outputs/zero-shot-ref.wav", 0.5)  # 变慢
    test_all_methods("/workspace/echoloc/codes/fairseq_ecvc/examples/celsds/temp_outputs/zero-shot-ref.wav", 2.0)  # 变快