





"""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


UpperCamelCase_									=logging.get_logger(__name__)




class 			_a						:






				def __init__(			self		:   Optional[Any],			lowerCAmelCase__		:   Dict = None,			lowerCAmelCase__		:   Any = None,			lowerCAmelCase__		:   Union[str, Any]=None,			lowerCAmelCase__		:   Dict=None							)     ->				List[Any]:




										'''simple docstring'''



										if not conversation_id:
																_UpperCamelCase      : Union[str, Any]         =   uuid.uuida()
										if past_user_inputs is None:
																_UpperCamelCase      : List[Any]         =   []
										if generated_responses is None:
																_UpperCamelCase      : Tuple         =   []

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






				def __eq__(			self		:   int,			lowerCAmelCase__		:   Optional[Any]							)     ->				List[Any]:




										'''simple docstring'''



										if not isinstance(SCREAMING_SNAKE_CASE_,			SCREAMING_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     snake_case     (			self		:   Union[str, Any],			lowerCAmelCase__		:   str,			lowerCAmelCase__		:   Optional[Any] = False							)     ->				List[str]:




										'''simple docstring'''



										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}\"."""							)
																						_UpperCamelCase      : Tuple         =   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:
																_UpperCamelCase      : Any         =   text






				def     snake_case     (			self		:   Optional[Any]							)     ->				Optional[int]:




										'''simple docstring'''



										if self.new_user_input:
																self.past_user_inputs.append(self.new_user_input							)
										_UpperCamelCase      : List[str]         =   None






				def     snake_case     (			self		:   Optional[int],			lowerCAmelCase__		:   Tuple							)     ->				int:




										'''simple docstring'''



										self.generated_responses.append(SCREAMING_SNAKE_CASE_							)






				def     snake_case     (			self		:   Optional[int]							)     ->				Optional[Any]:




										'''simple docstring'''



										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		:   Union[str, Any]							)     ->				Optional[Any]:




										'''simple docstring'''



										_UpperCamelCase      : Dict         =   f"""Conversation id: {self.uuid} \n"""
										for is_user, text in self.iter_texts():
																_UpperCamelCase      : List[Any]         =   """user""" if is_user else """bot"""
																output += f"""{name} >> {text} \n"""
										return output




@add_end_docstrings(
    _lowerCAmelCase     ,    r'''
        min_length_for_response (`int`, *optional*, defaults to 32):
            The minimum length (in number of tokens) for a response.
        minimum_tokens (`int`, *optional*, defaults to 10):
            The minimum length of tokens to leave for a response.
    '''     ,    )
class 			_a						(				_lowerCAmelCase    ):






				def __init__(			self		:   int,			*lowerCAmelCase__		:   str,			**lowerCAmelCase__		:   List[str]							)     ->				str:




										'''simple docstring'''



										super().__init__(*SCREAMING_SNAKE_CASE_,			**SCREAMING_SNAKE_CASE_							)
										if self.tokenizer.pad_token_id is None:
																_UpperCamelCase      : List[str]         =   self.tokenizer.eos_token






				def     snake_case     (			self		:   int,			lowerCAmelCase__		:   Union[str, Any]=None,			lowerCAmelCase__		:   str=None,			lowerCAmelCase__		:   Optional[int]=None,			**lowerCAmelCase__		:   List[str]							)     ->				Optional[int]:




										'''simple docstring'''



										_UpperCamelCase      : Tuple         =   {}
										_UpperCamelCase      : Dict         =   {}
										_UpperCamelCase      : str         =   {}

										if min_length_for_response is not None:
																_UpperCamelCase      : Optional[Any]         =   min_length_for_response
										if minimum_tokens is not None:
																_UpperCamelCase      : Dict         =   minimum_tokens

										if "max_length" in generate_kwargs:
																_UpperCamelCase      : List[str]         =   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:
																_UpperCamelCase      : Optional[Any]         =   clean_up_tokenization_spaces

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






				def __call__(			self		:   Any,			lowerCAmelCase__		:   List[str],			lowerCAmelCase__		:   Optional[int]=0,			**lowerCAmelCase__		:   Any							)     ->				List[str]:




										'''simple docstring'''



										_UpperCamelCase      : Tuple         =   super().__call__(SCREAMING_SNAKE_CASE_,			num_workers=SCREAMING_SNAKE_CASE_,			**SCREAMING_SNAKE_CASE_							)
										if isinstance(SCREAMING_SNAKE_CASE_,			SCREAMING_SNAKE_CASE_							) and len(SCREAMING_SNAKE_CASE_							) == 1:
																return outputs[0]
										return outputs






				def     snake_case     (			self		:   Union[str, Any],			lowerCAmelCase__		:   List[Any],			lowerCAmelCase__		:   Optional[Any]=3_2							)     ->				Dict[str, Any]:




										'''simple docstring'''



										if not isinstance(SCREAMING_SNAKE_CASE_,			SCREAMING_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'''							):
																_UpperCamelCase      : Optional[Any]         =   self.tokenizer._build_conversation_input_ids(SCREAMING_SNAKE_CASE_							)
										else:
																# If the tokenizer cannot handle conversations, we default to only the old version
																_UpperCamelCase      : Any         =   self._legacy_parse_and_tokenize(SCREAMING_SNAKE_CASE_							)

										if self.framework == "pt":
																_UpperCamelCase      : List[Any]         =   torch.LongTensor([input_ids]							)
										elif self.framework == "tf":
																_UpperCamelCase      : List[Any]         =   tf.constant([input_ids]							)
										return {"input_ids": input_ids, "conversation": conversation}






				def     snake_case     (			self		:   Tuple,			lowerCAmelCase__		:   Optional[Any],			lowerCAmelCase__		:   Tuple=1_0,			**lowerCAmelCase__		:   List[Any]							)     ->				Dict:




										'''simple docstring'''



										_UpperCamelCase      : Any         =   generate_kwargs.get('''max_length''',			self.model.config.max_length							)

										_UpperCamelCase      : Optional[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})"""							)
																_UpperCamelCase      : Tuple         =   max_length - minimum_tokens
																_UpperCamelCase      : Union[str, Any]         =   model_inputs["""input_ids"""][:, -trim:]
																if "attention_mask" in model_inputs:
																						_UpperCamelCase      : List[Any]         =   model_inputs["""attention_mask"""][:, -trim:]
										_UpperCamelCase      : str         =   model_inputs.pop('''conversation'''							)
										_UpperCamelCase      : int         =   max_length
										_UpperCamelCase      : List[Any]         =   self.model.generate(**SCREAMING_SNAKE_CASE_,			**SCREAMING_SNAKE_CASE_							)
										if self.model.config.is_encoder_decoder:
																_UpperCamelCase      : Any         =   1
										else:
																_UpperCamelCase      : Optional[Any]         =   n
										return {"output_ids": output_ids[:, start_position:], "conversation": conversation}






				def     snake_case     (			self		:   Any,			lowerCAmelCase__		:   Dict,			lowerCAmelCase__		:   Optional[Any]=True							)     ->				Union[str, Any]:




										'''simple docstring'''



										_UpperCamelCase      : Optional[Any]         =   model_outputs["""output_ids"""]
										_UpperCamelCase      : Optional[int]         =   self.tokenizer.decode(
										    output_ids[0],			skip_special_tokens=SCREAMING_SNAKE_CASE_,			clean_up_tokenization_spaces=SCREAMING_SNAKE_CASE_,			)
										_UpperCamelCase      : str         =   model_outputs["""conversation"""]
										conversation.mark_processed()
										conversation.append_response(SCREAMING_SNAKE_CASE_							)
										return conversation






				def     snake_case     (			self		:   Optional[int],			lowerCAmelCase__		:   Optional[Any]							)     ->				Dict:




										'''simple docstring'''



										_UpperCamelCase      : List[str]         =   self.tokenizer.eos_token_id
										_UpperCamelCase      : str         =   []
										for is_user, text in conversation.iter_texts():
																if eos_token_id is not None:
																						input_ids.extend(self.tokenizer.encode(SCREAMING_SNAKE_CASE_,			add_special_tokens=SCREAMING_SNAKE_CASE_							) + [eos_token_id]							)
																else:
																						input_ids.extend(self.tokenizer.encode(SCREAMING_SNAKE_CASE_,			add_special_tokens=SCREAMING_SNAKE_CASE_							)							)

										if len(SCREAMING_SNAKE_CASE_							) > self.tokenizer.model_max_length:
																_UpperCamelCase      : Optional[Any]         =   input_ids[-self.tokenizer.model_max_length :]
										return input_ids










"""simple docstring"""




from ...processing_utils import ProcessorMixin




class 			_a						(				_lowerCAmelCase    ):
				UpperCamelCase								=   ['''image_processor''', '''feature_extractor''']
				UpperCamelCase								=   '''TvltImageProcessor'''
				UpperCamelCase								=   '''TvltFeatureExtractor'''






				def __init__(			self		:   Union[str, Any],			lowerCAmelCase__		:   str,			lowerCAmelCase__		:   List[str]							)     ->				Optional[Any]:




										'''simple docstring'''



										super().__init__(image_processor=lowerCAmelCase__,			feature_extractor=lowerCAmelCase__							)

										_UpperCamelCase      : List[str]         =   image_processor
										_UpperCamelCase      : int         =   feature_extractor






				def __call__(			self		:   List[str],			lowerCAmelCase__		:   Optional[int]=None,			lowerCAmelCase__		:   str=None,			lowerCAmelCase__		:   Dict=None,			lowerCAmelCase__		:   str=None,			lowerCAmelCase__		:   Optional[int]=False,			lowerCAmelCase__		:   str=False,			*lowerCAmelCase__		:   List[str],			**lowerCAmelCase__		:   Optional[int],			)     ->				Dict:




										'''simple docstring'''



										if images is None and audio is None:
																raise ValueError('''You need to specify either an `images` or `audio` input to process.'''							)

										_UpperCamelCase      : Optional[int]         =   None
										if images is not None:
																_UpperCamelCase      : Optional[int]         =   self.image_processor(lowerCAmelCase__,			mask_pixel=lowerCAmelCase__,			*lowerCAmelCase__,			**lowerCAmelCase__							)
										if images_mixed is not None:
																_UpperCamelCase      : str         =   self.image_processor(lowerCAmelCase__,			is_mixed=lowerCAmelCase__,			*lowerCAmelCase__,			**lowerCAmelCase__							)
										if audio is not None:
																_UpperCamelCase      : Union[str, Any]         =   self.feature_extractor(
																    lowerCAmelCase__,			*lowerCAmelCase__,			sampling_rate=lowerCAmelCase__,			mask_audio=lowerCAmelCase__,			**lowerCAmelCase__							)

										_UpperCamelCase      : str         =   {}
										if audio is not None:
																output_dict.update(lowerCAmelCase__							)
										if images is not None:
																output_dict.update(lowerCAmelCase__							)
										if images_mixed_dict is not None:
																output_dict.update(lowerCAmelCase__							)
										return output_dict






				@property
				def     snake_case     (			self		:   List[str]							)     ->				Tuple:




										'''simple docstring'''



										_UpperCamelCase      : List[str]         =   self.image_processor.model_input_names
										_UpperCamelCase      : List[Any]         =   self.feature_extractor.model_input_names
										return list(dict.fromkeys(image_processor_input_names + feature_extractor_input_names							)							)



