




'''simple docstring'''




import inspect
import unittest

from transformers import DecisionTransformerConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
     import torch

     from transformers import DecisionTransformerModel
     from transformers.models.decision_transformer.modeling_decision_transformer import (
         DECISION_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
     )


class   __UpperCamelCase       :






       def __init__(							self,		lowerCAmelCase,		lowerCAmelCase=13,		lowerCAmelCase=7,		lowerCAmelCase=6,		lowerCAmelCase=17,		lowerCAmelCase=23,		lowerCAmelCase=11,		lowerCAmelCase=True,		):



              """simple docstring"""







              lowerCamelCase_							=parent
              lowerCamelCase_							=batch_size
              lowerCamelCase_							=seq_length
              lowerCamelCase_							=act_dim
              lowerCamelCase_							=state_dim
              lowerCamelCase_							=hidden_size
              lowerCamelCase_							=max_length
              lowerCamelCase_							=is_training






       def      lowercase__     (							self  ):



              """simple docstring"""







              lowerCamelCase_							=floats_tensor((self.batch_size, self.seq_length, self.state_dim)  )
              lowerCamelCase_							=floats_tensor((self.batch_size, self.seq_length, self.act_dim)  )
              lowerCamelCase_							=floats_tensor((self.batch_size, self.seq_length, 1)  )
              lowerCamelCase_							=floats_tensor((self.batch_size, self.seq_length, 1)  )
              lowerCamelCase_							=ids_tensor((self.batch_size, self.seq_length),		vocab_size=1_000  )
              lowerCamelCase_							=random_attention_mask((self.batch_size, self.seq_length)  )

              lowerCamelCase_							=self.get_config()

              return (
                  config,
                  states,
                  actions,
                  rewards,
                  returns_to_go,
                  timesteps,
                  attention_mask,
              )






       def      lowercase__     (							self  ):



              """simple docstring"""







              return DecisionTransformerConfig(
                  batch_size=self.batch_size,		seq_length=self.seq_length,		act_dim=self.act_dim,		state_dim=self.state_dim,		hidden_size=self.hidden_size,		max_length=self.max_length,		)






       def      lowercase__     (							self,		lowerCAmelCase,		lowerCAmelCase,		lowerCAmelCase,		lowerCAmelCase,		lowerCAmelCase,		lowerCAmelCase,		lowerCAmelCase,		):



              """simple docstring"""







              lowerCamelCase_							=DecisionTransformerModel(config=lowerCAmelCase  )
              model.to(lowerCAmelCase  )
              model.eval()
              lowerCamelCase_							=model(lowerCAmelCase,		lowerCAmelCase,		lowerCAmelCase,		lowerCAmelCase,		lowerCAmelCase,		lowerCAmelCase  )

              self.parent.assertEqual(result.state_preds.shape,		states.shape  )
              self.parent.assertEqual(result.action_preds.shape,		actions.shape  )
              self.parent.assertEqual(result.return_preds.shape,		returns_to_go.shape  )
              self.parent.assertEqual(
                  result.last_hidden_state.shape,		(self.batch_size, self.seq_length * 3, self.hidden_size)  )  # seq length *3 as there are 3 modelities: states, returns and actions






       def      lowercase__     (							self  ):



              """simple docstring"""







              lowerCamelCase_							=self.prepare_config_and_inputs()
              (
                  (
                  lowerCamelCase_
              ),      (
                  lowerCamelCase_
              ),      (
                  lowerCamelCase_
              ),      (
                  lowerCamelCase_
              ),      (
                  lowerCamelCase_
              ),      (
                  lowerCamelCase_
              ),      (
                  lowerCamelCase_
              ),      
              )							=config_and_inputs
              lowerCamelCase_							={
                  '''states''': states,
                  '''actions''': actions,
                  '''rewards''': rewards,
                  '''returns_to_go''': returns_to_go,
                  '''timesteps''': timesteps,
                  '''attention_mask''': attention_mask,
              }
              return config, inputs_dict





@require_torch
class   __UpperCamelCase       ( lowerCamelCase__			,					lowerCamelCase__			,					lowerCamelCase__			,					unittest.TestCase				):
       lowercase	:    str      						=(DecisionTransformerModel,) if is_torch_available() else ()
       lowercase	:    Optional[Any]      						=()
       lowercase	:    List[str]      						={'feature-extraction': DecisionTransformerModel} if is_torch_available() else {}

       # Ignoring of a failing test from GenerationTesterMixin, as the model does not use inputs_ids
       lowercase	:    Dict      						=False

       # Ignoring of a failing tests from ModelTesterMixin, as the model does not implement these features
       lowercase	:    Tuple      						=False
       lowercase	:    Any      						=False
       lowercase	:    Any      						=False
       lowercase	:    Any      						=False
       lowercase	:    List[str]      						=False
       lowercase	:    Any      						=False
       lowercase	:    Dict      						=False
       lowercase	:    List[Any]      						=False
       lowercase	:    Optional[int]      						=False






       def      lowercase__     (							self  ):



              """simple docstring"""







              lowerCamelCase_							=DecisionTransformerModelTester(self  )
              lowerCamelCase_							=ConfigTester(self,		config_class=lowerCAmelCase,		hidden_size=37  )






       def      lowercase__     (							self  ):



              """simple docstring"""







              self.config_tester.run_common_tests()






       def      lowercase__     (							self  ):



              """simple docstring"""







              lowerCamelCase_							=self.model_tester.prepare_config_and_inputs()
              self.model_tester.create_and_check_model(*lowerCAmelCase  )






       @slow
       def      lowercase__     (							self  ):



              """simple docstring"""







              for model_name in DECISION_TRANSFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                     lowerCamelCase_							=DecisionTransformerModel.from_pretrained(lowerCAmelCase  )
                     self.assertIsNotNone(lowerCAmelCase  )






       def      lowercase__     (							self  ):



              """simple docstring"""







              lowerCamelCase_,      lowerCamelCase_							=self.model_tester.prepare_config_and_inputs_for_common()

              for model_class in self.all_model_classes:
                     lowerCamelCase_							=model_class(lowerCAmelCase  )
                     lowerCamelCase_							=inspect.signature(model.forward  )
                     # signature.parameters is an OrderedDict => so arg_names order is deterministic
                     lowerCamelCase_							=[*signature.parameters.keys()]

                     lowerCamelCase_							=[
                         '''states''',
                         '''actions''',
                         '''rewards''',
                         '''returns_to_go''',
                         '''timesteps''',
                         '''attention_mask''',
                     ]

                     self.assertListEqual(arg_names[: len(lowerCAmelCase  )],		lowerCAmelCase  )





@require_torch
class   __UpperCamelCase       ( unittest.TestCase				):






       @slow
       def      lowercase__     (							self  ):



              """simple docstring"""







              lowerCamelCase_							=2  # number of steps of autoregressive prediction we will perform
              lowerCamelCase_							=10  # defined by the RL environment, may be normalized
              lowerCamelCase_							=DecisionTransformerModel.from_pretrained('''edbeeching/decision-transformer-gym-hopper-expert'''  )
              lowerCamelCase_							=model.to(lowerCAmelCase  )
              lowerCamelCase_							=model.config
              torch.manual_seed(0  )
              lowerCamelCase_							=torch.randn(1,		1,		config.state_dim  ).to(device=lowerCAmelCase,		dtype=torch.floataa  )  # env.reset()

              lowerCamelCase_							=torch.tensor(
                  [[0.2_4_2_7_9_3, -0.2_8_6_9_3_0_7_4, 0.8_7_4_2_6_1_3], [0.6_7_8_1_5_2_7_4, -0.0_8_1_0_1_0_8_5, -0.1_2_9_5_2_1_4_7]],		device=lowerCAmelCase  )

              lowerCamelCase_							=torch.tensor(lowerCAmelCase,		device=lowerCAmelCase,		dtype=torch.floataa  ).reshape(1,		1,		1  )
              lowerCamelCase_							=state
              lowerCamelCase_							=torch.zeros(1,		0,		config.act_dim,		device=lowerCAmelCase,		dtype=torch.floataa  )
              lowerCamelCase_							=torch.zeros(1,		0,		device=lowerCAmelCase,		dtype=torch.floataa  )
              lowerCamelCase_							=torch.tensor(0,		device=lowerCAmelCase,		dtype=torch.long  ).reshape(1,		1  )

              for step in range(lowerCAmelCase  ):
                     lowerCamelCase_							=torch.cat([actions, torch.zeros(1,		1,		config.act_dim,		device=lowerCAmelCase  )],		dim=1  )
                     lowerCamelCase_							=torch.cat([rewards, torch.zeros(1,		1,		device=lowerCAmelCase  )],		dim=1  )

                     lowerCamelCase_							=torch.ones(1,		states.shape[1]  ).to(dtype=torch.long,		device=states.device  )

                     with torch.no_grad():
                            lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_							=model(
                                states=lowerCAmelCase,		actions=lowerCAmelCase,		rewards=lowerCAmelCase,		returns_to_go=lowerCAmelCase,		timesteps=lowerCAmelCase,		attention_mask=lowerCAmelCase,		return_dict=lowerCAmelCase,		)

                     self.assertEqual(action_pred.shape,		actions.shape  )
                     self.assertTrue(torch.allclose(action_pred[0, -1],		expected_outputs[step],		atol=1e-4  )  )
                     lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_,      lowerCamelCase_							=(  # env.step(action)
                         torch.randn(1,		1,		config.state_dim  ).to(device=lowerCAmelCase,		dtype=torch.floataa  ),
                         1.0,
                         False,
                         {},
                     )

                     lowerCamelCase_							=action_pred[0, -1]
                     lowerCamelCase_							=torch.cat([states, state],		dim=1  )
                     lowerCamelCase_							=returns_to_go[0, -1] - reward
                     lowerCamelCase_							=torch.cat([returns_to_go, pred_return.reshape(1,		1,		1  )],		dim=1  )
                     lowerCamelCase_							=torch.cat(
                         [timesteps, torch.ones((1, 1),		device=lowerCAmelCase,		dtype=torch.long  ) * (step + 1)],		dim=1  )





"""simple docstring"""




from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block
@dataclass
class 			UpperCamelCase__(	__A       ):
						lowerCAmelCase__     :  torch.FloatTensor






class 			UpperCamelCase__(	nn.Module       ):





						def __init__(				self						,__UpperCAmelCase=3						,__UpperCAmelCase=3						,__UpperCAmelCase=("DownEncoderBlock2D",)						,__UpperCAmelCase=(64,)						,__UpperCAmelCase=2						,__UpperCAmelCase=32						,__UpperCAmelCase="silu"						,__UpperCAmelCase=True						,)    -> Union[str, Any]:
													super().__init__()
													A__		   =							layers_per_block

													A__		   =							torch.nn.Convad(
													    __UpperCAmelCase						,block_out_channels[0]						,kernel_size=3						,stride=1						,padding=1						,)

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

													# down
													A__		   =							block_out_channels[0]
													for i, down_block_type in enumerate(__UpperCAmelCase   ):
																				A__		   =							output_channel
																				A__		   =							block_out_channels[i]
																				A__		   =							i == len(__UpperCAmelCase   ) - 1

																				A__		   =							get_down_block(
																				    __UpperCAmelCase						,num_layers=self.layers_per_block						,in_channels=__UpperCAmelCase						,out_channels=__UpperCAmelCase						,add_downsample=not is_final_block						,resnet_eps=1e-6						,downsample_padding=0						,resnet_act_fn=__UpperCAmelCase						,resnet_groups=__UpperCAmelCase						,attention_head_dim=__UpperCAmelCase						,temb_channels=__UpperCAmelCase						,)
																				self.down_blocks.append(__UpperCAmelCase   )

													# mid
													A__		   =							UNetMidBlockaD(
													    in_channels=block_out_channels[-1]						,resnet_eps=1e-6						,resnet_act_fn=__UpperCAmelCase						,output_scale_factor=1						,resnet_time_scale_shift='default'						,attention_head_dim=block_out_channels[-1]						,resnet_groups=__UpperCAmelCase						,temb_channels=__UpperCAmelCase						,)

													# out
													A__		   =							nn.GroupNorm(num_channels=block_out_channels[-1]						,num_groups=__UpperCAmelCase						,eps=1e-6   )
													A__		   =							nn.SiLU()

													A__		   =							2 * out_channels if double_z else out_channels
													A__		   =							nn.Convad(block_out_channels[-1]						,__UpperCAmelCase						,3						,padding=1   )

													A__		   =							False





						def  snake_case__		(				self						,__UpperCAmelCase   )    -> Optional[int]:
													A__		   =							x
													A__		   =							self.conv_in(__UpperCAmelCase   )

													if self.training and self.gradient_checkpointing:

																				def create_custom_forward(__UpperCAmelCase   ):
																											def custom_forward(*__UpperCAmelCase   ):
																																		return module(*__UpperCAmelCase   )

																											return custom_forward

																				# down
																				if is_torch_version('>='						,'1.11.0'   ):
																											for down_block in self.down_blocks:
																																		A__		   =							torch.utils.checkpoint.checkpoint(
																																		    create_custom_forward(__UpperCAmelCase   )						,__UpperCAmelCase						,use_reentrant=__UpperCAmelCase   )
																											# middle
																											A__		   =							torch.utils.checkpoint.checkpoint(
																											    create_custom_forward(self.mid_block   )						,__UpperCAmelCase						,use_reentrant=__UpperCAmelCase   )
																				else:
																											for down_block in self.down_blocks:
																																		A__		   =							torch.utils.checkpoint.checkpoint(create_custom_forward(__UpperCAmelCase   )						,__UpperCAmelCase   )
																											# middle
																											A__		   =							torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block   )						,__UpperCAmelCase   )

													else:
																				# down
																				for down_block in self.down_blocks:
																											A__		   =							down_block(__UpperCAmelCase   )

																				# middle
																				A__		   =							self.mid_block(__UpperCAmelCase   )

													# post-process
													A__		   =							self.conv_norm_out(__UpperCAmelCase   )
													A__		   =							self.conv_act(__UpperCAmelCase   )
													A__		   =							self.conv_out(__UpperCAmelCase   )

													return sample






class 			UpperCamelCase__(	nn.Module       ):





						def __init__(				self						,__UpperCAmelCase=3						,__UpperCAmelCase=3						,__UpperCAmelCase=("UpDecoderBlock2D",)						,__UpperCAmelCase=(64,)						,__UpperCAmelCase=2						,__UpperCAmelCase=32						,__UpperCAmelCase="silu"						,__UpperCAmelCase="group"						,)    -> Any:
													super().__init__()
													A__		   =							layers_per_block

													A__		   =							nn.Convad(
													    __UpperCAmelCase						,block_out_channels[-1]						,kernel_size=3						,stride=1						,padding=1						,)

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

													A__		   =							in_channels if norm_type == 'spatial' else None

													# mid
													A__		   =							UNetMidBlockaD(
													    in_channels=block_out_channels[-1]						,resnet_eps=1e-6						,resnet_act_fn=__UpperCAmelCase						,output_scale_factor=1						,resnet_time_scale_shift='default' if norm_type == 'group' else norm_type						,attention_head_dim=block_out_channels[-1]						,resnet_groups=__UpperCAmelCase						,temb_channels=__UpperCAmelCase						,)

													# up
													A__		   =							list(reversed(__UpperCAmelCase   )   )
													A__		   =							reversed_block_out_channels[0]
													for i, up_block_type in enumerate(__UpperCAmelCase   ):
																				A__		   =							output_channel
																				A__		   =							reversed_block_out_channels[i]

																				A__		   =							i == len(__UpperCAmelCase   ) - 1

																				A__		   =							get_up_block(
																				    __UpperCAmelCase						,num_layers=self.layers_per_block + 1						,in_channels=__UpperCAmelCase						,out_channels=__UpperCAmelCase						,prev_output_channel=__UpperCAmelCase						,add_upsample=not is_final_block						,resnet_eps=1e-6						,resnet_act_fn=__UpperCAmelCase						,resnet_groups=__UpperCAmelCase						,attention_head_dim=__UpperCAmelCase						,temb_channels=__UpperCAmelCase						,resnet_time_scale_shift=__UpperCAmelCase						,)
																				self.up_blocks.append(__UpperCAmelCase   )
																				A__		   =							output_channel

													# out
													if norm_type == "spatial":
																				A__		   =							SpatialNorm(block_out_channels[0]						,__UpperCAmelCase   )
													else:
																				A__		   =							nn.GroupNorm(num_channels=block_out_channels[0]						,num_groups=__UpperCAmelCase						,eps=1e-6   )
													A__		   =							nn.SiLU()
													A__		   =							nn.Convad(block_out_channels[0]						,__UpperCAmelCase						,3						,padding=1   )

													A__		   =							False





						def  snake_case__		(				self						,__UpperCAmelCase						,__UpperCAmelCase=None   )    -> Dict:
													A__		   =							z
													A__		   =							self.conv_in(__UpperCAmelCase   )

													A__		   =							next(iter(self.up_blocks.parameters()   )   ).dtype
													if self.training and self.gradient_checkpointing:

																				def create_custom_forward(__UpperCAmelCase   ):
																											def custom_forward(*__UpperCAmelCase   ):
																																		return module(*__UpperCAmelCase   )

																											return custom_forward

																				if is_torch_version('>='						,'1.11.0'   ):
																											# middle
																											A__		   =							torch.utils.checkpoint.checkpoint(
																											    create_custom_forward(self.mid_block   )						,__UpperCAmelCase						,__UpperCAmelCase						,use_reentrant=__UpperCAmelCase   )
																											A__		   =							sample.to(__UpperCAmelCase   )

																											# up
																											for up_block in self.up_blocks:
																																		A__		   =							torch.utils.checkpoint.checkpoint(
																																		    create_custom_forward(__UpperCAmelCase   )						,__UpperCAmelCase						,__UpperCAmelCase						,use_reentrant=__UpperCAmelCase   )
																				else:
																											# middle
																											A__		   =							torch.utils.checkpoint.checkpoint(
																											    create_custom_forward(self.mid_block   )						,__UpperCAmelCase						,__UpperCAmelCase   )
																											A__		   =							sample.to(__UpperCAmelCase   )

																											# up
																											for up_block in self.up_blocks:
																																		A__		   =							torch.utils.checkpoint.checkpoint(create_custom_forward(__UpperCAmelCase   )						,__UpperCAmelCase						,__UpperCAmelCase   )
													else:
																				# middle
																				A__		   =							self.mid_block(__UpperCAmelCase						,__UpperCAmelCase   )
																				A__		   =							sample.to(__UpperCAmelCase   )

																				# up
																				for up_block in self.up_blocks:
																											A__		   =							up_block(__UpperCAmelCase						,__UpperCAmelCase   )

        # post-process
													if latent_embeds is None:
																				A__		   =							self.conv_norm_out(__UpperCAmelCase   )
													else:
																				A__		   =							self.conv_norm_out(__UpperCAmelCase						,__UpperCAmelCase   )
													A__		   =							self.conv_act(__UpperCAmelCase   )
													A__		   =							self.conv_out(__UpperCAmelCase   )

													return sample






class 			UpperCamelCase__(	nn.Module       ):





						def __init__(				self						,__UpperCAmelCase						,__UpperCAmelCase						,__UpperCAmelCase						,__UpperCAmelCase=None						,__UpperCAmelCase="random"						,__UpperCAmelCase=False						,__UpperCAmelCase=True   )    -> Tuple:
													super().__init__()
													A__		   =							n_e
													A__		   =							vq_embed_dim
													A__		   =							beta
													A__		   =							legacy

													A__		   =							nn.Embedding(self.n_e						,self.vq_embed_dim   )
													self.embedding.weight.data.uniform_(-1.0 / self.n_e						,1.0 / self.n_e   )

													A__		   =							remap
													if self.remap is not None:
																				self.register_buffer('used'						,torch.tensor(np.load(self.remap   )   )   )
																				A__		   =							self.used.shape[0]
																				A__		   =							unknown_index  # "random" or "extra" or integer
																				if self.unknown_index == "extra":
																											A__		   =							self.re_embed
																											A__		   =							self.re_embed + 1
																				print(
																				    f'''Remapping {self.n_e} indices to {self.re_embed} indices. '''
																				    f'''Using {self.unknown_index} for unknown indices.'''   )
													else:
																				A__		   =							n_e

													A__		   =							sane_index_shape





						def  snake_case__		(				self						,__UpperCAmelCase   )    -> List[str]:
													A__		   =							inds.shape
													assert len(__UpperCAmelCase   ) > 1
													A__		   =							inds.reshape(ishape[0]						,-1   )
													A__		   =							self.used.to(__UpperCAmelCase   )
													A__		   =							(inds[:, :, None] == used[None, None, ...]).long()
													A__		   =							match.argmax(-1   )
													A__		   =							match.sum(2   ) < 1
													if self.unknown_index == "random":
																				A__		   =							torch.randint(0						,self.re_embed						,size=new[unknown].shape   ).to(device=new.device   )
													else:
																				A__		   =							self.unknown_index
													return new.reshape(__UpperCAmelCase   )





						def  snake_case__		(				self						,__UpperCAmelCase   )    -> Optional[Any]:
													A__		   =							inds.shape
													assert len(__UpperCAmelCase   ) > 1
													A__		   =							inds.reshape(ishape[0]						,-1   )
													A__		   =							self.used.to(__UpperCAmelCase   )
													if self.re_embed > self.used.shape[0]:  # extra token
																				A__		   =							0  # simply set to zero
													A__		   =							torch.gather(used[None, :][inds.shape[0] * [0], :]						,1						,__UpperCAmelCase   )
													return back.reshape(__UpperCAmelCase   )





						def  snake_case__		(				self						,__UpperCAmelCase   )    -> Dict:
													# reshape z -> (batch, height, width, channel) and flatten
													A__		   =							z.permute(0						,2						,3						,1   ).contiguous()
													A__		   =							z.view(-1						,self.vq_embed_dim   )

													# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
													A__		   =							torch.argmin(torch.cdist(__UpperCAmelCase						,self.embedding.weight   )						,dim=1   )

													A__		   =							self.embedding(__UpperCAmelCase   ).view(z.shape   )
													A__		   =							None
													A__		   =							None

													# compute loss for embedding
													if not self.legacy:
																				A__		   =							self.beta * torch.mean((z_q.detach() - z) ** 2   ) + torch.mean((z_q - z.detach()) ** 2   )
													else:
																				A__		   =							torch.mean((z_q.detach() - z) ** 2   ) + self.beta * torch.mean((z_q - z.detach()) ** 2   )

													# preserve gradients
													A__		   =							z + (z_q - z).detach()

													# reshape back to match original input shape
													A__		   =							z_q.permute(0						,3						,1						,2   ).contiguous()

													if self.remap is not None:
																				A__		   =							min_encoding_indices.reshape(z.shape[0]						,-1   )  # add batch axis
																				A__		   =							self.remap_to_used(__UpperCAmelCase   )
																				A__		   =							min_encoding_indices.reshape(-1						,1   )  # flatten

													if self.sane_index_shape:
																				A__		   =							min_encoding_indices.reshape(z_q.shape[0]						,z_q.shape[2]						,z_q.shape[3]   )

													return z_q, loss, (perplexity, min_encodings, min_encoding_indices)





						def  snake_case__		(				self						,__UpperCAmelCase						,__UpperCAmelCase   )    -> str:
													# shape specifying (batch, height, width, channel)
													if self.remap is not None:
																				A__		   =							indices.reshape(shape[0]						,-1   )  # add batch axis
																				A__		   =							self.unmap_to_all(__UpperCAmelCase   )
																				A__		   =							indices.reshape(-1   )  # flatten again

													# get quantized latent vectors
													A__		   =							self.embedding(__UpperCAmelCase   )

													if shape is not None:
																				A__		   =							z_q.view(__UpperCAmelCase   )
																				# reshape back to match original input shape
																				A__		   =							z_q.permute(0						,3						,1						,2   ).contiguous()

													return z_q






class 			UpperCamelCase__(	__A       ):





						def __init__(				self						,__UpperCAmelCase						,__UpperCAmelCase=False   )    -> Any:
													A__		   =							parameters
													A__   ,     A__		   =							torch.chunk(__UpperCAmelCase						,2						,dim=1   )
													A__		   =							torch.clamp(self.logvar						,-3_0.0						,2_0.0   )
													A__		   =							deterministic
													A__		   =							torch.exp(0.5 * self.logvar   )
													A__		   =							torch.exp(self.logvar   )
													if self.deterministic:
																				A__		   =							A__		   =							torch.zeros_like(
																				    self.mean						,device=self.parameters.device						,dtype=self.parameters.dtype   )





						def  snake_case__		(				self						,__UpperCAmelCase = None   )    -> torch.FloatTensor:
													# make sure sample is on the same device as the parameters and has same dtype
													A__		   =							randn_tensor(
													    self.mean.shape						,generator=__UpperCAmelCase						,device=self.parameters.device						,dtype=self.parameters.dtype   )
													A__		   =							self.mean + self.std * sample
													return x





						def  snake_case__		(				self						,__UpperCAmelCase=None   )    -> int:
													if self.deterministic:
																				return torch.Tensor([0.0]   )
													else:
																				if other is None:
																											return 0.5 * torch.sum(torch.pow(self.mean						,2   ) + self.var - 1.0 - self.logvar						,dim=[1, 2, 3]   )
																				else:
																											return 0.5 * torch.sum(
																											    torch.pow(self.mean - other.mean						,2   ) / other.var
																											    + self.var / other.var
																											    - 1.0
																											    - self.logvar
																											    + other.logvar						,dim=[1, 2, 3]						,)





						def  snake_case__		(				self						,__UpperCAmelCase						,__UpperCAmelCase=[1, 2, 3]   )    -> List[Any]:
													if self.deterministic:
																				return torch.Tensor([0.0]   )
													A__		   =							np.log(2.0 * np.pi   )
													return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean						,2   ) / self.var						,dim=__UpperCAmelCase   )





						def  snake_case__		(				self   )    -> Optional[Any]:
													return self.mean





