
'''simple docstring'''
import dataclasses
import json
import warnings
from dataclasses import dataclass, field
from time import time
from typing import List

from ..utils import logging


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







def 		A_(     A						:		str=None		,     A						:		Union[str, Any]=None):
							return field(default_factory=lambda: default		,     metadata=A)





@dataclass
class 			SCREAMING_SNAKE_CASE__      :
				lowerCAmelCase_     			=							list_field(
				    default=[]							,       metadata={
				        """help""": (
				            """Model checkpoints to be provided to the AutoModel classes. Leave blank to benchmark the base version"""
				            """ of all available models"""
				        )
				    }							,       )

				lowerCAmelCase_     			=							list_field(
				    default=[8]							,       metadata={"""help""": """List of batch sizes for which memory and time performance will be evaluated"""})

				lowerCAmelCase_     			=							list_field(
				    default=[8, 32, 1_28, 5_12]							,       metadata={"""help""": """List of sequence lengths for which memory and time performance will be evaluated"""}							,       )

				lowerCAmelCase_     			=							field(
				    default=snake_case_							,       metadata={"""help""": """Whether to benchmark inference of model. Inference can be disabled via --no-inference."""}							,       )
				lowerCAmelCase_     			=							field(
				    default=snake_case_							,       metadata={"""help""": """Whether to run on available cuda devices. Cuda can be disabled via --no-cuda."""}							,       )
				lowerCAmelCase_     			=							field(
				    default=snake_case_							,       metadata={"""help""": """Whether to run on available tpu devices. TPU can be disabled via --no-tpu."""})
				lowerCAmelCase_     			=							field(default=snake_case_							,       metadata={"""help""": """Use FP16 to accelerate inference."""})
				lowerCAmelCase_     			=							field(default=snake_case_							,       metadata={"""help""": """Benchmark training of model"""})
				lowerCAmelCase_     			=							field(default=snake_case_							,       metadata={"""help""": """Verbose memory tracing"""})
				lowerCAmelCase_     			=							field(
				    default=snake_case_							,       metadata={"""help""": """Whether to perform speed measurements. Speed measurements can be disabled via --no-speed."""}							,       )
				lowerCAmelCase_     			=							field(
				    default=snake_case_							,       metadata={
				        """help""": """Whether to perform memory measurements. Memory measurements can be disabled via --no-memory"""
				    }							,       )
				lowerCAmelCase_     			=							field(default=snake_case_							,       metadata={"""help""": """Trace memory line by line"""})
				lowerCAmelCase_     			=							field(default=snake_case_							,       metadata={"""help""": """Save result to a CSV file"""})
				lowerCAmelCase_     			=							field(default=snake_case_							,       metadata={"""help""": """Save all print statements in a log file"""})
				lowerCAmelCase_     			=							field(default=snake_case_							,       metadata={"""help""": """Whether to print environment information"""})
				lowerCAmelCase_     			=							field(
				    default=snake_case_							,       metadata={
				        """help""": (
				            """Whether to use multiprocessing for memory and speed measurement. It is highly recommended to use"""
				            """ multiprocessing for accurate CPU and GPU memory measurements. This option should only be disabled"""
				            """ for debugging / testing and on TPU."""
				        )
				    }							,       )
				lowerCAmelCase_     			=							field(
				    default=f"inference_time_{round(time())}.csv"							,       metadata={"""help""": """CSV filename used if saving time results to csv."""}							,       )
				lowerCAmelCase_     			=							field(
				    default=f"inference_memory_{round(time())}.csv"							,       metadata={"""help""": """CSV filename used if saving memory results to csv."""}							,       )
				lowerCAmelCase_     			=							field(
				    default=f"train_time_{round(time())}.csv"							,       metadata={"""help""": """CSV filename used if saving time results to csv for training."""}							,       )
				lowerCAmelCase_     			=							field(
				    default=f"train_memory_{round(time())}.csv"							,       metadata={"""help""": """CSV filename used if saving memory results to csv for training."""}							,       )
				lowerCAmelCase_     			=							field(
				    default=f"env_info_{round(time())}.csv"							,       metadata={"""help""": """CSV filename used if saving environment information."""}							,       )
				lowerCAmelCase_     			=							field(
				    default=f"log_{round(time())}.csv"							,       metadata={"""help""": """Log filename used if print statements are saved in log."""}							,       )
				lowerCAmelCase_     			=							field(default=3							,       metadata={"""help""": """Times an experiment will be run."""})
				lowerCAmelCase_     			=							field(
				    default=snake_case_							,       metadata={
				        """help""": (
				            """Instead of loading the model as defined in `config.architectures` if exists, just load the pretrain"""
				            """ model weights."""
				        )
				    }							,       )







				def 						UpperCAmelCase_    (       self     )->		int:







											'''simple docstring'''






											warnings.warn(
											    F'''The class {self.__class__} is deprecated. Hugging Face Benchmarking utils'''
											    ' are deprecated in general and it is advised to use external Benchmarking libraries '
											    ' to benchmark Transformer models.'						,				A_						,				)







				def 						UpperCAmelCase_    (       self     )->		Tuple:







											'''simple docstring'''






											return json.dumps(dataclasses.asdict(self     )						,				indent=2     )







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







											'''simple docstring'''






											if len(self.models     ) <= 0:
																		raise ValueError(
																		    'Please make sure you provide at least one model name / model identifier, *e.g.* `--models'
																		    ' bert-base-cased` or `args.models = [\'bert-base-cased\'].'     )
											return self.models







				@property
				def 						UpperCAmelCase_    (       self     )->		Any:







											'''simple docstring'''






											if not self.multi_process:
																		return False
											elif self.is_tpu:
																		logger.info('Multiprocessing is currently not possible on TPU.'     )
																		return False
											else:
																		return True






'''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					:    Optional[Any]           = logging.get_logger(__name__)


class 			SCREAMING_SNAKE_CASE__      :







				def __init__(       self						,				A_ = None						,				A_ = None						,				A_=None						,				A_=None     )->		Optional[Any]:







											'''simple docstring'''






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

											UpperCamelCase   			=     conversation_id
											UpperCamelCase   			=     past_user_inputs
											UpperCamelCase   			=     generated_responses
											UpperCamelCase   			=     text







				def __eq__(       self						,				A_     )->		List[Any]:







											'''simple docstring'''






											if not isinstance(A_						,				A_     ):
																		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 						UpperCAmelCase_    (       self						,				A_						,				A_ = False     )->		int:







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







				def 						UpperCAmelCase_    (       self     )->		Any:







											'''simple docstring'''






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







				def 						UpperCAmelCase_    (       self						,				A_     )->		int:







											'''simple docstring'''






											self.generated_responses.append(A_     )







				def 						UpperCAmelCase_    (       self     )->		List[str]:







											'''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     )->		Any:







											'''simple docstring'''






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


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







				def __init__(       self						,				*A_						,				**A_     )->		Any:







											'''simple docstring'''






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







				def 						UpperCAmelCase_    (       self						,				A_=None						,				A_=None						,				A_=None						,				**A_     )->		Union[str, Any]:







											'''simple docstring'''






											UpperCamelCase   			=     {}
											UpperCamelCase   			=     {}
											UpperCamelCase   			=     {}

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

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

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







				def __call__(       self						,				A_						,				A_=0						,				**A_     )->		Any:







											'''simple docstring'''






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







				def 						UpperCAmelCase_    (       self						,				A_						,				A_=32     )->		Dict[str, Any]:







											'''simple docstring'''






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

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







				def 						UpperCAmelCase_    (       self						,				A_						,				A_=10						,				**A_     )->		Optional[Any]:







											'''simple docstring'''






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

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







				def 						UpperCAmelCase_    (       self						,				A_						,				A_=True     )->		Tuple:







											'''simple docstring'''






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







				def 						UpperCAmelCase_    (       self						,				A_     )->		Dict:







											'''simple docstring'''






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

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




