



'''simple docstring'''




# Copyright 2021 The HuggingFace 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.

import argparse

from .config import config_command_parser
from .config_args import default_config_file, load_config_from_file  # noqa: F401
from .default import default_command_parser
from .update import update_command_parser



def 	a			( lowerCamelCase__=None    ):
			'''simple docstring'''





			A_	:						int	=     argparse.ArgumentParser(add_help=lowerCamelCase__						,      allow_abbrev=lowerCamelCase__    )
			# The main config parser
			A_	:						int	=     config_command_parser(lowerCamelCase__    )
			# The subparser to add commands to
			A_	:						int	=     config_parser.add_subparsers(title="""subcommands"""						,      dest="""subcommand"""    )

			# Then add other parsers with the parent parser
			default_command_parser(lowerCamelCase__						,      parents=[parent_parser]    )
			update_command_parser(lowerCamelCase__						,      parents=[parent_parser]    )

			return config_parser



def 	a			( ):
			'''simple docstring'''





			A_	:						Optional[int]	=     get_config_parser()
			A_	:						List[str]	=     config_parser.parse_args()

			if not hasattr(lowerCamelCase__						,      """func"""    ):
						config_parser.print_help()
						exit(1    )

			# Run
			args.func(lowerCamelCase__    )


if __name__ == "__main__":
						main()




'''simple docstring'''




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							:Union[str, Any]				    =					logging.get_logger(__name__)







@add_end_docstrings(__UpperCAmelCase )
class 	_lowerCAmelCase							(			__UpperCAmelCase ):


	def __init__(self       ,      **lowercase							):
				super().__init__(**lowercase							)

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


	def __call__(self       ,      lowercase       ,      **lowercase							):
				return super().__call__(lowercase       ,      **lowercase							)


	def     _a	(self       ,      **lowercase							):
				A_	:						Tuple	=     {}
				if "candidate_labels" in kwargs:
							A_	:						Dict	=     kwargs["""candidate_labels"""]
				if "hypothesis_template" in kwargs:
							A_	:						Optional[Any]	=     kwargs["""hypothesis_template"""]

				return preprocess_params, {}, {}


	def     _a	(self       ,      lowercase       ,      lowercase=None       ,      lowercase="This is a sound of {}."							):
				if isinstance(lowercase       ,      lowercase							):
							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(lowercase							).content
							else:
										with open(lowercase       ,      """rb"""							) as f:
													A_	:						List[str]	=     f.read()

				if isinstance(lowercase       ,      lowercase							):
							A_	:						List[Any]	=     ffmpeg_read(lowercase       ,      self.feature_extractor.sampling_rate							)

				if not isinstance(lowercase       ,      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_	:						int	=     self.feature_extractor(
				    [audio]       ,      sampling_rate=self.feature_extractor.sampling_rate       ,      return_tensors="""pt"""							)
				A_	:						List[Any]	=     candidate_labels
				A_	:						str	=     [hypothesis_template.format(lowercase							) for x in candidate_labels]
				A_	:						Optional[Any]	=     self.tokenizer(lowercase       ,      return_tensors=self.framework       ,      padding=lowercase							)
				A_	:						Optional[Any]	=     [text_inputs]
				return inputs


	def     _a	(self       ,      lowercase							):
				A_	:						Union[str, Any]	=     model_inputs.pop("""candidate_labels"""							)
				A_	:						List[Any]	=     model_inputs.pop("""text_inputs"""							)
				if isinstance(text_inputs[0]       ,      lowercase							):
							A_	:						Union[str, Any]	=     text_inputs[0]
				else:
							# Batching case.
							A_	:						Optional[int]	=     text_inputs[0][0]

				A_	:						str	=     self.model(**lowercase       ,      **lowercase							)

				A_	:						Union[str, Any]	=     {
				    """candidate_labels""": candidate_labels,
				    """logits""": outputs.logits_per_audio,
				}
				return model_outputs


	def     _a	(self       ,      lowercase							):
				A_	:						Union[str, Any]	=     model_outputs.pop("""candidate_labels"""							)
				A_	:						List[Any]	=     model_outputs["""logits"""][0]

				if self.framework == "pt":
							A_	:						Optional[Any]	=     logits.softmax(dim=0							)
							A_	:						str	=     probs.tolist()
				else:
							raise ValueError("""`tf` framework not supported."""							)

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