


import argparse

import huggingface_hub
import k_diffusion as K
import torch

from diffusers import UNetaDConditionModel


__lowerCamelCase   	=  '''pcuenq/k-upscaler'''



def 			_a      (    __UpperCamelCase	,	__UpperCamelCase	,	*, __UpperCamelCase	,	__UpperCamelCase			):
     a_     :  Optional[Any]				      =     {
         # norm1
         F'''{diffusers_resnet_prefix}.norm1.linear.weight''': checkpoint[F'''{resnet_prefix}.main.0.mapper.weight'''],
         F'''{diffusers_resnet_prefix}.norm1.linear.bias''': checkpoint[F'''{resnet_prefix}.main.0.mapper.bias'''],
         # conv1
         F'''{diffusers_resnet_prefix}.conv1.weight''': checkpoint[F'''{resnet_prefix}.main.2.weight'''],
         F'''{diffusers_resnet_prefix}.conv1.bias''': checkpoint[F'''{resnet_prefix}.main.2.bias'''],
         # norm2
         F'''{diffusers_resnet_prefix}.norm2.linear.weight''': checkpoint[F'''{resnet_prefix}.main.4.mapper.weight'''],
         F'''{diffusers_resnet_prefix}.norm2.linear.bias''': checkpoint[F'''{resnet_prefix}.main.4.mapper.bias'''],
         # conv2
         F'''{diffusers_resnet_prefix}.conv2.weight''': checkpoint[F'''{resnet_prefix}.main.6.weight'''],
         F'''{diffusers_resnet_prefix}.conv2.bias''': checkpoint[F'''{resnet_prefix}.main.6.bias'''],
     }

     if resnet.conv_shortcut is not None:
          rv.update(
              {
                  F'''{diffusers_resnet_prefix}.conv_shortcut.weight''': checkpoint[F'''{resnet_prefix}.skip.weight'''],
              }			)

     return rv



def 			_a      (    __UpperCamelCase	,	*, __UpperCamelCase	,	__UpperCamelCase			):
     a_      ,      a_      ,      a_     :  str				      =     checkpoint[F'''{attention_prefix}.qkv_proj.weight'''].chunk(3	,	dim=0			)
     a_      ,      a_      ,      a_     :  Tuple				      =     checkpoint[F'''{attention_prefix}.qkv_proj.bias'''].chunk(3	,	dim=0			)
     a_     :  List[str]				      =     {
         # norm
         F'''{diffusers_attention_prefix}.norm1.linear.weight''': checkpoint[F'''{attention_prefix}.norm_in.mapper.weight'''],
         F'''{diffusers_attention_prefix}.norm1.linear.bias''': checkpoint[F'''{attention_prefix}.norm_in.mapper.bias'''],
         # to_q
         F'''{diffusers_attention_prefix}.attn1.to_q.weight''': weight_q.squeeze(-1			).squeeze(-1			),
         F'''{diffusers_attention_prefix}.attn1.to_q.bias''': bias_q,
         # to_k
         F'''{diffusers_attention_prefix}.attn1.to_k.weight''': weight_k.squeeze(-1			).squeeze(-1			),
         F'''{diffusers_attention_prefix}.attn1.to_k.bias''': bias_k,
         # to_v
         F'''{diffusers_attention_prefix}.attn1.to_v.weight''': weight_v.squeeze(-1			).squeeze(-1			),
         F'''{diffusers_attention_prefix}.attn1.to_v.bias''': bias_v,
         # to_out
         F'''{diffusers_attention_prefix}.attn1.to_out.0.weight''': checkpoint[F'''{attention_prefix}.out_proj.weight''']
         .squeeze(-1			)
         .squeeze(-1			),
         F'''{diffusers_attention_prefix}.attn1.to_out.0.bias''': checkpoint[F'''{attention_prefix}.out_proj.bias'''],
     }

     return rv



def 			_a      (    __UpperCamelCase	,	*, __UpperCamelCase	,	__UpperCamelCase	,	__UpperCamelCase			):
     a_      ,      a_     :  str				      =     checkpoint[F'''{attention_prefix}.kv_proj.weight'''].chunk(2	,	dim=0			)
     a_      ,      a_     :  List[Any]				      =     checkpoint[F'''{attention_prefix}.kv_proj.bias'''].chunk(2	,	dim=0			)

     a_     :  Any				      =     {
         # norm2 (ada groupnorm)
         F'''{diffusers_attention_prefix}.norm{diffusers_attention_index}.linear.weight''': checkpoint[
             F'''{attention_prefix}.norm_dec.mapper.weight'''
         ],
         F'''{diffusers_attention_prefix}.norm{diffusers_attention_index}.linear.bias''': checkpoint[
             F'''{attention_prefix}.norm_dec.mapper.bias'''
         ],
         # layernorm on encoder_hidden_state
         F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.norm_cross.weight''': checkpoint[
             F'''{attention_prefix}.norm_enc.weight'''
         ],
         F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.norm_cross.bias''': checkpoint[
             F'''{attention_prefix}.norm_enc.bias'''
         ],
         # to_q
         F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_q.weight''': checkpoint[
             F'''{attention_prefix}.q_proj.weight'''
         ]
         .squeeze(-1			)
         .squeeze(-1			),
         F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_q.bias''': checkpoint[
             F'''{attention_prefix}.q_proj.bias'''
         ],
         # to_k
         F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_k.weight''': weight_k.squeeze(-1			).squeeze(-1			),
         F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_k.bias''': bias_k,
         # to_v
         F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_v.weight''': weight_v.squeeze(-1			).squeeze(-1			),
         F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_v.bias''': bias_v,
         # to_out
         F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_out.0.weight''': checkpoint[
             F'''{attention_prefix}.out_proj.weight'''
         ]
         .squeeze(-1			)
         .squeeze(-1			),
         F'''{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_out.0.bias''': checkpoint[
             F'''{attention_prefix}.out_proj.bias'''
         ],
     }

     return rv



def 			_a      (    __UpperCamelCase	,	__UpperCamelCase	,	__UpperCamelCase	,	__UpperCamelCase			):
     a_     :  Optional[int]				      =     """inner_model.u_net.u_blocks""" if block_type == """up""" else """inner_model.u_net.d_blocks"""
     a_     :  int				      =     F'''{block_prefix}.{block_idx}'''

     a_     :  Tuple				      =     {}

     if not hasattr(__UpperCamelCase	,	"""attentions"""			):
          a_     :  Dict				      =     1  # resnet only
     elif not block.attentions[0].add_self_attention:
          a_     :  Union[str, Any]				      =     2  # resnet -> cross-attention
     else:
          a_     :  Dict				      =     3  # resnet -> self-attention -> cross-attention)

     for resnet_idx, resnet in enumerate(block.resnets			):
          # diffusers_resnet_prefix = f"{diffusers_up_block_prefix}.resnets.{resnet_idx}"
          a_     :  Union[str, Any]				      =     F'''{block_type}_blocks.{block_idx}.resnets.{resnet_idx}'''
          a_     :  Tuple				      =     n * resnet_idx if block_type == """up""" else n * resnet_idx + 1
          a_     :  List[Any]				      =     F'''{block_prefix}.{idx}''' if block_type == """up""" else F'''{block_prefix}.{idx}'''

          diffusers_checkpoint.update(
              resnet_to_diffusers_checkpoint(
                  __UpperCamelCase	,	__UpperCamelCase	,	diffusers_resnet_prefix=__UpperCamelCase	,	resnet_prefix=__UpperCamelCase			)			)

     if hasattr(__UpperCamelCase	,	"""attentions"""			):
          for attention_idx, attention in enumerate(block.attentions			):
               a_     :  List[Any]				      =     F'''{block_type}_blocks.{block_idx}.attentions.{attention_idx}'''
               a_     :  Optional[int]				      =     n * attention_idx + 1 if block_type == """up""" else n * attention_idx + 2
               a_     :  Optional[int]				      =     F'''{block_prefix}.{idx}'''
               a_     :  Any				      =     F'''{block_prefix}.{idx }'''
               a_     :  Any				      =     1 if not attention.add_self_attention else 2
               a_     :  List[str]				      =     (
                   n * attention_idx + cross_attention_index
                   if block_type == """up"""
                   else n * attention_idx + cross_attention_index + 1
               )
               a_     :  int				      =     F'''{block_prefix}.{idx }'''

               diffusers_checkpoint.update(
                   cross_attn_to_diffusers_checkpoint(
                       __UpperCamelCase	,	diffusers_attention_prefix=__UpperCamelCase	,	diffusers_attention_index=2	,	attention_prefix=__UpperCamelCase	,	)			)

               if attention.add_self_attention is True:
                    diffusers_checkpoint.update(
                        self_attn_to_diffusers_checkpoint(
                            __UpperCamelCase	,	diffusers_attention_prefix=__UpperCamelCase	,	attention_prefix=__UpperCamelCase	,	)			)

     return diffusers_checkpoint



def 			_a      (    __UpperCamelCase	,	__UpperCamelCase			):
     a_     :  Tuple				      =     {}

     # pre-processing
     diffusers_checkpoint.update(
         {
             """conv_in.weight""": checkpoint["""inner_model.proj_in.weight"""],
             """conv_in.bias""": checkpoint["""inner_model.proj_in.bias"""],
         }			)

     # timestep and class embedding
     diffusers_checkpoint.update(
         {
             """time_proj.weight""": checkpoint["""inner_model.timestep_embed.weight"""].squeeze(-1			),
             """time_embedding.linear_1.weight""": checkpoint["""inner_model.mapping.0.weight"""],
             """time_embedding.linear_1.bias""": checkpoint["""inner_model.mapping.0.bias"""],
             """time_embedding.linear_2.weight""": checkpoint["""inner_model.mapping.2.weight"""],
             """time_embedding.linear_2.bias""": checkpoint["""inner_model.mapping.2.bias"""],
             """time_embedding.cond_proj.weight""": checkpoint["""inner_model.mapping_cond.weight"""],
         }			)

     # down_blocks
     for down_block_idx, down_block in enumerate(model.down_blocks			):
          diffusers_checkpoint.update(block_to_diffusers_checkpoint(__UpperCamelCase	,	__UpperCamelCase	,	__UpperCamelCase	,	"""down"""			)			)

     # up_blocks
     for up_block_idx, up_block in enumerate(model.up_blocks			):
          diffusers_checkpoint.update(block_to_diffusers_checkpoint(__UpperCamelCase	,	__UpperCamelCase	,	__UpperCamelCase	,	"""up"""			)			)

     # post-processing
     diffusers_checkpoint.update(
         {
             """conv_out.weight""": checkpoint["""inner_model.proj_out.weight"""],
             """conv_out.bias""": checkpoint["""inner_model.proj_out.bias"""],
         }			)

     return diffusers_checkpoint



def 			_a      (    __UpperCamelCase			):
     a_     :  Union[str, Any]				      =     original_config["""input_channels"""] + original_config["""unet_cond_dim"""]
     a_     :  List[str]				      =     original_config["""input_channels"""] + (1 if original_config["""has_variance"""] else 0)

     a_     :  int				      =     original_config["""channels"""]

     assert (
         len(set(original_config["""depths"""]			)			) == 1
     ), "UNet2DConditionModel currently do not support blocks with different number of layers"
     a_     :  str				      =     original_config["""depths"""][0]

     a_     :  str				      =     original_config["""mapping_cond_dim"""]
     a_     :  str				      =     original_config["""cross_cond_dim"""]

     a_     :  Optional[Any]				      =     []
     a_     :  str				      =     []
     for s, c in zip(original_config["""self_attn_depths"""]	,	original_config["""cross_attn_depths"""]			):
          if s:
               a_     :  Any				      =     """self"""
               a_     :  Dict				      =     """cross""" if c else None
          elif c:
               a_     :  int				      =     """cross"""
               a_     :  List[Any]				      =     None
          else:
               a_     :  int				      =     None
               a_     :  List[Any]				      =     None
          attna_types.append(__UpperCamelCase			)
          attna_types.append(__UpperCamelCase			)

     a_     :  Dict				      =     UNetaDConditionModel(
         in_channels=__UpperCamelCase	,	out_channels=__UpperCamelCase	,	down_block_types=("""KDownBlock2D""", """KCrossAttnDownBlock2D""", """KCrossAttnDownBlock2D""", """KCrossAttnDownBlock2D""")	,	mid_block_type=__UpperCamelCase	,	up_block_types=("""KCrossAttnUpBlock2D""", """KCrossAttnUpBlock2D""", """KCrossAttnUpBlock2D""", """KUpBlock2D""")	,	block_out_channels=__UpperCamelCase	,	layers_per_block=__UpperCamelCase	,	act_fn="""gelu"""	,	norm_num_groups=__UpperCamelCase	,	cross_attention_dim=__UpperCamelCase	,	attention_head_dim=6_4	,	time_cond_proj_dim=__UpperCamelCase	,	resnet_time_scale_shift="""scale_shift"""	,	time_embedding_type="""fourier"""	,	timestep_post_act="""gelu"""	,	conv_in_kernel=1	,	conv_out_kernel=1	,	)

     return unet



def 			_a      (    __UpperCamelCase			):
     a_     :  Tuple				      =     torch.device("""cuda""" if torch.cuda.is_available() else """cpu"""			)

     a_     :  int				      =     huggingface_hub.hf_hub_download(__UpperCamelCase	,	"""config_laion_text_cond_latent_upscaler_2.json"""			)
     a_     :  List[Any]				      =     huggingface_hub.hf_hub_download(
         __UpperCamelCase	,	"""laion_text_cond_latent_upscaler_2_1_00470000_slim.pth"""			)
     print(F'''loading original model configuration from {orig_config_path}'''			)
     print(F'''loading original model checkpoint from {orig_weights_path}'''			)

     print("""converting to diffusers unet"""			)
     a_     :  Dict				      =     K.config.load_config(open(__UpperCamelCase			)			)["""model"""]
     a_     :  int				      =     unet_model_from_original_config(__UpperCamelCase			)

     a_     :  Optional[Any]				      =     torch.load(__UpperCamelCase	,	map_location=__UpperCamelCase			)["""model_ema"""]
     a_     :  str				      =     unet_to_diffusers_checkpoint(__UpperCamelCase	,	__UpperCamelCase			)

     model.load_state_dict(__UpperCamelCase	,	strict=__UpperCamelCase			)
     model.save_pretrained(args.dump_path			)
     print(F'''saving converted unet model in {args.dump_path}'''			)


if __name__ == "__main__":
       __lowerCamelCase   	=  argparse.ArgumentParser()

       parser.add_argument('''--dump_path''', default=None, type=str, required=True, help='''Path to the output model.''')
       __lowerCamelCase   	=  parser.parse_args()

       main(args)







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


__lowerCamelCase   	=  logging.get_logger(__name__)


class 	a__						(    lowerCAmelCase_     ):
       lowerCamelCase__:  PyTorchBenchmarkArguments
       lowerCamelCase__:  PretrainedConfig
       lowerCamelCase__:  str				     =  "PyTorch"

       @property
       def     UpperCAmelCase( self   :  str     ):
            return torch.__version__

       def     UpperCAmelCase( self   :  str   ,				lowerCamelCase_   :  str   ,				lowerCamelCase_   :  int   ,				lowerCamelCase_   :  int     ):
            a_     :  Tuple				      =     self._prepare_inference_func(lowerCamelCase_   ,				lowerCamelCase_   ,				lowerCamelCase_     )
            return self._measure_speed(_inference     )

       def     UpperCAmelCase( self   :  Optional[Any]   ,				lowerCamelCase_   :  str   ,				lowerCamelCase_   :  int   ,				lowerCamelCase_   :  int     ):
            a_     :  List[str]				      =     self._prepare_inference_func(lowerCamelCase_   ,				lowerCamelCase_   ,				lowerCamelCase_     )
            return self._measure_memory(_inference     )

       def     UpperCAmelCase( self   :  List[str]   ,				lowerCamelCase_   :  str   ,				lowerCamelCase_   :  int   ,				lowerCamelCase_   :  int     ):
            a_     :  str				      =     self._prepare_train_func(lowerCamelCase_   ,				lowerCamelCase_   ,				lowerCamelCase_     )
            return self._measure_speed(_train     )

       def     UpperCAmelCase( self   :  str   ,				lowerCamelCase_   :  str   ,				lowerCamelCase_   :  int   ,				lowerCamelCase_   :  int     ):
            a_     :  List[Any]				      =     self._prepare_train_func(lowerCamelCase_   ,				lowerCamelCase_   ,				lowerCamelCase_     )
            return self._measure_memory(_train     )

       def     UpperCAmelCase( self   :  Union[str, Any]   ,				lowerCamelCase_   :  str   ,				lowerCamelCase_   :  int   ,				lowerCamelCase_   :  int     ):
            a_     :  Union[str, Any]				      =     self.config_dict[model_name]

            if self.args.torchscript:
                 a_     :  Optional[Any]				      =     True

            a_     :  Any				      =     (
                hasattr(lowerCamelCase_   ,				"""architectures"""     )
                and isinstance(config.architectures   ,				lowerCamelCase_     )
                and len(config.architectures     ) > 0
            )
            if not self.args.only_pretrain_model and has_model_class_in_config:
                 try:
                      a_     :  Union[str, Any]				      =     config.architectures[0]
                      a_     :  Any				      =     __import__("""transformers"""   ,				fromlist=[model_class]     )
                      a_     :  List[str]				      =     getattr(lowerCamelCase_   ,				lowerCamelCase_     )
                      a_     :  Dict				      =     model_cls(lowerCamelCase_     )
                 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:
                 a_     :  str				      =     MODEL_MAPPING[config.__class__](lowerCamelCase_     )

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

            # encoder-decoder has vocab size saved differently
            a_     :  Union[str, Any]				      =     config.vocab_size if hasattr(lowerCamelCase_   ,				"""vocab_size"""     ) else config.encoder.vocab_size
            a_     :  Dict				      =     torch.randint(lowerCamelCase_   ,				(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():
                      a_     :  Optional[int]				      =     torch.jit.trace(lowerCamelCase_   ,				lowerCamelCase_     )
            else:
                 a_     :  Optional[Any]				      =     model

            def encoder_decoder_forward():
                 with torch.no_grad():
                      a_     :  Any				      =     inference_model(lowerCamelCase_   ,				decoder_input_ids=lowerCamelCase_     )
                 return outputs

            def encoder_forward():
                 with torch.no_grad():
                      a_     :  Dict				      =     inference_model(lowerCamelCase_     )
                 return outputs

            a_     :  Union[str, Any]				      =     encoder_decoder_forward if config.is_encoder_decoder else encoder_forward
            return _forward

       def     UpperCAmelCase( self   :  int   ,				lowerCamelCase_   :  str   ,				lowerCamelCase_   :  int   ,				lowerCamelCase_   :  int     ):
            a_     :  Optional[int]				      =     self.config_dict[model_name]

            a_     :  str				      =     (
                hasattr(lowerCamelCase_   ,				"""architectures"""     )
                and isinstance(config.architectures   ,				lowerCamelCase_     )
                and len(config.architectures     ) > 0
            )
            if not self.args.only_pretrain_model and has_model_class_in_config:
                 try:
                      a_     :  Optional[int]				      =     config.architectures[0]
                      a_     :  List[Any]				      =     __import__("""transformers"""   ,				fromlist=[model_class]     )
                      a_     :  List[str]				      =     getattr(lowerCamelCase_   ,				lowerCamelCase_     )
                      a_     :  Any				      =     model_cls(lowerCamelCase_     )
                 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:
                 a_     :  Union[str, Any]				      =     MODEL_WITH_LM_HEAD_MAPPING[config.__class__](lowerCamelCase_     )

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

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

            # encoder-decoder has vocab size saved differently
            a_     :  Any				      =     config.vocab_size if hasattr(lowerCamelCase_   ,				"""vocab_size"""     ) else config.encoder.vocab_size
            a_     :  Tuple				      =     torch.randint(lowerCamelCase_   ,				(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():
                 a_     :  Dict				      =     train_model(lowerCamelCase_   ,				labels=lowerCamelCase_     )[0]
                 loss.backward()
                 return loss

            def compute_loss_and_backprob_encoder_decoder():
                 a_     :  Optional[int]				      =     train_model(lowerCamelCase_   ,				decoder_input_ids=lowerCamelCase_   ,				labels=lowerCamelCase_     )[0]
                 loss.backward()
                 return loss

            a_     :  Optional[Any]				      =     (
                compute_loss_and_backprob_encoder_decoder
                if config.is_encoder_decoder
                else compute_loss_and_backprob_encoder
            )
            return _train

       def     UpperCAmelCase( self   :  str   ,				lowerCamelCase_   :  Optional[int]     ):
            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(
                          lowerCamelCase_   ,				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
                 a_     :  List[str]				      =     timeit.repeat(
                     lowerCamelCase_   ,				repeat=self.args.repeat   ,				number=1_0   ,				)

                 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(lowerCamelCase_     ) / 1_0.0
            except RuntimeError as e:
                 self.print_fn(F'''Doesn\'t fit on GPU. {e}'''     )
                 return "N/A"

       def     UpperCAmelCase( self   :  Tuple   ,				lowerCamelCase_   :  Callable[[], None]     ):
            try:
                 if self.args.trace_memory_line_by_line:
                      a_     :  Dict				      =     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."""     )
                           a_     :  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()
                           a_     :  List[Any]				      =     nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx     )
                           a_     :  Tuple				      =     nvml.nvmlDeviceGetMemoryInfo(lowerCamelCase_     )
                           a_     :  Any				      =     meminfo.used
                           a_     :  Any				      =     Memory(lowerCamelCase_     )
                           # shutdown nvml
                           nvml.nvmlShutdown()
                 else:
                      # cpu
                      a_     :  Union[str, Any]				      =     measure_peak_memory_cpu(lowerCamelCase_     )
                      a_     :  Optional[Any]				      =     Memory(lowerCamelCase_     ) if isinstance(lowerCamelCase_   ,				lowerCamelCase_     ) else memory_bytes

                 if self.args.trace_memory_line_by_line:
                      a_     :  Dict				      =     stop_memory_tracing(lowerCamelCase_     )
                 else:
                      a_     :  Any				      =     None

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



