




'''simple docstring'''





import torch
from torch import nn

from ...configuration_utils import ConfigMixin, register_to_config
from ...models import ModelMixin


class    lowerCamelCase    (		_lowerCamelCase				,_lowerCamelCase						):






			'''simple docstring'''





			@register_to_config
			def __init__( self			:   Tuple       ,						*,
			    lowerCamelCase_			:   int = 4       ,						lowerCamelCase_			:   int = 768       ,						lowerCamelCase_			:   int       ,						lowerCamelCase_			:   List[Any]       ,						)						->   Dict:
								super().__init__()

								__magic_name__							:	Optional[int]    							=							nn.Parameter(torch.zeros(lowerCamelCase_  )  )

								# parameters for additional clip time embeddings
								__magic_name__							:	Union[str, Any]    							=							nn.Linear(lowerCamelCase_       ,						lowerCamelCase_  )
								__magic_name__							:	Optional[int]    							=							nn.Linear(lowerCamelCase_       ,						lowerCamelCase_  )

								# parameters for encoder hidden states
								__magic_name__							:	str    							=							clip_extra_context_tokens
								__magic_name__							:	List[Any]    							=							nn.Linear(
								    lowerCamelCase_       ,						self.clip_extra_context_tokens * cross_attention_dim  )
								__magic_name__							:	Union[str, Any]    							=							nn.Linear(lowerCamelCase_       ,						lowerCamelCase_  )
								__magic_name__							:	Tuple    							=							nn.LayerNorm(lowerCamelCase_  )


			def 							UpperCAmelCase__       ( self			:   List[Any]       ,						*, lowerCamelCase_			:   Tuple       ,						lowerCamelCase_			:   Dict       ,						lowerCamelCase_			:   Union[str, Any]       ,						lowerCamelCase_			:   List[str]  )						->   Dict:
								if do_classifier_free_guidance:
													# Add the classifier free guidance embeddings to the image embeddings
													__magic_name__							:	str    							=							image_embeddings.shape[0]
													__magic_name__							:	Optional[Any]    							=							self.learned_classifier_free_guidance_embeddings.unsqueeze(0  )
													__magic_name__							:	Dict    							=							classifier_free_guidance_embeddings.expand(
													    lowerCamelCase_       ,						-1  )
													__magic_name__							:	Optional[Any]    							=							torch.cat([classifier_free_guidance_embeddings, image_embeddings]       ,						dim=0  )

								# The image embeddings batch size and the text embeddings batch size are equal
								assert image_embeddings.shape[0] == prompt_embeds.shape[0]

								__magic_name__							:	Tuple    							=							prompt_embeds.shape[0]

								# "Specifically, we modify the architecture described in Nichol et al. (2021) by projecting and
								# adding CLIP embeddings to the existing timestep embedding, ...
								__magic_name__							:	Optional[int]    							=							self.embedding_proj(lowerCamelCase_  )
								__magic_name__							:	Tuple    							=							self.clip_image_embeddings_project_to_time_embeddings(lowerCamelCase_  )
								__magic_name__							:	str    							=							time_projected_image_embeddings + time_projected_prompt_embeds

								# ... and by projecting CLIP embeddings into four
								# extra tokens of context that are concatenated to the sequence of outputs from the GLIDE text encoder"
								__magic_name__							:	Optional[int]    							=							self.clip_extra_context_tokens_proj(lowerCamelCase_  )
								__magic_name__							:	Dict    							=							clip_extra_context_tokens.reshape(lowerCamelCase_       ,						-1       ,						self.clip_extra_context_tokens  )
								__magic_name__							:	int    							=							clip_extra_context_tokens.permute(0       ,						2       ,						1  )

								__magic_name__							:	Dict    							=							self.encoder_hidden_states_proj(lowerCamelCase_  )
								__magic_name__							:	int    							=							self.text_encoder_hidden_states_norm(lowerCamelCase_  )
								__magic_name__							:	Optional[int]    							=							torch.cat([clip_extra_context_tokens, text_encoder_hidden_states]       ,						dim=1  )

								return text_encoder_hidden_states, additive_clip_time_embeddings









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


__lowerCamelCase   :						Dict				=			WebClient(token=os.environ['''CI_SLACK_BOT_TOKEN'''])







def 		lowercase__   (		__A:	List[Any]				):





					'''simple docstring'''






					__magic_name__							:	int    							=							test_results.split(''' '''				)

					__magic_name__							:	Optional[Any]    							=							0
					__magic_name__							:	int    							=							0

					# When the output is short enough, the output is surrounded by = signs: "== OUTPUT =="
					# When it is too long, those signs are not present.
					__magic_name__							:	Optional[Any]    							=							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 		lowercase__   (		__A:	Dict				):





					'''simple docstring'''






					__magic_name__							:	Tuple    							=							{}
					__magic_name__							:	List[Any]    							=							None
					__magic_name__							:	int    							=							False
					for line in failures_short_lines.split('''\n'''				):
										if re.search(r'''_ \[doctest\]'''							,__A				):
															__magic_name__							:	Dict    							=							True
															__magic_name__							:	Any    							=							line.split(''' '''				)[2]
										elif in_error and not line.split(''' '''				)[0].isdigit():
															__magic_name__							:	List[Any]    							=							line
															__magic_name__							:	List[Any]    							=							False

					return failures


class    lowerCamelCase    :






			'''simple docstring'''





			def __init__( self			:   Union[str, Any]       ,						lowerCamelCase_			:   str       ,						lowerCamelCase_			:   Dict  )						->   List[str]:
								__magic_name__							:	Union[str, Any]    							=							title

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

								# Failures and success of the modeling tests
								__magic_name__							:	List[Any]    							=							doc_test_results

			@property
			def 							UpperCAmelCase__       ( self			:   List[str]  )						->   str:
								__magic_name__							:	Union[str, Any]    							=							[self._time_spent]
								__magic_name__							:	str    							=							0

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

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

								__magic_name__							,    __magic_name__							,    __magic_name__							:	Dict    							=							total_secs // 3600, (total_secs % 3600) // 60, total_secs % 60
								return F'''{int(lowerCamelCase_  )}h{int(lowerCamelCase_  )}m{int(lowerCamelCase_  )}s'''

			@property
			def 							UpperCAmelCase__       ( self			:   Dict  )						->   Dict:
								return {"type": "header", "text": {"type": "plain_text", "text": self.title}}

			@property
			def 							UpperCAmelCase__       ( self			:   Optional[Any]  )						->   Dict:
								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 							UpperCAmelCase__       ( self			:   Union[str, Any]  )						->   Dict:
								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 							UpperCAmelCase__       ( self			:   Dict  )						->   Dict:
								__magic_name__							:	List[Any]    							=							40
								__magic_name__							:	Optional[Any]    							=							{k: v['''failed'''] for k, v in doc_test_results.items() if isinstance(lowerCamelCase_       ,						lowerCamelCase_  )}

								__magic_name__							:	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 							UpperCAmelCase__       ( self			:   Dict  )						->   str:
								__magic_name__							:	List[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 							UpperCAmelCase__       ( )						->   List[Any]:
								__magic_name__							:	Dict    							=							[
								    {
								        '''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 							UpperCAmelCase__       ( self			:   Union[str, Any]  )						->   Union[str, Any]:
								print('''Sending the following payload'''  )
								print(json.dumps({'''blocks''': json.loads(self.payload  )}  )  )

								__magic_name__							:	Tuple    							=							F'''{self.n_failures} failures out of {self.n_tests} tests,''' if self.n_failures else '''All tests passed.'''

								__magic_name__							:	List[Any]    							=							client.chat_postMessage(
								    channel=os.environ['''CI_SLACK_CHANNEL_ID_DAILY''']       ,						blocks=self.payload       ,						text=lowerCamelCase_       ,						)

			def 							UpperCAmelCase__       ( self			:   Tuple       ,						lowerCamelCase_			:   List[Any]       ,						lowerCamelCase_			:   Optional[int]       ,						lowerCamelCase_			:   Any       ,						lowerCamelCase_			:   Any  )						->   Optional[Any]:
								__magic_name__							:	Optional[Any]    							=							''''''
								for key, value in failures.items():
													__magic_name__							:	int    							=							value[:200] + ''' [Truncated]''' if len(lowerCamelCase_  ) > 250 else value
													failures_text += F'''*{key}*\n_{value}_\n\n'''

								__magic_name__							:	Dict    							=							job_name
								__magic_name__							:	Tuple    							=							{'''type''': '''section''', '''text''': {'''type''': '''mrkdwn''', '''text''': text}}

								if job_link is not None:
													__magic_name__							:	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 							UpperCAmelCase__       ( self			:   Any  )						->   Tuple:
								if self.thread_ts is None:
													raise ValueError('''Can only post reply if a post has been made.'''  )

								__magic_name__							:	List[str]    							=							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'''  )

								__magic_name__							:	int    							=							sorted(self.doc_test_results.items()       ,						key=lambda lowerCamelCase_  : t[0]  )
								for job, job_result in sorted_dict:
													if len(job_result['''failures''']  ):
																		__magic_name__							:	List[Any]    							=							F'''*Num failures* :{len(job_result['failed']  )} \n'''
																		__magic_name__							:	str    							=							job_result['''failures''']
																		__magic_name__							:	str    							=							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 		lowercase__   (		):





					'''simple docstring'''






					__magic_name__							:	List[Any]    							=							os.environ['''GITHUB_RUN_ID''']
					__magic_name__							:	Optional[int]    							=							F'''https://api.github.com/repos/huggingface/transformers/actions/runs/{run_id}/jobs?per_page=100'''
					__magic_name__							:	Any    							=							requests.get(__A				).json()
					__magic_name__							:	List[str]    							=							{}

					try:
										jobs.update({job['''name''']: job['''html_url'''] for job in result['''jobs''']}				)
										__magic_name__							:	Optional[int]    							=							math.ceil((result['''total_count'''] - 1_0_0) / 1_0_0				)

										for i in range(__A				):
															__magic_name__							:	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 		lowercase__   (		__A:	str				):





					'''simple docstring'''






					__magic_name__							:	List[str]    							=							{}

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

					return _artifact







def 		lowercase__   (		):





					'''simple docstring'''






					class    lowerCamelCase    :






								'''simple docstring'''





								def __init__( self			:   Optional[Any]       ,						lowerCamelCase_			:   str  )						->   int:
													__magic_name__							:	int    							=							name
													__magic_name__							:	Dict    							=							[]

								def __str__( self			:   List[str]  )						->   Union[str, Any]:
													return self.name


								def 							UpperCAmelCase__       ( self			:   List[Any]       ,						lowerCamelCase_			:   str  )						->   Optional[Any]:
													self.paths.append({'''name''': self.name, '''path''': path}  )






					__magic_name__							:	Dict[str, Artifact]    							=							{}

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

															_available_artifacts[artifact_name].add_path(__A				)

					return _available_artifacts


if __name__ == "__main__":
	__lowerCamelCase   :						Optional[int]				=			get_job_links()
	__lowerCamelCase   :						Tuple				=			retrieve_available_artifacts()

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

	# Link to the GitHub Action job
	__lowerCamelCase   :						Optional[int]				=			github_actions_job_links.get('''run_doctests''')

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

		__lowerCamelCase   :						Dict				=			extract_first_line_failure(artifact['''failures_short'''])
		for line in artifact["summary_short"].split('''\n'''):
			if re.search('''FAILED''', line):
				__lowerCamelCase   :						List[Any]				=			line.replace('''FAILED ''', '''''')
				__lowerCamelCase   :						int				=			line.split()[0].replace('''\n''', '''''')

				if "::" in line:
					__lowerCamelCase	,		__lowerCamelCase   :						List[str]				=			line.split('''::''')
				else:
					__lowerCamelCase	,		__lowerCamelCase   :						Optional[int]				=			line, line

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

						__lowerCamelCase   :						List[Any]				=			all_failures[test] if test in all_failures else '''N/A'''
						__lowerCamelCase   :						int				=			failure
						break

	__lowerCamelCase   :						List[Any]				=			Message('''🤗 Results of the doc tests.''', doc_test_results)
	message.post()
	message.post_reply()



