






"""simple docstring"""





import uuid
from typing import Any, Dict, List, Optional, Union

from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
							import tensorflow as tf

if is_torch_available():
							import torch


__lowerCamelCase							=  logging.get_logger(__name__)



class       __A		:

					def __init__(			self  :      List[Any]     ,		__snake_case  :      str = None     ,		__snake_case  :      uuid.UUID = None     ,		__snake_case  :      Tuple=None     ,		__snake_case  :      List[Any]=None					)			->							Optional[Any]:
											if not conversation_id:
																	__magic_name__:	Optional[Any]							=       uuid.uuida()
											if past_user_inputs is None:
																	__magic_name__:	Union[str, Any]							=       []
											if generated_responses is None:
																	__magic_name__:	Tuple							=       []

											__magic_name__:	uuid.UUID							=       conversation_id
											__magic_name__:	List[str]							=       past_user_inputs
											__magic_name__:	List[str]							=       generated_responses
											__magic_name__:	Optional[str]							=       text

					def __eq__(			self  :      List[Any]     ,		__snake_case  :      int					)			->							List[str]:
											if not isinstance(__snake_case     ,		__snake_case					):
																	return False
											if self.uuid == other.uuid:
																	return True
											return (
											    self.new_user_input == other.new_user_input
											    and self.past_user_inputs == other.past_user_inputs
											    and self.generated_responses == other.generated_responses
											)

					def   lowerCamelCase__   (			self  :      List[str]     ,		__snake_case  :      str     ,		__snake_case  :      bool = False					)			->							Any:
											if self.new_user_input:
																	if overwrite:
																							logger.warning(
																							    F'User input added while unprocessed input was existing: "{self.new_user_input}" was overwritten '
																							    F'with: "{text}".'					)
																							__magic_name__:	List[str]							=       text
																	else:
																							logger.warning(
																							    F'User input added while unprocessed input was existing: "{self.new_user_input}" new input '
																							    F'ignored: "{text}". Set `overwrite` to True to overwrite unprocessed user input'					)
											else:
																	__magic_name__:	Union[str, Any]							=       text

					def   lowerCamelCase__   (			self  :      Any					)			->							Optional[int]:
											if self.new_user_input:
																	self.past_user_inputs.append(self.new_user_input					)
											__magic_name__:	Any							=       None

					def   lowerCamelCase__   (			self  :      Union[str, Any]     ,		__snake_case  :      str					)			->							str:
											self.generated_responses.append(__snake_case					)

					def   lowerCamelCase__   (			self  :      str					)			->							List[str]:
											for user_input, generated_response in zip(self.past_user_inputs     ,		self.generated_responses					):
																	yield True, user_input
																	yield False, generated_response
											if self.new_user_input:
																	yield True, self.new_user_input






					def __repr__(			self  :      Tuple					)			->							Tuple:
											__magic_name__:	Any							=       F'Conversation id: {self.uuid} \n'
											for is_user, text in self.iter_texts():
																	__magic_name__:	List[str]							=       """user""" if is_user else """bot"""
																	output += F'{name} >> {text} \n'
											return output



@add_end_docstrings(
    SCREAMING_SNAKE_CASE_     ,R"\n        min_length_for_response (`int`, *optional*, defaults to 32):\n            The minimum length (in number of tokens) for a response.\n        minimum_tokens (`int`, *optional*, defaults to 10):\n            The minimum length of tokens to leave for a response.\n    "     ,)
class       __A		(   SCREAMING_SNAKE_CASE_			):

					def __init__(			self  :      Union[str, Any]     ,		*__snake_case  :      Any     ,		**__snake_case  :      Optional[int]					)			->							int:
											super().__init__(*__snake_case     ,		**__snake_case					)
											if self.tokenizer.pad_token_id is None:
																	__magic_name__:	Optional[int]							=       self.tokenizer.eos_token

					def   lowerCamelCase__   (			self  :      Optional[Any]     ,		__snake_case  :      str=None     ,		__snake_case  :      List[str]=None     ,		__snake_case  :      int=None     ,		**__snake_case  :      Optional[int]					)			->							Dict:
											__magic_name__:	List[Any]							=       {}
											__magic_name__:	int							=       {}
											__magic_name__:	List[str]							=       {}

											if min_length_for_response is not None:
																	__magic_name__:	Union[str, Any]							=       min_length_for_response
											if minimum_tokens is not None:
																	__magic_name__:	Optional[int]							=       minimum_tokens

											if "max_length" in generate_kwargs:
																	__magic_name__:	Tuple							=       generate_kwargs["""max_length"""]
																	# self.max_length = generate_kwargs.get("max_length", self.model.config.max_length)
											if clean_up_tokenization_spaces is not None:
																	__magic_name__:	Optional[int]							=       clean_up_tokenization_spaces

											if generate_kwargs:
																	forward_params.update(__snake_case					)
											return preprocess_params, forward_params, postprocess_params

					def __call__(			self  :      List[str]     ,		__snake_case  :      Union[Conversation, List[Conversation]]     ,		__snake_case  :      int=0     ,		**__snake_case  :      Dict					)			->							Dict:
											__magic_name__:	List[str]							=       super().__call__(__snake_case     ,		num_workers=__snake_case     ,		**__snake_case					)
											if isinstance(__snake_case     ,		__snake_case					) and len(__snake_case					) == 1:
																	return outputs[0]
											return outputs

					def   lowerCamelCase__   (			self  :      Union[str, Any]     ,		__snake_case  :      Conversation     ,		__snake_case  :      Tuple=3_2					)			->							Dict[str, Any]:
											if not isinstance(__snake_case     ,		__snake_case					):
																	raise ValueError("""ConversationalPipeline, expects Conversation as inputs"""					)
											if conversation.new_user_input is None:
																	raise ValueError(
																	    F'Conversation with UUID {type(conversation.uuid					)} does not contain new user input to process. '
																	    """Add user inputs with the conversation's `add_user_input` method"""					)
											if hasattr(self.tokenizer     ,		"""_build_conversation_input_ids"""					):
																	__magic_name__:	int							=       self.tokenizer._build_conversation_input_ids(__snake_case					)
											else:
																	# If the tokenizer cannot handle conversations, we default to only the old version
																	__magic_name__:	Union[str, Any]							=       self._legacy_parse_and_tokenize(__snake_case					)

											if self.framework == "pt":
																	__magic_name__:	int							=       torch.LongTensor([input_ids]					)
											elif self.framework == "tf":
																	__magic_name__:	Tuple							=       tf.constant([input_ids]					)
											return {"input_ids": input_ids, "conversation": conversation}

					def   lowerCamelCase__   (			self  :      Optional[Any]     ,		__snake_case  :      Any     ,		__snake_case  :      int=1_0     ,		**__snake_case  :      Optional[Any]					)			->							Any:
											__magic_name__:	Any							=       generate_kwargs.get("""max_length"""     ,		self.model.config.max_length					)

											__magic_name__:	int							=       model_inputs["""input_ids"""].shape[1]
											if max_length - minimum_tokens < n:
																	logger.warning(F'Conversation input is to long ({n}), trimming it to ({max_length} - {minimum_tokens})'					)
																	__magic_name__:	Any							=       max_length - minimum_tokens
																	__magic_name__:	Union[str, Any]							=       model_inputs["""input_ids"""][:, -trim:]
																	if "attention_mask" in model_inputs:
																							__magic_name__:	Union[str, Any]							=       model_inputs["""attention_mask"""][:, -trim:]
											__magic_name__:	Tuple							=       model_inputs.pop("""conversation"""					)
											__magic_name__:	Any							=       max_length
											__magic_name__:	Tuple							=       self.model.generate(**__snake_case     ,		**__snake_case					)
											if self.model.config.is_encoder_decoder:
																	__magic_name__:	List[str]							=       1
											else:
																	__magic_name__:	List[Any]							=       n
											return {"output_ids": output_ids[:, start_position:], "conversation": conversation}

					def   lowerCamelCase__   (			self  :      List[Any]     ,		__snake_case  :      str     ,		__snake_case  :      Tuple=True					)			->							Any:
											__magic_name__:	Tuple							=       model_outputs["""output_ids"""]
											__magic_name__:	List[Any]							=       self.tokenizer.decode(
											    output_ids[0]     ,		skip_special_tokens=__snake_case     ,		clean_up_tokenization_spaces=__snake_case     ,		)
											__magic_name__:	Tuple							=       model_outputs["""conversation"""]
											conversation.mark_processed()
											conversation.append_response(__snake_case					)
											return conversation






					def   lowerCamelCase__   (			self  :      Union[str, Any]     ,		__snake_case  :      Conversation					)			->							Dict:
											__magic_name__:	Dict							=       self.tokenizer.eos_token_id
											__magic_name__:	Optional[int]							=       []
											for is_user, text in conversation.iter_texts():
																	if eos_token_id is not None:
																							input_ids.extend(self.tokenizer.encode(__snake_case     ,		add_special_tokens=__snake_case					) + [eos_token_id]					)
																	else:
																							input_ids.extend(self.tokenizer.encode(__snake_case     ,		add_special_tokens=__snake_case					)					)

											if len(__snake_case					) > self.tokenizer.model_max_length:
																	__magic_name__:	Tuple							=       input_ids[-self.tokenizer.model_max_length :]
											return input_ids











"""simple docstring"""





# 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.
from ..models.whisper import WhisperForConditionalGeneration, WhisperProcessor
from .base import PipelineTool



class       __A		(   SCREAMING_SNAKE_CASE_			):
					UpperCAmelCase__								=				"openai/whisper-base"
					UpperCAmelCase__								=				(
					    "This is a tool that transcribes an audio into text. It takes an input named `audio` and returns the "
					    "transcribed text."
					)
					UpperCAmelCase__								=				"transcriber"
					UpperCAmelCase__								=				WhisperProcessor
					UpperCAmelCase__								=				WhisperForConditionalGeneration

					UpperCAmelCase__								=				["audio"]
					UpperCAmelCase__								=				["text"]

					def   lowerCamelCase__   (			self  :      Dict     ,		__snake_case  :      List[str]					)			->							Dict:
											return self.pre_processor(__snake_case     ,		return_tensors="""pt"""					).input_features

					def   lowerCamelCase__   (			self  :      int     ,		__snake_case  :      Union[str, Any]					)			->							List[str]:
											return self.model.generate(inputs=__snake_case					)






					def   lowerCamelCase__   (			self  :      Union[str, Any]     ,		__snake_case  :      List[Any]					)			->							Any:
											return self.pre_processor.batch_decode(__snake_case     ,		skip_special_tokens=__snake_case					)[0]



