




from dataclasses import dataclass
from typing import Optional, Tuple, Union

import torch
import torch.nn as nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .embeddings import GaussianFourierProjection, TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin
from .unet_ad_blocks import get_down_block, get_mid_block, get_out_block, get_up_block


@dataclass
class 							UpperCamelCase_       (					UpperCAmelCase__	):

    '''simple docstring'''






    UpperCAmelCase__				    = 42


class 							UpperCamelCase_       (					UpperCAmelCase__						,   UpperCAmelCase__	):

    '''simple docstring'''






    @register_to_config
    def __init__(     self :      int   ,					UpperCAmelCase__ :      int = 65_536   ,					UpperCAmelCase__ :      Optional[int] = None   ,					UpperCAmelCase__ :      int = 2   ,					UpperCAmelCase__ :      int = 2   ,					UpperCAmelCase__ :      int = 0   ,					UpperCAmelCase__ :      str = "fourier"   ,					UpperCAmelCase__ :      bool = True   ,					UpperCAmelCase__ :      bool = False   ,					UpperCAmelCase__ :      float = 0.0   ,					UpperCAmelCase__ :      Tuple[str] = ("DownBlock1DNoSkip", "DownBlock1D", "AttnDownBlock1D")   ,					UpperCAmelCase__ :      Tuple[str] = ("AttnUpBlock1D", "UpBlock1D", "UpBlock1DNoSkip")   ,					UpperCAmelCase__ :      Tuple[str] = "UNetMidBlock1D"   ,					UpperCAmelCase__ :      str = None   ,					UpperCAmelCase__ :      Tuple[int] = (32, 32, 64)   ,					UpperCAmelCase__ :      str = None   ,					UpperCAmelCase__ :      int = 8   ,					UpperCAmelCase__ :      int = 1   ,					UpperCAmelCase__ :      bool = False   ,					)		->Tuple:





           '''simple docstring'''

           super().__init__()
           A__ 					=       sample_size

           # time
           if time_embedding_type == "fourier":
                  A__ 					=       GaussianFourierProjection(
                      embedding_size=8   ,					set_W_to_weight=UpperCAmelCase__   ,					log=UpperCAmelCase__   ,					flip_sin_to_cos=UpperCAmelCase__)
                  A__ 					=       2 * block_out_channels[0]
           elif time_embedding_type == "positional":
                  A__ 					=       Timesteps(
                      block_out_channels[0]   ,					flip_sin_to_cos=UpperCAmelCase__   ,					downscale_freq_shift=UpperCAmelCase__)
                  A__ 					=       block_out_channels[0]

           if use_timestep_embedding:
                  A__ 					=       block_out_channels[0] * 4
                  A__ 					=       TimestepEmbedding(
                      in_channels=UpperCAmelCase__   ,					time_embed_dim=UpperCAmelCase__   ,					act_fn=UpperCAmelCase__   ,					out_dim=block_out_channels[0]   ,					)

           A__ 					=       nn.ModuleList([])
           A__ 					=       None
           A__ 					=       nn.ModuleList([])
           A__ 					=       None

           # down
           A__ 					=       in_channels
           for i, down_block_type in enumerate(UpperCAmelCase__):
                  A__ 					=       output_channel
                  A__ 					=       block_out_channels[i]

                  if i == 0:
                         input_channel += extra_in_channels

                  A__ 					=       i == len(UpperCAmelCase__) - 1

                  A__ 					=       get_down_block(
                      UpperCAmelCase__   ,					num_layers=UpperCAmelCase__   ,					in_channels=UpperCAmelCase__   ,					out_channels=UpperCAmelCase__   ,					temb_channels=block_out_channels[0]   ,					add_downsample=not is_final_block or downsample_each_block   ,					)
                  self.down_blocks.append(UpperCAmelCase__)

           # mid
           A__ 					=       get_mid_block(
               UpperCAmelCase__   ,					in_channels=block_out_channels[-1]   ,					mid_channels=block_out_channels[-1]   ,					out_channels=block_out_channels[-1]   ,					embed_dim=block_out_channels[0]   ,					num_layers=UpperCAmelCase__   ,					add_downsample=UpperCAmelCase__   ,					)

           # up
           A__ 					=       list(reversed(UpperCAmelCase__))
           A__ 					=       reversed_block_out_channels[0]
           if out_block_type is None:
                  A__ 					=       out_channels
           else:
                  A__ 					=       block_out_channels[0]

           for i, up_block_type in enumerate(UpperCAmelCase__):
                  A__ 					=       output_channel
                  A__ 					=       (
                      reversed_block_out_channels[i + 1] if i < len(UpperCAmelCase__) - 1 else final_upsample_channels
                  )

                  A__ 					=       i == len(UpperCAmelCase__) - 1

                  A__ 					=       get_up_block(
                      UpperCAmelCase__   ,					num_layers=UpperCAmelCase__   ,					in_channels=UpperCAmelCase__   ,					out_channels=UpperCAmelCase__   ,					temb_channels=block_out_channels[0]   ,					add_upsample=not is_final_block   ,					)
                  self.up_blocks.append(UpperCAmelCase__)
                  A__ 					=       output_channel

           # out
           A__ 					=       norm_num_groups if norm_num_groups is not None else min(block_out_channels[0] // 4   ,					32)
           A__ 					=       get_out_block(
               out_block_type=UpperCAmelCase__   ,					num_groups_out=UpperCAmelCase__   ,					embed_dim=block_out_channels[0]   ,					out_channels=UpperCAmelCase__   ,					act_fn=UpperCAmelCase__   ,					fc_dim=block_out_channels[-1] // 4   ,					)





    def 			SCREAMING_SNAKE_CASE				(     self :      Union[str, Any]   ,					UpperCAmelCase__ :      torch.FloatTensor   ,					UpperCAmelCase__ :      Union[torch.Tensor, float, int]   ,					UpperCAmelCase__ :      bool = True   ,					)		->Union[UNetaDOutput, Tuple]:





           '''simple docstring'''

           A__ 					=       timestep
           if not torch.is_tensor(UpperCAmelCase__):
                  A__ 					=       torch.tensor([timesteps]   ,					dtype=torch.long   ,					device=sample.device)
           elif torch.is_tensor(UpperCAmelCase__) and len(timesteps.shape) == 0:
                  A__ 					=       timesteps[None].to(sample.device)

           A__ 					=       self.time_proj(UpperCAmelCase__)
           if self.config.use_timestep_embedding:
                  A__ 					=       self.time_mlp(UpperCAmelCase__)
           else:
                  A__ 					=       timestep_embed[..., None]
                  A__ 					=       timestep_embed.repeat([1, 1, sample.shape[2]]).to(sample.dtype)
                  A__ 					=       timestep_embed.broadcast_to((sample.shape[:1] + timestep_embed.shape[1:]))

           # 2. down
           A__ 					=       ()
           for downsample_block in self.down_blocks:
                  A__					,				A__ 					=       downsample_block(hidden_states=UpperCAmelCase__   ,					temb=UpperCAmelCase__)
                  down_block_res_samples += res_samples

           # 3. mid
           if self.mid_block:
                  A__ 					=       self.mid_block(UpperCAmelCase__   ,					UpperCAmelCase__)

           # 4. up
           for i, upsample_block in enumerate(self.up_blocks):
                  A__ 					=       down_block_res_samples[-1:]
                  A__ 					=       down_block_res_samples[:-1]
                  A__ 					=       upsample_block(UpperCAmelCase__   ,					res_hidden_states_tuple=UpperCAmelCase__   ,					temb=UpperCAmelCase__)

           # 5. post-process
           if self.out_block:
                  A__ 					=       self.out_block(UpperCAmelCase__   ,					UpperCAmelCase__)

           if not return_dict:
                  return (sample,)

           return UNetaDOutput(sample=UpperCAmelCase__)




"""simple docstring"""






import os
import shutil
import tempfile
from unittest import TestCase
from unittest.mock import patch

import numpy as np
from datasets import Dataset

from transformers.models.realm.configuration_realm import RealmConfig
from transformers.models.realm.retrieval_realm import _REALM_BLOCK_RECORDS_FILENAME, RealmRetriever
from transformers.models.realm.tokenization_realm import VOCAB_FILES_NAMES, RealmTokenizer





class 				a__					(						A__       ):



     def  lowerCamelCase_						(							self :List[str]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Optional[Any]   	=tempfile.mkdtemp()
      UpperCamelCase_     :					str   	=5

      # Realm tok
      UpperCamelCase_     :					List[Any]   	=[
          '[UNK]',
          '[CLS]',
          '[SEP]',
          '[PAD]',
          '[MASK]',
          'test',
          'question',
          'this',
          'is',
          'the',
          'first',
          'second',
          'third',
          'fourth',
          'fifth',
          'record',
          'want',
          '##want',
          '##ed',
          'wa',
          'un',
          'runn',
          '##ing',
          ',',
          'low',
          'lowest',
      ]
      UpperCamelCase_     :					List[str]   	=os.path.join(self.tmpdirname      ,  'realm_tokenizer'   )
      os.makedirs(_lowerCamelCase      ,  exist_ok=_lowerCamelCase   )
      UpperCamelCase_     :					int   	=os.path.join(_lowerCamelCase      ,  VOCAB_FILES_NAMES['vocab_file']   )
      with open(self.vocab_file      ,  'w'      ,  encoding='utf-8'   ) as vocab_writer:
       vocab_writer.write(''.join([x + '\n' for x in vocab_tokens]   )   )

      UpperCamelCase_     :					str   	=os.path.join(self.tmpdirname      ,  'realm_block_records'   )
      os.makedirs(_lowerCamelCase      ,  exist_ok=_lowerCamelCase   )



     def  lowerCamelCase_						(							self :Union[str, Any]   ):







      '''simple docstring'''



      return RealmTokenizer.from_pretrained(os.path.join(self.tmpdirname      ,  'realm_tokenizer'   )   )



     def  lowerCamelCase_						(							self :Dict   ):







      '''simple docstring'''



      shutil.rmtree(self.tmpdirname   )



     def  lowerCamelCase_						(							self :Dict   ):







      '''simple docstring'''



      UpperCamelCase_     :					List[str]   	=RealmConfig(num_block_records=self.num_block_records   )
      return config



     def  lowerCamelCase_						(							self :Optional[int]   ):







      '''simple docstring'''



      UpperCamelCase_     :					str   	=Dataset.from_dict(
          {
              'id': ['0', '1'],
              'question': ['foo', 'bar'],
              'answers': [['Foo', 'Bar'], ['Bar']],
          }   )
      return dataset



     def  lowerCamelCase_						(							self :Optional[Any]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Union[str, Any]   	=np.array(
          [
              b'This is the first record',
              b'This is the second record',
              b'This is the third record',
              b'This is the fourth record',
              b'This is the fifth record',
              b'This is a longer longer longer record',
          ]      ,  dtype=_lowerCamelCase      ,  )
      return block_records



     def  lowerCamelCase_						(							self :Any   ):







      '''simple docstring'''



      UpperCamelCase_     :					List[Any]   	=RealmRetriever(
          block_records=self.get_dummy_block_records()      ,  tokenizer=self.get_tokenizer()      ,  )
      return retriever



     def  lowerCamelCase_						(							self :int   ):







      '''simple docstring'''



      UpperCamelCase_     :					Any   	=self.get_config()
      UpperCamelCase_     :					Tuple   	=self.get_dummy_retriever()
      UpperCamelCase_     :					Tuple   	=retriever.tokenizer

      UpperCamelCase_     :					List[str]   	=np.array([0, 3]      ,  dtype='long'   )
      UpperCamelCase_     :					Optional[Any]   	=tokenizer(['Test question']   ).input_ids
      UpperCamelCase_     :					Union[str, Any]   	=tokenizer(
          ['the fourth']      ,  add_special_tokens=_lowerCamelCase      ,  return_token_type_ids=_lowerCamelCase      ,  return_attention_mask=_lowerCamelCase      ,  ).input_ids
      UpperCamelCase_     :					Union[str, Any]   	=config.reader_seq_len

      UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_     :					Any   	=retriever(
          _lowerCamelCase      ,  _lowerCamelCase      ,  answer_ids=_lowerCamelCase      ,  max_length=_lowerCamelCase      ,  return_tensors='np'   )

      self.assertEqual(len(_lowerCamelCase   )      ,  2   )
      self.assertEqual(len(_lowerCamelCase   )      ,  2   )
      self.assertEqual(len(_lowerCamelCase   )      ,  2   )
      self.assertEqual(concat_inputs.input_ids.shape      ,  (2, 10)   )
      self.assertEqual(concat_inputs.attention_mask.shape      ,  (2, 10)   )
      self.assertEqual(concat_inputs.token_type_ids.shape      ,  (2, 10)   )
      self.assertEqual(concat_inputs.special_tokens_mask.shape      ,  (2, 10)   )
      self.assertEqual(
          tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[0]   )      ,  ['[CLS]', 'test', 'question', '[SEP]', 'this', 'is', 'the', 'first', 'record', '[SEP]']      ,  )
      self.assertEqual(
          tokenizer.convert_ids_to_tokens(concat_inputs.input_ids[1]   )      ,  ['[CLS]', 'test', 'question', '[SEP]', 'this', 'is', 'the', 'fourth', 'record', '[SEP]']      ,  )



     def  lowerCamelCase_						(							self :Optional[int]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Optional[int]   	=self.get_config()
      UpperCamelCase_     :					int   	=self.get_dummy_retriever()
      UpperCamelCase_     :					Any   	=retriever.tokenizer

      UpperCamelCase_     :					Optional[int]   	=np.array([0, 3, 5]      ,  dtype='long'   )
      UpperCamelCase_     :					Optional[int]   	=tokenizer(['Test question']   ).input_ids
      UpperCamelCase_     :					int   	=tokenizer(
          ['the fourth', 'longer longer']      ,  add_special_tokens=_lowerCamelCase      ,  return_token_type_ids=_lowerCamelCase      ,  return_attention_mask=_lowerCamelCase      ,  ).input_ids
      UpperCamelCase_     :					Union[str, Any]   	=config.reader_seq_len

      UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_	,     UpperCamelCase_     :					Union[str, Any]   	=retriever(
          _lowerCamelCase      ,  _lowerCamelCase      ,  answer_ids=_lowerCamelCase      ,  max_length=_lowerCamelCase      ,  return_tensors='np'   )

      self.assertEqual([False, True, True]      ,  _lowerCamelCase   )
      self.assertEqual([[-1, -1, -1], [6, -1, -1], [6, 7, 8]]      ,  _lowerCamelCase   )
      self.assertEqual([[-1, -1, -1], [7, -1, -1], [7, 8, 9]]      ,  _lowerCamelCase   )




     def  lowerCamelCase_						(							self :List[Any]   ):







      '''simple docstring'''



      UpperCamelCase_     :					Tuple   	=self.get_dummy_retriever()
      retriever.save_pretrained(os.path.join(self.tmpdirname      ,  'realm_block_records'   )   )

      # Test local path
      UpperCamelCase_     :					int   	=retriever.from_pretrained(os.path.join(self.tmpdirname      ,  'realm_block_records'   )   )
      self.assertEqual(retriever.block_records[0]      ,  b'This is the first record'   )

      # Test mocked remote path
      with patch('transformers.models.realm.retrieval_realm.hf_hub_download'   ) as mock_hf_hub_download:
       UpperCamelCase_     :					Dict   	=os.path.join(
           os.path.join(self.tmpdirname      ,  'realm_block_records'   )      ,  _REALM_BLOCK_RECORDS_FILENAME   )
       UpperCamelCase_     :					int   	=RealmRetriever.from_pretrained('google/realm-cc-news-pretrained-openqa'   )

      self.assertEqual(retriever.block_records[0]      ,  b'This is the first record'   )
