





import json
import os
from typing import Optional

import numpy as np

from ...feature_extraction_utils import BatchFeature
from ...processing_utils import ProcessorMixin
from ...utils import logging
from ...utils.hub import get_file_from_repo
from ..auto import AutoTokenizer


UpperCamelCase__         =      logging.get_logger(__name__)
class       a__   ( snake_case__				):
						_a							:   List[str] =   """AutoTokenizer"""
						_a							:   List[Any] =   ["""tokenizer"""]

						_a							:   Tuple =   {
						    """semantic_prompt""": 1,
						    """coarse_prompt""": 2,
						    """fine_prompt""": 2,
						}






						def __init__(       self		,      _A		,      _A=None      ):




											"""simple docstring"""



											super().__init__(_A      )

											__lowerCAmelCase           =	speaker_embeddings






						@classmethod
						def     __SCREAMING_SNAKE_CASE(       cls		,      _A		,      _A="speaker_embeddings_path.json"		,      **_A      ):




											"""simple docstring"""



											if speaker_embeddings_dict_path is not None:
																__lowerCAmelCase           =	get_file_from_repo(
																    _A		,      _A		,      subfolder=kwargs.pop("subfolder"		,      _A      )		,      cache_dir=kwargs.pop("cache_dir"		,      _A      )		,      force_download=kwargs.pop("force_download"		,      _A      )		,      proxies=kwargs.pop("proxies"		,      _A      )		,      resume_download=kwargs.pop("resume_download"		,      _A      )		,      local_files_only=kwargs.pop("local_files_only"		,      _A      )		,      use_auth_token=kwargs.pop("use_auth_token"		,      _A      )		,      revision=kwargs.pop("revision"		,      _A      )		,      )
																if speaker_embeddings_path is None:
																					logger.warning(
																					    f"""`{os.path.join(_A		,      _A      )}` does not exists
                    , no preloaded speaker embeddings will be used - Make sure to provide a correct path to the json
                    dictionnary if wanted, otherwise set `speaker_embeddings_dict_path=None`."""      )
																					__lowerCAmelCase           =	None
																else:
																					with open(_A      ) as speaker_embeddings_json:
																										__lowerCAmelCase           =	json.load(_A      )
											else:
																__lowerCAmelCase           =	None

											__lowerCAmelCase           =	AutoTokenizer.from_pretrained(_A		,      **_A      )

											return cls(tokenizer=_A		,      speaker_embeddings=_A      )






						def     __SCREAMING_SNAKE_CASE(       self		,      _A		,      _A="speaker_embeddings_path.json"		,      _A="speaker_embeddings"		,      _A = False		,      **_A		,      ):




											"""simple docstring"""



											if self.speaker_embeddings is not None:
																os.makedirs(os.path.join(_A		,      _A		,      "v2"      )		,      exist_ok=_A      )

																__lowerCAmelCase           =	{}

																__lowerCAmelCase           =	save_directory

																for prompt_key in self.speaker_embeddings:
																					if prompt_key != "repo_or_path":
																										__lowerCAmelCase           =	self._load_voice_preset(_A      )

																										__lowerCAmelCase           =	{}
																										for key in self.speaker_embeddings[prompt_key]:
																															np.save(
																															    os.path.join(
																															        embeddings_dict["repo_or_path"]		,      _A		,      f"""{prompt_key}_{key}"""      )		,      voice_preset[key]		,      allow_pickle=_A		,      )
																															__lowerCAmelCase           =	os.path.join(_A		,      f"""{prompt_key}_{key}.npy"""      )

																										__lowerCAmelCase           =	tmp_dict

																with open(os.path.join(_A		,      _A      )		,      "w"      ) as fp:
																					json.dump(_A		,      _A      )

											super().save_pretrained(_A		,      _A		,      **_A      )






						def     __SCREAMING_SNAKE_CASE(       self		,      _A = None		,      **_A      ):




											"""simple docstring"""



											__lowerCAmelCase           =	self.speaker_embeddings[voice_preset]

											__lowerCAmelCase           =	{}
											for key in ["semantic_prompt", "coarse_prompt", "fine_prompt"]:
																if key not in voice_preset_paths:
																					raise ValueError(
																					    f"""Voice preset unrecognized, missing {key} as a key in self.speaker_embeddings[{voice_preset}]."""      )

																__lowerCAmelCase           =	get_file_from_repo(
																    self.speaker_embeddings.get("repo_or_path"		,      "/"      )		,      voice_preset_paths[key]		,      subfolder=kwargs.pop("subfolder"		,      _A      )		,      cache_dir=kwargs.pop("cache_dir"		,      _A      )		,      force_download=kwargs.pop("force_download"		,      _A      )		,      proxies=kwargs.pop("proxies"		,      _A      )		,      resume_download=kwargs.pop("resume_download"		,      _A      )		,      local_files_only=kwargs.pop("local_files_only"		,      _A      )		,      use_auth_token=kwargs.pop("use_auth_token"		,      _A      )		,      revision=kwargs.pop("revision"		,      _A      )		,      )
																if path is None:
																					raise ValueError(
																					    f"""`{os.path.join(self.speaker_embeddings.get('repo_or_path'		,      '/'      )		,      voice_preset_paths[key]      )}` does not exists
                    , no preloaded voice preset will be used - Make sure to provide correct paths to the {voice_preset}
                    embeddings."""      )

																__lowerCAmelCase           =	np.load(_A      )

											return voice_preset_dict






						def     __SCREAMING_SNAKE_CASE(       self		,      _A = None      ):




											"""simple docstring"""



											for key in ["semantic_prompt", "coarse_prompt", "fine_prompt"]:
																if key not in voice_preset:
																					raise ValueError(f"""Voice preset unrecognized, missing {key} as a key."""      )

																if not isinstance(voice_preset[key]		,      np.ndarray      ):
																					raise ValueError(f"""{key} voice preset must be a {str(self.preset_shape[key]      )}D ndarray."""      )

																if len(voice_preset[key].shape      ) != self.preset_shape[key]:
																					raise ValueError(f"""{key} voice preset must be a {str(self.preset_shape[key]      )}D ndarray."""      )






						def __call__(       self		,      _A=None		,      _A=None		,      _A="pt"		,      _A=2_5_6		,      _A=False		,      _A=True		,      _A=False		,      **_A		,      ):




											"""simple docstring"""



											if voice_preset is not None and not isinstance(_A		,      _A      ):
																if (
																    isinstance(_A		,      _A      )
																    and self.speaker_embeddings is not None
																    and voice_preset in self.speaker_embeddings
																):
																					__lowerCAmelCase           =	self._load_voice_preset(_A      )

																else:
																					if isinstance(_A		,      _A      ) and not voice_preset.endswith(".npz"      ):
																										__lowerCAmelCase           =	voice_preset + '.npz'

																					__lowerCAmelCase           =	np.load(_A      )

											if voice_preset is not None:
																self._validate_voice_preset_dict(_A		,      **_A      )
																__lowerCAmelCase           =	BatchFeature(data=_A		,      tensor_type=_A      )

											__lowerCAmelCase           =	self.tokenizer(
											    _A		,      return_tensors=_A		,      padding="max_length"		,      max_length=_A		,      return_attention_mask=_A		,      return_token_type_ids=_A		,      add_special_tokens=_A		,      **_A		,      )

											if voice_preset is not None:
																__lowerCAmelCase           =	voice_preset

											return encoded_text










from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..utils import maybe_allow_in_graph
from .activations import get_activation
from .attention_processor import Attention
from .embeddings import CombinedTimestepLabelEmbeddings
@maybe_allow_in_graph
class       a__   ( nn.Module				):






						def __init__(       self		,      _A		,      _A		,      _A		,      _A=0.0		,      _A = None		,      _A = "geglu"		,      _A = None		,      _A = False		,      _A = False		,      _A = False		,      _A = False		,      _A = True		,      _A = "layer_norm"		,      _A = False		,      ):




											"""simple docstring"""



											super().__init__()
											__lowerCAmelCase           =	only_cross_attention

											__lowerCAmelCase           =	(num_embeds_ada_norm is not None) and norm_type == "ada_norm_zero"
											__lowerCAmelCase           =	(num_embeds_ada_norm is not None) and norm_type == "ada_norm"

											if norm_type in ("ada_norm", "ada_norm_zero") and num_embeds_ada_norm is None:
																raise ValueError(
																    f"""`norm_type` is set to {norm_type}, but `num_embeds_ada_norm` is not defined. Please make sure to"""
																    f""" define `num_embeds_ada_norm` if setting `norm_type` to {norm_type}."""      )

											# Define 3 blocks. Each block has its own normalization layer.
											# 1. Self-Attn
											if self.use_ada_layer_norm:
																__lowerCAmelCase           =	AdaLayerNorm(_A		,      _A      )
											elif self.use_ada_layer_norm_zero:
																__lowerCAmelCase           =	AdaLayerNormZero(_A		,      _A      )
											else:
																__lowerCAmelCase           =	nn.LayerNorm(_A		,      elementwise_affine=_A      )
											__lowerCAmelCase           =	Attention(
											    query_dim=_A		,      heads=_A		,      dim_head=_A		,      dropout=_A		,      bias=_A		,      cross_attention_dim=cross_attention_dim if only_cross_attention else None		,      upcast_attention=_A		,      )

											# 2. Cross-Attn
											if cross_attention_dim is not None or double_self_attention:
																# We currently only use AdaLayerNormZero for self attention where there will only be one attention block.
																# I.e. the number of returned modulation chunks from AdaLayerZero would not make sense if returned during
																# the second cross attention block.
																__lowerCAmelCase           =	(
																    AdaLayerNorm(_A		,      _A      )
																    if self.use_ada_layer_norm
																    else nn.LayerNorm(_A		,      elementwise_affine=_A      )
																)
																__lowerCAmelCase           =	Attention(
																    query_dim=_A		,      cross_attention_dim=cross_attention_dim if not double_self_attention else None		,      heads=_A		,      dim_head=_A		,      dropout=_A		,      bias=_A		,      upcast_attention=_A		,      )  # is self-attn if encoder_hidden_states is none
											else:
																__lowerCAmelCase           =	None
																__lowerCAmelCase           =	None

											# 3. Feed-forward
											__lowerCAmelCase           =	nn.LayerNorm(_A		,      elementwise_affine=_A      )
											__lowerCAmelCase           =	FeedForward(_A		,      dropout=_A		,      activation_fn=_A		,      final_dropout=_A      )

											# let chunk size default to None
											__lowerCAmelCase           =	None
											__lowerCAmelCase           =	0






						def     __SCREAMING_SNAKE_CASE(       self		,      _A		,      _A      ):




											"""simple docstring"""



											__lowerCAmelCase           =	chunk_size
											__lowerCAmelCase           =	dim






						def     __SCREAMING_SNAKE_CASE(       self		,      _A		,      _A = None		,      _A = None		,      _A = None		,      _A = None		,      _A = None		,      _A = None		,      ):




											"""simple docstring"""



											if self.use_ada_layer_norm:
																__lowerCAmelCase           =	self.norma(_A		,      _A      )
											elif self.use_ada_layer_norm_zero:
																__lowerCAmelCase       ,							__lowerCAmelCase       ,							__lowerCAmelCase       ,							__lowerCAmelCase       ,							__lowerCAmelCase           =	self.norma(
																    _A		,      _A		,      _A		,      hidden_dtype=hidden_states.dtype      )
											else:
																__lowerCAmelCase           =	self.norma(_A      )

											__lowerCAmelCase           =	cross_attention_kwargs if cross_attention_kwargs is not None else {}

											__lowerCAmelCase           =	self.attna(
											    _A		,      encoder_hidden_states=encoder_hidden_states if self.only_cross_attention else None		,      attention_mask=_A		,      **_A		,      )
											if self.use_ada_layer_norm_zero:
																__lowerCAmelCase           =	gate_msa.unsqueeze(1      ) * attn_output
											__lowerCAmelCase           =	attn_output + hidden_states

											# 2. Cross-Attention
											if self.attna is not None:
																__lowerCAmelCase           =	(
																    self.norma(_A		,      _A      ) if self.use_ada_layer_norm else self.norma(_A      )
																)

																__lowerCAmelCase           =	self.attna(
																    _A		,      encoder_hidden_states=_A		,      attention_mask=_A		,      **_A		,      )
																__lowerCAmelCase           =	attn_output + hidden_states

											# 3. Feed-forward
											__lowerCAmelCase           =	self.norma(_A      )

											if self.use_ada_layer_norm_zero:
																__lowerCAmelCase           =	norm_hidden_states * (1 + scale_mlp[:, None]) + shift_mlp[:, None]

											if self._chunk_size is not None:
																# "feed_forward_chunk_size" can be used to save memory
																if norm_hidden_states.shape[self._chunk_dim] % self._chunk_size != 0:
																					raise ValueError(
																					    f"""`hidden_states` dimension to be chunked: {norm_hidden_states.shape[self._chunk_dim]} has to be divisible by chunk size: {self._chunk_size}. Make sure to set an appropriate `chunk_size` when calling `unet.enable_forward_chunking`."""      )

																__lowerCAmelCase           =	norm_hidden_states.shape[self._chunk_dim] // self._chunk_size
																__lowerCAmelCase           =	torch.cat(
																    [self.ff(_A      ) for hid_slice in norm_hidden_states.chunk(_A		,      dim=self._chunk_dim      )]		,      dim=self._chunk_dim		,      )
											else:
																__lowerCAmelCase           =	self.ff(_A      )

											if self.use_ada_layer_norm_zero:
																__lowerCAmelCase           =	gate_mlp.unsqueeze(1      ) * ff_output

											__lowerCAmelCase           =	ff_output + hidden_states

											return hidden_states






class       a__   ( nn.Module				):






						def __init__(       self		,      _A		,      _A = None		,      _A = 4		,      _A = 0.0		,      _A = "geglu"		,      _A = False		,      ):




											"""simple docstring"""



											super().__init__()
											__lowerCAmelCase           =	int(dim * mult      )
											__lowerCAmelCase           =	dim_out if dim_out is not None else dim

											if activation_fn == "gelu":
																__lowerCAmelCase           =	GELU(_A		,      _A      )
											if activation_fn == "gelu-approximate":
																__lowerCAmelCase           =	GELU(_A		,      _A		,      approximate="tanh"      )
											elif activation_fn == "geglu":
																__lowerCAmelCase           =	GEGLU(_A		,      _A      )
											elif activation_fn == "geglu-approximate":
																__lowerCAmelCase           =	ApproximateGELU(_A		,      _A      )

											__lowerCAmelCase           =	nn.ModuleList([]      )
											# project in
											self.net.append(_A      )
											# project dropout
											self.net.append(nn.Dropout(_A      )      )
											# project out
											self.net.append(nn.Linear(_A		,      _A      )      )
											# FF as used in Vision Transformer, MLP-Mixer, etc. have a final dropout
											if final_dropout:
																self.net.append(nn.Dropout(_A      )      )






						def     __SCREAMING_SNAKE_CASE(       self		,      _A      ):




											"""simple docstring"""



											for module in self.net:
																__lowerCAmelCase           =	module(_A      )
											return hidden_states






class       a__   ( nn.Module				):






						def __init__(       self		,      _A		,      _A		,      _A = "none"      ):




											"""simple docstring"""



											super().__init__()
											__lowerCAmelCase           =	nn.Linear(_A		,      _A      )
											__lowerCAmelCase           =	approximate






						def     __SCREAMING_SNAKE_CASE(       self		,      _A      ):




											"""simple docstring"""



											if gate.device.type != "mps":
																return F.gelu(_A		,      approximate=self.approximate      )
											# mps: gelu is not implemented for float16
											return F.gelu(gate.to(dtype=torch.floataa      )		,      approximate=self.approximate      ).to(dtype=gate.dtype      )






						def     __SCREAMING_SNAKE_CASE(       self		,      _A      ):




											"""simple docstring"""



											__lowerCAmelCase           =	self.proj(_A      )
											__lowerCAmelCase           =	self.gelu(_A      )
											return hidden_states






class       a__   ( nn.Module				):






						def __init__(       self		,      _A		,      _A      ):




											"""simple docstring"""



											super().__init__()
											__lowerCAmelCase           =	nn.Linear(_A		,      dim_out * 2      )






						def     __SCREAMING_SNAKE_CASE(       self		,      _A      ):




											"""simple docstring"""



											if gate.device.type != "mps":
																return F.gelu(_A      )
											# mps: gelu is not implemented for float16
											return F.gelu(gate.to(dtype=torch.floataa      )      ).to(dtype=gate.dtype      )






						def     __SCREAMING_SNAKE_CASE(       self		,      _A      ):




											"""simple docstring"""



											__lowerCAmelCase       ,							__lowerCAmelCase           =	self.proj(_A      ).chunk(2		,      dim=-1      )
											return hidden_states * self.gelu(_A      )






class       a__   ( nn.Module				):






						def __init__(       self		,      _A		,      _A      ):




											"""simple docstring"""



											super().__init__()
											__lowerCAmelCase           =	nn.Linear(_A		,      _A      )






						def     __SCREAMING_SNAKE_CASE(       self		,      _A      ):




											"""simple docstring"""



											__lowerCAmelCase           =	self.proj(_A      )
											return x * torch.sigmoid(1.7_02 * x      )






class       a__   ( nn.Module				):






						def __init__(       self		,      _A		,      _A      ):




											"""simple docstring"""



											super().__init__()
											__lowerCAmelCase           =	nn.Embedding(_A		,      _A      )
											__lowerCAmelCase           =	nn.SiLU()
											__lowerCAmelCase           =	nn.Linear(_A		,      embedding_dim * 2      )
											__lowerCAmelCase           =	nn.LayerNorm(_A		,      elementwise_affine=_A      )






						def     __SCREAMING_SNAKE_CASE(       self		,      _A		,      _A      ):




											"""simple docstring"""



											__lowerCAmelCase           =	self.linear(self.silu(self.emb(_A      )      )      )
											__lowerCAmelCase       ,							__lowerCAmelCase           =	torch.chunk(_A		,      2      )
											__lowerCAmelCase           =	self.norm(_A      ) * (1 + scale) + shift
											return x






class       a__   ( nn.Module				):






						def __init__(       self		,      _A		,      _A      ):




											"""simple docstring"""



											super().__init__()

											__lowerCAmelCase           =	CombinedTimestepLabelEmbeddings(_A		,      _A      )

											__lowerCAmelCase           =	nn.SiLU()
											__lowerCAmelCase           =	nn.Linear(_A		,      6 * embedding_dim		,      bias=_A      )
											__lowerCAmelCase           =	nn.LayerNorm(_A		,      elementwise_affine=_A		,      eps=1E-6      )






						def     __SCREAMING_SNAKE_CASE(       self		,      _A		,      _A		,      _A		,      _A=None      ):




											"""simple docstring"""



											__lowerCAmelCase           =	self.linear(self.silu(self.emb(_A		,      _A		,      hidden_dtype=_A      )      )      )
											__lowerCAmelCase       ,							__lowerCAmelCase       ,							__lowerCAmelCase       ,							__lowerCAmelCase       ,							__lowerCAmelCase       ,							__lowerCAmelCase           =	emb.chunk(6		,      dim=1      )
											__lowerCAmelCase           =	self.norm(_A      ) * (1 + scale_msa[:, None]) + shift_msa[:, None]
											return x, gate_msa, shift_mlp, scale_mlp, gate_mlp






class       a__   ( nn.Module				):






						def __init__(       self		,      _A		,      _A		,      _A		,      _A = None		,      _A = 1E-5      ):




											"""simple docstring"""



											super().__init__()
											__lowerCAmelCase           =	num_groups
											__lowerCAmelCase           =	eps

											if act_fn is None:
																__lowerCAmelCase           =	None
											else:
																__lowerCAmelCase           =	get_activation(_A      )

											__lowerCAmelCase           =	nn.Linear(_A		,      out_dim * 2      )






						def     __SCREAMING_SNAKE_CASE(       self		,      _A		,      _A      ):




											"""simple docstring"""



											if self.act:
																__lowerCAmelCase           =	self.act(_A      )
											__lowerCAmelCase           =	self.linear(_A      )
											__lowerCAmelCase           =	emb[:, :, None, None]
											__lowerCAmelCase       ,							__lowerCAmelCase           =	emb.chunk(2		,      dim=1      )

											__lowerCAmelCase           =	F.group_norm(_A		,      self.num_groups		,      eps=self.eps      )
											__lowerCAmelCase           =	x * (1 + scale) + shift
											return x




