


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

import requests






def 							_lowercase    (		SCREAMING_SNAKE_CASE_   :		Optional[Any]       ,  SCREAMING_SNAKE_CASE_   :		Dict=None	):





			"""simple docstring"""




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

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

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

						for i in range(SCREAMING_SNAKE_CASE_	):
									UpperCamelCase         =					requests.get(url + f'&page={i + 2}'       ,  headers=SCREAMING_SNAKE_CASE_	).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 							_lowercase    (		SCREAMING_SNAKE_CASE_   :		List[str]       ,  SCREAMING_SNAKE_CASE_   :		Optional[Any]=None	):





			"""simple docstring"""




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

			UpperCamelCase         =					f'https://api.github.com/repos/huggingface/transformers/actions/runs/{worflow_run_id}/artifacts?per_page=100'
			UpperCamelCase         =					requests.get(SCREAMING_SNAKE_CASE_       ,  headers=SCREAMING_SNAKE_CASE_	).json()
			UpperCamelCase         =					{}

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

						for i in range(SCREAMING_SNAKE_CASE_	):
									UpperCamelCase         =					requests.get(url + f'&page={i + 2}'       ,  headers=SCREAMING_SNAKE_CASE_	).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 							_lowercase    (		SCREAMING_SNAKE_CASE_   :		Any       ,  SCREAMING_SNAKE_CASE_   :		List[str]       ,  SCREAMING_SNAKE_CASE_   :		str       ,  SCREAMING_SNAKE_CASE_   :		Optional[int]	):





			"""simple docstring"""




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

			UpperCamelCase         =					requests.get(SCREAMING_SNAKE_CASE_       ,  headers=SCREAMING_SNAKE_CASE_       ,  allow_redirects=SCREAMING_SNAKE_CASE_	)
			UpperCamelCase         =					result.headers["""Location"""]
			UpperCamelCase         =					requests.get(SCREAMING_SNAKE_CASE_       ,  allow_redirects=SCREAMING_SNAKE_CASE_	)
			UpperCamelCase         =					os.path.join(SCREAMING_SNAKE_CASE_       ,  f'{artifact_name}.zip'	)
			with open(SCREAMING_SNAKE_CASE_       ,  """wb"""	) as fp:
						fp.write(response.content	)






def 							_lowercase    (		SCREAMING_SNAKE_CASE_   :		Dict       ,  SCREAMING_SNAKE_CASE_   :		List[str]=None	):





			"""simple docstring"""




			UpperCamelCase         =					[]
			UpperCamelCase         =					[]
			UpperCamelCase         =					None

			with zipfile.ZipFile(SCREAMING_SNAKE_CASE_	) as z:
						for filename in z.namelist():
									if not os.path.isdir(SCREAMING_SNAKE_CASE_	):
												# read the file
												if filename in ["failures_line.txt", "summary_short.txt", "job_name.txt"]:
															with z.open(SCREAMING_SNAKE_CASE_	) as f:
																		for line in f:
																					UpperCamelCase         =					line.decode("""UTF-8"""	).strip()
																					if filename == "failures_line.txt":
																								try:
																											# `error_line` is the place where `error` occurs
																											UpperCamelCase         =					line[: line.index(""": """	)]
																											UpperCamelCase         =					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
																								UpperCamelCase         =					line[len("""FAILED """	) :]
																								failed_tests.append(SCREAMING_SNAKE_CASE_	)
																					elif filename == "job_name.txt":
																								UpperCamelCase         =					line

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

			UpperCamelCase         =					None
			if job_name and job_links:
						UpperCamelCase         =					job_links.get(SCREAMING_SNAKE_CASE_       ,  SCREAMING_SNAKE_CASE_	)

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

			return result






def 							_lowercase    (		SCREAMING_SNAKE_CASE_   :		Dict       ,  SCREAMING_SNAKE_CASE_   :		Tuple=None	):





			"""simple docstring"""




			UpperCamelCase         =					[]

			UpperCamelCase         =					[os.path.join(SCREAMING_SNAKE_CASE_       ,  SCREAMING_SNAKE_CASE_	) for p in os.listdir(SCREAMING_SNAKE_CASE_	) if p.endswith(""".zip"""	)]
			for p in paths:
						errors.extend(get_errors_from_single_artifact(SCREAMING_SNAKE_CASE_       ,  job_links=SCREAMING_SNAKE_CASE_	)	)

			return errors






def 							_lowercase    (		SCREAMING_SNAKE_CASE_   :		int       ,  SCREAMING_SNAKE_CASE_   :		List[Any]=None	):





			"""simple docstring"""




			UpperCamelCase         =					Counter()
			counter.update([x[1] for x in logs]	)
			UpperCamelCase         =					counter.most_common()
			UpperCamelCase         =					{}
			for error, count in counts:
						if error_filter is None or error not in error_filter:
									UpperCamelCase         =					{"""count""": count, """failed_tests""": [(x[2], x[0]) for x in logs if x[1] == error]}

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






def 							_lowercase    (		SCREAMING_SNAKE_CASE_   :		List[str]	):





			"""simple docstring"""




			UpperCamelCase         =					test.split("""::"""	)[0]
			if test.startswith("""tests/models/"""	):
						UpperCamelCase         =					test.split("""/"""	)[2]
			else:
						UpperCamelCase         =					None

			return test






def 							_lowercase    (		SCREAMING_SNAKE_CASE_   :		Any       ,  SCREAMING_SNAKE_CASE_   :		Dict=None	):





			"""simple docstring"""




			UpperCamelCase         =					[(x[0], x[1], get_model(x[2]	)) for x in logs]
			UpperCamelCase         =					[x for x in logs if x[2] is not None]
			UpperCamelCase         =					{x[2] for x in logs}

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

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






def 							_lowercase    (		SCREAMING_SNAKE_CASE_   :		Optional[Any]	):





			"""simple docstring"""




			UpperCamelCase         =					"""| no. | error | status |"""
			UpperCamelCase         =					"""|-:|:-|:-|"""
			UpperCamelCase         =					[header, sep]
			for error in reduced_by_error:
						UpperCamelCase         =					reduced_by_error[error]["""count"""]
						UpperCamelCase         =					f'| {count} | {error[:100]} |  |'
						lines.append(SCREAMING_SNAKE_CASE_	)

			return "\n".join(SCREAMING_SNAKE_CASE_	)






def 							_lowercase    (		SCREAMING_SNAKE_CASE_   :		int	):





			"""simple docstring"""




			UpperCamelCase         =					"""| model | no. of errors | major error | count |"""
			UpperCamelCase         =					"""|-:|-:|-:|-:|"""
			UpperCamelCase         =					[header, sep]
			for model in reduced_by_model:
						UpperCamelCase         =					reduced_by_model[model]["""count"""]
						UpperCamelCase  ,  UpperCamelCase         =					list(reduced_by_model[model]["""errors"""].items()	)[0]
						UpperCamelCase         =					f'| {model} | {count} | {error[:60]} | {_count} |'
						lines.append(SCREAMING_SNAKE_CASE_	)

			return "\n".join(SCREAMING_SNAKE_CASE_	)


if __name__ == "__main__":
			__snake_case            =   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.")
			__snake_case            =   parser.parse_args()

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

			__snake_case            =   get_job_links(args.workflow_run_id, token=args.token)
			__snake_case            =   {}
			# 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:
												__snake_case            =   k.find(" / ")
												__snake_case            =   k[index + len(" / ") :]
									__snake_case            =   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)

			__snake_case            =   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)

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

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

			# print the top 30 most common test errors
			__snake_case            =   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)

			__snake_case            =   reduce_by_error(errors)
			__snake_case            =   reduce_by_model(errors)

			__snake_case            =   make_github_table(reduced_by_error)
			__snake_case            =   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)








import logging
import os

import quant_trainer
import torch
from torch.utils.data import DataLoader

from transformers import Trainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput


__snake_case            =   logging.getLogger(__name__)

if is_torch_tpu_available(check_device=False):
			import torch_xla.core.xla_model as xm
			import torch_xla.debug.metrics as met




class    UpperCAmelCase					(       __snake_case    ):




			def __init__(						self							:		Dict       ,       *__magic_name__							:		Union[str, Any]       ,       __magic_name__							:		Any=None       ,       __magic_name__							:		Union[str, Any]=None       ,       __magic_name__							:		Any=None       ,       **__magic_name__							:		Optional[Any]						):







						"""simple docstring"""







						super().__init__(*__magic_name__       ,       **__magic_name__						)
						UpperCamelCase         =					eval_examples
						UpperCamelCase         =					post_process_function
						UpperCamelCase         =					quant_trainer_args
						UpperCamelCase         =					1_2_8  # default number of calibration samples




			def   lowerCamelCase_					(						self							:		str       ,       __magic_name__							:		List[Any]=None						):







						"""simple docstring"""







						if calib_dataset is None and self.calib_dataset is None:
									raise ValueError("""Trainer: calibration requires an calib_dataset."""						)
						UpperCamelCase         =					calib_dataset if calib_dataset is not None else self.calib_dataset

						UpperCamelCase         =					self._remove_unused_columns(__magic_name__       ,       description="""Calibration"""						)

						return DataLoader(
						    __magic_name__       ,       batch_size=self.args.eval_batch_size       ,       collate_fn=self.data_collator       ,       drop_last=self.args.dataloader_drop_last       ,       num_workers=self.args.dataloader_num_workers       ,       pin_memory=self.args.dataloader_pin_memory       ,       shuffle=__magic_name__       ,       )




			def   lowerCamelCase_					(						self							:		Tuple       ,       __magic_name__							:		List[str]=None						):







						"""simple docstring"""







						UpperCamelCase         =					self.train_dataset if calib_dataset is None else calib_dataset
						UpperCamelCase         =					self.get_calib_dataloader(__magic_name__						)

						UpperCamelCase         =					self.model
						quant_trainer.configure_model(__magic_name__       ,       self.quant_trainer_args       ,       calib=__magic_name__						)
						model.eval()
						quant_trainer.enable_calibration(__magic_name__						)

						logger.info("""***** Running calibration *****"""						)
						logger.info(F'  Num examples = {self.calib_num}'						)
						logger.info(F'  Batch size = {calib_dataloader.batch_size}'						)

						for step, inputs in enumerate(__magic_name__						):
									# Prediction step
									UpperCamelCase  ,  UpperCamelCase  ,  UpperCamelCase         =					self.prediction_step(__magic_name__       ,       __magic_name__       ,       prediction_loss_only=__magic_name__						)
									if (step + 1) * calib_dataloader.batch_size >= self.calib_num:
												break

						quant_trainer.finish_calibration(__magic_name__       ,       self.quant_trainer_args						)
						UpperCamelCase         =					model




			def   lowerCamelCase_					(						self							:		Any       ,       __magic_name__							:		Optional[int]=None       ,       __magic_name__							:		Optional[Any]=None       ,       __magic_name__							:		Union[str, Any]=None       ,       __magic_name__							:		str = "eval"						):







						"""simple docstring"""







						UpperCamelCase         =					self.eval_dataset if eval_dataset is None else eval_dataset
						UpperCamelCase         =					self.get_eval_dataloader(__magic_name__						)
						UpperCamelCase         =					self.eval_examples if eval_examples is None else eval_examples

						# Temporarily disable metric computation, we will do it in the loop here.
						UpperCamelCase         =					self.compute_metrics
						UpperCamelCase         =					None
						UpperCamelCase         =					self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
						try:
									UpperCamelCase         =					eval_loop(
									    __magic_name__       ,       description="""Evaluation"""       ,       prediction_loss_only=True if compute_metrics is None else None       ,       ignore_keys=__magic_name__       ,       )
						finally:
									UpperCamelCase         =					compute_metrics

						if self.post_process_function is not None and self.compute_metrics is not None:
									UpperCamelCase         =					self.post_process_function(__magic_name__       ,       __magic_name__       ,       output.predictions						)
									UpperCamelCase         =					self.compute_metrics(__magic_name__						)

									# Prefix all keys with metric_key_prefix + '_'
									for key in list(metrics.keys()						):
												if not key.startswith(F'{metric_key_prefix}_'						):
															UpperCamelCase         =					metrics.pop(__magic_name__						)

									self.log(__magic_name__						)
						else:
									UpperCamelCase         =					{}

						if self.args.tpu_metrics_debug or self.args.debug:
									# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
									xm.master_print(met.metrics_report()						)

						UpperCamelCase         =					self.callback_handler.on_evaluate(self.args       ,       self.state       ,       self.control       ,       __magic_name__						)
						return metrics




			def   lowerCamelCase_					(						self							:		Dict       ,       __magic_name__							:		List[str]       ,       __magic_name__							:		List[Any]       ,       __magic_name__							:		List[Any]=None       ,       __magic_name__							:		str = "test"						):







						"""simple docstring"""







						UpperCamelCase         =					self.get_test_dataloader(__magic_name__						)

						# Temporarily disable metric computation, we will do it in the loop here.
						UpperCamelCase         =					self.compute_metrics
						UpperCamelCase         =					None
						UpperCamelCase         =					self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
						try:
									UpperCamelCase         =					eval_loop(
									    __magic_name__       ,       description="""Prediction"""       ,       prediction_loss_only=True if compute_metrics is None else None       ,       ignore_keys=__magic_name__       ,       )
						finally:
									UpperCamelCase         =					compute_metrics

						if self.post_process_function is None or self.compute_metrics is None:
									return output

						UpperCamelCase         =					self.post_process_function(__magic_name__       ,       __magic_name__       ,       output.predictions       ,       """predict"""						)
						UpperCamelCase         =					self.compute_metrics(__magic_name__						)

						# Prefix all keys with metric_key_prefix + '_'
						for key in list(metrics.keys()						):
									if not key.startswith(F'{metric_key_prefix}_'						):
												UpperCamelCase         =					metrics.pop(__magic_name__						)

						return PredictionOutput(predictions=predictions.predictions       ,       label_ids=predictions.label_ids       ,       metrics=__magic_name__						)





			def   lowerCamelCase_					(						self							:		List[str]       ,       __magic_name__							:		Optional[Any]="./"						):







						"""simple docstring"""







						UpperCamelCase         =					self.eval_dataset
						UpperCamelCase         =					self.get_eval_dataloader(__magic_name__						)

						UpperCamelCase         =					next(iter(__magic_name__						)						)

						# saving device - to make it consistent
						UpperCamelCase         =					torch.device("""cuda""" if torch.cuda.is_available() else """cpu"""						)

						# convert to tuple
						UpperCamelCase         =					tuple(v.to(__magic_name__						) for k, v in batch.items()						)

						logger.info("""Converting model to be onnx compatible"""						)
						from pytorch_quantization.nn import TensorQuantizer

						UpperCamelCase         =					True

						UpperCamelCase         =					self.model.to(__magic_name__						)

						model.eval()
						model.float()

						UpperCamelCase         =					model.module if hasattr(__magic_name__       ,       """module"""						) else model
						quant_trainer.configure_model(__magic_name__       ,       self.quant_trainer_args						)

						UpperCamelCase         =					os.path.join(__magic_name__       ,       """model.onnx"""						)
						logger.info(F'exporting model to {output_model_file}'						)

						UpperCamelCase         =					{0: """batch_size""", 1: """seq_len"""}

						torch.onnx.export(
						    __magic_name__       ,       __magic_name__       ,       __magic_name__       ,       export_params=__magic_name__       ,       opset_version=1_3       ,       do_constant_folding=__magic_name__       ,       input_names=["""input_ids""", """attention_mask""", """token_type_ids"""]       ,       output_names=["""output_start_logits""", """output_end_logits"""]       ,       dynamic_axes={
						        """input_ids""": axes,
						        """attention_mask""": axes,
						        """token_type_ids""": axes,
						        """output_start_logits""": axes,
						        """output_end_logits""": axes,
						    }       ,       verbose=__magic_name__       ,       )
						logger.info("""onnx export finished"""						)




