
"""simple docstring"""




import os
import tempfile
import unittest

import numpy as np

from diffusers.utils import is_flax_available
from diffusers.utils.testing_utils import require_flax, slow


if is_flax_available():
  import jax
  import jax.numpy as jnp
  from flax.jax_utils import replicate
  from flax.training.common_utils import shard

  from diffusers import FlaxDDIMScheduler, FlaxDiffusionPipeline, FlaxStableDiffusionPipeline







@require_flax
class   _lowercase						(       unittest.TestCase				):


     '''simple docstring'''





     def  lowerCAmelCase__       (      self  )->					int:






          with tempfile.TemporaryDirectory() as tmpdirname:
               # pipeline has Flax weights
               UpperCAmelCase__							: List[str] 							=						FlaxDiffusionPipeline.from_pretrained(
                   "hf-internal-testing/tiny-stable-diffusion-pipe" ,			safety_checker=lowerCamelCase__ ,			cache_dir=lowerCamelCase__  )

               UpperCAmelCase__							: List[str] 							=						[t[-1] for t in os.walk(os.path.join(lowerCamelCase__ ,			os.listdir(lowerCamelCase__  )[0] ,			"snapshots"  )  )]
               UpperCAmelCase__							: Optional[int] 							=						[item for sublist in all_root_files for item in sublist]

               # None of the downloaded files should be a PyTorch file even if we have some here:
               # https://huggingface.co/hf-internal-testing/tiny-stable-diffusion-pipe/blob/main/unet/diffusion_pytorch_model.bin
               assert not any(f.endswith(".bin"  ) for f in files  )







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


     '''simple docstring'''





     def  lowerCAmelCase__       (      self  )->					Optional[int]:
          UpperCAmelCase__							: Optional[Any] 							=						FlaxStableDiffusionPipeline.from_pretrained(
              "hf-internal-testing/tiny-stable-diffusion-pipe" ,			safety_checker=lowerCamelCase__  )

          UpperCAmelCase__							: List[Any] 							=						(
              '''A cinematic film still of Morgan Freeman starring as Jimi Hendrix, portrait, 40mm lens, shallow depth of'''
              ''' field, close up, split lighting, cinematic'''
          )

          UpperCAmelCase__							: str 							=						jax.random.PRNGKey(0  )
          UpperCAmelCase__							: List[str] 							=						4

          UpperCAmelCase__							: Tuple 							=						jax.device_count()
          UpperCAmelCase__							: Union[str, Any] 							=						num_samples * [prompt]
          UpperCAmelCase__							: Dict 							=						pipeline.prepare_inputs(lowerCamelCase__  )

          # shard inputs and rng
          UpperCAmelCase__							: int 							=						replicate(lowerCamelCase__  )
          UpperCAmelCase__							: Optional[int] 							=						jax.random.split(lowerCamelCase__ ,			lowerCamelCase__  )
          UpperCAmelCase__							: Any 							=						shard(lowerCamelCase__  )

          UpperCAmelCase__							: Any 							=						pipeline(lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__ ,			jit=lowerCamelCase__  ).images

          assert images.shape == (num_samples, 1, 64, 64, 3)
          if jax.device_count() == 8:
               assert np.abs(np.abs(images[0, 0, :2, :2, -2:] ,			dtype=np.floataa  ).sum() - 4.151_4745  ) < 1E-3
               assert np.abs(np.abs(lowerCamelCase__ ,			dtype=np.floataa  ).sum() - 4_99_47.8_75  ) < 5E-1

          UpperCAmelCase__							: str 							=						pipeline.numpy_to_pil(np.asarray(images.reshape((num_samples,) + images.shape[-3:]  )  )  )
          assert len(lowerCamelCase__  ) == num_samples



     def  lowerCAmelCase__       (      self  )->					Tuple:
          UpperCAmelCase__							: Optional[int] 							=						FlaxStableDiffusionPipeline.from_pretrained(
              "CompVis/stable-diffusion-v1-4" ,			revision="flax" ,			safety_checker=lowerCamelCase__  )

          UpperCAmelCase__							: str 							=						(
              '''A cinematic film still of Morgan Freeman starring as Jimi Hendrix, portrait, 40mm lens, shallow depth of'''
              ''' field, close up, split lighting, cinematic'''
          )

          UpperCAmelCase__							: str 							=						jax.random.PRNGKey(0  )
          UpperCAmelCase__							: Any 							=						50

          UpperCAmelCase__							: Optional[int] 							=						jax.device_count()
          UpperCAmelCase__							: Any 							=						num_samples * [prompt]
          UpperCAmelCase__							: str 							=						pipeline.prepare_inputs(lowerCamelCase__  )

          # shard inputs and rng
          UpperCAmelCase__							: int 							=						replicate(lowerCamelCase__  )
          UpperCAmelCase__							: List[Any] 							=						jax.random.split(lowerCamelCase__ ,			lowerCamelCase__  )
          UpperCAmelCase__							: Any 							=						shard(lowerCamelCase__  )

          UpperCAmelCase__							: List[str] 							=						pipeline(lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__ ,			jit=lowerCamelCase__  ).images

          assert images.shape == (num_samples, 1, 5_12, 5_12, 3)
          if jax.device_count() == 8:
               assert np.abs((np.abs(images[0, 0, :2, :2, -2:] ,			dtype=np.floataa  ).sum() - 0.0565_2401)  ) < 1E-3
               assert np.abs((np.abs(lowerCamelCase__ ,			dtype=np.floataa  ).sum() - 2_38_38_08.2)  ) < 5E-1



     def  lowerCAmelCase__       (      self  )->					Union[str, Any]:
          UpperCAmelCase__							: Optional[int] 							=						FlaxStableDiffusionPipeline.from_pretrained(
              "CompVis/stable-diffusion-v1-4" ,			revision="bf16" ,			dtype=jnp.bfloataa ,			safety_checker=lowerCamelCase__  )

          UpperCAmelCase__							: int 							=						(
              '''A cinematic film still of Morgan Freeman starring as Jimi Hendrix, portrait, 40mm lens, shallow depth of'''
              ''' field, close up, split lighting, cinematic'''
          )

          UpperCAmelCase__							: Dict 							=						jax.random.PRNGKey(0  )
          UpperCAmelCase__							: Union[str, Any] 							=						50

          UpperCAmelCase__							: List[str] 							=						jax.device_count()
          UpperCAmelCase__							: List[str] 							=						num_samples * [prompt]
          UpperCAmelCase__							: List[Any] 							=						pipeline.prepare_inputs(lowerCamelCase__  )

          # shard inputs and rng
          UpperCAmelCase__							: str 							=						replicate(lowerCamelCase__  )
          UpperCAmelCase__							: List[Any] 							=						jax.random.split(lowerCamelCase__ ,			lowerCamelCase__  )
          UpperCAmelCase__							: Optional[int] 							=						shard(lowerCamelCase__  )

          UpperCAmelCase__							: Any 							=						pipeline(lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__ ,			jit=lowerCamelCase__  ).images

          assert images.shape == (num_samples, 1, 5_12, 5_12, 3)
          if jax.device_count() == 8:
               assert np.abs((np.abs(images[0, 0, :2, :2, -2:] ,			dtype=np.floataa  ).sum() - 0.0400_3906)  ) < 1E-3
               assert np.abs((np.abs(lowerCamelCase__ ,			dtype=np.floataa  ).sum() - 2_37_35_16.75)  ) < 5E-1



     def  lowerCAmelCase__       (      self  )->					int:
          UpperCAmelCase__							: List[str] 							=						FlaxStableDiffusionPipeline.from_pretrained(
              "CompVis/stable-diffusion-v1-4" ,			revision="bf16" ,			dtype=jnp.bfloataa  )

          UpperCAmelCase__							: Tuple 							=						(
              '''A cinematic film still of Morgan Freeman starring as Jimi Hendrix, portrait, 40mm lens, shallow depth of'''
              ''' field, close up, split lighting, cinematic'''
          )

          UpperCAmelCase__							: List[Any] 							=						jax.random.PRNGKey(0  )
          UpperCAmelCase__							: List[str] 							=						50

          UpperCAmelCase__							: Optional[Any] 							=						jax.device_count()
          UpperCAmelCase__							: Optional[int] 							=						num_samples * [prompt]
          UpperCAmelCase__							: Dict 							=						pipeline.prepare_inputs(lowerCamelCase__  )

          # shard inputs and rng
          UpperCAmelCase__							: Optional[int] 							=						replicate(lowerCamelCase__  )
          UpperCAmelCase__							: str 							=						jax.random.split(lowerCamelCase__ ,			lowerCamelCase__  )
          UpperCAmelCase__							: Union[str, Any] 							=						shard(lowerCamelCase__  )

          UpperCAmelCase__							: Union[str, Any] 							=						pipeline(lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__ ,			jit=lowerCamelCase__  ).images

          assert images.shape == (num_samples, 1, 5_12, 5_12, 3)
          if jax.device_count() == 8:
               assert np.abs((np.abs(images[0, 0, :2, :2, -2:] ,			dtype=np.floataa  ).sum() - 0.0400_3906)  ) < 1E-3
               assert np.abs((np.abs(lowerCamelCase__ ,			dtype=np.floataa  ).sum() - 2_37_35_16.75)  ) < 5E-1



     def  lowerCAmelCase__       (      self  )->					List[Any]:
          UpperCAmelCase__							: List[str] 							=						FlaxDDIMScheduler(
              beta_start=0.0_0085 ,			beta_end=0.012 ,			beta_schedule="scaled_linear" ,			set_alpha_to_one=lowerCamelCase__ ,			steps_offset=1 ,			)

          UpperCAmelCase__							: Tuple 							=						FlaxStableDiffusionPipeline.from_pretrained(
              "CompVis/stable-diffusion-v1-4" ,			revision="bf16" ,			dtype=jnp.bfloataa ,			scheduler=lowerCamelCase__ ,			safety_checker=lowerCamelCase__ ,			)
          UpperCAmelCase__							: List[Any] 							=						scheduler.create_state()

          UpperCAmelCase__							: Union[str, Any] 							=						scheduler_state

          UpperCAmelCase__							: List[str] 							=						(
              '''A cinematic film still of Morgan Freeman starring as Jimi Hendrix, portrait, 40mm lens, shallow depth of'''
              ''' field, close up, split lighting, cinematic'''
          )

          UpperCAmelCase__							: Tuple 							=						jax.random.PRNGKey(0  )
          UpperCAmelCase__							: Optional[int] 							=						50

          UpperCAmelCase__							: int 							=						jax.device_count()
          UpperCAmelCase__							: Dict 							=						num_samples * [prompt]
          UpperCAmelCase__							: str 							=						pipeline.prepare_inputs(lowerCamelCase__  )

          # shard inputs and rng
          UpperCAmelCase__							: List[Any] 							=						replicate(lowerCamelCase__  )
          UpperCAmelCase__							: Optional[Any] 							=						jax.random.split(lowerCamelCase__ ,			lowerCamelCase__  )
          UpperCAmelCase__							: Union[str, Any] 							=						shard(lowerCamelCase__  )

          UpperCAmelCase__							: Dict 							=						pipeline(lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__ ,			jit=lowerCamelCase__  ).images

          assert images.shape == (num_samples, 1, 5_12, 5_12, 3)
          if jax.device_count() == 8:
               assert np.abs((np.abs(images[0, 0, :2, :2, -2:] ,			dtype=np.floataa  ).sum() - 0.0_4504_3945)  ) < 1E-3
               assert np.abs((np.abs(lowerCamelCase__ ,			dtype=np.floataa  ).sum() - 2_34_76_93.5)  ) < 5E-1



     def  lowerCAmelCase__       (      self  )->					Union[str, Any]:
          UpperCAmelCase__							: List[Any] 							=						(
              '''A cinematic film still of Morgan Freeman starring as Jimi Hendrix, portrait, 40mm lens, shallow depth of'''
              ''' field, close up, split lighting, cinematic'''
          )

          UpperCAmelCase__							: str 							=						jax.device_count()
          UpperCAmelCase__							: Union[str, Any] 							=						num_samples * [prompt]
          UpperCAmelCase__							: str 							=						jax.random.split(jax.random.PRNGKey(0  ) ,			lowerCamelCase__  )

          UpperCAmelCase__							: Dict 							=						FlaxStableDiffusionPipeline.from_pretrained(
              "CompVis/stable-diffusion-v1-4" ,			revision="bf16" ,			dtype=jnp.bfloataa ,			safety_checker=lowerCamelCase__ ,			)

          UpperCAmelCase__							: List[str] 							=						replicate(lowerCamelCase__  )
          UpperCAmelCase__							: Tuple 							=						pipeline.prepare_inputs(lowerCamelCase__  )
          UpperCAmelCase__							: int 							=						shard(lowerCamelCase__  )
          UpperCAmelCase__							: List[Any] 							=						pipeline(lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__ ,			jit=lowerCamelCase__  ).images
          assert images.shape == (num_samples, 1, 5_12, 5_12, 3)
          UpperCAmelCase__							: Union[str, Any] 							=						images[2, 0, 2_56, 10:17, 1]

          # With memory efficient attention
          UpperCAmelCase__							: Tuple 							=						FlaxStableDiffusionPipeline.from_pretrained(
              "CompVis/stable-diffusion-v1-4" ,			revision="bf16" ,			dtype=jnp.bfloataa ,			safety_checker=lowerCamelCase__ ,			use_memory_efficient_attention=lowerCamelCase__ ,			)

          UpperCAmelCase__							: Any 							=						replicate(lowerCamelCase__  )
          UpperCAmelCase__							: List[str] 							=						pipeline.prepare_inputs(lowerCamelCase__  )
          UpperCAmelCase__							: int 							=						shard(lowerCamelCase__  )
          UpperCAmelCase__							: Dict 							=						pipeline(lowerCamelCase__ ,			lowerCamelCase__ ,			lowerCamelCase__ ,			jit=lowerCamelCase__  ).images
          assert images_eff.shape == (num_samples, 1, 5_12, 5_12, 3)
          UpperCAmelCase__							: Union[str, Any] 							=						images[2, 0, 2_56, 10:17, 1]

          # I checked the results visually and they are very similar. However, I saw that the max diff is `1` and the `sum`
          # over the 8 images is exactly `256`, which is very suspicious. Testing a random slice for now.
          assert abs(slice_eff - slice  ).max() < 1E-2





"""simple docstring"""




import warnings
from pathlib import Path
from typing import List, Tuple, Union

import fire
from torch import nn

from transformers import AutoModelForSeqaSeqLM, AutoTokenizer, PreTrainedModel
from transformers.utils import logging


A__      :	Tuple          =    logging.get_logger(__name__)



def 							a__ (				lowerCAmelCase	:       nn.ModuleList       ,    lowerCAmelCase	:       nn.ModuleList       ,    lowerCAmelCase	:       List[int]		):






     '''simple docstring'''
     UpperCAmelCase__							: Optional[Any] 							=						nn.ModuleList([src_layers[i] for i in layers_to_copy]		)
     assert len(lowerCAmelCase		) == len(lowerCAmelCase		), F"{len(lowerCAmelCase		)} != {len(lowerCAmelCase		)}"
     dest_layers.load_state_dict(layers_to_copy.state_dict()		)


A__      :	List[Any]          =    {
    # maps  num layers in teacher -> num_layers in student -> which teacher layers to copy.
    # 12: bart, 16: pegasus, 6: marian/Helsinki-NLP
    12: {
        1: [0],  # This says that if the teacher has 12 layers and the student has 1, copy layer 0 of the teacher
        2: [0, 6],
        3: [0, 6, 11],
        4: [0, 4, 8, 11],
        6: [0, 2, 4, 7, 9, 11],
        9: [0, 1, 2, 4, 5, 7, 9, 10, 11],
        12: list(range(12)),
    },
    16: {  # maps  num layers in student -> which teacher layers to copy
        1: [0],
        2: [0, 15],
        3: [0, 8, 15],
        4: [0, 5, 10, 15],
        6: [0, 3, 6, 9, 12, 15],
        8: [0, 2, 4, 6, 8, 10, 12, 15],
        9: [0, 1, 3, 5, 7, 9, 11, 13, 15],
        12: [0, 1, 2, 3, 4, 5, 6, 7, 9, 11, 13, 15],
        16: list(range(16)),
    },
    6: {1: [0], 2: [0, 5], 3: [0, 2, 5], 4: [0, 1, 3, 5], 6: list(range(6))},
}
A__      :	List[Any]          =    {
    # maps  num layers in student -> which teacher layers to copy.
    6: {1: [5], 2: [3, 5], 3: [1, 4, 5], 4: [1, 2, 4, 5]},
    12: {1: [11], 2: [5, 11], 3: [3, 7, 11], 6: [1, 3, 5, 8, 10, 11]},
    16: {1: [15], 4: [4, 9, 12, 15], 8: [1, 3, 5, 7, 9, 11, 13, 15]},
}



def 							a__ (				lowerCAmelCase	:       Dict       ,    lowerCAmelCase	:       Dict		):






     '''simple docstring'''
     try:
          UpperCAmelCase__							: Tuple 							=						LAYERS_TO_COPY[n_teacher][n_student]
          return val
     except KeyError:
          if n_student != n_teacher:
               warnings.warn(
                   F"no hardcoded layers to copy for teacher {n_teacher} -> student {n_student}, defaulting to first"
                   F" {n_student}"		)
          return list(range(lowerCAmelCase		)		)



def 							a__ (				lowerCAmelCase	:       int       ,    lowerCAmelCase	:       Tuple		):






     '''simple docstring'''
     if n_student > n_teacher:
          raise ValueError(F"Cannot perform intermediate supervision for student {n_student} > teacher {n_teacher}"		)
     elif n_teacher == n_student:
          return list(range(lowerCAmelCase		)		)
     elif n_student == 1:
          return [n_teacher - 1]
     else:
          return LAYERS_TO_SUPERVISE[n_teacher][n_student]



def 							a__ (				lowerCAmelCase	:       Union[str, PreTrainedModel]       ,    lowerCAmelCase	:       Union[str, Path] = "student"       ,    lowerCAmelCase	:       Union[int, None] = None       ,    lowerCAmelCase	:       Union[int, None] = None       ,    lowerCAmelCase	:       List[str]=False       ,    lowerCAmelCase	:       List[str]=None       ,    lowerCAmelCase	:       List[str]=None       ,    **lowerCAmelCase	:       List[str]       ,    ):






     '''simple docstring'''
     UpperCAmelCase__							: List[str] 							=						"encoder_layers and decoder_layers cannot be both None-- you would just have an identical teacher."
     assert (e is not None) or (d is not None), _msg
     if isinstance(lowerCAmelCase       ,    lowerCAmelCase		):
          AutoTokenizer.from_pretrained(lowerCAmelCase		).save_pretrained(lowerCAmelCase		)  # purely for convenience
          UpperCAmelCase__							: List[str] 							=						AutoModelForSeqaSeqLM.from_pretrained(lowerCAmelCase		).eval()
     else:
          assert isinstance(lowerCAmelCase       ,    lowerCAmelCase		), F"teacher must be a model or string got type {type(lowerCAmelCase		)}"
     UpperCAmelCase__							: int 							=						teacher.config.to_diff_dict()

     try:
          UpperCAmelCase__							,	UpperCAmelCase__							: Optional[Any] 							=						teacher.config.encoder_layers, teacher.config.decoder_layers
          if e is None:
               UpperCAmelCase__							: Tuple 							=						teacher_e
          if d is None:
               UpperCAmelCase__							: str 							=						teacher_d
          init_kwargs.update({"encoder_layers": e, "decoder_layers": d}		)
     except AttributeError:  # T5
          if hasattr(teacher.config       ,    "num_encoder_layers"		):
               UpperCAmelCase__							,	UpperCAmelCase__							: Union[str, Any] 							=						teacher.config.num_encoder_layers, teacher.config.num_decoder_layers
          else:
               UpperCAmelCase__							,	UpperCAmelCase__							: Union[str, Any] 							=						teacher.config.num_layers, teacher.config.num_decoder_layers
          if e is None:
               UpperCAmelCase__							: Optional[Any] 							=						teacher_e
          if d is None:
               UpperCAmelCase__							: Optional[Any] 							=						teacher_d
          if hasattr(teacher.config       ,    "num_encoder_layers"		):
               init_kwargs.update({"num_encoder_layers": e, "num_decoder_layers": d}		)
          else:
               init_kwargs.update({"num_layers": e, "num_decoder_layers": d}		)

    # Kwargs to instantiate student: teacher kwargs with updated layer numbers + **extra_config_kwargs
     init_kwargs.update(lowerCAmelCase		)

     # Copy weights
     UpperCAmelCase__							: Tuple 							=						teacher.config_class(**lowerCAmelCase		)
     UpperCAmelCase__							: List[str] 							=						AutoModelForSeqaSeqLM.from_config(lowerCAmelCase		)
     # Start by copying the full teacher state dict this will copy the first N teacher layers to the student.
     UpperCAmelCase__							: Optional[int] 							=						student.load_state_dict(teacher.state_dict()       ,    strict=lowerCAmelCase		)
     assert info.missing_keys == [], info.missing_keys  # every student key should have a teacher keys.

     if copy_first_teacher_layers:  # Our copying is done. We just log and save
          UpperCAmelCase__							,	UpperCAmelCase__							: int 							=						list(range(lowerCAmelCase		)		), list(range(lowerCAmelCase		)		)
          logger.info(
              F"Copied encoder layers {e_layers_to_copy} and decoder layers {d_layers_to_copy}. Saving them to"
              F" {save_path}"		)
          student.save_pretrained(lowerCAmelCase		)
          return student, e_layers_to_copy, d_layers_to_copy

     # Decide which layers of the teacher to copy. Not exactly alternating -- we try to keep first and last layer.
     if e_layers_to_copy is None:
          UpperCAmelCase__							: List[int] 							=						pick_layers_to_copy(lowerCAmelCase       ,    lowerCAmelCase		)
     if d_layers_to_copy is None:
          UpperCAmelCase__							: List[int] 							=						pick_layers_to_copy(lowerCAmelCase       ,    lowerCAmelCase		)

     try:
          if hasattr(
              lowerCAmelCase       ,    "prophetnet"		):  # For ProphetNet, student.model.encoder.layers is called student.prophetnet.encoder.layers
               copy_layers(teacher.prophetnet.encoder.layers       ,    student.prophetnet.encoder.layers       ,    lowerCAmelCase		)
               copy_layers(teacher.prophetnet.decoder.layers       ,    student.prophetnet.decoder.layers       ,    lowerCAmelCase		)
          else:
               copy_layers(teacher.model.encoder.layers       ,    student.model.encoder.layers       ,    lowerCAmelCase		)
               copy_layers(teacher.model.decoder.layers       ,    student.model.decoder.layers       ,    lowerCAmelCase		)
     except AttributeError:  # For t5, student.model.encoder.layers is called student.encoder.block
          copy_layers(teacher.encoder.block       ,    student.encoder.block       ,    lowerCAmelCase		)
          copy_layers(teacher.decoder.block       ,    student.decoder.block       ,    lowerCAmelCase		)
     logger.info(
         F"Copied encoder layers {e_layers_to_copy} and decoder layers {d_layers_to_copy}. Saving them to {save_path}"		)
     UpperCAmelCase__							: int 							=						{
         "teacher_type": teacher.config.model_type,
         "copied_encoder_layers": e_layers_to_copy,
         "copied_decoder_layers": d_layers_to_copy,
     }
     student.save_pretrained(lowerCAmelCase		)
     # Save information about copying for easier reproducibility

     return student, e_layers_to_copy, d_layers_to_copy


if __name__ == "__main__":
    fire.Fire(create_student_by_copying_alternating_layers)



