
"""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 					lowerCamelCase   (     _snake_case ):
    UpperCAmelCase__	:      List[Any]     							=					botoa.client('iam' )

    UpperCAmelCase__	:      Optional[int]     							=					{
        '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=_snake_case    ,AssumeRolePolicyDocument=json.dumps(_snake_case    ,indent=2 ) )
        UpperCAmelCase__	:      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=_snake_case    ,PolicyName=F'''{role_name}_policy_permission'''    ,PolicyDocument=json.dumps(_snake_case    ,indent=2 )    ,)
    except iam_client.exceptions.EntityAlreadyExistsException:
        print(F'''role {role_name} already exists. Using existing one''' )
def 					lowerCamelCase   (     _snake_case ):
    UpperCAmelCase__	:      Dict     							=					botoa.client('iam' )
    return iam_client.get_role(RoleName=_snake_case )["Role"]["Arn"]

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

        UpperCAmelCase__	:      List[Any]     							=					_ask_field('AWS Secret Access Key: ' )
        UpperCAmelCase__	:      Optional[Any]     							=					aws_secret_access_key

    UpperCAmelCase__	:      Tuple     							=					_ask_field('Enter your AWS Region: [us-east-1]'    ,default='us-east-1' )
    UpperCAmelCase__	:      Union[str, Any]     							=					aws_region

    UpperCAmelCase__	:      Tuple     							=					_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']    ,_snake_case    ,)
    if role_management == 0:
        UpperCAmelCase__	:      Tuple     							=					_ask_field('Enter your IAM role name: ' )
    else:
        UpperCAmelCase__	:      List[Any]     							=					'accelerate_sagemaker_execution_role'
        print(F'''Accelerate will create an iam role "{iam_role_name}" using the provided credentials''' )
        _create_iam_role_for_sagemaker(_snake_case )

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

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

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

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

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

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

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



"""simple docstring"""



import logging
import random

import ray

from transformers import RagConfig, RagRetriever, RagTokenizer
from transformers.models.rag.retrieval_rag import CustomHFIndex


UpperCamelCase__          =	logging.getLogger(__name__)





class 				a			:



      def __init__(   self      ):
          UpperCAmelCase__	:      Union[str, Any]     							=					False



      def  __snake_case							(   self     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_      ):
          if not self.initialized:
              UpperCAmelCase__	:      Optional[Any]     							=					RagRetriever(
                  UpperCamelCase_     ,	question_encoder_tokenizer=UpperCamelCase_     ,	generator_tokenizer=UpperCamelCase_     ,	index=UpperCamelCase_     ,	init_retrieval=UpperCamelCase_     ,	)
              UpperCAmelCase__	:      Union[str, Any]     							=					True



      def  __snake_case							(   self      ):
          self.retriever.index.init_index()



      def  __snake_case							(   self     ,	UpperCamelCase_     ,	UpperCamelCase_      ):
          UpperCAmelCase__							,							UpperCAmelCase__	:      Union[str, Any]     							=					self.retriever._main_retrieve(UpperCamelCase_     ,	UpperCamelCase_      )
          return doc_ids, retrieved_doc_embeds





class 				a			(					lowercase      ):



      def __init__(   self     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_=None      ):
          if index is not None and index.is_initialized() and len(UpperCamelCase_      ) > 0:
              raise ValueError(
                  'When using Ray for distributed fine-tuning, '
                  'you\'ll need to provide the paths instead, '
                  'as the dataset and the index are loaded '
                  'separately. More info in examples/rag/use_own_knowledge_dataset.py '      )
          super().__init__(
              UpperCamelCase_     ,	question_encoder_tokenizer=UpperCamelCase_     ,	generator_tokenizer=UpperCamelCase_     ,	index=UpperCamelCase_     ,	init_retrieval=UpperCamelCase_     ,	)
          UpperCAmelCase__	:      Union[str, Any]     							=					retrieval_workers
          if len(self.retrieval_workers      ) > 0:
              ray.get(
                  [
                      worker.create_rag_retriever.remote(UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_      )
                      for worker in self.retrieval_workers
                  ]      )



      def  __snake_case							(   self      ):
          logger.info('initializing retrieval'      )

          if len(self.retrieval_workers      ) > 0:
              ray.get([worker.init_retrieval.remote() for worker in self.retrieval_workers]      )
          else:
              # Non-distributed training. Load index into this same process.
              self.index.init_index()



      def  __snake_case							(   self     ,	UpperCamelCase_     ,	UpperCamelCase_      ):
          if len(self.retrieval_workers      ) > 0:
              # Select a random retrieval actor.
              UpperCAmelCase__	:      Optional[Any]     							=					self.retrieval_workers[random.randint(0     ,	len(self.retrieval_workers      ) - 1      )]
              UpperCAmelCase__							,							UpperCAmelCase__	:      List[Any]     							=					ray.get(random_worker.retrieve.remote(UpperCamelCase_     ,	UpperCamelCase_      )      )
          else:
              UpperCAmelCase__							,							UpperCAmelCase__	:      Dict     							=					self._main_retrieve(UpperCamelCase_     ,	UpperCamelCase_      )
          return retrieved_doc_embeds, doc_ids, self.index.get_doc_dicts(UpperCamelCase_      )



      @classmethod
      def  __snake_case							(   cls     ,	UpperCamelCase_     ,	UpperCamelCase_=None     ,	**UpperCamelCase_      ):
          return super(UpperCamelCase_     ,	cls      ).get_tokenizers(UpperCamelCase_     ,	UpperCamelCase_     ,	**UpperCamelCase_      )



      @classmethod
      def  __snake_case							(   cls     ,	UpperCamelCase_     ,	UpperCamelCase_     ,	UpperCamelCase_=None     ,	**UpperCamelCase_      ):
          UpperCAmelCase__	:      Union[str, Any]     							=					kwargs.pop('config'     ,	UpperCamelCase_      ) or RagConfig.from_pretrained(UpperCamelCase_     ,	**UpperCamelCase_      )
          UpperCAmelCase__	:      Union[str, Any]     							=					RagTokenizer.from_pretrained(UpperCamelCase_     ,	config=UpperCamelCase_      )
          UpperCAmelCase__	:      List[Any]     							=					rag_tokenizer.question_encoder
          UpperCAmelCase__	:      Optional[Any]     							=					rag_tokenizer.generator
          if indexed_dataset is not None:
              UpperCAmelCase__	:      List[str]     							=					'custom'
              UpperCAmelCase__	:      Tuple     							=					CustomHFIndex(config.retrieval_vector_size     ,	UpperCamelCase_      )
          else:
              UpperCAmelCase__	:      Optional[Any]     							=					cls._build_index(UpperCamelCase_      )
          return cls(
              UpperCamelCase_     ,	question_encoder_tokenizer=UpperCamelCase_     ,	generator_tokenizer=UpperCamelCase_     ,	retrieval_workers=UpperCamelCase_     ,	index=UpperCamelCase_     ,	)

