




import warnings
from contextlib import contextmanager

from ...processing_utils import ProcessorMixin




class  __lowerCAmelCase							(				lowerCAmelCase):
       _a			       =    ['''image_processor''', '''tokenizer''']
       _a			       =    '''AutoImageProcessor'''
       _a			       =    '''AutoTokenizer'''






       def __init__( self:       Any							,       _lowerCAmelCase:       List[Any]=None							,       _lowerCAmelCase:       int=None							,       **_lowerCAmelCase:       str					):
         lowercase     :Optional[Any]   						=    None
         if "feature_extractor" in kwargs:
           warnings.warn(
               "The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`"
               " instead."							,       _lowerCAmelCase							,       )
           lowercase     :Dict   						=    kwargs.pop("feature_extractor"					)

         lowercase     :int   						=    image_processor if image_processor is not None else feature_extractor
         if image_processor is None:
           raise ValueError("You need to specify an `image_processor`."					)
         if tokenizer is None:
           raise ValueError("You need to specify a `tokenizer`."					)

         super().__init__(_lowerCAmelCase							,       _lowerCAmelCase					)
         lowercase     :Optional[int]   						=    self.image_processor
         lowercase     :List[Any]   						=    False






       def __call__( self:       Optional[int]							,       *_lowerCAmelCase:       Optional[Any]							,       **_lowerCAmelCase:       Optional[int]					):
         # For backward compatibility
         if self._in_target_context_manager:
           return self.current_processor(*_lowerCAmelCase							,       **_lowerCAmelCase					)

         lowercase     :int   						=    kwargs.pop("images"							,       _lowerCAmelCase					)
         lowercase     :Optional[int]   						=    kwargs.pop("text"							,       _lowerCAmelCase					)
         if len(_lowerCAmelCase					) > 0:
           lowercase     :Optional[Any]   						=    args[0]
           lowercase     :str   						=    args[1:]

         if images is None and text is None:
           raise ValueError("You need to specify either an `images` or `text` input to process."					)

         if images is not None:
           lowercase     :Any   						=    self.image_processor(_lowerCAmelCase							,       *_lowerCAmelCase							,       **_lowerCAmelCase					)
         if text is not None:
           lowercase     :Optional[Any]   						=    self.tokenizer(_lowerCAmelCase							,       **_lowerCAmelCase					)

         if text is None:
           return inputs
         elif images is None:
           return encodings
         else:
           lowercase     :Union[str, Any]   						=    encodings["input_ids"]
           return inputs






       def        SCREAMING_SNAKE_CASE     ( self:       List[str]							,       *_lowerCAmelCase:       Union[str, Any]							,       **_lowerCAmelCase:       str					):
         return self.tokenizer.batch_decode(*_lowerCAmelCase							,       **_lowerCAmelCase					)






       def        SCREAMING_SNAKE_CASE     ( self:       Optional[Any]							,       *_lowerCAmelCase:       Dict							,       **_lowerCAmelCase:       Any					):
         return self.tokenizer.decode(*_lowerCAmelCase							,       **_lowerCAmelCase					)






       @contextmanager
       def        SCREAMING_SNAKE_CASE     ( self:       Optional[Any]					):
         warnings.warn(
             "`as_target_processor` is deprecated and will be removed in v5 of Transformers. You can process your "
             "labels by using the argument `text` of the regular `__call__` method (either in the same call as "
             "your images inputs, or in a separate call."					)
         lowercase     :Union[str, Any]   						=    True
         lowercase     :Tuple   						=    self.tokenizer
         yield
         lowercase     :Optional[int]   						=    self.image_processor
         lowercase     :str   						=    False






       @property
       def        SCREAMING_SNAKE_CASE     ( self:       List[str]					):
         warnings.warn(
             "`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead."							,       _lowerCAmelCase							,       )
         return self.image_processor_class






       @property
       def        SCREAMING_SNAKE_CASE     ( self:       int					):
         warnings.warn(
             "`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead."							,       _lowerCAmelCase							,       )
         return self.image_processor









import tempfile
import unittest

import torch

from diffusers import UNetaDConditionModel
from diffusers.training_utils import EMAModel
from diffusers.utils.testing_utils import enable_full_determinism, skip_mps, torch_device


enable_full_determinism()




class  __lowerCAmelCase							(				unittest.TestCase):
       _a			       =    '''hf-internal-testing/tiny-stable-diffusion-pipe'''
       _a			       =    1
       _a			       =    77
       _a			       =    32
       _a			       =    4
       _a			       =    _a			       =    64
       _a			       =    torch.manual_seed(0)






       def        SCREAMING_SNAKE_CASE     ( self:       Union[str, Any]							,       _lowerCAmelCase:       int=0.99_99					):
         lowercase     :Optional[int]   						=    UNetaDConditionModel.from_pretrained(self.model_id							,       subfolder="unet"					)
         lowercase     :Union[str, Any]   						=    unet.to(_lowerCAmelCase					)
         lowercase     :Any   						=    EMAModel(unet.parameters()							,       decay=_lowerCAmelCase							,       model_cls=_lowerCAmelCase							,       model_config=unet.config					)
         return unet, ema_unet






       def        SCREAMING_SNAKE_CASE     ( self:       List[str]					):
         lowercase     :Optional[Any]   						=    torch.randn(
             self.batch_size							,       self.num_in_channels							,       self.latent_height							,       self.latent_width							,       generator=self.generator					).to(_lowerCAmelCase					)
         lowercase     :Dict   						=    torch.randint(0							,       10_00							,       size=(self.batch_size,)							,       generator=self.generator					).to(_lowerCAmelCase					)
         lowercase     :List[Any]   						=    torch.randn(
             self.batch_size							,       self.prompt_length							,       self.text_encoder_hidden_dim							,       generator=self.generator					).to(_lowerCAmelCase					)
         return noisy_latents, timesteps, encoder_hidden_states






       def        SCREAMING_SNAKE_CASE     ( self:       Optional[int]							,       _lowerCAmelCase:       Optional[Any]					):
         lowercase     :Any   						=    {}
         for k, param in unet.state_dict().items():
           lowercase     :Optional[int]   						=    torch.randn_like(_lowerCAmelCase					) + (param * torch.randn_like(_lowerCAmelCase					))
           updated_state_dict.update({k: updated_param}					)
         unet.load_state_dict(_lowerCAmelCase					)
         return unet






       def        SCREAMING_SNAKE_CASE     ( self:       int					):
         lowercase						,     lowercase     :List[str]   						=    self.get_models()
         # Take the first (hypothetical) EMA step.
         ema_unet.step(unet.parameters()					)
         assert ema_unet.optimization_step == 1

         # Take two more.
         for _ in range(2					):
           ema_unet.step(unet.parameters()					)
         assert ema_unet.optimization_step == 3






       def        SCREAMING_SNAKE_CASE     ( self:       Any					):
         lowercase						,     lowercase     :str   						=    self.get_models()
         # Since the `unet` is not being updated (i.e., backprop'd)
         # there won't be any difference between the `params` of `unet`
         # and `ema_unet` even if we call `ema_unet.step(unet.parameters())`.
         ema_unet.step(unet.parameters()					)
         lowercase     :Tuple   						=    list(unet.parameters()					)
         for s_param, param in zip(ema_unet.shadow_params							,       _lowerCAmelCase					):
           assert torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase					)

         # The above holds true even if we call `ema.step()` multiple times since
         # `unet` params are still not being updated.
         for _ in range(4					):
           ema_unet.step(unet.parameters()					)
         for s_param, param in zip(ema_unet.shadow_params							,       _lowerCAmelCase					):
           assert torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase					)






       def        SCREAMING_SNAKE_CASE     ( self:       Dict					):
         lowercase						,     lowercase     :Optional[int]   						=    self.get_models()
         # Here we simulate the parameter updates for `unet`. Since there might
         # be some parameters which are initialized to zero we take extra care to
         # initialize their values to something non-zero before the multiplication.
         lowercase     :List[str]   						=    self.simulate_backprop(_lowerCAmelCase					)

         # Take the EMA step.
         ema_unet.step(unet_pseudo_updated_step_one.parameters()					)

         # Now the EMA'd parameters won't be equal to the original model parameters.
         lowercase     :str   						=    list(unet_pseudo_updated_step_one.parameters()					)
         for s_param, param in zip(ema_unet.shadow_params							,       _lowerCAmelCase					):
           assert ~torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase					)

         # Ensure this is the case when we take multiple EMA steps.
         for _ in range(4					):
           ema_unet.step(unet.parameters()					)
         for s_param, param in zip(ema_unet.shadow_params							,       _lowerCAmelCase					):
           assert ~torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase					)






       def        SCREAMING_SNAKE_CASE     ( self:       Optional[int]					):
         # If we call EMA step after a backpropagation consecutively for two times,
         # the shadow params from those two steps should be different.
         lowercase						,     lowercase     :Optional[Any]   						=    self.get_models()

         # First backprop + EMA
         lowercase     :Union[str, Any]   						=    self.simulate_backprop(_lowerCAmelCase					)
         ema_unet.step(unet_step_one.parameters()					)
         lowercase     :int   						=    ema_unet.shadow_params

         # Second backprop + EMA
         lowercase     :Optional[int]   						=    self.simulate_backprop(_lowerCAmelCase					)
         ema_unet.step(unet_step_two.parameters()					)
         lowercase     :Optional[int]   						=    ema_unet.shadow_params

         for step_one, step_two in zip(_lowerCAmelCase							,       _lowerCAmelCase					):
           assert ~torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase					)






       def        SCREAMING_SNAKE_CASE     ( self:       Union[str, Any]					):
         # If there's no decay even if there are backprops, EMA steps
         # won't take any effect i.e., the shadow params would remain the
         # same.
         lowercase						,     lowercase     :Any   						=    self.get_models(decay=0.0					)
         lowercase     :List[Any]   						=    self.simulate_backprop(_lowerCAmelCase					)
         ema_unet.step(unet_step_one.parameters()					)
         lowercase     :List[str]   						=    ema_unet.shadow_params

         lowercase     :str   						=    self.simulate_backprop(_lowerCAmelCase					)
         ema_unet.step(unet_step_two.parameters()					)
         lowercase     :List[str]   						=    ema_unet.shadow_params

         for step_one, step_two in zip(_lowerCAmelCase							,       _lowerCAmelCase					):
           assert torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase					)






       @skip_mps
       def        SCREAMING_SNAKE_CASE     ( self:       List[Any]					):
         lowercase						,     lowercase     :Optional[Any]   						=    self.get_models()
         lowercase						,     lowercase						,     lowercase     :str   						=    self.get_dummy_inputs()

         with tempfile.TemporaryDirectory() as tmpdir:
           ema_unet.save_pretrained(_lowerCAmelCase					)
           lowercase     :Optional[int]   						=    UNetaDConditionModel.from_pretrained(_lowerCAmelCase							,       model_cls=_lowerCAmelCase					)
           lowercase     :Tuple   						=    loaded_unet.to(unet.device					)

         # Since no EMA step has been performed the outputs should match.
         lowercase     :Dict   						=    unet(_lowerCAmelCase							,       _lowerCAmelCase							,       _lowerCAmelCase					).sample
         lowercase     :Any   						=    loaded_unet(_lowerCAmelCase							,       _lowerCAmelCase							,       _lowerCAmelCase					).sample

         assert torch.allclose(_lowerCAmelCase							,       _lowerCAmelCase							,       atol=1e-4					)



