





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        __lowerCAmelCase (					UpperCamelCase__		,       UpperCamelCase__		,       UpperCamelCase__		,       unittest.TestCase):
      _lowercase       :      Optional[Any]        =    StableDiffusionImageVariationPipeline
      _lowercase       :      Optional[int]        =    IMAGE_VARIATION_PARAMS
      _lowercase       :      Optional[int]        =    IMAGE_VARIATION_BATCH_PARAMS
      _lowercase       :      Optional[int]        =    frozenset([])
      # TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
      _lowercase       :      str        =    frozenset([])







      def      _lowercase						(					self							)     ->		str:



        '''simple docstring'''




        torch.manual_seed(0							)
        a__			:      List[Any]            =UNetaDConditionModel(
            block_out_channels=(3_2, 6_4)	,   layers_per_block=2	,   sample_size=3_2	,   in_channels=4	,   out_channels=4	,   down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")	,   up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")	,   cross_attention_dim=3_2	,   )
        a__			:      Union[str, Any]            =PNDMScheduler(skip_prk_steps=lowerCAmelCase__							)
        torch.manual_seed(0							)
        a__			:      int            =AutoencoderKL(
            block_out_channels=[3_2, 6_4]	,   in_channels=3	,   out_channels=3	,   down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]	,   up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]	,   latent_channels=4	,   )
        torch.manual_seed(0							)
        a__			:      List[str]            =CLIPVisionConfig(
            hidden_size=3_2	,   projection_dim=3_2	,   intermediate_size=3_7	,   layer_norm_eps=1E-05	,   num_attention_heads=4	,   num_hidden_layers=5	,   image_size=3_2	,   patch_size=4	,   )
        a__			:      Tuple            =CLIPVisionModelWithProjection(lowerCAmelCase__							)
        a__			:      int            =CLIPImageProcessor(crop_size=3_2	,   size=3_2							)

        a__			:      Optional[int]            ={
            "unet": unet,
            "scheduler": scheduler,
            "vae": vae,
            "image_encoder": image_encoder,
            "feature_extractor": feature_extractor,
            "safety_checker": None,
        }
        return components







      def      _lowercase						(					self	,   lowerCAmelCase__	,   lowerCAmelCase__=0							)     ->		List[str]:



        '''simple docstring'''




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







      def      _lowercase						(					self							)     ->		Optional[int]:



        '''simple docstring'''




        a__			:      Tuple            ="cpu"  # ensure determinism for the device-dependent torch.Generator
        a__			:      Optional[Any]            =self.get_dummy_components()
        a__			:      List[str]            =StableDiffusionImageVariationPipeline(**lowerCAmelCase__							)
        a__			:      Optional[Any]            =sd_pipe.to(lowerCAmelCase__							)
        sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__							)

        a__			:      Any            =self.get_dummy_inputs(lowerCAmelCase__							)
        a__			:      Optional[Any]            =sd_pipe(**lowerCAmelCase__							).images
        a__			:      int            =image[0, -3:, -3:, -1]

        assert image.shape == (1, 6_4, 6_4, 3)
        a__			:      int            =np.array([0.52_39, 0.57_23, 0.47_96, 0.50_49, 0.55_50, 0.46_85, 0.53_29, 0.48_91, 0.49_21]							)

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







      def      _lowercase						(					self							)     ->		List[Any]:



        '''simple docstring'''




        a__			:      int            ="cpu"  # ensure determinism for the device-dependent torch.Generator
        a__			:      Any            =self.get_dummy_components()
        a__			:      Tuple            =StableDiffusionImageVariationPipeline(**lowerCAmelCase__							)
        a__			:      Dict            =sd_pipe.to(lowerCAmelCase__							)
        sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__							)

        a__			:      Union[str, Any]            =self.get_dummy_inputs(lowerCAmelCase__							)
        a__			:      Optional[int]            =2 * [inputs["image"]]
        a__			:      Optional[Any]            =sd_pipe(**lowerCAmelCase__							)

        a__			:      Union[str, Any]            =output.images

        a__			:      Any            =image[-1, -3:, -3:, -1]

        assert image.shape == (2, 6_4, 6_4, 3)
        a__			:      List[str]            =np.array([0.68_92, 0.56_37, 0.58_36, 0.57_71, 0.62_54, 0.64_09, 0.55_80, 0.55_69, 0.52_89]							)

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







      def      _lowercase						(					self							)     ->		Union[str, Any]:



        '''simple docstring'''




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







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







      def      _lowercase						(					self							)     ->		Union[str, Any]:



        '''simple docstring'''




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







      def      _lowercase						(					self	,   lowerCAmelCase__	,   lowerCAmelCase__="cpu"	,   lowerCAmelCase__=torch.floataa	,   lowerCAmelCase__=0							)     ->		int:



        '''simple docstring'''




        a__			:      int            =torch.Generator(device=lowerCAmelCase__							).manual_seed(lowerCAmelCase__							)
        a__			:      List[Any]            =load_image(
            "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
            "/stable_diffusion_imgvar/input_image_vermeer.png"							)
        a__			:      int            =np.random.RandomState(lowerCAmelCase__							).standard_normal((1, 4, 6_4, 6_4)							)
        a__			:      Any            =torch.from_numpy(lowerCAmelCase__							).to(device=lowerCAmelCase__	,   dtype=lowerCAmelCase__							)
        a__			:      str            ={
            "image": init_image,
            "latents": latents,
            "generator": generator,
            "num_inference_steps": 3,
            "guidance_scale": 7.5,
            "output_type": "numpy",
        }
        return inputs







      def      _lowercase						(					self							)     ->		List[Any]:



        '''simple docstring'''




        a__			:      int            =StableDiffusionImageVariationPipeline.from_pretrained(
            "lambdalabs/sd-image-variations-diffusers"	,   safety_checker=lowerCAmelCase__							)
        a__			:      Union[str, Any]            =sd_pipe.to(lowerCAmelCase__							)
        sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__							)

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

        assert image.shape == (1, 5_1_2, 5_1_2, 3)
        a__			:      Any            =np.array([0.8_44_91, 0.9_07_89, 0.7_57_08, 0.7_87_34, 0.8_34_85, 0.7_00_99, 0.6_69_38, 0.6_87_27, 0.6_13_79]							)
        assert np.abs(image_slice - expected_slice							).max() < 6E-3







      def      _lowercase						(					self							)     ->		Union[str, Any]:



        '''simple docstring'''




        a__			:      Tuple            =0

        def callback_fn(lowerCAmelCase__	,   lowerCAmelCase__	,   lowerCAmelCase__							) -> None:
          a__			:      int            =True
          nonlocal number_of_steps
          number_of_steps += 1
          if step == 1:
            a__			:      Union[str, Any]            =latents.detach().cpu().numpy()
            assert latents.shape == (1, 4, 6_4, 6_4)
            a__			:      Any            =latents[0, -3:, -3:, -1]
            a__			:      Any            =np.array(
                [-0.16_21, 0.28_37, -0.79_79, -0.12_21, -1.30_57, 0.76_81, -2.11_91, 0.04_64, 1.63_09]							)

            assert np.abs(latents_slice.flatten() - expected_slice							).max() < 5E-2
          elif step == 2:
            a__			:      List[Any]            =latents.detach().cpu().numpy()
            assert latents.shape == (1, 4, 6_4, 6_4)
            a__			:      Union[str, Any]            =latents[0, -3:, -3:, -1]
            a__			:      Dict            =np.array([0.62_99, 1.75_00, 1.19_92, -2.15_82, -1.89_94, 0.73_34, -0.70_90, 1.01_37, 1.52_73]							)

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

        a__			:      str            =False

        a__			:      List[Any]            =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__			:      Optional[int]            =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      _lowercase						(					self							)     ->		List[str]:



        '''simple docstring'''




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

        a__			:      Dict            ="fusing/sd-image-variations-diffusers"
        a__			:      Optional[int]            =StableDiffusionImageVariationPipeline.from_pretrained(
            lowerCAmelCase__	,   safety_checker=lowerCAmelCase__	,   torch_dtype=torch.floataa							)
        a__			:      Optional[Any]            =pipe.to(lowerCAmelCase__							)
        pipe.set_progress_bar_config(disable=lowerCAmelCase__							)
        pipe.enable_attention_slicing(1							)
        pipe.enable_sequential_cpu_offload()

        a__			:      Any            =self.get_inputs(lowerCAmelCase__	,   dtype=torch.floataa							)
        a__			:      str            =pipe(**lowerCAmelCase__							)

        a__			:      Tuple            =torch.cuda.max_memory_allocated()
        # make sure that less than 2.6 GB is allocated
        assert mem_bytes < 2.6 * 1_0**9







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







      def      _lowercase						(					self							)     ->		Dict:



        '''simple docstring'''




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







      def      _lowercase						(					self	,   lowerCAmelCase__	,   lowerCAmelCase__="cpu"	,   lowerCAmelCase__=torch.floataa	,   lowerCAmelCase__=0							)     ->		Dict:



        '''simple docstring'''




        a__			:      Tuple            =torch.Generator(device=lowerCAmelCase__							).manual_seed(lowerCAmelCase__							)
        a__			:      List[str]            =load_image(
            "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main"
            "/stable_diffusion_imgvar/input_image_vermeer.png"							)
        a__			:      Tuple            =np.random.RandomState(lowerCAmelCase__							).standard_normal((1, 4, 6_4, 6_4)							)
        a__			:      Any            =torch.from_numpy(lowerCAmelCase__							).to(device=lowerCAmelCase__	,   dtype=lowerCAmelCase__							)
        a__			:      str            ={
            "image": init_image,
            "latents": latents,
            "generator": generator,
            "num_inference_steps": 5_0,
            "guidance_scale": 7.5,
            "output_type": "numpy",
        }
        return inputs







      def      _lowercase						(					self							)     ->		Any:



        '''simple docstring'''




        a__			:      Optional[int]            =StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"							)
        sd_pipe.to(lowerCAmelCase__							)
        sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__							)

        a__			:      Union[str, Any]            =self.get_inputs(lowerCAmelCase__							)
        a__			:      Dict            =sd_pipe(**lowerCAmelCase__							).images[0]

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







      def      _lowercase						(					self							)     ->		Optional[int]:



        '''simple docstring'''




        a__			:      List[Any]            =StableDiffusionImageVariationPipeline.from_pretrained("fusing/sd-image-variations-diffusers"							)
        a__			:      Union[str, Any]            =DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config							)
        sd_pipe.to(lowerCAmelCase__							)
        sd_pipe.set_progress_bar_config(disable=lowerCAmelCase__							)

        a__			:      Any            =self.get_inputs(lowerCAmelCase__							)
        a__			:      Union[str, Any]            =2_5
        a__			:      List[str]            =sd_pipe(**lowerCAmelCase__							).images[0]

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













def   _A (	SCREAMING_SNAKE_CASE  :     int		):





  """simple docstring"""



  barcode //= 10  # exclude the last digit
  a__			:      int            =False
  a__			:      List[str]            =0

  # extract and check each digit
  while barcode != 0:
    a__			:      Optional[Any]            =1 if checker else 3
    s += mult * (barcode % 10)
    barcode //= 10
    a__			:      Union[str, Any]            =not checker

  return (10 - (s % 10)) % 10
def   _A (	SCREAMING_SNAKE_CASE  :     int		):





  """simple docstring"""



  return len(str(SCREAMING_SNAKE_CASE		)		) == 13 and get_check_digit(SCREAMING_SNAKE_CASE		) == barcode % 10
def   _A (	SCREAMING_SNAKE_CASE  :     str		):





  """simple docstring"""



  if str(SCREAMING_SNAKE_CASE		).isalpha():
    a__			:      Tuple            =f'''Barcode \'{barcode}\' has alphabetic characters.'''
    raise ValueError(SCREAMING_SNAKE_CASE		)
  elif int(SCREAMING_SNAKE_CASE		) < 0:
    raise ValueError("The entered barcode has a negative value. Try again."		)
  else:
    return int(SCREAMING_SNAKE_CASE		)


if __name__ == "__main__":
     import doctest

     doctest.testmod()
     UpperCAmelCase			:			List[str]	  = get_barcode(input("""Barcode: """).strip())

     if is_valid(barcode):
          print(F"""'{barcode}' is a valid barcode.""")
     else:
          print(F"""'{barcode}' is NOT a valid barcode.""")






