From 49e8e9d8fc1209c347aa2c2c65c6eb067b9f79d4 Mon Sep 17 00:00:00 2001
From: zhu-gu-an <76513567+zhu-gu-an@users.noreply.github.com>
Date: 星期六, 13 一月 2024 13:54:00 +0800
Subject: [PATCH] add triton paraformer large online (#1242)
---
funasr/datasets/preprocessor.py | 155 +++++++++++++++++++++++++++++----------------------
1 files changed, 88 insertions(+), 67 deletions(-)
diff --git a/funasr/datasets/preprocessor.py b/funasr/datasets/preprocessor.py
index c6623f8..966cc94 100644
--- a/funasr/datasets/preprocessor.py
+++ b/funasr/datasets/preprocessor.py
@@ -10,12 +10,12 @@
import numpy as np
import scipy.signal
-import soundfile
+import librosa
import jieba
-from funasr.text.build_tokenizer import build_tokenizer
-from funasr.text.cleaner import TextCleaner
-from funasr.text.token_id_converter import TokenIDConverter
+from funasr.tokenizer.build_tokenizer import build_tokenizer
+from funasr.tokenizer.cleaner import TextCleaner
+from funasr.tokenizer.token_id_converter import TokenIDConverter
class AbsPreprocessor(ABC):
@@ -201,7 +201,7 @@
self.seg_dict = None
if seg_dict_file is not None:
self.seg_dict = {}
- with open(seg_dict_file) as f:
+ with open(seg_dict_file, "r", encoding="utf8") as f:
lines = f.readlines()
for line in lines:
s = line.strip().split()
@@ -284,7 +284,7 @@
if self.rirs is not None and self.rir_apply_prob >= np.random.random():
rir_path = np.random.choice(self.rirs)
if rir_path is not None:
- rir, _ = soundfile.read(
+ rir, _ = librosa.load(
rir_path, dtype=np.float64, always_2d=True
)
@@ -310,28 +310,31 @@
noise_db = np.random.uniform(
self.noise_db_low, self.noise_db_high
)
- with soundfile.SoundFile(noise_path) as f:
- if f.frames == nsamples:
- noise = f.read(dtype=np.float64, always_2d=True)
- elif f.frames < nsamples:
- offset = np.random.randint(0, nsamples - f.frames)
- # noise: (Time, Nmic)
- noise = f.read(dtype=np.float64, always_2d=True)
- # Repeat noise
- noise = np.pad(
- noise,
- [(offset, nsamples - f.frames - offset), (0, 0)],
- mode="wrap",
- )
- else:
- offset = np.random.randint(0, f.frames - nsamples)
- f.seek(offset)
- # noise: (Time, Nmic)
- noise = f.read(
- nsamples, dtype=np.float64, always_2d=True
- )
- if len(noise) != nsamples:
- raise RuntimeError(f"Something wrong: {noise_path}")
+
+ audio_data = librosa.load(noise_path, dtype='float32')[0][None, :]
+ frames = len(audio_data[0])
+ if frames == nsamples:
+ noise = audio_data
+ elif frames < nsamples:
+ offset = np.random.randint(0, nsamples - frames)
+ # noise: (Time, Nmic)
+ noise = audio_data
+ # Repeat noise
+ noise = np.pad(
+ noise,
+ [(offset, nsamples - frames - offset), (0, 0)],
+ mode="wrap",
+ )
+ else:
+ noise = audio_data[:, nsamples]
+ # offset = np.random.randint(0, frames - nsamples)
+ # f.seek(offset)
+ # noise: (Time, Nmic)
+ # noise = f.read(
+ # nsamples, dtype=np.float64, always_2d=True
+ # )
+ # if len(noise) != nsamples:
+ # raise RuntimeError(f"Something wrong: {noise_path}")
# noise: (Nmic, Time)
noise = noise.T
@@ -702,55 +705,73 @@
return line
@classmethod
- def split_words_jieba(cls, text: str):
- input_list = text.split()
- token_list_all = []
- langauge_list = []
- token_list_tmp = []
- language_flag = None
- for token in input_list:
- if cls.isEnglish(token) and language_flag == 'Chinese':
+ def split_words(cls, text: str , seg_jieba: bool):
+ if seg_jieba == True:
+ input_list = text.split()
+ token_list_all = []
+ langauge_list = []
+ token_list_tmp = []
+ language_flag = None
+ for token in input_list:
+ if cls.isEnglish(token) and language_flag == 'Chinese':
+ token_list_all.append(token_list_tmp)
+ langauge_list.append('Chinese')
+ token_list_tmp = []
+ elif not cls.isEnglish(token) and language_flag == 'English':
+ token_list_all.append(token_list_tmp)
+ langauge_list.append('English')
+ token_list_tmp = []
+
+ token_list_tmp.append(token)
+
+ if cls.isEnglish(token):
+ language_flag = 'English'
+ else:
+ language_flag = 'Chinese'
+
+ if token_list_tmp:
token_list_all.append(token_list_tmp)
- langauge_list.append('Chinese')
- token_list_tmp = []
- elif not cls.isEnglish(token) and language_flag == 'English':
- token_list_all.append(token_list_tmp)
- langauge_list.append('English')
- token_list_tmp = []
+ langauge_list.append(language_flag)
- token_list_tmp.append(token)
+ result_list = []
+ for token_list_tmp, language_flag in zip(token_list_all, langauge_list):
+ if language_flag == 'English':
+ result_list.extend(token_list_tmp)
+ else:
+ seg_list = jieba.cut(cls.join_chinese_and_english(token_list_tmp), HMM=False)
+ result_list.extend(seg_list)
- if cls.isEnglish(token):
- language_flag = 'English'
- else:
- language_flag = 'Chinese'
+ return result_list
- if token_list_tmp:
- token_list_all.append(token_list_tmp)
- langauge_list.append(language_flag)
+ else:
+ words = []
+ segs = text.split()
+ for seg in segs:
+ # There is no space in seg.
+ current_word = ""
+ for c in seg:
+ if len(c.encode()) == 1:
+ # This is an ASCII char.
+ current_word += c
+ else:
+ # This is a Chinese char.
+ if len(current_word) > 0:
+ words.append(current_word)
+ current_word = ""
+ words.append(c)
+ if len(current_word) > 0:
+ words.append(current_word)
+ return words
- result_list = []
- for token_list_tmp, language_flag in zip(token_list_all, langauge_list):
- if language_flag == 'English':
- result_list.extend(token_list_tmp)
- else:
- seg_list = jieba.cut(cls.join_chinese_and_english(token_list_tmp), HMM=False)
- result_list.extend(seg_list)
-
- return result_list
def __call__(
self, uid: str, data: Dict[str, Union[list, str, np.ndarray]]
) -> Dict[str, Union[list, np.ndarray]]:
# Split words.
- if isinstance(data[self.text_name], str):
- if self.seg_jieba:
- # jieba.load_userdict(seg_dict_file)
- split_text = self.split_words_jieba(data[self.text_name])
- else:
- split_text = self.split_words(data[self.text_name])
- else:
- split_text = data[self.text_name]
+ data_in = data[self.text_name]
+ if isinstance(data[self.text_name], list):
+ data_in = " ".join(data[self.text_name])
+ split_text = self.split_words(data_in, self.seg_jieba)
data[self.text_name] = " ".join(split_text)
data = self._speech_process(data)
data = self._text_process(data)
--
Gitblit v1.9.1