"""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 							SCREAMING_SNAKE_CASE     (			_lowerCamelCase						:       Tuple	)       ->      Optional[Any]:
  _lowerCAmelCase						:       List[Any]  			=     botoa.client("""iam"""	)

  _lowerCAmelCase						:       Dict  			=     {
      """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=_lowerCamelCase			,AssumeRolePolicyDocument=json.dumps(_lowerCamelCase			,indent=2	)	)
    _lowerCAmelCase						:       Union[str, Any]  			=     {
        """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=_lowerCamelCase			,PolicyName=f"{role_name}_policy_permission"			,PolicyDocument=json.dumps(_lowerCamelCase			,indent=2	)			,)
  except iam_client.exceptions.EntityAlreadyExistsException:
    print(f"role {role_name} already exists. Using existing one"	)





def 							SCREAMING_SNAKE_CASE     (			_lowerCamelCase						:       int	)       ->      Optional[int]:
  _lowerCAmelCase						:       List[str]  			=     botoa.client("""iam"""	)
  return iam_client.get_role(RoleName=_lowerCamelCase	)["Role"]["Arn"]






def 							SCREAMING_SNAKE_CASE     (			)       ->      int:
  _lowerCAmelCase						:       List[Any]  			=     _ask_options(
      """How do you want to authorize?"""			,["""AWS Profile""", """Credentials (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY) """]			,_lowerCamelCase			,)
  _lowerCAmelCase						:       Optional[Any]  			=     None
  if credentials_configuration == 0:
    _lowerCAmelCase						:       Union[str, Any]  			=     _ask_field("""Enter your AWS Profile name: [default] """			,default="""default"""	)
    _lowerCAmelCase						:       List[Any]  			=     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`"""	)
    _lowerCAmelCase						:       str  			=     _ask_field("""AWS Access Key ID: """	)
    _lowerCAmelCase						:       List[str]  			=     aws_access_key_id

    _lowerCAmelCase						:       List[Any]  			=     _ask_field("""AWS Secret Access Key: """	)
    _lowerCAmelCase						:       int  			=     aws_secret_access_key

  _lowerCAmelCase						:       Any  			=     _ask_field("""Enter your AWS Region: [us-east-1]"""			,default="""us-east-1"""	)
  _lowerCAmelCase						:       int  			=     aws_region

  _lowerCAmelCase						:       Union[str, Any]  			=     _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"""]			,_lowerCamelCase			,)
  if role_management == 0:
    _lowerCAmelCase						:       Any  			=     _ask_field("""Enter your IAM role name: """	)
  else:
    _lowerCAmelCase						:       Optional[int]  			=     """accelerate_sagemaker_execution_role"""
    print(f"Accelerate will create an iam role \"{iam_role_name}\" using the provided credentials"	)
    _create_iam_role_for_sagemaker(_lowerCamelCase	)

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

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

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

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

    if use_custom_options:
      _lowerCAmelCase						:       Dict  			=     _ask_options(
          """Which mode do you want to use?"""			,_lowerCamelCase			,lambda _lowerCamelCase	: TORCH_DYNAMO_MODES[int(_lowerCamelCase	)]			,default="""default"""			,)
      _lowerCAmelCase						:       List[Any]  			=     _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=_lowerCamelCase			,error_message="""Please enter yes or no."""			,)
      _lowerCAmelCase						:       Dict  			=     _ask_field(
          """Do you want to enable dynamic shape tracing? [yes/NO]: """			,_convert_yes_no_to_bool			,default=_lowerCamelCase			,error_message="""Please enter yes or no."""			,)
  _lowerCAmelCase						:       Any  			=     """Which EC2 instance type you want to use for your training?"""
  if distributed_type != SageMakerDistributedType.NO:
    _lowerCAmelCase						:       List[Any]  			=     _ask_options(
        _lowerCamelCase			,_lowerCamelCase			,lambda _lowerCamelCase	: SAGEMAKER_PARALLEL_EC2_INSTANCES[int(_lowerCamelCase	)]	)
  else:
    eca_instance_query += "? [ml.p3.2xlarge]:"
    _lowerCAmelCase						:       int  			=     _ask_field(_lowerCamelCase			,lambda _lowerCamelCase	: str(_lowerCamelCase	).lower()			,default="""ml.p3.2xlarge"""	)

  _lowerCAmelCase						:       int  			=     1
  if distributed_type in (SageMakerDistributedType.DATA_PARALLEL, SageMakerDistributedType.MODEL_PARALLEL):
    _lowerCAmelCase						:       List[Any]  			=     _ask_field(
        """How many machines do you want use? [1]: """			,_lowerCamelCase			,default=1			,)

  _lowerCAmelCase						:       Optional[Any]  			=     _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=_lowerCamelCase			,compute_environment=ComputeEnvironment.AMAZON_SAGEMAKER			,distributed_type=_lowerCamelCase			,use_cpu=_lowerCamelCase			,dynamo_config=_lowerCamelCase			,eca_instance_type=_lowerCamelCase			,profile=_lowerCamelCase			,region=_lowerCamelCase			,iam_role_name=_lowerCamelCase			,mixed_precision=_lowerCamelCase			,num_machines=_lowerCamelCase			,sagemaker_inputs_file=_lowerCamelCase			,sagemaker_metrics_file=_lowerCamelCase			,)









import os
import unittest

from transformers.models.transfo_xl.tokenization_transfo_xl import VOCAB_FILES_NAMES, TransfoXLTokenizer

from ...test_tokenization_common import TokenizerTesterMixin




class 					snake_case__						(       lowerCAmelCase_				,     unittest.TestCase     ):






     """simple docstring"""


     _SCREAMING_SNAKE_CASE						=	TransfoXLTokenizer
     _SCREAMING_SNAKE_CASE						=	False
     _SCREAMING_SNAKE_CASE						=	False






     def    lowercase_   (       self  :  Optional[int]					)   ->Any:
            super().setUp()

            snake_case__							:	Tuple   	=	[
                '<unk>',
                '[CLS]',
                '[SEP]',
                'want',
                'unwanted',
                'wa',
                'un',
                'running',
                ',',
                'low',
                'l',
            ]
            snake_case__							:	Any   	=	os.path.join(self.tmpdirname, VOCAB_FILES_NAMES['vocab_file']					)
            with open(self.vocab_file, 'w', encoding='utf-8'					) as vocab_writer:
                   vocab_writer.write(''.join([x + '\n' for x in vocab_tokens]					)					)






     def    lowercase_   (       self  :  Union[str, Any], **_snake_case  :  List[Any]					)   ->Dict:
            snake_case__							:	str   	=	True
            return TransfoXLTokenizer.from_pretrained(self.tmpdirname, **_snake_case					)






     def    lowercase_   (       self  :  Optional[Any], _snake_case  :  str					)   ->Dict:
            snake_case__							:	List[Any]   	=	'<unk> UNwanted , running'
            snake_case__							:	List[Any]   	=	'<unk> unwanted, running'
            return input_text, output_text






     def    lowercase_   (       self  :  List[Any]					)   ->Tuple:
            snake_case__							:	Dict   	=	TransfoXLTokenizer(vocab_file=self.vocab_file, lower_case=_snake_case					)

            snake_case__							:	str   	=	tokenizer.tokenize('<unk> UNwanted , running'					)
            self.assertListEqual(_snake_case, ['<unk>', 'unwanted', ',', 'running']					)

            self.assertListEqual(tokenizer.convert_tokens_to_ids(_snake_case					), [0, 4, 8, 7]					)






     def    lowercase_   (       self  :  List[str]					)   ->List[Any]:
            snake_case__							:	str   	=	TransfoXLTokenizer(lower_case=_snake_case					)

            self.assertListEqual(
                tokenizer.tokenize(' \tHeLLo ! how  \n Are yoU ?  '					), ['hello', '!', 'how', 'are', 'you', '?']					)






     def    lowercase_   (       self  :  Optional[int]					)   ->Optional[Any]:
            snake_case__							:	Optional[int]   	=	TransfoXLTokenizer(lower_case=_snake_case					)

            self.assertListEqual(
                tokenizer.tokenize(' \tHeLLo ! how  \n Are yoU ?  '					), ['HeLLo', '!', 'how', 'Are', 'yoU', '?']					)






     def    lowercase_   (       self  :  Optional[int]					)   ->Union[str, Any]:
            snake_case__							:	List[Any]   	=	TransfoXLTokenizer(lower_case=_snake_case					)
            snake_case__							:	Dict   	=	'Hello (bracket) and side-scrolled [and] Henry\'s $5,000 with 3.34 m. What\'s up!?'
            snake_case__							:	List[Any]   	=	[
                'Hello',
                '(',
                'bracket',
                ')',
                'and',
                'side',
                '@-@',
                'scrolled',
                '[',
                'and',
                ']',
                'Henry',
                '\'s',
                '$',
                '5',
                '@,@',
                '000',
                'with',
                '3',
                '@.@',
                '34',
                'm',
                '.',
                'What',
                '\'s',
                'up',
                '!',
                '?',
            ]

            self.assertListEqual(tokenizer.tokenize(_snake_case					), _snake_case					)

            self.assertEqual(tokenizer.convert_tokens_to_string(_snake_case					), _snake_case					)






     def    lowercase_   (       self  :  Dict					)   ->Any:
            snake_case__							:	Dict   	=	self.get_tokenizer()
            snake_case__							:	Optional[Any]   	=	len(_snake_case					)

            tokenizer.add_tokens(['new1', 'new2']					)
            tokenizer.move_added_token('new1', 1					)

            # Check that moved token is not copied (duplicate)
            self.assertEqual(len(_snake_case					), original_len + 2					)
            # Check that token is moved to specified id
            self.assertEqual(tokenizer.encode('new1'					), [1]					)
            self.assertEqual(tokenizer.decode([1]					), 'new1'					)






