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


UpperCAmelCase    					=   logging.get_logger(__name__)

UpperCAmelCase    					=   {
    "transfo-xl-wt103": "https://huggingface.co/transfo-xl-wt103/resolve/main/config.json",
}

class 						a   (				__magic_name__    ):
				_snake_case		    = '''transfo-xl'''
				_snake_case		    = ['''mems''']
				_snake_case		    = {
				    '''n_token''': '''vocab_size''',
				    '''hidden_size''': '''d_model''',
				    '''num_attention_heads''': '''n_head''',
				    '''num_hidden_layers''': '''n_layer''',
				}




				def __init__(							self	:				List[str],    SCREAMING_SNAKE_CASE_	:				Union[str, Any]=26_77_35,    SCREAMING_SNAKE_CASE_	:				int=[2_00_00, 4_00_00, 20_00_00],    SCREAMING_SNAKE_CASE_	:				Tuple=10_24,    SCREAMING_SNAKE_CASE_	:				int=10_24,    SCREAMING_SNAKE_CASE_	:				str=16,    SCREAMING_SNAKE_CASE_	:				Union[str, Any]=64,    SCREAMING_SNAKE_CASE_	:				List[Any]=40_96,    SCREAMING_SNAKE_CASE_	:				int=4,    SCREAMING_SNAKE_CASE_	:				List[Any]=False,    SCREAMING_SNAKE_CASE_	:				Any=18,    SCREAMING_SNAKE_CASE_	:				List[Any]=16_00,    SCREAMING_SNAKE_CASE_	:				Optional[int]=10_00,    SCREAMING_SNAKE_CASE_	:				Optional[Any]=True,    SCREAMING_SNAKE_CASE_	:				Optional[int]=True,    SCREAMING_SNAKE_CASE_	:				str=0,    SCREAMING_SNAKE_CASE_	:				Optional[int]=-1,    SCREAMING_SNAKE_CASE_	:				List[str]=True,    SCREAMING_SNAKE_CASE_	:				Dict=0.1,    SCREAMING_SNAKE_CASE_	:				str=0.0,    SCREAMING_SNAKE_CASE_	:				Optional[Any]=True,    SCREAMING_SNAKE_CASE_	:				Optional[Any]="normal",    SCREAMING_SNAKE_CASE_	:				str=0.01,    SCREAMING_SNAKE_CASE_	:				Union[str, Any]=0.01,    SCREAMING_SNAKE_CASE_	:				List[str]=0.02,    SCREAMING_SNAKE_CASE_	:				Optional[int]=1e-5,    SCREAMING_SNAKE_CASE_	:				str=0,    **SCREAMING_SNAKE_CASE_	:				Optional[int],    ):
										snake_case       :     Any =    vocab_size
										snake_case       :     str =    []
										self.cutoffs.extend(SCREAMING_SNAKE_CASE_      )
										if proj_share_all_but_first:
																snake_case       :     List[Any] =    [False] + [True] * len(self.cutoffs      )
										else:
																snake_case       :     Any =    [False] + [False] * len(self.cutoffs      )
										snake_case       :     List[Any] =    d_model
										snake_case       :     Optional[int] =    d_embed
										snake_case       :     int =    d_head
										snake_case       :     Dict =    d_inner
										snake_case       :     Tuple =    div_val
										snake_case       :     Union[str, Any] =    pre_lnorm
										snake_case       :     Optional[int] =    n_layer
										snake_case       :     Dict =    n_head
										snake_case       :     List[str] =    mem_len
										snake_case       :     str =    same_length
										snake_case       :     Any =    attn_type
										snake_case       :     List[Any] =    clamp_len
										snake_case       :     Dict =    sample_softmax
										snake_case       :     Dict =    adaptive
										snake_case       :     Any =    dropout
										snake_case       :     Any =    dropatt
										snake_case       :     str =    untie_r
										snake_case       :     Optional[Any] =    init
										snake_case       :     str =    init_range
										snake_case       :     int =    proj_init_std
										snake_case       :     Dict =    init_std
										snake_case       :     Any =    layer_norm_epsilon
										super().__init__(eos_token_id=SCREAMING_SNAKE_CASE_,    **SCREAMING_SNAKE_CASE_      )




				@property
				def 	__snake_case							(							self	:				List[str]      ):
										# Message copied from Transformer-XL documentation
										logger.info(F"""The model {self.model_type} is one of the few models that has no sequence length limit."""      )
										return -1




				@max_position_embeddings.setter
				def 	__snake_case							(							self	:				Optional[Any],    SCREAMING_SNAKE_CASE_	:				List[Any]      ):
										# Message copied from Transformer-XL documentation
										raise NotImplementedError(
										    F"""The model {self.model_type} is one of the few models that has no sequence length limit."""      )



'''simple docstring'''





import argparse
import json
import math
import os
import time
import traceback
import zipfile
from collections import Counter

import requests

def      A (       A_							:					List[Any]    ,							A_							:					int=None       ):
						snake_case       :     Optional[int] =    None
						if token is not None:
												snake_case       :     List[Any] =    {'''Accept''': '''application/vnd.github+json''', '''Authorization''': F"""Bearer {token}"""}

						snake_case       :     str =    F"""https://api.github.com/repos/huggingface/transformers/actions/runs/{workflow_run_id}/jobs?per_page=100"""
						snake_case       :     List[Any] =    requests.get(A_    ,							headers=A_       ).json()
						snake_case       :     List[Any] =    {}

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

												for i in range(A_       ):
																		snake_case       :     Optional[Any] =    requests.get(url + F"""&page={i + 2}"""    ,							headers=A_       ).json()
																		job_links.update({job['''name''']: job['''html_url'''] for job in result['''jobs''']}       )

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

						return {}

def      A (       A_							:					int    ,							A_							:					Optional[int]=None       ):
						snake_case       :     Any =    None
						if token is not None:
												snake_case       :     str =    {'''Accept''': '''application/vnd.github+json''', '''Authorization''': F"""Bearer {token}"""}

						snake_case       :     Tuple =    F"""https://api.github.com/repos/huggingface/transformers/actions/runs/{worflow_run_id}/artifacts?per_page=100"""
						snake_case       :     List[Any] =    requests.get(A_    ,							headers=A_       ).json()
						snake_case       :     List[Any] =    {}

						try:
												artifacts.update({artifact['''name''']: artifact['''archive_download_url'''] for artifact in result['''artifacts''']}       )
												snake_case       :     Dict =    math.ceil((result['''total_count'''] - 100) / 100       )

												for i in range(A_       ):
																		snake_case       :     List[Any] =    requests.get(url + F"""&page={i + 2}"""    ,							headers=A_       ).json()
																		artifacts.update({artifact['''name''']: artifact['''archive_download_url'''] for artifact in result['''artifacts''']}       )

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

						return {}

def      A (       A_							:					List[Any]    ,							A_							:					List[Any]    ,							A_							:					Any    ,							A_							:					str       ):
						snake_case       :     Any =    None
						if token is not None:
												snake_case       :     int =    {'''Accept''': '''application/vnd.github+json''', '''Authorization''': F"""Bearer {token}"""}

						snake_case       :     Any =    requests.get(A_    ,							headers=A_    ,							allow_redirects=A_       )
						snake_case       :     Union[str, Any] =    result.headers['''Location''']
						snake_case       :     str =    requests.get(A_    ,							allow_redirects=A_       )
						snake_case       :     int =    os.path.join(A_    ,							F"""{artifact_name}.zip"""       )
						with open(A_    ,							'''wb'''       ) as fp:
												fp.write(response.content       )

def      A (       A_							:					int    ,							A_							:					Any=None       ):
						snake_case       :     Optional[int] =    []
						snake_case       :     Any =    []
						snake_case       :     str =    None

						with zipfile.ZipFile(A_       ) as z:
												for filename in z.namelist():
																		if not os.path.isdir(A_       ):
																								# read the file
																								if filename in ["failures_line.txt", "summary_short.txt", "job_name.txt"]:
																														with z.open(A_       ) as f:
																																				for line in f:
																																										snake_case       :     Optional[int] =    line.decode('''UTF-8'''       ).strip()
																																										if filename == "failures_line.txt":
																																																try:
																																																						# `error_line` is the place where `error` occurs
																																																						snake_case       :     List[str] =    line[: line.index(''': '''       )]
																																																						snake_case       :     List[Any] =    line[line.index(''': '''       ) + len(''': '''       ) :]
																																																						errors.append([error_line, error]       )
																																																except Exception:
																																																						# skip un-related lines
																																																						pass
																																										elif filename == "summary_short.txt" and line.startswith('''FAILED '''       ):
																																																# `test` is the test method that failed
																																																snake_case       :     List[Any] =    line[len('''FAILED '''       ) :]
																																																failed_tests.append(A_       )
																																										elif filename == "job_name.txt":
																																																snake_case       :     Optional[int] =    line

						if len(A_       ) != len(A_       ):
												raise ValueError(
												    F"""`errors` and `failed_tests` should have the same number of elements. Got {len(A_       )} for `errors` """
												    F"""and {len(A_       )} for `failed_tests` instead. The test reports in {artifact_zip_path} have some"""
												    ''' problem.'''       )

						snake_case       :     int =    None
						if job_name and job_links:
												snake_case       :     int =    job_links.get(A_    ,							A_       )

						# A list with elements of the form (line of error, error, failed test)
						snake_case       :     Tuple =    [x + [y] + [job_link] for x, y in zip(A_    ,							A_       )]

						return result

def      A (       A_							:					Optional[Any]    ,							A_							:					List[str]=None       ):
						snake_case       :     str =    []

						snake_case       :     str =    [os.path.join(A_    ,							A_       ) for p in os.listdir(A_       ) if p.endswith('''.zip'''       )]
						for p in paths:
												errors.extend(get_errors_from_single_artifact(A_    ,							job_links=A_       )       )

						return errors

def      A (       A_							:					Dict    ,							A_							:					Optional[Any]=None       ):
						snake_case       :     Any =    Counter()
						counter.update([x[1] for x in logs]       )
						snake_case       :     Optional[Any] =    counter.most_common()
						snake_case       :     Dict =    {}
						for error, count in counts:
												if error_filter is None or error not in error_filter:
																		snake_case       :     str =    {'''count''': count, '''failed_tests''': [(x[2], x[0]) for x in logs if x[1] == error]}

						snake_case       :     Dict =    dict(sorted(r.items()    ,							key=lambda A_       : item[1]["count"]    ,							reverse=A_       )       )
						return r

def      A (       A_							:					Tuple       ):
						snake_case       :     Tuple =    test.split('''::'''       )[0]
						if test.startswith('''tests/models/'''       ):
												snake_case       :     Optional[int] =    test.split('''/'''       )[2]
						else:
												snake_case       :     int =    None

						return test

def      A (       A_							:					Optional[Any]    ,							A_							:					Dict=None       ):
						snake_case       :     List[str] =    [(x[0], x[1], get_model(x[2]       )) for x in logs]
						snake_case       :     int =    [x for x in logs if x[2] is not None]
						snake_case       :     int =    {x[2] for x in logs}

						snake_case       :     List[str] =    {}
						for test in tests:
												snake_case       :     Dict =    Counter()
												# count by errors in `test`
												counter.update([x[1] for x in logs if x[2] == test]       )
												snake_case       :     Optional[Any] =    counter.most_common()
												snake_case       :     str =    {error: count for error, count in counts if (error_filter is None or error not in error_filter)}
												snake_case       :     List[Any] =    sum(error_counts.values()       )
												if n_errors > 0:
																		snake_case       :     Optional[Any] =    {'''count''': n_errors, '''errors''': error_counts}

						snake_case       :     Optional[int] =    dict(sorted(r.items()    ,							key=lambda A_       : item[1]["count"]    ,							reverse=A_       )       )
						return r

def      A (       A_							:					Optional[Any]       ):
						snake_case       :     Dict =    '''| no. | error | status |'''
						snake_case       :     Union[str, Any] =    '''|-:|:-|:-|'''
						snake_case       :     List[Any] =    [header, sep]
						for error in reduced_by_error:
												snake_case       :     Optional[int] =    reduced_by_error[error]['''count''']
												snake_case       :     List[str] =    F"""| {count} | {error[:100]} |  |"""
												lines.append(A_       )

						return "\n".join(A_       )

def      A (       A_							:					Optional[int]       ):
						snake_case       :     Union[str, Any] =    '''| model | no. of errors | major error | count |'''
						snake_case       :     Dict =    '''|-:|-:|-:|-:|'''
						snake_case       :     Optional[Any] =    [header, sep]
						for model in reduced_by_model:
												snake_case       :     str =    reduced_by_model[model]['''count''']
												snake_case,					snake_case       :     Tuple =    list(reduced_by_model[model]['''errors'''].items()       )[0]
												snake_case       :     Dict =    F"""| {model} | {count} | {error[:60]} | {_count} |"""
												lines.append(A_       )

						return "\n".join(A_       )


if __name__ == "__main__":
						UpperCAmelCase    					=   argparse.ArgumentParser()
						# Required parameters
						parser.add_argument("--workflow_run_id", type=str, required=True, help="A GitHub Actions workflow run id.")
						parser.add_argument(
						    "--output_dir",
						    type=str,
						    required=True,
						    help="Where to store the downloaded artifacts and other result files.",
						)
						parser.add_argument("--token", default=None, type=str, help="A token that has actions:read permission.")
						UpperCAmelCase    					=   parser.parse_args()

						os.makedirs(args.output_dir, exist_ok=True)

						UpperCAmelCase    					=   get_job_links(args.workflow_run_id, token=args.token)
						UpperCAmelCase    					=   {}
						# To deal with `workflow_call` event, where a job name is the combination of the job names in the caller and callee.
						# For example, `PyTorch 1.11 / Model tests (models/albert, single-gpu)`.
						if _job_links:
												for k, v in _job_links.items():
																		# This is how GitHub actions combine job names.
																		if " / " in k:
																								UpperCAmelCase    					=   k.find(" / ")
																								UpperCAmelCase    					=   k[index + len(" / ") :]
																		UpperCAmelCase    					=   v
						with open(os.path.join(args.output_dir, "job_links.json"), "w", encoding="UTF-8") as fp:
												json.dump(job_links, fp, ensure_ascii=False, indent=4)

						UpperCAmelCase    					=   get_artifacts_links(args.workflow_run_id, token=args.token)
						with open(os.path.join(args.output_dir, "artifacts.json"), "w", encoding="UTF-8") as fp:
												json.dump(artifacts, fp, ensure_ascii=False, indent=4)

						for idx, (name, url) in enumerate(artifacts.items()):
												download_artifact(name, url, args.output_dir, args.token)
												# Be gentle to GitHub
												time.sleep(1)

						UpperCAmelCase    					=   get_all_errors(args.output_dir, job_links=job_links)

						# `e[1]` is the error
						UpperCAmelCase    					=   Counter()
						counter.update([e[1] for e in errors])

						# print the top 30 most common test errors
						UpperCAmelCase    					=   counter.most_common(30)
						for item in most_common:
												print(item)

						with open(os.path.join(args.output_dir, "errors.json"), "w", encoding="UTF-8") as fp:
												json.dump(errors, fp, ensure_ascii=False, indent=4)

						UpperCAmelCase    					=   reduce_by_error(errors)
						UpperCAmelCase    					=   reduce_by_model(errors)

						UpperCAmelCase    					=   make_github_table(reduced_by_error)
						UpperCAmelCase    					=   make_github_table_per_model(reduced_by_model)

						with open(os.path.join(args.output_dir, "reduced_by_error.txt"), "w", encoding="UTF-8") as fp:
												fp.write(sa)
						with open(os.path.join(args.output_dir, "reduced_by_model.txt"), "w", encoding="UTF-8") as fp:
												fp.write(sa)

