

'''simple docstring'''




# Copyright 2021 The HuggingFace 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.
import json
import os

from ...utils.constants import SAGEMAKER_PARALLEL_EC2_INSTANCES, TORCH_DYNAMO_MODES
from ...utils.dataclasses import ComputeEnvironment, SageMakerDistributedType
from ...utils.imports import is_botoa_available
from .config_args import SageMakerConfig
from .config_utils import (
    DYNAMO_BACKENDS,
    _ask_field,
    _ask_options,
    _convert_dynamo_backend,
    _convert_mixed_precision,
    _convert_sagemaker_distributed_mode,
    _convert_yes_no_to_bool,
)


if is_botoa_available():
      import botoa  # noqa: F401




def 						UpperCamelCase	(			a			)				->     Union[str, Any]:






     '''simple docstring'''
     __magic_name__						       =    botoa.client('''iam'''			)

     __magic_name__						       =    {
         '''Version''': '''2012-10-17''',
         '''Statement''': [
             {'''Effect''': '''Allow''', '''Principal''': {'''Service''': '''sagemaker.amazonaws.com'''}, '''Action''': '''sts:AssumeRole'''}
         ],
     }
     try:
          # create the role, associated with the chosen trust policy
          iam_client.create_role(
              RoleName=a		,						AssumeRolePolicyDocument=json.dumps(a		,						indent=2			)			)
          __magic_name__						       =    {
              '''Version''': '''2012-10-17''',
              '''Statement''': [
                  {
                      '''Effect''': '''Allow''',
                      '''Action''': [
                          '''sagemaker:*''',
                          '''ecr:GetDownloadUrlForLayer''',
                          '''ecr:BatchGetImage''',
                          '''ecr:BatchCheckLayerAvailability''',
                          '''ecr:GetAuthorizationToken''',
                          '''cloudwatch:PutMetricData''',
                          '''cloudwatch:GetMetricData''',
                          '''cloudwatch:GetMetricStatistics''',
                          '''cloudwatch:ListMetrics''',
                          '''logs:CreateLogGroup''',
                          '''logs:CreateLogStream''',
                          '''logs:DescribeLogStreams''',
                          '''logs:PutLogEvents''',
                          '''logs:GetLogEvents''',
                          '''s3:CreateBucket''',
                          '''s3:ListBucket''',
                          '''s3:GetBucketLocation''',
                          '''s3:GetObject''',
                          '''s3:PutObject''',
                      ],
                      '''Resource''': '''*''',
                  }
              ],
          }
          # attach policy to role
          iam_client.put_role_policy(
              RoleName=a		,						PolicyName=F'''{role_name}_policy_permission'''		,						PolicyDocument=json.dumps(a		,						indent=2			)		,						)
     except iam_client.exceptions.EntityAlreadyExistsException:
          print(F'''role {role_name} already exists. Using existing one'''			)




def 						UpperCamelCase	(			a			)				->     int:






     '''simple docstring'''
     __magic_name__						       =    botoa.client('''iam'''			)
     return iam_client.get_role(RoleName=a			)["Role"]["Arn"]




def 						UpperCamelCase	(			)				->     int:






     '''simple docstring'''
     __magic_name__						       =    _ask_options(
         '''How do you want to authorize?'''		,						['''AWS Profile''', '''Credentials (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY) ''']		,						a		,						)
     __magic_name__						       =    None
     if credentials_configuration == 0:
          __magic_name__						       =    _ask_field('''Enter your AWS Profile name: [default] '''		,						default='''default'''			)
          __magic_name__						       =    aws_profile
     else:
          print(
              '''Note you will need to provide AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY when you launch you training script with,'''
              '''`accelerate launch --aws_access_key_id XXX --aws_secret_access_key YYY`'''			)
          __magic_name__						       =    _ask_field('''AWS Access Key ID: '''			)
          __magic_name__						       =    aws_access_key_id

          __magic_name__						       =    _ask_field('''AWS Secret Access Key: '''			)
          __magic_name__						       =    aws_secret_access_key

     __magic_name__						       =    _ask_field('''Enter your AWS Region: [us-east-1]'''		,						default='''us-east-1'''			)
     __magic_name__						       =    aws_region

     __magic_name__						       =    _ask_options(
         '''Do you already have an IAM Role for executing Amazon SageMaker Training Jobs?'''		,						['''Provide IAM Role name''', '''Create new IAM role using credentials''']		,						a		,						)
     if role_management == 0:
          __magic_name__						       =    _ask_field('''Enter your IAM role name: '''			)
     else:
          __magic_name__						       =    '''accelerate_sagemaker_execution_role'''
          print(F'''Accelerate will create an iam role "{iam_role_name}" using the provided credentials'''			)
          _create_iam_role_for_sagemaker(a			)

     __magic_name__						       =    _ask_field(
         '''Do you want to use custom Docker image? [yes/NO]: '''		,						_convert_yes_no_to_bool		,						default=a		,						error_message='''Please enter yes or no.'''		,						)
     __magic_name__						       =    None
     if is_custom_docker_image:
          __magic_name__						       =    _ask_field('''Enter your Docker image: '''		,						lambda a			: str(a			).lower()			)

     __magic_name__						       =    _ask_field(
         '''Do you want to provide SageMaker input channels with data locations? [yes/NO]: '''		,						_convert_yes_no_to_bool		,						default=a		,						error_message='''Please enter yes or no.'''		,						)
     __magic_name__						       =    None
     if is_sagemaker_inputs_enabled:
          __magic_name__						       =    _ask_field(
              '''Enter the path to the SageMaker inputs TSV file with columns (channel_name, data_location): '''		,						lambda a			: str(a			).lower()		,						)

     __magic_name__						       =    _ask_field(
         '''Do you want to enable SageMaker metrics? [yes/NO]: '''		,						_convert_yes_no_to_bool		,						default=a		,						error_message='''Please enter yes or no.'''		,						)
     __magic_name__						       =    None
     if is_sagemaker_metrics_enabled:
          __magic_name__						       =    _ask_field(
              '''Enter the path to the SageMaker metrics TSV file with columns (metric_name, metric_regex): '''		,						lambda a			: str(a			).lower()		,						)

     __magic_name__						       =    _ask_options(
         '''What is the distributed mode?'''		,						['''No distributed training''', '''Data parallelism''']		,						_convert_sagemaker_distributed_mode		,						)
     __magic_name__						       =    {}
     __magic_name__						       =    _ask_field(
         '''Do you wish to optimize your script with torch dynamo?[yes/NO]:'''		,						_convert_yes_no_to_bool		,						default=a		,						error_message='''Please enter yes or no.'''		,						)
     if use_dynamo:
          __magic_name__						       =    '''dynamo_'''
          __magic_name__						       =    _ask_options(
              '''Which dynamo backend would you like to use?'''		,						[x.lower() for x in DYNAMO_BACKENDS]		,						_convert_dynamo_backend		,						default=2		,						)
          __magic_name__						       =    _ask_field(
              '''Do you want to customize the defaults sent to torch.compile? [yes/NO]: '''		,						_convert_yes_no_to_bool		,						default=a		,						error_message='''Please enter yes or no.'''		,						)

          if use_custom_options:
               __magic_name__						       =    _ask_options(
                   '''Which mode do you want to use?'''		,						a		,						lambda a			: TORCH_DYNAMO_MODES[int(a			)]		,						default='''default'''		,						)
               __magic_name__						       =    _ask_field(
                   '''Do you want the fullgraph mode or it is ok to break model into several subgraphs? [yes/NO]: '''		,						_convert_yes_no_to_bool		,						default=a		,						error_message='''Please enter yes or no.'''		,						)
               __magic_name__						       =    _ask_field(
                   '''Do you want to enable dynamic shape tracing? [yes/NO]: '''		,						_convert_yes_no_to_bool		,						default=a		,						error_message='''Please enter yes or no.'''		,						)
     __magic_name__						       =    '''Which EC2 instance type you want to use for your training?'''
     if distributed_type != SageMakerDistributedType.NO:
          __magic_name__						       =    _ask_options(
              a		,						a		,						lambda a			: SAGEMAKER_PARALLEL_EC2_INSTANCES[int(a			)]			)
     else:
          eca_instance_query += "? [ml.p3.2xlarge]:"
          __magic_name__						       =    _ask_field(a		,						lambda a			: str(a			).lower()		,						default='''ml.p3.2xlarge'''			)

     __magic_name__						       =    1
     if distributed_type in (SageMakerDistributedType.DATA_PARALLEL, SageMakerDistributedType.MODEL_PARALLEL):
          __magic_name__						       =    _ask_field(
              '''How many machines do you want use? [1]: '''		,						a		,						default=1		,						)

     __magic_name__						       =    _ask_options(
         '''Do you wish to use FP16 or BF16 (mixed precision)?'''		,						['''no''', '''fp16''', '''bf16''', '''fp8''']		,						_convert_mixed_precision		,						)

     if use_dynamo and mixed_precision == "no":
          print(
              '''Torch dynamo used without mixed precision requires TF32 to be efficient. Accelerate will enable it by default when launching your scripts.'''			)

     return SageMakerConfig(
         image_uri=a		,						compute_environment=ComputeEnvironment.AMAZON_SAGEMAKER		,						distributed_type=a		,						use_cpu=a		,						dynamo_config=a		,						eca_instance_type=a		,						profile=a		,						region=a		,						iam_role_name=a		,						mixed_precision=a		,						num_machines=a		,						sagemaker_inputs_file=a		,						sagemaker_metrics_file=a		,						)









'''simple docstring'''




import unittest

from transformers import (
    MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
    TextaTextGenerationPipeline,
    pipeline,
)
from transformers.testing_utils import is_pipeline_test, require_tf, require_torch
from transformers.utils import is_torch_available

from .test_pipelines_common import ANY


if is_torch_available():
      import torch






@is_pipeline_test
class   _SCREAMING_SNAKE_CASE   (   unittest.TestCase     ):
 __SCREAMING_SNAKE_CASE					:List[Any]      	=					MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING
 __SCREAMING_SNAKE_CASE					:Any      	=					TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING



 def 			snake_case__  (   self		:   Tuple			,				a__		:   Tuple			,				a__		:   int			,				a__		:   int      ):
      __magic_name__						       =    TextaTextGenerationPipeline(model=a__			,				tokenizer=a__      )
      return generator, ["Something to write", "Something else"]



 def 			snake_case__  (   self		:   List[str]			,				a__		:   List[Any]			,				a__		:   List[str]      ):
      __magic_name__						       =    generator('''Something there'''      )
      self.assertEqual(a__			,				[{'''generated_text''': ANY(a__      )}]      )
      # These are encoder decoder, they don't just append to incoming string
      self.assertFalse(outputs[0]['''generated_text'''].startswith('''Something there'''      )      )

      __magic_name__						       =    generator(['''This is great !''', '''Something else''']			,				num_return_sequences=2			,				do_sample=a__      )
      self.assertEqual(
          a__			,				[
              [{'''generated_text''': ANY(a__      )}, {'''generated_text''': ANY(a__      )}],
              [{'''generated_text''': ANY(a__      )}, {'''generated_text''': ANY(a__      )}],
          ]			,				)

      __magic_name__						       =    generator(
          ['''This is great !''', '''Something else''']			,				num_return_sequences=2			,				batch_size=2			,				do_sample=a__      )
      self.assertEqual(
          a__			,				[
              [{'''generated_text''': ANY(a__      )}, {'''generated_text''': ANY(a__      )}],
              [{'''generated_text''': ANY(a__      )}, {'''generated_text''': ANY(a__      )}],
          ]			,				)

      with self.assertRaises(a__      ):
           generator(4      )



 @require_torch
 def 			snake_case__  (   self		:   Any      ):
      __magic_name__						       =    pipeline('''text2text-generation'''			,				model='''patrickvonplaten/t5-tiny-random'''			,				framework='''pt'''      )
      # do_sample=False necessary for reproducibility
      __magic_name__						       =    generator('''Something there'''			,				do_sample=a__      )
      self.assertEqual(a__			,				[{'''generated_text''': ''''''}]      )

      __magic_name__						       =    3
      __magic_name__						       =    generator(
          '''Something there'''			,				num_return_sequences=a__			,				num_beams=a__			,				)
      __magic_name__						       =    [
          {'''generated_text''': '''Beide Beide Beide Beide Beide Beide Beide Beide Beide'''},
          {'''generated_text''': '''Beide Beide Beide Beide Beide Beide Beide Beide'''},
          {'''generated_text''': ''''''},
      ]
      self.assertEqual(a__			,				a__      )

      __magic_name__						       =    generator('''This is a test'''			,				do_sample=a__			,				num_return_sequences=2			,				return_tensors=a__      )
      self.assertEqual(
          a__			,				[
              {'''generated_token_ids''': ANY(torch.Tensor      )},
              {'''generated_token_ids''': ANY(torch.Tensor      )},
          ]			,				)
      __magic_name__						       =    generator.model.config.eos_token_id
      __magic_name__						       =    '''<pad>'''
      __magic_name__						       =    generator(
          ['''This is a test''', '''This is a second test''']			,				do_sample=a__			,				num_return_sequences=2			,				batch_size=2			,				return_tensors=a__			,				)
      self.assertEqual(
          a__			,				[
              [
                  {'''generated_token_ids''': ANY(torch.Tensor      )},
                  {'''generated_token_ids''': ANY(torch.Tensor      )},
              ],
              [
                  {'''generated_token_ids''': ANY(torch.Tensor      )},
                  {'''generated_token_ids''': ANY(torch.Tensor      )},
              ],
          ]			,				)



 @require_tf
 def 			snake_case__  (   self		:   int      ):
      __magic_name__						       =    pipeline('''text2text-generation'''			,				model='''patrickvonplaten/t5-tiny-random'''			,				framework='''tf'''      )
      # do_sample=False necessary for reproducibility
      __magic_name__						       =    generator('''Something there'''			,				do_sample=a__      )
      self.assertEqual(a__			,				[{'''generated_text''': ''''''}]      )






