from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


lowerCAmelCase__     						=			logging.get_logger(__name__)
class     a_			(						SCREAMING_SNAKE_CASE						):






 '''simple docstring'''






 UpperCAmelCase_     			= ['input_features', 'attention_mask']




 def __init__(      self :  Tuple      ,   lowercase__ :  List[str]=80      ,   lowercase__ :  Optional[int]=16_000      ,   lowercase__ :  Any=80      ,   lowercase__ :  Optional[Any]=0.0      ,   lowercase__ :  int=True      ,   lowercase__ :  Optional[int]=True      ,   lowercase__ :  Any=True      ,   **lowercase__ :  Optional[Any]      ,   ):







        '''simple docstring'''



        super().__init__(feature_size=lowercase__      ,   sampling_rate=lowercase__      ,   padding_value=lowercase__      ,   **lowercase__)
        lowerCAmelCase__					     = num_mel_bins
        lowerCAmelCase__					     = do_ceptral_normalize
        lowerCAmelCase__					     = normalize_means
        lowerCAmelCase__					     = normalize_vars
        lowerCAmelCase__					     = True




 def 		__snake_case						(      self :  List[Any]      ,   lowercase__ :  np.ndarray      ,   ):







        '''simple docstring'''



        lowerCAmelCase__					     = waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
        lowerCAmelCase__					     = torch.from_numpy(lowercase__).unsqueeze(0)
        lowerCAmelCase__					     = ta_kaldi.fbank(lowercase__      ,   num_mel_bins=self.num_mel_bins      ,   sample_frequency=self.sampling_rate)
        return features.numpy()




 @staticmethod
 def 		__snake_case						(      lowercase__ :  np.ndarray      ,   lowercase__ :  int      ,   lowercase__ :  Optional[bool] = True      ,   lowercase__ :  Optional[bool] = True      ,   lowercase__ :  float = 0.0      ,   ):







        '''simple docstring'''



        if normalize_means:
               lowerCAmelCase__					     = x[:input_length].mean(axis=0)
               lowerCAmelCase__					     = np.subtract(lowercase__      ,   lowercase__)
        if normalize_vars:
               lowerCAmelCase__					     = x[:input_length].std(axis=0)
               lowerCAmelCase__					     = np.divide(lowercase__      ,   lowercase__)

        if input_length < x.shape[0]:
               lowerCAmelCase__					     = padding_value

        # make sure array is in float32
        lowerCAmelCase__					     = x.astype(np.floataa)

        return x




 def 		__snake_case						(      self :  Dict      ,   lowercase__ :  List[np.ndarray]      ,   lowercase__ :  Optional[np.ndarray] = None):







        '''simple docstring'''



        lowerCAmelCase__					     = attention_mask.sum(-1) if attention_mask is not None else [x.shape[0] for x in input_features]
        return [
            self.utterance_cmvn(lowercase__      ,   lowercase__      ,   self.normalize_means      ,   self.normalize_vars      ,   self.padding_value)
            for x, n in zip(lowercase__      ,   lowercase__)
        ]




 def __call__(      self :  List[str]      ,   lowercase__ :  Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]      ,   lowercase__ :  Union[bool, str, PaddingStrategy] = False      ,   lowercase__ :  Optional[int] = None      ,   lowercase__ :  bool = False      ,   lowercase__ :  Optional[int] = None      ,   lowercase__ :  Optional[Union[str, TensorType]] = None      ,   lowercase__ :  Optional[int] = None      ,   lowercase__ :  Optional[bool] = None      ,   **lowercase__ :  Union[str, Any]      ,   ):







        '''simple docstring'''



        if sampling_rate is not None:
               if sampling_rate != self.sampling_rate:
                      raise ValueError(
                          F"""The model corresponding to this feature extractor: {self} was trained using a sampling rate of"""
                          F""" {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with"""
                          F""" {self.sampling_rate} and not {sampling_rate}.""")
        else:
               logger.warning(
                   'It is strongly recommended to pass the `sampling_rate` argument to this function. '
                   'Failing to do so can result in silent errors that might be hard to debug.')

        lowerCAmelCase__					     = isinstance(lowercase__      ,   np.ndarray) and len(raw_speech.shape) > 1
        if is_batched_numpy and len(raw_speech.shape) > 2:
               raise ValueError(F"""Only mono-channel audio is supported for input to {self}""")
        lowerCAmelCase__					     = is_batched_numpy or (
            isinstance(lowercase__      ,   (list, tuple)) and (isinstance(raw_speech[0]      ,   (np.ndarray, tuple, list)))
        )

        if is_batched:
               lowerCAmelCase__					     = [np.asarray(lowercase__      ,   dtype=np.floataa) for speech in raw_speech]
        elif not is_batched and not isinstance(lowercase__      ,   np.ndarray):
               lowerCAmelCase__					     = np.asarray(lowercase__      ,   dtype=np.floataa)
        elif isinstance(lowercase__      ,   np.ndarray) and raw_speech.dtype is np.dtype(np.floataa):
               lowerCAmelCase__					     = raw_speech.astype(np.floataa)

        # always return batch
        if not is_batched:
               lowerCAmelCase__					     = [raw_speech]

        # extract fbank features
        lowerCAmelCase__					     = [self._extract_fbank_features(lowercase__) for waveform in raw_speech]

        # convert into correct format for padding
        lowerCAmelCase__					     = BatchFeature({'input_features': features})

        lowerCAmelCase__					     = self.pad(
            lowercase__      ,   padding=lowercase__      ,   max_length=lowercase__      ,   truncation=lowercase__      ,   pad_to_multiple_of=lowercase__      ,   return_attention_mask=lowercase__      ,   **lowercase__      ,   )

        # make sure list is in array format
        lowerCAmelCase__					     = padded_inputs.get('input_features')
        if isinstance(input_features[0]      ,   lowercase__):
               lowerCAmelCase__					     = [np.asarray(lowercase__      ,   dtype=np.floataa) for feature in input_features]

        lowerCAmelCase__					     = padded_inputs.get('attention_mask')
        if attention_mask is not None:
               lowerCAmelCase__					     = [np.asarray(lowercase__      ,   dtype=np.intaa) for array in attention_mask]

        # Utterance-level cepstral mean and variance normalization
        if self.do_ceptral_normalize:
               lowerCAmelCase__					     = (
                   np.array(lowercase__      ,   dtype=np.intaa)
                   if self._get_padding_strategies(lowercase__      ,   max_length=lowercase__) is not PaddingStrategy.DO_NOT_PAD
                   else None
               )
               lowerCAmelCase__					     = self.normalize(
                   padded_inputs['input_features']      ,   attention_mask=lowercase__)

        if return_tensors is not None:
               lowerCAmelCase__					     = padded_inputs.convert_to_tensors(lowercase__)

        return padded_inputs

import argparse
import math
import os

import torch
from neural_compressor.utils.pytorch import load
from PIL import Image
from transformers import CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, StableDiffusionPipeline, UNetaDConditionModel






def 				__lowerCamelCase		(						):
       lowerCAmelCase__					     = argparse.ArgumentParser()
       parser.add_argument(
           '-m'     ,  '--pretrained_model_name_or_path'     ,  type=lowerCAmelCase__     ,  default=lowerCAmelCase__     ,  required=lowerCAmelCase__     ,  help='Path to pretrained model or model identifier from huggingface.co/models.'     ,  )
       parser.add_argument(
           '-c'     ,  '--caption'     ,  type=lowerCAmelCase__     ,  default='robotic cat with wings'     ,  help='Text used to generate images.'     ,  )
       parser.add_argument(
           '-n'     ,  '--images_num'     ,  type=lowerCAmelCase__     ,  default=4     ,  help='How much images to generate.'     ,  )
       parser.add_argument(
           '-s'     ,  '--seed'     ,  type=lowerCAmelCase__     ,  default=4_2     ,  help='Seed for random process.'     ,  )
       parser.add_argument(
           '-ci'     ,  '--cuda_id'     ,  type=lowerCAmelCase__     ,  default=0     ,  help='cuda_id.'     ,  )
       lowerCAmelCase__					     = parser.parse_args()
       return args






def 				__lowerCamelCase		(						lowerCAmelCase__     ,  lowerCAmelCase__     ,  lowerCAmelCase__     ):
       if not len(lowerCAmelCase__     ) == rows * cols:
              raise ValueError('The specified number of rows and columns are not correct.'     )

       lowerCAmelCase__     , lowerCAmelCase__					     = imgs[0].size
       lowerCAmelCase__					     = Image.new('RGB'     ,  size=(cols * w, rows * h)     )
       lowerCAmelCase__     , lowerCAmelCase__					     = grid.size

       for i, img in enumerate(lowerCAmelCase__     ):
              grid.paste(lowerCAmelCase__     ,  box=(i % cols * w, i // cols * h)     )
       return grid






def 				__lowerCamelCase		(						lowerCAmelCase__     ,  lowerCAmelCase__="robotic cat with wings"     ,  lowerCAmelCase__=7.5     ,  lowerCAmelCase__=5_0     ,  lowerCAmelCase__=1     ,  lowerCAmelCase__=4_2     ,  ):
       lowerCAmelCase__					     = torch.Generator(pipeline.device     ).manual_seed(lowerCAmelCase__     )
       lowerCAmelCase__					     = pipeline(
           lowerCAmelCase__     ,  guidance_scale=lowerCAmelCase__     ,  num_inference_steps=lowerCAmelCase__     ,  generator=lowerCAmelCase__     ,  num_images_per_prompt=lowerCAmelCase__     ,  ).images
       lowerCAmelCase__					     = int(math.sqrt(lowerCAmelCase__     )     )
       lowerCAmelCase__					     = image_grid(lowerCAmelCase__     ,  rows=_rows     ,  cols=num_images_per_prompt // _rows     )
       return grid, images


lowerCAmelCase__     						=			parse_args()
# Load models and create wrapper for stable diffusion
lowerCAmelCase__     						=			CLIPTokenizer.from_pretrained(args.pretrained_model_name_or_path, subfolder='tokenizer')
lowerCAmelCase__     						=			CLIPTextModel.from_pretrained(args.pretrained_model_name_or_path, subfolder='text_encoder')
lowerCAmelCase__     						=			AutoencoderKL.from_pretrained(args.pretrained_model_name_or_path, subfolder='vae')
lowerCAmelCase__     						=			UNetaDConditionModel.from_pretrained(args.pretrained_model_name_or_path, subfolder='unet')

lowerCAmelCase__     						=			StableDiffusionPipeline.from_pretrained(
    args.pretrained_model_name_or_path, text_encoder=text_encoder, vae=vae, unet=unet, tokenizer=tokenizer
)
lowerCAmelCase__     						=			lambda images, clip_input: (images, False)
if os.path.exists(os.path.join(args.pretrained_model_name_or_path, 'best_model.pt')):
    lowerCAmelCase__     						=			load(args.pretrained_model_name_or_path, model=unet)
    unet.eval()
    setattr(pipeline, 'unet', unet)
else:
    lowerCAmelCase__     						=			unet.to(torch.device('cuda', args.cuda_id))
lowerCAmelCase__     						=			pipeline.to(unet.device)
lowerCAmelCase__						,		lowerCAmelCase__     						=			generate_images(pipeline, prompt=args.caption, num_images_per_prompt=args.images_num, seed=args.seed)
grid.save(os.path.join(args.pretrained_model_name_or_path, '{}.png'.format('_'.join(args.caption.split()))))
lowerCAmelCase__     						=			os.path.join(args.pretrained_model_name_or_path, '_'.join(args.caption.split()))
os.makedirs(dirname, exist_ok=True)
for idx, image in enumerate(images):
    image.save(os.path.join(dirname, '{}.png'.format(idx + 1)))
