



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     _UpperCAmelCase			(       unittest.TestCase     ):




     def 		snake_case_		(				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=a__ ,						)
         assert hasattr(self ,						'''env''')




     def 		snake_case_		(				self ,						a__=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=a__ ,						instance_type=self.instance_type ,						debugger_hook_config=a__ ,						hyperparameters={**self.env.hyperparameters, '''model_name_or_path''': self.model_name_or_path} ,						metric_definitions=self.env.metric_definitions ,						py_version='''py36''' ,						)




     def 		snake_case_		(				self ,						a__):
         TrainingJobAnalytics(a__).export_csv(F"{self.env.test_path}/{job_name}_metrics.csv")




     def 		snake_case_		(				self):
         # create estimator
         A__		 =					self.create_estimator()

         # run training
         estimator.fit()

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

         # extract kpis
         A__		 =					list(result_metrics_df[result_metrics_df.metric_name == '''eval_accuracy''']['''value'''])
         A__		 =					list(result_metrics_df[result_metrics_df.metric_name == '''eval_loss''']['''value'''])
         # get train time from SageMaker job, this includes starting, preprocessing, stopping
         A__		 =					(
             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} ,						a__)





# 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     _UpperCAmelCase			(       A__     ):
     UpperCamelCase__		    =				'''Salesforce/blip-image-captioning-base'''
     UpperCamelCase__		    =				(
         '''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.'''
     )
     UpperCamelCase__		    =				'''image_captioner'''
     UpperCamelCase__		    =				AutoModelForVisionaSeq

     UpperCamelCase__		    =				['''image''']
     UpperCamelCase__		    =				['''text''']




     def __init__(				self ,						*a__ ,						**a__):
         requires_backends(self ,						['''vision'''])
         super().__init__(*a__ ,						**a__)




     def 		snake_case_		(				self ,						a__):
         return self.pre_processor(images=a__ ,						return_tensors='''pt''')




     def 		snake_case_		(				self ,						a__):
         return self.model.generate(**a__)




     def 		snake_case_		(				self ,						a__):
         return self.pre_processor.batch_decode(a__ ,						skip_special_tokens=a__)[0].strip()
