import argparse
import logging
import math
import time

import torch

from transformers import TransfoXLCorpus, TransfoXLLMHeadModel


logging.basicConfig(
    format='%(asctime)s - %(levelname)s - %(name)s - %(message)s', datefmt='%m/%d/%Y %H:%M:%S', level=logging.INFO
)
lowerCAmelCase__							= logging.getLogger(__name__)






def       __lowerCamelCase		(						):
       lowerCAmelCase__     	=					argparse.ArgumentParser(description='PyTorch Transformer Language Model'	)
       parser.add_argument('--model_name'	,  type=lowerCAmelCase__	,  default='transfo-xl-wt103'	,  help='pretrained model name'	)
       parser.add_argument(
           '--split'	,  type=lowerCAmelCase__	,  default='test'	,  choices=['all', 'valid', 'test']	,  help='which split to evaluate'	)
       parser.add_argument('--batch_size'	,  type=lowerCAmelCase__	,  default=1_0	,  help='batch size'	)
       parser.add_argument('--tgt_len'	,  type=lowerCAmelCase__	,  default=1_2_8	,  help='number of tokens to predict'	)
       parser.add_argument('--ext_len'	,  type=lowerCAmelCase__	,  default=0	,  help='length of the extended context'	)
       parser.add_argument('--mem_len'	,  type=lowerCAmelCase__	,  default=1_6_0_0	,  help='length of the retained previous heads'	)
       parser.add_argument('--clamp_len'	,  type=lowerCAmelCase__	,  default=1_0_0_0	,  help='max positional embedding index'	)
       parser.add_argument('--no_cuda'	,  action='store_true'	,  help='Do not use CUDA even though CUA is available'	)
       parser.add_argument('--work_dir'	,  type=lowerCAmelCase__	,  required=lowerCAmelCase__	,  help='path to the work_dir'	)
       parser.add_argument('--no_log'	,  action='store_true'	,  help='do not log the eval result'	)
       parser.add_argument('--same_length'	,  action='store_true'	,  help='set same length attention with masking'	)
       parser.add_argument('--server_ip'	,  type=lowerCAmelCase__	,  default=''	,  help='Can be used for distant debugging.'	)
       parser.add_argument('--server_port'	,  type=lowerCAmelCase__	,  default=''	,  help='Can be used for distant debugging.'	)
       lowerCAmelCase__     	=					parser.parse_args()
       assert args.ext_len >= 0, "extended context length must be non-negative"

       if args.server_ip and args.server_port:
              # Distant debugging - see https://code.visualstudio.com/docs/python/debugging#_attach-to-a-local-script
              import ptvsd

              print('Waiting for debugger attach'	)
              ptvsd.enable_attach(address=(args.server_ip, args.server_port)	,  redirect_output=lowerCAmelCase__	)
              ptvsd.wait_for_attach()

       lowerCAmelCase__     	=					torch.device('cuda' if torch.cuda.is_available() and not args.no_cuda else 'cpu'	)
       logger.info('device: {}'.format(lowerCAmelCase__	)	)

       # Load a pre-processed dataset
       # You can also build the corpus yourself using TransfoXLCorpus methods
       # The pre-processing involve computing word frequencies to prepare the Adaptive input and SoftMax
       # and tokenizing the dataset
       # The pre-processed corpus is a convertion (using the conversion script )
       lowerCAmelCase__     	=					TransfoXLCorpus.from_pretrained(args.model_name	)

       lowerCAmelCase__     	=					corpus.get_iterator('valid'	,  args.batch_size	,  args.tgt_len	,  device=lowerCAmelCase__	,  ext_len=args.ext_len	)
       lowerCAmelCase__     	=					corpus.get_iterator('test'	,  args.batch_size	,  args.tgt_len	,  device=lowerCAmelCase__	,  ext_len=args.ext_len	)

       # Load a pre-trained model
       lowerCAmelCase__     	=					TransfoXLLMHeadModel.from_pretrained(args.model_name	)
       model.to(lowerCAmelCase__	)

       logger.info(
           'Evaluating with bsz {} tgt_len {} ext_len {} mem_len {} clamp_len {}'.format(
               args.batch_size	,  args.tgt_len	,  args.ext_len	,  args.mem_len	,  args.clamp_len	)	)

       model.reset_memory_length(args.mem_len	)
       if args.clamp_len > 0:
              lowerCAmelCase__     	=					args.clamp_len
       if args.same_length:
              lowerCAmelCase__     	=					True

       ###############################################################################
       # Evaluation code
       ###############################################################################
       def evaluate(lowerCAmelCase__	):
              # Turn on evaluation mode which disables dropout.
              model.eval()
              lowerCAmelCase__	,					lowerCAmelCase__     	=					0, 0.0
              lowerCAmelCase__     	=					time.time()
              with torch.no_grad():
                     lowerCAmelCase__     	=					None
                     for idx, (data, target, seq_len) in enumerate(lowerCAmelCase__	):
                            lowerCAmelCase__     	=					model(lowerCAmelCase__	,  lm_labels=lowerCAmelCase__	,  mems=lowerCAmelCase__	)
                            lowerCAmelCase__	,					lowerCAmelCase__	,					lowerCAmelCase__     	=					ret
                            lowerCAmelCase__     	=					loss.mean()
                            total_loss += seq_len * loss.item()
                            total_len += seq_len
                     lowerCAmelCase__     	=					time.time() - start_time
              logger.info('Time : {:.2f}s, {:.2f}ms/segment'.format(lowerCAmelCase__	,  1_0_0_0 * total_time / (idx + 1)	)	)
              return total_loss / total_len

       # Run on test data.
       if args.split == "all":
              lowerCAmelCase__     	=					evaluate(lowerCAmelCase__	)
              lowerCAmelCase__     	=					evaluate(lowerCAmelCase__	)
       elif args.split == "valid":
              lowerCAmelCase__     	=					evaluate(lowerCAmelCase__	)
              lowerCAmelCase__     	=					None
       elif args.split == "test":
              lowerCAmelCase__     	=					evaluate(lowerCAmelCase__	)
              lowerCAmelCase__     	=					None

       def format_log(lowerCAmelCase__	,  lowerCAmelCase__	):
              lowerCAmelCase__     	=					'| {0} loss {1:5.2f} | {0} ppl {2:9.3f} '.format(lowerCAmelCase__	,  lowerCAmelCase__	,  math.exp(lowerCAmelCase__	)	)
              return log_str

       lowerCAmelCase__     	=					''
       if valid_loss is not None:
              log_str += format_log(lowerCAmelCase__	,  'valid'	)
       if test_loss is not None:
              log_str += format_log(lowerCAmelCase__	,  'test'	)

       logger.info('=' * 1_0_0	)
       logger.info(lowerCAmelCase__	)
       logger.info('=' * 1_0_0	)


if __name__ == "__main__":
    main()

import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()
class 				a_ (						SCREAMING_SNAKE_CASE       ,     SCREAMING_SNAKE_CASE       ,     SCREAMING_SNAKE_CASE       ,     unittest.TestCase						):






 '''simple docstring'''






 UpperCAmelCase_	 =					StableDiffusionImageVariationPipeline
 UpperCAmelCase_	 =					IMAGE_VARIATION_PARAMS
 UpperCAmelCase_	 =					IMAGE_VARIATION_BATCH_PARAMS
 UpperCAmelCase_	 =					frozenset([]						)
 # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
 UpperCAmelCase_	 =					frozenset([]						)




 def 		__snake_case						(			self					:  Optional[Any]):







        '''simple docstring'''



        torch.manual_seed(0)
        lowerCAmelCase__     	=					UNetaDConditionModel(
            block_out_channels=(32, 64)			,    layers_per_block=2			,    sample_size=32			,    in_channels=4			,    out_channels=4			,    down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')			,    up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')			,    cross_attention_dim=32			,    )
        lowerCAmelCase__     	=					PNDMScheduler(skip_prk_steps=lowercase__)
        torch.manual_seed(0)
        lowerCAmelCase__     	=					AutoencoderKL(
            block_out_channels=[32, 64]			,    in_channels=3			,    out_channels=3			,    down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']			,    up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']			,    latent_channels=4			,    )
        torch.manual_seed(0)
        lowerCAmelCase__     	=					CLIPVisionConfig(
            hidden_size=32			,    projection_dim=32			,    intermediate_size=37			,    layer_norm_eps=1e-05			,    num_attention_heads=4			,    num_hidden_layers=5			,    image_size=32			,    patch_size=4			,    )
        lowerCAmelCase__     	=					CLIPVisionModelWithProjection(lowercase__)
        lowerCAmelCase__     	=					CLIPImageProcessor(crop_size=32			,    size=32)

        lowerCAmelCase__     	=					{
            'unet': unet,
            'scheduler': scheduler,
            'vae': vae,
            'image_encoder': image_encoder,
            'feature_extractor': feature_extractor,
            'safety_checker': None,
        }
        return components




 def 		__snake_case						(			self					:  Optional[int]			,    lowercase__					:  Dict			,    lowercase__					:  Tuple=0):







        '''simple docstring'''



        lowerCAmelCase__     	=					floats_tensor((1, 3, 32, 32)			,    rng=random.Random(lowercase__))
        lowerCAmelCase__     	=					image.cpu().permute(0			,    2			,    3			,    1)[0]
        lowerCAmelCase__     	=					Image.fromarray(np.uinta(lowercase__)).convert('RGB').resize((32, 32))
        if str(lowercase__).startswith('mps'):
               lowerCAmelCase__     	=					torch.manual_seed(lowercase__)
        else:
               lowerCAmelCase__     	=					torch.Generator(device=lowercase__).manual_seed(lowercase__)
        lowerCAmelCase__     	=					{
            'image': image,
            'generator': generator,
            'num_inference_steps': 2,
            'guidance_scale': 6.0,
            'output_type': 'numpy',
        }
        return inputs




 def 		__snake_case						(			self					:  Union[str, Any]):







        '''simple docstring'''



        lowerCAmelCase__     	=					'cpu'  # ensure determinism for the device-dependent torch.Generator
        lowerCAmelCase__     	=					self.get_dummy_components()
        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline(**lowercase__)
        lowerCAmelCase__     	=					sd_pipe.to(lowercase__)
        sd_pipe.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__     	=					self.get_dummy_inputs(lowercase__)
        lowerCAmelCase__     	=					sd_pipe(**lowercase__).images
        lowerCAmelCase__     	=					image[0, -3:, -3:, -1]

        assert image.shape == (1, 64, 64, 3)
        lowerCAmelCase__     	=					np.array([0.5_239, 0.5_723, 0.4_796, 0.5_049, 0.5_550, 0.4_685, 0.5_329, 0.4_891, 0.4_921])

        assert np.abs(image_slice.flatten() - expected_slice).max() < 1e-3




 def 		__snake_case						(			self					:  Any):







        '''simple docstring'''



        lowerCAmelCase__     	=					'cpu'  # ensure determinism for the device-dependent torch.Generator
        lowerCAmelCase__     	=					self.get_dummy_components()
        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline(**lowercase__)
        lowerCAmelCase__     	=					sd_pipe.to(lowercase__)
        sd_pipe.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__     	=					self.get_dummy_inputs(lowercase__)
        lowerCAmelCase__     	=					2 * [inputs['image']]
        lowerCAmelCase__     	=					sd_pipe(**lowercase__)

        lowerCAmelCase__     	=					output.images

        lowerCAmelCase__     	=					image[-1, -3:, -3:, -1]

        assert image.shape == (2, 64, 64, 3)
        lowerCAmelCase__     	=					np.array([0.6_892, 0.5_637, 0.5_836, 0.5_771, 0.6_254, 0.6_409, 0.5_580, 0.5_569, 0.5_289])

        assert np.abs(image_slice.flatten() - expected_slice).max() < 1e-3




 def 		__snake_case						(			self					:  Optional[int]):







        '''simple docstring'''



        super().test_inference_batch_single_identical(expected_max_diff=3e-3)






@slow
@require_torch_gpu
class 				a_ (						unittest.TestCase						):






 '''simple docstring'''






 def 		__snake_case						(			self					:  List[Any]):







        '''simple docstring'''



        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()




 def 		__snake_case						(			self					:  List[Any]			,    lowercase__					:  List[str]			,    lowercase__					:  Optional[int]="cpu"			,    lowercase__					:  List[str]=torch.floataa			,    lowercase__					:  Dict=0):







        '''simple docstring'''



        lowerCAmelCase__     	=					torch.Generator(device=lowercase__).manual_seed(lowercase__)
        lowerCAmelCase__     	=					load_image(
            'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
            '/stable_diffusion_imgvar/input_image_vermeer.png')
        lowerCAmelCase__     	=					np.random.RandomState(lowercase__).standard_normal((1, 4, 64, 64))
        lowerCAmelCase__     	=					torch.from_numpy(lowercase__).to(device=lowercase__			,    dtype=lowercase__)
        lowerCAmelCase__     	=					{
            'image': init_image,
            'latents': latents,
            'generator': generator,
            'num_inference_steps': 3,
            'guidance_scale': 7.5,
            'output_type': 'numpy',
        }
        return inputs




 def 		__snake_case						(			self					:  List[str]):







        '''simple docstring'''



        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline.from_pretrained(
            'lambdalabs/sd-image-variations-diffusers'			,    safety_checker=lowercase__)
        lowerCAmelCase__     	=					sd_pipe.to(lowercase__)
        sd_pipe.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__     	=					self.get_inputs(lowercase__)
        lowerCAmelCase__     	=					sd_pipe(**lowercase__).images
        lowerCAmelCase__     	=					image[0, -3:, -3:, -1].flatten()

        assert image.shape == (1, 512, 512, 3)
        lowerCAmelCase__     	=					np.array([0.84_491, 0.90_789, 0.75_708, 0.78_734, 0.83_485, 0.70_099, 0.66_938, 0.68_727, 0.61_379])
        assert np.abs(image_slice - expected_slice).max() < 6e-3




 def 		__snake_case						(			self					:  Optional[Any]):







        '''simple docstring'''



        lowerCAmelCase__     	=					0

        def callback_fn(lowercase__					:  int			,    lowercase__					:  int			,    lowercase__					:  torch.FloatTensor) -> None:
               lowerCAmelCase__     	=					True
               nonlocal number_of_steps
               number_of_steps += 1
               if step == 1:
                      lowerCAmelCase__     	=					latents.detach().cpu().numpy()
                      assert latents.shape == (1, 4, 64, 64)
                      lowerCAmelCase__     	=					latents[0, -3:, -3:, -1]
                      lowerCAmelCase__     	=					np.array(
                          [-0.1_621, 0.2_837, -0.7_979, -0.1_221, -1.3_057, 0.7_681, -2.1_191, 0.0_464, 1.6_309])

                      assert np.abs(latents_slice.flatten() - expected_slice).max() < 5e-2
               elif step == 2:
                      lowerCAmelCase__     	=					latents.detach().cpu().numpy()
                      assert latents.shape == (1, 4, 64, 64)
                      lowerCAmelCase__     	=					latents[0, -3:, -3:, -1]
                      lowerCAmelCase__     	=					np.array([0.6_299, 1.7_500, 1.1_992, -2.1_582, -1.8_994, 0.7_334, -0.7_090, 1.0_137, 1.5_273])

                      assert np.abs(latents_slice.flatten() - expected_slice).max() < 5e-2

        lowerCAmelCase__     	=					False

        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline.from_pretrained(
            'fusing/sd-image-variations-diffusers'			,    safety_checker=lowercase__			,    torch_dtype=torch.floataa			,    )
        pipe.to(lowercase__)
        pipe.set_progress_bar_config(disable=lowercase__)
        pipe.enable_attention_slicing()

        lowerCAmelCase__     	=					self.get_inputs(lowercase__			,    dtype=torch.floataa)
        pipe(**lowercase__			,    callback=lowercase__			,    callback_steps=1)
        assert callback_fn.has_been_called
        assert number_of_steps == inputs["num_inference_steps"]




 def 		__snake_case						(			self					:  str):







        '''simple docstring'''



        torch.cuda.empty_cache()
        torch.cuda.reset_max_memory_allocated()
        torch.cuda.reset_peak_memory_stats()

        lowerCAmelCase__     	=					'fusing/sd-image-variations-diffusers'
        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline.from_pretrained(
            lowercase__			,    safety_checker=lowercase__			,    torch_dtype=torch.floataa)
        lowerCAmelCase__     	=					pipe.to(lowercase__)
        pipe.set_progress_bar_config(disable=lowercase__)
        pipe.enable_attention_slicing(1)
        pipe.enable_sequential_cpu_offload()

        lowerCAmelCase__     	=					self.get_inputs(lowercase__			,    dtype=torch.floataa)
        lowerCAmelCase__     	=					pipe(**lowercase__)

        lowerCAmelCase__     	=					torch.cuda.max_memory_allocated()
        # make sure that less than 2.6 GB is allocated
        assert mem_bytes < 2.6 * 10**9






@nightly
@require_torch_gpu
class 				a_ (						unittest.TestCase						):






 '''simple docstring'''






 def 		__snake_case						(			self					:  str):







        '''simple docstring'''



        super().tearDown()
        gc.collect()
        torch.cuda.empty_cache()




 def 		__snake_case						(			self					:  int			,    lowercase__					:  List[str]			,    lowercase__					:  Dict="cpu"			,    lowercase__					:  Optional[int]=torch.floataa			,    lowercase__					:  Optional[Any]=0):







        '''simple docstring'''



        lowerCAmelCase__     	=					torch.Generator(device=lowercase__).manual_seed(lowercase__)
        lowerCAmelCase__     	=					load_image(
            'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
            '/stable_diffusion_imgvar/input_image_vermeer.png')
        lowerCAmelCase__     	=					np.random.RandomState(lowercase__).standard_normal((1, 4, 64, 64))
        lowerCAmelCase__     	=					torch.from_numpy(lowercase__).to(device=lowercase__			,    dtype=lowercase__)
        lowerCAmelCase__     	=					{
            'image': init_image,
            'latents': latents,
            'generator': generator,
            'num_inference_steps': 50,
            'guidance_scale': 7.5,
            'output_type': 'numpy',
        }
        return inputs




 def 		__snake_case						(			self					:  Optional[Any]):







        '''simple docstring'''



        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers')
        sd_pipe.to(lowercase__)
        sd_pipe.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__     	=					self.get_inputs(lowercase__)
        lowerCAmelCase__     	=					sd_pipe(**lowercase__).images[0]

        lowerCAmelCase__     	=					load_numpy(
            'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
            '/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy')
        lowerCAmelCase__     	=					np.abs(expected_image - image).max()
        assert max_diff < 1e-3




 def 		__snake_case						(			self					:  Tuple):







        '''simple docstring'''



        lowerCAmelCase__     	=					StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers')
        lowerCAmelCase__     	=					DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config)
        sd_pipe.to(lowercase__)
        sd_pipe.set_progress_bar_config(disable=lowercase__)

        lowerCAmelCase__     	=					self.get_inputs(lowercase__)
        lowerCAmelCase__     	=					25
        lowerCAmelCase__     	=					sd_pipe(**lowercase__).images[0]

        lowerCAmelCase__     	=					load_numpy(
            'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
            '/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy')
        lowerCAmelCase__     	=					np.abs(expected_image - image).max()
        assert max_diff < 1e-3
