


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(a)
class      lowerCAmelCase__       (				a):




     '''simple docstring'''


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

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


     def __call__(						self ,						__lowerCamelCase ,						**__lowerCamelCase)			->				Any:
            return super().__call__(__lowerCamelCase ,						**__lowerCamelCase)


     def  _lowerCamelCase (						self ,						**__lowerCamelCase)			->				Any:
            _A      :       Dict				    = {}
            if "candidate_labels" in kwargs:
                   _A      :       Any				    = kwargs["candidate_labels"]
            if "hypothesis_template" in kwargs:
                   _A      :       Optional[int]				    = kwargs["hypothesis_template"]

            return preprocess_params, {}, {}


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

            if isinstance(__lowerCamelCase ,						__lowerCamelCase):
                   _A      :       Any				    = ffmpeg_read(__lowerCamelCase ,						self.feature_extractor.sampling_rate)

            if not isinstance(__lowerCamelCase ,						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")

            _A      :       Dict				    = self.feature_extractor(
                [audio] ,						sampling_rate=self.feature_extractor.sampling_rate ,						return_tensors="pt")
            _A      :       Optional[int]				    = candidate_labels
            _A      :       Dict				    = [hypothesis_template.format(__lowerCamelCase) for x in candidate_labels]
            _A      :       List[str]				    = self.tokenizer(__lowerCamelCase ,						return_tensors=self.framework ,						padding=__lowerCamelCase)
            _A      :       int				    = [text_inputs]
            return inputs


     def  _lowerCamelCase (						self ,						__lowerCamelCase)			->				int:
            _A      :       Union[str, Any]				    = model_inputs.pop("candidate_labels")
            _A      :       Optional[Any]				    = model_inputs.pop("text_inputs")
            if isinstance(text_inputs[0] ,						__lowerCamelCase):
                   _A      :       str				    = text_inputs[0]
            else:
                   # Batching case.
                   _A      :       Dict				    = text_inputs[0][0]

            _A      :       List[Any]				    = self.model(**__lowerCamelCase ,						**__lowerCamelCase)

            _A      :       str				    = {
                "candidate_labels": candidate_labels,
                "logits": outputs.logits_per_audio,
            }
            return model_outputs


     def  _lowerCamelCase (						self ,						__lowerCamelCase)			->				Any:
            _A      :       int				    = model_outputs.pop("candidate_labels")
            _A      :       Any				    = model_outputs["logits"][0]

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

            _A      :       Tuple				    = [
                {"score": score, "label": candidate_label}
                for score, candidate_label in sorted(zip(__lowerCamelCase ,						__lowerCamelCase) ,						key=lambda __lowerCamelCase: -x[0])
            ]
            return result




import pickle

import numpy as np
from matplotlib import pyplot as plt







class      lowerCAmelCase__       :




     '''simple docstring'''


     def __init__(						self ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase=0.2 ,						__lowerCamelCase=0.2)			->				str:
            _A      :       Optional[int]				    = bp_numa
            _A      :       Dict				    = bp_numa
            _A      :       Tuple				    = bp_numa
            _A      :       List[str]				    = conva_get[:2]
            _A      :       Tuple				    = conva_get[2]
            _A      :       Optional[int]				    = size_pa
            _A      :       Optional[Any]				    = rate_w
            _A      :       Optional[Any]				    = rate_t
            _A      :       Union[str, Any]				    = [
                np.mat(-1 * np.random.rand(self.conva[0] ,						self.conva[0]) + 0.5)
                for i in range(self.conva[1])
            ]
            _A      :       int				    = np.mat(-1 * np.random.rand(self.num_bpa ,						self.num_bpa) + 0.5)
            _A      :       Dict				    = np.mat(-1 * np.random.rand(self.num_bpa ,						self.num_bpa) + 0.5)
            _A      :       Any				    = -2 * np.random.rand(self.conva[1]) + 1
            _A      :       Optional[int]				    = -2 * np.random.rand(self.num_bpa) + 1
            _A      :       Optional[Any]				    = -2 * np.random.rand(self.num_bpa) + 1


     def  _lowerCamelCase (						self ,						__lowerCamelCase)			->				Dict:
            # save model dict with pickle
            _A      :       Dict				    = {
                "num_bp1": self.num_bpa,
                "num_bp2": self.num_bpa,
                "num_bp3": self.num_bpa,
                "conv1": self.conva,
                "step_conv1": self.step_conva,
                "size_pooling1": self.size_poolinga,
                "rate_weight": self.rate_weight,
                "rate_thre": self.rate_thre,
                "w_conv1": self.w_conva,
                "wkj": self.wkj,
                "vji": self.vji,
                "thre_conv1": self.thre_conva,
                "thre_bp2": self.thre_bpa,
                "thre_bp3": self.thre_bpa,
            }
            with open(__lowerCamelCase ,						"wb") as f:
                   pickle.dump(__lowerCamelCase ,						__lowerCamelCase)

            print(F"Model saved： {save_path}")


     @classmethod
     def  _lowerCamelCase (						cls ,						__lowerCamelCase)			->				Any:
            # read saved model
            with open(__lowerCamelCase ,						"rb") as f:
                   _A      :       Any				    = pickle.load(__lowerCamelCase)  # noqa: S301

            _A      :       Optional[int]				    = model_dic.get("conv1")
            conv_get.append(model_dic.get("step_conv1"))
            _A      :       str				    = model_dic.get("size_pooling1")
            _A      :       List[str]				    = model_dic.get("num_bp1")
            _A      :       Union[str, Any]				    = model_dic.get("num_bp2")
            _A      :       List[Any]				    = model_dic.get("num_bp3")
            _A      :       Dict				    = model_dic.get("rate_weight")
            _A      :       List[Any]				    = model_dic.get("rate_thre")
            # create model instance
            _A      :       str				    = CNN(__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase)
            # modify model parameter
            _A      :       List[Any]				    = model_dic.get("w_conv1")
            _A      :       Union[str, Any]				    = model_dic.get("wkj")
            _A      :       str				    = model_dic.get("vji")
            _A      :       List[str]				    = model_dic.get("thre_conv1")
            _A      :       Optional[Any]				    = model_dic.get("thre_bp2")
            _A      :       Dict				    = model_dic.get("thre_bp3")
            return conv_ins


     def  _lowerCamelCase (						self ,						__lowerCamelCase)			->				Dict:
            return 1 / (1 + np.exp(-1 * x))


     def  _lowerCamelCase (						self ,						__lowerCamelCase)			->				Union[str, Any]:
            return round(__lowerCamelCase ,						3)


     def  _lowerCamelCase (						self ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase)			->				Union[str, Any]:
            # convolution process
            _A      :       Tuple				    = convs[0]
            _A      :       Union[str, Any]				    = convs[1]
            _A      :       List[Any]				    = np.shape(__lowerCamelCase)[0]
            # get the data slice of original image data, data_focus
            _A      :       Tuple				    = []
            for i_focus in range(0 ,						size_data - size_conv + 1 ,						__lowerCamelCase):
                   for j_focus in range(0 ,						size_data - size_conv + 1 ,						__lowerCamelCase):
                          _A      :       Optional[int]				    = data[
                              i_focus : i_focus + size_conv, j_focus : j_focus + size_conv
                          ]
                          data_focus.append(__lowerCamelCase)
        # calculate the feature map of every single kernel, and saved as list of matrix
            _A      :       Optional[Any]				    = []
            _A      :       Optional[int]				    = int((size_data - size_conv) / conv_step + 1)
            for i_map in range(__lowerCamelCase):
                   _A      :       Optional[int]				    = []
                   for i_focus in range(len(__lowerCamelCase)):
                          _A      :       Any				    = (
                              np.sum(np.multiply(data_focus[i_focus] ,						w_convs[i_map]))
                              - thre_convs[i_map]
                          )
                          featuremap.append(self.sig(__lowerCamelCase))
                   _A      :       Optional[Any]				    = np.asmatrix(__lowerCamelCase).reshape(
                       __lowerCamelCase ,						__lowerCamelCase)
                   data_featuremap.append(__lowerCamelCase)

            # expanding the data slice to One dimenssion
            _A      :       Optional[Any]				    = []
            for each_focus in data_focus:
                   focusa_list.extend(self.Expand_Mat(__lowerCamelCase))
            _A      :       Dict				    = np.asarray(__lowerCamelCase)
            return focus_list, data_featuremap


     def  _lowerCamelCase (						self ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase="average_pool")			->				Dict:
            # pooling process
            _A      :       Optional[Any]				    = len(featuremaps[0])
            _A      :       str				    = int(size_map / size_pooling)
            _A      :       Optional[int]				    = []
            for i_map in range(len(__lowerCamelCase)):
                   _A      :       int				    = featuremaps[i_map]
                   _A      :       Optional[int]				    = []
                   for i_focus in range(0 ,						__lowerCamelCase ,						__lowerCamelCase):
                          for j_focus in range(0 ,						__lowerCamelCase ,						__lowerCamelCase):
                                 _A      :       str				    = feature_map[
                                     i_focus : i_focus + size_pooling,
                                     j_focus : j_focus + size_pooling,
                                 ]
                                 if pooling_type == "average_pool":
                                        # average pooling
                                        map_pooled.append(np.average(__lowerCamelCase))
                                 elif pooling_type == "max_pooling":
                                        # max pooling
                                        map_pooled.append(np.max(__lowerCamelCase))
                   _A      :       Tuple				    = np.asmatrix(__lowerCamelCase).reshape(__lowerCamelCase ,						__lowerCamelCase)
                   featuremap_pooled.append(__lowerCamelCase)
            return featuremap_pooled


     def  _lowerCamelCase (						self ,						__lowerCamelCase)			->				Tuple:
            # expanding three dimension data to one dimension list
            _A      :       Tuple				    = []
            for i in range(len(__lowerCamelCase)):
                   _A      :       Union[str, Any]				    = np.shape(data[i])
                   _A      :       List[Any]				    = data[i].reshape(1 ,						shapes[0] * shapes[1])
                   _A      :       Optional[Any]				    = data_listed.getA().tolist()[0]
                   data_expanded.extend(__lowerCamelCase)
            _A      :       Optional[Any]				    = np.asarray(__lowerCamelCase)
            return data_expanded


     def  _lowerCamelCase (						self ,						__lowerCamelCase)			->				Union[str, Any]:
            # expanding matrix to one dimension list
            _A      :       List[Any]				    = np.asarray(__lowerCamelCase)
            _A      :       Union[str, Any]				    = np.shape(__lowerCamelCase)
            _A      :       Dict				    = data_mat.reshape(1 ,						shapes[0] * shapes[1])
            return data_expanded


     def  _lowerCamelCase (						self ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase)			->				Optional[int]:
            _A      :       Dict				    = []
            _A      :       Any				    = 0
            for i_map in range(__lowerCamelCase):
                   _A      :       Union[str, Any]				    = np.ones((size_map, size_map))
                   for i in range(0 ,						__lowerCamelCase ,						__lowerCamelCase):
                          for j in range(0 ,						__lowerCamelCase ,						__lowerCamelCase):
                                 _A      :       List[Any]				    = pd_pool[
                                     i_pool
                                 ]
                                 _A      :       Tuple				    = i_pool + 1
                   _A      :       Optional[Any]				    = np.multiply(
                       __lowerCamelCase ,						np.multiply(out_map[i_map] ,						(1 - out_map[i_map])))
                   pd_all.append(__lowerCamelCase)
            return pd_all


     def  _lowerCamelCase (						self ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase ,						__lowerCamelCase=bool)			->				Union[str, Any]:
            # model traning
            print("----------------------Start Training-------------------------")
            print((" - - Shape: Train_Data  ", np.shape(__lowerCamelCase)))
            print((" - - Shape: Teach_Data  ", np.shape(__lowerCamelCase)))
            _A      :       Tuple				    = 0
            _A      :       Dict				    = []
            _A      :       Optional[Any]				    = 1_0_0_0_0
            while rp < n_repeat and mse >= error_accuracy:
                   _A      :       Union[str, Any]				    = 0
                   print(F"-------------Learning Time {rp}--------------")
                   for p in range(len(__lowerCamelCase)):
                          # print('------------Learning Image: %d--------------'%p)
                          _A      :       str				    = np.asmatrix(datas_train[p])
                          _A      :       Union[str, Any]				    = np.asarray(datas_teach[p])
                          _A    ,			_A      :       Any				    = self.convolute(
                              __lowerCamelCase ,						self.conva ,						self.w_conva ,						self.thre_conva ,						conv_step=self.step_conva ,						)
                          _A      :       Optional[Any]				    = self.pooling(__lowerCamelCase ,						self.size_poolinga)
                          _A      :       Optional[int]				    = np.shape(__lowerCamelCase)
                          _A      :       List[str]				    = self._expand(__lowerCamelCase)
                          _A      :       Tuple				    = data_bp_input

                          _A      :       int				    = np.dot(__lowerCamelCase ,						self.vji.T) - self.thre_bpa
                          _A      :       List[Any]				    = self.sig(__lowerCamelCase)
                          _A      :       Union[str, Any]				    = np.dot(__lowerCamelCase ,						self.wkj.T) - self.thre_bpa
                          _A      :       List[str]				    = self.sig(__lowerCamelCase)

                          # --------------Model Leaning ------------------------
                          # calculate error and gradient---------------
                          _A      :       int				    = np.multiply(
                              (data_teach - bp_outa) ,						np.multiply(__lowerCamelCase ,						(1 - bp_outa)))
                          _A      :       Optional[Any]				    = np.multiply(
                              np.dot(__lowerCamelCase ,						self.wkj) ,						np.multiply(__lowerCamelCase ,						(1 - bp_outa)))
                          _A      :       Union[str, Any]				    = np.dot(__lowerCamelCase ,						self.vji)

                          _A      :       Any				    = pd_i_all / (self.size_poolinga * self.size_poolinga)
                          _A      :       Dict				    = pd_conva_pooled.T.getA().tolist()
                          _A      :       Optional[Any]				    = self._calculate_gradient_from_pool(
                              __lowerCamelCase ,						__lowerCamelCase ,						shape_featuremapa[0] ,						shape_featuremapa[1] ,						self.size_poolinga ,						)
                          # weight and threshold learning process---------
                          # convolution layer
                          for k_conv in range(self.conva[1]):
                                 _A      :       int				    = self._expand_mat(pd_conva_all[k_conv])
                                 _A      :       Optional[int]				    = self.rate_weight * np.dot(__lowerCamelCase ,						__lowerCamelCase)

                                 _A      :       List[Any]				    = self.w_conva[k_conv] + delta_w.reshape(
                                     (self.conva[0], self.conva[0]))

                                 _A      :       Any				    = (
                                     self.thre_conva[k_conv]
                                     - np.sum(pd_conva_all[k_conv]) * self.rate_thre
                                 )
                          # all connected layer
                          _A      :       Tuple				    = self.wkj + pd_k_all.T * bp_outa * self.rate_weight
                          _A      :       int				    = self.vji + pd_j_all.T * bp_outa * self.rate_weight
                          _A      :       Tuple				    = self.thre_bpa - pd_k_all * self.rate_thre
                          _A      :       List[str]				    = self.thre_bpa - pd_j_all * self.rate_thre
                          # calculate the sum error of all single image
                          _A      :       Optional[int]				    = np.sum(abs(data_teach - bp_outa))
                          error_count += errors
                          # print('   ----Teach      ',data_teach)
                          # print('   ----BP_output  ',bp_out3)
                   _A      :       Any				    = rp + 1
                   _A      :       Dict				    = error_count / patterns
                   all_mse.append(__lowerCamelCase)

            def draw_error():
                   _A      :       Optional[int]				    = [error_accuracy for i in range(int(n_repeat * 1.2))]
                   plt.plot(__lowerCamelCase ,						"+-")
                   plt.plot(__lowerCamelCase ,						"r--")
                   plt.xlabel("Learning Times")
                   plt.ylabel("All_mse")
                   plt.grid(__lowerCamelCase ,						alpha=0.5)
                   plt.show()

            print("------------------Training Complished---------------------")
            print((" - - Training epoch: ", rp, F"     - - Mse: {mse:.6f}"))
            if draw_e:
                   draw_error()
            return mse


     def  _lowerCamelCase (						self ,						__lowerCamelCase)			->				int:
            # model predict
            _A      :       Union[str, Any]				    = []
            print("-------------------Start Testing-------------------------")
            print((" - - Shape: Test_Data  ", np.shape(__lowerCamelCase)))
            for p in range(len(__lowerCamelCase)):
                   _A      :       int				    = np.asmatrix(datas_test[p])
                   _A    ,			_A      :       List[Any]				    = self.convolute(
                       __lowerCamelCase ,						self.conva ,						self.w_conva ,						self.thre_conva ,						conv_step=self.step_conva ,						)
                   _A      :       str				    = self.pooling(__lowerCamelCase ,						self.size_poolinga)
                   _A      :       Optional[int]				    = self._expand(__lowerCamelCase)

                   _A      :       List[Any]				    = data_bp_input
                   _A      :       Optional[int]				    = bp_outa * self.vji.T - self.thre_bpa
                   _A      :       int				    = self.sig(__lowerCamelCase)
                   _A      :       int				    = bp_outa * self.wkj.T - self.thre_bpa
                   _A      :       Optional[int]				    = self.sig(__lowerCamelCase)
                   produce_out.extend(bp_outa.getA().tolist())
            _A      :       int				    = [list(map(self.do_round ,						__lowerCamelCase)) for each in produce_out]
            return np.asarray(__lowerCamelCase)


     def  _lowerCamelCase (						self ,						__lowerCamelCase)			->				Dict:
            # return the data of image after convoluting process so we can check it out
            _A      :       Optional[int]				    = np.asmatrix(__lowerCamelCase)
            _A    ,			_A      :       Tuple				    = self.convolute(
                __lowerCamelCase ,						self.conva ,						self.w_conva ,						self.thre_conva ,						conv_step=self.step_conva ,						)
            _A      :       Union[str, Any]				    = self.pooling(__lowerCamelCase ,						self.size_poolinga)

            return data_conveda, data_pooleda


if __name__ == "__main__":
    pass
