



"""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				:  Optional[int] =       WebClient(token=os.environ['CI_SLACK_BOT_TOKEN'])


def      _lowercase							(    __snake_case      )		->  List[str]:
			__lowerCAmelCase :    List[Any]				=			test_results.split(" "      )

			__lowerCAmelCase :    List[str]				=			0
			__lowerCAmelCase :    Union[str, Any]				=			0

			# When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
			# When it is too long, those signs are not present.
			__lowerCAmelCase :    Union[str, Any]				=			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      _lowercase							(    __snake_case      )		->  Dict:
			__lowerCAmelCase :    Any				=			{}
			__lowerCAmelCase :    int				=			None
			__lowerCAmelCase :    Optional[int]				=			False
			for line in failures_short_lines.split("\n"      ):
						if re.search(r"_ \[doctest\]"   ,UpperCamelCase__      ):
									__lowerCAmelCase :    Dict				=			True
									__lowerCAmelCase :    Tuple				=			line.split(" "      )[2]
						elif in_error and not line.split(" "      )[0].isdigit():
									__lowerCAmelCase :    Any				=			line
									__lowerCAmelCase :    Dict				=			False

			return failures



class      A__  :




	'''simple docstring'''





	def __init__(	self:						str			,     _SCREAMING_SNAKE_CASE:						str			,     _SCREAMING_SNAKE_CASE:						Dict)							->			Optional[Any]:

				"""simple docstring"""

				__lowerCAmelCase :    Optional[Any]				=			title

				__lowerCAmelCase :    str				=			doc_test_results["time_spent"].split(",")[0]
				__lowerCAmelCase :    Optional[Any]				=			doc_test_results["success"]
				__lowerCAmelCase :    Dict				=			doc_test_results["failures"]
				__lowerCAmelCase :    Optional[int]				=			self.n_success + self.n_failures

				# Failures and success of the modeling tests
				__lowerCAmelCase :    Tuple				=			doc_test_results

	@property
	def 				_SCREAMING_SNAKE_CASE  (	self:						int)							->			str:

				"""simple docstring"""

				__lowerCAmelCase :    str				=			[self._time_spent]
				__lowerCAmelCase :    int				=			0

				for time in time_spent:
							__lowerCAmelCase :    int				=			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(__a) == 1:
										__lowerCAmelCase :    Any				=			[0, 0, time_parts[0]]

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

				__lowerCAmelCase		,      __lowerCAmelCase		,      __lowerCAmelCase :    Optional[int]				=			total_secs // 3600, (total_secs % 3600) // 60, total_secs % 60
				return F"""{int(__a)}h{int(__a)}m{int(__a)}s"""

	@property
	def 				_SCREAMING_SNAKE_CASE  (	self:						str)							->			Dict:

				"""simple docstring"""

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

	@property
	def 				_SCREAMING_SNAKE_CASE  (	self:						Optional[int])							->			Dict:

				"""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 				_SCREAMING_SNAKE_CASE  (	self:						int)							->			Dict:

				"""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 				_SCREAMING_SNAKE_CASE  (	self:						str)							->			Dict:

				"""simple docstring"""

				__lowerCAmelCase :    str				=			40
				__lowerCAmelCase :    Optional[int]				=			{k: v["failed"] for k, v in doc_test_results.items() if isinstance(__a			,     __a)}

				__lowerCAmelCase :    Optional[int]				=			""
				for category, failures in category_failures.items():
							if len(__a) == 0:
										continue

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

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

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

	@property
	def 				_SCREAMING_SNAKE_CASE  (	self:						Tuple)							->			str:

				"""simple docstring"""

				__lowerCAmelCase :    Dict				=			[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(__a)

	@staticmethod
	def 				_SCREAMING_SNAKE_CASE  (	)							->			List[Any]:

				"""simple docstring"""

				__lowerCAmelCase :    str				=			[
				    {
				        "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(__a)}))

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

	def 				_SCREAMING_SNAKE_CASE  (	self:						Union[str, Any])							->			List[str]:

				"""simple docstring"""

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

				__lowerCAmelCase :    str				=			F"""{self.n_failures} failures out of {self.n_tests} tests,""" if self.n_failures else "All tests passed."

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

	def 				_SCREAMING_SNAKE_CASE  (	self:						Optional[int]			,     _SCREAMING_SNAKE_CASE:						int			,     _SCREAMING_SNAKE_CASE:						str			,     _SCREAMING_SNAKE_CASE:						Any			,     _SCREAMING_SNAKE_CASE:						Any)							->			Optional[int]:

				"""simple docstring"""

				__lowerCAmelCase :    int				=			""
				for key, value in failures.items():
							__lowerCAmelCase :    str				=			value[:200] + " [Truncated]" if len(__a) > 250 else value
							failures_text += F"""*{key}*\n_{value}_\n\n"""

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

				if job_link is not None:
							__lowerCAmelCase :    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 				_SCREAMING_SNAKE_CASE  (	self:						str)							->			List[Any]:

				"""simple docstring"""

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

				__lowerCAmelCase :    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")

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

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

										time.sleep(1)






def      _lowercase							(    )		->  Union[str, Any]:
			__lowerCAmelCase :    int				=			os.environ["GITHUB_RUN_ID"]
			__lowerCAmelCase :    Optional[int]				=			F"""https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100"""
			__lowerCAmelCase :    Optional[int]				=			requests.get(UpperCamelCase__      ).json()
			__lowerCAmelCase :    List[Any]				=			{}

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

						for i in range(UpperCamelCase__      ):
									__lowerCAmelCase :    List[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."   ,UpperCamelCase__      )

			return {}


def      _lowercase							(    __snake_case      )		->  Optional[Any]:
			__lowerCAmelCase :    Dict				=			{}

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

			return _artifact


def      _lowercase							(    )		->  Any:



			class      A__  :




				'''simple docstring'''





				def __init__(	self:						Dict			,     _SCREAMING_SNAKE_CASE:						str)							->			str:

							"""simple docstring"""

							__lowerCAmelCase :    Optional[int]				=			name
							__lowerCAmelCase :    Tuple				=			[]

				def __str__(	self:						List[Any])							->			Optional[Any]:

							"""simple docstring"""

							return self.name







				def 				_SCREAMING_SNAKE_CASE  (	self:						List[str]			,     _SCREAMING_SNAKE_CASE:						str)							->			Dict:

							"""simple docstring"""

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






			__lowerCAmelCase :    Any				=			{}

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

									_available_artifacts[artifact_name].add_path(UpperCamelCase__      )

			return _available_artifacts


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

	__snake_case				:  List[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				:  Optional[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				:  Optional[int] =       available_artifacts["doc_tests_gpu_test_reports"].paths[0]
	__snake_case				:  List[Any] =       retrieve_artifact(artifact_path['name'])
	if "stats" in artifact:
		__snake_case				:  int =       handle_test_results(artifact['stats'])
		__snake_case				:  str =       failed
		__snake_case				:  Tuple =       success
		__snake_case				:  List[str] =       time_spent[1:-1] + ", "

		__snake_case				:  str =       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				:  List[str] =       line.split()[0].replace('\n', '')

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

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

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

	__snake_case				:  Any =       Message('🤗 Results of the doc tests.', doc_test_results)
	message.post()
	message.post_reply()
'''simple docstring'''


import logging
from pathlib import Path

import numpy as np
import pytorch_lightning as pl
import torch
from pytorch_lightning.callbacks import EarlyStopping, ModelCheckpoint
from pytorch_lightning.utilities import rank_zero_only
from utils_rag import save_json





def    lowerCAmelCase       (		UpperCamelCase__					: Tuple							):


	"""simple docstring"""
	__UpperCAmelCase      					=      filter(lambda UpperCamelCase__							: p.requires_grad     ,       model.parameters()							)
	__UpperCAmelCase      					=      sum([np.prod(p.size()							) for p in model_parameters]							)
	return params


__lowerCAmelCase						:							List[str]    							=						logging.getLogger(__name__)





def    lowerCAmelCase       (		UpperCamelCase__					: Any     ,       UpperCamelCase__					: str							):


	"""simple docstring"""
	if metric == "rouge2":
		__UpperCAmelCase      					=      '''{val_avg_rouge2:.4f}-{step_count}'''
	elif metric == "bleu":
		__UpperCAmelCase      					=      '''{val_avg_bleu:.4f}-{step_count}'''
	elif metric == "em":
		__UpperCAmelCase      					=      '''{val_avg_em:.4f}-{step_count}'''
	elif metric == "loss":
		__UpperCAmelCase      					=      '''{val_avg_loss:.4f}-{step_count}'''
	else:
		raise NotImplementedError(
		    f"""seq2seq callbacks only support rouge2 and bleu, got {metric}, You can make your own by adding to this"""
		    ''' function.'''							)

	__UpperCAmelCase      					=      ModelCheckpoint(
	    dirpath=UpperCamelCase__     ,       filename=UpperCamelCase__     ,       monitor=f"""val_{metric}"""     ,       mode='''max'''     ,       save_top_k=1     ,       every_n_epochs=1     ,       )

	return checkpoint_callback





def    lowerCAmelCase       (		UpperCamelCase__					: Tuple     ,       UpperCamelCase__					: Optional[Any]							):


	"""simple docstring"""
	return EarlyStopping(
	    monitor=f"""val_{metric}"""     ,       mode='''min''' if '''loss''' in metric else '''max'''     ,       patience=UpperCamelCase__     ,       verbose=UpperCamelCase__     ,       )






class     A ( pl.Callback ):





							def    snake_case__    (   self					:				Union[str, Any]  ,		__a					:				int  ,		__a					:				Any						)				->						Optional[Any]:
								__UpperCAmelCase      					=      {f"""lr_group_{i}""": param['''lr'''] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups						)}
								pl_module.logger.log_metrics(__a						)





							@rank_zero_only
							def    snake_case__    (   self					:				Union[str, Any]  ,		__a					:				pl.Trainer  ,		__a					:				pl.LightningModule  ,		__a					:				str  ,		__a					:				Any=True						)				->						None:
								logger.info(f"""***** {type_path} results at step {trainer.global_step:05d} *****"""						)
								__UpperCAmelCase      					=      trainer.callback_metrics
								trainer.logger.log_metrics({k: v for k, v in metrics.items() if k not in ['''log''', '''progress_bar''', '''preds''']}						)
								# Log results
								__UpperCAmelCase      					=      Path(pl_module.hparams.output_dir						)
								if type_path == "test":
									__UpperCAmelCase      					=      od / '''test_results.txt'''
									__UpperCAmelCase      					=      od / '''test_generations.txt'''
								else:
									# this never gets hit. I prefer not to save intermediate generations, and results are in metrics.json
									# If people want this it will be easy enough to add back.
									__UpperCAmelCase      					=      od / f"""{type_path}_results/{trainer.global_step:05d}.txt"""
									__UpperCAmelCase      					=      od / f"""{type_path}_generations/{trainer.global_step:05d}.txt"""
									results_file.parent.mkdir(exist_ok=__a						)
									generations_file.parent.mkdir(exist_ok=__a						)
								with open(__a  ,		'''a+'''						) as writer:
									for key in sorted(__a						):
										if key in ["log", "progress_bar", "preds"]:
											continue
										__UpperCAmelCase      					=      metrics[key]
										if isinstance(__a  ,		torch.Tensor						):
											__UpperCAmelCase      					=      val.item()
										__UpperCAmelCase      					=      f"""{key}: {val:.6f}\n"""
										writer.write(__a						)

								if not save_generations:
									return

								if "preds" in metrics:
									__UpperCAmelCase      					=      '''\n'''.join(metrics['''preds''']						)
									generations_file.open('''w+'''						).write(__a						)





							@rank_zero_only
							def    snake_case__    (   self					:				List[Any]  ,		__a					:				Optional[int]  ,		__a					:				Union[str, Any]						)				->						List[Any]:
								try:
									__UpperCAmelCase      					=      pl_module.model.model.num_parameters()
								except AttributeError:
									__UpperCAmelCase      					=      pl_module.model.num_parameters()

								__UpperCAmelCase      					=      count_trainable_parameters(__a						)
								# mp stands for million parameters
								trainer.logger.log_metrics({'''n_params''': npars, '''mp''': npars / 1e6, '''grad_mp''': n_trainable_pars / 1e6}						)





							@rank_zero_only
							def    snake_case__    (   self					:				List[str]  ,		__a					:				pl.Trainer  ,		__a					:				pl.LightningModule						)				->						Optional[int]:
								save_json(pl_module.metrics  ,		pl_module.metrics_save_path						)
								return self._write_logs(__a  ,		__a  ,		'''test'''						)






							@rank_zero_only
							def    snake_case__    (   self					:				Any  ,		__a					:				pl.Trainer  ,		__a					:				Union[str, Any]						)				->						Tuple:
								save_json(pl_module.metrics  ,		pl_module.metrics_save_path						)
								# Uncommenting this will save val generations
								# return self._write_logs(trainer, pl_module, "valid")
