

from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer
from ...configuration_utils import PretrainedConfig
from ...onnx import OnnxConfig, OnnxConfigWithPast, OnnxSeqaSeqConfigWithPast
from ...onnx.utils import compute_effective_axis_dimension
from ...utils import TensorType, is_torch_available, logging


A_       : Dict         =			logging.get_logger(__name__)

A_       : Optional[Any]         =			{
    '''Helsinki-NLP/opus-mt-en-de''': '''https://huggingface.co/Helsinki-NLP/opus-mt-en-de/resolve/main/config.json''',
    # See all Marian models at https://huggingface.co/models?filter=marian
}



class     _a						(__snake_case				):


					'''simple docstring'''






					UpperCAmelCase__:   Union[str, Any]  							=       """marian"""
					UpperCAmelCase__:   Optional[Any]  							=       ["""past_key_values"""]
					UpperCAmelCase__:   int  							=       {"""num_attention_heads""": """encoder_attention_heads""", """hidden_size""": """d_model"""}






					def __init__( self    ,       A__=5_8101    ,       A__=None    ,       A__=1024    ,       A__=12    ,       A__=4096    ,       A__=16    ,       A__=12    ,       A__=4096    ,       A__=16    ,       A__=0.0    ,       A__=0.0    ,       A__=True    ,       A__=True    ,       A__="gelu"    ,       A__=1024    ,       A__=0.1    ,       A__=0.0    ,       A__=0.0    ,       A__=0.0_2    ,       A__=5_8100    ,       A__=False    ,       A__=5_8100    ,       A__=0    ,       A__=0    ,       A__=True    ,       **A__    ,       ):
						A__							:    int        =		vocab_size
						A__							:    Optional[Any]        =		decoder_vocab_size or vocab_size
						A__							:    Tuple        =		max_position_embeddings
						A__							:    List[Any]        =		d_model
						A__							:    Any        =		encoder_ffn_dim
						A__							:    str        =		encoder_layers
						A__							:    str        =		encoder_attention_heads
						A__							:    Dict        =		decoder_ffn_dim
						A__							:    str        =		decoder_layers
						A__							:    Optional[int]        =		decoder_attention_heads
						A__							:    Optional[int]        =		dropout
						A__							:    Optional[int]        =		attention_dropout
						A__							:    Optional[int]        =		activation_dropout
						A__							:    int        =		activation_function
						A__							:    str        =		init_std
						A__							:    Optional[int]        =		encoder_layerdrop
						A__							:    Tuple        =		decoder_layerdrop
						A__							:    Union[str, Any]        =		use_cache
						A__							:    Optional[Any]        =		encoder_layers
						A__							:    Union[str, Any]        =		scale_embedding  # scale factor will be sqrt(d_model) if True
						A__							:    List[Any]        =		share_encoder_decoder_embeddings
						super().__init__(
						    pad_token_id=__SCREAMING_SNAKE_CASE    ,       eos_token_id=__SCREAMING_SNAKE_CASE    ,       is_encoder_decoder=__SCREAMING_SNAKE_CASE    ,       decoder_start_token_id=__SCREAMING_SNAKE_CASE    ,       forced_eos_token_id=__SCREAMING_SNAKE_CASE    ,       **__SCREAMING_SNAKE_CASE    ,       )



class     _a						(__snake_case				):


					'''simple docstring'''






					@property
					# Copied from transformers.models.bart.configuration_bart.BartOnnxConfig.inputs
					def   __A			( self			):
						if self.task in ["default", "seq2seq-lm"]:
							A__							:    Union[str, Any]        =		OrderedDict(
							    [
							        ("""input_ids""", {0: """batch""", 1: """encoder_sequence"""}),
							        ("""attention_mask""", {0: """batch""", 1: """encoder_sequence"""}),
							    ]			)

							if self.use_past:
								A__							:    int        =		{0: """batch"""}
								A__							:    int        =		{0: """batch""", 1: """past_decoder_sequence + sequence"""}
							else:
								A__							:    Dict        =		{0: """batch""", 1: """decoder_sequence"""}
								A__							:    Optional[int]        =		{0: """batch""", 1: """decoder_sequence"""}

							if self.use_past:
								self.fill_with_past_key_values_(__SCREAMING_SNAKE_CASE    ,       direction="""inputs"""			)
						elif self.task == "causal-lm":
							# TODO: figure this case out.
							A__							:    Tuple        =		OrderedDict(
							    [
							        ("""input_ids""", {0: """batch""", 1: """encoder_sequence"""}),
							        ("""attention_mask""", {0: """batch""", 1: """encoder_sequence"""}),
							    ]			)
							if self.use_past:
								A__   ,   A__							:    List[str]        =		self.num_layers
								for i in range(__SCREAMING_SNAKE_CASE			):
									A__							:    Union[str, Any]        =		{0: """batch""", 2: """past_sequence + sequence"""}
									A__							:    Optional[Any]        =		{0: """batch""", 2: """past_sequence + sequence"""}
						else:
							A__							:    Dict        =		OrderedDict(
							    [
							        ("""input_ids""", {0: """batch""", 1: """encoder_sequence"""}),
							        ("""attention_mask""", {0: """batch""", 1: """encoder_sequence"""}),
							        ("""decoder_input_ids""", {0: """batch""", 1: """decoder_sequence"""}),
							        ("""decoder_attention_mask""", {0: """batch""", 1: """decoder_sequence"""}),
							    ]			)

						return common_inputs






					@property
					# Copied from transformers.models.bart.configuration_bart.BartOnnxConfig.outputs
					def   __A			( self			):
						if self.task in ["default", "seq2seq-lm"]:
							A__							:    Any        =		super().outputs
						else:
							A__							:    Union[str, Any]        =		super(__SCREAMING_SNAKE_CASE    ,       self			).outputs
							if self.use_past:
								A__   ,   A__							:    Optional[int]        =		self.num_layers
								for i in range(__SCREAMING_SNAKE_CASE			):
									A__							:    Dict        =		{0: """batch""", 2: """past_sequence + sequence"""}
									A__							:    Dict        =		{0: """batch""", 2: """past_sequence + sequence"""}
						return common_outputs






					def   __A			( self    ,       A__    ,       A__ = -1    ,       A__ = -1    ,       A__ = False    ,       A__ = None    ,       ):
						A__							:    Any        =		self._generate_dummy_inputs_for_encoder_and_decoder(
						    __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE			)

						# Generate decoder inputs
						A__							:    Union[str, Any]        =		seq_length if not self.use_past else 1
						A__							:    Any        =		self._generate_dummy_inputs_for_encoder_and_decoder(
						    __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE			)
						A__							:    str        =		{F"""decoder_{name}""": tensor for name, tensor in decoder_inputs.items()}
						A__							:    Any        =		dict(**__SCREAMING_SNAKE_CASE    ,       **__SCREAMING_SNAKE_CASE			)

						if self.use_past:
							if not is_torch_available():
								raise ValueError("""Cannot generate dummy past_keys inputs without PyTorch installed."""			)
							else:
								import torch
							A__   ,   A__							:    List[Any]        =		common_inputs["""input_ids"""].shape
							A__							:    Any        =		common_inputs["""decoder_input_ids"""].shape[1]
							A__   ,   A__							:    Tuple        =		self.num_attention_heads
							A__							:    Dict        =		(
							    batch,
							    num_encoder_attention_heads,
							    encoder_seq_length,
							    self._config.hidden_size // num_encoder_attention_heads,
							)
							A__							:    int        =		decoder_seq_length + 3
							A__							:    Tuple        =		(
							    batch,
							    num_decoder_attention_heads,
							    decoder_past_length,
							    self._config.hidden_size // num_decoder_attention_heads,
							)

							A__							:    Optional[int]        =		torch.cat(
							    [common_inputs["""decoder_attention_mask"""], torch.ones(__SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE			)]    ,       dim=1			)

							A__							:    List[Any]        =		[]
							# If the number of encoder and decoder layers are present in the model configuration, both are considered
							A__   ,   A__							:    List[str]        =		self.num_layers
							A__							:    Optional[Any]        =		min(__SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE			)
							A__							:    Tuple        =		max(__SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE			) - min_num_layers
							A__							:    List[Any]        =		"""encoder""" if num_encoder_layers > num_decoder_layers else """decoder"""

							for _ in range(__SCREAMING_SNAKE_CASE			):
								common_inputs["past_key_values"].append(
								    (
								        torch.zeros(__SCREAMING_SNAKE_CASE			),
								        torch.zeros(__SCREAMING_SNAKE_CASE			),
								        torch.zeros(__SCREAMING_SNAKE_CASE			),
								        torch.zeros(__SCREAMING_SNAKE_CASE			),
								    )			)
							# TODO: test this.
							A__							:    Tuple        =		encoder_shape if remaining_side_name == """encoder""" else decoder_shape
							for _ in range(__SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE			):
								common_inputs["past_key_values"].append((torch.zeros(__SCREAMING_SNAKE_CASE			), torch.zeros(__SCREAMING_SNAKE_CASE			))			)
						return common_inputs






					def   __A			( self    ,       A__    ,       A__ = -1    ,       A__ = -1    ,       A__ = False    ,       A__ = None    ,       ):
						A__							:    int        =		self._generate_dummy_inputs_for_encoder_and_decoder(
						    __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE			)

						if self.use_past:
							if not is_torch_available():
								raise ValueError("""Cannot generate dummy past_keys inputs without PyTorch installed."""			)
							else:
								import torch
							A__   ,   A__							:    List[str]        =		common_inputs["""input_ids"""].shape
							# Not using the same length for past_key_values
							A__							:    Any        =		seqlen + 2
							A__   ,   A__							:    Dict        =		self.num_layers
							A__   ,   A__							:    List[Any]        =		self.num_attention_heads
							A__							:    int        =		(
							    batch,
							    num_encoder_attention_heads,
							    past_key_values_length,
							    self._config.hidden_size // num_encoder_attention_heads,
							)

							A__							:    List[str]        =		common_inputs["""attention_mask"""].dtype
							A__							:    Dict        =		torch.cat(
							    [common_inputs["""attention_mask"""], torch.ones(__SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       dtype=__SCREAMING_SNAKE_CASE			)]    ,       dim=1			)
							A__							:    Optional[Any]        =		[
							    (torch.zeros(__SCREAMING_SNAKE_CASE			), torch.zeros(__SCREAMING_SNAKE_CASE			)) for _ in range(__SCREAMING_SNAKE_CASE			)
							]
						return common_inputs






					def   __A			( self    ,       A__    ,       A__ = -1    ,       A__ = -1    ,       A__ = False    ,       A__ = None    ,       ):
						# Copied from OnnxConfig.generate_dummy_inputs
						# Did not use super(OnnxConfigWithPast, self).generate_dummy_inputs for code clarity.
						# If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
						A__							:    Any        =		compute_effective_axis_dimension(
						    __SCREAMING_SNAKE_CASE    ,       fixed_dimension=OnnxConfig.default_fixed_batch    ,       num_token_to_add=0			)

						# If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
						A__							:    str        =		tokenizer.num_special_tokens_to_add(__SCREAMING_SNAKE_CASE			)
						A__							:    List[Any]        =		compute_effective_axis_dimension(
						    __SCREAMING_SNAKE_CASE    ,       fixed_dimension=OnnxConfig.default_fixed_sequence    ,       num_token_to_add=__SCREAMING_SNAKE_CASE			)

						# Generate dummy inputs according to compute batch and sequence
						A__							:    str        =		[""" """.join([tokenizer.unk_token]			) * seq_length] * batch_size
						A__							:    Optional[Any]        =		dict(tokenizer(__SCREAMING_SNAKE_CASE    ,       return_tensors=__SCREAMING_SNAKE_CASE			)			)
						return common_inputs






					def   __A			( self    ,       A__    ,       A__ = -1    ,       A__ = -1    ,       A__ = False    ,       A__ = None    ,       ):
						if self.task in ["default", "seq2seq-lm"]:
							A__							:    Optional[Any]        =		self._generate_dummy_inputs_for_default_and_seqaseq_lm(
							    __SCREAMING_SNAKE_CASE    ,       batch_size=__SCREAMING_SNAKE_CASE    ,       seq_length=__SCREAMING_SNAKE_CASE    ,       is_pair=__SCREAMING_SNAKE_CASE    ,       framework=__SCREAMING_SNAKE_CASE			)

						else:
							A__							:    List[str]        =		self._generate_dummy_inputs_for_causal_lm(
							    __SCREAMING_SNAKE_CASE    ,       batch_size=__SCREAMING_SNAKE_CASE    ,       seq_length=__SCREAMING_SNAKE_CASE    ,       is_pair=__SCREAMING_SNAKE_CASE    ,       framework=__SCREAMING_SNAKE_CASE			)

						return common_inputs






					def   __A			( self    ,       A__    ,       A__    ,       A__    ,       A__			):
						if self.task in ["default", "seq2seq-lm"]:
							A__							:    Optional[Any]        =		super()._flatten_past_key_values_(__SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE			)
						else:
							A__							:    Any        =		super(__SCREAMING_SNAKE_CASE    ,       self			)._flatten_past_key_values_(
							    __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE    ,       __SCREAMING_SNAKE_CASE			)






					@property
					def   __A			( self			):
						return 1e-4











"""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 typing import TYPE_CHECKING

from ..models.auto import AutoModelForVisionaSeq
from ..utils import requires_backends
from .base import PipelineTool


if TYPE_CHECKING:
  from PIL import Image






class      a						(					__snake_case		):
    SCREAMING_SNAKE_CASE						:     Dict						= """Salesforce/blip-image-captioning-base"""
    SCREAMING_SNAKE_CASE						:     int						= (
        """This is a tool that generates a description of an image. It takes an input named `image` which should be the """
        """image to caption, and returns a text that contains the description in English."""
    )
    SCREAMING_SNAKE_CASE						:     Optional[int]						= """image_captioner"""
    SCREAMING_SNAKE_CASE						:     Optional[Any]						= AutoModelForVisionaSeq

    SCREAMING_SNAKE_CASE						:     int						= ["""image"""]
    SCREAMING_SNAKE_CASE						:     Optional[Any]						= ["""text"""]






    def __init__(       self     :							Optional[Any]       ,						*__SCREAMING_SNAKE_CASE     :							Optional[Any]       ,						**__SCREAMING_SNAKE_CASE     :							List[Any]					)     ->		Optional[Any]:
      requires_backends(self       ,						['vision']					)
      super().__init__(*__SCREAMING_SNAKE_CASE       ,						**__SCREAMING_SNAKE_CASE					)






    def    UpperCamelCase       (       self     :							List[Any]       ,						__SCREAMING_SNAKE_CASE     :							"Image"					)     ->		Dict:
      return self.pre_processor(images=__SCREAMING_SNAKE_CASE       ,						return_tensors='pt'					)






    def    UpperCamelCase       (       self     :							List[str]       ,						__SCREAMING_SNAKE_CASE     :							Tuple					)     ->		str:
      return self.model.generate(**__SCREAMING_SNAKE_CASE					)






    def    UpperCamelCase       (       self     :							Union[str, Any]       ,						__SCREAMING_SNAKE_CASE     :							List[str]					)     ->		Optional[Any]:
      return self.pre_processor.batch_decode(__SCREAMING_SNAKE_CASE       ,						skip_special_tokens=__SCREAMING_SNAKE_CASE					)[0].strip()




