




"""simple docstring"""


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

import numpy as np



def 	lowercase_				(			__UpperCAmelCase    ,		__UpperCAmelCase				)						->    np.array:
      lowerCAmelCase__      :			int   =	f"""{sampling_rate}"""
      lowerCAmelCase__      :			List[Any]   =	"""1"""
      lowerCAmelCase__      :			Dict   =	"""f32le"""
      lowerCAmelCase__      :			Optional[Any]   =	[
          """ffmpeg""",
          """-i""",
          """pipe:0""",
          """-ac""",
          ac,
          """-ar""",
          ar,
          """-f""",
          format_for_conversion,
          """-hide_banner""",
          """-loglevel""",
          """quiet""",
          """pipe:1""",
      ]

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



def 	lowercase_				(			__UpperCAmelCase    ,		__UpperCAmelCase    ,		__UpperCAmelCase = "f32le"    ,		)						->    Any:
      lowerCAmelCase__      :			List[Any]   =	f"""{sampling_rate}"""
      lowerCAmelCase__      :			int   =	"""1"""
      if format_for_conversion == "s16le":
            lowerCAmelCase__      :			List[str]   =	2
      elif format_for_conversion == "f32le":
            lowerCAmelCase__      :			Optional[Any]   =	4
      else:
            raise ValueError(f"""Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`"""				)

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

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



def 	lowercase_				(			__UpperCAmelCase    ,		__UpperCAmelCase    ,		__UpperCAmelCase = None    ,		__UpperCAmelCase = None    ,		__UpperCAmelCase = "f32le"    ,		)						->    List[str]:
      if stream_chunk_s is not None:
            lowerCAmelCase__      :			List[Any]   =	stream_chunk_s
      else:
            lowerCAmelCase__      :			Tuple   =	chunk_length_s

      lowerCAmelCase__      :			int   =	ffmpeg_microphone(__UpperCAmelCase    ,		__UpperCAmelCase    ,		format_for_conversion=__UpperCAmelCase				)
      if format_for_conversion == "s16le":
            lowerCAmelCase__      :			Optional[int]   =	np.intaa
            lowerCAmelCase__      :			Optional[int]   =	2
      elif format_for_conversion == "f32le":
            lowerCAmelCase__      :			Tuple   =	np.floataa
            lowerCAmelCase__      :			Tuple   =	4
      else:
            raise ValueError(f"""Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`"""				)

      if stride_length_s is None:
            lowerCAmelCase__      :			List[Any]   =	chunk_length_s / 6
      lowerCAmelCase__      :			str   =	int(round(sampling_rate * chunk_length_s				)				) * size_of_sample
      if isinstance(__UpperCAmelCase    ,		(int, float)				):
            lowerCAmelCase__      :			str   =	[stride_length_s, stride_length_s]

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



def 	lowercase_				(			__UpperCAmelCase    ,		__UpperCAmelCase    ,		__UpperCAmelCase    ,		__UpperCAmelCase = False				)						->    str:
      lowerCAmelCase__      :			int   =	B""""""
      lowerCAmelCase__ ,   lowerCAmelCase__      :			Optional[int]   =	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}"""				)
      lowerCAmelCase__      :			str   =	0
      for raw in iterator:
            acc += raw
            if stream and len(__UpperCAmelCase				) < chunk_len:
                  lowerCAmelCase__      :			Union[str, Any]   =	(_stride_left, 0)
                  yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
            else:
                  while len(__UpperCAmelCase				) >= chunk_len:
                        # We are flushing the accumulator
                        lowerCAmelCase__      :			Tuple   =	(_stride_left, stride_right)
                        lowerCAmelCase__      :			Dict   =	{"""raw""": acc[:chunk_len], """stride""": stride}
                        if stream:
                              lowerCAmelCase__      :			Dict   =	False
                        yield item
                        lowerCAmelCase__      :			Optional[Any]   =	stride_left
                        lowerCAmelCase__      :			List[Any]   =	acc[chunk_len - stride_left - stride_right :]
    # Last chunk
      if len(__UpperCAmelCase				) > stride_left:
            lowerCAmelCase__      :			Any   =	{"""raw""": acc, """stride""": (_stride_left, 0)}
            if stream:
                  lowerCAmelCase__      :			Optional[Any]   =	False
            yield item



def 	lowercase_				(			__UpperCAmelCase    ,		__UpperCAmelCase				)						->    List[str]:
      lowerCAmelCase__      :			Tuple   =	2**24  # 16Mo
      try:
            with subprocess.Popen(__UpperCAmelCase    ,		stdout=subprocess.PIPE    ,		bufsize=__UpperCAmelCase				) as ffmpeg_process:
                  while True:
                        lowerCAmelCase__      :			List[Any]   =	ffmpeg_process.stdout.read(__UpperCAmelCase				)
                        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 Optional, Tuple

import jax
import jax.numpy as jnp
from flax import linen as nn
from flax.core.frozen_dict import FrozenDict
from transformers import CLIPConfig, FlaxPreTrainedModel
from transformers.models.clip.modeling_flax_clip import FlaxCLIPVisionModule



def 	lowercase_				(			__UpperCAmelCase    ,		__UpperCAmelCase    ,		__UpperCAmelCase=1E-12				)						->    Dict:
      lowerCAmelCase__      :			Optional[int]   =	jnp.divide(emb_a.T    ,		jnp.clip(jnp.linalg.norm(__UpperCAmelCase    ,		axis=1				)    ,		a_min=__UpperCAmelCase				)				).T
      lowerCAmelCase__      :			Any   =	jnp.divide(emb_a.T    ,		jnp.clip(jnp.linalg.norm(__UpperCAmelCase    ,		axis=1				)    ,		a_min=__UpperCAmelCase				)				).T
      return jnp.matmul(__UpperCAmelCase    ,		norm_emb_a.T				)


class _lowerCamelCase    (					nn.Module						):
   _lowerCamelCase    :CLIPConfig
   _lowerCamelCase    :jnp.dtype     				=     jnp.floataa






   def 	_lowerCAmelCase   (		self       :		List[str]   )			->						Tuple:

         """simple docstring"""
         lowerCAmelCase__      :			Optional[int]   =	FlaxCLIPVisionModule(self.config.vision_config   )
         lowerCAmelCase__      :			Dict   =	nn.Dense(self.config.projection_dim   ,     use_bias=UpperCamelCase   ,     dtype=self.dtype   )

         lowerCAmelCase__      :			Optional[int]   =	self.param("""concept_embeds"""   ,     jax.nn.initializers.ones   ,     (17, self.config.projection_dim)   )
         lowerCAmelCase__      :			Optional[int]   =	self.param(
             """special_care_embeds"""   ,     jax.nn.initializers.ones   ,     (3, self.config.projection_dim)   )

         lowerCAmelCase__      :			List[Any]   =	self.param("""concept_embeds_weights"""   ,     jax.nn.initializers.ones   ,     (17,)   )
         lowerCAmelCase__      :			List[Any]   =	self.param("""special_care_embeds_weights"""   ,     jax.nn.initializers.ones   ,     (3,)   )






   def __call__(		self       :		str   ,     UpperCamelCase       :		List[Any]   )			->						Union[str, Any]:

         """simple docstring"""
         lowerCAmelCase__      :			Union[str, Any]   =	self.vision_model(UpperCamelCase   )[1]
         lowerCAmelCase__      :			List[Any]   =	self.visual_projection(UpperCamelCase   )

         lowerCAmelCase__      :			str   =	jax_cosine_distance(UpperCamelCase   ,     self.special_care_embeds   )
         lowerCAmelCase__      :			Optional[int]   =	jax_cosine_distance(UpperCamelCase   ,     self.concept_embeds   )

         # increase this value to create a stronger `nfsw` filter
         # at the cost of increasing the possibility of filtering benign image inputs
         lowerCAmelCase__      :			Any   =	0.0

         lowerCAmelCase__      :			List[Any]   =	special_cos_dist - self.special_care_embeds_weights[None, :] + adjustment
         lowerCAmelCase__      :			Any   =	jnp.round(UpperCamelCase   ,     3   )
         lowerCAmelCase__      :			List[str]   =	jnp.any(special_scores > 0   ,     axis=1   ,     keepdims=UpperCamelCase   )
         # Use a lower threshold if an image has any special care concept
         lowerCAmelCase__      :			Tuple   =	is_special_care * 0.01

         lowerCAmelCase__      :			List[Any]   =	cos_dist - self.concept_embeds_weights[None, :] + special_adjustment
         lowerCAmelCase__      :			Dict   =	jnp.round(UpperCamelCase   ,     3   )
         lowerCAmelCase__      :			List[str]   =	jnp.any(concept_scores > 0   ,     axis=1   )

         return has_nsfw_concepts






class _lowerCamelCase    (					a_						):
   _lowerCamelCase    :Optional[Any]     				=     CLIPConfig
   _lowerCamelCase    :Dict     				=     "clip_input"
   _lowerCamelCase    :int     				=     FlaxStableDiffusionSafetyCheckerModule






   def __init__(		self       :		int   ,     UpperCamelCase       :		CLIPConfig   ,     UpperCamelCase       :		Optional[Tuple] = None   ,     UpperCamelCase       :		int = 0   ,     UpperCamelCase       :		jnp.dtype = jnp.floataa   ,     UpperCamelCase       :		bool = True   ,     **UpperCamelCase       :		int   ,     )			->						str:

         """simple docstring"""
         if input_shape is None:
               lowerCAmelCase__      :			str   =	(1, 2_24, 2_24, 3)
         lowerCAmelCase__      :			Union[str, Any]   =	self.module_class(config=UpperCamelCase   ,     dtype=UpperCamelCase   ,     **UpperCamelCase   )
         super().__init__(UpperCamelCase   ,     UpperCamelCase   ,     input_shape=UpperCamelCase   ,     seed=UpperCamelCase   ,     dtype=UpperCamelCase   ,     _do_init=_do_init   )






   def 	_lowerCAmelCase   (		self       :		Dict   ,     UpperCamelCase       :		jax.random.KeyArray   ,     UpperCamelCase       :		Tuple   ,     UpperCamelCase       :		FrozenDict = None   )			->						FrozenDict:

         """simple docstring"""
         # init input tensor
         lowerCAmelCase__      :			Union[str, Any]   =	jax.random.normal(UpperCamelCase   ,     UpperCamelCase   )

         lowerCAmelCase__ ,   lowerCAmelCase__      :			Union[str, Any]   =	jax.random.split(UpperCamelCase   )
         lowerCAmelCase__      :			List[Any]   =	{"""params""": params_rng, """dropout""": dropout_rng}

         lowerCAmelCase__      :			str   =	self.module.init(UpperCamelCase   ,     UpperCamelCase   )["""params"""]

         return random_params






   def __call__(		self       :		Dict   ,     UpperCamelCase       :		Tuple   ,     UpperCamelCase       :		dict = None   ,     )			->						List[Any]:

         """simple docstring"""
         lowerCAmelCase__      :			Union[str, Any]   =	jnp.transpose(UpperCamelCase   ,     (0, 2, 3, 1)   )

         return self.module.apply(
             {"""params""": params or self.params}   ,     jnp.array(UpperCamelCase   ,     dtype=jnp.floataa   )   ,     rngs={}   ,     )





