'''simple docstring'''




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 	__UpperCAmelCase				(				snake_case_						,      snake_case_						,      snake_case_						,      unittest.TestCase     ):





       '''simple docstring'''



       __lowercase					:					List[str]    =				StableDiffusionImageVariationPipeline
       __lowercase					:					int    =				IMAGE_VARIATION_PARAMS
       __lowercase					:					Optional[int]    =				IMAGE_VARIATION_BATCH_PARAMS
       __lowercase					:					Dict    =				frozenset([]     )
       # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
       __lowercase					:					List[str]    =				frozenset([]     )

       def 					__A       (      self )		->     Optional[Any]:
           torch.manual_seed(0 )
           A_     =   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    ,     )
           A_     =   PNDMScheduler(skip_prk_steps=lowerCamelCase_ )
           torch.manual_seed(0 )
           A_     =   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 )
           A_     =   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    ,     )
           A_     =   CLIPVisionModelWithProjection(lowerCamelCase_ )
           A_     =   CLIPImageProcessor(crop_size=32    ,     size=32 )

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

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

       def 					__A       (      self )		->     Any:
           A_     =   'cpu'  # ensure determinism for the device-dependent torch.Generator
           A_     =   self.get_dummy_components()
           A_     =   StableDiffusionImageVariationPipeline(**lowerCamelCase_ )
           A_     =   sd_pipe.to(lowerCamelCase_ )
           sd_pipe.set_progress_bar_config(disable=lowerCamelCase_ )

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

           assert image.shape == (1, 64, 64, 3)
           A_     =   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 					__A       (      self )		->     Optional[int]:
           A_     =   'cpu'  # ensure determinism for the device-dependent torch.Generator
           A_     =   self.get_dummy_components()
           A_     =   StableDiffusionImageVariationPipeline(**lowerCamelCase_ )
           A_     =   sd_pipe.to(lowerCamelCase_ )
           sd_pipe.set_progress_bar_config(disable=lowerCamelCase_ )

           A_     =   self.get_dummy_inputs(lowerCamelCase_ )
           A_     =   2 * [inputs['image']]
           A_     =   sd_pipe(**lowerCamelCase_ )

           A_     =   output.images

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

           assert image.shape == (2, 64, 64, 3)
           A_     =   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 					__A       (      self )		->     Union[str, Any]:
           super().test_inference_batch_single_identical(expected_max_diff=3E-3 )




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





       '''simple docstring'''



       def 					__A       (      self )		->     Dict:
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()

       def 					__A       (      self    ,     _SCREAMING_SNAKE_CASE    ,     _SCREAMING_SNAKE_CASE="cpu"    ,     _SCREAMING_SNAKE_CASE=torch.floataa    ,     _SCREAMING_SNAKE_CASE=0 )		->     Dict:
           A_     =   torch.Generator(device=lowerCamelCase_ ).manual_seed(lowerCamelCase_ )
           A_     =   load_image(
               '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
               '''/stable_diffusion_imgvar/input_image_vermeer.png''' )
           A_     =   np.random.RandomState(lowerCamelCase_ ).standard_normal((1, 4, 64, 64) )
           A_     =   torch.from_numpy(lowerCamelCase_ ).to(device=lowerCamelCase_    ,     dtype=lowerCamelCase_ )
           A_     =   {
               'image': init_image,
               'latents': latents,
               'generator': generator,
               'num_inference_steps': 3,
               'guidance_scale': 7.5,
               'output_type': 'numpy',
           }
           return inputs

       def 					__A       (      self )		->     Tuple:
           A_     =   StableDiffusionImageVariationPipeline.from_pretrained(
               '''lambdalabs/sd-image-variations-diffusers'''    ,     safety_checker=lowerCamelCase_ )
           A_     =   sd_pipe.to(lowerCamelCase_ )
           sd_pipe.set_progress_bar_config(disable=lowerCamelCase_ )

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

           assert image.shape == (1, 512, 512, 3)
           A_     =   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 					__A       (      self )		->     Tuple:
           A_     =   0

           def callback_fn(_SCREAMING_SNAKE_CASE    ,     _SCREAMING_SNAKE_CASE    ,     _SCREAMING_SNAKE_CASE ) -> None:
               A_     =   True
               nonlocal number_of_steps
               number_of_steps += 1
               if step == 1:
                   A_     =   latents.detach().cpu().numpy()
                   assert latents.shape == (1, 4, 64, 64)
                   A_     =   latents[0, -3:, -3:, -1]
                   A_     =   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:
                   A_     =   latents.detach().cpu().numpy()
                   assert latents.shape == (1, 4, 64, 64)
                   A_     =   latents[0, -3:, -3:, -1]
                   A_     =   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

           A_     =   False

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

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

       def 					__A       (      self )		->     Tuple:
           torch.cuda.empty_cache()
           torch.cuda.reset_max_memory_allocated()
           torch.cuda.reset_peak_memory_stats()

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

           A_     =   self.get_inputs(lowerCamelCase_    ,     dtype=torch.floataa )
           A_     =   pipe(**lowerCamelCase_ )

           A_     =   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 	__UpperCAmelCase				(				unittest.TestCase     ):





       '''simple docstring'''



       def 					__A       (      self )		->     Optional[int]:
           super().tearDown()
           gc.collect()
           torch.cuda.empty_cache()

       def 					__A       (      self    ,     _SCREAMING_SNAKE_CASE    ,     _SCREAMING_SNAKE_CASE="cpu"    ,     _SCREAMING_SNAKE_CASE=torch.floataa    ,     _SCREAMING_SNAKE_CASE=0 )		->     Tuple:
           A_     =   torch.Generator(device=lowerCamelCase_ ).manual_seed(lowerCamelCase_ )
           A_     =   load_image(
               '''https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'''
               '''/stable_diffusion_imgvar/input_image_vermeer.png''' )
           A_     =   np.random.RandomState(lowerCamelCase_ ).standard_normal((1, 4, 64, 64) )
           A_     =   torch.from_numpy(lowerCamelCase_ ).to(device=lowerCamelCase_    ,     dtype=lowerCamelCase_ )
           A_     =   {
               'image': init_image,
               'latents': latents,
               'generator': generator,
               'num_inference_steps': 50,
               'guidance_scale': 7.5,
               'output_type': 'numpy',
           }
           return inputs

       def 					__A       (      self )		->     Any:
           A_     =   StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers''' )
           sd_pipe.to(lowerCamelCase_ )
           sd_pipe.set_progress_bar_config(disable=lowerCamelCase_ )

           A_     =   self.get_inputs(lowerCamelCase_ )
           A_     =   sd_pipe(**lowerCamelCase_ ).images[0]

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

       def 					__A       (      self )		->     Union[str, Any]:
           A_     =   StableDiffusionImageVariationPipeline.from_pretrained('''fusing/sd-image-variations-diffusers''' )
           A_     =   DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config )
           sd_pipe.to(lowerCamelCase_ )
           sd_pipe.set_progress_bar_config(disable=lowerCamelCase_ )

           A_     =   self.get_inputs(lowerCamelCase_ )
           A_     =   25
           A_     =   sd_pipe(**lowerCamelCase_ ).images[0]

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














"""simple docstring"""
import bsa
import requests






def        lowerCamelCase_  (						_lowerCamelCase				):
 return {
     "name": soup.ha.a.text,
     "genre": soup.find('span'			,   class_='genre'				).text.strip(),
     "rating": soup.strong.text,
     "page_link": f'''https://www.imdb.com{soup.a.get('href'				)}''',
 }






def        lowerCamelCase_  (						_lowerCamelCase = 5				):
 lowerCamelCase__     :			Optional[int]     				=						int(float(_lowerCamelCase				)				)
 if num_movies < 1:
  return ()
 lowerCamelCase__     :			Any     				=						(
     'https://www.imdb.com/search/title?title_type='
     f'''feature&sort=num_votes,desc&count={num_movies}'''
 )
 lowerCamelCase__     :			Union[str, Any]     				=						bsa.BeautifulSoup(requests.get(_lowerCamelCase				).content			,   'html.parser'				)
 return tuple(
     get_movie_data_from_soup(_lowerCamelCase				)
     for movie in source.find_all('div'			,   class_='lister-item mode-advanced'				)				)


if __name__ == "__main__":
      import json

      A_			: str				=			int(input("How many movies would you like to see? "))
      print(
          ", ".join(
              json.dumps(movie, indent=4) for movie in get_imdb_top_movies(num_movies)
          )
      )

