



"""simple docstring"""




from typing import Any, Callable, Dict, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker


_lowerCAmelCase    : Union[str, Any]							=							"CompVis/stable-diffusion-v1-1"
_lowerCAmelCase    : int							=							"CompVis/stable-diffusion-v1-2"
_lowerCAmelCase    : List[Any]							=							"CompVis/stable-diffusion-v1-3"
_lowerCAmelCase    : Any							=							"CompVis/stable-diffusion-v1-4"






class      UpperCAmelCase_  (    _UpperCamelCase						):



      def __init__(					self					:     Union[str, Any]					,     A					:     AutoencoderKL					,     A					:     CLIPTextModel					,     A					:     CLIPTokenizer					,     A					:     UNetaDConditionModel					,     A					:     Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]					,     A					:     StableDiffusionSafetyChecker					,     A					:     CLIPImageProcessor					,     A					:     bool = True					,     ):
           super()._init_()

           _UpperCAmelCase      :     Union[str, Any]        =   StableDiffusionPipeline.from_pretrained(A						)
           _UpperCAmelCase      :     Union[str, Any]        =   StableDiffusionPipeline.from_pretrained(A						)
           _UpperCAmelCase      :     List[str]        =   StableDiffusionPipeline.from_pretrained(A						)
           _UpperCAmelCase      :     Optional[int]        =   StableDiffusionPipeline(
               vae=A					,     text_encoder=A					,     tokenizer=A					,     unet=A					,     scheduler=A					,     safety_checker=A					,     feature_extractor=A					,     requires_safety_checker=A					,     )

           self.register_modules(pipelinea=self.pipea					,     pipelinea=self.pipea					,     pipelinea=self.pipea					,     pipelinea=self.pipea						)



      @property
      def 						snake_case_ (					self					:     Any						):
           return {k: getattr(self					,     A						) for k in self.config.keys() if not k.startswith("_"						)}



      def 						snake_case_ (					self					:     Tuple					,     A					:     Optional[Union[str, int]] = "auto"						):
           if slice_size == "auto":
                # half the attention head size is usually a good trade-off between
                # speed and memory
                _UpperCAmelCase      :     str        =   self.unet.config.attention_head_dim // 2
           self.unet.set_attention_slice(A						)



      def 						snake_case_ (					self					:     int						):
           self.enable_attention_slicing(A						)



      @torch.no_grad()
      def 						snake_case_ (					self					:     Optional[Any]					,     A					:     Union[str, List[str]]					,     A					:     int = 5_1_2					,     A					:     int = 5_1_2					,     A					:     int = 5_0					,     A					:     float = 7.5					,     A					:     Optional[Union[str, List[str]]] = None					,     A					:     Optional[int] = 1					,     A					:     float = 0.0					,     A					:     Optional[torch.Generator] = None					,     A					:     Optional[torch.FloatTensor] = None					,     A					:     Optional[str] = "pil"					,     A					:     bool = True					,     A					:     Optional[Callable[[int, int, torch.FloatTensor], None]] = None					,     A					:     int = 1					,     **A					:     Any					,     ):
           return self.pipea(
               prompt=A					,     height=A					,     width=A					,     num_inference_steps=A					,     guidance_scale=A					,     negative_prompt=A					,     num_images_per_prompt=A					,     eta=A					,     generator=A					,     latents=A					,     output_type=A					,     return_dict=A					,     callback=A					,     callback_steps=A					,     **A					,     )



      @torch.no_grad()
      def 						snake_case_ (					self					:     str					,     A					:     Union[str, List[str]]					,     A					:     int = 5_1_2					,     A					:     int = 5_1_2					,     A					:     int = 5_0					,     A					:     float = 7.5					,     A					:     Optional[Union[str, List[str]]] = None					,     A					:     Optional[int] = 1					,     A					:     float = 0.0					,     A					:     Optional[torch.Generator] = None					,     A					:     Optional[torch.FloatTensor] = None					,     A					:     Optional[str] = "pil"					,     A					:     bool = True					,     A					:     Optional[Callable[[int, int, torch.FloatTensor], None]] = None					,     A					:     int = 1					,     **A					:     List[Any]					,     ):
           return self.pipea(
               prompt=A					,     height=A					,     width=A					,     num_inference_steps=A					,     guidance_scale=A					,     negative_prompt=A					,     num_images_per_prompt=A					,     eta=A					,     generator=A					,     latents=A					,     output_type=A					,     return_dict=A					,     callback=A					,     callback_steps=A					,     **A					,     )



      @torch.no_grad()
      def 						snake_case_ (					self					:     Any					,     A					:     Union[str, List[str]]					,     A					:     int = 5_1_2					,     A					:     int = 5_1_2					,     A					:     int = 5_0					,     A					:     float = 7.5					,     A					:     Optional[Union[str, List[str]]] = None					,     A					:     Optional[int] = 1					,     A					:     float = 0.0					,     A					:     Optional[torch.Generator] = None					,     A					:     Optional[torch.FloatTensor] = None					,     A					:     Optional[str] = "pil"					,     A					:     bool = True					,     A					:     Optional[Callable[[int, int, torch.FloatTensor], None]] = None					,     A					:     int = 1					,     **A					:     Tuple					,     ):
           return self.pipea(
               prompt=A					,     height=A					,     width=A					,     num_inference_steps=A					,     guidance_scale=A					,     negative_prompt=A					,     num_images_per_prompt=A					,     eta=A					,     generator=A					,     latents=A					,     output_type=A					,     return_dict=A					,     callback=A					,     callback_steps=A					,     **A					,     )



      @torch.no_grad()
      def 						snake_case_ (					self					:     Any					,     A					:     Union[str, List[str]]					,     A					:     int = 5_1_2					,     A					:     int = 5_1_2					,     A					:     int = 5_0					,     A					:     float = 7.5					,     A					:     Optional[Union[str, List[str]]] = None					,     A					:     Optional[int] = 1					,     A					:     float = 0.0					,     A					:     Optional[torch.Generator] = None					,     A					:     Optional[torch.FloatTensor] = None					,     A					:     Optional[str] = "pil"					,     A					:     bool = True					,     A					:     Optional[Callable[[int, int, torch.FloatTensor], None]] = None					,     A					:     int = 1					,     **A					:     int					,     ):
           return self.pipea(
               prompt=A					,     height=A					,     width=A					,     num_inference_steps=A					,     guidance_scale=A					,     negative_prompt=A					,     num_images_per_prompt=A					,     eta=A					,     generator=A					,     latents=A					,     output_type=A					,     return_dict=A					,     callback=A					,     callback_steps=A					,     **A					,     )



      @torch.no_grad()
      def 						snake_case_ (					self					:     Optional[int]					,     A					:     Union[str, List[str]]					,     A					:     int = 5_1_2					,     A					:     int = 5_1_2					,     A					:     int = 5_0					,     A					:     float = 7.5					,     A					:     Optional[Union[str, List[str]]] = None					,     A					:     Optional[int] = 1					,     A					:     float = 0.0					,     A					:     Optional[torch.Generator] = None					,     A					:     Optional[torch.FloatTensor] = None					,     A					:     Optional[str] = "pil"					,     A					:     bool = True					,     A					:     Optional[Callable[[int, int, torch.FloatTensor], None]] = None					,     A					:     int = 1					,     **A					:     str					,     ):
           _UpperCAmelCase      :     List[str]        =   "cuda" if torch.cuda.is_available() else "cpu"
           self.to(A						)

           # Checks if the height and width are divisible by 8 or not
           if height % 8 != 0 or width % 8 != 0:
                raise ValueError(f'`height` and `width` must be divisible by 8 but are {height} and {width}.'						)

           # Get first result from Stable Diffusion Checkpoint v1.1
           _UpperCAmelCase      :     Any        =   self.textaimg_sda_a(
               prompt=A					,     height=A					,     width=A					,     num_inference_steps=A					,     guidance_scale=A					,     negative_prompt=A					,     num_images_per_prompt=A					,     eta=A					,     generator=A					,     latents=A					,     output_type=A					,     return_dict=A					,     callback=A					,     callback_steps=A					,     **A					,     )

           # Get first result from Stable Diffusion Checkpoint v1.2
           _UpperCAmelCase      :     Tuple        =   self.textaimg_sda_a(
               prompt=A					,     height=A					,     width=A					,     num_inference_steps=A					,     guidance_scale=A					,     negative_prompt=A					,     num_images_per_prompt=A					,     eta=A					,     generator=A					,     latents=A					,     output_type=A					,     return_dict=A					,     callback=A					,     callback_steps=A					,     **A					,     )

           # Get first result from Stable Diffusion Checkpoint v1.3
           _UpperCAmelCase      :     Dict        =   self.textaimg_sda_a(
               prompt=A					,     height=A					,     width=A					,     num_inference_steps=A					,     guidance_scale=A					,     negative_prompt=A					,     num_images_per_prompt=A					,     eta=A					,     generator=A					,     latents=A					,     output_type=A					,     return_dict=A					,     callback=A					,     callback_steps=A					,     **A					,     )

           # Get first result from Stable Diffusion Checkpoint v1.4
           _UpperCAmelCase      :     Dict        =   self.textaimg_sda_a(
               prompt=A					,     height=A					,     width=A					,     num_inference_steps=A					,     guidance_scale=A					,     negative_prompt=A					,     num_images_per_prompt=A					,     eta=A					,     generator=A					,     latents=A					,     output_type=A					,     return_dict=A					,     callback=A					,     callback_steps=A					,     **A					,     )

           # Get all result images into a single list and pass it via StableDiffusionPipelineOutput for final result
           return StableDiffusionPipelineOutput([resa[0], resa[0], resa[0], resa[0]]						)









"""simple docstring"""




import io
import json
import unittest

from parameterized import parameterized

from transformers import FSMTForConditionalGeneration, FSMTTokenizer
from transformers.testing_utils import get_tests_dir, require_torch, slow, torch_device
from utils import calculate_bleu


_lowerCAmelCase    : Union[str, Any]							=							get_tests_dir() + "/test_data/fsmt/fsmt_val_data.json"
with io.open(filename, "r", encoding="utf-8") as f:
 _lowerCAmelCase    : Tuple							=							json.load(f)






@require_torch
class      UpperCAmelCase_  (    unittest.TestCase						):



      def 						snake_case_ (					self					:     str					,     A					:     Union[str, Any]						):
           return FSMTTokenizer.from_pretrained(A						)



      def 						snake_case_ (					self					:     Union[str, Any]					,     A					:     Union[str, Any]						):
           _UpperCAmelCase      :     List[Any]        =   FSMTForConditionalGeneration.from_pretrained(A						).to(A						)
           if torch_device == "cuda":
                model.half()
           return model



      @parameterized.expand(
          [
              ["en-ru", 26.0],
              ["ru-en", 22.0],
              ["en-de", 22.0],
              ["de-en", 29.0],
          ]						)
      @slow
      def 						snake_case_ (					self					:     Any					,     A					:     Dict					,     A					:     List[str]						):
           # note: this test is not testing the best performance since it only evals a small batch
           # but it should be enough to detect a regression in the output quality
           _UpperCAmelCase      :     Any        =   f'facebook/wmt19-{pair}'
           _UpperCAmelCase      :     Dict        =   self.get_tokenizer(A						)
           _UpperCAmelCase      :     Optional[int]        =   self.get_model(A						)

           _UpperCAmelCase      :     int        =   bleu_data[pair]["src"]
           _UpperCAmelCase      :     Optional[int]        =   bleu_data[pair]["tgt"]

           _UpperCAmelCase      :     List[str]        =   tokenizer(A					,     return_tensors="pt"					,     truncation=A					,     padding="longest"						).to(A						)
           _UpperCAmelCase      :     List[str]        =   model.generate(
               input_ids=batch.input_ids					,     num_beams=8					,     )
           _UpperCAmelCase      :     Any        =   tokenizer.batch_decode(
               A					,     skip_special_tokens=A					,     clean_up_tokenization_spaces=A						)
           _UpperCAmelCase      :     Any        =   calculate_bleu(A					,     A						)
           print(A						)
           self.assertGreaterEqual(scores["bleu"]					,     A						)
