



'''simple docstring'''






import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPanoramaPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


@skip_mps
class   _lowercase						(							lowerCAmelCase__		,					lowerCAmelCase__		,					unittest.TestCase				):
   _UpperCamelCase       		=   StableDiffusionPanoramaPipeline
   _UpperCamelCase       		=   TEXT_TO_IMAGE_PARAMS
   _UpperCamelCase       		=   TEXT_TO_IMAGE_BATCH_PARAMS
   _UpperCamelCase       		=   TEXT_TO_IMAGE_IMAGE_PARAMS
   _UpperCamelCase       		=   TEXT_TO_IMAGE_IMAGE_PARAMS


   def 					snake_case				( self					):
       torch.manual_seed(0					)
       A       :       Dict							=						UNetaDConditionModel(
           block_out_channels=(32, 64)    ,   layers_per_block=1    ,   sample_size=32    ,   in_channels=4    ,   out_channels=4    ,   down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')    ,   up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')    ,   cross_attention_dim=32    ,   )
       A       :       Optional[int]							=						DDIMScheduler()
       torch.manual_seed(0					)
       A       :       int							=						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       :       Dict							=						CLIPTextConfig(
           bos_token_id=0    ,   eos_token_id=2    ,   hidden_size=32    ,   intermediate_size=37    ,   layer_norm_eps=1E-0_5    ,   num_attention_heads=4    ,   num_hidden_layers=5    ,   pad_token_id=1    ,   vocab_size=1_000    ,   )
       A       :       Any							=						CLIPTextModel(_SCREAMING_SNAKE_CASE					)
       A       :       Optional[Any]							=						CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''					)

       A       :       Any							=						{
           '''unet''': unet,
           '''scheduler''': scheduler,
           '''vae''': vae,
           '''text_encoder''': text_encoder,
           '''tokenizer''': tokenizer,
           '''safety_checker''': None,
           '''feature_extractor''': None,
       }
       return components


   def 					snake_case				( self    ,   _UpperCAmelCase    ,   _UpperCAmelCase=0					):
       A       :       List[str]							=						torch.manual_seed(_SCREAMING_SNAKE_CASE					)
       A       :       str							=						{
           '''prompt''': '''a photo of the dolomites''',
           '''generator''': generator,
           # Setting height and width to None to prevent OOMs on CPU.
           '''height''': None,
           '''width''': None,
           '''num_inference_steps''': 1,
           '''guidance_scale''': 6.0,
           '''output_type''': '''numpy''',
       }
       return inputs


   def 					snake_case				( self					):
       A       :       Any							=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
       A       :       Optional[int]							=						self.get_dummy_components()
       A       :       Optional[int]							=						StableDiffusionPanoramaPipeline(**_SCREAMING_SNAKE_CASE					)
       A       :       Union[str, Any]							=						sd_pipe.to(_SCREAMING_SNAKE_CASE					)
       sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)

       A       :       str							=						self.get_dummy_inputs(_SCREAMING_SNAKE_CASE					)
       A       :       Union[str, Any]							=						sd_pipe(**_SCREAMING_SNAKE_CASE					).images
       A       :       str							=						image[0, -3:, -3:, -1]
       assert image.shape == (1, 64, 64, 3)

       A       :       Dict							=						np.array([0.6186, 0.5374, 0.4915, 0.4135, 0.4114, 0.4563, 0.5128, 0.4977, 0.4757]					)

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


   def 					snake_case				( self					):
       super().test_inference_batch_consistent(batch_sizes=[1, 2]					)


   def 					snake_case				( self					):
       super().test_inference_batch_single_identical(batch_size=2    ,   expected_max_diff=3.2_5E-3					)


   def 					snake_case				( self					):
       A       :       Tuple							=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
       A       :       List[Any]							=						self.get_dummy_components()
       A       :       str							=						StableDiffusionPanoramaPipeline(**_SCREAMING_SNAKE_CASE					)
       A       :       int							=						sd_pipe.to(_SCREAMING_SNAKE_CASE					)
       sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)

       A       :       Optional[Any]							=						self.get_dummy_inputs(_SCREAMING_SNAKE_CASE					)
       A       :       Any							=						'''french fries'''
       A       :       List[Any]							=						sd_pipe(**_SCREAMING_SNAKE_CASE    ,   negative_prompt=_SCREAMING_SNAKE_CASE					)
       A       :       str							=						output.images
       A       :       Union[str, Any]							=						image[0, -3:, -3:, -1]

       assert image.shape == (1, 64, 64, 3)

       A       :       str							=						np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757]					)

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


   def 					snake_case				( self					):
       A       :       Optional[Any]							=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
       A       :       Optional[Any]							=						self.get_dummy_components()
       A       :       List[str]							=						StableDiffusionPanoramaPipeline(**_SCREAMING_SNAKE_CASE					)
       A       :       Dict							=						sd_pipe.to(_SCREAMING_SNAKE_CASE					)
       sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)

       A       :       Optional[Any]							=						self.get_dummy_inputs(_SCREAMING_SNAKE_CASE					)
       A       :       Union[str, Any]							=						sd_pipe(**_SCREAMING_SNAKE_CASE    ,   view_batch_size=2					)
       A       :       str							=						output.images
       A       :       int							=						image[0, -3:, -3:, -1]

       assert image.shape == (1, 64, 64, 3)

       A       :       Union[str, Any]							=						np.array([0.6187, 0.5375, 0.4915, 0.4136, 0.4114, 0.4563, 0.5128, 0.4976, 0.4757]					)

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


   def 					snake_case				( self					):
       A       :       Any							=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
       A       :       Optional[Any]							=						self.get_dummy_components()
       A       :       Tuple							=						EulerAncestralDiscreteScheduler(
           beta_start=0.00085    ,   beta_end=0.012    ,   beta_schedule='''scaled_linear'''					)
       A       :       Optional[Any]							=						StableDiffusionPanoramaPipeline(**_SCREAMING_SNAKE_CASE					)
       A       :       Union[str, Any]							=						sd_pipe.to(_SCREAMING_SNAKE_CASE					)
       sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)

       A       :       Any							=						self.get_dummy_inputs(_SCREAMING_SNAKE_CASE					)
       A       :       List[Any]							=						sd_pipe(**_SCREAMING_SNAKE_CASE					).images
       A       :       Any							=						image[0, -3:, -3:, -1]

       assert image.shape == (1, 64, 64, 3)

       A       :       str							=						np.array([0.4024, 0.6510, 0.4901, 0.5378, 0.5813, 0.5622, 0.4795, 0.4467, 0.4952]					)

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


   def 					snake_case				( self					):
       A       :       List[str]							=						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
       A       :       int							=						self.get_dummy_components()
       A       :       int							=						PNDMScheduler(
           beta_start=0.00085    ,   beta_end=0.012    ,   beta_schedule='''scaled_linear'''    ,   skip_prk_steps=_SCREAMING_SNAKE_CASE					)
       A       :       Dict							=						StableDiffusionPanoramaPipeline(**_SCREAMING_SNAKE_CASE					)
       A       :       Dict							=						sd_pipe.to(_SCREAMING_SNAKE_CASE					)
       sd_pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)

       A       :       Union[str, Any]							=						self.get_dummy_inputs(_SCREAMING_SNAKE_CASE					)
       A       :       Any							=						sd_pipe(**_SCREAMING_SNAKE_CASE					).images
       A       :       List[str]							=						image[0, -3:, -3:, -1]

       assert image.shape == (1, 64, 64, 3)

       A       :       int							=						np.array([0.6391, 0.6291, 0.4861, 0.5134, 0.5552, 0.4578, 0.5032, 0.5023, 0.4539]					)

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







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


   def 					snake_case				( self					):
       super().tearDown()
       gc.collect()
       torch.cuda.empty_cache()


   def 					snake_case				( self    ,   _UpperCAmelCase=0					):
       A       :       Optional[Any]							=						torch.manual_seed(_SCREAMING_SNAKE_CASE					)
       A       :       Any							=						{
           '''prompt''': '''a photo of the dolomites''',
           '''generator''': generator,
           '''num_inference_steps''': 3,
           '''guidance_scale''': 7.5,
           '''output_type''': '''numpy''',
       }
       return inputs


   def 					snake_case				( self					):
       A       :       int							=						'''stabilityai/stable-diffusion-2-base'''
       A       :       Tuple							=						DDIMScheduler.from_pretrained(_SCREAMING_SNAKE_CASE    ,   subfolder='''scheduler'''					)
       A       :       int							=						StableDiffusionPanoramaPipeline.from_pretrained(_SCREAMING_SNAKE_CASE    ,   scheduler=_SCREAMING_SNAKE_CASE    ,   safety_checker=_SCREAMING_SNAKE_CASE					)
       pipe.to(_SCREAMING_SNAKE_CASE					)
       pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)
       pipe.enable_attention_slicing()

       A       :       Any							=						self.get_inputs()
       A       :       List[str]							=						pipe(**_SCREAMING_SNAKE_CASE					).images
       A       :       str							=						image[0, -3:, -3:, -1].flatten()

       assert image.shape == (1, 512, 2_048, 3)

       A       :       List[Any]							=						np.array(
           [
               0.36968392,
               0.27025372,
               0.32446766,
               0.28379387,
               0.36363274,
               0.30733347,
               0.27100027,
               0.27054125,
               0.25536096,
           ]					)

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


   def 					snake_case				( self					):
       A       :       Optional[Any]							=						StableDiffusionPanoramaPipeline.from_pretrained(
           '''stabilityai/stable-diffusion-2-base'''    ,   safety_checker=_SCREAMING_SNAKE_CASE					)
       A       :       Any							=						LMSDiscreteScheduler.from_config(pipe.scheduler.config					)
       pipe.to(_SCREAMING_SNAKE_CASE					)
       pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)
       pipe.enable_attention_slicing()

       A       :       Optional[Any]							=						self.get_inputs()
       A       :       List[Any]							=						pipe(**_SCREAMING_SNAKE_CASE					).images
       A       :       Dict							=						image[0, -3:, -3:, -1].flatten()

       assert image.shape == (1, 512, 2_048, 3)

       A       :       int							=						np.array(
           [
               [
                   0.0,
                   0.0,
                   0.0,
                   0.0,
                   0.0,
                   0.0,
                   0.0,
                   0.0,
                   0.0,
               ]
           ]					)

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


   def 					snake_case				( self					):
       A       :       List[Any]							=						0

       def callback_fn(_UpperCAmelCase    ,   _UpperCAmelCase    ,   _UpperCAmelCase					) -> None:
           A       :       Optional[int]							=						True
           nonlocal number_of_steps
           number_of_steps += 1
           if step == 1:
               A       :       Dict							=						latents.detach().cpu().numpy()
               assert latents.shape == (1, 4, 64, 256)
               A       :       Optional[Any]							=						latents[0, -3:, -3:, -1]

               A       :       str							=						np.array(
                   [
                       0.18681869,
                       0.33907816,
                       0.5361276,
                       0.14432865,
                       -0.02856611,
                       -0.73941123,
                       0.23397987,
                       0.47322682,
                       -0.37823164,
                   ]					)
               assert np.abs(latents_slice.flatten() - expected_slice					).max() < 5E-2
           elif step == 2:
               A       :       Tuple							=						latents.detach().cpu().numpy()
               assert latents.shape == (1, 4, 64, 256)
               A       :       List[str]							=						latents[0, -3:, -3:, -1]

               A       :       List[Any]							=						np.array(
                   [
                       0.18539645,
                       0.33987248,
                       0.5378559,
                       0.14437142,
                       -0.02455261,
                       -0.7338317,
                       0.23990755,
                       0.47356272,
                       -0.3786505,
                   ]					)

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

       A       :       str							=						False

       A       :       List[Any]							=						'''stabilityai/stable-diffusion-2-base'''
       A       :       Union[str, Any]							=						DDIMScheduler.from_pretrained(_SCREAMING_SNAKE_CASE    ,   subfolder='''scheduler'''					)
       A       :       Dict							=						StableDiffusionPanoramaPipeline.from_pretrained(_SCREAMING_SNAKE_CASE    ,   scheduler=_SCREAMING_SNAKE_CASE    ,   safety_checker=_SCREAMING_SNAKE_CASE					)
       A       :       Any							=						pipe.to(_SCREAMING_SNAKE_CASE					)
       pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)
       pipe.enable_attention_slicing()

       A       :       List[Any]							=						self.get_inputs()
       pipe(**_SCREAMING_SNAKE_CASE    ,   callback=_SCREAMING_SNAKE_CASE    ,   callback_steps=1					)
       assert callback_fn.has_been_called
       assert number_of_steps == 3


   def 					snake_case				( self					):
       torch.cuda.empty_cache()
       torch.cuda.reset_max_memory_allocated()
       torch.cuda.reset_peak_memory_stats()

       A       :       Union[str, Any]							=						'''stabilityai/stable-diffusion-2-base'''
       A       :       Optional[int]							=						DDIMScheduler.from_pretrained(_SCREAMING_SNAKE_CASE    ,   subfolder='''scheduler'''					)
       A       :       str							=						StableDiffusionPanoramaPipeline.from_pretrained(_SCREAMING_SNAKE_CASE    ,   scheduler=_SCREAMING_SNAKE_CASE    ,   safety_checker=_SCREAMING_SNAKE_CASE					)
       A       :       List[Any]							=						pipe.to(_SCREAMING_SNAKE_CASE					)
       pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE					)
       pipe.enable_attention_slicing(1					)
       pipe.enable_sequential_cpu_offload()

       A       :       Tuple							=						self.get_inputs()
       A       :       Tuple							=						pipe(**_SCREAMING_SNAKE_CASE					)

       A       :       Optional[Any]							=						torch.cuda.max_memory_allocated()
       # make sure that less than 5.2 GB is allocated
       assert mem_bytes < 5.5 * 10**9







'''simple docstring'''






from pathlib import Path
from typing import List

from transformers import is_torch_available, is_vision_available
from transformers.testing_utils import get_tests_dir, is_tool_test
from transformers.tools.agent_types import AGENT_TYPE_MAPPING, AgentAudio, AgentImage, AgentText


if is_torch_available():
     import torch

if is_vision_available():
     from PIL import Image


snake_case_     =	["""text""", """image""", """audio"""]
def 							_lowerCamelCase(  UpperCamelCase__    : List[str]		)		->  str:
    A       :       int							=						[]

    for input_type in input_types:
        if input_type == "text":
            inputs.append('''Text input'''		)
        elif input_type == "image":
            inputs.append(
                Image.open(Path(get_tests_dir('''fixtures/tests_samples/COCO'''		)		) / '''000000039769.png'''		).resize((512, 512)		)		)
        elif input_type == "audio":
            inputs.append(torch.ones(3_000		)		)
        elif isinstance(UpperCamelCase__    ,       UpperCamelCase__		):
            inputs.append(create_inputs(UpperCamelCase__		)		)
        else:
            raise ValueError(F'''Invalid type requested: {input_type}'''		)

    return inputs
def 							_lowerCamelCase(  UpperCamelCase__    : List		)		->  Tuple:
    A       :       Optional[int]							=						[]

    for output in outputs:
        if isinstance(UpperCamelCase__    ,       (str, AgentText)		):
            output_types.append('''text'''		)
        elif isinstance(UpperCamelCase__    ,       (Image.Image, AgentImage)		):
            output_types.append('''image'''		)
        elif isinstance(UpperCamelCase__    ,       (torch.Tensor, AgentAudio)		):
            output_types.append('''audio'''		)
        else:
            raise ValueError(F'''Invalid output: {output}'''		)

    return output_types





@is_tool_test
class   _lowercase						:


   def 					snake_case				( self					):
       self.assertTrue(hasattr(self.tool    ,   '''inputs'''					)					)
       self.assertTrue(hasattr(self.tool    ,   '''outputs'''					)					)

       A       :       Any							=						self.tool.inputs
       for _input in inputs:
           if isinstance(_input    ,   _UpperCAmelCase					):
               for __input in _input:
                   self.assertTrue(__input in authorized_types					)
           else:
               self.assertTrue(_input in authorized_types					)

       A       :       Tuple							=						self.tool.outputs
       for _output in outputs:
           self.assertTrue(_output in authorized_types					)


   def 					snake_case				( self					):
       A       :       Any							=						create_inputs(self.tool.inputs					)
       A       :       Dict							=						self.tool(*_UpperCAmelCase					)

       # There is a single output
       if len(self.tool.outputs					) == 1:
           A       :       Optional[int]							=						[outputs]

       self.assertListEqual(output_types(_UpperCAmelCase					)    ,   self.tool.outputs					)


   def 					snake_case				( self					):
       self.assertTrue(hasattr(self.tool    ,   '''description'''					)					)
       self.assertTrue(hasattr(self.tool    ,   '''default_checkpoint'''					)					)
       self.assertTrue(self.tool.description.startswith('''This is a tool that'''					)					)


   def 					snake_case				( self					):
       A       :       List[str]							=						create_inputs(self.tool.inputs					)
       A       :       Dict							=						self.tool(*_UpperCAmelCase					)

       if not isinstance(_UpperCAmelCase    ,   _UpperCAmelCase					):
           A       :       int							=						[outputs]

       self.assertEqual(len(_UpperCAmelCase					)    ,   len(self.tool.outputs					)					)

       for output, output_type in zip(_UpperCAmelCase    ,   self.tool.outputs					):
           A       :       Tuple							=						AGENT_TYPE_MAPPING[output_type]
           self.assertTrue(isinstance(_UpperCAmelCase    ,   _UpperCAmelCase					)					)


   def 					snake_case				( self					):
       A       :       Tuple							=						create_inputs(self.tool.inputs					)

       A       :       Dict							=						[]

       for _input, input_type in zip(_UpperCAmelCase    ,   self.tool.inputs					):
           if isinstance(_UpperCAmelCase    ,   _UpperCAmelCase					):
               _inputs.append([AGENT_TYPE_MAPPING[_input_type](_input					) for _input_type in input_type]					)
           else:
               _inputs.append(AGENT_TYPE_MAPPING[input_type](_input					)					)

        # Should not raise an error
       A       :       List[Any]							=						self.tool(*_UpperCAmelCase					)

       if not isinstance(_UpperCAmelCase    ,   _UpperCAmelCase					):
           A       :       Any							=						[outputs]

       self.assertEqual(len(_UpperCAmelCase					)    ,   len(self.tool.outputs					)					)
