




'''simple docstring'''





import json
import os
import shutil
import sys
import tempfile
import unittest
from unittest import TestCase
from unittest.mock import patch

import faiss
import numpy as np
from datasets import Dataset

from transformers import BartConfig, BartTokenizer, DPRConfig, DPRQuestionEncoderTokenizer, RagConfig
from transformers.file_utils import is_datasets_available, is_faiss_available, is_psutil_available, is_torch_available
from transformers.integrations import is_ray_available
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES as DPR_VOCAB_FILES_NAMES
from transformers.models.rag.retrieval_rag import CustomHFIndex, RagRetriever
from transformers.models.roberta.tokenization_roberta import VOCAB_FILES_NAMES as BART_VOCAB_FILES_NAMES
from transformers.testing_utils import require_ray


sys.path.append(os.path.join(os.getcwd()))  # noqa: E402 # noqa: E402 # isort:skip

if is_torch_available():
  from distributed_pytorch_retriever import RagPyTorchDistributedRetriever  # noqa: E402 # isort:skip
else:
  a		:	List[Any]       =      None

if is_ray_available():
  import ray  # noqa: E402 # isort:skip
  from distributed_ray_retriever import RagRayDistributedRetriever, RayRetriever  # noqa: E402 # isort:skip
else:
  a		:	Optional[int]       =      None
  a		:	Tuple       =      None
  a		:	Tuple       =      None





def UpperCAmelCase    (       UpperCAmelCase_						):
    if not (is_datasets_available() and is_faiss_available() and is_psutil_available()):
        _a				:							Tuple							       =      unittest.skip("""test requires Datasets, Faiss, psutil"""						)(UpperCAmelCase_						)
    return test_case





@require_distributed_retrieval
class      UpperCamelCase       (	snake_case__       ):



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Union[str, Any]							       =      tempfile.mkdtemp()
      _a				:							Optional[Any]							       =      8

      # DPR tok
      _a				:							int							       =      [
          """[UNK]""",
          """[CLS]""",
          """[SEP]""",
          """[PAD]""",
          """[MASK]""",
          """want""",
          """##want""",
          """##ed""",
          """wa""",
          """un""",
          """runn""",
          """##ing""",
          """,""",
          """low""",
          """lowest""",
      ]
      _a				:							Tuple							       =      os.path.join(self.tmpdirname		,     """dpr_tokenizer""" )
      os.makedirs(__lowerCAmelCase		,     exist_ok=__lowerCAmelCase )
      _a				:							Any							       =      os.path.join(__lowerCAmelCase		,     DPR_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] ) )

      # BART tok
      _a				:							Union[str, Any]							       =      [
          """l""",
          """o""",
          """w""",
          """e""",
          """r""",
          """s""",
          """t""",
          """i""",
          """d""",
          """n""",
          """\u0120""",
          """\u0120l""",
          """\u0120n""",
          """\u0120lo""",
          """\u0120low""",
          """er""",
          """\u0120lowest""",
          """\u0120newer""",
          """\u0120wider""",
          """<unk>""",
      ]
      _a				:							Tuple							       =      dict(zip(__lowerCAmelCase		,     range(len(__lowerCAmelCase ) ) ) )
      _a				:							Any							       =      ["""#version: 0.2""", """\u0120 l""", """\u0120l o""", """\u0120lo w""", """e r""", """"""]
      _a				:							Tuple							       =      {"""unk_token""": """<unk>"""}

      _a				:							Optional[Any]							       =      os.path.join(self.tmpdirname		,     """bart_tokenizer""" )
      os.makedirs(__lowerCAmelCase		,     exist_ok=__lowerCAmelCase )
      _a				:							List[Any]							       =      os.path.join(__lowerCAmelCase		,     BART_VOCAB_FILES_NAMES["""vocab_file"""] )
      _a				:							Union[str, Any]							       =      os.path.join(__lowerCAmelCase		,     BART_VOCAB_FILES_NAMES["""merges_file"""] )
      with open(self.vocab_file		,     """w"""		,     encoding="""utf-8""" ) as fp:
          fp.write(json.dumps(__lowerCAmelCase ) + """\n""" )
      with open(self.merges_file		,     """w"""		,     encoding="""utf-8""" ) as fp:
          fp.write("""\n""".join(__lowerCAmelCase ) )



  def     a_							(			self ):







      '''simple docstring'''


      return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname		,     """dpr_tokenizer""" ) )



  def     a_							(			self ):







      '''simple docstring'''


      return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname		,     """bart_tokenizer""" ) )



  def     a_							(			self ):







      '''simple docstring'''


      shutil.rmtree(self.tmpdirname )



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Tuple							       =      Dataset.from_dict(
          {
              """id""": ["""0""", """1"""],
              """text""": ["""foo""", """bar"""],
              """title""": ["""Foo""", """Bar"""],
              """embeddings""": [np.ones(self.retrieval_vector_size ), 2 * np.ones(self.retrieval_vector_size )],
          } )
      dataset.add_faiss_index("""embeddings"""		,     string_factory="""Flat"""		,     metric_type=faiss.METRIC_INNER_PRODUCT )
      return dataset



  def     a_							(			self		,     __lowerCAmelCase		,     __lowerCAmelCase=1_2345 ):







      '''simple docstring'''


      _a				:							str							       =      self.get_dummy_dataset()
      _a				:							Any							       =      RagConfig(
          retrieval_vector_size=self.retrieval_vector_size		,     question_encoder=DPRConfig().to_dict()		,     generator=BartConfig().to_dict()		,     )
      with patch("""transformers.models.rag.retrieval_rag.load_dataset""" ) as mock_load_dataset:
          _a				:							Optional[Any]							       =      dataset
          _a				:							Tuple							       =      RagPyTorchDistributedRetriever(
              __lowerCAmelCase		,     question_encoder_tokenizer=self.get_dpr_tokenizer()		,     generator_tokenizer=self.get_bart_tokenizer()		,     )
          if init_retrieval:
              retriever.init_retrieval(__lowerCAmelCase )
      return retriever



  def     a_							(			self		,     __lowerCAmelCase ):







      '''simple docstring'''


      ray.init(local_mode=__lowerCAmelCase )
      _a				:							Union[str, Any]							       =      RagConfig(
          retrieval_vector_size=self.retrieval_vector_size		,     question_encoder=DPRConfig().to_dict()		,     generator=BartConfig().to_dict()		,     )
      _a				:							List[str]							       =      ray.remote(__lowerCAmelCase )
      _a				:							Any							       =      [remote_cls.remote() for _ in range(1 )]
      with patch("""transformers.models.rag.retrieval_rag.load_dataset""" ) as mock_load_dataset:
          _a				:							str							       =      self.get_dummy_dataset()
          _a				:							int							       =      RagRayDistributedRetriever(
              __lowerCAmelCase		,     question_encoder_tokenizer=self.get_dpr_tokenizer()		,     generator_tokenizer=self.get_bart_tokenizer()		,     retrieval_workers=__lowerCAmelCase		,     )
          if init_retrieval:
              retriever.init_retrieval()
      return retriever



  def     a_							(			self		,     __lowerCAmelCase		,     __lowerCAmelCase		,     __lowerCAmelCase=1_2345 ):







      '''simple docstring'''


      _a				:							Optional[int]							       =      self.get_dummy_dataset()
      _a				:							List[Any]							       =      RagConfig(
          retrieval_vector_size=self.retrieval_vector_size		,     question_encoder=DPRConfig().to_dict()		,     generator=BartConfig().to_dict()		,     index_name="""custom"""		,     )
      if from_disk:
          _a				:							Any							       =      os.path.join(self.tmpdirname		,     """dataset""" )
          _a				:							str							       =      os.path.join(self.tmpdirname		,     """index.faiss""" )
          dataset.get_index("""embeddings""" ).save(os.path.join(self.tmpdirname		,     """index.faiss""" ) )
          dataset.drop_index("""embeddings""" )
          dataset.save_to_disk(os.path.join(self.tmpdirname		,     """dataset""" ) )
          del dataset
          _a				:							List[Any]							       =      RagPyTorchDistributedRetriever(
              __lowerCAmelCase		,     question_encoder_tokenizer=self.get_dpr_tokenizer()		,     generator_tokenizer=self.get_bart_tokenizer()		,     )
      else:
          _a				:							Any							       =      RagPyTorchDistributedRetriever(
              __lowerCAmelCase		,     question_encoder_tokenizer=self.get_dpr_tokenizer()		,     generator_tokenizer=self.get_bart_tokenizer()		,     index=CustomHFIndex(config.retrieval_vector_size		,     __lowerCAmelCase )		,     )
      if init_retrieval:
          retriever.init_retrieval(__lowerCAmelCase )
      return retriever



  def     a_							(			self		,     __lowerCAmelCase		,     __lowerCAmelCase ):







      '''simple docstring'''


      ray.init(local_mode=__lowerCAmelCase )
      _a				:							Any							       =      self.get_dummy_dataset()
      _a				:							str							       =      RagConfig(
          retrieval_vector_size=self.retrieval_vector_size		,     question_encoder=DPRConfig().to_dict()		,     generator=BartConfig().to_dict()		,     index_name="""custom"""		,     )
      _a				:							int							       =      ray.remote(__lowerCAmelCase )
      _a				:							Optional[Any]							       =      [remote_cls.remote() for _ in range(1 )]
      if from_disk:
          _a				:							List[Any]							       =      os.path.join(self.tmpdirname		,     """dataset""" )
          _a				:							Optional[int]							       =      os.path.join(self.tmpdirname		,     """index.faiss""" )
          dataset.get_index("""embeddings""" ).save(os.path.join(self.tmpdirname		,     """index.faiss""" ) )
          dataset.drop_index("""embeddings""" )
          dataset.save_to_disk(os.path.join(self.tmpdirname		,     """dataset""" ) )
          del dataset
          _a				:							Optional[int]							       =      RagRayDistributedRetriever(
              __lowerCAmelCase		,     question_encoder_tokenizer=self.get_dpr_tokenizer()		,     generator_tokenizer=self.get_bart_tokenizer()		,     retrieval_workers=__lowerCAmelCase		,     index=CustomHFIndex.load_from_disk(
                  vector_size=config.retrieval_vector_size		,     dataset_path=config.passages_path		,     index_path=config.index_path		,     )		,     )
      else:
          _a				:							Tuple							       =      RagRayDistributedRetriever(
              __lowerCAmelCase		,     question_encoder_tokenizer=self.get_dpr_tokenizer()		,     generator_tokenizer=self.get_bart_tokenizer()		,     retrieval_workers=__lowerCAmelCase		,     index=CustomHFIndex(config.retrieval_vector_size		,     __lowerCAmelCase )		,     )
      if init_retrieval:
          retriever.init_retrieval()
      return retriever



  def     a_							(			self		,     __lowerCAmelCase		,     __lowerCAmelCase		,     __lowerCAmelCase ):







      '''simple docstring'''


      _a       , _a       , _a				:							Union[str, Any]							       =      retriever.retrieve(__lowerCAmelCase		,     n_docs=__lowerCAmelCase )
      self.assertEqual(retrieved_doc_embeds.shape		,     (2, n_docs, self.retrieval_vector_size) )
      self.assertEqual(len(__lowerCAmelCase )		,     2 )
      self.assertEqual(sorted(doc_dicts[0] )		,     ["""embeddings""", """id""", """text""", """title"""] )
      self.assertEqual(len(doc_dicts[0]["""id"""] )		,     __lowerCAmelCase )
      self.assertEqual(doc_dicts[0]["""id"""][0]		,     """1""" )  # max inner product is reached with second doc
      self.assertEqual(doc_dicts[1]["""id"""][0]		,     """0""" )  # max inner product is reached with first doc
      self.assertListEqual(doc_ids.tolist()		,     [[1], [0]] )



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Dict							       =      1
      _a				:							Any							       =      np.array(
          [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]		,     dtype=np.floataa )

      self.distributed_retriever_check(
          self.get_dummy_pytorch_distributed_retriever(init_retrieval=__lowerCAmelCase )		,     __lowerCAmelCase		,     __lowerCAmelCase )



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Dict							       =      1
      _a				:							int							       =      np.array(
          [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]		,     dtype=np.floataa )

      self.distributed_retriever_check(
          self.get_dummy_custom_hf_index_pytorch_retriever(init_retrieval=__lowerCAmelCase		,     from_disk=__lowerCAmelCase )		,     __lowerCAmelCase		,     __lowerCAmelCase		,     )



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Tuple							       =      1
      _a				:							int							       =      np.array(
          [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]		,     dtype=np.floataa )

      self.distributed_retriever_check(
          self.get_dummy_custom_hf_index_pytorch_retriever(init_retrieval=__lowerCAmelCase		,     from_disk=__lowerCAmelCase )		,     __lowerCAmelCase		,     __lowerCAmelCase		,     )



  @require_ray
  def     a_							(			self ):







      '''simple docstring'''


      _a				:							str							       =      1
      _a				:							Union[str, Any]							       =      np.array(
          [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]		,     dtype=np.floataa )

      self.distributed_retriever_check(
          self.get_dummy_ray_distributed_retriever(init_retrieval=__lowerCAmelCase )		,     __lowerCAmelCase		,     __lowerCAmelCase )
      ray.shutdown()



  @require_ray
  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Dict							       =      1
      _a				:							Optional[int]							       =      np.array(
          [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]		,     dtype=np.floataa )
      with self.assertRaises(__lowerCAmelCase ):
          self.distributed_retriever_check(
              self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=__lowerCAmelCase		,     from_disk=__lowerCAmelCase )		,     __lowerCAmelCase		,     __lowerCAmelCase		,     )
      ray.shutdown()



  @require_ray
  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Tuple							       =      1
      _a				:							Optional[Any]							       =      np.array(
          [np.ones(self.retrieval_vector_size ), -np.ones(self.retrieval_vector_size )]		,     dtype=np.floataa )

      self.distributed_retriever_check(
          self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=__lowerCAmelCase		,     from_disk=__lowerCAmelCase )		,     __lowerCAmelCase		,     __lowerCAmelCase )
      ray.shutdown()








'''simple docstring'''






import timeit
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_auto import MODEL_MAPPING, MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_torch_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_torch_available():
      import torch

      from .benchmark_args import PyTorchBenchmarkArguments


if is_pyanvml_available():
      import pyanvml.pyanvml as nvml


lowercase						:   List[Any]      						=							logging.get_logger(__name__)

class 	SCREAMING_SNAKE_CASE_		(							lowerCamelCase_					):
 __snake_case						:  PyTorchBenchmarkArguments
 __snake_case						:  PretrainedConfig
 __snake_case						:  str       =   "PyTorch"



 @property
 def  SCREAMING_SNAKE_CASE(	self			):






        '''simple docstring'''




        return torch.__version__



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		str    =   self._prepare_inference_func(A ,		A ,		A			)
        return self._measure_speed(_inference			)



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		int    =   self._prepare_inference_func(A ,		A ,		A			)
        return self._measure_memory(_inference			)



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		Union[str, Any]    =   self._prepare_train_func(A ,		A ,		A			)
        return self._measure_speed(_train			)



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		Any    =   self._prepare_train_func(A ,		A ,		A			)
        return self._measure_memory(_train			)



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		Any    =   self.config_dict[model_name]

        if self.args.torchscript:
               lowercase	:		Any    =   True

        lowercase	:		Union[str, Any]    =   (
            hasattr(A ,		'''architectures'''			)
            and isinstance(config.architectures ,		A			)
            and len(config.architectures			) > 0
        )
        if not self.args.only_pretrain_model and has_model_class_in_config:
               try:
                      lowercase	:		Any    =   config.architectures[0]
                      lowercase	:		Dict    =   __import__('''transformers''' ,		fromlist=[model_class]			)
                      lowercase	:		List[str]    =   getattr(A ,		A			)
                      lowercase	:		Dict    =   model_cls(A			)
               except ImportError:
                      raise ImportError(
                          f"""{model_class} does not exist. If you just want to test the pretrained model, you might want to"""
                          ''' set `--only_pretrain_model` or `args.only_pretrain_model=True`.'''			)
        else:
               lowercase	:		Optional[int]    =   MODEL_MAPPING[config.__class__](A			)

        model.eval()
        model.to(self.args.device			)

        # encoder-decoder has vocab size saved differently
        lowercase	:		str    =   config.vocab_size if hasattr(A ,		'''vocab_size'''			) else config.encoder.vocab_size
        lowercase	:		Union[str, Any]    =   torch.randint(A ,		(batch_size, sequence_length) ,		dtype=torch.long ,		device=self.args.device			)

        if self.args.fpaa:
               logger.info('''Running training in Mixed Precision...'''			)
               if not self.args.is_gpu:
                      raise ValueError('''Mixed precision is possible only for GPU.'''			)
               # amp seems to have memory leaks so that memory usage
               # is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
               model.half()

        if self.args.torchscript:
               with torch.no_grad():
                      lowercase	:		int    =   torch.jit.trace(A ,		A			)
        else:
               lowercase	:		Any    =   model

        def encoder_decoder_forward():
               with torch.no_grad():
                      lowercase	:		int    =   inference_model(A ,		decoder_input_ids=A			)
               return outputs

        def encoder_forward():
               with torch.no_grad():
                      lowercase	:		Tuple    =   inference_model(A			)
               return outputs

        lowercase	:		int    =   encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
        return _forward



 def  SCREAMING_SNAKE_CASE(	self ,		A ,		A ,		A			):






        '''simple docstring'''




        lowercase	:		Optional[int]    =   self.config_dict[model_name]

        lowercase	:		Optional[Any]    =   (
            hasattr(A ,		'''architectures'''			)
            and isinstance(config.architectures ,		A			)
            and len(config.architectures			) > 0
        )
        if not self.args.only_pretrain_model and has_model_class_in_config:
               try:
                      lowercase	:		str    =   config.architectures[0]
                      lowercase	:		List[str]    =   __import__('''transformers''' ,		fromlist=[model_class]			)
                      lowercase	:		int    =   getattr(A ,		A			)
                      lowercase	:		Any    =   model_cls(A			)
               except ImportError:
                      raise ImportError(
                          f"""{model_class} does not exist. If you just want to test the pretrained model, you might want to"""
                          ''' set `--only_pretrain_model` or `args.only_pretrain_model=True`.'''			)
        else:
               lowercase	:		Union[str, Any]    =   MODEL_WITH_LM_HEAD_MAPPING[config.__class__](A			)

        if self.args.torchscript:
               raise NotImplementedError('''Training for torchscript is currently not implemented'''			)
        else:
               lowercase	:		Dict    =   model

        model.train()
        model.to(self.args.device			)

        # encoder-decoder has vocab size saved differently
        lowercase	:		Union[str, Any]    =   config.vocab_size if hasattr(A ,		'''vocab_size'''			) else config.encoder.vocab_size
        lowercase	:		List[Any]    =   torch.randint(A ,		(batch_size, sequence_length) ,		dtype=torch.long ,		device=self.args.device			)

        if self.args.fpaa:
               logger.info('''Running training in Mixed Precision...'''			)
               if not self.args.is_gpu:
                      raise ValueError('''Mixed precision is possible only for GPU.'''			)

               # amp seems to have memory leaks so that memory usage
               # is measured using .half() for now https://github.com/NVIDIA/apex/issues/439
               model.half()

        def compute_loss_and_backprob_encoder():
               lowercase	:		Optional[Any]    =   train_model(A ,		labels=A			)[0]
               loss.backward()
               return loss

        def compute_loss_and_backprob_encoder_decoder():
               lowercase	:		List[Any]    =   train_model(A ,		decoder_input_ids=A ,		labels=A			)[0]
               loss.backward()
               return loss

        lowercase	:		Any    =   (
            compute_loss_and_backprob_encoder_decoder
            if config.is_encoder_decoder
            else compute_loss_and_backprob_encoder
        )
        return _train



 def  SCREAMING_SNAKE_CASE(	self ,		A			):






        '''simple docstring'''




        try:
               if self.args.is_tpu or self.args.torchscript:
                      # run additional 10 times to stabilize compilation for tpu and torchscript
                      logger.info('''Do inference on TPU or torchscript. Running model 5 times to stabilize compilation'''			)
                      timeit.repeat(
                          A ,		repeat=1 ,		number=5 ,		)

               # as written in https://docs.python.org/2/library/timeit.html#timeit.Timer.repeat, min should be taken rather than the average
               lowercase	:		str    =   timeit.repeat(
                   A ,		repeat=self.args.repeat ,		number=10 ,		)

               if self.args.is_tpu and self.args.torch_xla_tpu_print_metrics:
                      import torch_xla.debug.metrics as met

                      self.print_fn(met.metrics_report()			)

               return min(A			) / 1_0.0
        except RuntimeError as e:
               self.print_fn(f"""Doesn't fit on GPU. {e}"""			)
               return "N/A"





 def  SCREAMING_SNAKE_CASE(	self ,		A			):






        '''simple docstring'''




        try:
               if self.args.trace_memory_line_by_line:
                      lowercase	:		int    =   start_memory_tracing('''transformers'''			)

               if self.args.is_tpu:
                      # tpu
                      raise NotImplementedError(
                          '''Memory Benchmarking is currently not implemented for TPU. Please disable memory benchmarking with'''
                          ''' `--no-memory` or `args.memory=False`'''			)
               elif self.args.is_gpu:
                      if not is_pyanvml_available():
                             logger.warning(
                                 '''py3nvml not installed, we won\'t log GPU memory usage. '''
                                 '''Install py3nvml (pip install py3nvml) to log information about GPU.'''			)
                             lowercase	:		str    =   '''N/A'''
                      else:
                             logger.info(
                                 '''Measuring total GPU usage on GPU device. Make sure to not have additional processes running'''
                                 ''' on the same GPU.'''			)
                             # init nvml
                             nvml.nvmlInit()
                             func()
                             lowercase	:		List[Any]    =   nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx			)
                             lowercase	:		List[Any]    =   nvml.nvmlDeviceGetMemoryInfo(A			)
                             lowercase	:		int    =   meminfo.used
                             lowercase	:		Dict    =   Memory(A			)
                             # shutdown nvml
                             nvml.nvmlShutdown()
               else:
                      # cpu
                      lowercase	:		Optional[Any]    =   measure_peak_memory_cpu(A			)
                      lowercase	:		Union[str, Any]    =   Memory(A			) if isinstance(A ,		A			) else memory_bytes

               if self.args.trace_memory_line_by_line:
                      lowercase	:		Tuple    =   stop_memory_tracing(A			)
               else:
                      lowercase	:		int    =   None

               return memory, summary
        except RuntimeError as e:
               self.print_fn(f"""Doesn't fit on GPU. {e}"""			)
               return "N/A", None




