



import textwrap

import pyarrow as pa
import pytest

from datasets import Features, Image
from datasets.packaged_modules.text.text import Text

from ..utils import require_pil





@pytest.fixture
def 						__lowerCamelCase       (      __magic_name__	:  Optional[int]						):
       a__:       str											=tmp_path / "text.txt"
       a__:       Dict											=textwrap.dedent(
           "\\n        Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.\n        Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat.\n        Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.\n        Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.\n\n        Second paragraph:\n        Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua.\n        Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat.\n        Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur.\n        Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.\n        "						)
       with open(__magic_name__     , "w"     , encoding="utf-8"						) as f:
              f.write(__magic_name__						)
       return str(__magic_name__						)





@pytest.fixture
def 						__lowerCamelCase       (      __magic_name__	:  List[Any]     , __magic_name__	:  List[Any]						):
       a__:       Union[str, Any]											=tmp_path / "text_with_image.txt"
       with open(__magic_name__     , "w"     , encoding="utf-8"						) as f:
              f.write(__magic_name__						)
       return str(__magic_name__						)





@pytest.mark.parametrize("keep_linebreaks"     , [True, False]						)
def 						__lowerCamelCase       (      __magic_name__	:  Optional[int]     , __magic_name__	:  Dict						):
       with open(__magic_name__     , encoding="utf-8"						) as f:
              a__:       Tuple											=f.read().splitlines(keepends=__magic_name__						)
       a__:       Any											=Text(keep_linebreaks=__magic_name__     , encoding="utf-8"						)
       a__:       Optional[Any]											=text._generate_tables([[text_file]]						)
       a__:       List[Any]											=pa.concat_tables([table for _, table in generator]						).to_pydict()["text"]
       assert generated_content == expected_content





@require_pil
def 						__lowerCamelCase       (      __magic_name__	:  Any						):
       with open(__magic_name__     , encoding="utf-8"						) as f:
              a__:       Dict											=f.read().splitlines()[0]
       a__:       str											=Text(encoding="utf-8"     , features=Features({"image": Image()}						)						)
       a__:       Any											=text._generate_tables([[text_file_with_image]]						)
       a__:       Tuple											=pa.concat_tables([table for _, table in generator]						)
       assert pa_table.schema.field("image"						).type == Image()()
       a__:       str											=pa_table.to_pydict()["image"]
       assert generated_content == [{"path": image_file, "bytes": None}]







@pytest.mark.parametrize("sample_by"     , ["line", "paragraph", "document"]						)
def 						__lowerCamelCase       (      __magic_name__	:  Tuple     , __magic_name__	:  Optional[int]						):
       with open(__magic_name__     , encoding="utf-8"						) as f:
              a__:       str											=f.read()
       if sample_by == "line":
              a__:       Union[str, Any]											=expected_content.splitlines()
       elif sample_by == "paragraph":
              a__:       Union[str, Any]											=expected_content.split("\n\n"						)
       elif sample_by == "document":
              a__:       Union[str, Any]											=[expected_content]
       a__:       Union[str, Any]											=Text(sample_by=__magic_name__     , encoding="utf-8"     , chunksize=100						)
       a__:       List[Any]											=text._generate_tables([[text_file]]						)
       a__:       Dict											=pa.concat_tables([table for _, table in generator]						).to_pydict()["text"]
       assert generated_content == expected_content










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

__UpperCAmelCase						  =     logging.get_logger(__name__)





def 						__lowerCamelCase       (      __magic_name__	:  bytes     , __magic_name__	:  int						):
							a__:       Optional[int]											=F"{sampling_rate}"
							a__:       Union[str, Any]											="1"
							a__:       List[str]											="f32le"
							a__:       Dict											=[
							    "ffmpeg",
							    "-i",
							    "pipe:0",
							    "-ac",
							    ac,
							    "-ar",
							    ar,
							    "-f",
							    format_for_conversion,
							    "-hide_banner",
							    "-loglevel",
							    "quiet",
							    "pipe:1",
							]

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

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







@add_end_docstrings(_a   )
class 	lowerCamelCase__ (		_a   ):


						def __init__( self						:  Union[str, Any]      ,       *_a						:  Dict      ,       **_a						:  Tuple			):
													# Default, might be overriden by the model.config.
													a__:       List[Any]											=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						:  Tuple      ,       _a						:  Union[np.ndarray, bytes, str]      ,       **_a						:  Union[str, Any]      ,       ):
													return super().__call__(_a      ,       **_a			)


						def   _lowerCamelCase			( self						:  int      ,       _a						:  Tuple=None      ,       **_a						:  int			):
													# No parameters on this pipeline right now
													a__:       Optional[int]											={}
													if top_k is not None:
																				if top_k > self.model.config.num_labels:
																											a__:       Union[str, Any]											=self.model.config.num_labels
																				a__:       Union[str, Any]											=top_k
													return {}, {}, postprocess_params


						def   _lowerCamelCase			( self						:  Any      ,       _a						:  Union[str, Any]			):
													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__:       List[str]											=requests.get(_a			).content
																				else:
																											with open(_a      ,       "rb"			) as f:
																																		a__:       Dict											=f.read()

													if isinstance(_a      ,       _a			):
																				a__:       Tuple											=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__:       Tuple											=inputs.pop("raw"      ,       _a			)
																				if _inputs is None:
																											# Remove path which will not be used from `datasets`.
																											inputs.pop("path"      ,       _a			)
																											a__:       int											=inputs.pop("array"      ,       _a			)
																				a__:       Optional[Any]											=inputs.pop("sampling_rate"			)
																				a__:       Optional[int]											=_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__:       Union[str, Any]											=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__:       Union[str, Any]											=self.feature_extractor(
													    _a      ,       sampling_rate=self.feature_extractor.sampling_rate      ,       return_tensors="pt"			)
													return processed


						def   _lowerCamelCase			( self						:  Dict      ,       _a						:  Tuple			):
													a__:       str											=self.model(**_a			)
													return model_outputs







						def   _lowerCamelCase			( self						:  Optional[Any]      ,       _a						:  Tuple      ,       _a						:  Tuple=5			):
													a__:       Tuple											=model_outputs.logits[0].softmax(-1			)
													a__							,			a__:       Any											=probs.topk(_a			)

													a__:       Union[str, Any]											=scores.tolist()
													a__:       Dict											=ids.tolist()

													a__:       Optional[int]											=[{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(_a      ,       _a			)]

													return labels





