
"""simple docstring"""

from logging import getLogger
from typing import Any, Callable, List, Optional, Union

import numpy as np
import PIL
import torch

from ...schedulers import DDPMScheduler
from ..onnx_utils import ORT_TO_NP_TYPE, OnnxRuntimeModel
from ..pipeline_utils import ImagePipelineOutput
from . import StableDiffusionUpscalePipeline


UpperCamelCase      :							int          =					getLogger(__name__)


UpperCamelCase      :							Tuple          =					4
UpperCamelCase      :							List[Any]          =					7

UpperCamelCase      :							Optional[Any]          =					{
    "float16": torch.floataa,
    "float32": torch.floataa,
}






def 				A		(							snake_case :Dict						) ->      Union[str, Any]:
 if isinstance(__UpperCamelCase							,   torch.Tensor						):
  return image
 elif isinstance(__UpperCamelCase							,   PIL.Image.Image						):
  __UpperCamelCase							    =			[image]

 if isinstance(image[0]							,   PIL.Image.Image						):
  __UpperCamelCase							    =			image[0].size
  __UpperCamelCase							    =			(x - x % 6_4 for x in (w, h))  # resize to integer multiple of 32

  __UpperCamelCase							    =			[np.array(i.resize((w, h)						)						)[None, :] for i in image]
  __UpperCamelCase							    =			np.concatenate(__UpperCamelCase							,   axis=0						)
  __UpperCamelCase							    =			np.array(__UpperCamelCase						).astype(np.floataa						) / 2_5_5.0
  __UpperCamelCase							    =			image.transpose(0							,   3							,   1							,   2						)
  __UpperCamelCase							    =			2.0 * image - 1.0
  __UpperCamelCase							    =			torch.from_numpy(__UpperCamelCase						)
 elif isinstance(image[0]							,   torch.Tensor						):
  __UpperCamelCase							    =			torch.cat(__UpperCamelCase							,   dim=0						)

 return image





class __lowerCAmelCase						( lowerCAmelCase_     ):





     def __init__(      self      ,	__UpperCAmelCase      ,	__UpperCAmelCase      ,	__UpperCAmelCase      ,	__UpperCAmelCase      ,	__UpperCAmelCase      ,	__UpperCAmelCase      ,	__UpperCAmelCase = 350      ,	):

      '''simple docstring'''




      super().__init__(
          vae=lowerCamelCase_      ,	text_encoder=lowerCamelCase_      ,	tokenizer=lowerCamelCase_      ,	unet=lowerCamelCase_      ,	low_res_scheduler=lowerCamelCase_      ,	scheduler=lowerCamelCase_      ,	safety_checker=lowerCamelCase_      ,	feature_extractor=lowerCamelCase_      ,	watermarker=lowerCamelCase_      ,	max_noise_level=lowerCamelCase_      ,	)





     def __call__(      self      ,	__UpperCAmelCase      ,	__UpperCAmelCase      ,	__UpperCAmelCase = 75      ,	__UpperCAmelCase = 9.0      ,	__UpperCAmelCase = 20      ,	__UpperCAmelCase = None      ,	__UpperCAmelCase = 1      ,	__UpperCAmelCase = 0.0      ,	__UpperCAmelCase = None      ,	__UpperCAmelCase = None      ,	__UpperCAmelCase = None      ,	__UpperCAmelCase = None      ,	__UpperCAmelCase = "pil"      ,	__UpperCAmelCase = True      ,	__UpperCAmelCase = None      ,	__UpperCAmelCase = 1      ,	):

      '''simple docstring'''




      self.check_inputs(lowerCamelCase_      ,	lowerCamelCase_      ,	lowerCamelCase_      ,	lowerCamelCase_ )

      # 2. Define call parameters
      if prompt is not None and isinstance(lowerCamelCase_      ,	lowerCamelCase_ ):
       __UpperCamelCase							    =			1
      elif prompt is not None and isinstance(lowerCamelCase_      ,	lowerCamelCase_ ):
       __UpperCamelCase							    =			len(lowerCamelCase_ )
      else:
       __UpperCamelCase							    =			prompt_embeds.shape[0]

      __UpperCamelCase							    =			self._execution_device
      # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
      # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
      # corresponds to doing no classifier free guidance.
      __UpperCamelCase							    =			guidance_scale > 1.0

      # 3. Encode input prompt
      __UpperCamelCase							    =			self._encode_prompt(
          lowerCamelCase_      ,	lowerCamelCase_      ,	lowerCamelCase_      ,	lowerCamelCase_      ,	lowerCamelCase_      ,	prompt_embeds=lowerCamelCase_      ,	negative_prompt_embeds=lowerCamelCase_      ,	)

      __UpperCamelCase							    =			ORT_TO_PT_TYPE[str(text_embeddings.dtype )]

      # 4. Preprocess image
      __UpperCamelCase							    =			preprocess(lowerCamelCase_ )
      __UpperCamelCase							    =			image.cpu()

      # 5. set timesteps
      self.scheduler.set_timesteps(lowerCamelCase_      ,	device=lowerCamelCase_ )
      __UpperCamelCase							    =			self.scheduler.timesteps

      # 5. Add noise to image
      __UpperCamelCase							    =			torch.tensor([noise_level]      ,	dtype=torch.long      ,	device=lowerCamelCase_ )
      __UpperCamelCase							    =			torch.randn(image.shape      ,	generator=lowerCamelCase_      ,	device=lowerCamelCase_      ,	dtype=lowerCamelCase_ )
      __UpperCamelCase							    =			self.low_res_scheduler.add_noise(lowerCamelCase_      ,	lowerCamelCase_      ,	lowerCamelCase_ )

      __UpperCamelCase							    =			2 if do_classifier_free_guidance else 1
      __UpperCamelCase							    =			np.concatenate([image] * batch_multiplier * num_images_per_prompt )
      __UpperCamelCase							    =			np.concatenate([noise_level] * image.shape[0] )

      # 6. Prepare latent variables
      __UpperCamelCase							    =			image.shape[2:]
      __UpperCamelCase							    =			self.prepare_latents(
          batch_size * num_images_per_prompt      ,	lowerCamelCase_      ,	lowerCamelCase_      ,	lowerCamelCase_      ,	lowerCamelCase_      ,	lowerCamelCase_      ,	lowerCamelCase_      ,	lowerCamelCase_      ,	)

      # 7. Check that sizes of image and latents match
      __UpperCamelCase							    =			image.shape[1]
      if NUM_LATENT_CHANNELS + num_channels_image != NUM_UNET_INPUT_CHANNELS:
       raise ValueError(
           'Incorrect configuration settings! The config of `pipeline.unet` expects'
           F' {NUM_UNET_INPUT_CHANNELS} but received `num_channels_latents`: {NUM_LATENT_CHANNELS} +'
           F' `num_channels_image`: {num_channels_image} '
           F' = {NUM_LATENT_CHANNELS+num_channels_image}. Please verify the config of'
           ' `pipeline.unet` or your `image` input.' )

      # 8. Prepare extra step kwargs. TODO: Logic should ideally just be moved out of the pipeline
      __UpperCamelCase							    =			self.prepare_extra_step_kwargs(lowerCamelCase_      ,	lowerCamelCase_ )

      __UpperCamelCase							    =			next(
          (input.type for input in self.unet.model.get_inputs() if input.name == 'timestep')      ,	'tensor(float)' )
      __UpperCamelCase							    =			ORT_TO_NP_TYPE[timestep_dtype]

      # 9. Denoising loop
      __UpperCamelCase							    =			len(lowerCamelCase_ ) - num_inference_steps * self.scheduler.order
      with self.progress_bar(total=lowerCamelCase_ ) as progress_bar:
       for i, t in enumerate(lowerCamelCase_ ):
        # expand the latents if we are doing classifier free guidance
        __UpperCamelCase							    =			np.concatenate([latents] * 2 ) if do_classifier_free_guidance else latents

        # concat latents, mask, masked_image_latents in the channel dimension
        __UpperCamelCase							    =			self.scheduler.scale_model_input(lowerCamelCase_      ,	lowerCamelCase_ )
        __UpperCamelCase							    =			np.concatenate([latent_model_input, image]      ,	axis=1 )

        # timestep to tensor
        __UpperCamelCase							    =			np.array([t]      ,	dtype=lowerCamelCase_ )

        # predict the noise residual
        __UpperCamelCase							    =			self.unet(
            sample=lowerCamelCase_      ,	timestep=lowerCamelCase_      ,	encoder_hidden_states=lowerCamelCase_      ,	class_labels=noise_level.astype(np.intaa )      ,	)[0]

        # perform guidance
        if do_classifier_free_guidance:
         __UpperCamelCase							    =			np.split(lowerCamelCase_      ,	2 )
        __UpperCamelCase							    =			noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

        # compute the previous noisy sample x_t -> x_t-1
        __UpperCamelCase							    =			self.scheduler.step(
            torch.from_numpy(lowerCamelCase_ )      ,	lowerCamelCase_      ,	lowerCamelCase_      ,	**lowerCamelCase_ ).prev_sample

        # call the callback, if provided
        if i == len(lowerCamelCase_ ) - 1 or ((i + 1) > num_warmup_steps and (i + 1) % self.scheduler.order == 0):
         progress_bar.update()
         if callback is not None and i % callback_steps == 0:
          callback(lowerCamelCase_      ,	lowerCamelCase_      ,	lowerCamelCase_ )

        # 10. Post-processing
      __UpperCamelCase							    =			self.decode_latents(latents.float() )

      # 11. Convert to PIL
      if output_type == "pil":
       __UpperCamelCase							    =			self.numpy_to_pil(lowerCamelCase_ )

      if not return_dict:
       return (image,)

      return ImagePipelineOutput(images=lowerCamelCase_ )





     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase ):

      '''simple docstring'''




      __UpperCamelCase							    =			1 / 0.0_8_3_3_3 * latents
      __UpperCamelCase							    =			self.vae(latent_sample=lowerCamelCase_ )[0]
      __UpperCamelCase							    =			np.clip(image / 2 + 0.5      ,	0      ,	1 )
      __UpperCamelCase							    =			image.transpose((0, 2, 3, 1) )
      return image







     def 		UpperCAmelCase   (      self      ,	__UpperCAmelCase      ,	__UpperCAmelCase      ,	__UpperCAmelCase      ,	__UpperCAmelCase      ,	__UpperCAmelCase      ,	__UpperCAmelCase = None      ,	__UpperCAmelCase = None      ,	):

      '''simple docstring'''




      if prompt is not None and isinstance(lowerCamelCase_      ,	lowerCamelCase_ ):
       __UpperCamelCase							    =			1
      elif prompt is not None and isinstance(lowerCamelCase_      ,	lowerCamelCase_ ):
       __UpperCamelCase							    =			len(lowerCamelCase_ )
      else:
       __UpperCamelCase							    =			prompt_embeds.shape[0]

      if prompt_embeds is None:
       __UpperCamelCase							    =			self.tokenizer(
           lowerCamelCase_      ,	padding='max_length'      ,	max_length=self.tokenizer.model_max_length      ,	truncation=lowerCamelCase_      ,	return_tensors='pt'      ,	)
       __UpperCamelCase							    =			text_inputs.input_ids
       __UpperCamelCase							    =			self.tokenizer(lowerCamelCase_      ,	padding='longest'      ,	return_tensors='pt' ).input_ids

       if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(
           lowerCamelCase_      ,	lowerCamelCase_ ):
        __UpperCamelCase							    =			self.tokenizer.batch_decode(
            untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1] )
        logger.warning(
            'The following part of your input was truncated because CLIP can only handle sequences up to'
            F' {self.tokenizer.model_max_length} tokens: {removed_text}' )

       # no positional arguments to text_encoder
       __UpperCamelCase							    =			self.text_encoder(
           input_ids=text_input_ids.int().to(lowerCamelCase_ )      ,	)
       __UpperCamelCase							    =			prompt_embeds[0]

      __UpperCamelCase							    =			prompt_embeds.shape
      # duplicate text embeddings for each generation per prompt, using mps friendly method
      __UpperCamelCase							    =			prompt_embeds.repeat(1      ,	lowerCamelCase_ )
      __UpperCamelCase							    =			prompt_embeds.reshape(bs_embed * num_images_per_prompt      ,	lowerCamelCase_      ,	-1 )

      # get unconditional embeddings for classifier free guidance
      if do_classifier_free_guidance and negative_prompt_embeds is None:
       __UpperCamelCase							    =			42
       if negative_prompt is None:
        __UpperCamelCase							    =			[""""""] * batch_size
       elif type(lowerCamelCase_ ) is not type(lowerCamelCase_ ):
        raise TypeError(
            F'`negative_prompt` should be the same type to `prompt`, but got {type(lowerCamelCase_ )} !='
            F' {type(lowerCamelCase_ )}.' )
       elif isinstance(lowerCamelCase_      ,	lowerCamelCase_ ):
        __UpperCamelCase							    =			[negative_prompt]
       elif batch_size != len(lowerCamelCase_ ):
        raise ValueError(
            F'`negative_prompt`: {negative_prompt} has batch size {len(lowerCamelCase_ )}, but `prompt`:'
            F' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'
            ' the batch size of `prompt`.' )
       else:
        __UpperCamelCase							    =			negative_prompt

       __UpperCamelCase							    =			text_input_ids.shape[-1]
       __UpperCamelCase							    =			self.tokenizer(
           lowerCamelCase_      ,	padding='max_length'      ,	max_length=lowerCamelCase_      ,	truncation=lowerCamelCase_      ,	return_tensors='pt'      ,	)

       # if hasattr(uncond_input, "attention_mask"):
       #     attention_mask = uncond_input.attention_mask.to(device)
       # else:
       #     attention_mask = None

       __UpperCamelCase							    =			self.text_encoder(
           input_ids=uncond_input.input_ids.int().to(lowerCamelCase_ )      ,	)
       __UpperCamelCase							    =			uncond_embeddings[0]

      if do_classifier_free_guidance:
       __UpperCamelCase							    =			uncond_embeddings.shape[1]
       # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
       __UpperCamelCase							    =			uncond_embeddings.repeat(1      ,	lowerCamelCase_ )
       __UpperCamelCase							    =			uncond_embeddings.reshape(batch_size * num_images_per_prompt      ,	lowerCamelCase_      ,	-1 )

       # For classifier free guidance, we need to do two forward passes.
       # Here we concatenate the unconditional and text embeddings into a single batch
       # to avoid doing two forward passes
       __UpperCamelCase							    =			np.concatenate([uncond_embeddings, prompt_embeds] )

      return prompt_embeds







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



