


"""simple docstring"""







from typing import Any, Dict, List, Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_vision_available():
		from PIL import Image

		from ..image_utils import load_image

if is_torch_available():
		import torch

		from transformers.modeling_outputs import BaseModelOutput

		from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING

a__       :							Optional[Any]        =  logging.get_logger(__name__)




@add_end_docstrings(UpperCamelCase)
class 				UpperCamelCase_			(				UpperCamelCase):

	"""simple docstring"""





	def __init__(				self    :       Any    ,    **UpperCAmelCase__    :       Optional[Any]     ) ->			Optional[int]:
						super().__init__(**UpperCAmelCase__     )

						if self.framework == "tf":
											raise ValueError(F"""The {self.__class__} is only available in PyTorch."""     )

						requires_backends(self    ,    "vision"     )
						self.check_model_type(UpperCAmelCase__     )




	def __call__(				self    :       str    ,    UpperCAmelCase__    :       Union[str, "Image.Image", List[Dict[str, Any]]]    ,    UpperCAmelCase__    :       Union[str, List[str]] = None    ,    **UpperCAmelCase__    :       Dict    ,    ) ->			List[Any]:
						if "text_queries" in kwargs:
											__SCREAMING_SNAKE_CASE		     =       kwargs.pop("text_queries"     )

						if isinstance(UpperCAmelCase__    ,    (str, Image.Image)     ):
											__SCREAMING_SNAKE_CASE		     =       {"image": image, "candidate_labels": candidate_labels}
						else:
											__SCREAMING_SNAKE_CASE		     =       image
						__SCREAMING_SNAKE_CASE		     =       super().__call__(UpperCAmelCase__    ,    **UpperCAmelCase__     )
						return results




	def 						UpperCAmelCase_   (				self    :       List[Any]    ,    **UpperCAmelCase__    :       Union[str, Any]     ) ->			List[Any]:
						__SCREAMING_SNAKE_CASE		     =       {}
						if "threshold" in kwargs:
											__SCREAMING_SNAKE_CASE		     =       kwargs["threshold"]
						if "top_k" in kwargs:
											__SCREAMING_SNAKE_CASE		     =       kwargs["top_k"]
						return {}, {}, postprocess_params




	def 						UpperCAmelCase_   (				self    :       int    ,    UpperCAmelCase__    :       Optional[Any]     ) ->			Optional[Any]:
						__SCREAMING_SNAKE_CASE		     =       load_image(inputs["image"]     )
						__SCREAMING_SNAKE_CASE		     =       inputs["candidate_labels"]
						if isinstance(UpperCAmelCase__    ,    UpperCAmelCase__     ):
											__SCREAMING_SNAKE_CASE		     =       candidate_labels.split(","     )

						__SCREAMING_SNAKE_CASE		     =       torch.tensor([[image.height, image.width]]    ,    dtype=torch.intaa     )
						for i, candidate_label in enumerate(UpperCAmelCase__     ):
											__SCREAMING_SNAKE_CASE		     =       self.tokenizer(UpperCAmelCase__    ,    return_tensors=self.framework     )
											__SCREAMING_SNAKE_CASE		     =       self.image_processor(UpperCAmelCase__    ,    return_tensors=self.framework     )
											yield {
											    "is_last": i == len(UpperCAmelCase__     ) - 1,
											    "target_size": target_size,
											    "candidate_label": candidate_label,
											    **text_inputs,
											    **image_features,
											}




	def 						UpperCAmelCase_   (				self    :       Optional[int]    ,    UpperCAmelCase__    :       Tuple     ) ->			int:
						__SCREAMING_SNAKE_CASE		     =       model_inputs.pop("target_size"     )
						__SCREAMING_SNAKE_CASE		     =       model_inputs.pop("candidate_label"     )
						__SCREAMING_SNAKE_CASE		     =       model_inputs.pop("is_last"     )

						__SCREAMING_SNAKE_CASE		     =       self.model(**UpperCAmelCase__     )

						__SCREAMING_SNAKE_CASE		     =       {"target_size": target_size, "candidate_label": candidate_label, "is_last": is_last, **outputs}
						return model_outputs




	def 						UpperCAmelCase_   (				self    :       Optional[int]    ,    UpperCAmelCase__    :       Tuple    ,    UpperCAmelCase__    :       List[str]=0.1    ,    UpperCAmelCase__    :       Optional[Any]=None     ) ->			List[str]:
						__SCREAMING_SNAKE_CASE		     =       []
						for model_output in model_outputs:
											__SCREAMING_SNAKE_CASE		     =       model_output["candidate_label"]
											__SCREAMING_SNAKE_CASE		     =       BaseModelOutput(UpperCAmelCase__     )
											__SCREAMING_SNAKE_CASE		     =       self.image_processor.post_process_object_detection(
											    outputs=UpperCAmelCase__    ,    threshold=UpperCAmelCase__    ,    target_sizes=model_output["target_size"]     )[0]

											for index in outputs["scores"].nonzero():
																__SCREAMING_SNAKE_CASE		     =       outputs["scores"][index].item()
																__SCREAMING_SNAKE_CASE		     =       self._get_bounding_box(outputs["boxes"][index][0]     )

																__SCREAMING_SNAKE_CASE		     =       {"score": score, "label": label, "box": box}
																results.append(UpperCAmelCase__     )

						__SCREAMING_SNAKE_CASE		     =       sorted(UpperCAmelCase__    ,    key=lambda UpperCAmelCase__     : x["score"]    ,    reverse=UpperCAmelCase__     )
						if top_k:
											__SCREAMING_SNAKE_CASE		     =       results[:top_k]

						return results




	def 						UpperCAmelCase_   (				self    :       Dict    ,    UpperCAmelCase__    :       "torch.Tensor"     ) ->			Dict[str, int]:
						if self.framework != "pt":
											raise ValueError("The ZeroShotObjectDetectionPipeline is only available in PyTorch."     )
						__SCREAMING_SNAKE_CASE     ,						__SCREAMING_SNAKE_CASE     ,						__SCREAMING_SNAKE_CASE     ,						__SCREAMING_SNAKE_CASE		     =       box.int().tolist()
						__SCREAMING_SNAKE_CASE		     =       {
						    "xmin": xmin,
						    "ymin": ymin,
						    "xmax": xmax,
						    "ymax": ymax,
						}
						return bbox




"""simple docstring"""







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.py",
            "model_name_or_path": "distilbert-base-cased",
            "instance_type": "ml.p3.16xlarge",
            "results": {"train_runtime": 650, "eval_accuracy": 0.7, "eval_loss": 0.6},
        },
        {
            "framework": "pytorch",
            "script": "run_ddp.py",
            "model_name_or_path": "distilbert-base-cased",
            "instance_type": "ml.p3.16xlarge",
            "results": {"train_runtime": 600, "eval_accuracy": 0.7, "eval_loss": 0.6},
        },
        {
            "framework": "tensorflow",
            "script": "run_tf_dist.py",
            "model_name_or_path": "distilbert-base-cased",
            "instance_type": "ml.p3.16xlarge",
            "results": {"train_runtime": 600, "eval_accuracy": 0.6, "eval_loss": 0.7},
        },
    ])
class 				UpperCamelCase_			(				unittest.TestCase):

	"""simple docstring"""





	def 						UpperCAmelCase_   (				self    :       Dict     ) ->			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=UpperCAmelCase__    ,    )
						assert hasattr(self    ,    "env"     )




	def 						UpperCAmelCase_   (				self    :       Tuple    ,    UpperCAmelCase__    :       int     ) ->			Any:
						__SCREAMING_SNAKE_CASE		     =       F"""{self.env.base_job_name}-{instance_count}-{'ddp' if 'ddp' in self.script else 'smd'}"""
						# distributed data settings
						__SCREAMING_SNAKE_CASE		     =       {"smdistributed": {"dataparallel": {"enabled": True}}} if self.script != "run_ddp.py" else None

						# 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=UpperCAmelCase__    ,    instance_count=UpperCAmelCase__    ,    instance_type=self.instance_type    ,    debugger_hook_config=UpperCAmelCase__    ,    hyperparameters={**self.env.distributed_hyperparameters, "model_name_or_path": self.model_name_or_path}    ,    metric_definitions=self.env.metric_definitions    ,    distribution=UpperCAmelCase__    ,    py_version="py36"    ,    )




	def 						UpperCAmelCase_   (				self    :       Optional[int]    ,    UpperCAmelCase__    :       Optional[int]     ) ->			Optional[Any]:
						TrainingJobAnalytics(UpperCAmelCase__     ).export_csv(F"""{self.env.test_path}/{job_name}_metrics.csv"""     )




	@parameterized.expand([(2,)]     )
	def 						UpperCAmelCase_   (				self    :       Dict    ,    UpperCAmelCase__    :       Optional[Any]     ) ->			List[str]:
						# create estimator
						__SCREAMING_SNAKE_CASE		     =       self.create_estimator(UpperCAmelCase__     )

						# 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"    ,    9_9_9_9_9_9     )
						)

						# 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__     )
