'''simple docstring'''




import unittest

import torch

from diffusers import UNetaDModel
from diffusers.utils import floats_tensor, slow, torch_device

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin

class   _lowercase(     _lowerCamelCase   ,_lowerCamelCase   ,unittest.TestCase				):



   """simple docstring"""





   __lowerCamelCase    =  UNetaDModel
   __lowerCamelCase    =  '''sample'''





   @property
   def 			snake_case	(							self:    str	):
    __UpperCAmelCase         =					4
    __UpperCAmelCase         =					14
    __UpperCAmelCase         =					16

    __UpperCAmelCase         =					floats_tensor((batch_size, num_features, seq_len)	).to(a	)
    __UpperCAmelCase         =					torch.tensor([10] * batch_size	).to(a	)

    return {"sample": noise, "timestep": time_step}





   @property
   def 			snake_case	(							self:    Dict	):
    return (4, 14, 16)





   @property
   def 			snake_case	(							self:    Optional[Any]	):
    return (4, 14, 16)





   def 			snake_case	(							self:    int	):
    pass





   def 			snake_case	(							self:    Optional[int]	):
    pass





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    List[str]	):
    super().test_determinism()





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    Optional[int]	):
    super().test_outputs_equivalence()





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    Optional[int]	):
    super().test_from_save_pretrained()





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    Optional[int]	):
    super().test_from_save_pretrained_variant()





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    Union[str, Any]	):
    super().test_model_from_pretrained()





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    int	):
    super().test_output()





   def 			snake_case	(							self:    List[Any]	):
    __UpperCAmelCase         =					{
        'block_out_channels': (32, 64, 128, 256),
        'in_channels': 14,
        'out_channels': 14,
        'time_embedding_type': 'positional',
        'use_timestep_embedding': True,
        'flip_sin_to_cos': False,
        'freq_shift': 1.0,
        'out_block_type': 'OutConv1DBlock',
        'mid_block_type': 'MidResTemporalBlock1D',
        'down_block_types': ('DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D'),
        'up_block_types': ('UpResnetBlock1D', 'UpResnetBlock1D', 'UpResnetBlock1D'),
        'act_fn': 'mish',
    }
    __UpperCAmelCase         =					self.dummy_input
    return init_dict, inputs_dict





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    int	):
    __UpperCAmelCase  ,			__UpperCAmelCase         =					UNetaDModel.from_pretrained(
        'bglick13/hopper-medium-v2-value-function-hor32'			,output_loading_info=a			,subfolder='unet'	)
    self.assertIsNotNone(a	)
    self.assertEqual(len(loading_info['missing_keys']	)			,0	)

    model.to(a	)
    __UpperCAmelCase         =					model(**self.dummy_input	)

    assert image is not None, "Make sure output is not None"





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    Optional[Any]	):
    __UpperCAmelCase         =					UNetaDModel.from_pretrained('bglick13/hopper-medium-v2-value-function-hor32'			,subfolder='unet'	)
    torch.manual_seed(0	)
    if torch.cuda.is_available():
     torch.cuda.manual_seed_all(0	)

    __UpperCAmelCase         =					model.config.in_channels
    __UpperCAmelCase         =					16
    __UpperCAmelCase         =					torch.randn((1, seq_len, num_features)	).permute(
        0			,2			,1	)  # match original, we can update values and remove
    __UpperCAmelCase         =					torch.full((num_features,)			,0	)

    with torch.no_grad():
     __UpperCAmelCase         =					model(a			,a	).sample.permute(0			,2			,1	)

    __UpperCAmelCase         =					output[0, -3:, -3:].flatten()
    # fmt: off
    __UpperCAmelCase         =					torch.tensor([-2.137172, 1.1426016, 0.3688687, -0.766922, 0.7303146, 0.11038864, -0.4760633, 0.13270172, 0.02591348]	)
    # fmt: on
    self.assertTrue(torch.allclose(a			,a			,rtol=1e-3	)	)





   def 			snake_case	(							self:    str	):
    # Not implemented yet for this UNet
    pass





   @slow
   def 			snake_case	(							self:    List[str]	):
    __UpperCAmelCase         =					'harmonai/maestro-150k'
    __UpperCAmelCase         =					UNetaDModel.from_pretrained(a			,subfolder='unet'	)
    model.to(a	)

    __UpperCAmelCase         =					65536
    __UpperCAmelCase         =					torch.sin(torch.arange(a	)[None, None, :].repeat(1			,2			,1	)	).to(a	)
    __UpperCAmelCase         =					torch.tensor([1]	).to(a	)

    with torch.no_grad():
     __UpperCAmelCase         =					model(a			,a	).sample

    __UpperCAmelCase         =					output.abs().sum()
    __UpperCAmelCase         =					output.abs().max()

    assert (output_sum - 224.0896).abs() < 0.5
    assert (output_max - 0.0607).abs() < 4e-4




class   _lowercase(     _lowerCamelCase   ,_lowerCamelCase   ,unittest.TestCase				):



   """simple docstring"""





   __lowerCamelCase    =  UNetaDModel
   __lowerCamelCase    =  '''sample'''





   @property
   def 			snake_case	(							self:    str	):
    __UpperCAmelCase         =					4
    __UpperCAmelCase         =					14
    __UpperCAmelCase         =					16

    __UpperCAmelCase         =					floats_tensor((batch_size, num_features, seq_len)	).to(a	)
    __UpperCAmelCase         =					torch.tensor([10] * batch_size	).to(a	)

    return {"sample": noise, "timestep": time_step}





   @property
   def 			snake_case	(							self:    str	):
    return (4, 14, 16)





   @property
   def 			snake_case	(							self:    Optional[int]	):
    return (4, 14, 1)





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    str	):
    super().test_determinism()





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    Optional[Any]	):
    super().test_outputs_equivalence()





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    Union[str, Any]	):
    super().test_from_save_pretrained()





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    Tuple	):
    super().test_from_save_pretrained_variant()





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    Any	):
    super().test_model_from_pretrained()





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    Optional[int]	):
    # UNetRL is a value-function is different output shape
    __UpperCAmelCase  ,			__UpperCAmelCase         =					self.prepare_init_args_and_inputs_for_common()
    __UpperCAmelCase         =					self.model_class(**a	)
    model.to(a	)
    model.eval()

    with torch.no_grad():
     __UpperCAmelCase         =					model(**a	)

     if isinstance(a			,a	):
      __UpperCAmelCase         =					output.sample

    self.assertIsNotNone(a	)
    __UpperCAmelCase         =					torch.Size((inputs_dict['sample'].shape[0], 1)	)
    self.assertEqual(output.shape			,a			,'Input and output shapes do not match'	)





   def 			snake_case	(							self:    Optional[Any]	):
    pass





   def 			snake_case	(							self:    Any	):
    pass





   def 			snake_case	(							self:    str	):
    __UpperCAmelCase         =					{
        'in_channels': 14,
        'out_channels': 14,
        'down_block_types': ['DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D', 'DownResnetBlock1D'],
        'up_block_types': [],
        'out_block_type': 'ValueFunction',
        'mid_block_type': 'ValueFunctionMidBlock1D',
        'block_out_channels': [32, 64, 128, 256],
        'layers_per_block': 1,
        'downsample_each_block': True,
        'use_timestep_embedding': True,
        'freq_shift': 1.0,
        'flip_sin_to_cos': False,
        'time_embedding_type': 'positional',
        'act_fn': 'mish',
    }
    __UpperCAmelCase         =					self.dummy_input
    return init_dict, inputs_dict





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    List[Any]	):
    __UpperCAmelCase  ,			__UpperCAmelCase         =					UNetaDModel.from_pretrained(
        'bglick13/hopper-medium-v2-value-function-hor32'			,output_loading_info=a			,subfolder='value_function'	)
    self.assertIsNotNone(a	)
    self.assertEqual(len(vf_loading_info['missing_keys']	)			,0	)

    value_function.to(a	)
    __UpperCAmelCase         =					value_function(**self.dummy_input	)

    assert image is not None, "Make sure output is not None"





   @unittest.skipIf(torch_device == 'mps'			,'mish op not supported in MPS'	)
   def 			snake_case	(							self:    List[str]	):
    __UpperCAmelCase  ,			__UpperCAmelCase         =					UNetaDModel.from_pretrained(
        'bglick13/hopper-medium-v2-value-function-hor32'			,output_loading_info=a			,subfolder='value_function'	)
    torch.manual_seed(0	)
    if torch.cuda.is_available():
     torch.cuda.manual_seed_all(0	)

    __UpperCAmelCase         =					value_function.config.in_channels
    __UpperCAmelCase         =					14
    __UpperCAmelCase         =					torch.randn((1, seq_len, num_features)	).permute(
        0			,2			,1	)  # match original, we can update values and remove
    __UpperCAmelCase         =					torch.full((num_features,)			,0	)

    with torch.no_grad():
     __UpperCAmelCase         =					value_function(a			,a	).sample

    # fmt: off
    __UpperCAmelCase         =					torch.tensor([165.25] * seq_len	)
    # fmt: on
    self.assertTrue(torch.allclose(a			,a			,rtol=1e-3	)	)





   def 			snake_case	(							self:    Dict	):
    # Not implemented yet for this UNet
    pass




'''simple docstring'''




from typing import Union

import torch
from PIL import Image
from torchvision import transforms as tfms
from tqdm.auto import tqdm
from transformers import CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)

class   _lowercase(     _lowerCamelCase				):



   """simple docstring"""





   def __init__(							self:    List[str]			,a:    AutoencoderKL			,a:    CLIPTextModel			,a:    CLIPTokenizer			,a:    UNetaDConditionModel			,a:    Union[PNDMScheduler, LMSDiscreteScheduler, DDIMScheduler]			,):
    super().__init__()

    self.register_modules(vae=a			,text_encoder=a			,tokenizer=a			,unet=a			,scheduler=a	)





   def 			snake_case	(							self:    Union[str, Any]			,a:    Any	):
    with torch.no_grad():
     __UpperCAmelCase         =					self.vae.encode(tfms.ToTensor()(a	).unsqueeze(0	).to(self.device	) * 2 - 1	)
     __UpperCAmelCase         =					0.18215 * latent.latent_dist.sample()
    return latent





   def 			snake_case	(							self:    Any			,a:    List[str]	):
    __UpperCAmelCase         =					(1 / 0.18215) * latent
    with torch.no_grad():
     __UpperCAmelCase         =					self.vae.decode(a	).sample
    __UpperCAmelCase         =					(img / 2 + 0.5).clamp(0			,1	)
    __UpperCAmelCase         =					img.detach().cpu().permute(0			,2			,3			,1	).numpy()
    __UpperCAmelCase         =					(img * 255).round().astype('uint8'	)
    return Image.fromarray(img[0]	)





   def 			snake_case	(							self:    Any			,a:    str	):
    __UpperCAmelCase         =					self.tokenizer(
        a			,padding='max_length'			,max_length=self.tokenizer.model_max_length			,truncation=a			,return_tensors='pt'			,)

    __UpperCAmelCase         =					self.text_encoder(text_input.input_ids.to(self.device	)	)[0]

    __UpperCAmelCase         =					self.tokenizer(
        ''			,padding='max_length'			,max_length=self.tokenizer.model_max_length			,truncation=a			,return_tensors='pt'			,)

    __UpperCAmelCase         =					self.text_encoder(uncond_input.input_ids.to(self.device	)	)[0]

    return torch.cat([uncond_embedding, text_embedding]	)





   def __call__(							self:    Optional[Any]			,a:    Image.Image			,a:    str			,a:    float = 0.3			,a:    float = 0.6			,a:    float = 0.5			,a:    int = 42			,a:    int = 50			,a:    float = 7.5			,):
    __UpperCAmelCase         =					steps - int(kmin * steps	)
    __UpperCAmelCase         =					steps - int(kmax * steps	)

    __UpperCAmelCase         =					self.prep_text(a	)

    self.scheduler.set_timesteps(a	)

    __UpperCAmelCase  ,			__UpperCAmelCase         =					img.size
    __UpperCAmelCase         =					self.encode(a	)

    torch.manual_seed(a	)
    __UpperCAmelCase         =					torch.randn(
        (1, self.unet.config.in_channels, height // 8, width // 8)			,).to(self.device	)

    __UpperCAmelCase         =					self.scheduler.add_noise(
        a			,a			,timesteps=self.scheduler.timesteps[tmax]			,)

    __UpperCAmelCase         =					torch.cat([latents] * 2	)

    __UpperCAmelCase         =					self.scheduler.scale_model_input(a			,self.scheduler.timesteps[tmax]	)

    with torch.no_grad():
     __UpperCAmelCase         =					self.unet(
         a			,self.scheduler.timesteps[tmax]			,encoder_hidden_states=a			,).sample

    __UpperCAmelCase  ,			__UpperCAmelCase         =					pred.chunk(2	)
    __UpperCAmelCase         =					pred_uncond + guidance_scale * (pred_text - pred_uncond)

    __UpperCAmelCase         =					self.scheduler.step(a			,self.scheduler.timesteps[tmax]			,a	).prev_sample

    for i, t in enumerate(tqdm(self.scheduler.timesteps	)	):
     if i > tmax:
      if i < tmin:  # layout generation phase
       __UpperCAmelCase         =					self.scheduler.add_noise(
           a			,a			,timesteps=a			,)

       __UpperCAmelCase         =					(mix_factor * latents) + (
           1 - mix_factor
       ) * orig_latents  # interpolating between layout noise and conditionally generated noise to preserve layout sematics
       __UpperCAmelCase         =					torch.cat([input] * 2	)

      else:  # content generation phase
       __UpperCAmelCase         =					torch.cat([latents] * 2	)

      __UpperCAmelCase         =					self.scheduler.scale_model_input(a			,a	)

      with torch.no_grad():
       __UpperCAmelCase         =					self.unet(
           a			,a			,encoder_hidden_states=a			,).sample

      __UpperCAmelCase  ,			__UpperCAmelCase         =					pred.chunk(2	)
      __UpperCAmelCase         =					pred_uncond + guidance_scale * (pred_text - pred_uncond)

      __UpperCAmelCase         =					self.scheduler.step(a			,a			,a	).prev_sample

    return self.decode(a	)



