




# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import torch

from ..models.speechta import SpeechTaForTextToSpeech, SpeechTaHifiGan, SpeechTaProcessor
from ..utils import is_datasets_available
from .base import PipelineTool


if is_datasets_available():
		from datasets import load_dataset




class 					snake_case__							(UpperCamelCase__     ):
				"""simple docstring"""



				SCREAMING_SNAKE_CASE_ :							List[str]											=      """microsoft/speecht5_tts"""
				SCREAMING_SNAKE_CASE_ :							Union[str, Any]											=      (
				    """This is a tool that reads an English text out loud. It takes an input named `text` which should contain the """
				    """text to read (in English) and returns a waveform object containing the sound."""
				)
				SCREAMING_SNAKE_CASE_ :							Dict											=      """text_reader"""
				SCREAMING_SNAKE_CASE_ :							List[str]											=      SpeechTaProcessor
				SCREAMING_SNAKE_CASE_ :							str											=      SpeechTaForTextToSpeech
				SCREAMING_SNAKE_CASE_ :							str											=      SpeechTaHifiGan

				SCREAMING_SNAKE_CASE_ :							Optional[Any]											=      ["""text"""]
				SCREAMING_SNAKE_CASE_ :							Any											=      ["""audio"""]





				def 							__UpperCAmelCase							(   self       :  Tuple							)      ->  str:
							if self.post_processor is None:
										a					=    """microsoft/speecht5_hifigan"""
							super().setup()





				def 							__UpperCAmelCase							(   self       :  Optional[int]      ,   __lowerCamelCase       :  Tuple      ,   __lowerCamelCase       :  List[Any]=None							)      ->  Optional[Any]:
							a					=    self.pre_processor(text=_a      ,   return_tensors="pt"      ,   truncation=_a							)

							if speaker_embeddings is None:
										if not is_datasets_available():
													raise ImportError("Datasets needs to be installed if not passing speaker embeddings."							)

										a					=    load_dataset("Matthijs/cmu-arctic-xvectors"      ,   split="validation"							)
										a					=    torch.tensor(embeddings_dataset[73_05]["xvector"]							).unsqueeze(0							)

							return {"input_ids": inputs["input_ids"], "speaker_embeddings": speaker_embeddings}





				def 							__UpperCAmelCase							(   self       :  str      ,   __lowerCamelCase       :  Tuple							)      ->  List[Any]:
							with torch.no_grad():
										return self.model.generate_speech(**_a							)





				def 							__UpperCAmelCase							(   self       :  List[Any]      ,   __lowerCamelCase       :  Optional[int]							)      ->  List[str]:
							with torch.no_grad():
										return self.post_processor(_a							).cpu().detach()












from typing import List, Optional, TypeVar

from .arrow_dataset import Dataset, _concatenate_map_style_datasets, _interleave_map_style_datasets
from .dataset_dict import DatasetDict, IterableDatasetDict
from .info import DatasetInfo
from .iterable_dataset import IterableDataset, _concatenate_iterable_datasets, _interleave_iterable_datasets
from .splits import NamedSplit
from .utils import logging
from .utils.py_utils import Literal


__lowerCAmelCase	:       List[Any]     =	logging.get_logger(__name__)


__lowerCAmelCase	:       List[Any]     =	TypeVar('DatasetType', Dataset, IterableDataset)




def     __magic_name__       (				A	:      List[DatasetType],						A	:      Optional[List[float]] = None,						A	:      Optional[int] = None,						A	:      Optional[DatasetInfo] = None,						A	:      Optional[NamedSplit] = None,						A	:      Literal["first_exhausted", "all_exhausted"] = "first_exhausted",						):






			'''simple docstring'''







			from .arrow_dataset import Dataset
			from .iterable_dataset import IterableDataset

			if not datasets:
						raise ValueError("Unable to interleave an empty list of datasets."     )
			for i, dataset in enumerate(A     ):
						if not isinstance(A,						(Dataset, IterableDataset)     ):
									if isinstance(A,						(DatasetDict, IterableDatasetDict)     ):
												if not dataset:
															raise ValueError(
															    F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} """
															    "is an empty dataset dictionary."     )
												raise ValueError(
												    F"""Dataset at position {i} has at least one split: {list(A     )}\n"""
												    F"""Please pick one to interleave with the other datasets, for example: dataset['{next(iter(A     )     )}']"""     )
									raise ValueError(
									    F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(A     ).__name__}."""     )
						if i == 0:
									a					,	a					=    (
									    (Dataset, IterableDataset) if isinstance(A,						A     ) else (IterableDataset, Dataset)
									)
						elif not isinstance(A,						A     ):
									raise ValueError(
									    F"""Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."""     )
			if stopping_strategy not in ["first_exhausted", "all_exhausted"]:
						raise ValueError(F"""{stopping_strategy} is not supported. Please enter a valid stopping_strategy."""     )
			if dataset_type is Dataset:
						return _interleave_map_style_datasets(
						    A,						A,						A,						info=A,						split=A,						stopping_strategy=A     )
			else:
						return _interleave_iterable_datasets(
						    A,						A,						A,						info=A,						split=A,						stopping_strategy=A     )






def     __magic_name__       (				A	:      List[DatasetType],						A	:      Optional[DatasetInfo] = None,						A	:      Optional[NamedSplit] = None,						A	:      int = 0,						):






			'''simple docstring'''







			if not dsets:
						raise ValueError("Unable to concatenate an empty list of datasets."     )
			for i, dataset in enumerate(A     ):
						if not isinstance(A,						(Dataset, IterableDataset)     ):
									if isinstance(A,						(DatasetDict, IterableDatasetDict)     ):
												if not dataset:
															raise ValueError(
															    F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} """
															    "is an empty dataset dictionary."     )
												raise ValueError(
												    F"""Dataset at position {i} has at least one split: {list(A     )}\n"""
												    F"""Please pick one to interleave with the other datasets, for example: dataset['{next(iter(A     )     )}']"""     )
									raise ValueError(
									    F"""Expected a list of Dataset objects or a list of IterableDataset objects, but element at position {i} is a {type(A     ).__name__}."""     )
						if i == 0:
									a					,	a					=    (
									    (Dataset, IterableDataset) if isinstance(A,						A     ) else (IterableDataset, Dataset)
									)
						elif not isinstance(A,						A     ):
									raise ValueError(
									    F"""Unable to interleave a {dataset_type.__name__} (at position 0) with a {other_type.__name__} (at position {i}). Expected a list of Dataset objects or a list of IterableDataset objects."""     )
			if dataset_type is Dataset:
						return _concatenate_map_style_datasets(A,						info=A,						split=A,						axis=A     )
			else:
						return _concatenate_iterable_datasets(A,						info=A,						split=A,						axis=A     )






