



import subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_torch_available():
				from ..models.auto.modeling_auto import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

_lowercase:				Tuple       =	logging.get_logger(__name__)






def 					a(   A :					bytes				,  A :					int )	-> List[Any]:
					"""simple docstring"""





					a        =     f'''{sampling_rate}'''
					a        =     "1"
					a        =     "f32le"
					a        =     [
					    "ffmpeg",
					    "-i",
					    "pipe:0",
					    "-ac",
					    ac,
					    "-ar",
					    ar,
					    "-f",
					    format_for_conversion,
					    "-hide_banner",
					    "-loglevel",
					    "quiet",
					    "pipe:1",
					]

					try:
										a        =     subprocess.Popen(lowercase__				,  stdin=subprocess.PIPE				,  stdout=subprocess.PIPE )
					except FileNotFoundError:
										raise ValueError("ffmpeg was not found but is required to load audio files from filename" )
					a        =     ffmpeg_process.communicate(lowercase__ )
					a        =     output_stream[0]

					a        =     np.frombuffer(lowercase__				,  np.floataa )
					if audio.shape[0] == 0:
										raise ValueError("Malformed soundfile" )
					return audio







@add_end_docstrings(lowerCAmelCase			)
class 			_lowercase							(     lowerCAmelCase			):





	"""simple docstring"""




	def __init__(self	, *lowerCamelCase_	, **lowerCamelCase_	):
						"""simple docstring"""

						a        =     5
						super().__init__(*A_	, **A_	)

						if self.framework != "pt":
											raise ValueError(F'''The {self.__class__} is only available in PyTorch.'''	)

						self.check_model_type(A_	)




	def __call__(self	, lowerCamelCase_	, **lowerCamelCase_	, ):
						"""simple docstring"""

						return super().__call__(A_	, **A_	)




	def   UpperCamelCase_ (self	, lowerCamelCase_=None	, **lowerCamelCase_	):
						"""simple docstring"""

						a        =     {}
						if top_k is not None:
											if top_k > self.model.config.num_labels:
																a        =     self.model.config.num_labels
											a        =     top_k
						return {}, {}, postprocess_params




	def   UpperCamelCase_ (self	, lowerCamelCase_	):
						"""simple docstring"""

						if isinstance(A_	, A_	):
											if inputs.startswith("http://"	) or inputs.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        =     requests.get(A_	).content
											else:
																with open(A_	, "rb"	) as f:
																					a        =     f.read()

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

						if isinstance(A_	, A_	):
											# Accepting `"array"` which is the key defined in `datasets` for
											# better integration
											if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
																raise ValueError(
																    "When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a "
																    "\"raw\" key containing the numpy array representing the audio and a \"sampling_rate\" key, "
																    "containing the sampling_rate associated with that array"	)

											a        =     inputs.pop("raw"	, A_	)
											if _inputs is None:
																# Remove path which will not be used from `datasets`.
																inputs.pop("path"	, A_	)
																a        =     inputs.pop("array"	, A_	)
											a        =     inputs.pop("sampling_rate"	)
											a        =     _inputs
											if in_sampling_rate != self.feature_extractor.sampling_rate:
																import torch

																if is_torchaudio_available():
																					from torchaudio import functional as F
																else:
																					raise ImportError(
																					    "torchaudio is required to resample audio samples in AudioClassificationPipeline. "
																					    "The torchaudio package can be installed through: `pip install torchaudio`."	)

																a        =     F.resample(
																    torch.from_numpy(A_	)	, A_	, self.feature_extractor.sampling_rate	).numpy()

						if not isinstance(A_	, np.ndarray	):
											raise ValueError("We expect a numpy ndarray as input"	)
						if len(inputs.shape	) != 1:
											raise ValueError("We expect a single channel audio input for AudioClassificationPipeline"	)

						a        =     self.feature_extractor(
						    A_	, sampling_rate=self.feature_extractor.sampling_rate	, return_tensors="pt"	)
						return processed




	def   UpperCamelCase_ (self	, lowerCamelCase_	):
						"""simple docstring"""

						a        =     self.model(**A_	)
						return model_outputs




	def   UpperCamelCase_ (self	, lowerCamelCase_	, lowerCamelCase_=5	):
						"""simple docstring"""

						a        =     model_outputs.logits[0].softmax(-1	)
						a  ,       a        =     probs.topk(A_	)

						a        =     scores.tolist()
						a        =     ids.tolist()

						a        =     [{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(A_	, A_	)]

						return labels





import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device


if is_torch_available():
				import torch

				from transformers import XLMProphetNetForConditionalGeneration, XLMProphetNetTokenizer



@require_torch
class 			_lowercase							(     unittest.TestCase			):





	"""simple docstring"""




	@slow
	def   UpperCamelCase_ (self	):
						"""simple docstring"""

						a        =     XLMProphetNetForConditionalGeneration.from_pretrained("microsoft/xprophetnet-large-wiki100-cased"	)
						model.to(lowerCamelCase_	)

						# encoder-decoder outputs
						a        =     torch.tensor([[17, 96208, 103471, 2]]	).to(lowerCamelCase_	)
						a        =     torch.tensor(
						    [[2, 250, 9953, 34, 69489, 1620, 32, 118424, 624, 210, 105, 2913, 1032, 351]]	).to(lowerCamelCase_	)
						a        =     model(
						    input_ids=lowerCamelCase_	, attention_mask=lowerCamelCase_	, encoder_outputs=lowerCamelCase_	, decoder_input_ids=lowerCamelCase_	)
						a        =     output[0]
						a        =     torch.Size((1, 14, 250012)	)
						self.assertEqual(output_predited_logis.shape	, lowerCamelCase_	)
						a        =     torch.tensor(
						    [[[-6.3986, -8.2391, 12.5189], [-6.3289, -8.0864, 12.6211], [-6.2418, -8.0445, 12.7968]]]	).to(lowerCamelCase_	)
						self.assertTrue(torch.allclose(output_predited_logis[:, :3, :3]	, lowerCamelCase_	, atol=1E-4	)	)

						# encoder outputs
						a        =     model.prophetnet.encoder(lowerCamelCase_	)[0]
						a        =     torch.tensor(
						    [[[-1.4260, -0.7628, 0.8453], [-1.4719, -0.1391, 0.7807], [-1.7678, 0.0114, 0.4646]]]	).to(lowerCamelCase_	)
						a        =     torch.Size((1, 4, 1024)	)
						self.assertEqual(encoder_outputs.shape	, lowerCamelCase_	)
						self.assertTrue(torch.allclose(encoder_outputs[:, :3, :3]	, lowerCamelCase_	, atol=1E-4	)	)

						# decoder outputs
						a        =     model.prophetnet.decoder(
						    lowerCamelCase_	, encoder_hidden_states=lowerCamelCase_	, )
						a        =     decoder_outputs[1].view(1	, model.config.ngram	, 14	, -1	)
						a        =     model.lm_head(lowerCamelCase_	)
						a        =     predicting_streams_logits[:, 0]
						self.assertTrue(torch.allclose(next_first_stream_logits[:, :3, :3]	, lowerCamelCase_	, atol=1E-4	)	)




	@slow
	def   UpperCamelCase_ (self	):
						"""simple docstring"""

						a        =     XLMProphetNetForConditionalGeneration.from_pretrained(
						    "microsoft/xprophetnet-large-wiki100-cased-xglue-ntg"	)
						model.to(lowerCamelCase_	)

						a        =     torch.tensor([[17, 96208, 103471, 2]]	).to(lowerCamelCase_	)
						a        =     torch.tensor(
						    [[2, 250, 9953, 34, 69489, 1620, 32, 118424, 624, 210, 105, 2913, 1032, 351]]	).to(lowerCamelCase_	)
						a        =     model(
						    input_ids=lowerCamelCase_	, attention_mask=lowerCamelCase_	, encoder_outputs=lowerCamelCase_	, decoder_input_ids=lowerCamelCase_	)
						a        =     output[0]
						a        =     torch.Size((1, 14, 250012)	)
						self.assertEqual(output_predited_logis.shape	, lowerCamelCase_	)
						# compare the actual values for a slice.
						a        =     torch.tensor(
						    [[[-9.2253, -9.7173, -6.3529], [-7.6701, -9.0145, -1.9382], [-8.0195, -7.0004, -0.1523]]]	).to(lowerCamelCase_	)

						self.assertTrue(torch.allclose(output_predited_logis[:, :3, :3]	, lowerCamelCase_	, atol=1E-4	)	)




	@slow
	def   UpperCamelCase_ (self	):
						"""simple docstring"""

						a        =     XLMProphetNetForConditionalGeneration.from_pretrained(
						    "microsoft/xprophetnet-large-wiki100-cased-xglue-ntg"	)
						model.to(lowerCamelCase_	)
						a        =     512

						a        =     XLMProphetNetTokenizer.from_pretrained("microsoft/xprophetnet-large-wiki100-cased-xglue-ntg"	)

						a        =     (
						    "Microsoft Corporation intends to officially end free support for the Windows 7 operating system after"
						    " January 14, 2020, according to the official portal of the organization. From that day, users of this"
						    " system will not be able to receive security updates, which could make their computers vulnerable to"
						    " cyber attacks."
						)
						a        =     (
						    "орпорация Microsoft намерена официально прекратить бесплатную поддержку операционной системы Windows 7"
						    " после 14 января 2020 года, сообщается на официальном портале организации . С указанного дня пользователи"
						    " этой системы не смогут получать обновления безопасности, из-за чего их компьютеры могут стать уязвимыми"
						    " к кибератакам."
						)
						a        =     (
						    "根据该组织的官方门户网站，微软公司打算在2020年1月14日之后正式终止对Windows 7操作系统的免费支持。从那时起，该系统的用户将无法接收安全更新，这可能会使他们的计算机容易受到网络攻击。"
						)

						a        =     tokenizer(
						    [EN_SENTENCE, RU_SENTENCE, ZH_SENTENCE]	, padding=lowerCamelCase_	, max_length=255	, return_tensors="pt"	).input_ids
						a        =     input_ids.to(lowerCamelCase_	)

						a        =     model.generate(
						    lowerCamelCase_	, num_beams=10	, length_penalty=1.0	, no_repeat_ngram_size=3	, early_stopping=lowerCamelCase_	)
						a        =     [tokenizer.decode(lowerCamelCase_	, skip_special_tokens=lowerCamelCase_	) for g in summary_ids]
						a        =     "Microsoft to end Windows 7 free support after January 14, 2020"
						a        =     "Microsoft намерена прекратить бесплатную поддержку Windows 7 после 14 января 2020 года"
						a        =     "微软打算终止对Windows 7操作系统的免费支持"
						self.assertListEqual(
						    [EXPECTED_TITLE_EN, EXPECTED_TITLE_RU, EXPECTED_TITLE_ZH]	, lowerCamelCase_	, )

						a        =     model.generate(
						    lowerCamelCase_	, num_beams=1	, length_penalty=1.0	, no_repeat_ngram_size=3	, early_stopping=lowerCamelCase_	)
						a        =     [
						    tokenizer.convert_ids_to_tokens(lowerCamelCase_	, skip_special_tokens=lowerCamelCase_	) for g in summary_ids_beama
						]
						a        =     "▁Microsoft ▁to ▁end ▁free ▁support ▁for ▁Windows ▁7".split(" "	)
						a        =     "▁Microsoft ▁намерен а ▁прекрати ть ▁бес плат ную ▁поддержку ▁Windows ▁7 ▁после ▁14 ▁января ▁2020 ▁года".split(
						    " "	)
						a        =     "微软 公司 打算 终止 对 Windows ▁7 操作 系统的 免费 支持".split(" "	)
						self.assertListEqual(
						    [EXPECTED_TITLE_EN_BEAM1_TOK, EXPECTED_TITLE_RU_BEAM1_TOK, EXPECTED_TITLE_ZH_BEAM1_TOK]	, lowerCamelCase_	, )
