






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   UpperCamelCase  (       snake_case__     :   Any  ):




     '''simple docstring'''






     __snake_case			:Union[str, Any]          =						[]
     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   UpperCamelCase  (       snake_case__     :   Optional[int]	,snake_case__     :   Optional[Any]  ):




     '''simple docstring'''






     __snake_case			:Optional[int]          =						[]
     for d in reversed(__snake_case  ):
          idx.append(flat_idx % d  )
          __snake_case			:Optional[Any]          =						flat_idx // d

     return tuple(reversed(__snake_case  )  )






@torch.jit.ignore
def   UpperCamelCase  (       snake_case__     :   List[str]	,snake_case__     :   Any	,snake_case__     :   Union[str, Any]	,snake_case__     :   str = None	,snake_case__     :   Optional[int] = None	,):




     '''simple docstring'''






     def reduce_edge_list(snake_case__     :   Union[str, Any]  ) -> None:
          __snake_case			:Optional[Any]          =						True
          for i in range(len(__snake_case  )  ):
               __snake_case			:Union[str, Any]          =						-1 * (i + 1)
               l[reversed_idx] &= tally
               __snake_case			:Union[str, Any]          =						l[reversed_idx]

     if start_edges is None:
          __snake_case			:int          =						[s == 0 for s in start]
          reduce_edge_list(__snake_case  )
     if end_edges is None:
          __snake_case			:List[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  ),)]

     __snake_case			:Any          =						[]
     __snake_case			:Optional[Any]          =						[]

     # 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

     __snake_case			:Tuple          =						tuple(__snake_case  )
     __snake_case			:Union[str, Any]          =						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

          __snake_case			:List[Any]          =						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

          __snake_case			:Tuple          =						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()  )
          __snake_case			: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   UpperCamelCase  (       snake_case__     :   str	,snake_case__     :   Optional[Any]	,snake_case__     :   str	,snake_case__     :   List[str]  ):




     '''simple docstring'''






     __snake_case			:Optional[Any]          =						t.shape[:no_batch_dims]
     __snake_case			:List[str]          =						list(_flat_idx_to_idx(__snake_case	,__snake_case  )  )
     # _get_minimal_slice_set is inclusive
     __snake_case			:Any          =						list(_flat_idx_to_idx(flat_end - 1	,__snake_case  )  )

     # Get an ordered list of slices to perform
     __snake_case			:Any          =						_get_minimal_slice_set(
         __snake_case	,__snake_case	,__snake_case	,)

     __snake_case			: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   UpperCamelCase  (       snake_case__     :   List[Any]	,snake_case__     :   Tuple	,snake_case__     :   List[str]	,snake_case__     :   Optional[Any]	,snake_case__     :   Dict = False	,snake_case__     :   str = None	,snake_case__     :   Dict = False	,):




     '''simple docstring'''






     if not (len(__snake_case  ) > 0):
          raise ValueError("""Must provide at least one input"""  )

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

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

     __snake_case			:List[str]          =						tensor_tree_map(_prep_inputs	,__snake_case  )
     __snake_case			:List[str]          =						None
     if _out is not None:
          __snake_case			:List[str]          =						tensor_tree_map(lambda snake_case__  : t.view([-1] + list(t.shape[no_batch_dims:]  )  )	,_out  )

     __snake_case			:Optional[int]          =						1
     for d in orig_batch_dims:
          flat_batch_dim *= d

     __snake_case			:Union[str, Any]          =						flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

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

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

          __snake_case			:Tuple          =						tensor_tree_map(__snake_case	,__snake_case  )

          # Run the layer on the chunk
          __snake_case			:Optional[Any]          =						layer(**__snake_case  )

          # Allocate space for the output
          if out is None:
               __snake_case			:int          =						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__     :   List[Any]	,snake_case__     :   List[str]  ) -> 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:
                                   __snake_case			: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:
                         __snake_case			:List[Any]          =						xa
          elif isinstance(__snake_case	,torch.Tensor  ):
               if _add_into_out:
                    out[i : i + chunk_size] += output_chunk
               else:
                    __snake_case			:str          =						output_chunk
          else:
               raise ValueError("""Not supported"""  )

          i += chunk_size

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

     return out




class 			snake_case__   :

       '''simple docstring'''



       def __init__(							self			,						a__ = 5_12			,						) ->      List[Any]:






            '''simple docstring'''





            __snake_case			:str          =						max_chunk_size
            __snake_case			:Optional[int]          =						None
            __snake_case			:List[str]          =						None



       def 					__lowercase						(							self			,						a__			,						a__			,						a__    ) ->      int:






            '''simple docstring'''





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

            if min_chunk_size >= self.max_chunk_size:
                 return min_chunk_size

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

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

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

            return candidates[min_viable_chunk_size_index]



       def 					__lowercase						(							self			,						a__			,						a__    ) ->      bool:






            '''simple docstring'''





            __snake_case			:List[Any]          =						True
            for aa, aa in zip(__a			,						__a    ):
                 assert type(__a    ) == type(__a    )
                 if isinstance(__a			,						(list, tuple)    ):
                      consistent &= self._compare_arg_caches(__a			,						__a    )
                 elif isinstance(__a			,						__a    ):
                      __snake_case			:List[Any]          =						[v for _, v in sorted(aa.items()			,						key=lambda a__    : x[0]    )]
                      __snake_case			:Union[str, Any]          =						[v for _, v in sorted(aa.items()			,						key=lambda a__    : x[0]    )]
                      consistent &= self._compare_arg_caches(__a			,						__a    )
                 else:
                      consistent &= aa == aa

            return consistent



       def 					__lowercase						(							self			,						a__			,						a__			,						a__			,						) ->      int:






            '''simple docstring'''





            __snake_case			:List[str]          =						True
            __snake_case			:List[str]          =						tree_map(lambda a__    : a.shape if isinstance(__a			,						torch.Tensor    ) else a			,						__a			,						__a    )
            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(__a    )
                 __snake_case			:Any          =						self._compare_arg_caches(self.cached_arg_data			,						__a    )
            else:
                 # Otherwise, we can reuse the precomputed value
                 __snake_case			:Dict          =						False

            if not consistent:
                 __snake_case			:Any          =						self._determine_favorable_chunk_size(
                     __a			,						__a			,						__a			,						)
                 __snake_case			:List[str]          =						arg_data

            assert self.cached_chunk_size is not None

            return self.cached_chunk_size









import copy
import os
from typing import TYPE_CHECKING, List, Union


if TYPE_CHECKING:
     pass

from ...configuration_utils import PretrainedConfig
from ...utils import logging


lowerCamelCase__							=     logging.get_logger(__name__)

lowerCamelCase__							=     {
    """kakaobrain/align-base""": """https://huggingface.co/kakaobrain/align-base/resolve/main/config.json""",
}




class 			snake_case__   (							lowercase_):

       '''simple docstring'''
       lowerCamelCase							:				Union[str, Any] 				= "align_text_model"



       def __init__(							self			,						a__=3_05_22			,						a__=7_68			,						a__=12			,						a__=12			,						a__=30_72			,						a__="gelu"			,						a__=0.1			,						a__=0.1			,						a__=5_12			,						a__=2			,						a__=0.02			,						a__=1e-12			,						a__=0			,						a__="absolute"			,						a__=True			,						**a__			,						) ->      List[str]:






            '''simple docstring'''





            super().__init__(**a__    )

            __snake_case			:Optional[int]          =						vocab_size
            __snake_case			:List[str]          =						hidden_size
            __snake_case			:Optional[Any]          =						num_hidden_layers
            __snake_case			:int          =						num_attention_heads
            __snake_case			:Optional[Any]          =						hidden_act
            __snake_case			:Union[str, Any]          =						intermediate_size
            __snake_case			:int          =						hidden_dropout_prob
            __snake_case			:Optional[Any]          =						attention_probs_dropout_prob
            __snake_case			:List[str]          =						max_position_embeddings
            __snake_case			:List[str]          =						type_vocab_size
            __snake_case			:Union[str, Any]          =						initializer_range
            __snake_case			:str          =						layer_norm_eps
            __snake_case			:Any          =						position_embedding_type
            __snake_case			:List[str]          =						use_cache
            __snake_case			:Optional[int]          =						pad_token_id



       @classmethod
       def 					__lowercase						(							cls			,						a__			,						**a__    ) ->      "PretrainedConfig":






            '''simple docstring'''





            cls._set_token_in_kwargs(a__    )

            __snake_case       , __snake_case			:Tuple          =						cls.get_config_dict(a__			,						**a__    )

            # get the text config dict if we are loading from AlignConfig
            if config_dict.get("""model_type"""    ) == "align":
                 __snake_case			:Any          =						config_dict["""text_config"""]

            if "model_type" in config_dict and hasattr(cls			,						"""model_type"""    ) and config_dict["model_type"] != cls.model_type:
                 logger.warning(
                     F'''You are using a model of type {config_dict['model_type']} to instantiate a model of type '''
                     F'''{cls.model_type}. This is not supported for all configurations of models and can yield errors.'''    )

            return cls.from_dict(a__			,						**a__    )




class 			snake_case__   (							lowercase_):

       '''simple docstring'''
       lowerCamelCase							:				Optional[int] 				= "align_vision_model"



       def __init__(							self			,						a__ = 3			,						a__ = 6_00			,						a__ = 2.0			,						a__ = 3.1			,						a__ = 8			,						a__ = [3, 3, 5, 3, 5, 5, 3]			,						a__ = [32, 16, 24, 40, 80, 1_12, 1_92]			,						a__ = [16, 24, 40, 80, 1_12, 1_92, 3_20]			,						a__ = []			,						a__ = [1, 2, 2, 2, 1, 2, 1]			,						a__ = [1, 2, 2, 3, 3, 4, 1]			,						a__ = [1, 6, 6, 6, 6, 6, 6]			,						a__ = 0.25			,						a__ = "swish"			,						a__ = 25_60			,						a__ = "mean"			,						a__ = 0.02			,						a__ = 0.0_01			,						a__ = 0.99			,						a__ = 0.2			,						**a__			,						) ->      List[Any]:






            '''simple docstring'''





            super().__init__(**a__    )

            __snake_case			:Union[str, Any]          =						num_channels
            __snake_case			:List[str]          =						image_size
            __snake_case			:int          =						width_coefficient
            __snake_case			:int          =						depth_coefficient
            __snake_case			:List[Any]          =						depth_divisor
            __snake_case			:Any          =						kernel_sizes
            __snake_case			:Optional[int]          =						in_channels
            __snake_case			:Optional[int]          =						out_channels
            __snake_case			:int          =						depthwise_padding
            __snake_case			:List[str]          =						strides
            __snake_case			:Union[str, Any]          =						num_block_repeats
            __snake_case			:Dict          =						expand_ratios
            __snake_case			:Union[str, Any]          =						squeeze_expansion_ratio
            __snake_case			:Any          =						hidden_act
            __snake_case			:Optional[Any]          =						hidden_dim
            __snake_case			:Union[str, Any]          =						pooling_type
            __snake_case			:Union[str, Any]          =						initializer_range
            __snake_case			:Optional[Any]          =						batch_norm_eps
            __snake_case			:List[Any]          =						batch_norm_momentum
            __snake_case			:Optional[int]          =						drop_connect_rate
            __snake_case			:Union[str, Any]          =						sum(a__    ) * 4



       @classmethod
       def 					__lowercase						(							cls			,						a__			,						**a__    ) ->      "PretrainedConfig":






            '''simple docstring'''





            cls._set_token_in_kwargs(a__    )

            __snake_case       , __snake_case			:int          =						cls.get_config_dict(a__			,						**a__    )

            # get the vision config dict if we are loading from AlignConfig
            if config_dict.get("""model_type"""    ) == "align":
                 __snake_case			:str          =						config_dict["""vision_config"""]

            if "model_type" in config_dict and hasattr(cls			,						"""model_type"""    ) and config_dict["model_type"] != cls.model_type:
                 logger.warning(
                     F'''You are using a model of type {config_dict['model_type']} to instantiate a model of type '''
                     F'''{cls.model_type}. This is not supported for all configurations of models and can yield errors.'''    )

            return cls.from_dict(a__			,						**a__    )




class 			snake_case__   (							lowercase_):

       '''simple docstring'''
       lowerCamelCase							:				str 				= "align"
       lowerCamelCase							:				Union[str, Any] 				= True



       def __init__(							self			,						a__=None			,						a__=None			,						a__=6_40			,						a__=1.0			,						a__=0.02			,						**a__			,						) ->      Dict:






            '''simple docstring'''





            super().__init__(**a__    )

            if text_config is None:
                 __snake_case			:Union[str, Any]          =						{}
                 logger.info("""text_config is None. Initializing the AlignTextConfig with default values."""    )

            if vision_config is None:
                 __snake_case			:str          =						{}
                 logger.info("""vision_config is None. Initializing the AlignVisionConfig with default values."""    )

            __snake_case			:List[Any]          =						AlignTextConfig(**a__    )
            __snake_case			:Tuple          =						AlignVisionConfig(**a__    )

            __snake_case			:Tuple          =						projection_dim
            __snake_case			:int          =						temperature_init_value
            __snake_case			:Any          =						initializer_range



       @classmethod
       def 					__lowercase						(							cls			,						a__			,						a__			,						**a__    ) ->      str:






            '''simple docstring'''





            return cls(text_config=text_config.to_dict()			,						vision_config=vision_config.to_dict()			,						**a__    )



       def 					__lowercase						(							self    ) ->      Dict:






            '''simple docstring'''





            __snake_case			:Optional[Any]          =						copy.deepcopy(self.__dict__    )
            __snake_case			:Dict          =						self.text_config.to_dict()
            __snake_case			:Union[str, Any]          =						self.vision_config.to_dict()
            __snake_case			:List[Any]          =						self.__class__.model_type
            return output


