




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


lowercase__							:	Optional[int]											=		WebClient(token=os.environ["CI_SLACK_BOT_TOKEN"])


def 		lowerCamelCase__       (						_A				):







	'''simple docstring'''





	snake_case_						       =	test_results.split(" "				)

	snake_case_						       =	0
	snake_case_						       =	0

	# When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
	# When it is too long, those signs are not present.
	snake_case_						       =	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 		lowerCamelCase__       (						_A				):







	'''simple docstring'''





	snake_case_						       =	{}
	snake_case_						       =	None
	snake_case_						       =	False
	for line in failures_short_lines.split("\n"				):
		if re.search(R"_ \[doctest\]"	,      _A				):
			snake_case_						       =	True
			snake_case_						       =	line.split(" "				)[2]
		elif in_error and not line.split(" "				)[0].isdigit():
			snake_case_						       =	line
			snake_case_						       =	False

	return failures
class 						UpperCAmelCase  :







						'''simple docstring'''





						def __init__(		self				:	Union[str, Any]			,      __lowercase				:	str			,      __lowercase				:	Dict		):






							"""simple docstring"""






							snake_case_						       =	title

							snake_case_						       =	doc_test_results["time_spent"].split(","		)[0]
							snake_case_						       =	doc_test_results["success"]
							snake_case_						       =	doc_test_results["failures"]
							snake_case_						       =	self.n_success + self.n_failures

							# Failures and success of the modeling tests
							snake_case_						       =	doc_test_results






						@property
						def 			snake_case__		(		self				:	Dict		):






							"""simple docstring"""






							snake_case_						       =	[self._time_spent]
							snake_case_						       =	0

							for time in time_spent:
								snake_case_						       =	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(__lowercase		) == 1:
									snake_case_						       =	[0, 0, time_parts[0]]

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

							snake_case_       , snake_case_       , snake_case_						       =	total_secs // 36_00, (total_secs % 36_00) // 60, total_secs % 60
							return f"{int(__lowercase		)}h{int(__lowercase		)}m{int(__lowercase		)}s"






						@property
						def 			snake_case__		(		self				:	Optional[int]		):






							"""simple docstring"""






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






						@property
						def 			snake_case__		(		self				:	Optional[int]		):






							"""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 			snake_case__		(		self				:	List[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 			snake_case__		(		self				:	str		):






							"""simple docstring"""






							snake_case_						       =	40
							snake_case_						       =	{k: v["failed"] for k, v in doc_test_results.items() if isinstance(__lowercase			,      __lowercase		)}

							snake_case_						       =	""
							for category, failures in category_failures.items():
								if len(__lowercase		) == 0:
									continue

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

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

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






						@property
						def 			snake_case__		(		self				:	Union[str, Any]		):






							"""simple docstring"""






							snake_case_						       =	[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(__lowercase		)






						@staticmethod
						def 			snake_case__		(		):






							"""simple docstring"""






							snake_case_						       =	[
							    {
							        "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(__lowercase		)}		)		)

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






						def 			snake_case__		(		self				:	Tuple		):






							"""simple docstring"""






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

							snake_case_						       =	f"{self.n_failures} failures out of {self.n_tests} tests," if self.n_failures else "All tests passed."

							snake_case_						       =	client.chat_postMessage(
							    channel=os.environ["CI_SLACK_CHANNEL_ID_DAILY"]			,      blocks=self.payload			,      text=__lowercase			,      )






						def 			snake_case__		(		self				:	Any			,      __lowercase				:	Dict			,      __lowercase				:	Tuple			,      __lowercase				:	List[str]			,      __lowercase				:	Any		):






							"""simple docstring"""






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

							snake_case_						       =	job_name
							snake_case_						       =	{"type": "section", "text": {"type": "mrkdwn", "text": text}}

							if job_link is not None:
								snake_case_						       =	{
								    "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 			snake_case__		(		self				:	Dict		):






							"""simple docstring"""






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

							snake_case_						       =	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"		)

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

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

									time.sleep(1		)







def 		lowerCamelCase__       (						):







	'''simple docstring'''





	snake_case_						       =	os.environ["GITHUB_RUN_ID"]
	snake_case_						       =	f"https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100"
	snake_case_						       =	requests.get(_A				).json()
	snake_case_						       =	{}

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

		for i in range(_A				):
			snake_case_						       =	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 		lowerCamelCase__       (						_A				):







	'''simple docstring'''





	snake_case_						       =	{}

	if os.path.exists(_A				):
		snake_case_						       =	os.listdir(_A				)
		for file in files:
			try:
				with open(os.path.join(_A	,      _A				)	,      encoding="utf-8"				) as f:
					snake_case_						       =	f.read()
			except UnicodeDecodeError as e:
				raise ValueError(f"Could not open {os.path.join(_A	,      _A				)}."				) from e

	return _artifact


def 		lowerCamelCase__       (						):







	'''simple docstring'''





	class 						UpperCAmelCase  :







							'''simple docstring'''





							def __init__(		self				:	Union[str, Any]			,      __lowercase				:	str		):






								"""simple docstring"""






								snake_case_						       =	name
								snake_case_						       =	[]






							def __str__(		self				:	List[Any]		):






								"""simple docstring"""






								return self.name






							def 			snake_case__		(		self				:	Optional[Any]			,      __lowercase				:	str		):






								"""simple docstring"""






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







	snake_case_						       =	{}

	snake_case_						       =	filter(os.path.isdir	,      os.listdir()				)
	for directory in directories:
		snake_case_						       =	directory
		if artifact_name not in _available_artifacts:
			snake_case_						       =	Artifact(_A				)

			_available_artifacts[artifact_name].add_path(_A				)

	return _available_artifacts


if __name__ == "__main__":
		lowercase__							:	Tuple											=		get_job_links()
		lowercase__							:	Dict											=		retrieve_available_artifacts()

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

		# Link to the GitHub Action job
		lowercase__							:	Union[str, Any]											=		github_actions_job_links.get("run_doctests")

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

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

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

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

												lowercase__							:	Optional[int]											=		all_failures[test] if test in all_failures else "N/A"
												lowercase__							:	Optional[Any]											=		failure
												break

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






import argparse

import torch

from transformers import MobileBertConfig, MobileBertForPreTraining, load_tf_weights_in_mobilebert
from transformers.utils import logging


logging.set_verbosity_info()


def 		lowerCamelCase__       (						_A	,      _A	,      _A				):







	'''simple docstring'''





	snake_case_						       =	MobileBertConfig.from_json_file(_A				)
	print(f"Building PyTorch model from configuration: {config}"				)
	snake_case_						       =	MobileBertForPreTraining(_A				)
	# Load weights from tf checkpoint
	snake_case_						       =	load_tf_weights_in_mobilebert(_A	,      _A	,      _A				)
	# Save pytorch-model
	print(f"Save PyTorch model to {pytorch_dump_path}"				)
	torch.save(model.state_dict()	,      _A				)


if __name__ == "__main__":
		lowercase__							:	Dict											=		argparse.ArgumentParser()
		# Required parameters
		parser.add_argument(
		    "--tf_checkpoint_path", default=None, type=str, required=True, help="Path to the TensorFlow checkpoint path."
		)
		parser.add_argument(
		    "--mobilebert_config_file",
		    default=None,
		    type=str,
		    required=True,
		    help=(
		        "The config json file corresponding to the pre-trained MobileBERT model. \n"
		        "This specifies the model architecture."
		    ),
		)
		parser.add_argument(
		    "--pytorch_dump_path", default=None, type=str, required=True, help="Path to the output PyTorch model."
		)
		lowercase__							:	Tuple											=		parser.parse_args()
		convert_tf_checkpoint_to_pytorch(args.tf_checkpoint_path, args.mobilebert_config_file, args.pytorch_dump_path)
