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


__UpperCamelCase		:       Any									=     WebClient(token=os.environ['CI_SLACK_BOT_TOKEN'])
def       A	(    _lowercase						):
						SCREAMING_SNAKE_CASE							:    int							=							test_results.split(''' '''						)

						SCREAMING_SNAKE_CASE							:    Tuple							=							0
						SCREAMING_SNAKE_CASE							:    str							=							0

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

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

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

						return failures





class 					lowercase__						:


						def __init__(       self	:				List[Any]    ,		UpperCamelCase__	:				str    ,		UpperCamelCase__	:				Dict						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Tuple							=							title

												SCREAMING_SNAKE_CASE							:    Tuple							=							doc_test_results['''time_spent'''].split(''','''						)[0]
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							doc_test_results['''success''']
												SCREAMING_SNAKE_CASE							:    Dict							=							doc_test_results['''failures''']
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							self.n_success + self.n_failures

												# Failures and success of the modeling tests
												SCREAMING_SNAKE_CASE							:    Dict							=							doc_test_results


						@property
						def        __A       (       self	:				Any						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    List[Any]							=							[self._time_spent]
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							0

												for time in time_spent:
																		SCREAMING_SNAKE_CASE							:    Union[str, 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(UpperCamelCase__						) == 1:
																								SCREAMING_SNAKE_CASE							:    str							=							[0, 0, time_parts[0]]

																		SCREAMING_SNAKE_CASE				,  SCREAMING_SNAKE_CASE				,  SCREAMING_SNAKE_CASE							:    List[Any]							=							int(time_parts[0]						), int(time_parts[1]						), float(time_parts[2]						)
																		total_secs += hours * 3600 + minutes * 60 + seconds

												SCREAMING_SNAKE_CASE				,  SCREAMING_SNAKE_CASE				,  SCREAMING_SNAKE_CASE							:    List[Any]							=							total_secs // 3600, (total_secs % 3600) // 60, total_secs % 60
												return f"""{int(UpperCamelCase__						)}h{int(UpperCamelCase__						)}m{int(UpperCamelCase__						)}s"""


						@property
						def        __A       (       self	:				List[str]						):





												'''simple docstring'''





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


						@property
						def        __A       (       self	:				List[str]						):





												'''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        __A       (       self	:				Union[str, 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        __A       (       self	:				Dict						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							40
												SCREAMING_SNAKE_CASE							:    Any							=							{k: v['''failed'''] for k, v in doc_test_results.items() if isinstance(UpperCamelCase__    ,		UpperCamelCase__						)}

												SCREAMING_SNAKE_CASE							:    Any							=							''''''
												for category, failures in category_failures.items():
																		if len(UpperCamelCase__						) == 0:
																								continue

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

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

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


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





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    int							=							[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(UpperCamelCase__						)


						@staticmethod
						def        __A       (       ):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Any							=							[
												    {
												        '''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(UpperCamelCase__						)}						)						)

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


						def        __A       (       self	:				int						):





												'''simple docstring'''





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

												SCREAMING_SNAKE_CASE							:    Optional[int]							=							f"""{self.n_failures} failures out of {self.n_tests} tests,""" if self.n_failures else '''All tests passed.'''

												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							client.chat_postMessage(
												    channel=os.environ['''CI_SLACK_CHANNEL_ID_DAILY''']    ,		blocks=self.payload    ,		text=UpperCamelCase__    ,		)


						def        __A       (       self	:				Dict    ,		UpperCamelCase__	:				Optional[int]    ,		UpperCamelCase__	:				List[Any]    ,		UpperCamelCase__	:				Dict    ,		UpperCamelCase__	:				str						):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    Optional[int]							=							''''''
												for key, value in failures.items():
																		SCREAMING_SNAKE_CASE							:    Optional[Any]							=							value[:200] + ''' [Truncated]''' if len(UpperCamelCase__						) > 250 else value
																		failures_text += f"""*{key}*\n_{value}_\n\n"""

												SCREAMING_SNAKE_CASE							:    List[Any]							=							job_name
												SCREAMING_SNAKE_CASE							:    int							=							{'''type''': '''section''', '''text''': {'''type''': '''mrkdwn''', '''text''': text}}

												if job_link is not None:
																		SCREAMING_SNAKE_CASE							:    Optional[Any]							=							{
																		    '''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        __A       (       self	:				Dict						):





												'''simple docstring'''





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

												SCREAMING_SNAKE_CASE							:    Optional[int]							=							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'''						)

												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							sorted(self.doc_test_results.items()    ,		key=lambda UpperCamelCase__						: t[0]						)
												for job, job_result in sorted_dict:
																		if len(job_result['''failures''']						):
																								SCREAMING_SNAKE_CASE							:    int							=							f"""*Num failures* :{len(job_result['failed']						)} \n"""
																								SCREAMING_SNAKE_CASE							:    List[str]							=							job_result['''failures''']
																								SCREAMING_SNAKE_CASE							:    Any							=							self.get_reply_blocks(UpperCamelCase__    ,		UpperCamelCase__    ,		UpperCamelCase__    ,		text=UpperCamelCase__						)

																								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=UpperCamelCase__    ,		thread_ts=self.thread_ts['''ts''']    ,		)

																								time.sleep(1						)






def       A	(    ):
						SCREAMING_SNAKE_CASE							:    Optional[int]							=							os.environ['''GITHUB_RUN_ID''']
						SCREAMING_SNAKE_CASE							:    Optional[Any]							=							f"""https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100"""
						SCREAMING_SNAKE_CASE							:    Dict							=							requests.get(_lowercase						).json()
						SCREAMING_SNAKE_CASE							:    Dict							=							{}

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

												for i in range(_lowercase						):
																		SCREAMING_SNAKE_CASE							:    List[str]							=							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.'''		,  _lowercase						)

						return {}
def       A	(    _lowercase						):
						SCREAMING_SNAKE_CASE							:    Dict							=							{}

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

						return _artifact
def       A	(    ):





						class 					lowercase__						:


												def __init__(       self	:				int    ,		UpperCamelCase__	:				str						):





																		'''simple docstring'''





																		SCREAMING_SNAKE_CASE							:    Dict							=							name
																		SCREAMING_SNAKE_CASE							:    Dict							=							[]


												def __str__(       self	:				str						):





																		'''simple docstring'''





																		return self.name



												def        __A       (       self	:				Tuple    ,		UpperCamelCase__	:				str						):





																		'''simple docstring'''





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






						SCREAMING_SNAKE_CASE							:    Dict[str, Artifact]							=							{}

						SCREAMING_SNAKE_CASE							:    Tuple							=							filter(os.path.isdir		,  os.listdir()						)
						for directory in directories:
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							directory
												if artifact_name not in _available_artifacts:
																		SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							Artifact(_lowercase						)

																		_available_artifacts[artifact_name].add_path(_lowercase						)

						return _available_artifacts


if __name__ == "__main__":
					__UpperCamelCase		:       List[Any]									=     get_job_links()
					__UpperCamelCase		:       List[str]									=     retrieve_available_artifacts()

					__UpperCamelCase		:       str									=     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'
					__UpperCamelCase		:       List[Any]									=     {
					    v: {
					        'failed': [],
					        'failures': {},
					    }
					    for v in docs.values()
					}

					# Link to the GitHub Action job
					__UpperCamelCase		:       str									=     github_actions_job_links.get('run_doctests')

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

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

																				if "::" in line:
																									__UpperCamelCase						,			__UpperCamelCase		:       List[Any]									=     line.split('::')
																				else:
																									__UpperCamelCase						,			__UpperCamelCase		:       Optional[Any]									=     line, line

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

																														__UpperCamelCase		:       int									=     all_failures[test] if test in all_failures else 'N/A'
																														__UpperCamelCase		:       List[str]									=     failure
																														break

					__UpperCamelCase		:       Union[str, Any]									=     Message('🤗 Results of the doc tests.', doc_test_results)
					message.post()
					message.post_reply()






from ...configuration_utils import PretrainedConfig
from ...utils import logging


__UpperCamelCase		:       Any									=     logging.get_logger(__name__)

__UpperCamelCase		:       Optional[int]									=     {
    'facebook/nllb-moe-54B': 'https://huggingface.co/facebook/nllb-moe-54b/resolve/main/config.json',
}





class 					lowercase__						(   UpperCamelCase_):
						UpperCamelCase_     =		"""nllb-moe"""
						UpperCamelCase_     =		["""past_key_values"""]
						UpperCamelCase_     =		{"""num_attention_heads""": """encoder_attention_heads""", """hidden_size""": """d_model"""}


						def __init__(       self	:				List[str]    ,		UpperCamelCase__	:				List[str]=12_8112    ,		UpperCamelCase__	:				str=1024    ,		UpperCamelCase__	:				Optional[int]=12    ,		UpperCamelCase__	:				Union[str, Any]=4096    ,		UpperCamelCase__	:				Optional[Any]=16    ,		UpperCamelCase__	:				Optional[Any]=12    ,		UpperCamelCase__	:				str=4096    ,		UpperCamelCase__	:				Dict=16    ,		UpperCamelCase__	:				Any=0.05    ,		UpperCamelCase__	:				Any=0.05    ,		UpperCamelCase__	:				Optional[Any]=True    ,		UpperCamelCase__	:				Optional[Any]=True    ,		UpperCamelCase__	:				List[Any]="relu"    ,		UpperCamelCase__	:				Union[str, Any]=1024    ,		UpperCamelCase__	:				Optional[int]=0.1    ,		UpperCamelCase__	:				Any=0.1    ,		UpperCamelCase__	:				Optional[Any]=0.0    ,		UpperCamelCase__	:				str=0.02    ,		UpperCamelCase__	:				Tuple=2    ,		UpperCamelCase__	:				List[str]=True    ,		UpperCamelCase__	:				Dict=False    ,		UpperCamelCase__	:				Any="float32"    ,		UpperCamelCase__	:				List[Any]=False    ,		UpperCamelCase__	:				Tuple=128    ,		UpperCamelCase__	:				Tuple=64    ,		UpperCamelCase__	:				List[str]=4    ,		UpperCamelCase__	:				Union[str, Any]=4    ,		UpperCamelCase__	:				Dict=0.001    ,		UpperCamelCase__	:				Optional[Any]=0.001    ,		UpperCamelCase__	:				Optional[Any]="all"    ,		UpperCamelCase__	:				Tuple=False    ,		UpperCamelCase__	:				List[Any]=False    ,		UpperCamelCase__	:				List[Any]=1.0    ,		UpperCamelCase__	:				str=0.2    ,		UpperCamelCase__	:				List[Any]=1    ,		UpperCamelCase__	:				List[Any]=0    ,		UpperCamelCase__	:				Optional[int]=2    ,		UpperCamelCase__	:				Tuple=False    ,		**UpperCamelCase__	:				Union[str, Any]    ,		):





												'''simple docstring'''





												SCREAMING_SNAKE_CASE							:    List[str]							=							vocab_size
												SCREAMING_SNAKE_CASE							:    Dict							=							max_position_embeddings
												SCREAMING_SNAKE_CASE							:    int							=							d_model
												SCREAMING_SNAKE_CASE							:    Any							=							encoder_ffn_dim
												SCREAMING_SNAKE_CASE							:    List[Any]							=							encoder_layers
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							encoder_attention_heads
												SCREAMING_SNAKE_CASE							:    List[str]							=							decoder_ffn_dim
												SCREAMING_SNAKE_CASE							:    Dict							=							decoder_layers
												SCREAMING_SNAKE_CASE							:    List[Any]							=							decoder_attention_heads
												SCREAMING_SNAKE_CASE							:    List[str]							=							dropout
												SCREAMING_SNAKE_CASE							:    Any							=							attention_dropout
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							activation_dropout
												SCREAMING_SNAKE_CASE							:    List[Any]							=							activation_function
												SCREAMING_SNAKE_CASE							:    Union[str, Any]							=							init_std
												SCREAMING_SNAKE_CASE							:    int							=							encoder_layerdrop
												SCREAMING_SNAKE_CASE							:    List[Any]							=							decoder_layerdrop
												SCREAMING_SNAKE_CASE							:    Any							=							use_cache
												SCREAMING_SNAKE_CASE							:    str							=							encoder_layers
												SCREAMING_SNAKE_CASE							:    List[Any]							=							scale_embedding  # scale factor will be sqrt(d_model) if True
												SCREAMING_SNAKE_CASE							:    List[str]							=							router_z_loss_coef
												SCREAMING_SNAKE_CASE							:    List[str]							=							router_aux_loss_coef
												SCREAMING_SNAKE_CASE							:    int							=							decoder_sparse_step
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							encoder_sparse_step
												SCREAMING_SNAKE_CASE							:    List[str]							=							num_experts
												SCREAMING_SNAKE_CASE							:    int							=							expert_capacity
												SCREAMING_SNAKE_CASE							:    Any							=							router_bias
												if router_dtype not in ["float32", "float16", "bfloat16"]:
																		raise ValueError(f"""`router_dtype` must be one of 'float32', 'float16' or 'bfloat16', got {router_dtype}"""						)
												SCREAMING_SNAKE_CASE							:    str							=							router_dtype

												SCREAMING_SNAKE_CASE							:    List[Any]							=							router_ignore_padding_tokens
												SCREAMING_SNAKE_CASE							:    int							=							batch_prioritized_routing
												SCREAMING_SNAKE_CASE							:    str							=							second_expert_policy
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							normalize_router_prob_before_dropping
												SCREAMING_SNAKE_CASE							:    Optional[Any]							=							moe_eval_capacity_token_fraction
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							moe_token_dropout
												SCREAMING_SNAKE_CASE							:    Optional[int]							=							output_router_logits
												super().__init__(
												    pad_token_id=UpperCamelCase__    ,		bos_token_id=UpperCamelCase__    ,		eos_token_id=UpperCamelCase__    ,		is_encoder_decoder=UpperCamelCase__    ,		decoder_start_token_id=UpperCamelCase__    ,		**UpperCamelCase__    ,		)





