



"""simple docstring"""






import logging
import math
from functools import partial
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union

import torch

from .tensor_utils import tensor_tree_map, tree_map


def      _lowercase							(    __snake_case      )		->  List[Tuple[int, ...]]:
			__lowerCAmelCase :    str				=			[]
			if isinstance(__snake_case   ,__snake_case      ):
						for v in tree.values():
									shapes.extend(_fetch_dims(__snake_case      )      )
			elif isinstance(__snake_case   ,(list, tuple)      ):
						for t in tree:
									shapes.extend(_fetch_dims(__snake_case      )      )
			elif isinstance(__snake_case   ,torch.Tensor      ):
						shapes.append(tree.shape      )
			else:
						raise ValueError("Not supported"      )

			return shapes


@torch.jit.ignore
def      _lowercase							(    __snake_case   ,__snake_case      )		->  Tuple[int, ...]:
			__lowerCAmelCase :    Union[str, Any]				=			[]
			for d in reversed(__snake_case      ):
						idx.append(flat_idx % d      )
						__lowerCAmelCase :    Optional[Any]				=			flat_idx // d

			return tuple(reversed(__snake_case      )      )


@torch.jit.ignore
def      _lowercase							(    __snake_case   ,__snake_case   ,__snake_case   ,__snake_case = None   ,__snake_case = None   ,)		->  List[Tuple[slice, ...]]:

			# start_edges and end_edges both indicate whether, starting from any given
			# dimension, the start/end index is at the top/bottom edge of the
			# corresponding tensor, modeled as a tree
			def reduce_edge_list(__snake_case      ) -> None:
						__lowerCAmelCase :    int				=			True
						for i in range(len(__snake_case      )      ):
									__lowerCAmelCase :    str				=			-1 * (i + 1)
									l[reversed_idx] &= tally
									__lowerCAmelCase :    Optional[int]				=			l[reversed_idx]

			if start_edges is None:
						__lowerCAmelCase :    Union[str, Any]				=			[s == 0 for s in start]
						reduce_edge_list(__snake_case      )
			if end_edges is None:
						__lowerCAmelCase :    Union[str, Any]				=			[e == (d - 1) for e, d in zip(__snake_case   ,__snake_case      )]
						reduce_edge_list(__snake_case      )

			# Base cases. Either start/end are empty and we're done, or the final,
			# one-dimensional tensor can be simply sliced
			if len(__snake_case      ) == 0:
						return [()]
			elif len(__snake_case      ) == 1:
						return [(slice(start[0]   ,end[0] + 1      ),)]

			__lowerCAmelCase :    List[Tuple[slice, ...]]				=			[]
			__lowerCAmelCase :    List[slice]				=			[]

			# Dimensions common to start and end can be selected directly
			for s, e in zip(__snake_case   ,__snake_case      ):
						if s == e:
									path_list.append(slice(__snake_case   ,s + 1      )      )
						else:
									break

			__lowerCAmelCase :    Tuple[slice, ...]				=			tuple(__snake_case      )
			__lowerCAmelCase :    Optional[int]				=			len(__snake_case      )

			# start == end, and we're done
			if divergence_idx == len(__snake_case      ):
						return [path]

			def upper() -> Tuple[Tuple[slice, ...], ...]:
						assert start_edges is not None
						assert end_edges is not None

						__lowerCAmelCase :    Optional[int]				=			start[divergence_idx]
						return tuple(
						    path + (slice(__snake_case   ,sdi + 1      ),) + s
						    for s in _get_minimal_slice_set(
						        start[divergence_idx + 1 :]   ,[d - 1 for d in dims[divergence_idx + 1 :]]   ,dims[divergence_idx + 1 :]   ,start_edges=start_edges[divergence_idx + 1 :]   ,end_edges=[True for _ in end_edges[divergence_idx + 1 :]]   ,)      )

			def lower() -> Tuple[Tuple[slice, ...], ...]:
						assert start_edges is not None
						assert end_edges is not None

						__lowerCAmelCase :    Any				=			end[divergence_idx]
						return tuple(
						    path + (slice(__snake_case   ,edi + 1      ),) + s
						    for s in _get_minimal_slice_set(
						        [0 for _ in start[divergence_idx + 1 :]]   ,end[divergence_idx + 1 :]   ,dims[divergence_idx + 1 :]   ,start_edges=[True for _ in start_edges[divergence_idx + 1 :]]   ,end_edges=end_edges[divergence_idx + 1 :]   ,)      )

			# If both start and end are at the edges of the subtree rooted at
			# divergence_idx, we can just select the whole subtree at once
			if start_edges[divergence_idx] and end_edges[divergence_idx]:
						slices.append(path + (slice(start[divergence_idx]   ,end[divergence_idx] + 1      ),)      )
			# If just start is at the edge, we can grab almost all of the subtree,
			# treating only the ragged bottom edge as an edge case
			elif start_edges[divergence_idx]:
						slices.append(path + (slice(start[divergence_idx]   ,end[divergence_idx]      ),)      )
						slices.extend(lower()      )
			# Analogous to the previous case, but the top is ragged this time
			elif end_edges[divergence_idx]:
						slices.extend(upper()      )
						slices.append(path + (slice(start[divergence_idx] + 1   ,end[divergence_idx] + 1      ),)      )
			# If both sides of the range are ragged, we need to handle both sides
			# separately. If there's contiguous meat in between them, we can index it
			# in one big chunk
			else:
						slices.extend(upper()      )
						__lowerCAmelCase :    str				=			end[divergence_idx] - start[divergence_idx]
						if middle_ground > 1:
									slices.append(path + (slice(start[divergence_idx] + 1   ,end[divergence_idx]      ),)      )
						slices.extend(lower()      )

			return slices


@torch.jit.ignore
def      _lowercase							(    __snake_case   ,__snake_case   ,__snake_case   ,__snake_case      )		->  torch.Tensor:
			__lowerCAmelCase :    str				=			t.shape[:no_batch_dims]
			__lowerCAmelCase :    Optional[int]				=			list(_flat_idx_to_idx(__snake_case   ,__snake_case      )      )
			# _get_minimal_slice_set is inclusive
			__lowerCAmelCase :    int				=			list(_flat_idx_to_idx(flat_end - 1   ,__snake_case      )      )

			# Get an ordered list of slices to perform
			__lowerCAmelCase :    Optional[int]				=			_get_minimal_slice_set(
			    __snake_case   ,__snake_case   ,__snake_case   ,)

			__lowerCAmelCase :    Union[str, Any]				=			[t[s] for s in slices]

			return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]      ) for s in sliced_tensors]      )


def      _lowercase							(    __snake_case   ,__snake_case   ,__snake_case   ,__snake_case   ,__snake_case = False   ,__snake_case = None   ,__snake_case = False   ,)		->  Any:
			if not (len(__snake_case      ) > 0):
						raise ValueError("Must provide at least one input"      )

			__lowerCAmelCase :    List[Any]				=			[shape[:no_batch_dims] for shape in _fetch_dims(__snake_case      )]
			__lowerCAmelCase :    Any				=			tuple([max(__snake_case      ) for s in zip(*__snake_case      )]      )

			def _prep_inputs(__snake_case      ) -> torch.Tensor:
						if not low_mem:
									if not sum(t.shape[:no_batch_dims]      ) == no_batch_dims:
												__lowerCAmelCase :    int				=			t.expand(orig_batch_dims + t.shape[no_batch_dims:]      )
									__lowerCAmelCase :    Tuple				=			t.reshape(-1   ,*t.shape[no_batch_dims:]      )
						else:
									__lowerCAmelCase :    Optional[Any]				=			t.expand(orig_batch_dims + t.shape[no_batch_dims:]      )
						return t

			__lowerCAmelCase :    Dict[str, Any]				=			tensor_tree_map(_prep_inputs   ,__snake_case      )
			__lowerCAmelCase :    Optional[Any]				=			None
			if _out is not None:
						__lowerCAmelCase :    Optional[int]				=			tensor_tree_map(lambda __snake_case      : t.view([-1] + list(t.shape[no_batch_dims:]      )      )   ,_out      )

			__lowerCAmelCase :    Optional[Any]				=			1
			for d in orig_batch_dims:
						flat_batch_dim *= d

			__lowerCAmelCase :    Tuple				=			flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

			def _select_chunk(__snake_case      ) -> torch.Tensor:
						return t[i : i + chunk_size] if t.shape[0] != 1 else t

			__lowerCAmelCase :    int				=			0
			__lowerCAmelCase :    List[str]				=			prepped_outputs
			for _ in range(__snake_case      ):
						# Chunk the input
						if not low_mem:
									__lowerCAmelCase :    Union[str, Any]				=			_select_chunk
						else:
									__lowerCAmelCase :    Union[str, Any]				=			partial(
									    _chunk_slice   ,flat_start=__snake_case   ,flat_end=min(__snake_case   ,i + chunk_size      )   ,no_batch_dims=len(__snake_case      )   ,)

						__lowerCAmelCase :    Dict[str, Any]				=			tensor_tree_map(__snake_case   ,__snake_case      )

						# Run the layer on the chunk
						__lowerCAmelCase :    List[str]				=			layer(**__snake_case      )

						# Allocate space for the output
						if out is None:
									__lowerCAmelCase :    Tuple				=			tensor_tree_map(lambda __snake_case      : t.new_zeros((flat_batch_dim,) + t.shape[1:]      )   ,__snake_case      )

						# Put the chunk in its pre-allocated space
						if isinstance(__snake_case   ,__snake_case      ):

									def assign(__snake_case   ,__snake_case      ) -> None:
												for k, v in da.items():
															if isinstance(__snake_case   ,__snake_case      ):
																		assign(__snake_case   ,da[k]      )
															else:
																		if _add_into_out:
																					v[i : i + chunk_size] += da[k]
																		else:
																					__lowerCAmelCase :    Union[str, Any]				=			da[k]

									assign(__snake_case   ,__snake_case      )
						elif isinstance(__snake_case   ,__snake_case      ):
									for xa, xa in zip(__snake_case   ,__snake_case      ):
												if _add_into_out:
															xa[i : i + chunk_size] += xa
												else:
															__lowerCAmelCase :    Optional[Any]				=			xa
						elif isinstance(__snake_case   ,torch.Tensor      ):
									if _add_into_out:
												out[i : i + chunk_size] += output_chunk
									else:
												__lowerCAmelCase :    List[Any]				=			output_chunk
						else:
									raise ValueError("Not supported"      )

						i += chunk_size

			__lowerCAmelCase :    Tuple				=			tensor_tree_map(lambda __snake_case      : t.view(orig_batch_dims + t.shape[1:]      )   ,__snake_case      )

			return out





class      A__  :




	'''simple docstring'''





	def __init__(	self:						Dict			,     _SCREAMING_SNAKE_CASE:						int = 512			,     )							->			Dict:

				"""simple docstring"""

				__lowerCAmelCase :    List[Any]				=			max_chunk_size
				__lowerCAmelCase :    Optional[int]				=			None
				__lowerCAmelCase :    Optional[tuple]				=			None

	def 				_SCREAMING_SNAKE_CASE  (	self:						Dict			,     _SCREAMING_SNAKE_CASE:						Callable			,     _SCREAMING_SNAKE_CASE:						tuple			,     _SCREAMING_SNAKE_CASE:						int)							->			int:

				"""simple docstring"""

				logging.info("Tuning chunk size...")

				if min_chunk_size >= self.max_chunk_size:
							return min_chunk_size

				__lowerCAmelCase :    List[int]				=			[2**l for l in range(int(math.log(self.max_chunk_size			,     2)) + 1)]
				__lowerCAmelCase :    Any				=			[c for c in candidates if c > min_chunk_size]
				__lowerCAmelCase :    int				=			[min_chunk_size] + candidates
				candidates[-1] += 4

				def test_chunk_size(_SCREAMING_SNAKE_CASE:						int) -> bool:
							try:
										with torch.no_grad():
													fn(*_SCREAMING_SNAKE_CASE			,     chunk_size=_SCREAMING_SNAKE_CASE)
										return True
							except RuntimeError:
										return False

				__lowerCAmelCase :    int				=			0
				__lowerCAmelCase :    Optional[Any]				=			len(_SCREAMING_SNAKE_CASE) - 1
				while i > min_viable_chunk_size_index:
							__lowerCAmelCase :    List[Any]				=			test_chunk_size(candidates[i])
							if not viable:
										__lowerCAmelCase :    List[str]				=			(min_viable_chunk_size_index + i) // 2
							else:
										__lowerCAmelCase :    str				=			i
										__lowerCAmelCase :    Tuple				=			(i + len(_SCREAMING_SNAKE_CASE) - 1) // 2

				return candidates[min_viable_chunk_size_index]

	def 				_SCREAMING_SNAKE_CASE  (	self:						str			,     _SCREAMING_SNAKE_CASE:						Iterable			,     _SCREAMING_SNAKE_CASE:						Iterable)							->			bool:

				"""simple docstring"""

				__lowerCAmelCase :    Optional[Any]				=			True
				for aa, aa in zip(_SCREAMING_SNAKE_CASE			,     _SCREAMING_SNAKE_CASE):
							assert type(_SCREAMING_SNAKE_CASE) == type(_SCREAMING_SNAKE_CASE)
							if isinstance(_SCREAMING_SNAKE_CASE			,     (list, tuple)):
										consistent &= self._compare_arg_caches(_SCREAMING_SNAKE_CASE			,     _SCREAMING_SNAKE_CASE)
							elif isinstance(_SCREAMING_SNAKE_CASE			,     _SCREAMING_SNAKE_CASE):
										__lowerCAmelCase :    Optional[int]				=			[v for _, v in sorted(aa.items()			,     key=lambda _SCREAMING_SNAKE_CASE: x[0])]
										__lowerCAmelCase :    Optional[int]				=			[v for _, v in sorted(aa.items()			,     key=lambda _SCREAMING_SNAKE_CASE: x[0])]
										consistent &= self._compare_arg_caches(_SCREAMING_SNAKE_CASE			,     _SCREAMING_SNAKE_CASE)
							else:
										consistent &= aa == aa

				return consistent







	def 				_SCREAMING_SNAKE_CASE  (	self:						int			,     _SCREAMING_SNAKE_CASE:						Callable			,     _SCREAMING_SNAKE_CASE:						tuple			,     _SCREAMING_SNAKE_CASE:						int			,     )							->			int:

				"""simple docstring"""

				__lowerCAmelCase :    Any				=			True
				__lowerCAmelCase :    tuple				=			tree_map(lambda _SCREAMING_SNAKE_CASE: a.shape if isinstance(_SCREAMING_SNAKE_CASE			,     torch.Tensor) else a			,     _SCREAMING_SNAKE_CASE			,     _SCREAMING_SNAKE_CASE)
				if self.cached_arg_data is not None:
							# If args have changed shape/value, we need to re-tune
							assert len(self.cached_arg_data) == len(_SCREAMING_SNAKE_CASE)
							__lowerCAmelCase :    Union[str, Any]				=			self._compare_arg_caches(self.cached_arg_data			,     _SCREAMING_SNAKE_CASE)
				else:
							# Otherwise, we can reuse the precomputed value
							__lowerCAmelCase :    Tuple				=			False

				if not consistent:
							__lowerCAmelCase :    List[Any]				=			self._determine_favorable_chunk_size(
							    _SCREAMING_SNAKE_CASE			,     _SCREAMING_SNAKE_CASE			,     _SCREAMING_SNAKE_CASE			,     )
							__lowerCAmelCase :    Tuple				=			arg_data

				assert self.cached_chunk_size is not None

				return self.cached_chunk_size




"""simple docstring"""






import argparse

import torch

from transformers import BertForMaskedLM


if __name__ == "__main__":
	__snake_case				:  Optional[Any] =       argparse.ArgumentParser(
	    description=(
	        'Extraction some layers of the full BertForMaskedLM or RObertaForMaskedLM for Transfer Learned'
	        ' Distillation'
	    )
	)
	parser.add_argument('--model_type', default='bert', choices=['bert'])
	parser.add_argument('--model_name', default='bert-base-uncased', type=str)
	parser.add_argument('--dump_checkpoint', default='serialization_dir/tf_bert-base-uncased_0247911.pth', type=str)
	parser.add_argument('--vocab_transform', action='store_true')
	__snake_case				:  List[Any] =       parser.parse_args()

	if args.model_type == "bert":
		__snake_case				:  int =       BertForMaskedLM.from_pretrained(args.model_name)
		__snake_case				:  Tuple =       'bert'
	else:
		raise ValueError('args.model_type should be "bert".')

	__snake_case				:  List[Any] =       model.state_dict()
	__snake_case				:  Any =       {}

	for w in ["word_embeddings", "position_embeddings"]:
		__snake_case				:  List[str] =       state_dict[F"""{prefix}.embeddings.{w}.weight"""]
	for w in ["weight", "bias"]:
		__snake_case				:  Optional[int] =       state_dict[F"""{prefix}.embeddings.LayerNorm.{w}"""]

	__snake_case				:  List[str] =       0
	for teacher_idx in [0, 2, 4, 7, 9, 11]:
		for w in ["weight", "bias"]:
			__snake_case				:  str =       state_dict[
			    F"""{prefix}.encoder.layer.{teacher_idx}.attention.self.query.{w}"""
			]
			__snake_case				:  Tuple =       state_dict[
			    F"""{prefix}.encoder.layer.{teacher_idx}.attention.self.key.{w}"""
			]
			__snake_case				:  str =       state_dict[
			    F"""{prefix}.encoder.layer.{teacher_idx}.attention.self.value.{w}"""
			]

			__snake_case				:  Any =       state_dict[
			    F"""{prefix}.encoder.layer.{teacher_idx}.attention.output.dense.{w}"""
			]
			__snake_case				:  Union[str, Any] =       state_dict[
			    F"""{prefix}.encoder.layer.{teacher_idx}.attention.output.LayerNorm.{w}"""
			]

			__snake_case				:  Union[str, Any] =       state_dict[
			    F"""{prefix}.encoder.layer.{teacher_idx}.intermediate.dense.{w}"""
			]
			__snake_case				:  Any =       state_dict[
			    F"""{prefix}.encoder.layer.{teacher_idx}.output.dense.{w}"""
			]
			__snake_case				:  Optional[int] =       state_dict[
			    F"""{prefix}.encoder.layer.{teacher_idx}.output.LayerNorm.{w}"""
			]
		std_idx += 1

	__snake_case				:  Union[str, Any] =       state_dict['cls.predictions.decoder.weight']
	__snake_case				:  Union[str, Any] =       state_dict['cls.predictions.bias']
	if args.vocab_transform:
		for w in ["weight", "bias"]:
			__snake_case				:  Union[str, Any] =       state_dict[F"""cls.predictions.transform.dense.{w}"""]
			__snake_case				:  List[str] =       state_dict[F"""cls.predictions.transform.LayerNorm.{w}"""]

	print(F"""N layers selected for distillation: {std_idx}""")
	print(F"""Number of params transferred for distillation: {len(compressed_sd.keys())}""")

	print(F"""Save transferred checkpoint to {args.dump_checkpoint}.""")
	torch.save(compressed_sd, args.dump_checkpoint)