


"""simple docstring"""






import copy
import os
from typing import Union

from ...configuration_utils import PretrainedConfig
from ...models.auto.modeling_auto import MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
from ...utils import logging
from ..auto import CONFIG_MAPPING


_lowerCamelCase     =  logging.get_logger(__name__)

_lowerCamelCase     =  {
    'salesforce/blip2-opt-2.7b': 'https://huggingface.co/salesforce/blip2-opt-2.7b/resolve/main/config.json',
}


class 			lowerCamelCase_		(			lowercase   ):





				"""simple docstring"""







				_lowerCAmelCase      :      Dict      							=					"blip_2_vision_model"



				def __init__(						self       ,		UpperCAmelCase__=1408       ,		UpperCAmelCase__=6144       ,		UpperCAmelCase__=39       ,		UpperCAmelCase__=16       ,		UpperCAmelCase__=224       ,		UpperCAmelCase__=14       ,		UpperCAmelCase__="gelu"       ,		UpperCAmelCase__=0.00_001       ,		UpperCAmelCase__=0.0       ,		UpperCAmelCase__=1e-10       ,		UpperCAmelCase__=True       ,		**UpperCAmelCase__       ,		):
								super().__init__(**UpperCAmelCase__     )

								SCREAMING_SNAKE_CASE__     							=				hidden_size
								SCREAMING_SNAKE_CASE__     							=				intermediate_size
								SCREAMING_SNAKE_CASE__     							=				num_hidden_layers
								SCREAMING_SNAKE_CASE__     							=				num_attention_heads
								SCREAMING_SNAKE_CASE__     							=				patch_size
								SCREAMING_SNAKE_CASE__     							=				image_size
								SCREAMING_SNAKE_CASE__     							=				initializer_range
								SCREAMING_SNAKE_CASE__     							=				attention_dropout
								SCREAMING_SNAKE_CASE__     							=				layer_norm_eps
								SCREAMING_SNAKE_CASE__     							=				hidden_act
								SCREAMING_SNAKE_CASE__     							=				qkv_bias






				@classmethod
				def   lowerCAmelCase__		(						cls       ,		UpperCAmelCase__       ,		**UpperCAmelCase__     ):
								cls._set_token_in_kwargs(UpperCAmelCase__     )

								SCREAMING_SNAKE_CASE__							,      SCREAMING_SNAKE_CASE__     							=				cls.get_config_dict(UpperCAmelCase__       ,		**UpperCAmelCase__     )

								# get the vision config dict if we are loading from Blip2Config
								if config_dict.get("model_type"     ) == "blip-2":
												SCREAMING_SNAKE_CASE__     							=				config_dict["vision_config"]

								if "model_type" in config_dict and hasattr(cls       ,		"model_type"     ) and config_dict["model_type"] != cls.model_type:
												logger.warning(
												    f'''You are using a model of type {config_dict["model_type"]} to instantiate a model of type '''
												    f'''{cls.model_type}. This is not supported for all configurations of models and can yield errors.'''     )

								return cls.from_dict(UpperCAmelCase__       ,		**UpperCAmelCase__     )


class 			lowerCamelCase_		(			lowercase   ):





				"""simple docstring"""







				_lowerCAmelCase      :      Tuple      							=					"blip_2_qformer"



				def __init__(						self       ,		UpperCAmelCase__=3_0522       ,		UpperCAmelCase__=768       ,		UpperCAmelCase__=12       ,		UpperCAmelCase__=12       ,		UpperCAmelCase__=3072       ,		UpperCAmelCase__="gelu"       ,		UpperCAmelCase__=0.1       ,		UpperCAmelCase__=0.1       ,		UpperCAmelCase__=512       ,		UpperCAmelCase__=0.02       ,		UpperCAmelCase__=1e-12       ,		UpperCAmelCase__=0       ,		UpperCAmelCase__="absolute"       ,		UpperCAmelCase__=2       ,		UpperCAmelCase__=1408       ,		**UpperCAmelCase__       ,		):
								super().__init__(pad_token_id=UpperCAmelCase__       ,		**UpperCAmelCase__     )

								SCREAMING_SNAKE_CASE__     							=				vocab_size
								SCREAMING_SNAKE_CASE__     							=				hidden_size
								SCREAMING_SNAKE_CASE__     							=				num_hidden_layers
								SCREAMING_SNAKE_CASE__     							=				num_attention_heads
								SCREAMING_SNAKE_CASE__     							=				hidden_act
								SCREAMING_SNAKE_CASE__     							=				intermediate_size
								SCREAMING_SNAKE_CASE__     							=				hidden_dropout_prob
								SCREAMING_SNAKE_CASE__     							=				attention_probs_dropout_prob
								SCREAMING_SNAKE_CASE__     							=				max_position_embeddings
								SCREAMING_SNAKE_CASE__     							=				initializer_range
								SCREAMING_SNAKE_CASE__     							=				layer_norm_eps
								SCREAMING_SNAKE_CASE__     							=				position_embedding_type
								SCREAMING_SNAKE_CASE__     							=				cross_attention_frequency
								SCREAMING_SNAKE_CASE__     							=				encoder_hidden_size






				@classmethod
				def   lowerCAmelCase__		(						cls       ,		UpperCAmelCase__       ,		**UpperCAmelCase__     ):
								cls._set_token_in_kwargs(UpperCAmelCase__     )

								SCREAMING_SNAKE_CASE__							,      SCREAMING_SNAKE_CASE__     							=				cls.get_config_dict(UpperCAmelCase__       ,		**UpperCAmelCase__     )

								# get the qformer config dict if we are loading from Blip2Config
								if config_dict.get("model_type"     ) == "blip-2":
												SCREAMING_SNAKE_CASE__     							=				config_dict["qformer_config"]

								if "model_type" in config_dict and hasattr(cls       ,		"model_type"     ) and config_dict["model_type"] != cls.model_type:
												logger.warning(
												    f'''You are using a model of type {config_dict["model_type"]} to instantiate a model of type '''
												    f'''{cls.model_type}. This is not supported for all configurations of models and can yield errors.'''     )

								return cls.from_dict(UpperCAmelCase__       ,		**UpperCAmelCase__     )


class 			lowerCamelCase_		(			lowercase   ):





				"""simple docstring"""







				_lowerCAmelCase      :      str      							=					"blip-2"
				_lowerCAmelCase      :      List[Any]      							=					True



				def __init__(						self       ,		UpperCAmelCase__=None       ,		UpperCAmelCase__=None       ,		UpperCAmelCase__=None       ,		UpperCAmelCase__=32       ,		**UpperCAmelCase__     ):
								super().__init__(**UpperCAmelCase__     )

								if vision_config is None:
												SCREAMING_SNAKE_CASE__     							=				{}
												logger.info("vision_config is None. initializing the Blip2VisionConfig with default values."     )

								if qformer_config is None:
												SCREAMING_SNAKE_CASE__     							=				{}
												logger.info("qformer_config is None. Initializing the Blip2QFormerConfig with default values."     )

								if text_config is None:
												SCREAMING_SNAKE_CASE__     							=				{}
												logger.info("text_config is None. Initializing the text config with default values (`OPTConfig`)."     )

								SCREAMING_SNAKE_CASE__     							=				BlipaVisionConfig(**UpperCAmelCase__     )
								SCREAMING_SNAKE_CASE__     							=				BlipaQFormerConfig(**UpperCAmelCase__     )
								SCREAMING_SNAKE_CASE__     							=				text_config["model_type"] if "model_type" in text_config else "opt"
								SCREAMING_SNAKE_CASE__     							=				CONFIG_MAPPING[text_model_type](**UpperCAmelCase__     )

								SCREAMING_SNAKE_CASE__     							=				self.text_config.tie_word_embeddings
								SCREAMING_SNAKE_CASE__     							=				self.text_config.is_encoder_decoder

								SCREAMING_SNAKE_CASE__     							=				num_query_tokens
								SCREAMING_SNAKE_CASE__     							=				self.vision_config.hidden_size
								SCREAMING_SNAKE_CASE__     							=				self.text_config.model_type in MODEL_FOR_CAUSAL_LM_MAPPING_NAMES
								SCREAMING_SNAKE_CASE__     							=				1.0
								SCREAMING_SNAKE_CASE__     							=				0.02



				@classmethod
				def   lowerCAmelCase__		(						cls       ,		UpperCAmelCase__       ,		UpperCAmelCase__       ,		UpperCAmelCase__       ,		**UpperCAmelCase__       ,		):
								return cls(
								    vision_config=vision_config.to_dict()       ,		qformer_config=qformer_config.to_dict()       ,		text_config=text_config.to_dict()       ,		**UpperCAmelCase__       ,		)






				def   lowerCAmelCase__		(						self     ):
								SCREAMING_SNAKE_CASE__     							=				copy.deepcopy(self.__dict__     )
								SCREAMING_SNAKE_CASE__     							=				self.vision_config.to_dict()
								SCREAMING_SNAKE_CASE__     							=				self.qformer_config.to_dict()
								SCREAMING_SNAKE_CASE__     							=				self.text_config.to_dict()
								SCREAMING_SNAKE_CASE__     							=				self.__class__.model_type
								return output




"""simple docstring"""






import json
import os
import subprocess
import unittest
from ast import literal_eval

import pytest
from parameterized import parameterized_class

from . import is_sagemaker_available


if is_sagemaker_available():
	from sagemaker import Session, TrainingJobAnalytics
	from sagemaker.huggingface import HuggingFace


@pytest.mark.skipif(
    literal_eval(os.getenv("TEST_SAGEMAKER"       ,  "False"   )   ) is not True       ,  reason="Skipping test because should only be run when releasing minor transformers version"       ,  )
@pytest.mark.usefixtures("sm_env"   )
@parameterized_class(
    [
        {
            "framework": "pytorch",
            "script": "run_glue.py",
            "model_name_or_path": "distilbert-base-cased",
            "instance_type": "ml.g4dn.xlarge",
            "results": {"train_runtime": 650, "eval_accuracy": 0.6, "eval_loss": 0.9},
        },
        {
            "framework": "tensorflow",
            "script": "run_tf.py",
            "model_name_or_path": "distilbert-base-cased",
            "instance_type": "ml.g4dn.xlarge",
            "results": {"train_runtime": 600, "eval_accuracy": 0.3, "eval_loss": 0.9},
        },
    ]   )
class 			lowerCamelCase_		(			unittest.TestCase   ):





				"""simple docstring"""







				def   lowerCAmelCase__		(						self     ):
								if self.framework == "pytorch":
												subprocess.run(
												    f'''cp ./examples/pytorch/text-classification/run_glue.py {self.env.test_path}/run_glue.py'''.split()       ,		encoding="utf-8"       ,		check=UpperCAmelCase__       ,		)
								assert hasattr(self       ,		"env"     )



				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__=1     ):
								# creates estimator
								return HuggingFace(
								    entry_point=self.script       ,		source_dir=self.env.test_path       ,		role=self.env.role       ,		image_uri=self.env.image_uri       ,		base_job_name=f'''{self.env.base_job_name}-single'''       ,		instance_count=UpperCAmelCase__       ,		instance_type=self.instance_type       ,		debugger_hook_config=UpperCAmelCase__       ,		hyperparameters={**self.env.hyperparameters, "model_name_or_path": self.model_name_or_path}       ,		metric_definitions=self.env.metric_definitions       ,		py_version="py36"       ,		)



				def   lowerCAmelCase__		(						self       ,		UpperCAmelCase__     ):
								TrainingJobAnalytics(UpperCAmelCase__     ).export_csv(f'''{self.env.test_path}/{job_name}_metrics.csv'''     )






				def   lowerCAmelCase__		(						self     ):
								# create estimator
								SCREAMING_SNAKE_CASE__     							=				self.create_estimator()

								# run training
								estimator.fit()

								# result dataframe
								SCREAMING_SNAKE_CASE__     							=				TrainingJobAnalytics(estimator.latest_training_job.name     ).dataframe()

								# extract kpis
								SCREAMING_SNAKE_CASE__     							=				list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"]     )
								SCREAMING_SNAKE_CASE__     							=				list(result_metrics_df[result_metrics_df.metric_name == "eval_loss"]["value"]     )
								# get train time from SageMaker job, this includes starting, preprocessing, stopping
								SCREAMING_SNAKE_CASE__     							=				(
								    Session().describe_training_job(estimator.latest_training_job.name     ).get("TrainingTimeInSeconds"       ,		99_9999     )
								)

								# assert kpis
								assert train_runtime <= self.results["train_runtime"]
								assert all(t >= self.results["eval_accuracy"] for t in eval_accuracy     )
								assert all(t <= self.results["eval_loss"] for t in eval_loss     )

								# dump tests result into json file to share in PR
								with open(f'''{estimator.latest_training_job.name}.json'''       ,		"w"     ) as outfile:
												json.dump({"train_time": train_runtime, "eval_accuracy": eval_accuracy, "eval_loss": eval_loss}       ,		UpperCAmelCase__     )
