
"""simple docstring"""
import logging
import os
import sys
import threading
from logging import (
    CRITICAL,  # NOQA
    DEBUG,  # NOQA
    ERROR,  # NOQA
    FATAL,  # NOQA
    INFO,  # NOQA
    NOTSET,  # NOQA
    WARN,  # NOQA
    WARNING,  # NOQA
)
from typing import Optional

from tqdm import auto as tqdm_lib


lowercase		:						Any =					threading.Lock()
lowercase		:						Optional[logging.Handler] =					None

lowercase		:						Tuple =					{
    'debug': logging.DEBUG,
    'info': logging.INFO,
    'warning': logging.WARNING,
    'error': logging.ERROR,
    'critical': logging.CRITICAL,
}

lowercase		:						List[str] =					logging.WARNING

lowercase		:						Dict =					True


def     __UpperCAmelCase    (	)     ->			str:


					'''simple docstring'''
					__a:						Tuple		    =					os.getenv("""DIFFUSERS_VERBOSITY"""	,				lowerCamelCase_ )
					if env_level_str:
										if env_level_str in log_levels:
															return log_levels[env_level_str]
										else:
															logging.getLogger().warning(
															    f"""Unknown option DIFFUSERS_VERBOSITY={env_level_str}, """
															    f"""has to be one of: { ', '.join(log_levels.keys() ) }""" )
					return _default_log_level


def     __UpperCAmelCase    (	)     ->			str:


					'''simple docstring'''
					return __name__.split(""".""" )[0]


def     __UpperCAmelCase    (	)     ->			logging.Logger:


					'''simple docstring'''
					return logging.getLogger(_get_library_name() )


def     __UpperCAmelCase    (	)     ->			None:


					'''simple docstring'''
					global _default_handler

					with _lock:
										if _default_handler:
															# This library has already configured the library root logger.
															return
										__a:						List[Any]		    =					logging.StreamHandler()  # Set sys.stderr as stream.
										__a:						int		    =					sys.stderr.flush

										# Apply our default configuration to the library root logger.
										__a:						List[Any]		    =					_get_library_root_logger()
										library_root_logger.addHandler(_default_handler )
										library_root_logger.setLevel(_get_default_logging_level() )
										__a:						Dict		    =					False


def     __UpperCAmelCase    (	)     ->			None:


					'''simple docstring'''
					global _default_handler

					with _lock:
										if not _default_handler:
															return

										__a:						int		    =					_get_library_root_logger()
										library_root_logger.removeHandler(_default_handler )
										library_root_logger.setLevel(logging.NOTSET )
										__a:						str		    =					None


def     __UpperCAmelCase    (	)     ->			Dict:


					'''simple docstring'''
					return log_levels


def     __UpperCAmelCase    (	lowerCamelCase_ = None )     ->			logging.Logger:


					'''simple docstring'''

					if name is None:
										__a:						int		    =					_get_library_name()

					_configure_library_root_logger()
					return logging.getLogger(lowerCamelCase_ )


def     __UpperCAmelCase    (	)     ->			int:


					'''simple docstring'''
					_configure_library_root_logger()
					return _get_library_root_logger().getEffectiveLevel()


def     __UpperCAmelCase    (	lowerCamelCase_ )     ->			None:


					'''simple docstring'''
					_configure_library_root_logger()
					_get_library_root_logger().setLevel(lowerCamelCase_ )


def     __UpperCAmelCase    (	)     ->			Optional[int]:


					'''simple docstring'''
					return set_verbosity(lowerCamelCase_ )


def     __UpperCAmelCase    (	)     ->			Optional[int]:


					'''simple docstring'''
					return set_verbosity(lowerCamelCase_ )


def     __UpperCAmelCase    (	)     ->			Tuple:


					'''simple docstring'''
					return set_verbosity(lowerCamelCase_ )


def     __UpperCAmelCase    (	)     ->			List[Any]:


					'''simple docstring'''
					return set_verbosity(lowerCamelCase_ )


def     __UpperCAmelCase    (	)     ->			None:


					'''simple docstring'''
					_configure_library_root_logger()

					assert _default_handler is not None
					_get_library_root_logger().removeHandler(_default_handler )


def     __UpperCAmelCase    (	)     ->			None:


					'''simple docstring'''
					_configure_library_root_logger()

					assert _default_handler is not None
					_get_library_root_logger().addHandler(_default_handler )


def     __UpperCAmelCase    (	lowerCamelCase_ )     ->			None:


					'''simple docstring'''
					_configure_library_root_logger()

					assert handler is not None
					_get_library_root_logger().addHandler(lowerCamelCase_ )


def     __UpperCAmelCase    (	lowerCamelCase_ )     ->			None:


					'''simple docstring'''
					_configure_library_root_logger()

					assert handler is not None and handler not in _get_library_root_logger().handlers
					_get_library_root_logger().removeHandler(lowerCamelCase_ )


def     __UpperCAmelCase    (	)     ->			None:


					'''simple docstring'''
					_configure_library_root_logger()
					__a:						str		    =					False


def     __UpperCAmelCase    (	)     ->			None:


					'''simple docstring'''
					_configure_library_root_logger()
					__a:						Tuple		    =					True


def     __UpperCAmelCase    (	)     ->			None:


					'''simple docstring'''
					__a:						Any		    =					_get_library_root_logger().handlers

					for handler in handlers:
										__a:						int		    =					logging.Formatter("""[%(levelname)s|%(filename)s:%(lineno)s] %(asctime)s >> %(message)s""" )
										handler.setFormatter(lowerCamelCase_ )


def     __UpperCAmelCase    (	)     ->			None:


					'''simple docstring'''
					__a:						int		    =					_get_library_root_logger().handlers

					for handler in handlers:
										handler.setFormatter(lowerCamelCase_ )


def     __UpperCAmelCase    (	self	,				*lowerCamelCase_	,				**lowerCamelCase_ )     ->			Union[str, Any]:


					'''simple docstring'''
					__a:						Optional[Any]		    =					os.getenv("""DIFFUSERS_NO_ADVISORY_WARNINGS"""	,				lowerCamelCase_ )
					if no_advisory_warnings:
										return
					self.warning(*lowerCamelCase_	,				**lowerCamelCase_ )


lowercase		:						Tuple =					warning_advice




class    _snake_case  :






							def __init__(					self						:			Dict						, *snake_case__						:			Dict						, **snake_case__						:			List[Any]    ):  # pylint: disable=unused-argument






												"""simple docstring"""



												__a:						Dict		    =					args[0] if args else None






							def __iter__(					self						:			List[str]    ):






												"""simple docstring"""



												return iter(self._iterator    )






							def __getattr__(					self						:			List[str]						, snake_case__						:			Dict    ):






												"""simple docstring"""



												def empty_fn(*snake_case__						:			Optional[Any]						, **snake_case__						:			int    ):  # pylint: disable=unused-argument
																	return

												return empty_fn






							def __enter__(					self						:			Optional[Any]    ):






												"""simple docstring"""



												return self






							def __exit__(					self						:			int						, snake_case__						:			List[Any]						, snake_case__						:			Tuple						, snake_case__						:			str    ):






												"""simple docstring"""



												return






class    _snake_case  :






							def __call__(					self						:			Union[str, Any]						, *snake_case__						:			Optional[int]						, **snake_case__						:			Union[str, Any]    ):






												"""simple docstring"""



												if _tqdm_active:
																	return tqdm_lib.tqdm(*snake_case__						, **snake_case__    )
												else:
																	return EmptyTqdm(*snake_case__						, **snake_case__    )






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






												"""simple docstring"""



												__a:						Any		    =					None
												if _tqdm_active:
																	return tqdm_lib.tqdm.set_lock(*snake_case__						, **snake_case__    )






							def 							__lowercase  (					self						:			Optional[Any]    ):






												"""simple docstring"""



												if _tqdm_active:
																	return tqdm_lib.tqdm.get_lock()






lowercase		:						str =					_tqdm_cls()


def     __UpperCAmelCase    (	)     ->			bool:


					'''simple docstring'''
					global _tqdm_active
					return bool(_tqdm_active )


def     __UpperCAmelCase    (	)     ->			Dict:


					'''simple docstring'''
					global _tqdm_active
					__a:						List[Any]		    =					True




def     __UpperCAmelCase    (	)     ->			int:


					'''simple docstring'''
					global _tqdm_active
					__a:						Dict		    =					False


"""simple docstring"""
import torch
import torch.nn as nn
from transformers.modeling_utils import ModuleUtilsMixin
from transformers.models.ta.modeling_ta import (
    TaBlock,
    TaConfig,
    TaLayerNorm,
)

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




class    _snake_case  (						_lowercase ,				_lowercase ,				_lowercase       ):






							@register_to_config
							def __init__(					self						:			str						, snake_case__						:			int						, snake_case__						:			int						, snake_case__						:			int						, snake_case__						:			float						, snake_case__						:			int						, snake_case__						:			int						, snake_case__						:			int						, snake_case__						:			int						, snake_case__						:			str						, snake_case__						:			bool = False						, ):






												"""simple docstring"""



												super().__init__()

												__a:						Optional[Any]		    =					nn.Linear(snake_case__						, snake_case__						, bias=snake_case__    )

												__a:						Optional[Any]		    =					nn.Embedding(snake_case__						, snake_case__    )
												__a:						Any		    =					False

												__a:						Any		    =					nn.Dropout(p=snake_case__    )

												__a:						List[str]		    =					TaConfig(
												    d_model=snake_case__						, num_heads=snake_case__						, d_kv=snake_case__						, d_ff=snake_case__						, feed_forward_proj=snake_case__						, dropout_rate=snake_case__						, is_decoder=snake_case__						, is_encoder_decoder=snake_case__						, )
												__a:						Optional[int]		    =					nn.ModuleList()
												for lyr_num in range(snake_case__    ):
																	__a:						Union[str, Any]		    =					TaBlock(snake_case__    )
																	self.encoders.append(snake_case__    )

												__a:						List[str]		    =					TaLayerNorm(snake_case__    )
												__a:						Optional[int]		    =					nn.Dropout(p=snake_case__    )






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






												"""simple docstring"""



												__a:						str		    =					self.input_proj(snake_case__    )

												# terminal relative positional encodings
												__a:						Optional[int]		    =					encoder_inputs.shape[1]
												__a:						List[Any]		    =					torch.arange(snake_case__						, device=encoder_inputs.device    )

												__a:						Union[str, Any]		    =					encoder_inputs_mask.sum(-1    )
												__a:						str		    =					torch.roll(input_positions.unsqueeze(0    )						, tuple(seq_lens.tolist()    )						, dims=0    )
												x += self.position_encoding(snake_case__    )

												__a:						Any		    =					self.dropout_pre(snake_case__    )

												# inverted the attention mask
												__a:						Any		    =					encoder_inputs.size()
												__a:						str		    =					self.get_extended_attention_mask(snake_case__						, snake_case__    )

												for lyr in self.encoders:
																	__a:						Any		    =					lyr(snake_case__						, snake_case__    )[0]
												__a:						int		    =					self.layer_norm(snake_case__    )

												return self.dropout_post(snake_case__    ), encoder_inputs_mask
