


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

import pytest
from parameterized import parameterized, 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_model_parallelism.py",
            "model_name_or_path": "roberta-large",
            "instance_type": "ml.p3dn.24xlarge",
            "results": {"train_runtime": 1600, "eval_accuracy": 0.3, "eval_loss": 1.2},
        },
        {
            "framework": "pytorch",
            "script": "run_glue.py",
            "model_name_or_path": "roberta-large",
            "instance_type": "ml.p3dn.24xlarge",
            "results": {"train_runtime": 1600, "eval_accuracy": 0.3, "eval_loss": 1.2},
        },
    ]				)
class     a		(unittest.TestCase				):




       """simple docstring"""


       def    __snake_case (							self	:  Optional[int]    )     ->   Any:
              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=lowerCamelCase   ,     )
              assert hasattr(self   ,     "env"    )

       def    __snake_case (							self	:  Tuple   ,     lowerCamelCase	:  str    )     ->   List[str]:
              # configuration for running training on smdistributed Model Parallel
              __snake_case		:      int										=       {
                  "enabled": True,
                  "processes_per_host": 8,
              }
              __snake_case		:      Tuple										=       {
                  "enabled": True,
                  "parameters": {
                      "microbatches": 4,
                      "placement_strategy": "spread",
                      "pipeline": "interleaved",
                      "optimize": "speed",
                      "partitions": 4,
                      "ddp": True,
                  },
              }

              __snake_case		:      Dict										=       {"smdistributed": {"modelparallel": smp_options}, "mpi": mpi_options}

              __snake_case		:      List[Any]										=       "trainer" if self.script == "run_glue.py" else "smtrainer"
              # 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}-{instance_count}-smp-{name_extension}'   ,     instance_count=lowerCamelCase   ,     instance_type=self.instance_type   ,     debugger_hook_config=lowerCamelCase   ,     hyperparameters={
                      **self.env.hyperparameters,
                      "model_name_or_path": self.model_name_or_path,
                      "max_steps": 500,
                  }   ,     metric_definitions=self.env.metric_definitions   ,     distribution=lowerCamelCase   ,     py_version="py36"   ,     )

       def    __snake_case (							self	:  List[Any]   ,     lowerCamelCase	:  Optional[Any]    )     ->   Union[str, Any]:
              TrainingJobAnalytics(lowerCamelCase    ).export_csv(F'{self.env.test_path}/{job_name}_metrics.csv'    )






       @parameterized.expand([(1,)]    )
       def    __snake_case (							self	:  List[str]   ,     lowerCamelCase	:  Optional[int]    )     ->   Any:
              # create estimator
              __snake_case		:      List[str]										=       self.create_estimator(lowerCamelCase    )

              # run training
              estimator.fit()

              # result dataframe
              __snake_case		:      List[str]										=       TrainingJobAnalytics(estimator.latest_training_job.name    ).dataframe()

              # extract kpis
              __snake_case		:      Optional[Any]										=       list(result_metrics_df[result_metrics_df.metric_name == "eval_accuracy"]["value"]    )
              __snake_case		:      str										=       list(result_metrics_df[result_metrics_df.metric_name == "eval_loss"]["value"]    )
              # get train time from SageMaker job, this includes starting, preprocessing, stopping
              __snake_case		:      str										=       (
                  Session().describe_training_job(estimator.latest_training_job.name    ).get("TrainingTimeInSeconds"   ,     999999    )
              )

              # 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}   ,     lowerCamelCase    )





import warnings
from typing import List

import numpy as np

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding
from ...utils import is_flax_available, is_tf_available, is_torch_available

class     a		(_lowerCAmelCase				):




							"""simple docstring"""


							__UpperCAmelCase   :							List[str]					  =  ["image_processor", "tokenizer"]
							__UpperCAmelCase   :							str					  =  "OwlViTImageProcessor"
							__UpperCAmelCase   :							Dict					  =  ("CLIPTokenizer", "CLIPTokenizerFast")

							def __init__(							self	:  str   ,     lowerCamelCase	:  Any=None   ,     lowerCamelCase	:  Any=None   ,     **lowerCamelCase	:  Union[str, Any]    )     ->   List[Any]:
														__snake_case		:      List[Any]										=       None
														if "feature_extractor" in kwargs:
																					warnings.warn(
																					    "The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"
																					    " instead."   ,     lowerCamelCase   ,     )
																					__snake_case		:      List[Any]										=       kwargs.pop("feature_extractor"    )

														__snake_case		:      str										=       image_processor if image_processor is not None else feature_extractor
														if image_processor is None:
																					raise ValueError("You need to specify an `image_processor`."    )
														if tokenizer is None:
																					raise ValueError("You need to specify a `tokenizer`."    )

														super().__init__(lowerCamelCase   ,     lowerCamelCase    )

							def __call__(							self	:  Union[str, Any]   ,     lowerCamelCase	:  Tuple=None   ,     lowerCamelCase	:  int=None   ,     lowerCamelCase	:  Union[str, Any]=None   ,     lowerCamelCase	:  List[str]="max_length"   ,     lowerCamelCase	:  Dict="np"   ,     **lowerCamelCase	:  str    )     ->   List[Any]:

														if text is None and query_images is None and images is None:
																					raise ValueError(
																					    "You have to specify at least one text or query image or image. All three cannot be none."    )

														if text is not None:
																					if isinstance(lowerCamelCase   ,     lowerCamelCase    ) or (isinstance(lowerCamelCase   ,     lowerCamelCase    ) and not isinstance(text[0]   ,     lowerCamelCase    )):
																												__snake_case		:      Union[str, Any]										=       [self.tokenizer(lowerCamelCase   ,     padding=lowerCamelCase   ,     return_tensors=lowerCamelCase   ,     **lowerCamelCase    )]

																					elif isinstance(lowerCamelCase   ,     lowerCamelCase    ) and isinstance(text[0]   ,     lowerCamelCase    ):
																												__snake_case		:      Tuple										=       []

																												# Maximum number of queries across batch
																												__snake_case		:      str										=       max([len(lowerCamelCase    ) for t in text]    )

																												# Pad all batch samples to max number of text queries
																												for t in text:
																																			if len(lowerCamelCase    ) != max_num_queries:
																																										__snake_case		:      Dict										=       t + [" "] * (max_num_queries - len(lowerCamelCase    ))

																																			__snake_case		:      int										=       self.tokenizer(lowerCamelCase   ,     padding=lowerCamelCase   ,     return_tensors=lowerCamelCase   ,     **lowerCamelCase    )
																																			encodings.append(lowerCamelCase    )
																					else:
																												raise TypeError("Input text should be a string, a list of strings or a nested list of strings"    )

																					if return_tensors == "np":
																												__snake_case		:      Any										=       np.concatenate([encoding["input_ids"] for encoding in encodings]   ,     axis=0    )
																												__snake_case		:      Tuple										=       np.concatenate([encoding["attention_mask"] for encoding in encodings]   ,     axis=0    )

																					elif return_tensors == "jax" and is_flax_available():
																												import jax.numpy as jnp

																												__snake_case		:      List[Any]										=       jnp.concatenate([encoding["input_ids"] for encoding in encodings]   ,     axis=0    )
																												__snake_case		:      Any										=       jnp.concatenate([encoding["attention_mask"] for encoding in encodings]   ,     axis=0    )

																					elif return_tensors == "pt" and is_torch_available():
																												import torch

																												__snake_case		:      int										=       torch.cat([encoding["input_ids"] for encoding in encodings]   ,     dim=0    )
																												__snake_case		:      int										=       torch.cat([encoding["attention_mask"] for encoding in encodings]   ,     dim=0    )

																					elif return_tensors == "tf" and is_tf_available():
																												import tensorflow as tf

																												__snake_case		:      int										=       tf.stack([encoding["input_ids"] for encoding in encodings]   ,     axis=0    )
																												__snake_case		:      Dict										=       tf.stack([encoding["attention_mask"] for encoding in encodings]   ,     axis=0    )

																					else:
																												raise ValueError("Target return tensor type could not be returned"    )

																					__snake_case		:      Any										=       BatchEncoding()
																					__snake_case		:      Tuple										=       input_ids
																					__snake_case		:      int										=       attention_mask

														if query_images is not None:
																					__snake_case		:      List[Any]										=       BatchEncoding()
																					__snake_case		:      Union[str, Any]										=       self.image_processor(
																					    lowerCamelCase   ,     return_tensors=lowerCamelCase   ,     **lowerCamelCase    ).pixel_values
																					__snake_case		:      str										=       query_pixel_values

														if images is not None:
																					__snake_case		:      Optional[int]										=       self.image_processor(lowerCamelCase   ,     return_tensors=lowerCamelCase   ,     **lowerCamelCase    )

														if text is not None and images is not None:
																					__snake_case		:      List[str]										=       image_features.pixel_values
																					return encoding
														elif query_images is not None and images is not None:
																					__snake_case		:      int										=       image_features.pixel_values
																					return encoding
														elif text is not None or query_images is not None:
																					return encoding
														else:
																					return BatchEncoding(data=dict(**lowerCamelCase    )   ,     tensor_type=lowerCamelCase    )

							def    __snake_case (							self	:  Dict   ,     *lowerCamelCase	:  List[Any]   ,     **lowerCamelCase	:  Union[str, Any]    )     ->   str:
														return self.image_processor.post_process(*lowerCamelCase   ,     **lowerCamelCase    )

							def    __snake_case (							self	:  Union[str, Any]   ,     *lowerCamelCase	:  str   ,     **lowerCamelCase	:  List[str]    )     ->   Tuple:
														return self.image_processor.post_process_object_detection(*lowerCamelCase   ,     **lowerCamelCase    )

							def    __snake_case (							self	:  Optional[Any]   ,     *lowerCamelCase	:  Optional[Any]   ,     **lowerCamelCase	:  Optional[Any]    )     ->   Any:
														return self.image_processor.post_process_image_guided_detection(*lowerCamelCase   ,     **lowerCamelCase    )

							def    __snake_case (							self	:  List[Any]   ,     *lowerCamelCase	:  Tuple   ,     **lowerCamelCase	:  Optional[int]    )     ->   str:
														return self.tokenizer.batch_decode(*lowerCamelCase   ,     **lowerCamelCase    )

							def    __snake_case (							self	:  Union[str, Any]   ,     *lowerCamelCase	:  Tuple   ,     **lowerCamelCase	:  List[Any]    )     ->   Tuple:
														return self.tokenizer.decode(*lowerCamelCase   ,     **lowerCamelCase    )

							@property
							def    __snake_case (							self	:  Any    )     ->   Dict:
														warnings.warn(
														    "`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."   ,     lowerCamelCase   ,     )
														return self.image_processor_class






							@property
							def    __snake_case (							self	:  List[str]    )     ->   Union[str, Any]:
														warnings.warn(
														    "`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."   ,     lowerCamelCase   ,     )
														return self.image_processor

