






import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
     from .features import FeatureType


UpperCamelCase       ,							UpperCamelCase       ,							UpperCamelCase							       =    False, False, False




@dataclass
class    lowercase_    :
      A__	:		Optional[int]       			=				None
      A__	:		bool       			=				True
      A__	:		bool       			=				True
      A__	:		Optional[str]       			=				None
      # Automatically constructed
      A__	:		ClassVar[str]       			=				"dict"
      A__	:		ClassVar[Any]       			=				pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}     )
      A__	:		str       			=				field(default='''Audio''',    init=_UpperCAmelCase,    repr=_UpperCAmelCase     )






      def __call__(    self	)	->List[Any]:







        '''simple docstring'''


        return self.pa_type






      def 			lowerCamelCase__    (    self     ,     a_	)	->dict:







        '''simple docstring'''


        try:
          import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
        except ImportError as err:
          raise ImportError("To support encoding audio data, please install \'soundfile\'."	) from err
        if isinstance(a_     ,     a_	):
          return {"bytes": None, "path": value}
        elif isinstance(a_     ,     a_	):
          return {"bytes": value, "path": None}
        elif "array" in value:
          # convert the audio array to wav bytes
          _a							=	BytesIO()
          sf.write(a_     ,     value["array"]     ,     value["sampling_rate"]     ,     format="wav"	)
          return {"bytes": buffer.getvalue(), "path": None}
        elif value.get("path"	) is not None and os.path.isfile(value["path"]	):
          # we set "bytes": None to not duplicate the data if they're already available locally
          if value["path"].endswith("pcm"	):
            # "PCM" only has raw audio bytes
            if value.get("sampling_rate"	) is None:
              # At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
              raise KeyError("To use PCM files, please specify a \'sampling_rate\' in Audio object"	)
            if value.get("bytes"	):
              # If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
              _a							=	np.frombuffer(value["bytes"]     ,     dtype=np.intaa	).astype(np.floataa	) / 3_2_7_6_7
            else:
              _a							=	np.memmap(value["path"]     ,     dtype="h"     ,     mode="r"	).astype(np.floataa	) / 3_2_7_6_7

            _a							=	BytesIO(bytes()	)
            sf.write(a_     ,     a_     ,     value["sampling_rate"]     ,     format="wav"	)
            return {"bytes": buffer.getvalue(), "path": None}
          else:
            return {"bytes": None, "path": value.get("path"	)}
        elif value.get("bytes"	) is not None or value.get("path"	) is not None:
          # store the audio bytes, and path is used to infer the audio format using the file extension
          return {"bytes": value.get("bytes"	), "path": value.get("path"	)}
        else:
          raise ValueError(
              f'''An audio sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.'''	)






      def 			lowerCamelCase__    (    self     ,     a_     ,     a_ = None	)	->dict:







        '''simple docstring'''


        if not self.decode:
          raise RuntimeError("Decoding is disabled for this feature. Please use Audio(decode=True) instead."	)

        _a			,	_a							=	(value["path"], BytesIO(value["bytes"]	)) if value["bytes"] is not None else (value["path"], None)
        if path is None and file is None:
          raise ValueError(f'''An audio sample should have one of \'path\' or \'bytes\' but both are None in {value}.'''	)

        try:
          import librosa
          import soundfile as sf
        except ImportError as err:
          raise ImportError("To support decoding audio files, please install \'librosa\' and \'soundfile\'."	) from err

        _a							=	xsplitext(a_	)[1][1:].lower() if path is not None else None
        if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
          raise RuntimeError(
              "Decoding \'opus\' files requires system library \'libsndfile\'>=1.0.31, "
              "You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. "	)
        elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
          raise RuntimeError(
              "Decoding \'mp3\' files requires system library \'libsndfile\'>=1.1.0, "
              "You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. "	)

        if file is None:
          _a							=	token_per_repo_id or {}
          _a							=	path.split("::"	)[-1]
          try:
            _a							=	string_to_dict(a_     ,     config.HUB_DATASETS_URL	)["repo_id"]
            _a							=	token_per_repo_id[repo_id]
          except (ValueError, KeyError):
            _a							=	None

          with xopen(a_     ,     "rb"     ,     use_auth_token=a_	) as f:
            _a			,	_a							=	sf.read(a_	)

        else:
          _a			,	_a							=	sf.read(a_	)

        _a							=	array.T
        if self.mono:
          _a							=	librosa.to_mono(a_	)
        if self.sampling_rate and self.sampling_rate != sampling_rate:
          _a							=	librosa.resample(a_     ,     orig_sr=a_     ,     target_sr=self.sampling_rate	)
          _a							=	self.sampling_rate

        return {"path": path, "array": array, "sampling_rate": sampling_rate}






      def 			lowerCamelCase__    (    self	)	->Union["FeatureType", Dict[str, "FeatureType"]]:







        '''simple docstring'''


        from .features import Value

        if self.decode:
          raise ValueError("Cannot flatten a decoded Audio feature."	)
        return {
            "bytes": Value("binary"	),
            "path": Value("string"	),
        }






      def 			lowerCamelCase__    (    self     ,     a_	)	->pa.StructArray:







        '''simple docstring'''


        if pa.types.is_string(storage.type	):
          _a							=	pa.array([None] * len(a_	)     ,     type=pa.binary()	)
          _a							=	pa.StructArray.from_arrays([bytes_array, storage]     ,     ["bytes", "path"]     ,     mask=storage.is_null()	)
        elif pa.types.is_binary(storage.type	):
          _a							=	pa.array([None] * len(a_	)     ,     type=pa.string()	)
          _a							=	pa.StructArray.from_arrays([storage, path_array]     ,     ["bytes", "path"]     ,     mask=storage.is_null()	)
        elif pa.types.is_struct(storage.type	) and storage.type.get_all_field_indices("array"	):
          _a							=	pa.array([Audio().encode_example(a_	) if x is not None else None for x in storage.to_pylist()]	)
        elif pa.types.is_struct(storage.type	):
          if storage.type.get_field_index("bytes"	) >= 0:
            _a							=	storage.field("bytes"	)
          else:
            _a							=	pa.array([None] * len(a_	)     ,     type=pa.binary()	)
          if storage.type.get_field_index("path"	) >= 0:
            _a							=	storage.field("path"	)
          else:
            _a							=	pa.array([None] * len(a_	)     ,     type=pa.string()	)
          _a							=	pa.StructArray.from_arrays([bytes_array, path_array]     ,     ["bytes", "path"]     ,     mask=storage.is_null()	)
        return array_cast(a_     ,     self.pa_type	)






      def 			lowerCamelCase__    (    self     ,     a_	)	->pa.StructArray:







        '''simple docstring'''


        @no_op_if_value_is_null
        def path_to_bytes(a_	):
          with xopen(a_     ,     "rb"	) as f:
            _a							=	f.read()
          return bytes_

        _a							=	pa.array(
            [
                (path_to_bytes(x["path"]	) if x["bytes"] is None else x["bytes"]) if x is not None else None
                for x in storage.to_pylist()
            ]     ,     type=pa.binary()     ,     )
        _a							=	pa.array(
            [os.path.basename(a_	) if path is not None else None for path in storage.field("path"	).to_pylist()]     ,     type=pa.string()     ,     )
        _a							=	pa.StructArray.from_arrays([bytes_array, path_array]     ,     ["bytes", "path"]     ,     mask=bytes_array.is_null()	)
        return array_cast(a_     ,     self.pa_type	)










"""simple docstring"""




# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING

from ..models.auto import AutoModelForVisionaSeq
from ..utils import requires_backends
from .base import PipelineTool


if TYPE_CHECKING:
     from PIL import Image




class    lowercase_    (_UpperCAmelCase     ):
      A__	:		str       			=				'''Salesforce/blip-image-captioning-base'''
      A__	:		Optional[int]       			=				(
          '''This is a tool that generates a description of an image. It takes an input named `image` which should be the '''
          '''image to caption, and returns a text that contains the description in English.'''
      )
      A__	:		str       			=				'''image_captioner'''
      A__	:		Tuple       			=				AutoModelForVisionaSeq

      A__	:		Tuple       			=				['''image''']
      A__	:		List[Any]       			=				['''text''']






      def __init__(    self     ,     *a_     ,     **a_	)	->Union[str, Any]:







        '''simple docstring'''


        requires_backends(self     ,     ["vision"]	)
        super().__init__(*a_     ,     **a_	)






      def 			lowerCamelCase__    (    self     ,     a_	)	->Union[str, Any]:







        '''simple docstring'''


        return self.pre_processor(images=a_     ,     return_tensors="pt"	)






      def 			lowerCamelCase__    (    self     ,     a_	)	->List[Any]:







        '''simple docstring'''


        return self.model.generate(**a_	)






      def 			lowerCamelCase__    (    self     ,     a_	)	->Tuple:







        '''simple docstring'''


        return self.pre_processor.batch_decode(a_     ,     skip_special_tokens=a_	)[0].strip()


