| from pathlib import Path |
| import json |
|
|
| from transformers.models.mamba2.modeling_mamba2 import segment_sum |
|
|
| from lib.utils import cmd |
| from environment import TEST_DATA |
|
|
|
|
| def read_recording(folder: Path=TEST_DATA/"recordings", count_limit=None): |
| """读取录音文件夹，返回音频路径、文本, |
| """ |
| data_file = folder / 'data.json' |
| with open(data_file, encoding='utf-8') as f: |
| data = json.load(f) |
| count = 0 |
| for filename, text in data.items(): |
| count += 1 |
| if count_limit and count > count_limit: |
| break |
| wav_path = folder / filename |
| yield wav_path, text |
|
|
|
|
| def read_dataset(folder: Path= TEST_DATA / "AIShell", count_limit=None): |
| """line sample: {"audio": {"path": "dataset/audio/data_aishell/wav/test/S0916/BAC009S0916W0158.wav"}, "sentence": "顾客体验的核心是真善美", "duration": 3.22, "sentences": [{"start": 0, "end": 3.22, "text": "顾客体验的核心是真善美"}]}""" |
| with open(folder / "dataset/dataset.txt") as f: |
| lines =f.readlines() |
| count = 0 |
| for line in lines: |
| if count_limit and count > count_limit: |
| break |
| count += 1 |
| line = line.strip() |
| if not line: |
| continue |
| data = json.loads(line) |
|
|
| yield folder / data["audio"]["path"], data["sentence"] |
|
|
| def read_emilia(folder: Path=TEST_DATA/"ZH-B000000", count_limit=None): |
| """读取 emilia 数据集，返回音频路径、文本, |
| json 文件样例： |
| {"id": "ZH_B00000_S00110_W000000", "wav": "ZH_B00000/ZH_B00000_S00110/mp3/ZH_B00000_S00110_W000000.mp3", "text": "\u628a\u63e1\u6700\u524d\u6cbf\u7684\u91d1\u878d\u9886\u57df\u548c\u533a\u5757\u94fe\u6700\u65b0\u8d44\u8baf\u3002\u6211\u4eec\u4e00\u8d77\u6765\u4e86\u89e3\u4e00\u4e0b\u4eca\u5929\u5e02\u573a\u4e0a\u6709\u53d1\u751f\u54ea\u4e9b\u91cd\u8981\u4e8b\u4ef6\u3002", "duration": 7.963, "speaker": "ZH_B00000_S00110", "language": "zh", "dnsmos": 3.3808}""" |
| count = 0 |
| for json_file in sorted(folder.glob("*.json")): |
| count += 1 |
| if count_limit and count > count_limit: |
| break |
| with open(json_file, encoding="utf-8") as f: |
| data = json.load(f) |
| text = data["text"] |
| duration = data["duration"] |
| wav_path = folder /f'{json_file.stem}.wav' |
| if not wav_path.exists(): |
| mp3_path = folder / f'{json_file.stem}.mp3' |
| command=f"ffmpeg -i {mp3_path} -ac 1 -ar 16000 {wav_path}" |
| cmd(command) |
| yield wav_path, text |
|
|
| def read_wenet(folder: Path=TEST_DATA/"wenet", json_file="WenetSpeech_TEST_NET.json", count_limit=None): |
| """读取 wenet 数据集，返回音频路径、文本, |
| """ |
| count = 0 |
| with open(folder/json_file, encoding="utf-8") as f: |
| data = json.load(f) |
| audios = data["audios"] |
| for a in audios: |
| audio_file = Path(folder/a['path']) |
| if len(a["segments"])>=100: |
| continue |
| for seg in a["segments"]: |
| if count > count_limit: |
| return |
| seg_file = audio_file.parent / (seg["sid"]+".wav") |
| if not seg_file.exists(): |
| command = f"ffmpeg -i {audio_file} -ar 16000 -ac 1 -ss {seg['begin_time']} -to {seg['end_time']} {seg_file}" |
| cmd(command) |
| count +=1 |
| yield seg_file, seg["text"] |
|
|
| def read_libri(folder: Path=TEST_DATA/"LibriSpeech/test-clean", count_limit=None): |
| """读取 libri 数据集，返回音频路径、文本, |
| """ |
| count = 0 |
| for trans_file in sorted(folder.rglob("*trans.txt")): |
| with open(trans_file, encoding="utf-8") as f: |
| lines = f.readlines() |
| for line in lines: |
| if count_limit and count >= count_limit: |
| return |
| parts = line.strip().split(" ", 1) |
| if len(parts) != 2: |
| print("Invalid line:", line) |
| continue |
| file_id, text = parts |
| |
| flac_path = trans_file.parent / (file_id + ".flac") |
| wav_path = flac_path.with_suffix(".wav") |
| if not wav_path.exists(): |
| command = f"ffmpeg -i {flac_path} -ar 16000 -ac 1 {wav_path}" |
| cmd(command) |
| count += 1 |
| yield wav_path, text |
|
|
| if __name__ == '__main__': |
| read_wenet() |
