





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   : Any			=				logging.get_logger(__name__)


@add_end_docstrings(SCREAMING_SNAKE_CASE   )
class   A      (SCREAMING_SNAKE_CASE   ):
			'''simple docstring'''




			def __init__(      self       :   Union[str, Any]							,     **__lowerCAmelCase       :   List[Any]       )     ->     Optional[Any]:







					"""simple docstring"""






					super().__init__(**__lowerCAmelCase       )

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






			def __call__(      self       :   Optional[Any]							,     __lowerCAmelCase       :   Union[np.ndarray, bytes, str]							,     **__lowerCAmelCase       :   Dict       )     ->     Dict:







					"""simple docstring"""






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






			def        a_	(      self       :   List[str]							,     **__lowerCAmelCase       :   int       )     ->     Any:







					"""simple docstring"""






					A__  				=       {}
					if "candidate_labels" in kwargs:
							A__  				=       kwargs["""candidate_labels"""]
					if "hypothesis_template" in kwargs:
							A__  				=       kwargs["""hypothesis_template"""]

					return preprocess_params, {}, {}






			def        a_	(      self       :   List[str]							,     __lowerCAmelCase       :   Any							,     __lowerCAmelCase       :   Any=None							,     __lowerCAmelCase       :   Dict="This is a sound of {}."       )     ->     Tuple:







					"""simple docstring"""






					if isinstance(__lowerCAmelCase							,     __lowerCAmelCase       ):
							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__  				=       requests.get(__lowerCAmelCase       ).content
							else:
									with open(__lowerCAmelCase							,     """rb"""       ) as f:
											A__  				=       f.read()

					if isinstance(__lowerCAmelCase							,     __lowerCAmelCase       ):
							A__  				=       ffmpeg_read(__lowerCAmelCase							,     self.feature_extractor.sampling_rate       )

					if not isinstance(__lowerCAmelCase							,     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__  				=       self.feature_extractor(
					    [audio]							,     sampling_rate=self.feature_extractor.sampling_rate							,     return_tensors="""pt"""       )
					A__  				=       candidate_labels
					A__  				=       [hypothesis_template.format(__lowerCAmelCase       ) for x in candidate_labels]
					A__  				=       self.tokenizer(__lowerCAmelCase							,     return_tensors=self.framework							,     padding=__lowerCAmelCase       )
					A__  				=       [text_inputs]
					return inputs






			def        a_	(      self       :   Tuple							,     __lowerCAmelCase       :   Any       )     ->     Tuple:







					"""simple docstring"""






					A__  				=       model_inputs.pop("""candidate_labels"""       )
					A__  				=       model_inputs.pop("""text_inputs"""       )
					if isinstance(text_inputs[0]							,     __lowerCAmelCase       ):
							A__  				=       text_inputs[0]
					else:
							# Batching case.
							A__  				=       text_inputs[0][0]

					A__  				=       self.model(**__lowerCAmelCase							,     **__lowerCAmelCase       )

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







			def        a_	(      self       :   Tuple							,     __lowerCAmelCase       :   Any       )     ->     Optional[int]:







					"""simple docstring"""






					A__  				=       model_outputs.pop("""candidate_labels"""       )
					A__  				=       model_outputs["""logits"""][0]

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

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










import argparse
import math
import traceback

import dateutil.parser as date_parser
import requests







def        __lowerCamelCase (							__a       :str   )		->    Optional[int]:






		"""simple docstring"""

		A__  				=       {}

		A__  				=       job["""started_at"""]
		A__  				=       job["""completed_at"""]

		A__  				=       date_parser.parse(__a   )
		A__  				=       date_parser.parse(__a   )

		A__  				=       round((end_datetime - start_datetime).total_seconds() / 60.0   )

		A__  				=       start
		A__  				=       end
		A__  				=       duration_in_min

		return job_info







def        __lowerCamelCase (							__a       :Optional[Any]   ,					__a       :List[str]=None   )		->    List[Any]:






		"""simple docstring"""

		A__  				=       None
		if token is not None:
				A__  				=       {"""Accept""": """application/vnd.github+json""", """Authorization""": F'Bearer {token}'}

		A__  				=       F'https://api.github.com/repos/huggingface/transformers/actions/runs/{workflow_run_id}/jobs?per_page=100'
		A__  				=       requests.get(__a   ,					headers=__a   ).json()
		A__  				=       {}

		try:
				job_time.update({job["""name"""]: extract_time_from_single_job(__a   ) for job in result["""jobs"""]}   )
				A__  				=       math.ceil((result["""total_count"""] - 1_0_0) / 1_0_0   )

				for i in range(__a   ):
						A__  				=       requests.get(url + F'&page={i + 2}'   ,					headers=__a   ).json()
						job_time.update({job["""name"""]: extract_time_from_single_job(__a   ) for job in result["""jobs"""]}   )

				return job_time
		except Exception:
				print(F'Unknown error, could not fetch links:\n{traceback.format_exc()}'   )

		return {}


if __name__ == "__main__":
					A   : Optional[Any]			=				argparse.ArgumentParser()
					# Required parameters
					parser.add_argument('''--workflow_run_id''', type=str, required=True, help='''A GitHub Actions workflow run id.''')
					A   : Dict			=				parser.parse_args()

					A   : List[Any]			=				get_job_time(args.workflow_run_id)
					A   : int			=				dict(sorted(job_time.items(), key=lambda item: item[1]["duration"], reverse=True))

					for k, v in job_time.items():
										print(F'''{k}: {v["duration"]}''')



