"""simple docstring"""







import collections
import json
import math
import os
import re
import time
from fnmatch import fnmatch
from typing import Dict

import requests
from slack_sdk import WebClient


SCREAMING_SNAKE_CASE									=						WebClient(token=os.environ['CI_SLACK_BOT_TOKEN'])


def      __lowerCAmelCase(		__UpperCAmelCase			):






	"""simple docstring"""





	_lowercase :    List[Any]							=							test_results.split(' '			)

	_lowercase :    Optional[Any]							=							0
	_lowercase :    Tuple							=							0

	# When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
	# When it is too long, those signs are not present.
	_lowercase :    Dict							=							expressions[-2] if '=' in expressions[-1] else expressions[-1]

	for i, expression in enumerate(__UpperCAmelCase			):
		if "failed" in expression:
			failed += int(expressions[i - 1]			)
		if "passed" in expression:
			success += int(expressions[i - 1]			)

	return failed, success, time_spent


def      __lowerCAmelCase(		__UpperCAmelCase			):






	"""simple docstring"""





	_lowercase :    Dict							=							{}
	_lowercase :    Dict							=							None
	_lowercase :    Optional[Any]							=							False
	for line in failures_short_lines.split('\n'			):
		if re.search(R'_ \[doctest\]'     ,__UpperCAmelCase			):
			_lowercase :    int							=							True
			_lowercase :    List[str]							=							line.split(' '			)[2]
		elif in_error and not line.split(' '			)[0].isdigit():
			_lowercase :    Any							=							line
			_lowercase :    List[str]							=							False

	return failures



class  _lowerCamelCase			:







				def __init__( self    :       Tuple			,					lowerCamelCase_    :       str			,					lowerCamelCase_    :       Dict						):

					"""simple docstring"""






					_lowercase :    Tuple							=							title

					_lowercase :    int							=							doc_test_results['time_spent'].split(','						)[0]
					_lowercase :    Tuple							=							doc_test_results['success']
					_lowercase :    int							=							doc_test_results['failures']
					_lowercase :    Optional[Any]							=							self.n_success + self.n_failures

					# Failures and success of the modeling tests
					_lowercase :    Optional[Any]							=							doc_test_results







				@property
				def 							__UpperCAmelCase					( self    :       Optional[int]						):

					"""simple docstring"""






					_lowercase :    Union[str, Any]							=							[self._time_spent]
					_lowercase :    Any							=							0

					for time in time_spent:
						_lowercase :    List[Any]							=							time.split(':'						)

						# Time can be formatted as xx:xx:xx, as .xx, or as x.xx if the time spent was less than a minute.
						if len(lowerCamelCase_						) == 1:
							_lowercase :    str							=							[0, 0, time_parts[0]]

						_lowercase				,      _lowercase				,      _lowercase :    Optional[Any]							=							int(time_parts[0]						), int(time_parts[1]						), float(time_parts[2]						)
						total_secs += hours * 3_6_0_0 + minutes * 6_0 + seconds

					_lowercase				,      _lowercase				,      _lowercase :    Dict							=							total_secs // 3_6_0_0, (total_secs % 3_6_0_0) // 6_0, total_secs % 6_0
					return F'''{int(lowerCamelCase_						)}h{int(lowerCamelCase_						)}m{int(lowerCamelCase_						)}s'''







				@property
				def 							__UpperCAmelCase					( self    :       int						):

					"""simple docstring"""






					return {"type": "header", "text": {"type": "plain_text", "text": self.title}}







				@property
				def 							__UpperCAmelCase					( self    :       Optional[Any]						):

					"""simple docstring"""






					return {
					    "type": "section",
					    "text": {
					        "type": "plain_text",
					        "text": F'''🌞 There were no failures: all {self.n_tests} tests passed. The suite ran in {self.time}.''',
					        "emoji": True,
					    },
					    "accessory": {
					        "type": "button",
					        "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
					        "url": F'''https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}''',
					    },
					}







				@property
				def 							__UpperCAmelCase					( self    :       Optional[Any]						):

					"""simple docstring"""






					return {
					    "type": "section",
					    "text": {
					        "type": "plain_text",
					        "text": (
					            F'''There were {self.n_failures} failures, out of {self.n_tests} tests.\nThe suite ran in'''
					            F''' {self.time}.'''
					        ),
					        "emoji": True,
					    },
					    "accessory": {
					        "type": "button",
					        "text": {"type": "plain_text", "text": "Check Action results", "emoji": True},
					        "url": F'''https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}''',
					    },
					}







				@property
				def 							__UpperCAmelCase					( self    :       Any						):

					"""simple docstring"""






					_lowercase :    int							=							4_0
					_lowercase :    Tuple							=							{k: v['failed'] for k, v in doc_test_results.items() if isinstance(lowerCamelCase_			,					lowerCamelCase_						)}

					_lowercase :    List[Any]							=							''
					for category, failures in category_failures.items():
						if len(lowerCamelCase_						) == 0:
							continue

						if report != "":
							report += "\n\n"

						report += F'''*{category} failures*:'''.ljust(line_length // 2						).rjust(line_length // 2						) + "\n"
						report += "`"
						report += "`\n`".join(lowerCamelCase_						)
						report += "`"

					return {
					    "type": "section",
					    "text": {
					        "type": "mrkdwn",
					        "text": F'''The following examples had failures:\n\n\n{report}\n''',
					    },
					}







				@property
				def 							__UpperCAmelCase					( self    :       List[Any]						):

					"""simple docstring"""






					_lowercase :    Optional[Any]							=							[self.header]

					if self.n_failures > 0:
						blocks.append(self.failures						)

					if self.n_failures > 0:
						blocks.extend([self.category_failures]						)

					if self.n_failures == 0:
						blocks.append(self.no_failures						)

					return json.dumps(lowerCamelCase_						)







				@staticmethod
				def 							__UpperCAmelCase					( ):

					"""simple docstring"""






					_lowercase :    int							=							[
					    {
					        'type': 'section',
					        'text': {
					            'type': 'plain_text',
					            'text': 'There was an issue running the tests.',
					        },
					        'accessory': {
					            'type': 'button',
					            'text': {'type': 'plain_text', 'text': 'Check Action results', 'emoji': True},
					            'url': F'''https://github.com/huggingface/transformers/actions/runs/{os.environ['GITHUB_RUN_ID']}''',
					        },
					    }
					]

					print('Sending the following payload'						)
					print(json.dumps({'blocks': json.loads(lowerCamelCase_						)}						)						)

					client.chat_postMessage(
					    channel=os.environ['CI_SLACK_CHANNEL_ID_DAILY']			,					text='There was an issue running the tests.'			,					blocks=lowerCamelCase_			,					)







				def 							__UpperCAmelCase					( self    :       int						):

					"""simple docstring"""






					print('Sending the following payload'						)
					print(json.dumps({'blocks': json.loads(self.payload						)}						)						)

					_lowercase :    Tuple							=							F'''{self.n_failures} failures out of {self.n_tests} tests,''' if self.n_failures else 'All tests passed.'

					_lowercase :    Union[str, Any]							=							client.chat_postMessage(
					    channel=os.environ['CI_SLACK_CHANNEL_ID_DAILY']			,					blocks=self.payload			,					text=lowerCamelCase_			,					)







				def 							__UpperCAmelCase					( self    :       Any			,					lowerCamelCase_    :       List[Any]			,					lowerCamelCase_    :       List[Any]			,					lowerCamelCase_    :       Optional[Any]			,					lowerCamelCase_    :       List[str]						):

					"""simple docstring"""






					_lowercase :    str							=							''
					for key, value in failures.items():
						_lowercase :    Tuple							=							value[:2_0_0] + ' [Truncated]' if len(lowerCamelCase_						) > 2_5_0 else value
						failures_text += F'''*{key}*\n_{value}_\n\n'''

					_lowercase :    Optional[Any]							=							job_name
					_lowercase :    Optional[int]							=							{'type': 'section', 'text': {'type': 'mrkdwn', 'text': text}}

					if job_link is not None:
						_lowercase :    Tuple							=							{
						    'type': 'button',
						    'text': {'type': 'plain_text', 'text': 'GitHub Action job', 'emoji': True},
						    'url': job_link,
						}

					return [
					    {"type": "header", "text": {"type": "plain_text", "text": title.upper(), "emoji": True}},
					    content,
					    {"type": "section", "text": {"type": "mrkdwn", "text": failures_text}},
					]







				def 							__UpperCAmelCase					( self    :       Tuple						):

					"""simple docstring"""






					if self.thread_ts is None:
						raise ValueError('Can only post reply if a post has been made.'						)

					_lowercase :    List[Any]							=							self.doc_test_results.pop('job_link'						)
					self.doc_test_results.pop('failures'						)
					self.doc_test_results.pop('success'						)
					self.doc_test_results.pop('time_spent'						)

					_lowercase :    Any							=							sorted(self.doc_test_results.items()			,					key=lambda lowerCamelCase_						: t[0]						)
					for job, job_result in sorted_dict:
						if len(job_result['failures']						):
							_lowercase :    Tuple							=							F'''*Num failures* :{len(job_result['failed']						)} \n'''
							_lowercase :    Optional[Any]							=							job_result['failures']
							_lowercase :    List[str]							=							self.get_reply_blocks(lowerCamelCase_			,					lowerCamelCase_			,					lowerCamelCase_			,					text=lowerCamelCase_						)

							print('Sending the following reply'						)
							print(json.dumps({'blocks': blocks}						)						)

							client.chat_postMessage(
							    channel=os.environ['CI_SLACK_CHANNEL_ID_DAILY']			,					text=F'''Results for {job}'''			,					blocks=lowerCamelCase_			,					thread_ts=self.thread_ts['ts']			,					)

							time.sleep(1						)



def      __lowerCAmelCase(		):






	"""simple docstring"""





	_lowercase :    List[str]							=							os.environ['GITHUB_RUN_ID']
	_lowercase :    Union[str, Any]							=							F'''https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100'''
	_lowercase :    Optional[Any]							=							requests.get(__UpperCAmelCase			).json()
	_lowercase :    Dict							=							{}

	try:
		jobs.update({job['name']: job['html_url'] for job in result['jobs']}			)
		_lowercase :    Dict							=							math.ceil((result['total_count'] - 100) / 100			)

		for i in range(__UpperCAmelCase			):
			_lowercase :    int							=							requests.get(url + F'''&page={i + 2}'''			).json()
			jobs.update({job['name']: job['html_url'] for job in result['jobs']}			)

		return jobs
	except Exception as e:
		print('Unknown error, could not fetch links.'     ,__UpperCAmelCase			)

	return {}


def      __lowerCAmelCase(		__UpperCAmelCase			):






	"""simple docstring"""





	_lowercase :    Tuple							=							{}

	if os.path.exists(__UpperCAmelCase			):
		_lowercase :    Optional[Any]							=							os.listdir(__UpperCAmelCase			)
		for file in files:
			try:
				with open(os.path.join(__UpperCAmelCase     ,__UpperCAmelCase			)     ,encoding='utf-8'			) as f:
					_lowercase :    Optional[Any]							=							f.read()
			except UnicodeDecodeError as e:
				raise ValueError(F'''Could not open {os.path.join(__UpperCAmelCase     ,__UpperCAmelCase			)}.'''			) from e

	return _artifact


def      __lowerCAmelCase(		):






	"""simple docstring"""





	class  _lowerCamelCase			:







					def __init__( self    :       List[str]			,					lowerCamelCase_    :       str						):

						"""simple docstring"""






						_lowercase :    Tuple							=							name
						_lowercase :    int							=							[]







					def __str__( self    :       List[str]						):

						"""simple docstring"""






						return self.name







					def 							__UpperCAmelCase					( self    :       int			,					lowerCamelCase_    :       str						):

						"""simple docstring"""






						self.paths.append({'name': self.name, 'path': path}						)



	_lowercase :    Dict[str, Artifact]							=							{}

	_lowercase :    Optional[Any]							=							filter(os.path.isdir     ,os.listdir()			)
	for directory in directories:
		_lowercase :    List[Any]							=							directory
		if artifact_name not in _available_artifacts:
			_lowercase :    Optional[Any]							=							Artifact(__UpperCAmelCase			)

			_available_artifacts[artifact_name].add_path(__UpperCAmelCase			)

	return _available_artifacts


if __name__ == "__main__":
						SCREAMING_SNAKE_CASE									=						get_job_links()
						SCREAMING_SNAKE_CASE									=						retrieve_available_artifacts()

						SCREAMING_SNAKE_CASE									=						collections.OrderedDict(
						    [
						        ('*.py', 'API Examples'),
						        ('*.md', 'MD Examples'),
						    ]
						)

						# This dict will contain all the information relative to each doc test category:
						# - failed: list of failed tests
						# - failures: dict in the format 'test': 'error_message'
						SCREAMING_SNAKE_CASE									=						{
						    v: {
						        'failed': [],
						        'failures': {},
						    }
						    for v in docs.values()
						}

						# Link to the GitHub Action job
						SCREAMING_SNAKE_CASE									=						github_actions_job_links.get('run_doctests')

						SCREAMING_SNAKE_CASE									=						available_artifacts['doc_tests_gpu_test_reports'].paths[0]
						SCREAMING_SNAKE_CASE									=						retrieve_artifact(artifact_path['name'])
						if "stats" in artifact:
												SCREAMING_SNAKE_CASE							,    SCREAMING_SNAKE_CASE							,    SCREAMING_SNAKE_CASE									=						handle_test_results(artifact['stats'])
												SCREAMING_SNAKE_CASE									=						failed
												SCREAMING_SNAKE_CASE									=						success
												SCREAMING_SNAKE_CASE									=						time_spent[1:-1] + ', '

												SCREAMING_SNAKE_CASE									=						extract_first_line_failure(artifact['failures_short'])
												for line in artifact["summary_short"].split('\n'):
																		if re.search('FAILED', line):
																								SCREAMING_SNAKE_CASE									=						line.replace('FAILED ', '')
																								SCREAMING_SNAKE_CASE									=						line.split()[0].replace('\n', '')

																								if "::" in line:
																														SCREAMING_SNAKE_CASE							,    SCREAMING_SNAKE_CASE									=						line.split('::')
																								else:
																														SCREAMING_SNAKE_CASE							,    SCREAMING_SNAKE_CASE									=						line, line

																								for file_regex in docs.keys():
																														if fnmatch(file_path, file_regex):
																																				SCREAMING_SNAKE_CASE									=						docs[file_regex]
																																				doc_test_results[category]["failed"].append(test)

																																				SCREAMING_SNAKE_CASE									=						all_failures[test] if test in all_failures else 'N/A'
																																				SCREAMING_SNAKE_CASE									=						failure
																																				break

						SCREAMING_SNAKE_CASE									=						Message('🤗 Results of the doc tests.', doc_test_results)
						message.post()
						message.post_reply()




"""simple docstring"""







from typing import List, Optional, Union

import numpy as np

from ....audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
from ....feature_extraction_sequence_utils import SequenceFeatureExtractor
from ....feature_extraction_utils import BatchFeature
from ....file_utils import PaddingStrategy, TensorType
from ....utils import logging


SCREAMING_SNAKE_CASE									=						logging.get_logger(__name__)



class  _lowerCamelCase			(__lowerCamelCase     ):
				_snake_case     =  ["input_features", "attention_mask"]







				def __init__( self    :       int			,					lowerCamelCase_    :       List[str]=8_0			,					lowerCamelCase_    :       Tuple=1_6_0_0_0			,					lowerCamelCase_    :       Union[str, Any]=0.0			,					lowerCamelCase_    :       List[Any]=1_0			,					lowerCamelCase_    :       List[str]=2_5			,					lowerCamelCase_    :       List[Any]="hamming_window"			,					lowerCamelCase_    :       Tuple=3_2768.0			,					lowerCamelCase_    :       int=0.97			,					lowerCamelCase_    :       Optional[int]=1.0			,					lowerCamelCase_    :       Optional[Any]=True			,					lowerCamelCase_    :       Union[str, Any]=True			,					lowerCamelCase_    :       Optional[int]=False			,					**lowerCamelCase_    :       Optional[Any]			,					):

					"""simple docstring"""






					super().__init__(feature_size=lowerCamelCase_			,					sampling_rate=lowerCamelCase_			,					padding_value=lowerCamelCase_			,					**lowerCamelCase_						)

					_lowercase :    Dict							=							feature_size
					_lowercase :    Dict							=							sampling_rate
					_lowercase :    Tuple							=							padding_value
					_lowercase :    int							=							hop_length
					_lowercase :    Any							=							win_length
					_lowercase :    Union[str, Any]							=							frame_signal_scale
					_lowercase :    Tuple							=							preemphasis_coeff
					_lowercase :    Tuple							=							mel_floor
					_lowercase :    Tuple							=							normalize_means
					_lowercase :    List[Any]							=							normalize_vars
					_lowercase :    List[str]							=							win_function
					_lowercase :    int							=							return_attention_mask

					_lowercase :    Optional[Any]							=							win_length * sampling_rate // 1_0_0_0
					_lowercase :    Tuple							=							hop_length * sampling_rate // 1_0_0_0

					_lowercase :    str							=							optimal_fft_length(self.sample_size						)
					_lowercase :    Dict							=							(self.n_fft // 2) + 1







				def 							__UpperCAmelCase					( self    :       List[str]			,					lowerCamelCase_    :       np.array						):

					"""simple docstring"""






					if self.win_function == "hamming_window":
						_lowercase :    List[Any]							=							window_function(window_length=self.sample_size			,					name=self.win_function			,					periodic=lowerCamelCase_						)
					else:
						_lowercase :    Union[str, Any]							=							window_function(window_length=self.sample_size			,					name=self.win_function						)

					_lowercase :    Tuple							=							mel_filter_bank(
					    num_frequency_bins=self.n_freqs			,					num_mel_filters=self.feature_size			,					min_frequency=0.0			,					max_frequency=self.sampling_rate / 2.0			,					sampling_rate=self.sampling_rate			,					)

					_lowercase :    Tuple							=							spectrogram(
					    one_waveform * self.frame_signal_scale			,					window=lowerCamelCase_			,					frame_length=self.sample_size			,					hop_length=self.sample_stride			,					fft_length=self.n_fft			,					center=lowerCamelCase_			,					preemphasis=self.preemphasis_coeff			,					mel_filters=lowerCamelCase_			,					mel_floor=self.mel_floor			,					log_mel='log'			,					)
					return msfc_features.T







				def 							__UpperCAmelCase					( self    :       Optional[Any]			,					lowerCamelCase_    :       Dict			,					lowerCamelCase_    :       Dict			,					lowerCamelCase_    :       Tuple						):

					"""simple docstring"""






					if self.normalize_means:
						_lowercase :    Optional[int]							=							x[:input_length].mean(axis=0						)
						_lowercase :    int							=							np.subtract(lowerCamelCase_			,					lowerCamelCase_						)
					if self.normalize_vars:
						_lowercase :    int							=							x[:input_length].std(axis=0						)
						_lowercase :    Optional[Any]							=							np.divide(lowerCamelCase_			,					lowerCamelCase_						)

					if input_length < x.shape[0]:
						_lowercase :    Dict							=							padding_value

					# make sure array is in float32
					_lowercase :    Tuple							=							x.astype(np.floataa						)

					return x







				def 							__UpperCAmelCase					( self    :       List[Any]			,					lowerCamelCase_    :       List[np.ndarray]			,					lowerCamelCase_    :       Optional[np.ndarray] = None						):

					"""simple docstring"""






					_lowercase :    Dict							=							attention_mask.sum(-1						) if attention_mask is not None else [x.shape[0] for x in input_features]
					return [self._normalize_one(lowerCamelCase_			,					lowerCamelCase_			,					self.padding_value						) for x, n in zip(lowerCamelCase_			,					lowerCamelCase_						)]







				def __call__( self    :       Dict			,					lowerCamelCase_    :       Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]			,					lowerCamelCase_    :       Union[bool, str, PaddingStrategy] = False			,					lowerCamelCase_    :       Optional[int] = None			,					lowerCamelCase_    :       bool = False			,					lowerCamelCase_    :       Optional[int] = None			,					lowerCamelCase_    :       Optional[bool] = None			,					lowerCamelCase_    :       Optional[Union[str, TensorType]] = None			,					lowerCamelCase_    :       Optional[int] = None			,					**lowerCamelCase_    :       Optional[int]			,					):

					"""simple docstring"""






					if sampling_rate is not None:
						if sampling_rate != self.sampling_rate:
							raise ValueError(
							    F'''The model corresponding to this feature extractor: {self} was trained using a sampling rate of'''
							    F''' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'''
							    F''' {self.sampling_rate} and not {sampling_rate}.'''						)
					else:
						logger.warning(
						    'It is strongly recommended to pass the ``sampling_rate`` argument to this function. '
						    'Failing to do so can result in silent errors that might be hard to debug.'						)

					_lowercase :    Optional[Any]							=							isinstance(lowerCamelCase_			,					np.ndarray						) and len(raw_speech.shape						) > 1
					if is_batched_numpy and len(raw_speech.shape						) > 2:
						raise ValueError(F'''Only mono-channel audio is supported for input to {self}'''						)
					_lowercase :    Optional[int]							=							is_batched_numpy or (
					    isinstance(lowerCamelCase_			,					(list, tuple)						) and (isinstance(raw_speech[0]			,					(np.ndarray, tuple, list)						))
					)

					if is_batched:
						_lowercase :    str							=							[np.asarray(lowerCamelCase_			,					dtype=np.floataa						) for speech in raw_speech]
					elif not is_batched and not isinstance(lowerCamelCase_			,					np.ndarray						):
						_lowercase :    Tuple							=							np.asarray(lowerCamelCase_			,					dtype=np.floataa						)
					elif isinstance(lowerCamelCase_			,					np.ndarray						) and raw_speech.dtype is np.dtype(np.floataa						):
						_lowercase :    Optional[int]							=							raw_speech.astype(np.floataa						)

					# always return batch
					if not is_batched:
						_lowercase :    str							=							[raw_speech]

					# extract fbank features
					_lowercase :    Optional[Any]							=							[self._extract_mfsc_features(lowerCamelCase_						) for one_waveform in raw_speech]

					# convert into correct format for padding
					_lowercase :    Optional[int]							=							BatchFeature({'input_features': features}						)

					_lowercase :    Tuple							=							self.pad(
					    lowerCamelCase_			,					padding=lowerCamelCase_			,					max_length=lowerCamelCase_			,					truncation=lowerCamelCase_			,					pad_to_multiple_of=lowerCamelCase_			,					return_attention_mask=lowerCamelCase_			,					**lowerCamelCase_			,					)
					# make sure list is in array format
					_lowercase :    Dict							=							padded_inputs.get('input_features'						)
					if isinstance(input_features[0]			,					lowerCamelCase_						):
						_lowercase :    List[str]							=							[np.asarray(lowerCamelCase_			,					dtype=np.floataa						) for feature in input_features]

					_lowercase :    List[Any]							=							padded_inputs.get('attention_mask'						)
					if attention_mask is not None:
						_lowercase :    Union[str, Any]							=							[np.asarray(lowerCamelCase_			,					dtype=np.intaa						) for array in attention_mask]

					if self.normalize_means or self.normalize_vars:
						_lowercase :    int							=							(
						    np.array(lowerCamelCase_			,					dtype=np.intaa						)
						    if self._get_padding_strategies(lowerCamelCase_			,					max_length=lowerCamelCase_						) is not PaddingStrategy.DO_NOT_PAD
						    and padding
						    else None
						)
						_lowercase :    List[Any]							=							self.normalize(
						    padded_inputs['input_features']			,					attention_mask=lowerCamelCase_						)

					if return_tensors is not None:
						_lowercase :    Union[str, Any]							=							padded_inputs.convert_to_tensors(lowerCamelCase_						)

					return padded_inputs



