From d8b586e02cd14f7eed6b330bd4f110cb1e7f24ad Mon Sep 17 00:00:00 2001
From: 游雁 <zhifu.gzf@alibaba-inc.com>
Date: 星期二, 09 一月 2024 20:33:12 +0800
Subject: [PATCH] funasr1.0  modelscope

---
 funasr/utils/load_utils.py |   54 ++++++++++++++----------------------------------------
 1 files changed, 14 insertions(+), 40 deletions(-)

diff --git a/funasr/utils/load_utils.py b/funasr/utils/load_utils.py
index c5c3ffc..c82987f 100644
--- a/funasr/utils/load_utils.py
+++ b/funasr/utils/load_utils.py
@@ -10,29 +10,13 @@
 import logging
 from torch.nn.utils.rnn import pad_sequence
 try:
-	from urllib.parse import urlparse
-	from funasr.download.file import HTTPStorage
-	import tempfile
+	from funasr.download.file import download_from_url
 except:
 	print("urllib is not installed, if you infer from url, please install it first.")
-# def load_audio(data_or_path_or_list, fs: int=16000, audio_fs: int=16000):
-#
-# 	if isinstance(data_or_path_or_list, (list, tuple)):
-# 		return [load_audio(audio, fs=fs, audio_fs=audio_fs) for audio in data_or_path_or_list]
-#
-# 	if isinstance(data_or_path_or_list, str) and os.path.exists(data_or_path_or_list):
-# 		data_or_path_or_list, audio_fs = torchaudio.load(data_or_path_or_list)
-# 		data_or_path_or_list = data_or_path_or_list[0, :]
-# 	elif isinstance(data_or_path_or_list, np.ndarray): # audio sample point
-# 		data_or_path_or_list = np.squeeze(data_or_path_or_list) #[n_samples,]
-#
-# 	if audio_fs != fs:
-# 		resampler = torchaudio.transforms.Resample(audio_fs, fs)
-# 		data_or_path_or_list = resampler(data_or_path_or_list[None, :])[0, :]
-# 	return data_or_path_or_list
 
 
-def load_audio_text_image_video(data_or_path_or_list, fs: int = 16000, audio_fs: int = 16000, data_type=None, tokenizer=None):
+
+def load_audio_text_image_video(data_or_path_or_list, fs: int = 16000, audio_fs: int = 16000, data_type="sound", tokenizer=None):
 	if isinstance(data_or_path_or_list, (list, tuple)):
 		if data_type is not None and isinstance(data_type, (list, tuple)):
 
@@ -47,16 +31,22 @@
 
 			return data_or_path_or_list_ret
 		else:
-			return [load_audio_text_image_video(audio, fs=fs, audio_fs=audio_fs) for audio in data_or_path_or_list]
+			return [load_audio_text_image_video(audio, fs=fs, audio_fs=audio_fs, data_type=data_type) for audio in data_or_path_or_list]
 	if isinstance(data_or_path_or_list, str) and data_or_path_or_list.startswith('http'):
 		data_or_path_or_list = download_from_url(data_or_path_or_list)
 	if isinstance(data_or_path_or_list, str) and os.path.exists(data_or_path_or_list):
-		data_or_path_or_list, audio_fs = torchaudio.load(data_or_path_or_list)
-		data_or_path_or_list = data_or_path_or_list[0, :]
+		if data_type is None or data_type == "sound":
+			data_or_path_or_list, audio_fs = torchaudio.load(data_or_path_or_list)
+			data_or_path_or_list = data_or_path_or_list[0, :]
+		# elif data_type == "text" and tokenizer is not None:
+		# 	data_or_path_or_list = tokenizer.encode(data_or_path_or_list)
+	elif isinstance(data_or_path_or_list, str) and data_type == "text" and tokenizer is not None:
+		data_or_path_or_list = tokenizer.encode(data_or_path_or_list)
 	elif isinstance(data_or_path_or_list, np.ndarray):  # audio sample point
 		data_or_path_or_list = np.squeeze(data_or_path_or_list)  # [n_samples,]
-	elif isinstance(data_or_path_or_list, str) and data_type is not None and data_type == "text" and tokenizer is not None:
-		data_or_path_or_list = tokenizer.encode(data_or_path_or_list)
+	else:
+		pass
+		# print(f"unsupport data type: {data_or_path_or_list}, return raw data")
 		
 	if audio_fs != fs and data_type != "text":
 		resampler = torchaudio.transforms.Resample(audio_fs, fs)
@@ -107,19 +97,3 @@
 		data_len = torch.tensor([data_len])
 	return data.to(torch.float32), data_len.to(torch.int32)
 
-def download_from_url(url):
-	
-	result = urlparse(url)
-	file_path = None
-	if result.scheme is not None and len(result.scheme) > 0:
-		storage = HTTPStorage()
-		# bytes
-		data = storage.read(url)
-		work_dir = tempfile.TemporaryDirectory().name
-		if not os.path.exists(work_dir):
-			os.makedirs(work_dir)
-		file_path = os.path.join(work_dir, os.path.basename(url))
-		with open(file_path, 'wb') as fb:
-			fb.write(data)
-	assert file_path is not None, f"failed to download: {url}"
-	return file_path
\ No newline at end of file

--
Gitblit v1.9.1