
"""simple docstring"""




import datetime
import platform
import subprocess
from typing import Optional, Tuple, Union

import numpy as np


def   __magic_name__	(      __snake_case				:       bytes     , __snake_case				:       int  ) ->				np.array:
       lowercase    :				Any       		=							f"""{sampling_rate}"""
       lowercase    :				str       		=							"1"
       lowercase    :				Tuple       		=							"f32le"
       lowercase    :				str       		=							[
           "ffmpeg",
           "-i",
           "pipe:0",
           "-ac",
           ac,
           "-ar",
           ar,
           "-f",
           format_for_conversion,
           "-hide_banner",
           "-loglevel",
           "quiet",
           "pipe:1",
       ]

       try:
              with subprocess.Popen(__snake_case     , stdin=subprocess.PIPE     , stdout=subprocess.PIPE  ) as ffmpeg_process:
                     lowercase    :				str       		=							ffmpeg_process.communicate(__snake_case  )
       except FileNotFoundError as error:
              raise ValueError("ffmpeg was not found but is required to load audio files from filename"  ) from error
       lowercase    :				Any       		=							output_stream[0]
       lowercase    :				str       		=							np.frombuffer(__snake_case     , np.floataa  )
       if audio.shape[0] == 0:
              raise ValueError("Malformed soundfile"  )
       return audio


def   __magic_name__	(      __snake_case				:       int     , __snake_case				:       float     , __snake_case				:       str = "f32le"     , ) ->				Optional[int]:
       lowercase    :				Optional[int]       		=							f"""{sampling_rate}"""
       lowercase    :				int       		=							"1"
       if format_for_conversion == "s16le":
              lowercase    :				Optional[int]       		=							2
       elif format_for_conversion == "f32le":
              lowercase    :				Tuple       		=							4
       else:
              raise ValueError(f"""Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`"""  )

       lowercase    :				Optional[Any]       		=							platform.system()
       if system == "Linux":
              lowercase    :				Tuple       		=							"alsa"
              lowercase    :				Optional[Any]       		=							"default"
       elif system == "Darwin":
              lowercase    :				Any       		=							"avfoundation"
              lowercase    :				List[Any]       		=							":0"
       elif system == "Windows":
              lowercase    :				str       		=							"dshow"
              lowercase    :				Optional[int]       		=							"default"

       lowercase    :				Dict       		=							[
           "ffmpeg",
           "-f",
           format_,
           "-i",
           input_,
           "-ac",
           ac,
           "-ar",
           ar,
           "-f",
           format_for_conversion,
           "-fflags",
           "nobuffer",
           "-hide_banner",
           "-loglevel",
           "quiet",
           "pipe:1",
       ]
       lowercase    :				Dict       		=							int(round(sampling_rate * chunk_length_s  )  ) * size_of_sample
       lowercase    :				str       		=							_ffmpeg_stream(__snake_case     , __snake_case  )
       for item in iterator:
              yield item


def   __magic_name__	(      __snake_case				:       int     , __snake_case				:       float     , __snake_case				:       Optional[int] = None     , __snake_case				:       Optional[Union[Tuple[float, float], float]] = None     , __snake_case				:       str = "f32le"     , ) ->				Optional[Any]:
       if stream_chunk_s is not None:
              lowercase    :				int       		=							stream_chunk_s
       else:
              lowercase    :				Union[str, Any]       		=							chunk_length_s

       lowercase    :				Tuple       		=							ffmpeg_microphone(__snake_case     , __snake_case     , format_for_conversion=__snake_case  )
       if format_for_conversion == "s16le":
              lowercase    :				Dict       		=							np.intaa
              lowercase    :				Optional[Any]       		=							2
       elif format_for_conversion == "f32le":
              lowercase    :				List[Any]       		=							np.floataa
              lowercase    :				Union[str, Any]       		=							4
       else:
              raise ValueError(f"""Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`"""  )

       if stride_length_s is None:
              lowercase    :				Tuple       		=							chunk_length_s / 6
       lowercase    :				Optional[Any]       		=							int(round(sampling_rate * chunk_length_s  )  ) * size_of_sample
       if isinstance(__snake_case     , (int, float)  ):
              lowercase    :				Optional[Any]       		=							[stride_length_s, stride_length_s]

       lowercase    :				str       		=							int(round(sampling_rate * stride_length_s[0]  )  ) * size_of_sample
       lowercase    :				Tuple       		=							int(round(sampling_rate * stride_length_s[1]  )  ) * size_of_sample
       lowercase    :				str       		=							datetime.datetime.now()
       lowercase    :				List[Any]       		=							datetime.timedelta(seconds=__snake_case  )
       for item in chunk_bytes_iter(__snake_case     , __snake_case     , stride=(stride_left, stride_right)     , stream=__snake_case  ):
              # Put everything back in numpy scale
              lowercase    :				List[Any]       		=							np.frombuffer(item["raw"]     , dtype=__snake_case  )
              lowercase    :				Dict       		=							(
                  item["stride"][0] // size_of_sample,
                  item["stride"][1] // size_of_sample,
              )
              lowercase    :				Optional[Any]       		=							sampling_rate
              audio_time += delta
              if datetime.datetime.now() > audio_time + 10 * delta:
                     # We're late !! SKIP
                     continue
              yield item


def   __magic_name__	(      __snake_case				:       Optional[Any]     , __snake_case				:       int     , __snake_case				:       Tuple[int, int]     , __snake_case				:       bool = False  ) ->				Optional[Any]:
       lowercase    :				List[Any]       		=							B""
       lowercase		,    lowercase    :				Union[str, Any]       		=							stride
       if stride_left + stride_right >= chunk_len:
              raise ValueError(
                  f"""Stride needs to be strictly smaller than chunk_len: ({stride_left}, {stride_right}) vs {chunk_len}"""  )
       lowercase    :				List[str]       		=							0
       for raw in iterator:
              acc += raw
              if stream and len(__snake_case  ) < chunk_len:
                     lowercase    :				Union[str, Any]       		=							(_stride_left, 0)
                     yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
              else:
                     while len(__snake_case  ) >= chunk_len:
                            # We are flushing the accumulator
                            lowercase    :				Optional[int]       		=							(_stride_left, stride_right)
                            lowercase    :				Tuple       		=							{"raw": acc[:chunk_len], "stride": stride}
                            if stream:
                                   lowercase    :				int       		=							False
                            yield item
                            lowercase    :				List[Any]       		=							stride_left
                            lowercase    :				Any       		=							acc[chunk_len - stride_left - stride_right :]
    # Last chunk
       if len(__snake_case  ) > stride_left:
              lowercase    :				List[str]       		=							{"raw": acc, "stride": (_stride_left, 0)}
              if stream:
                     lowercase    :				Optional[int]       		=							False
              yield item


def   __magic_name__	(      __snake_case				:       Optional[Any]     , __snake_case				:       int  ) ->				Any:
       lowercase    :				Optional[int]       		=							2**24  # 16Mo
       try:
              with subprocess.Popen(__snake_case     , stdout=subprocess.PIPE     , bufsize=__snake_case  ) as ffmpeg_process:
                     while True:
                            lowercase    :				Dict       		=							ffmpeg_process.stdout.read(__snake_case  )
                            if raw == b"":
                                   break
                            yield raw
       except FileNotFoundError as error:
              raise ValueError("ffmpeg was not found but is required to stream audio files from filename"  ) from error




"""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


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

class 		a__						(      a_						):
   def __init__(			self	,    _a	,    _a	,    _a	,    _a	,    _a	,    _a	,    _a	,    _a = True	,    ):
          super()._init_()

          lowercase    :				Optional[Any]       		=							StableDiffusionPipeline.from_pretrained(_a				)
          lowercase    :				str       		=							StableDiffusionPipeline.from_pretrained(_a				)
          lowercase    :				Dict       		=							StableDiffusionPipeline.from_pretrained(_a				)
          lowercase    :				Union[str, Any]       		=							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 			__magic_name__			(			self				):
          return {k: getattr(self	,    _a				) for k in self.config.keys() if not k.startswith("_"				)}
   def 			__magic_name__			(			self	,    _a = "auto"				):
          if slice_size == "auto":
                 # half the attention head size is usually a good trade-off between
                 # speed and memory
                 lowercase    :				str       		=							self.unet.config.attention_head_dim // 2
          self.unet.set_attention_slice(_a				)
   def 			__magic_name__			(			self				):
          self.enable_attention_slicing(_a				)
   @torch.no_grad()
   def 			__magic_name__			(			self	,    _a	,    _a = 512	,    _a = 512	,    _a = 50	,    _a = 7.5	,    _a = None	,    _a = 1	,    _a = 0.0	,    _a = None	,    _a = None	,    _a = "pil"	,    _a = True	,    _a = None	,    _a = 1	,    **_a	,    ):
          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 			__magic_name__			(			self	,    _a	,    _a = 512	,    _a = 512	,    _a = 50	,    _a = 7.5	,    _a = None	,    _a = 1	,    _a = 0.0	,    _a = None	,    _a = None	,    _a = "pil"	,    _a = True	,    _a = None	,    _a = 1	,    **_a	,    ):
          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 			__magic_name__			(			self	,    _a	,    _a = 512	,    _a = 512	,    _a = 50	,    _a = 7.5	,    _a = None	,    _a = 1	,    _a = 0.0	,    _a = None	,    _a = None	,    _a = "pil"	,    _a = True	,    _a = None	,    _a = 1	,    **_a	,    ):
          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 			__magic_name__			(			self	,    _a	,    _a = 512	,    _a = 512	,    _a = 50	,    _a = 7.5	,    _a = None	,    _a = 1	,    _a = 0.0	,    _a = None	,    _a = None	,    _a = "pil"	,    _a = True	,    _a = None	,    _a = 1	,    **_a	,    ):
          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 			__magic_name__			(			self	,    _a	,    _a = 512	,    _a = 512	,    _a = 50	,    _a = 7.5	,    _a = None	,    _a = 1	,    _a = 0.0	,    _a = None	,    _a = None	,    _a = "pil"	,    _a = True	,    _a = None	,    _a = 1	,    **_a	,    ):
          lowercase    :				List[Any]       		=							"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
          lowercase    :				List[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
          lowercase    :				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.3
          lowercase    :				str       		=							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
          lowercase    :				Optional[int]       		=							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]]				)


