





import logging
import sys
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Union

import librosa
import torch
from datasets import DatasetDict, load_dataset
from packaging import version
from torch import nn

from transformers import (
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    WavaVecaConfig,
    WavaVecaFeatureExtractor,
    WavaVecaForPreTraining,
    is_apex_available,
    trainer_utils,
)
from transformers.models.wavaveca.modeling_wavaveca import _compute_mask_indices


if is_apex_available():
    from apex import amp

if version.parse(version.parse(torch.__version__).base_version) >= version.parse("""1.6"""):
    lowercase      :       Tuple									=  True
    from torch.cuda.amp import autocast


lowercase      :       List[Any]									=  logging.getLogger(__name__)





@dataclass
class  A__    :




   """simple docstring"""



   __A   :				str         =  field(
       metadata={'''help''': '''Path to pretrained model or model identifier from huggingface.co/models'''}	)
   __A   :				Optional[str]         =  field(
       default=__UpperCAmelCase ,  metadata={'''help''': '''Where do you want to store the pretrained models downloaded from huggingface.co'''} ,  )
   __A   :				Optional[bool]         =  field(
       default=__UpperCAmelCase ,  metadata={'''help''': '''Whether to freeze the feature extractor layers of the model.'''}	)
   __A   :				Optional[bool]         =  field(
       default=__UpperCAmelCase ,  metadata={'''help''': '''Whether to log verbose messages or not.'''} ,  )
   __A   :				Optional[float]         =  field(
       default=2.0 ,  metadata={'''help''': '''Maximum temperature for gumbel softmax.'''}	)
   __A   :				Optional[float]         =  field(
       default=0.5 ,  metadata={'''help''': '''Minimum temperature for gumbel softmax.'''}	)
   __A   :				Optional[float]         =  field(
       default=0.99_9995 ,  metadata={'''help''': '''Decay of gumbel temperature during training.'''}	)



def A_	(		A__		,      A__				)  ->		Tuple:
       logging.basicConfig(
           format='%(asctime)s - %(levelname)s - %(name)s - %(message)s'		,      datefmt='%m/%d/%Y %H:%M:%S'		,      handlers=[logging.StreamHandler(sys.stdout				)]		,      )
       a__						:	Optional[int] =				logging.WARNING
       if model_args.verbose_logging:
              a__						:	str =				logging.DEBUG
       elif trainer_utils.is_main_process(training_args.local_rank				):
              a__						:	int =				logging.INFO
       logger.setLevel(A__				)





@dataclass
class  A__    :




   """simple docstring"""



   __A   :				str         =  field(
       default=__UpperCAmelCase ,  metadata={'''help''': '''The name of the dataset to use (via the datasets library).'''}	)
   __A   :				Optional[str]         =  field(
       default=__UpperCAmelCase ,  metadata={'''help''': '''The configuration name of the dataset to use (via the datasets library).'''}	)
   __A   :				Optional[str]         =  field(
       default='''train''' ,  metadata={
           '''help''': '''The name of the training data set split to use (via the datasets library). Defaults to \'train\''''
       } ,  )
   __A   :				Optional[str]         =  field(
       default='''validation''' ,  metadata={
           '''help''': (
               '''The name of the validation data set split to use (via the datasets library). Defaults to \'validation\''''
           )
       } ,  )
   __A   :				Optional[str]         =  field(
       default='''file''' ,  metadata={'''help''': '''Column in the dataset that contains speech file path. Defaults to \'file\''''} ,  )
   __A   :				bool         =  field(
       default=__UpperCAmelCase ,  metadata={'''help''': '''Overwrite the cached preprocessed datasets or not.'''}	)
   __A   :				Optional[int]         =  field(
       default=1 ,  metadata={
           '''help''': '''The percentage of the train set used as validation set in case there\'s no validation split'''
       } ,  )
   __A   :				Optional[int]         =  field(
       default=__UpperCAmelCase ,  metadata={'''help''': '''The number of processes to use for the preprocessing.'''} ,  )
   __A   :				Optional[float]         =  field(
       default=20.0 ,  metadata={'''help''': '''Filter audio files that are longer than `max_duration_in_seconds` seconds'''}	)





@dataclass
class  A__    :




   """simple docstring"""



   __A   :				WavaVecaForPreTraining
   __A   :				WavaVecaFeatureExtractor
   __A   :				Union[bool, str]         =  "longest"
   __A   :				Optional[int]         =  None
   __A   :				Optional[int]         =  None

   def __call__(					self					,  lowercase)			->						Dict[str, torch.Tensor]:







          '''simple docstring'''

          a__						:	Optional[Any] =				self.feature_extractor.pad(
              lowercase					,  max_length=self.max_length					,  padding=self.padding					,  pad_to_multiple_of=self.pad_to_multiple_of					,  return_tensors='pt'					,  )
          a__						:	str =				self.model._get_feat_extract_output_lengths(batch['input_values'].shape[-1])

          a__						:	List[str] =				batch['input_values'].shape[0]

          # make sure that no loss is computed on padded inputs
          if batch["attention_mask"] is not None:
                 # compute real output lengths according to convolution formula
                 a__						:	Optional[int] =				self.model._get_feat_extract_output_lengths(batch['attention_mask'].sum(-1)).to(
                     torch.long)

                 a__						:	Optional[Any] =				torch.zeros(
                     (batch_size, mask_indices_seq_length)					,  dtype=torch.long					,  device=batch['input_values'].device)

                 # these two operations makes sure that all values
                 # before the output lengths indices are attended to
                 a__						:	Tuple =				1
                 a__						:	List[Any] =				attention_mask.flip([-1]).cumsum(-1).flip([-1]).bool()

          # sample randomly masked indices
          a__						:	Any =				_compute_mask_indices(
              (batch_size, mask_indices_seq_length)					,  self.model.config.mask_time_prob					,  self.model.config.mask_time_length					,  attention_mask=lowercase					,  min_masks=2					,  )

          return batch





class  A__    (							__UpperCAmelCase	):




   """simple docstring"""



   def __init__(					self					,  *lowercase					,  lowercase=1					,  lowercase=0					,  lowercase=1.0					,  **lowercase)			->						str:







          '''simple docstring'''

          super().__init__(*lowercase					,  **lowercase)
          a__						:	Dict =				0
          a__						:	int =				max_gumbel_temp
          a__						:	List[Any] =				min_gumbel_temp
          a__						:	List[str] =				gumbel_temp_decay

   def 							__lowercase		(					self					,  lowercase					,  lowercase)			->						torch.Tensor:







          '''simple docstring'''

          model.train()
          a__						:	Any =				self._prepare_inputs(lowercase)

          if self.use_amp:
                 with autocast():
                        a__						:	List[str] =				self.compute_loss(lowercase					,  lowercase)
          else:
                 a__						:	Optional[int] =				self.compute_loss(lowercase					,  lowercase)

          if self.args.n_gpu > 1 or self.deepspeed:
                 if model.module.config.ctc_loss_reduction == "mean":
                        a__						:	List[str] =				loss.mean()
                 elif model.module.config.ctc_loss_reduction == "sum":
                        a__						:	List[str] =				loss.sum() / (inputs['mask_time_indices']).sum()
                 else:
                        raise ValueError(F'{model.config.ctc_loss_reduction} is not valid. Choose one of [\'mean\', \'sum\']')

          if self.args.gradient_accumulation_steps > 1:
                 a__						:	Tuple =				loss / self.args.gradient_accumulation_steps

          if self.use_amp:
                 self.scaler.scale(lowercase).backward()
          elif self.use_apex:
                 with amp.scale_loss(lowercase					,  self.optimizer) as scaled_loss:
                        scaled_loss.backward()
          elif self.deepspeed:
                 self.deepspeed.backward(lowercase)
          else:
                 loss.backward()

          self.num_update_step += 1
          # make sure gumbel softmax temperature is decayed
          if self.args.n_gpu > 1 or self.deepspeed:
                 model.module.set_gumbel_temperature(
                     max(self.max_gumbel_temp * self.gumbel_temp_decay**self.num_update_step					,  self.min_gumbel_temp))
          else:
                 model.set_gumbel_temperature(
                     max(self.max_gumbel_temp * self.gumbel_temp_decay**self.num_update_step					,  self.min_gumbel_temp))

          return loss.detach()



def A_	(		)  ->		Optional[int]:
       # See all possible arguments in src/transformers/training_args.py
       # or by passing the --help flag to this script.
       # We now keep distinct sets of args, for a cleaner separation of concerns.

       a__						:	Tuple =				HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)				)

       a__ ,							a__ ,							a__						:	List[Any] =				parser.parse_args_into_dataclasses()
       configure_logger(A__		,      A__				)

       # Downloading and loading a dataset from the hub.
       a__						:	Dict =				load_dataset(data_args.dataset_name		,      data_args.dataset_config_name		,      cache_dir=model_args.cache_dir				)

       if "validation" not in datasets.keys():
              # make sure only "validation" and "train" keys remain"
              a__						:	Tuple =				DatasetDict()
              a__						:	Any =				load_dataset(
                  data_args.dataset_name		,      data_args.dataset_config_name		,      split=F'{data_args.train_split_name}[:{data_args.validation_split_percentage}%]'		,      cache_dir=model_args.cache_dir		,      )
              a__						:	List[Any] =				load_dataset(
                  data_args.dataset_name		,      data_args.dataset_config_name		,      split=F'{data_args.train_split_name}[{data_args.validation_split_percentage}%:]'		,      cache_dir=model_args.cache_dir		,      )
       else:
              # make sure only "validation" and "train" keys remain"
              a__						:	List[Any] =				DatasetDict()
              a__						:	str =				load_dataset(
                  data_args.dataset_name		,      data_args.dataset_config_name		,      split='validation'		,      cache_dir=model_args.cache_dir		,      )
              a__						:	Optional[Any] =				load_dataset(
                  data_args.dataset_name		,      data_args.dataset_config_name		,      split=F'{data_args.train_split_name}'		,      cache_dir=model_args.cache_dir		,      )

       # only normalized-inputs-training is supported
       a__						:	str =				WavaVecaFeatureExtractor.from_pretrained(
           model_args.model_name_or_path		,      cache_dir=model_args.cache_dir		,      do_normalize=A__				)

       def prepare_dataset(A__				):
              # check that all files have the correct sampling rate
              a__ ,							a__						:	Tuple =				librosa.load(batch[data_args.speech_file_column]		,      sr=feature_extractor.sampling_rate				)
              return batch

       # load audio files into numpy arrays
       a__						:	Tuple =				datasets.map(
           A__		,      num_proc=data_args.preprocessing_num_workers		,      remove_columns=datasets['train'].column_names				)

       # filter audio files that are too long
       a__						:	Tuple =				vectorized_datasets.filter(
           lambda A__				: len(data['speech']				) < int(data_args.max_duration_in_seconds * feature_extractor.sampling_rate				)				)

       def normalize(A__				):
              return feature_extractor(batch['speech']		,      sampling_rate=feature_extractor.sampling_rate				)

       # normalize and transform to `BatchFeatures`
       a__						:	Any =				vectorized_datasets.map(
           A__		,      batched=A__		,      num_proc=data_args.preprocessing_num_workers		,      load_from_cache_file=not data_args.overwrite_cache		,      remove_columns=vectorized_datasets['train'].column_names		,      )

       # pretraining is only supported for "newer" stable layer norm architecture
       # apply_spec_augment has to be True, mask_feature_prob has to be 0.0
       a__						:	Any =				WavaVecaConfig.from_pretrained(
           model_args.model_name_or_path		,      cache_dir=model_args.cache_dir		,      gradient_checkpointing=training_args.gradient_checkpointing		,      )

       if not config.do_stable_layer_norm or config.feat_extract_norm != "layer":
              raise ValueError(
                  'PreTraining is only supported for ``config.do_stable_layer_norm=True`` and'
                  ' ``config.feat_extract_norm=\'layer\''				)

       a__						:	Any =				WavaVecaForPreTraining(A__				)

       a__						:	int =				DataCollatorForWavaVecaPretraining(model=A__		,      feature_extractor=A__				)

       a__						:	Optional[int] =				WavaVecaPreTrainer(
           model=A__		,      data_collator=A__		,      args=A__		,      train_dataset=vectorized_datasets['train']		,      eval_dataset=vectorized_datasets['validation']		,      tokenizer=A__		,      max_gumbel_temp=model_args.max_gumbel_temperature		,      min_gumbel_temp=model_args.min_gumbel_temperature		,      gumbel_temp_decay=model_args.gumbel_temperature_decay		,      )
       trainer.train()


if __name__ == "__main__":
    main()







"""simple docstring"""
import gc
import math
import unittest

import torch

from diffusers import UNetaDModel
from diffusers.utils import floats_tensor, logging, slow, torch_all_close, torch_device
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


__a      =				logging.get_logger(__name__)

enable_full_determinism()
class 				lowerCamelCase  (			_lowerCAmelCase    ,						_lowerCAmelCase    ,						unittest.TestCase				):






					'''simple docstring'''





					_A		:			List[Any]						=     UNetaDModel
					_A		:			Union[str, Any]						=     """sample"""






					@property
					def    lowerCAmelCase_				(       self:							str						)   ->	Tuple:
											snake_case_ :List[str]						=			4
											snake_case_ :Tuple						=			3
											snake_case_ :Optional[Any]						=			(32, 32)

											snake_case_ :str						=			floats_tensor((batch_size, num_channels) + sizes						).to(snake_case						)
											snake_case_ :Union[str, Any]						=			torch.tensor([10]						).to(snake_case						)

											return {"sample": noise, "timestep": time_step}






					@property
					def    lowerCAmelCase_				(       self:							List[str]						)   ->	Dict:
											return (3, 32, 32)






					@property
					def    lowerCAmelCase_				(       self:							Optional[int]						)   ->	Optional[int]:
											return (3, 32, 32)






					def    lowerCAmelCase_				(       self:							Optional[int]						)   ->	Dict:
											snake_case_ :Any						=			{
											    """block_out_channels""": (32, 64),
											    """down_block_types""": ("""DownBlock2D""", """AttnDownBlock2D"""),
											    """up_block_types""": ("""AttnUpBlock2D""", """UpBlock2D"""),
											    """attention_head_dim""": 3,
											    """out_channels""": 3,
											    """in_channels""": 3,
											    """layers_per_block""": 2,
											    """sample_size""": 32,
											}
											snake_case_ :Tuple						=			self.dummy_input
											return init_dict, inputs_dict

class 				lowerCamelCase  (			_lowerCAmelCase    ,						_lowerCAmelCase    ,						unittest.TestCase				):






					'''simple docstring'''





					_A		:			List[str]						=     UNetaDModel
					_A		:			Union[str, Any]						=     """sample"""






					@property
					def    lowerCAmelCase_				(       self:							str						)   ->	str:
											snake_case_ :List[str]						=			4
											snake_case_ :Optional[int]						=			4
											snake_case_ :int						=			(32, 32)

											snake_case_ :Any						=			floats_tensor((batch_size, num_channels) + sizes						).to(snake_case						)
											snake_case_ :List[Any]						=			torch.tensor([10]						).to(snake_case						)

											return {"sample": noise, "timestep": time_step}






					@property
					def    lowerCAmelCase_				(       self:							Optional[int]						)   ->	Optional[int]:
											return (4, 32, 32)






					@property
					def    lowerCAmelCase_				(       self:							List[Any]						)   ->	int:
											return (4, 32, 32)






					def    lowerCAmelCase_				(       self:							Union[str, Any]						)   ->	List[Any]:
											snake_case_ :Dict						=			{
											    """sample_size""": 32,
											    """in_channels""": 4,
											    """out_channels""": 4,
											    """layers_per_block""": 2,
											    """block_out_channels""": (32, 64),
											    """attention_head_dim""": 32,
											    """down_block_types""": ("""DownBlock2D""", """DownBlock2D"""),
											    """up_block_types""": ("""UpBlock2D""", """UpBlock2D"""),
											}
											snake_case_ :List[str]						=			self.dummy_input
											return init_dict, inputs_dict






					def    lowerCAmelCase_				(       self:							Optional[int]						)   ->	Optional[Any]:
											snake_case_,       snake_case_ :List[str]						=			UNetaDModel.from_pretrained("""fusing/unet-ldm-dummy-update"""						,     output_loading_info=snake_case						)

											self.assertIsNotNone(snake_case						)
											self.assertEqual(len(loading_info["""missing_keys"""]						)						,     0						)

											model.to(snake_case						)
											snake_case_ :List[str]						=			model(**self.dummy_input						).sample

											assert image is not None, "Make sure output is not None"






					@unittest.skipIf(torch_device != """cuda"""						,     """This test is supposed to run on GPU"""						)
					def    lowerCAmelCase_				(       self:							Tuple						)   ->	Dict:
											snake_case_,       snake_case_ :Union[str, Any]						=			UNetaDModel.from_pretrained("""fusing/unet-ldm-dummy-update"""						,     output_loading_info=snake_case						)
											model.to(snake_case						)
											snake_case_ :Union[str, Any]						=			model(**self.dummy_input						).sample

											assert image is not None, "Make sure output is not None"






					@unittest.skipIf(torch_device != """cuda"""						,     """This test is supposed to run on GPU"""						)
					def    lowerCAmelCase_				(       self:							str						)   ->	Any:
											# by defautl model loading will use accelerate as `low_cpu_mem_usage=True`
											snake_case_,       snake_case_ :List[str]						=			UNetaDModel.from_pretrained("""fusing/unet-ldm-dummy-update"""						,     output_loading_info=snake_case						)
											model_accelerate.to(snake_case						)
											model_accelerate.eval()

											snake_case_ :List[Any]						=			torch.randn(
											    1						,     model_accelerate.config.in_channels						,     model_accelerate.config.sample_size						,     model_accelerate.config.sample_size						,     generator=torch.manual_seed(0						)						,     )
											snake_case_ :int						=			noise.to(snake_case						)
											snake_case_ :str						=			torch.tensor([10] * noise.shape[0]						).to(snake_case						)

											snake_case_ :Optional[int]						=			model_accelerate(snake_case						,     snake_case						)["""sample"""]

											# two models don't need to stay in the device at the same time
											del model_accelerate
											torch.cuda.empty_cache()
											gc.collect()

											snake_case_,       snake_case_ :str						=			UNetaDModel.from_pretrained(
											    """fusing/unet-ldm-dummy-update"""						,     output_loading_info=snake_case						,     low_cpu_mem_usage=snake_case						)
											model_normal_load.to(snake_case						)
											model_normal_load.eval()
											snake_case_ :int						=			model_normal_load(snake_case						,     snake_case						)["""sample"""]

											assert torch_all_close(snake_case						,     snake_case						,     rtol=1E-3						)






					def    lowerCAmelCase_				(       self:							Tuple						)   ->	Any:
											snake_case_ :Tuple						=			UNetaDModel.from_pretrained("""fusing/unet-ldm-dummy-update"""						)
											model.eval()
											model.to(snake_case						)

											snake_case_ :Optional[int]						=			torch.randn(
											    1						,     model.config.in_channels						,     model.config.sample_size						,     model.config.sample_size						,     generator=torch.manual_seed(0						)						,     )
											snake_case_ :int						=			noise.to(snake_case						)
											snake_case_ :List[Any]						=			torch.tensor([10] * noise.shape[0]						).to(snake_case						)

											with torch.no_grad():
																	snake_case_ :Union[str, Any]						=			model(snake_case						,     snake_case						).sample

											snake_case_ :Optional[int]						=			output[0, -1, -3:, -3:].flatten().cpu()
											# fmt: off
											snake_case_ :Dict						=			torch.tensor([-1_3.3_2_5_8, -2_0.1_1_0_0, -1_5.9_8_7_3, -1_7.6_6_1_7, -2_3.0_5_9_6, -1_7.9_4_1_9, -1_3.3_6_7_5, -1_6.1_8_8_9, -1_2.3_8_0_0]						)
											# fmt: on

											self.assertTrue(torch_all_close(snake_case						,     snake_case						,     rtol=1E-3						)						)

class 				lowerCamelCase  (			_lowerCAmelCase    ,						_lowerCAmelCase    ,						unittest.TestCase				):






					'''simple docstring'''





					_A		:			List[Any]						=     UNetaDModel
					_A		:			List[Any]						=     """sample"""






					@property
					def    lowerCAmelCase_				(       self:							Union[str, Any]						,     snake_case:							int=(32, 32)						)   ->	Tuple:
											snake_case_ :Union[str, Any]						=			4
											snake_case_ :Any						=			3

											snake_case_ :int						=			floats_tensor((batch_size, num_channels) + sizes						).to(snake_case						)
											snake_case_ :Any						=			torch.tensor(batch_size * [10]						).to(dtype=torch.intaa						,     device=snake_case						)

											return {"sample": noise, "timestep": time_step}






					@property
					def    lowerCAmelCase_				(       self:							Union[str, Any]						)   ->	Any:
											return (3, 32, 32)






					@property
					def    lowerCAmelCase_				(       self:							int						)   ->	Tuple:
											return (3, 32, 32)






					def    lowerCAmelCase_				(       self:							List[str]						)   ->	Tuple:
											snake_case_ :List[Any]						=			{
											    """block_out_channels""": [32, 64, 64, 64],
											    """in_channels""": 3,
											    """layers_per_block""": 1,
											    """out_channels""": 3,
											    """time_embedding_type""": """fourier""",
											    """norm_eps""": 1E-6,
											    """mid_block_scale_factor""": math.sqrt(2.0						),
											    """norm_num_groups""": None,
											    """down_block_types""": [
											        """SkipDownBlock2D""",
											        """AttnSkipDownBlock2D""",
											        """SkipDownBlock2D""",
											        """SkipDownBlock2D""",
											    ],
											    """up_block_types""": [
											        """SkipUpBlock2D""",
											        """SkipUpBlock2D""",
											        """AttnSkipUpBlock2D""",
											        """SkipUpBlock2D""",
											    ],
											}
											snake_case_ :int						=			self.dummy_input
											return init_dict, inputs_dict






					@slow
					def    lowerCAmelCase_				(       self:							Optional[Any]						)   ->	List[Any]:
											snake_case_,       snake_case_ :List[Any]						=			UNetaDModel.from_pretrained("""google/ncsnpp-celebahq-256"""						,     output_loading_info=snake_case						)
											self.assertIsNotNone(snake_case						)
											self.assertEqual(len(loading_info["""missing_keys"""]						)						,     0						)

											model.to(snake_case						)
											snake_case_ :Any						=			self.dummy_input
											snake_case_ :int						=			floats_tensor((4, 3) + (256, 256)						).to(snake_case						)
											snake_case_ :int						=			noise
											snake_case_ :int						=			model(**snake_case						)

											assert image is not None, "Make sure output is not None"






					@slow
					def    lowerCAmelCase_				(       self:							str						)   ->	Dict:
											snake_case_ :Dict						=			UNetaDModel.from_pretrained("""google/ncsnpp-celebahq-256"""						)
											model.to(snake_case						)

											snake_case_ :List[str]						=			4
											snake_case_ :Optional[int]						=			3
											snake_case_ :List[str]						=			(256, 256)

											snake_case_ :Tuple						=			torch.ones((batch_size, num_channels) + sizes						).to(snake_case						)
											snake_case_ :str						=			torch.tensor(batch_size * [1E-4]						).to(snake_case						)

											with torch.no_grad():
																	snake_case_ :Dict						=			model(snake_case						,     snake_case						).sample

											snake_case_ :int						=			output[0, -3:, -3:, -1].flatten().cpu()
											# fmt: off
											snake_case_ :Optional[Any]						=			torch.tensor([-4_8_4_2.8_6_9_1, -6_4_9_9.6_6_3_1, -3_8_0_0.1_9_5_3, -7_9_7_8.2_6_8_6, -1_0_9_8_0.7_1_2_9, -2_0_0_2_8.8_5_3_5, 8_1_4_8.2_8_2_2, 2_3_4_2.2_9_0_5, 5_6_7.7_6_0_8]						)
											# fmt: on

											self.assertTrue(torch_all_close(snake_case						,     snake_case						,     rtol=1E-2						)						)






					def    lowerCAmelCase_				(       self:							List[str]						)   ->	List[Any]:
											snake_case_ :Optional[Any]						=			UNetaDModel.from_pretrained("""fusing/ncsnpp-ffhq-ve-dummy-update"""						)
											model.to(snake_case						)

											snake_case_ :Optional[int]						=			4
											snake_case_ :Optional[Any]						=			3
											snake_case_ :Optional[Any]						=			(32, 32)

											snake_case_ :Dict						=			torch.ones((batch_size, num_channels) + sizes						).to(snake_case						)
											snake_case_ :Any						=			torch.tensor(batch_size * [1E-4]						).to(snake_case						)

											with torch.no_grad():
																	snake_case_ :str						=			model(snake_case						,     snake_case						).sample

											snake_case_ :int						=			output[0, -3:, -3:, -1].flatten().cpu()
											# fmt: off
											snake_case_ :int						=			torch.tensor([-0.0_3_2_5, -0.0_9_0_0, -0.0_8_6_9, -0.0_3_3_2, -0.0_7_2_5, -0.0_2_7_0, -0.0_1_0_1, 0.0_2_2_7, 0.0_2_5_6]						)
											# fmt: on

											self.assertTrue(torch_all_close(snake_case						,     snake_case						,     rtol=1E-2						)						)






					def    lowerCAmelCase_				(       self:							Dict						)   ->	Optional[Any]:
											# not required for this model
											pass
