






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 					a						( unittest.TestCase		):
    SCREAMING_SNAKE_CASE						:					str						=			"hf-internal-testing/tiny-stable-diffusion-pipe"
    SCREAMING_SNAKE_CASE						:					int						=			1
    SCREAMING_SNAKE_CASE						:					str						=			7_7
    SCREAMING_SNAKE_CASE						:					Dict						=			3_2
    SCREAMING_SNAKE_CASE						:					List[Any]						=			4
    SCREAMING_SNAKE_CASE						:					int						=			6_4
    SCREAMING_SNAKE_CASE						:					Optional[int]						=			torch.manual_seed(0		)






    def        UpperCamelCase	(	self					:							Union[str, Any]	,						__SCREAMING_SNAKE_CASE					:							Tuple=0.9_999 )					->		List[Any]:
      lowerCamelCase_   					=						UNetaDConditionModel.from_pretrained(self.model_id	,						subfolder='unet' )
      lowerCamelCase_   					=						unet.to(__A )
      lowerCamelCase_   					=						EMAModel(unet.parameters()	,						decay=__A	,						model_cls=__A	,						model_config=unet.config )
      return unet, ema_unet






    def        UpperCamelCase	(	self					:							str )					->		List[Any]:
      lowerCamelCase_   					=						torch.randn(
          self.batch_size	,						self.num_in_channels	,						self.latent_height	,						self.latent_width	,						generator=self.generator ).to(__A )
      lowerCamelCase_   					=						torch.randint(0	,						1000	,						size=(self.batch_size,)	,						generator=self.generator ).to(__A )
      lowerCamelCase_   					=						torch.randn(
          self.batch_size	,						self.prompt_length	,						self.text_encoder_hidden_dim	,						generator=self.generator ).to(__A )
      return noisy_latents, timesteps, encoder_hidden_states






    def        UpperCamelCase	(	self					:							int	,						__SCREAMING_SNAKE_CASE					:							Dict )					->		List[str]:
      lowerCamelCase_   					=						{}
      for k, param in unet.state_dict().items():
        lowerCamelCase_   					=						torch.randn_like(__A ) + (param * torch.randn_like(__A ))
        updated_state_dict.update({k: updated_param} )
      unet.load_state_dict(__A )
      return unet






    def        UpperCamelCase	(	self					:							Tuple )					->		Dict:
      lowerCamelCase_   					=						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        UpperCamelCase	(	self					:							Tuple )					->		str:
      lowerCamelCase_   					=						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() )
      lowerCamelCase_   					=						list(unet.parameters() )
      for s_param, param in zip(ema_unet.shadow_params	,						__A ):
        assert torch.allclose(__A	,						__A )

      # 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	,						__A ):
        assert torch.allclose(__A	,						__A )






    def        UpperCamelCase	(	self					:							Tuple )					->		List[str]:
      lowerCamelCase_   					=						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.
      lowerCamelCase_   					=						self.simulate_backprop(__A )

      # 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.
      lowerCamelCase_   					=						list(unet_pseudo_updated_step_one.parameters() )
      for s_param, param in zip(ema_unet.shadow_params	,						__A ):
        assert ~torch.allclose(__A	,						__A )

      # 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	,						__A ):
        assert ~torch.allclose(__A	,						__A )






    def        UpperCamelCase	(	self					:							Dict )					->		List[Any]:
      lowerCamelCase_   					=						self.get_models()

      # First backprop + EMA
      lowerCamelCase_   					=						self.simulate_backprop(__A )
      ema_unet.step(unet_step_one.parameters() )
      lowerCamelCase_   					=						ema_unet.shadow_params

      # Second backprop + EMA
      lowerCamelCase_   					=						self.simulate_backprop(__A )
      ema_unet.step(unet_step_two.parameters() )
      lowerCamelCase_   					=						ema_unet.shadow_params

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






    def        UpperCamelCase	(	self					:							Optional[int] )					->		List[Any]:
      lowerCamelCase_   					=						self.get_models(decay=0.0 )
      lowerCamelCase_   					=						self.simulate_backprop(__A )
      ema_unet.step(unet_step_one.parameters() )
      lowerCamelCase_   					=						ema_unet.shadow_params

      lowerCamelCase_   					=						self.simulate_backprop(__A )
      ema_unet.step(unet_step_two.parameters() )
      lowerCamelCase_   					=						ema_unet.shadow_params

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






    @skip_mps
    def        UpperCamelCase	(	self					:							int )					->		Optional[Any]:
      lowerCamelCase_   					=						self.get_models()
      lowerCamelCase_   					=						self.get_dummy_inputs()

      with tempfile.TemporaryDirectory() as tmpdir:
        ema_unet.save_pretrained(__A )
        lowerCamelCase_   					=						UNetaDConditionModel.from_pretrained(__A	,						model_cls=__A )
        lowerCamelCase_   					=						loaded_unet.to(unet.device )

      # Since no EMA step has been performed the outputs should match.
      lowerCamelCase_   					=						unet(__A	,						__A	,						__A ).sample
      lowerCamelCase_   					=						loaded_unet(__A	,						__A	,						__A ).sample

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












"""simple docstring"""





import enum
import inspect
from typing import Iterable, List, Optional, Tuple, Union






class 					a						( enum.Enum		):
    SCREAMING_SNAKE_CASE						:					str						=			"""timm"""
    SCREAMING_SNAKE_CASE						:					List[Any]						=			"""transformers"""







def 					lowerCamelCase__	(							_lowerCamelCase	:		Optional[Iterable[str]]		,    _lowerCamelCase	:		Optional[Iterable[int]]		,    _lowerCamelCase	:		Optional[Iterable[str]]			)				->    List[Any]:
  if stage_names is None:
    raise ValueError('Stage_names must be set for transformers backbones'			)

  if out_features is not None:
    if not isinstance(_lowerCamelCase		,    (list,)			):
      raise ValueError(F'''out_features must be a list {type(_lowerCamelCase			)}'''			)
    if any(feat not in stage_names for feat in out_features			):
      raise ValueError(F'''out_features must be a subset of stage_names: {stage_names} got {out_features}'''			)

  if out_indices is not None:
    if not isinstance(_lowerCamelCase		,    (list, tuple)			):
      raise ValueError(F'''out_indices must be a list or tuple, got {type(_lowerCamelCase			)}'''			)
    if any(idx >= len(_lowerCamelCase			) for idx in out_indices			):
      raise ValueError('out_indices must be valid indices for stage_names {stage_names}, got {out_indices}'			)

  if out_features is not None and out_indices is not None:
    if len(_lowerCamelCase			) != len(_lowerCamelCase			):
      raise ValueError('out_features and out_indices should have the same length if both are set'			)
    if out_features != [stage_names[idx] for idx in out_indices]:
      raise ValueError('out_features and out_indices should correspond to the same stages if both are set'			)







def 					lowerCamelCase__	(							_lowerCamelCase	:		Optional[List[str]]		,    _lowerCamelCase	:		Optional[Union[List[int], Tuple[int]]]		,    _lowerCamelCase	:		List[str]		,    )				->    Union[str, Any]:
  if out_indices is None and out_features is None:
    lowerCamelCase_   					=						[len(_lowerCamelCase			) - 1]
    lowerCamelCase_   					=						[stage_names[-1]]
  elif out_indices is None and out_features is not None:
    lowerCamelCase_   					=						[stage_names.index(_lowerCamelCase			) for layer in out_features]
  elif out_features is None and out_indices is not None:
    lowerCamelCase_   					=						[stage_names[idx] for idx in out_indices]
  return out_features, out_indices







def 					lowerCamelCase__	(							_lowerCamelCase	:		Optional[List[str]]		,    _lowerCamelCase	:		Optional[Union[List[int], Tuple[int]]]		,    _lowerCamelCase	:		List[str]		,    )				->    Tuple[List[str], List[int]]:
  verify_out_features_out_indices(out_features=_lowerCamelCase		,    out_indices=_lowerCamelCase		,    stage_names=_lowerCamelCase			)
  lowerCamelCase_					,     lowerCamelCase_   					=						_align_output_features_output_indices(
      out_features=_lowerCamelCase		,    out_indices=_lowerCamelCase		,    stage_names=_lowerCamelCase			)
  # Verify that the aligned out_features and out_indices are valid
  verify_out_features_out_indices(out_features=_lowerCamelCase		,    out_indices=_lowerCamelCase		,    stage_names=_lowerCamelCase			)
  return output_features, output_indices






class 					a						:
    SCREAMING_SNAKE_CASE						:					Optional[BackboneType]						=			None






    def        UpperCamelCase	(	self					:							Optional[Any]	,						__SCREAMING_SNAKE_CASE					:							List[str] )					->		None:
      if getattr(self	,						'_backbone'	,						__SCREAMING_SNAKE_CASE ) is None:
        raise ValueError('self._backbone must be set before calling _init_timm_backbone' )

      # These will diagree with the defaults for the transformers models e.g. for resnet50
      # the transformer model has out_features = ['stem', 'stage1', 'stage2', 'stage3', 'stage4']
      # the timm model has out_features = ['act', 'layer1', 'layer2', 'layer3', 'layer4']
      lowerCamelCase_   					=						[stage['module'] for stage in self._backbone.feature_info.info]
      lowerCamelCase_   					=						[stage['num_chs'] for stage in self._backbone.feature_info.info]
      lowerCamelCase_   					=						self._backbone.feature_info.out_indices
      lowerCamelCase_   					=						self._backbone.feature_info.module_name()

      # We verify the out indices and out features are valid
      verify_out_features_out_indices(
          out_features=__SCREAMING_SNAKE_CASE	,						out_indices=__SCREAMING_SNAKE_CASE	,						stage_names=self.stage_names )
      lowerCamelCase_					,     lowerCamelCase_   					=						out_features, out_indices






    def        UpperCamelCase	(	self					:							Tuple	,						__SCREAMING_SNAKE_CASE					:							Any )					->		None:
      lowerCamelCase_   					=						getattr(__SCREAMING_SNAKE_CASE	,						'stage_names' )
      lowerCamelCase_   					=						getattr(__SCREAMING_SNAKE_CASE	,						'out_features'	,						__SCREAMING_SNAKE_CASE )
      lowerCamelCase_   					=						getattr(__SCREAMING_SNAKE_CASE	,						'out_indices'	,						__SCREAMING_SNAKE_CASE )

      lowerCamelCase_   					=						stage_names
      lowerCamelCase_					,     lowerCamelCase_   					=						get_aligned_output_features_output_indices(
          out_features=__SCREAMING_SNAKE_CASE	,						out_indices=__SCREAMING_SNAKE_CASE	,						stage_names=__SCREAMING_SNAKE_CASE )
      # Number of channels for each stage. This is set in the transformer backbone model init
      lowerCamelCase_   					=						None






    def        UpperCamelCase	(	self					:							Any	,						__SCREAMING_SNAKE_CASE					:							Dict )					->		None:
      lowerCamelCase_   					=						config

      lowerCamelCase_   					=						getattr(__SCREAMING_SNAKE_CASE	,						'use_timm_backbone'	,						__SCREAMING_SNAKE_CASE )
      lowerCamelCase_   					=						BackboneType.TIMM if self.use_timm_backbone else BackboneType.TRANSFORMERS

      if self.backbone_type == BackboneType.TIMM:
        self._init_timm_backbone(__SCREAMING_SNAKE_CASE )
      elif self.backbone_type == BackboneType.TRANSFORMERS:
        self._init_transformers_backbone(__SCREAMING_SNAKE_CASE )
      else:
        raise ValueError(F'''backbone_type {self.backbone_type} not supported.''' )






    @property
    def        UpperCamelCase	(	self					:							Dict )					->		Any:
      return self._out_features






    @out_features.setter
    def        UpperCamelCase	(	self					:							Optional[Any]	,						__SCREAMING_SNAKE_CASE					:							List[str] )					->		List[Any]:
      lowerCamelCase_					,     lowerCamelCase_   					=						get_aligned_output_features_output_indices(
          out_features=__SCREAMING_SNAKE_CASE	,						out_indices=__SCREAMING_SNAKE_CASE	,						stage_names=self.stage_names )






    @property
    def        UpperCamelCase	(	self					:							List[Any] )					->		Dict:
      return self._out_indices






    @out_indices.setter
    def        UpperCamelCase	(	self					:							List[str]	,						__SCREAMING_SNAKE_CASE					:							Union[Tuple[int], List[int]] )					->		List[str]:
      lowerCamelCase_					,     lowerCamelCase_   					=						get_aligned_output_features_output_indices(
          out_features=__SCREAMING_SNAKE_CASE	,						out_indices=__SCREAMING_SNAKE_CASE	,						stage_names=self.stage_names )






    @property
    def        UpperCamelCase	(	self					:							Tuple )					->		Tuple:
      # the current backbones will output the number of channels for each stage
      # even if that stage is not in the out_features list.
      return {stage: self.num_features[i] for i, stage in enumerate(self.stage_names )}






    @property
    def        UpperCamelCase	(	self					:							Dict )					->		List[Any]:
      return [self.out_feature_channels[name] for name in self.out_features]






    def        UpperCamelCase	(	self					:							Optional[int]	,						*__SCREAMING_SNAKE_CASE					:							Any	,						**__SCREAMING_SNAKE_CASE					:							str )					->		Tuple:
      lowerCamelCase_   					=						dict(inspect.signature(self.forward ).parameters )
      lowerCamelCase_   					=						{k: v for k, v in kwargs.items() if k in signature}
      return self(*__SCREAMING_SNAKE_CASE	,						**__SCREAMING_SNAKE_CASE )






    def        UpperCamelCase	(	self					:							Optional[int]	,						__SCREAMING_SNAKE_CASE					:							Tuple	,						__SCREAMING_SNAKE_CASE					:							Optional[bool] = None	,						__SCREAMING_SNAKE_CASE					:							Optional[bool] = None	,						__SCREAMING_SNAKE_CASE					:							Optional[bool] = None	,						)					->		int:
      raise NotImplementedError('This method should be implemented by the derived class.' )






    def        UpperCamelCase	(	self					:							Union[str, Any] )					->		List[str]:
      lowerCamelCase_   					=						super().to_dict()
      lowerCamelCase_   					=						output.pop('_out_features' )
      lowerCamelCase_   					=						output.pop('_out_indices' )
      return output






class 					a						:






    @property
    def        UpperCamelCase	(	self					:							int )					->		Optional[Any]:
      return self._out_features






    @out_features.setter
    def        UpperCamelCase	(	self					:							List[str]	,						__SCREAMING_SNAKE_CASE					:							List[str] )					->		List[str]:
      lowerCamelCase_					,     lowerCamelCase_   					=						get_aligned_output_features_output_indices(
          out_features=__SCREAMING_SNAKE_CASE	,						out_indices=__SCREAMING_SNAKE_CASE	,						stage_names=self.stage_names )






    @property
    def        UpperCamelCase	(	self					:							Union[str, Any] )					->		Tuple:
      return self._out_indices






    @out_indices.setter
    def        UpperCamelCase	(	self					:							List[Any]	,						__SCREAMING_SNAKE_CASE					:							Union[Tuple[int], List[int]] )					->		str:
      lowerCamelCase_					,     lowerCamelCase_   					=						get_aligned_output_features_output_indices(
          out_features=__SCREAMING_SNAKE_CASE	,						out_indices=__SCREAMING_SNAKE_CASE	,						stage_names=self.stage_names )






    def        UpperCamelCase	(	self					:							int )					->		str:
      lowerCamelCase_   					=						super().to_dict()
      lowerCamelCase_   					=						output.pop('_out_features' )
      lowerCamelCase_   					=						output.pop('_out_indices' )
      return output




