| | |
| | | import time |
| | | import logging |
| | | from torch.nn.utils.rnn import pad_sequence |
| | | |
| | | try: |
| | | from urllib.parse import urlparse |
| | | from funasr.download.file import HTTPStorage |
| | | import tempfile |
| | | 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 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] |
| | | |
| | | 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 isinstance(data_len, (list, tuple)): |
| | | data_len = torch.tensor([data_len]) |
| | | return data.to(torch.float32), data_len.to(torch.int32) |
| | | 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 |