


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


_A			=						logging.get_logger(__name__)






@add_end_docstrings(A_		)
class    UpperCAmelCase__	( A_		):



    """simple docstring"""






    def __init__(      self ,       **A_   )  ->	Dict:
         super().__init__(**A_   )

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






    def __call__(      self ,       A_ ,       **A_   )  ->	Union[str, Any]:
         return super().__call__(A_ ,       **A_   )






    def _a (      self ,       **A_   )  ->	List[Any]:
         __UpperCamelCase   							={}
         if "candidate_labels" in kwargs:
              __UpperCamelCase   							=kwargs['candidate_labels']
         if "hypothesis_template" in kwargs:
              __UpperCamelCase   							=kwargs['hypothesis_template']

         return preprocess_params, {}, {}






    def _a (      self ,       A_ ,       A_=None ,       A_="This is a sound of {}."   )  ->	Any:
         if isinstance(A_ ,       A_   ):
              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
                   __UpperCamelCase   							=requests.get(A_   ).content
              else:
                   with open(A_ ,       'rb'   ) as f:
                        __UpperCamelCase   							=f.read()

         if isinstance(A_ ,       A_   ):
              __UpperCamelCase   							=ffmpeg_read(A_ ,       self.feature_extractor.sampling_rate   )

         if not isinstance(A_ ,       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'   )

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






    def _a (      self ,       A_   )  ->	int:
         __UpperCamelCase   							=model_inputs.pop('candidate_labels'   )
         __UpperCamelCase   							=model_inputs.pop('text_inputs'   )
         if isinstance(text_inputs[0] ,       A_   ):
              __UpperCamelCase   							=text_inputs[0]
         else:
              # Batching case.
              __UpperCamelCase   							=text_inputs[0][0]

         __UpperCamelCase   							=self.model(**A_ ,       **A_   )

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






    def _a (      self ,       A_   )  ->	Any:
         __UpperCamelCase   							=model_outputs.pop('candidate_labels'   )
         __UpperCamelCase   							=model_outputs['logits'][0]

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

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









import argparse

import torch

from transformers import BertForMaskedLM


if __name__ == "__main__":
   _A			=						argparse.ArgumentParser(
       description=(
           'Extraction some layers of the full BertForMaskedLM or RObertaForMaskedLM for Transfer Learned'
           ' Distillation'
       )
   )
   parser.add_argument('--model_type', default='bert', choices=['bert'])
   parser.add_argument('--model_name', default='bert-base-uncased', type=str)
   parser.add_argument('--dump_checkpoint', default='serialization_dir/tf_bert-base-uncased_0247911.pth', type=str)
   parser.add_argument('--vocab_transform', action='store_true')
   _A			=						parser.parse_args()

   if args.model_type == "bert":
      _A			=						BertForMaskedLM.from_pretrained(args.model_name)
      _A			=						'bert'
   else:
      raise ValueError('args.model_type should be "bert".')

   _A			=						model.state_dict()
   _A			=						{}

   for w in ["word_embeddings", "position_embeddings"]:
      _A			=						state_dict[f"""{prefix}.embeddings.{w}.weight"""]
   for w in ["weight", "bias"]:
      _A			=						state_dict[f"""{prefix}.embeddings.LayerNorm.{w}"""]

   _A			=						0
   for teacher_idx in [0, 2, 4, 7, 9, 11]:
      for w in ["weight", "bias"]:
         _A			=						state_dict[
             f"""{prefix}.encoder.layer.{teacher_idx}.attention.self.query.{w}"""
         ]
         _A			=						state_dict[
             f"""{prefix}.encoder.layer.{teacher_idx}.attention.self.key.{w}"""
         ]
         _A			=						state_dict[
             f"""{prefix}.encoder.layer.{teacher_idx}.attention.self.value.{w}"""
         ]

         _A			=						state_dict[
             f"""{prefix}.encoder.layer.{teacher_idx}.attention.output.dense.{w}"""
         ]
         _A			=						state_dict[
             f"""{prefix}.encoder.layer.{teacher_idx}.attention.output.LayerNorm.{w}"""
         ]

         _A			=						state_dict[
             f"""{prefix}.encoder.layer.{teacher_idx}.intermediate.dense.{w}"""
         ]
         _A			=						state_dict[
             f"""{prefix}.encoder.layer.{teacher_idx}.output.dense.{w}"""
         ]
         _A			=						state_dict[
             f"""{prefix}.encoder.layer.{teacher_idx}.output.LayerNorm.{w}"""
         ]
      std_idx += 1

   _A			=						state_dict['cls.predictions.decoder.weight']
   _A			=						state_dict['cls.predictions.bias']
   if args.vocab_transform:
      for w in ["weight", "bias"]:
         _A			=						state_dict[f"""cls.predictions.transform.dense.{w}"""]
         _A			=						state_dict[f"""cls.predictions.transform.LayerNorm.{w}"""]

   print(f"""N layers selected for distillation: {std_idx}""")
   print(f"""Number of params transferred for distillation: {len(compressed_sd.keys())}""")

   print(f"""Save transferred checkpoint to {args.dump_checkpoint}.""")
   torch.save(compressed_sd, args.dump_checkpoint)





