




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


_snake_case				:				Union[str, Any]       							=		logging.get_logger(__name__)

class       A   :







							def __init__(  self	:  int							,      lowerCAmelCase_	:  str = None							,      lowerCAmelCase_	:  uuid.UUID = None							,      lowerCAmelCase_	:  int=None							,      lowerCAmelCase_	:  Union[str, Any]=None )      ->							Any:


														"""simple docstring"""


														if not conversation_id:
																					_a     					=    uuid.uuida()
														if past_user_inputs is None:
																					_a     					=    []
														if generated_responses is None:
																					_a     					=    []

														_a     					=    conversation_id
														_a     					=    past_user_inputs
														_a     					=    generated_responses
														_a     					=    text







							def __eq__(  self	:  str							,      lowerCAmelCase_	:  Optional[int] )      ->							Union[str, Any]:


														"""simple docstring"""


														if not isinstance(lowerCAmelCase_							,      lowerCAmelCase_ ):
																					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	:  str							,      lowerCAmelCase_	:  str							,      lowerCAmelCase_	:  bool = False )      ->							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}".' )
																												_a     					=    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:
																					_a     					=    text







							def        __lowerCAmelCase      (  self	:  List[Any] )      ->							int:


														"""simple docstring"""


														if self.new_user_input:
																					self.past_user_inputs.append(self.new_user_input )
														_a     					=    None







							def        __lowerCAmelCase      (  self	:  List[str]							,      lowerCAmelCase_	:  str )      ->							Dict:


														"""simple docstring"""


														self.generated_responses.append(lowerCAmelCase_ )







							def        __lowerCAmelCase      (  self	:  str )      ->							List[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	:  Optional[int] )      ->							List[str]:


														"""simple docstring"""


														_a     					=    F'Conversation id: {self.uuid} \n'
														for is_user, text in self.iter_texts():
																					_a     					=    '''user''' if is_user else '''bot'''
																					output += F'{name} >> {text} \n'
														return output



@add_end_docstrings(
    _a					,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   (		_a   ):







							def __init__(  self	:  Optional[int]							,      *lowerCAmelCase_	:  Tuple							,      **lowerCAmelCase_	:  int )      ->							List[Any]:


														"""simple docstring"""


														super().__init__(*lowerCAmelCase_							,      **lowerCAmelCase_ )
														if self.tokenizer.pad_token_id is None:
																					_a     					=    self.tokenizer.eos_token







							def        __lowerCAmelCase      (  self	:  Dict							,      lowerCAmelCase_	:  Optional[Any]=None							,      lowerCAmelCase_	:  str=None							,      lowerCAmelCase_	:  Tuple=None							,      **lowerCAmelCase_	:  Tuple )      ->							Optional[int]:


														"""simple docstring"""


														_a     					=    {}
														_a     					=    {}
														_a     					=    {}

														if min_length_for_response is not None:
																					_a     					=    min_length_for_response
														if minimum_tokens is not None:
																					_a     					=    minimum_tokens

														if "max_length" in generate_kwargs:
																					_a     					=    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:
																					_a     					=    clean_up_tokenization_spaces

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







							def __call__(  self	:  Any							,      lowerCAmelCase_	:  Union[Conversation, List[Conversation]]							,      lowerCAmelCase_	:  Optional[Any]=0							,      **lowerCAmelCase_	:  List[str] )      ->							str:


														"""simple docstring"""


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







							def        __lowerCAmelCase      (  self	:  Optional[int]							,      lowerCAmelCase_	:  Conversation							,      lowerCAmelCase_	:  Any=32 )      ->							Dict[str, Any]:


														"""simple docstring"""


														if not isinstance(lowerCAmelCase_							,      lowerCAmelCase_ ):
																					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''' ):
																					_a     					=    self.tokenizer._build_conversation_input_ids(lowerCAmelCase_ )
														else:
																					# If the tokenizer cannot handle conversations, we default to only the old version
																					_a     					=    self._legacy_parse_and_tokenize(lowerCAmelCase_ )

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







							def        __lowerCAmelCase      (  self	:  str							,      lowerCAmelCase_	:  str							,      lowerCAmelCase_	:  Tuple=10							,      **lowerCAmelCase_	:  Union[str, Any] )      ->							List[str]:


														"""simple docstring"""


														_a     					=    generate_kwargs.get('''max_length'''							,      self.model.config.max_length )

														_a     					=    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})' )
																					_a     					=    max_length - minimum_tokens
																					_a     					=    model_inputs['''input_ids'''][:, -trim:]
																					if "attention_mask" in model_inputs:
																												_a     					=    model_inputs['''attention_mask'''][:, -trim:]
														_a     					=    model_inputs.pop('''conversation''' )
														_a     					=    max_length
														_a     					=    self.model.generate(**lowerCAmelCase_							,      **lowerCAmelCase_ )
														if self.model.config.is_encoder_decoder:
																					_a     					=    1
														else:
																					_a     					=    n
														return {"output_ids": output_ids[:, start_position:], "conversation": conversation}







							def        __lowerCAmelCase      (  self	:  str							,      lowerCAmelCase_	:  Dict							,      lowerCAmelCase_	:  int=True )      ->							Optional[Any]:


														"""simple docstring"""


														_a     					=    model_outputs['''output_ids''']
														_a     					=    self.tokenizer.decode(
														    output_ids[0]							,      skip_special_tokens=lowerCAmelCase_							,      clean_up_tokenization_spaces=lowerCAmelCase_							,      )
														_a     					=    model_outputs['''conversation''']
														conversation.mark_processed()
														conversation.append_response(lowerCAmelCase_ )
														return conversation







							def        __lowerCAmelCase      (  self	:  List[str]							,      lowerCAmelCase_	:  Conversation )      ->							Dict:


														"""simple docstring"""


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

														if len(lowerCAmelCase_ ) > self.tokenizer.model_max_length:
																					_a     					=    input_ids[-self.tokenizer.model_max_length :]
														return input_ids







'''simple docstring'''


from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ...utils.backbone_utils import BackboneConfigMixin, get_aligned_output_features_output_indices


_snake_case				:				Dict       							=		logging.get_logger(__name__)

_snake_case				:				Optional[Any]       							=		{
    'shi-labs/nat-mini-in1k-224': 'https://huggingface.co/shi-labs/nat-mini-in1k-224/resolve/main/config.json',
    # See all Nat models at https://huggingface.co/models?filter=nat
}

class       A   (		_a					,_a   ):
							lowercase_          =    'nat'

							lowercase_          =    {
							    'num_attention_heads': 'num_heads',
							    'num_hidden_layers': 'num_layers',
							}







							def __init__(  self	:  List[str]							,      lowerCAmelCase_	:  str=4							,      lowerCAmelCase_	:  Any=3							,      lowerCAmelCase_	:  List[Any]=64							,      lowerCAmelCase_	:  Dict=[3, 4, 6, 5]							,      lowerCAmelCase_	:  Dict=[2, 4, 8, 16]							,      lowerCAmelCase_	:  str=7							,      lowerCAmelCase_	:  Dict=3.0							,      lowerCAmelCase_	:  int=True							,      lowerCAmelCase_	:  Tuple=0.0							,      lowerCAmelCase_	:  str=0.0							,      lowerCAmelCase_	:  List[str]=0.1							,      lowerCAmelCase_	:  int="gelu"							,      lowerCAmelCase_	:  List[str]=0.0_2							,      lowerCAmelCase_	:  str=1e-5							,      lowerCAmelCase_	:  Tuple=0.0							,      lowerCAmelCase_	:  str=None							,      lowerCAmelCase_	:  int=None							,      **lowerCAmelCase_	:  List[Any]							,      )      ->							Any:


														"""simple docstring"""


														super().__init__(**lowerCAmelCase_ )

														_a     					=    patch_size
														_a     					=    num_channels
														_a     					=    embed_dim
														_a     					=    depths
														_a     					=    len(lowerCAmelCase_ )
														_a     					=    num_heads
														_a     					=    kernel_size
														_a     					=    mlp_ratio
														_a     					=    qkv_bias
														_a     					=    hidden_dropout_prob
														_a     					=    attention_probs_dropout_prob
														_a     					=    drop_path_rate
														_a     					=    hidden_act
														_a     					=    layer_norm_eps
														_a     					=    initializer_range
														# we set the hidden_size attribute in order to make Nat work with VisionEncoderDecoderModel
														# this indicates the channel dimension after the last stage of the model
														_a     					=    int(embed_dim * 2 ** (len(lowerCAmelCase_ ) - 1) )
														_a     					=    layer_scale_init_value
														_a     					=    ['''stem'''] + [F'stage{idx}' for idx in range(1							,      len(lowerCAmelCase_ ) + 1 )]
														_a					,							_a     					=    get_aligned_output_features_output_indices(
														    out_features=lowerCAmelCase_							,      out_indices=lowerCAmelCase_							,      stage_names=self.stage_names )

