

'''simple docstring'''






from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


lowerCAmelCase__      		=			logging.get_logger(__name__)






@add_end_docstrings(lowerCamelCase__   )
class     lowercase_			(lowerCamelCase__   ):




  """simple docstring"""




  def __init__(   self     :  Union[str, Any]      ,**lowercase__     :  Optional[Any]    ):
   super().__init__(**lowercase__    )

   if self.framework != "pt":
    raise ValueError(F"The {self.__class__} is only available in PyTorch."    )
   # No specific FOR_XXX available yet






  def __call__(   self     :  Optional[int]      ,lowercase__     :  Union[np.ndarray, bytes, str]      ,**lowercase__     :  int    ):
   return super().__call__(lowercase__      ,**lowercase__    )






  def       SCREAMING_SNAKE_CASE   (   self     :  str      ,**lowercase__     :  Tuple    ):
   __lowercase													=				{}
   if "candidate_labels" in kwargs:
    __lowercase													=				kwargs['''candidate_labels''']
   if "hypothesis_template" in kwargs:
    __lowercase													=				kwargs['''hypothesis_template''']

   return preprocess_params, {}, {}






  def       SCREAMING_SNAKE_CASE   (   self     :  Tuple      ,lowercase__     :  Optional[Any]      ,lowercase__     :  List[Any]=None      ,lowercase__     :  str="This is a sound of {}."    ):
   if isinstance(lowercase__      ,lowercase__    ):
    if audio.startswith('''http://'''    ) or audio.startswith('''https://'''    ):
     # We need to actually check for a real protocol, otherwise it's impossible to use a local file
     # like http_huggingface_co.png
     __lowercase													=				requests.get(lowercase__    ).content
    else:
     with open(lowercase__      ,'''rb'''    ) as f:
      __lowercase													=				f.read()

   if isinstance(lowercase__      ,lowercase__    ):
    __lowercase													=				ffmpeg_read(lowercase__      ,self.feature_extractor.sampling_rate    )

   if not isinstance(lowercase__      ,np.ndarray    ):
    raise ValueError('''We expect a numpy ndarray as input'''    )
   if len(audio.shape    ) != 1:
    raise ValueError('''We expect a single channel audio input for ZeroShotAudioClassificationPipeline'''    )

   __lowercase													=				self.feature_extractor(
       [audio]      ,sampling_rate=self.feature_extractor.sampling_rate      ,return_tensors='''pt'''    )
   __lowercase													=				candidate_labels
   __lowercase													=				[hypothesis_template.format(lowercase__    ) for x in candidate_labels]
   __lowercase													=				self.tokenizer(lowercase__      ,return_tensors=self.framework      ,padding=lowercase__    )
   __lowercase													=				[text_inputs]
   return inputs






  def       SCREAMING_SNAKE_CASE   (   self     :  List[Any]      ,lowercase__     :  Optional[int]    ):
   __lowercase													=				model_inputs.pop('''candidate_labels'''    )
   __lowercase													=				model_inputs.pop('''text_inputs'''    )
   if isinstance(text_inputs[0]      ,lowercase__    ):
    __lowercase													=				text_inputs[0]
   else:
    # Batching case.
    __lowercase													=				text_inputs[0][0]

   __lowercase													=				self.model(**lowercase__      ,**lowercase__    )

   __lowercase													=				{
       '''candidate_labels''': candidate_labels,
       '''logits''': outputs.logits_per_audio,
   }
   return model_outputs






  def       SCREAMING_SNAKE_CASE   (   self     :  Optional[Any]      ,lowercase__     :  Optional[Any]    ):
   __lowercase													=				model_outputs.pop('''candidate_labels'''    )
   __lowercase													=				model_outputs['''logits'''][0]

   if self.framework == "pt":
    __lowercase													=				logits.softmax(dim=0    )
    __lowercase													=				probs.tolist()
   else:
    raise ValueError('''`tf` framework not supported.'''    )

   __lowercase													=				[
       {'''score''': score, '''label''': candidate_label}
       for score, candidate_label in sorted(zip(lowercase__      ,lowercase__    )      ,key=lambda lowercase__    : -x[0]    )
   ]
   return result




'''simple docstring'''






import argparse
import json
import os

import torch
from transformers.file_utils import has_file

from diffusers import UNetaDConditionModel, UNetaDModel


lowerCAmelCase__      		=			False
lowerCAmelCase__      		=			True
lowerCAmelCase__      		=			False


if __name__ == "__main__":
 lowerCAmelCase__      		=			argparse.ArgumentParser()

 parser.add_argument(
     '''--repo_path''',
     default=None,
     type=str,
     required=True,
     help='''The config json file corresponding to the architecture.''',
 )

 parser.add_argument('''--dump_path''', default=None, type=str, required=True, help='''Path to the output model.''')

 lowerCAmelCase__      		=			parser.parse_args()

 lowerCAmelCase__      		=			{
     '''image_size''': '''sample_size''',
     '''num_res_blocks''': '''layers_per_block''',
     '''block_channels''': '''block_out_channels''',
     '''down_blocks''': '''down_block_types''',
     '''up_blocks''': '''up_block_types''',
     '''downscale_freq_shift''': '''freq_shift''',
     '''resnet_num_groups''': '''norm_num_groups''',
     '''resnet_act_fn''': '''act_fn''',
     '''resnet_eps''': '''norm_eps''',
     '''num_head_channels''': '''attention_head_dim''',
 }

 lowerCAmelCase__      		=			{
     '''time_steps''': '''time_proj''',
     '''mid''': '''mid_block''',
     '''downsample_blocks''': '''down_blocks''',
     '''upsample_blocks''': '''up_blocks''',
 }

 lowerCAmelCase__      		=			'''''' if has_file(args.repo_path, '''config.json''') else '''unet'''

 with open(os.path.join(args.repo_path, subfolder, '''config.json'''), '''r''', encoding='''utf-8''') as reader:
  lowerCAmelCase__      		=			reader.read()
  lowerCAmelCase__      		=			json.loads(text)

 if do_only_config:
  for key in config_parameters_to_change.keys():
   config.pop(key, None)

 if has_file(args.repo_path, '''config.json'''):
  lowerCAmelCase__      		=			UNetaDModel(**config)
 else:
  lowerCAmelCase__      		=			UNetaDConditionModel if '''ldm-text2im-large-256''' in args.repo_path else UNetaDModel
  lowerCAmelCase__      		=			class_name(**config)

 if do_only_config:
  model.save_config(os.path.join(args.repo_path, subfolder))

 lowerCAmelCase__      		=			dict(model.config)

 if do_only_renaming:
  for key, value in config_parameters_to_change.items():
   if key in config:
    lowerCAmelCase__      		=			config[key]
    del config[key]

  lowerCAmelCase__      		=			[k.replace('''UNetRes''', '''''') for k in config['''down_block_types''']]
  lowerCAmelCase__      		=			[k.replace('''UNetRes''', '''''') for k in config['''up_block_types''']]

 if do_only_weights:
  lowerCAmelCase__      		=			torch.load(os.path.join(args.repo_path, subfolder, '''diffusion_pytorch_model.bin'''))

  lowerCAmelCase__      		=			{}
  for param_key, param_value in state_dict.items():
   if param_key.endswith('''.op.bias''') or param_key.endswith('''.op.weight'''):
    continue
   lowerCAmelCase__      		=			False
   for key, new_key in key_parameters_to_change.items():
    if not has_changed and param_key.split('''.''')[0] == key:
     lowerCAmelCase__      		=			param_value
     lowerCAmelCase__      		=			True
   if not has_changed:
    lowerCAmelCase__      		=			param_value

  model.load_state_dict(new_state_dict)
  model.save_pretrained(os.path.join(args.repo_path, subfolder))

