




import logging

import torch
from torch import nn
from torch.nn import CrossEntropyLoss, MSELoss

from transformers.file_utils import add_start_docstrings, add_start_docstrings_to_model_forward
from transformers.models.albert.modeling_albert import (
    ALBERT_INPUTS_DOCSTRING,
    ALBERT_START_DOCSTRING,
    AlbertModel,
    AlbertPreTrainedModel,
    AlbertTransformer,
)


__a  :		Union[str, Any]		     =   logging.getLogger(__name__)







class 				UpperCAmelCase(					snake_case_					):



	"""simple docstring"""






	def    __a     (					self	,				lowerCamelCase	,				lowerCamelCase	,				lowerCamelCase=None	,				lowerCamelCase=None    ) ->      List[str]:





								"""simple docstring"""


								if current_layer == 0:
															lowercase__  :	str        =							self.embedding_hidden_mapping_in(lowerCamelCase    )
								else:
															lowercase__  :	List[Any]        =							hidden_states[0]

								lowercase__  :	Optional[Any]        =							int(self.config.num_hidden_layers / self.config.num_hidden_groups    )

								# Index of the hidden group
								lowercase__  :	Any        =							int(current_layer / (self.config.num_hidden_layers / self.config.num_hidden_groups)    )

								lowercase__  :	Union[str, Any]        =							self.albert_layer_groups[group_idx](
								    lowerCamelCase	,				lowerCamelCase	,				head_mask[group_idx * layers_per_group : (group_idx + 1) * layers_per_group]	,				)
								lowercase__  :	Tuple        =							layer_group_output[0]

								return (hidden_states,)







@add_start_docstrings(
    """The bare ALBERT Model transformer with PABEE outputting raw hidden-states without any specific head on top."""   ,     snake_case_   ,     )
class 				UpperCAmelCase(					snake_case_					):



	"""simple docstring"""






	def __init__(					self	,				lowerCamelCase    ) ->      Union[str, Any]:





								"""simple docstring"""


								super().__init__(lowerCamelCase    )

								lowercase__  :	Optional[int]        =							AlbertTransformerWithPabee(lowerCamelCase    )

								self.init_weights()
								lowercase__  :	List[str]        =							0
								lowercase__  :	List[str]        =							0
								lowercase__  :	Any        =							0

								lowercase__  :	Union[str, Any]        =							0

	def    __a     (					self	,				lowerCamelCase    ) ->      Optional[Any]:





								"""simple docstring"""


								lowercase__  :	Optional[Any]        =							threshold

	def    __a     (					self	,				lowerCamelCase    ) ->      Union[str, Any]:





								"""simple docstring"""


								lowercase__  :	int        =							patience

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





								"""simple docstring"""


								lowercase__  :	Optional[Any]        =							0
								lowercase__  :	Tuple        =							0

	def    __a     (					self    ) ->      List[str]:





								"""simple docstring"""


								lowercase__  :	int        =							self.inference_layers_num / self.inference_instances_num
								lowercase__  :	Any        =							(
								    f"""*** Patience = {self.patience} Avg. Inference Layers = {avg_inf_layers:.2f} Speed Up ="""
								    f""" {1 - avg_inf_layers / self.config.num_hidden_layers:.2f} ***"""
								)
								print(lowerCamelCase    )







	@add_start_docstrings_to_model_forward(lowerCamelCase    )
	def    __a     (					self	,				lowerCamelCase=None	,				lowerCamelCase=None	,				lowerCamelCase=None	,				lowerCamelCase=None	,				lowerCamelCase=None	,				lowerCamelCase=None	,				lowerCamelCase=None	,				lowerCamelCase=None	,				lowerCamelCase=False	,				) ->      List[Any]:





								"""simple docstring"""


								if input_ids is not None and inputs_embeds is not None:
															raise ValueError("You cannot specify both input_ids and inputs_embeds at the same time"    )
								elif input_ids is not None:
															lowercase__  :	Dict        =							input_ids.size()
								elif inputs_embeds is not None:
															lowercase__  :	Optional[int]        =							inputs_embeds.size()[:-1]
								else:
															raise ValueError("You have to specify either input_ids or inputs_embeds"    )

								lowercase__  :	Tuple        =							input_ids.device if input_ids is not None else inputs_embeds.device

								if attention_mask is None:
															lowercase__  :	Dict        =							torch.ones(lowerCamelCase	,				device=lowerCamelCase    )
								if token_type_ids is None:
															lowercase__  :	List[Any]        =							torch.zeros(lowerCamelCase	,				dtype=torch.long	,				device=lowerCamelCase    )

								lowercase__  :	List[Any]        =							attention_mask.unsqueeze(1    ).unsqueeze(2    )
								lowercase__  :	Tuple        =							extended_attention_mask.to(dtype=self.dtype    )  # fp16 compatibility
								lowercase__  :	Union[str, Any]        =							(1.0 - extended_attention_mask) * -1_00_00.0
								lowercase__  :	Tuple        =							self.get_head_mask(lowerCamelCase	,				self.config.num_hidden_layers    )

								lowercase__  :	Optional[int]        =							self.embeddings(
								    lowerCamelCase	,				position_ids=lowerCamelCase	,				token_type_ids=lowerCamelCase	,				inputs_embeds=lowerCamelCase    )
								lowercase__  :	Tuple        =							embedding_output

								if self.training:
															lowercase__  :	Optional[int]        =							[]
															for i in range(self.config.num_hidden_layers    ):
																						lowercase__  :	List[str]        =							self.encoder.adaptive_forward(
																						    lowerCamelCase	,				current_layer=lowerCamelCase	,				attention_mask=lowerCamelCase	,				head_mask=lowerCamelCase	,				)

																						lowercase__  :	Optional[int]        =							self.pooler_activation(self.pooler(encoder_outputs[0][:, 0]    )    )
																						lowercase__  :	Dict        =							output_layers[i](output_dropout(lowerCamelCase    )    )
																						res.append(lowerCamelCase    )
								elif self.patience == 0:  # Use all layers for inference
															lowercase__  :	int        =							self.encoder(lowerCamelCase	,				lowerCamelCase	,				head_mask=lowerCamelCase    )
															lowercase__  :	Dict        =							self.pooler_activation(self.pooler(encoder_outputs[0][:, 0]    )    )
															lowercase__  :	Union[str, Any]        =							[output_layers[self.config.num_hidden_layers - 1](lowerCamelCase    )]
								else:
															lowercase__  :	Optional[Any]        =							0
															lowercase__  :	Dict        =							None
															lowercase__  :	Optional[Any]        =							0
															for i in range(self.config.num_hidden_layers    ):
																						calculated_layer_num += 1
																						lowercase__  :	Optional[int]        =							self.encoder.adaptive_forward(
																						    lowerCamelCase	,				current_layer=lowerCamelCase	,				attention_mask=lowerCamelCase	,				head_mask=lowerCamelCase	,				)

																						lowercase__  :	Any        =							self.pooler_activation(self.pooler(encoder_outputs[0][:, 0]    )    )
																						lowercase__  :	Optional[int]        =							output_layers[i](lowerCamelCase    )
																						if regression:
																													lowercase__  :	str        =							logits.detach()
																													if patient_result is not None:
																																				lowercase__  :	List[str]        =							patient_result.detach()
																													if (patient_result is not None) and torch.abs(patient_result - labels    ) < self.regression_threshold:
																																				patient_counter += 1
																													else:
																																				lowercase__  :	int        =							0
																						else:
																													lowercase__  :	Dict        =							logits.detach().argmax(dim=1    )
																													if patient_result is not None:
																																				lowercase__  :	int        =							patient_result.detach().argmax(dim=1    )
																													if (patient_result is not None) and torch.all(labels.eq(lowerCamelCase    )    ):
																																				patient_counter += 1
																													else:
																																				lowercase__  :	List[str]        =							0

																						lowercase__  :	Optional[Any]        =							logits
																						if patient_counter == self.patience:
																													break
															lowercase__  :	Dict        =							[patient_result]
															self.inference_layers_num += calculated_layer_num
															self.inference_instances_num += 1

								return res







@add_start_docstrings(
    """Albert Model transformer with PABEE and a sequence classification/regression head on top (a linear layer on top of
    the pooled output) e.g. for GLUE tasks. """   ,     snake_case_   ,     )
class 				UpperCAmelCase(					snake_case_					):



	"""simple docstring"""






	def __init__(					self	,				lowerCamelCase    ) ->      Union[str, Any]:





								"""simple docstring"""


								super().__init__(lowerCamelCase    )
								lowercase__  :	Any        =							config.num_labels

								lowercase__  :	List[str]        =							AlbertModelWithPabee(lowerCamelCase    )
								lowercase__  :	Optional[Any]        =							nn.Dropout(config.classifier_dropout_prob    )
								lowercase__  :	List[Any]        =							nn.ModuleList(
								    [nn.Linear(config.hidden_size	,				self.config.num_labels    ) for _ in range(config.num_hidden_layers    )]    )

								self.init_weights()







	@add_start_docstrings_to_model_forward(lowerCamelCase    )
	def    __a     (					self	,				lowerCamelCase=None	,				lowerCamelCase=None	,				lowerCamelCase=None	,				lowerCamelCase=None	,				lowerCamelCase=None	,				lowerCamelCase=None	,				lowerCamelCase=None	,				) ->      Optional[int]:





								"""simple docstring"""


								lowercase__  :	List[Any]        =							self.albert(
								    input_ids=lowerCamelCase	,				attention_mask=lowerCamelCase	,				token_type_ids=lowerCamelCase	,				position_ids=lowerCamelCase	,				head_mask=lowerCamelCase	,				inputs_embeds=lowerCamelCase	,				output_dropout=self.dropout	,				output_layers=self.classifiers	,				regression=self.num_labels == 1	,				)

								lowercase__  :	List[str]        =							(logits[-1],)

								if labels is not None:
															lowercase__  :	Optional[int]        =							None
															lowercase__  :	Any        =							0
															for ix, logits_item in enumerate(lowerCamelCase    ):
																						if self.num_labels == 1:
																													#  We are doing regression
																													lowercase__  :	Any        =							MSELoss()
																													lowercase__  :	Optional[int]        =							loss_fct(logits_item.view(-1    )	,				labels.view(-1    )    )
																						else:
																													lowercase__  :	List[Any]        =							CrossEntropyLoss()
																													lowercase__  :	Tuple        =							loss_fct(logits_item.view(-1	,				self.num_labels    )	,				labels.view(-1    )    )
																						if total_loss is None:
																													lowercase__  :	Any        =							loss
																						else:
																													total_loss += loss * (ix + 1)
																						total_weights += ix + 1
															lowercase__  :	str        =							(total_loss / total_weights,) + outputs

								return outputs





import hashlib
import importlib.util
import json
import os
import pathlib
from types import ModuleType

import pytest
import requests

__a  :		str		     =   pathlib.Path.cwd().joinpath('''project_euler''')
__a  :		Tuple		     =   pathlib.Path.cwd().joinpath(
    '''scripts''', '''project_euler_answers.json'''
)

with open(PROJECT_EULER_ANSWERS_PATH) as file_handle:
						__a  :		dict[str, str]		     =   json.load(file_handle)
def  snake_case_					(		SCREAMING_SNAKE_CASE_    )      ->			ModuleType:
							lowercase__  :	List[str]        =							importlib.util.spec_from_file_location(file_path.name    ,str(SCREAMING_SNAKE_CASE_    )    )
							lowercase__  :	int        =							importlib.util.module_from_spec(SCREAMING_SNAKE_CASE_    )  # type: ignore
							spec.loader.exec_module(SCREAMING_SNAKE_CASE_    )  # type: ignore
							return module
def  snake_case_					(		)      ->			list[pathlib.Path]:
							lowercase__  :	List[str]        =							[]
							for problem_dir_path in PROJECT_EULER_DIR_PATH.iterdir():
														if problem_dir_path.is_file() or problem_dir_path.name.startswith("_"    ):
																					continue
														for file_path in problem_dir_path.iterdir():
																					if file_path.suffix != ".py" or file_path.name.startswith(("_", "test")    ):
																												continue
																					solution_file_paths.append(SCREAMING_SNAKE_CASE_    )
							return solution_file_paths
def  snake_case_					(		)      ->			str:
							with open(os.environ["GITHUB_EVENT_PATH"]    ) as file:
														lowercase__  :	str        =							json.load(SCREAMING_SNAKE_CASE_    )
							return event["pull_request"]["url"] + "/files"
def  snake_case_					(		)      ->			list[pathlib.Path]:
							lowercase__  :	Optional[Any]        =							[]
							lowercase__  :	Dict        =							{
							    "Accept": "application/vnd.github.v3+json",
							    "Authorization": "token " + os.environ["GITHUB_TOKEN"],
							}
							lowercase__  :	Union[str, Any]        =							requests.get(get_files_url()    ,headers=SCREAMING_SNAKE_CASE_    ).json()
							for file in files:
														lowercase__  :	Optional[Any]        =							pathlib.Path.cwd().joinpath(file["filename"]    )
														if (
														    filepath.suffix != ".py"
														    or filepath.name.startswith(("_", "test")    )
														    or not filepath.name.startswith("sol"    )
														):
																					continue
														solution_file_paths.append(SCREAMING_SNAKE_CASE_    )
							return solution_file_paths
def  snake_case_					(		)      ->			list[pathlib.Path]:
							if os.environ.get("CI"    ) and os.environ.get("GITHUB_EVENT_NAME"    ) == "pull_request":
														# Return only if there are any, otherwise default to all solutions
														if filepaths := added_solution_file_path():
																					return filepaths
							return all_solution_file_paths()




@pytest.mark.parametrize(
    "solution_path"    ,collect_solution_file_paths()    ,ids=lambda SCREAMING_SNAKE_CASE_    : F"""{path.parent.name}/{path.name}"""    ,)
def  snake_case_					(		SCREAMING_SNAKE_CASE_    )      ->			None:
							lowercase__  :	str        =							solution_path.parent.name[8:].zfill(3    )
							lowercase__  :	str        =							PROBLEM_ANSWERS[problem_number]
							lowercase__  :	Optional[Any]        =							convert_path_to_module(SCREAMING_SNAKE_CASE_    )
							lowercase__  :	str        =							str(solution_module.solution()    )  # type: ignore
							lowercase__  :	List[str]        =							hashlib.shaaaa(answer.encode()    ).hexdigest()
							assert (
							    answer == expected
							), F"""Expected solution to {problem_number} to have hash {expected}, got {answer}"""