import os
import tempfile
import unittest

import numpy as np
import torch

from diffusers.models import ModelMixin, UNetaDConditionModel
from diffusers.models.attention_processor import AttnProcessor, LoRAAttnProcessor
from diffusers.utils import (
    floats_tensor,
    logging,
    skip_mps,
    torch_device,
)
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


enable_full_determinism()

_SCREAMING_SNAKE_CASE     =     logging.get_logger(__name__)
def 					__lowercase						(    lowerCamelCase__			,					lowerCamelCase__ = True       )					->Any:

  '''simple docstring'''




  a__  		=  {}
  for name in model.attn_processors.keys():
    a__  		=  name.endswith("attn2.processor"       ) and not (
        name.startswith("transformer_in"       ) or "temp_attentions" in name.split("."       )
    )
    a__  		=  model.config.cross_attention_dim if has_cross_attention else None
    if name.startswith("mid_block"       ):
      a__  		=  model.config.block_out_channels[-1]
    elif name.startswith("up_blocks"       ):
      a__  		=  int(name[len("up_blocks."       )]       )
      a__  		=  list(reversed(model.config.block_out_channels       )       )[block_id]
    elif name.startswith("down_blocks"       ):
      a__  		=  int(name[len("down_blocks."       )]       )
      a__  		=  model.config.block_out_channels[block_id]
    elif name.startswith("transformer_in"       ):
      # Note that the `8 * ...` comes from: https://github.com/huggingface/diffusers/blob/7139f0e874f10b2463caa8cbd585762a309d12d6/src/diffusers/models/unet_3d_condition.py#L148
      a__  		=  8 * model.config.attention_head_dim

    a__  		=  LoRAAttnProcessor(hidden_size=lowerCamelCase__			,					cross_attention_dim=lowerCamelCase__       )
    a__  		=  lora_attn_procs[name].to(model.device       )

    if mock_weights:
      # add 1 to weights to mock trained weights
      with torch.no_grad():
        lora_attn_procs[name].to_q_lora.up.weight += 1
        lora_attn_procs[name].to_k_lora.up.weight += 1
        lora_attn_procs[name].to_v_lora.up.weight += 1
        lora_attn_procs[name].to_out_lora.up.weight += 1

  return lora_attn_procs







@skip_mps
class        UpperCAmelCase_					(				_a  ,  _a  ,  unittest.TestCase     ):
    UpperCAmelCase_      :							Union[str, Any]    			=				UNetaDConditionModel
    UpperCAmelCase_      :							str    			=				'sample'




    @property
    def     __lowerCamelCase			(       self	: Dict )						->			Tuple:
      a__  		=  4
      a__  		=  4
      a__  		=  4
      a__  		=  (32, 32)

      a__  		=  floats_tensor((batch_size, num_channels, num_frames) + sizes ).to(lowerCAmelCase )
      a__  		=  torch.tensor([10] ).to(lowerCAmelCase )
      a__  		=  floats_tensor((batch_size, 4, 32) ).to(lowerCAmelCase )

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




    @property
    def     __lowerCamelCase			(       self	: Tuple )						->			Optional[int]:
      return (4, 4, 32, 32)




    @property
    def     __lowerCamelCase			(       self	: Any )						->			Tuple:
      return (4, 4, 32, 32)




    def     __lowerCamelCase			(       self	: Dict )						->			Dict:
      a__  		=  {
          "block_out_channels": (32, 64),
          "down_block_types": (
              "CrossAttnDownBlock3D",
              "DownBlock3D",
          ),
          "up_block_types": ("UpBlock3D", "CrossAttnUpBlock3D"),
          "cross_attention_dim": 32,
          "attention_head_dim": 8,
          "out_channels": 4,
          "in_channels": 4,
          "layers_per_block": 1,
          "sample_size": 32,
      }
      a__  		=  self.dummy_input
      return init_dict, inputs_dict




    @unittest.skipIf(
        torch_device != "cuda" or not is_xformers_available()   ,   reason="XFormers attention is only available with CUDA and `xformers` installed"   ,   )
    def     __lowerCamelCase			(       self	: int )						->			Union[str, Any]:
      a__		,a__  		=  self.prepare_init_args_and_inputs_for_common()
      a__  		=  self.model_class(**lowerCAmelCase )

      model.enable_xformers_memory_efficient_attention()

      assert (
          model.mid_block.attentions[0].transformer_blocks[0].attna.processor.__class__.__name__
          == "XFormersAttnProcessor"
      ), "xformers is not enabled"




    def     __lowerCamelCase			(       self	: Optional[Any] )						->			Dict:
      a__		,a__  		=  self.prepare_init_args_and_inputs_for_common()

      a__  		=  32

      a__  		=  self.model_class(**lowerCAmelCase )
      model.to(lowerCAmelCase )
      model.eval()

      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase )

        if isinstance(lowerCAmelCase   ,   lowerCAmelCase ):
          a__  		=  output.sample

      self.assertIsNotNone(lowerCAmelCase )
      a__  		=  inputs_dict["sample"].shape
      self.assertEqual(output.shape   ,   lowerCAmelCase   ,   "Input and output shapes do not match" )




    def     __lowerCamelCase			(       self	: List[Any] )						->			str:
      a__		,a__  		=  self.prepare_init_args_and_inputs_for_common()
      a__  		=  self.model_class(**lowerCAmelCase )
      model.to(lowerCAmelCase )
      model.eval()

      with torch.no_grad():
        # Warmup pass when using mps (see #372)
        if torch_device == "mps" and isinstance(lowerCAmelCase   ,   lowerCAmelCase ):
          model(**self.dummy_input )

        a__  		=  model(**lowerCAmelCase )
        if isinstance(lowerCAmelCase   ,   lowerCAmelCase ):
          a__  		=  first.sample

        a__  		=  model(**lowerCAmelCase )
        if isinstance(lowerCAmelCase   ,   lowerCAmelCase ):
          a__  		=  second.sample

      a__  		=  first.cpu().numpy()
      a__  		=  second.cpu().numpy()
      a__  		=  out_a[~np.isnan(lowerCAmelCase )]
      a__  		=  out_a[~np.isnan(lowerCAmelCase )]
      a__  		=  np.amax(np.abs(out_a - out_a ) )
      self.assertLessEqual(lowerCAmelCase   ,   1e-5 )




    def     __lowerCamelCase			(       self	: int )						->			Union[str, Any]:
      a__		,a__  		=  self.prepare_init_args_and_inputs_for_common()

      a__  		=  8

      a__  		=  self.model_class(**lowerCAmelCase )
      model.to(lowerCAmelCase )
      model.eval()

      model.set_attention_slice("auto" )
      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase )
      assert output is not None

      model.set_attention_slice("max" )
      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase )
      assert output is not None

      model.set_attention_slice(2 )
      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase )
      assert output is not None




    def     __lowerCamelCase			(       self	: Tuple )						->			Dict:
      a__		,a__  		=  self.prepare_init_args_and_inputs_for_common()

      a__  		=  8

      a__  		=  self.model_class(**lowerCAmelCase )
      model.to(lowerCAmelCase )

      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase ).sample

      a__  		=  create_lora_layers(lowerCAmelCase )

      # make sure we can set a list of attention processors
      model.set_attn_processor(lowerCAmelCase )
      model.to(lowerCAmelCase )

      # test that attn processors can be set to itself
      model.set_attn_processor(model.attn_processors )

      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase   ,   cross_attention_kwargs={"scale": 0.0} ).sample
        a__  		=  model(**lowerCAmelCase   ,   cross_attention_kwargs={"scale": 0.5} ).sample
        a__  		=  model(**lowerCAmelCase   ,   cross_attention_kwargs={"scale": 0.5} ).sample

      assert (samplea - samplea).abs().max() < 3e-3
      assert (samplea - samplea).abs().max() < 3e-3

      # sample 2 and sample 3 should be different
      assert (samplea - samplea).abs().max() > 3e-3




    def     __lowerCamelCase			(       self	: List[Any] )						->			List[str]:
      a__		,a__  		=  self.prepare_init_args_and_inputs_for_common()

      a__  		=  8

      torch.manual_seed(0 )
      a__  		=  self.model_class(**lowerCAmelCase )
      model.to(lowerCAmelCase )

      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase ).sample

      a__  		=  create_lora_layers(lowerCAmelCase )
      model.set_attn_processor(lowerCAmelCase )

      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase   ,   cross_attention_kwargs={"scale": 0.5} ).sample

      with tempfile.TemporaryDirectory() as tmpdirname:
        model.save_attn_procs(lowerCAmelCase )
        self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase   ,   "pytorch_lora_weights.bin" ) ) )
        torch.manual_seed(0 )
        a__  		=  self.model_class(**lowerCAmelCase )
        new_model.to(lowerCAmelCase )
        new_model.load_attn_procs(lowerCAmelCase )

      with torch.no_grad():
        a__  		=  new_model(**lowerCAmelCase   ,   cross_attention_kwargs={"scale": 0.5} ).sample

      assert (sample - new_sample).abs().max() < 1e-3

      # LoRA and no LoRA should NOT be the same
      assert (sample - old_sample).abs().max() > 1e-4




    def     __lowerCamelCase			(       self	: int )						->			Optional[Any]:
      a__		,a__  		=  self.prepare_init_args_and_inputs_for_common()

      a__  		=  8

      torch.manual_seed(0 )
      a__  		=  self.model_class(**lowerCAmelCase )
      model.to(lowerCAmelCase )

      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase ).sample

      a__  		=  create_lora_layers(lowerCAmelCase )
      model.set_attn_processor(lowerCAmelCase )

      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase   ,   cross_attention_kwargs={"scale": 0.5} ).sample

      with tempfile.TemporaryDirectory() as tmpdirname:
        model.save_attn_procs(lowerCAmelCase   ,   safe_serialization=lowerCAmelCase )
        self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase   ,   "pytorch_lora_weights.safetensors" ) ) )
        torch.manual_seed(0 )
        a__  		=  self.model_class(**lowerCAmelCase )
        new_model.to(lowerCAmelCase )
        new_model.load_attn_procs(lowerCAmelCase )

      with torch.no_grad():
        a__  		=  new_model(**lowerCAmelCase   ,   cross_attention_kwargs={"scale": 0.5} ).sample

      assert (sample - new_sample).abs().max() < 3e-3

      # LoRA and no LoRA should NOT be the same
      assert (sample - old_sample).abs().max() > 1e-4




    def     __lowerCamelCase			(       self	: Tuple )						->			List[Any]:
      # enable deterministic behavior for gradient checkpointing
      a__		,a__  		=  self.prepare_init_args_and_inputs_for_common()

      a__  		=  8

      torch.manual_seed(0 )
      a__  		=  self.model_class(**lowerCAmelCase )
      model.to(lowerCAmelCase )

      a__  		=  create_lora_layers(lowerCAmelCase   ,   mock_weights=lowerCAmelCase )
      model.set_attn_processor(lowerCAmelCase )
      # Saving as torch, properly reloads with directly filename
      with tempfile.TemporaryDirectory() as tmpdirname:
        model.save_attn_procs(lowerCAmelCase )
        self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase   ,   "pytorch_lora_weights.bin" ) ) )
        torch.manual_seed(0 )
        a__  		=  self.model_class(**lowerCAmelCase )
        new_model.to(lowerCAmelCase )
        new_model.load_attn_procs(lowerCAmelCase   ,   weight_name="pytorch_lora_weights.bin" )




    def     __lowerCamelCase			(       self	: int )						->			str:
      a__		,a__  		=  self.prepare_init_args_and_inputs_for_common()

      a__  		=  8

      torch.manual_seed(0 )
      a__  		=  self.model_class(**lowerCAmelCase )
      model.to(lowerCAmelCase )

      a__  		=  create_lora_layers(lowerCAmelCase   ,   mock_weights=lowerCAmelCase )
      model.set_attn_processor(lowerCAmelCase )
      # Saving as torch, properly reloads with directly filename
      with tempfile.TemporaryDirectory() as tmpdirname:
        model.save_attn_procs(lowerCAmelCase )
        self.assertTrue(os.path.isfile(os.path.join(lowerCAmelCase   ,   "pytorch_lora_weights.bin" ) ) )
        torch.manual_seed(0 )
        a__  		=  self.model_class(**lowerCAmelCase )
        new_model.to(lowerCAmelCase )
        with self.assertRaises(lowerCAmelCase ) as e:
          new_model.load_attn_procs(lowerCAmelCase   ,   use_safetensors=lowerCAmelCase )
        self.assertIn("Error no file named pytorch_lora_weights.safetensors"   ,   str(e.exception ) )




    def     __lowerCamelCase			(       self	: List[str] )						->			List[Any]:
      a__		,a__  		=  self.prepare_init_args_and_inputs_for_common()

      a__  		=  8

      torch.manual_seed(0 )
      a__  		=  self.model_class(**lowerCAmelCase )
      model.to(lowerCAmelCase )

      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase ).sample

      a__  		=  create_lora_layers(lowerCAmelCase )
      model.set_attn_processor(lowerCAmelCase )

      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase   ,   cross_attention_kwargs={"scale": 0.0} ).sample

      model.set_attn_processor(AttnProcessor() )

      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase ).sample

      assert (sample - new_sample).abs().max() < 1e-4
      assert (sample - old_sample).abs().max() < 3e-3




    @unittest.skipIf(
        torch_device != "cuda" or not is_xformers_available()   ,   reason="XFormers attention is only available with CUDA and `xformers` installed"   ,   )
    def     __lowerCamelCase			(       self	: str )						->			Union[str, Any]:
      # enable deterministic behavior for gradient checkpointing
      a__		,a__  		=  self.prepare_init_args_and_inputs_for_common()

      a__  		=  4

      torch.manual_seed(0 )
      a__  		=  self.model_class(**lowerCAmelCase )
      model.to(lowerCAmelCase )
      a__  		=  create_lora_layers(lowerCAmelCase )
      model.set_attn_processor(lowerCAmelCase )

      # default
      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase ).sample

        model.enable_xformers_memory_efficient_attention()
        a__  		=  model(**lowerCAmelCase ).sample

        model.disable_xformers_memory_efficient_attention()
        a__  		=  model(**lowerCAmelCase ).sample

      assert (sample - on_sample).abs().max() < 1e-4
      assert (sample - off_sample).abs().max() < 1e-4




    def     __lowerCamelCase			(       self	: Any )						->			List[Any]:
      a__		,a__  		=  self.prepare_init_args_and_inputs_for_common()
      a__  		=  32

      a__  		=  self.model_class(**lowerCAmelCase )
      model.to(lowerCAmelCase )
      model.eval()

      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase )[0]

      model.enable_forward_chunking()
      with torch.no_grad():
        a__  		=  model(**lowerCAmelCase )[0]

      self.assertEqual(output.shape   ,   output_a.shape   ,   "Shape doesn't match" )
      assert np.abs(output.cpu() - output_a.cpu() ).max() < 1e-2


# (todo: sayakpaul) implement SLOW tests.


from collections.abc import Callable

_SCREAMING_SNAKE_CASE     =     Callable[[float], float]  # type alias for a real -> real function
def 					__lowercase						(    lowerCamelCase__			,					lowerCamelCase__			,					lowerCamelCase__			,					)					->float:

  '''simple docstring'''




  a__  		=  float(lowerCamelCase__       )
  while True:
    try:
      a__  		=  prev_guess - function(lowerCamelCase__       ) / derivative(lowerCamelCase__       )
    except ZeroDivisionError:
      raise ZeroDivisionError("Could not find root"       ) from None
    if abs(prev_guess - next_guess       ) < 10**-5:
      return next_guess
    a__  		=  next_guess
def 					__lowercase						(    lowerCamelCase__       )					->float:

  '''simple docstring'''




  return (x**3) - (2 * x) - 5
def 					__lowercase						(    lowerCamelCase__       )					->float:

  '''simple docstring'''




  return 3 * (x**2) - 2


if __name__ == "__main__":
     print(newton(f, fa, 3))

