


"""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


a__    :							Optional[Any]            =					logging.get_logger(__name__)




@add_end_docstrings(UpperCamelCase)
class 			UpperCamelCase_     (			UpperCamelCase):

	"""simple docstring"""





	def __init__(			self      :    Tuple      ,      **UpperCAmelCase__      :    List[Any]       )	->     Tuple:
						super().__init__(**UpperCAmelCase__       )

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




	def __call__(			self      :    int      ,      UpperCAmelCase__      :    Union[np.ndarray, bytes, str]      ,      **UpperCAmelCase__      :    Optional[int]       )	->     Tuple:
						return super().__call__(UpperCAmelCase__      ,      **UpperCAmelCase__       )




	def   UpperCAmelCase_				(			self      :    Any      ,      **UpperCAmelCase__      :    Tuple       )	->     Optional[int]:
						__SCREAMING_SNAKE_CASE						       =    {}
						if "candidate_labels" in kwargs:
											__SCREAMING_SNAKE_CASE						       =    kwargs["candidate_labels"]
						if "hypothesis_template" in kwargs:
											__SCREAMING_SNAKE_CASE						       =    kwargs["hypothesis_template"]

						return preprocess_params, {}, {}




	def   UpperCAmelCase_				(			self      :    int      ,      UpperCAmelCase__      :    Optional[Any]      ,      UpperCAmelCase__      :    int=None      ,      UpperCAmelCase__      :    int="This is a sound of {}."       )	->     List[str]:
						if isinstance(UpperCAmelCase__      ,      UpperCAmelCase__       ):
											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
																__SCREAMING_SNAKE_CASE						       =    requests.get(UpperCAmelCase__       ).content
											else:
																with open(UpperCAmelCase__      ,      "rb"       ) as f:
																					__SCREAMING_SNAKE_CASE						       =    f.read()

						if isinstance(UpperCAmelCase__      ,      UpperCAmelCase__       ):
											__SCREAMING_SNAKE_CASE						       =    ffmpeg_read(UpperCAmelCase__      ,      self.feature_extractor.sampling_rate       )

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

						__SCREAMING_SNAKE_CASE						       =    self.feature_extractor(
						    [audio]      ,      sampling_rate=self.feature_extractor.sampling_rate      ,      return_tensors="pt"       )
						__SCREAMING_SNAKE_CASE						       =    candidate_labels
						__SCREAMING_SNAKE_CASE						       =    [hypothesis_template.format(UpperCAmelCase__       ) for x in candidate_labels]
						__SCREAMING_SNAKE_CASE						       =    self.tokenizer(UpperCAmelCase__      ,      return_tensors=self.framework      ,      padding=UpperCAmelCase__       )
						__SCREAMING_SNAKE_CASE						       =    [text_inputs]
						return inputs




	def   UpperCAmelCase_				(			self      :    Any      ,      UpperCAmelCase__      :    Optional[int]       )	->     int:
						__SCREAMING_SNAKE_CASE						       =    model_inputs.pop("candidate_labels"       )
						__SCREAMING_SNAKE_CASE						       =    model_inputs.pop("text_inputs"       )
						if isinstance(text_inputs[0]      ,      UpperCAmelCase__       ):
											__SCREAMING_SNAKE_CASE						       =    text_inputs[0]
						else:
											# Batching case.
											__SCREAMING_SNAKE_CASE						       =    text_inputs[0][0]

						__SCREAMING_SNAKE_CASE						       =    self.model(**UpperCAmelCase__      ,      **UpperCAmelCase__       )

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




	def   UpperCAmelCase_				(			self      :    Any      ,      UpperCAmelCase__      :    List[Any]       )	->     List[str]:
						__SCREAMING_SNAKE_CASE						       =    model_outputs.pop("candidate_labels"       )
						__SCREAMING_SNAKE_CASE						       =    model_outputs["logits"][0]

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

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




"""simple docstring"""







from __future__ import annotations

import requests




def        UpperCAmelCase__	(lowerCAmelCase_       ):




					'''simple docstring'''


					__SCREAMING_SNAKE_CASE						       =    f"""https://hacker-news.firebaseio.com/v0/item/{story_id}.json?print=pretty"""
					return requests.get(lowerCAmelCase_       ).json()




def        UpperCAmelCase__	(lowerCAmelCase_ = 10       ):




					'''simple docstring'''


					__SCREAMING_SNAKE_CASE						       =    "https://hacker-news.firebaseio.com/v0/topstories.json?print=pretty"
					__SCREAMING_SNAKE_CASE						       =    requests.get(lowerCAmelCase_       ).json()[:max_stories]
					return [get_hackernews_story(lowerCAmelCase_       ) for story_id in story_ids]




def        UpperCAmelCase__	(lowerCAmelCase_ = 10       ):




					'''simple docstring'''


					__SCREAMING_SNAKE_CASE						       =    hackernews_top_stories(lowerCAmelCase_       )
					return "\n".join("* [{title}]({url})".format(**lowerCAmelCase_       ) for story in stories       )


if __name__ == "__main__":
		print(hackernews_top_stories_as_markdown())
