# 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) # 变快