
'''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


__snake_case:  str								=			WebClient(token=os.environ["CI_SLACK_BOT_TOKEN"])






def 	_snake_case						(      A_       :    str   ):
  """simple docstring"""







  a_      :	Any      =							test_results.split(""" """   )

  a_      :	str      =							0
  a_      :	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.
  a_      :	Optional[int]      =							expressions[-2] if """=""" in expressions[-1] else expressions[-1]

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

  return failed, success, time_spent






def 	_snake_case						(      A_       :    int   ):
  """simple docstring"""







  a_      :	Tuple      =							{}
  a_      :	Tuple      =							None
  a_      :	str      =							False
  for line in failures_short_lines.split("""\n"""   ):
    if re.search(R"""_ \[doctest\]"""     ,		A_   ):
      a_      :	List[str]      =							True
      a_      :	Union[str, Any]      =							line.split(""" """   )[2]
    elif in_error and not line.split(""" """   )[0].isdigit():
      a_      :	str      =							line
      a_      :	List[Any]      =							False

  return failures


class  _UpperCAmelCase :

     """simple docstring"""

     def __init__(      self							,		lowerCAmelCase_							,		lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	str      =							title

       a_      :	Any      =							doc_test_results["""time_spent"""].split(""",""" )[0]
       a_      :	Union[str, Any]      =							doc_test_results["""success"""]
       a_      :	List[Any]      =							doc_test_results["""failures"""]
       a_      :	Union[str, Any]      =							self.n_success + self.n_failures

       # Failures and success of the modeling tests
       a_      :	Optional[int]      =							doc_test_results


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

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

       for time in time_spent:
         a_      :	List[str]      =							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:
           a_      :	Any      =							[0, 0, time_parts[0]]

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

       a_      , a_      , a_      :	Any      =							total_secs // 36_00, (total_secs % 36_00) // 60, total_secs % 60
       return f'''{int(lowerCAmelCase_ )}h{int(lowerCAmelCase_ )}m{int(lowerCAmelCase_ )}s'''


     @property
     def 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

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


     @property
     def 							_lowerCAmelCase  (      self ):

       '''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 							_lowerCAmelCase  (      self ):

       '''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 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	int      =							40
       a_      :	Tuple      =							{k: v["""failed"""] for k, v in doc_test_results.items() if isinstance(lowerCAmelCase_							,		lowerCAmelCase_ )}

       a_      :	Optional[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 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

       a_      :	str      =							[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 							_lowerCAmelCase  (      ):

       '''simple docstring'''

       a_      :	Tuple      =							[
           {
               """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 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

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

       a_      :	List[Any]      =							f'''{self.n_failures} failures out of {self.n_tests} tests,''' if self.n_failures else """All tests passed."""

       a_      :	Any      =							client.chat_postMessage(
           channel=os.environ["""CI_SLACK_CHANNEL_ID_DAILY"""]							,		blocks=self.payload							,		text=lowerCAmelCase_							,		)


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_ ):

       '''simple docstring'''

       a_      :	int      =							""""""
       for key, value in failures.items():
         a_      :	Tuple      =							value[:2_00] + """ [Truncated]""" if len(lowerCAmelCase_ ) > 2_50 else value
         failures_text += f'''*{key}*\n_{value}_\n\n'''

       a_      :	Union[str, Any]      =							job_name
       a_      :	Optional[int]      =							{"""type""": """section""", """text""": {"""type""": """mrkdwn""", """text""": text}}

       if job_link is not None:
         a_      :	int      =							{
             """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 							_lowerCAmelCase  (      self ):

       '''simple docstring'''

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

       a_      :	Dict      =							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""" )

       a_      :	Optional[int]      =							sorted(self.doc_test_results.items()							,		key=lambda lowerCAmelCase_ : t[0] )
       for job, job_result in sorted_dict:
         if len(job_result["""failures"""] ):
           a_      :	Dict      =							f'''*Num failures* :{len(job_result['failed'] )} \n'''
           a_      :	Any      =							job_result["""failures"""]
           a_      :	Tuple      =							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 	_snake_case						(      ):
  """simple docstring"""







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

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

    for i in range(A_   ):
      a_      :	Optional[Any]      =							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."""     ,		A_   )

  return {}






def 	_snake_case						(      A_       :    str   ):
  """simple docstring"""







  a_      :	List[str]      =							{}

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

  return _artifact






def 	_snake_case						(      ):
  """simple docstring"""







  class  _UpperCAmelCase :

       """simple docstring"""

       def __init__(      self							,		lowerCAmelCase_ ):

         '''simple docstring'''

         a_      :	List[str]      =							name
         a_      :	int      =							[]


       def __str__(      self ):

         '''simple docstring'''

         return self.name






       def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_ ):

         '''simple docstring'''

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







  a_      :	Dict[str, Artifact]      =							{}

  a_      :	Dict      =							filter(os.path.isdir     ,		os.listdir()   )
  for directory in directories:
    a_      :	List[Any]      =							directory
    if artifact_name not in _available_artifacts:
      a_      :	str      =							Artifact(A_   )

      _available_artifacts[artifact_name].add_path(A_   )

  return _available_artifacts


if __name__ == "__main__":
    __snake_case:  Optional[int]								=			get_job_links()
    __snake_case:  str								=			retrieve_available_artifacts()

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

    # Link to the GitHub Action job
    __snake_case:  int								=			github_actions_job_links.get("run_doctests")

    __snake_case:  Tuple								=			available_artifacts["doc_tests_gpu_test_reports"].paths[0]
    __snake_case:  str								=			retrieve_artifact(artifact_path["name"])
    if "stats" in artifact:
        __snake_case							,__snake_case							,__snake_case:  int								=			handle_test_results(artifact["stats"])
        __snake_case:  List[Any]								=			failed
        __snake_case:  Optional[int]								=			success
        __snake_case:  Union[str, Any]								=			time_spent[1:-1] + ", "

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

                if "::" in line:
                    __snake_case							,__snake_case:  Any								=			line.split("::")
                else:
                    __snake_case							,__snake_case:  Any								=			line, line

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

                        __snake_case:  List[Any]								=			all_failures[test] if test in all_failures else "N/A"
                        __snake_case:  Union[str, Any]								=			failure
                        break

    __snake_case:  Tuple								=			Message("🤗 Results of the doc tests.", doc_test_results)
    message.post()
    message.post_reply()




'''simple docstring'''

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:  Dict								=			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 ( lowerCAmelCase__    ):

     """simple docstring"""

     def __init__(      self							,		*lowerCAmelCase_							,		lowerCAmelCase_=None							,		lowerCAmelCase_=None							,		lowerCAmelCase_=None							,		**lowerCAmelCase_ ):

       '''simple docstring'''

       super().__init__(*lowerCAmelCase_							,		**lowerCAmelCase_ )
       a_      :	Tuple      =							eval_examples
       a_      :	Optional[Any]      =							post_process_function
       a_      :	List[str]      =							quant_trainer_args
       a_      :	List[str]      =							1_28  # default number of calibration samples


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_=None ):

       '''simple docstring'''

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

       a_      :	Dict      =							self._remove_unused_columns(lowerCAmelCase_							,		description="""Calibration""" )

       return DataLoader(
           lowerCAmelCase_							,		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=lowerCAmelCase_							,		)


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_=None ):

       '''simple docstring'''

       a_      :	Union[str, Any]      =							self.train_dataset if calib_dataset is None else calib_dataset
       a_      :	int      =							self.get_calib_dataloader(lowerCAmelCase_ )

       a_      :	Dict      =							self.model
       quant_trainer.configure_model(lowerCAmelCase_							,		self.quant_trainer_args							,		calib=lowerCAmelCase_ )
       model.eval()
       quant_trainer.enable_calibration(lowerCAmelCase_ )

       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(lowerCAmelCase_ ):
         # Prediction step
         a_      , a_      , a_      :	Optional[int]      =							self.prediction_step(lowerCAmelCase_							,		lowerCAmelCase_							,		prediction_loss_only=lowerCAmelCase_ )
         if (step + 1) * calib_dataloader.batch_size >= self.calib_num:
           break

       quant_trainer.finish_calibration(lowerCAmelCase_							,		self.quant_trainer_args )
       a_      :	Optional[int]      =							model


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_=None							,		lowerCAmelCase_=None							,		lowerCAmelCase_=None							,		lowerCAmelCase_ = "eval" ):

       '''simple docstring'''

       a_      :	str      =							self.eval_dataset if eval_dataset is None else eval_dataset
       a_      :	List[str]      =							self.get_eval_dataloader(lowerCAmelCase_ )
       a_      :	str      =							self.eval_examples if eval_examples is None else eval_examples

       # Temporarily disable metric computation, we will do it in the loop here.
       a_      :	List[str]      =							self.compute_metrics
       a_      :	Tuple      =							None
       a_      :	Optional[Any]      =							self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
       try:
         a_      :	List[str]      =							eval_loop(
             lowerCAmelCase_							,		description="""Evaluation"""							,		prediction_loss_only=True if compute_metrics is None else None							,		ignore_keys=lowerCAmelCase_							,		)
       finally:
         a_      :	int      =							compute_metrics

       if self.post_process_function is not None and self.compute_metrics is not None:
         a_      :	Optional[Any]      =							self.post_process_function(lowerCAmelCase_							,		lowerCAmelCase_							,		output.predictions )
         a_      :	Tuple      =							self.compute_metrics(lowerCAmelCase_ )

         # Prefix all keys with metric_key_prefix + '_'
         for key in list(metrics.keys() ):
           if not key.startswith(f'''{metric_key_prefix}_''' ):
             a_      :	Optional[int]      =							metrics.pop(lowerCAmelCase_ )

         self.log(lowerCAmelCase_ )
       else:
         a_      :	Any      =							{}

       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() )

       a_      :	Optional[Any]      =							self.callback_handler.on_evaluate(self.args							,		self.state							,		self.control							,		lowerCAmelCase_ )
       return metrics


     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_=None							,		lowerCAmelCase_ = "test" ):

       '''simple docstring'''

       a_      :	List[Any]      =							self.get_test_dataloader(lowerCAmelCase_ )

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

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

       a_      :	Tuple      =							self.post_process_function(lowerCAmelCase_							,		lowerCAmelCase_							,		output.predictions							,		"""predict""" )
       a_      :	List[Any]      =							self.compute_metrics(lowerCAmelCase_ )

       # Prefix all keys with metric_key_prefix + '_'
       for key in list(metrics.keys() ):
         if not key.startswith(f'''{metric_key_prefix}_''' ):
           a_      :	Tuple      =							metrics.pop(lowerCAmelCase_ )

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






     def 							_lowerCAmelCase  (      self							,		lowerCAmelCase_="./" ):

       '''simple docstring'''

       a_      :	Any      =							self.eval_dataset
       a_      :	Tuple      =							self.get_eval_dataloader(lowerCAmelCase_ )

       a_      :	List[str]      =							next(iter(lowerCAmelCase_ ) )

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

       # convert to tuple
       a_      :	Optional[int]      =							tuple(v.to(lowerCAmelCase_ ) for k, v in batch.items() )

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

       a_      :	List[str]      =							True

       a_      :	str      =							self.model.to(lowerCAmelCase_ )

       model.eval()
       model.float()

       a_      :	Optional[int]      =							model.module if hasattr(lowerCAmelCase_							,		"""module""" ) else model
       quant_trainer.configure_model(lowerCAmelCase_							,		self.quant_trainer_args )

       a_      :	Union[str, Any]      =							os.path.join(lowerCAmelCase_							,		"""model.onnx""" )
       logger.info(f'''exporting model to {output_model_file}''' )

       a_      :	Dict      =							{0: """batch_size""", 1: """seq_len"""}

       torch.onnx.export(
           lowerCAmelCase_							,		lowerCAmelCase_							,		lowerCAmelCase_							,		export_params=lowerCAmelCase_							,		opset_version=13							,		do_constant_folding=lowerCAmelCase_							,		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=lowerCAmelCase_							,		)
       logger.info("""onnx export finished""" )


