





'''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__		(						__magic_name__  :   Optional[Any]      )				->			Dict:
 '''simple docstring'''






 snake_case__     :					int    =  botoa.client("""iam"""      )

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



def        UpperCamelCase__		(						__magic_name__  :   Any      )				->			Tuple:
 '''simple docstring'''






 snake_case__     :					List[str]    =  botoa.client("""iam"""      )
 return iam_client.get_role(RoleName=__magic_name__      )["Role"]["Arn"]





def        UpperCamelCase__		(						)				->			Tuple:
 '''simple docstring'''






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

  snake_case__     :					Optional[Any]    =  _ask_field("""AWS Secret Access Key: """      )
  snake_case__     :					List[str]    =  aws_secret_access_key

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

 snake_case__     :					int    =  _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"""]		,       __magic_name__		,       )
 if role_management == 0:
  snake_case__     :					Optional[Any]    =  _ask_field("""Enter your IAM role name: """      )
 else:
  snake_case__     :					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(__magic_name__      )

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

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

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

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

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

 snake_case__     :					Dict    =  1
 if distributed_type in (SageMakerDistributedType.DATA_PARALLEL, SageMakerDistributedType.MODEL_PARALLEL):
  snake_case__     :					Optional[Any]    =  _ask_field(
      """How many machines do you want use? [1]: """		,       __magic_name__		,       default=1		,       )

 snake_case__     :					Union[str, 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=__magic_name__		,       compute_environment=ComputeEnvironment.AMAZON_SAGEMAKER		,       distributed_type=__magic_name__		,       use_cpu=__magic_name__		,       dynamo_config=__magic_name__		,       eca_instance_type=__magic_name__		,       profile=__magic_name__		,       region=__magic_name__		,       iam_role_name=__magic_name__		,       mixed_precision=__magic_name__		,       num_machines=__magic_name__		,       sagemaker_inputs_file=__magic_name__		,       sagemaker_metrics_file=__magic_name__		,       )





"""simple docstring"""






import os
import unittest

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

from ...test_tokenization_common import TokenizerTesterMixin





class 					_A		(			lowerCAmelCase    ,       unittest.TestCase      ):
      snake_case__     :  str				    =      TransfoXLTokenizer
      snake_case__     :  Union[str, Any]				    =      False
      snake_case__     :  Union[str, Any]				    =      False

      def 							A__			(		self		):







            """simple docstring"""



            super().setUp()

            lowercase								=      [
                """<unk>""",
                """[CLS]""",
                """[SEP]""",
                """want""",
                """unwanted""",
                """wa""",
                """un""",
                """running""",
                """,""",
                """low""",
                """l""",
            ]
            lowercase								=      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 							A__			(		self						,			**__lowerCAmelCase		):







            """simple docstring"""



            lowercase								=      True
            return TransfoXLTokenizer.from_pretrained(self.tmpdirname						,			**__lowerCAmelCase		)

      def 							A__			(		self						,			__lowerCAmelCase		):







            """simple docstring"""



            lowercase								=      """<unk> UNwanted , running"""
            lowercase								=      """<unk> unwanted, running"""
            return input_text, output_text

      def 							A__			(		self		):







            """simple docstring"""



            lowercase								=      TransfoXLTokenizer(vocab_file=self.vocab_file						,			lower_case=__lowerCAmelCase		)

            lowercase								=      tokenizer.tokenize("""<unk> UNwanted , running"""		)
            self.assertListEqual(__lowerCAmelCase						,			["""<unk>""", """unwanted""", """,""", """running"""]		)

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

      def 							A__			(		self		):







            """simple docstring"""



            lowercase								=      TransfoXLTokenizer(lower_case=__lowerCAmelCase		)

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

      def 							A__			(		self		):







            """simple docstring"""



            lowercase								=      TransfoXLTokenizer(lower_case=__lowerCAmelCase		)

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

      def 							A__			(		self		):







            """simple docstring"""



            lowercase								=      TransfoXLTokenizer(lower_case=__lowerCAmelCase		)
            lowercase								=      """Hello (bracket) and side-scrolled [and] Henry's $5,000 with 3.34 m. What's up!?"""
            lowercase								=      [
                """Hello""",
                """(""",
                """bracket""",
                """)""",
                """and""",
                """side""",
                """@-@""",
                """scrolled""",
                """[""",
                """and""",
                """]""",
                """Henry""",
                """'s""",
                """$""",
                """5""",
                """@,@""",
                """000""",
                """with""",
                """3""",
                """@.@""",
                """34""",
                """m""",
                """.""",
                """What""",
                """'s""",
                """up""",
                """!""",
                """?""",
            ]

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

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







      def 							A__			(		self		):







            """simple docstring"""



            lowercase								=      self.get_tokenizer()
            lowercase								=      len(__lowerCAmelCase		)

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

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



