


import json
import os
import shutil
import tempfile
import unittest
from multiprocessing import get_context
from pathlib import Path

import datasets
import numpy as np
from datasets import load_dataset
from parameterized import parameterized

from transformers import AutoProcessor
from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.testing_utils import require_pyctcdecode, require_torch, require_torchaudio, slow
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_pyctcdecode_available, is_torch_available

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


if is_pyctcdecode_available():
	from huggingface_hub import snapshot_download
	from pyctcdecode import BeamSearchDecoderCTC

	from transformers.models.wavaveca_with_lm import WavaVecaProcessorWithLM
	from transformers.models.wavaveca_with_lm.processing_wavaveca_with_lm import WavaVecaDecoderWithLMOutput

if is_torch_available():
	from transformers import WavaVecaForCTC





@require_pyctcdecode
class       A    (   unittest.TestCase ):


							def 			_A       (self					):
									__lowercase=			'| <pad> <unk> <s> </s> a b c d e f g h i j k'.split()
									__lowercase=			dict(zip(lowerCAmelCase      ,				range(len(lowerCAmelCase					)					)					)					)

									__lowercase=			{
									    'unk_token': '<unk>',
									    'bos_token': '<s>',
									    'eos_token': '</s>',
									}
									__lowercase=			{
									    'feature_size': 1,
									    'padding_value': 0.0,
									    'sampling_rate': 1_6_0_0_0,
									    'return_attention_mask': False,
									    'do_normalize': True,
									}

									__lowercase=			tempfile.mkdtemp()
									__lowercase=			os.path.join(self.tmpdirname      ,				VOCAB_FILES_NAMES['vocab_file']					)
									__lowercase=			os.path.join(self.tmpdirname      ,				lowerCAmelCase					)
									with open(self.vocab_file      ,				'w'      ,				encoding='utf-8'					) as fp:
											fp.write(json.dumps(lowerCAmelCase					) + '\n'					)

									with open(self.feature_extraction_file      ,				'w'      ,				encoding='utf-8'					) as fp:
											fp.write(json.dumps(lowerCAmelCase					) + '\n'					)

									# load decoder from hub
									__lowercase=			'hf-internal-testing/ngram-beam-search-decoder'


							def 			_A       (self      ,				**lowerCAmelCase					):
									__lowercase=			self.add_kwargs_tokens_map.copy()
									kwargs.update(lowerCAmelCase					)
									return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname      ,				**lowerCAmelCase					)


							def 			_A       (self      ,				**lowerCAmelCase					):
									return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname      ,				**lowerCAmelCase					)


							def 			_A       (self      ,				**lowerCAmelCase					):
									return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name      ,				**lowerCAmelCase					)


							def 			_A       (self					):
									shutil.rmtree(self.tmpdirname					)


							def 			_A       (self					):
									__lowercase=			self.get_tokenizer()
									__lowercase=			self.get_feature_extractor()
									__lowercase=			self.get_decoder()

									__lowercase=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase      ,				feature_extractor=lowerCAmelCase      ,				decoder=lowerCAmelCase					)

									processor.save_pretrained(self.tmpdirname					)
									__lowercase=			WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname					)

									# tokenizer
									self.assertEqual(processor.tokenizer.get_vocab()      ,				tokenizer.get_vocab()					)
									self.assertIsInstance(processor.tokenizer      ,				lowerCAmelCase					)

									# feature extractor
									self.assertEqual(processor.feature_extractor.to_json_string()      ,				feature_extractor.to_json_string()					)
									self.assertIsInstance(processor.feature_extractor      ,				lowerCAmelCase					)

									# decoder
									self.assertEqual(processor.decoder._alphabet.labels      ,				decoder._alphabet.labels					)
									self.assertEqual(
									    processor.decoder.model_container[decoder._model_key]._unigram_set      ,				decoder.model_container[decoder._model_key]._unigram_set      ,				)
									self.assertIsInstance(processor.decoder      ,				lowerCAmelCase					)


							def 			_A       (self					):
									__lowercase=			WavaVecaProcessorWithLM(
									    tokenizer=self.get_tokenizer()      ,				feature_extractor=self.get_feature_extractor()      ,				decoder=self.get_decoder()					)
									processor.save_pretrained(self.tmpdirname					)

									# make sure that error is thrown when decoder alphabet doesn't match
									__lowercase=			WavaVecaProcessorWithLM.from_pretrained(
									    self.tmpdirname      ,				alpha=5.0      ,				beta=3.0      ,				score_boundary=-7.0      ,				unk_score_offset=3					)

									# decoder
									self.assertEqual(processor.language_model.alpha      ,				5.0					)
									self.assertEqual(processor.language_model.beta      ,				3.0					)
									self.assertEqual(processor.language_model.score_boundary      ,				-7.0					)
									self.assertEqual(processor.language_model.unk_score_offset      ,				3					)


							def 			_A       (self					):
									__lowercase=			self.get_tokenizer()
									# add token to trigger raise
									tokenizer.add_tokens(['xx']					)
									with self.assertRaisesRegex(lowerCAmelCase      ,				'include'					):
											WavaVecaProcessorWithLM(
											    tokenizer=lowerCAmelCase      ,				feature_extractor=self.get_feature_extractor()      ,				decoder=self.get_decoder()					)


							def 			_A       (self					):
									__lowercase=			self.get_feature_extractor()
									__lowercase=			self.get_tokenizer()
									__lowercase=			self.get_decoder()

									__lowercase=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase      ,				feature_extractor=lowerCAmelCase      ,				decoder=lowerCAmelCase					)

									__lowercase=			floats_list((3, 1_0_0_0)					)

									__lowercase=			feature_extractor(lowerCAmelCase      ,				return_tensors='np'					)
									__lowercase=			processor(lowerCAmelCase      ,				return_tensors='np'					)

									for key in input_feat_extract.keys():
											self.assertAlmostEqual(input_feat_extract[key].sum()      ,				input_processor[key].sum()      ,				delta=1E-2					)


							def 			_A       (self					):
									__lowercase=			self.get_feature_extractor()
									__lowercase=			self.get_tokenizer()
									__lowercase=			self.get_decoder()

									__lowercase=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase      ,				feature_extractor=lowerCAmelCase      ,				decoder=lowerCAmelCase					)

									__lowercase=			'This is a test string'

									__lowercase=			processor(text=lowerCAmelCase					)

									__lowercase=			tokenizer(lowerCAmelCase					)

									for key in encoded_tok.keys():
											self.assertListEqual(encoded_tok[key]      ,				encoded_processor[key]					)


							def 			_A       (self      ,				lowerCAmelCase=(2, 1_0, 1_6)      ,				lowerCAmelCase=7_7					):
									np.random.seed(lowerCAmelCase					)
									return np.random.rand(*lowerCAmelCase					)


							def 			_A       (self					):
									__lowercase=			self.get_feature_extractor()
									__lowercase=			self.get_tokenizer()
									__lowercase=			self.get_decoder()

									__lowercase=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase      ,				feature_extractor=lowerCAmelCase      ,				decoder=lowerCAmelCase					)

									__lowercase=			self._get_dummy_logits(shape=(1_0, 1_6)      ,				seed=1_3					)

									__lowercase=			processor.decode(lowerCAmelCase					)

									__lowercase=			decoder.decode_beams(lowerCAmelCase					)[0]

									self.assertEqual(decoded_decoder[0]      ,				decoded_processor.text					)
									self.assertEqual('</s> <s> </s>'      ,				decoded_processor.text					)
									self.assertEqual(decoded_decoder[-2]      ,				decoded_processor.logit_score					)
									self.assertEqual(decoded_decoder[-1]      ,				decoded_processor.lm_score					)


							@parameterized.expand([[None], ['fork'], ['spawn']]					)
							def 			_A       (self      ,				lowerCAmelCase					):
									__lowercase=			self.get_feature_extractor()
									__lowercase=			self.get_tokenizer()
									__lowercase=			self.get_decoder()

									__lowercase=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase      ,				feature_extractor=lowerCAmelCase      ,				decoder=lowerCAmelCase					)

									__lowercase=			self._get_dummy_logits()

									# note: pool should be instantiated *after* Wav2Vec2ProcessorWithLM.
									#       otherwise, the LM won't be available to the pool's sub-processes.
									# manual logic used to allow parameterized test for both pool=None and pool=Pool(...)
									if pool_context is None:
											__lowercase=			processor.batch_decode(lowerCAmelCase					)
									else:
											with get_context(lowerCAmelCase					).Pool() as pool:
													__lowercase=			processor.batch_decode(lowerCAmelCase      ,				lowerCAmelCase					)

									__lowercase=			list(lowerCAmelCase					)

									with get_context('fork'					).Pool() as p:
											__lowercase=			decoder.decode_beams_batch(lowerCAmelCase      ,				lowerCAmelCase					)

									__lowercase,							__lowercase,							__lowercase=			[], [], []
									for beams in decoded_beams:
											texts_decoder.append(beams[0][0]					)
											logit_scores_decoder.append(beams[0][-2]					)
											lm_scores_decoder.append(beams[0][-1]					)

									self.assertListEqual(lowerCAmelCase      ,				decoded_processor.text					)
									self.assertListEqual(['<s> <s> </s>', '<s> <s> <s>']      ,				decoded_processor.text					)
									self.assertListEqual(lowerCAmelCase      ,				decoded_processor.logit_score					)
									self.assertListEqual(lowerCAmelCase      ,				decoded_processor.lm_score					)


							def 			_A       (self					):
									__lowercase=			self.get_feature_extractor()
									__lowercase=			self.get_tokenizer()
									__lowercase=			self.get_decoder()

									__lowercase=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase      ,				feature_extractor=lowerCAmelCase      ,				decoder=lowerCAmelCase					)

									__lowercase=			self._get_dummy_logits()

									__lowercase=			1_5
									__lowercase=			-20.0
									__lowercase=			-4.0

									__lowercase=			processor.batch_decode(
									    lowerCAmelCase      ,				beam_width=lowerCAmelCase      ,				beam_prune_logp=lowerCAmelCase      ,				token_min_logp=lowerCAmelCase      ,				)
									__lowercase=			decoded_processor_out.text

									__lowercase=			list(lowerCAmelCase					)

									with get_context('fork'					).Pool() as pool:
											__lowercase=			decoder.decode_beams_batch(
											    lowerCAmelCase      ,				lowerCAmelCase      ,				beam_width=lowerCAmelCase      ,				beam_prune_logp=lowerCAmelCase      ,				token_min_logp=lowerCAmelCase      ,				)

									__lowercase=			[d[0][0] for d in decoded_decoder_out]
									__lowercase=			[d[0][2] for d in decoded_decoder_out]
									__lowercase=			[d[0][3] for d in decoded_decoder_out]

									self.assertListEqual(lowerCAmelCase      ,				lowerCAmelCase					)
									self.assertListEqual(['</s> <s> <s>', '<s> <s> <s>']      ,				lowerCAmelCase					)

									self.assertTrue(np.array_equal(lowerCAmelCase      ,				decoded_processor_out.logit_score					)					)
									self.assertTrue(np.allclose([-20.0_54, -18.4_47]      ,				lowerCAmelCase      ,				atol=1E-3					)					)

									self.assertTrue(np.array_equal(lowerCAmelCase      ,				decoded_processor_out.lm_score					)					)
									self.assertTrue(np.allclose([-15.5_54, -13.94_74]      ,				lowerCAmelCase      ,				atol=1E-3					)					)


							def 			_A       (self					):
									__lowercase=			self.get_feature_extractor()
									__lowercase=			self.get_tokenizer()
									__lowercase=			self.get_decoder()

									__lowercase=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase      ,				feature_extractor=lowerCAmelCase      ,				decoder=lowerCAmelCase					)

									__lowercase=			self._get_dummy_logits()

									__lowercase=			2.0
									__lowercase=			5.0
									__lowercase=			-20.0
									__lowercase=			True

									__lowercase=			processor.batch_decode(
									    lowerCAmelCase      ,				alpha=lowerCAmelCase      ,				beta=lowerCAmelCase      ,				unk_score_offset=lowerCAmelCase      ,				lm_score_boundary=lowerCAmelCase      ,				)
									__lowercase=			decoded_processor_out.text

									__lowercase=			list(lowerCAmelCase					)
									decoder.reset_params(
									    alpha=lowerCAmelCase      ,				beta=lowerCAmelCase      ,				unk_score_offset=lowerCAmelCase      ,				lm_score_boundary=lowerCAmelCase      ,				)

									with get_context('fork'					).Pool() as pool:
											__lowercase=			decoder.decode_beams_batch(
											    lowerCAmelCase      ,				lowerCAmelCase      ,				)

									__lowercase=			[d[0][0] for d in decoded_decoder_out]

									self.assertListEqual(lowerCAmelCase      ,				lowerCAmelCase					)
									self.assertListEqual(['<s> </s> <s> </s> </s>', '</s> </s> <s> </s> </s>']      ,				lowerCAmelCase					)
									__lowercase=			processor.decoder.model_container[processor.decoder._model_key]
									self.assertEqual(lm_model.alpha      ,				2.0					)
									self.assertEqual(lm_model.beta      ,				5.0					)
									self.assertEqual(lm_model.unk_score_offset      ,				-20.0					)
									self.assertEqual(lm_model.score_boundary      ,				lowerCAmelCase					)


							def 			_A       (self					):
									__lowercase=			WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'					)

									__lowercase=			processor.decoder.model_container[processor.decoder._model_key]
									__lowercase=			Path(language_model._kenlm_model.path.decode('utf-8'					)					).parent.parent.absolute()

									__lowercase=			os.listdir(lowerCAmelCase					)
									__lowercase=			['alphabet.json', 'language_model']

									downloaded_decoder_files.sort()
									expected_decoder_files.sort()

									# test that only decoder relevant files from
									# https://huggingface.co/hf-internal-testing/processor_with_lm/tree/main
									# are downloaded and none of the rest (e.g. README.md, ...)
									self.assertListEqual(lowerCAmelCase      ,				lowerCAmelCase					)


							def 			_A       (self					):
									__lowercase=			snapshot_download('hf-internal-testing/processor_with_lm'					)

									__lowercase=			WavaVecaProcessorWithLM.from_pretrained(lowerCAmelCase					)

									__lowercase=			processor.decoder.model_container[processor.decoder._model_key]
									__lowercase=			Path(language_model._kenlm_model.path.decode('utf-8'					)					).parent.parent.absolute()

									__lowercase=			os.listdir(lowerCAmelCase					)
									__lowercase=			os.listdir(lowerCAmelCase					)

									local_decoder_files.sort()
									expected_decoder_files.sort()

									# test that both decoder form hub and local files in cache are the same
									self.assertListEqual(lowerCAmelCase      ,				lowerCAmelCase					)


							def 			_A       (self					):
									__lowercase=			WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'					)
									__lowercase=			AutoProcessor.from_pretrained('hf-internal-testing/processor_with_lm'					)

									__lowercase=			floats_list((3, 1_0_0_0)					)

									__lowercase=			processor_wavaveca(lowerCAmelCase      ,				return_tensors='np'					)
									__lowercase=			processor_auto(lowerCAmelCase      ,				return_tensors='np'					)

									for key in input_wavaveca.keys():
											self.assertAlmostEqual(input_wavaveca[key].sum()      ,				input_auto[key].sum()      ,				delta=1E-2					)

									__lowercase=			self._get_dummy_logits()

									__lowercase=			processor_wavaveca.batch_decode(lowerCAmelCase					)
									__lowercase=			processor_auto.batch_decode(lowerCAmelCase					)

									self.assertListEqual(decoded_wavaveca.text      ,				decoded_auto.text					)


							def 			_A       (self					):
									__lowercase=			self.get_feature_extractor()
									__lowercase=			self.get_tokenizer()
									__lowercase=			self.get_decoder()

									__lowercase=			WavaVecaProcessorWithLM(tokenizer=lowerCAmelCase      ,				feature_extractor=lowerCAmelCase      ,				decoder=lowerCAmelCase					)

									self.assertListEqual(
									    processor.model_input_names      ,				feature_extractor.model_input_names      ,				msg='`processor` and `feature_extractor` model input names do not match'      ,				)


							@staticmethod
							def 			_A       (lowerCAmelCase      ,				lowerCAmelCase					):
									__lowercase=			[d[key] for d in offsets]
									return retrieved_list


							def 			_A       (self					):
									__lowercase=			WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'					)
									__lowercase=			self._get_dummy_logits()[0]

									__lowercase=			processor.decode(lowerCAmelCase      ,				output_word_offsets=lowerCAmelCase					)
									# check Wav2Vec2CTCTokenizerOutput keys for word
									self.assertEqual(len(outputs.keys()					)      ,				4					)
									self.assertTrue('text' in outputs					)
									self.assertTrue('word_offsets' in outputs					)
									self.assertTrue(isinstance(lowerCAmelCase      ,				lowerCAmelCase					)					)

									self.assertEqual(' '.join(self.get_from_offsets(outputs['word_offsets']      ,				'word'					)					)      ,				outputs.text					)
									self.assertListEqual(self.get_from_offsets(outputs['word_offsets']      ,				'word'					)      ,				['<s>', '<s>', '</s>']					)
									self.assertListEqual(self.get_from_offsets(outputs['word_offsets']      ,				'start_offset'					)      ,				[0, 2, 4]					)
									self.assertListEqual(self.get_from_offsets(outputs['word_offsets']      ,				'end_offset'					)      ,				[1, 3, 5]					)


							def 			_A       (self					):
									__lowercase=			WavaVecaProcessorWithLM.from_pretrained('hf-internal-testing/processor_with_lm'					)
									__lowercase=			self._get_dummy_logits()

									__lowercase=			processor.batch_decode(lowerCAmelCase      ,				output_word_offsets=lowerCAmelCase					)

									# check Wav2Vec2CTCTokenizerOutput keys for word
									self.assertEqual(len(outputs.keys()					)      ,				4					)
									self.assertTrue('text' in outputs					)
									self.assertTrue('word_offsets' in outputs					)
									self.assertTrue(isinstance(lowerCAmelCase      ,				lowerCAmelCase					)					)

									self.assertListEqual(
									    [' '.join(self.get_from_offsets(lowerCAmelCase      ,				'word'					)					) for o in outputs['word_offsets']]      ,				outputs.text					)
									self.assertListEqual(self.get_from_offsets(outputs['word_offsets'][0]      ,				'word'					)      ,				['<s>', '<s>', '</s>']					)
									self.assertListEqual(self.get_from_offsets(outputs['word_offsets'][0]      ,				'start_offset'					)      ,				[0, 2, 4]					)
									self.assertListEqual(self.get_from_offsets(outputs['word_offsets'][0]      ,				'end_offset'					)      ,				[1, 3, 5]					)







							@slow
							@require_torch
							@require_torchaudio
							def 			_A       (self					):
									import torch

									__lowercase=			load_dataset('common_voice'      ,				'en'      ,				split='train'      ,				streaming=lowerCAmelCase					)
									__lowercase=			ds.cast_column('audio'      ,				datasets.Audio(sampling_rate=1_6_0_0_0					)					)
									__lowercase=			iter(lowerCAmelCase					)
									__lowercase=			next(lowerCAmelCase					)

									__lowercase=			AutoProcessor.from_pretrained('patrickvonplaten/wav2vec2-base-100h-with-lm'					)
									__lowercase=			WavaVecaForCTC.from_pretrained('patrickvonplaten/wav2vec2-base-100h-with-lm'					)

									# compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
									__lowercase=			processor(sample['audio']['array']      ,				return_tensors='pt'					).input_values

									with torch.no_grad():
											__lowercase=			model(lowerCAmelCase					).logits.cpu().numpy()

									__lowercase=			processor.decode(logits[0]      ,				output_word_offsets=lowerCAmelCase					)

									__lowercase=			model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
									__lowercase=			[
									    {
									        'start_time': d['start_offset'] * time_offset,
									        'end_time': d['end_offset'] * time_offset,
									        'word': d['word'],
									    }
									    for d in output['word_offsets']
									]

									__lowercase=			'WHY DOES MILISANDRA LOOK LIKE SHE WANTS TO CONSUME JOHN SNOW ON THE RIVER AT THE WALL'

									# output words
									self.assertEqual(' '.join(self.get_from_offsets(lowerCAmelCase      ,				'word'					)					)      ,				lowerCAmelCase					)
									self.assertEqual(' '.join(self.get_from_offsets(lowerCAmelCase      ,				'word'					)					)      ,				output.text					)

									# output times
									__lowercase=			torch.tensor(self.get_from_offsets(lowerCAmelCase      ,				'start_time'					)					)
									__lowercase=			torch.tensor(self.get_from_offsets(lowerCAmelCase      ,				'end_time'					)					)

									# fmt: off
									__lowercase=			torch.tensor([1.41_99, 1.65_99, 2.25_99, 3.0, 3.24, 3.59_99, 3.79_99, 4.09_99, 4.26, 4.94, 5.28, 5.65_99, 5.78, 5.94, 6.32, 6.53_99, 6.65_99]					)
									__lowercase=			torch.tensor([1.53_99, 1.89_99, 2.9, 3.16, 3.53_99, 3.72, 4.01_99, 4.17_99, 4.76, 5.15_99, 5.55_99, 5.69_99, 5.86, 6.19_99, 6.38, 6.61_99, 6.94]					)
									# fmt: on

									self.assertTrue(torch.allclose(lowerCAmelCase      ,				lowerCAmelCase      ,				atol=0.01					)					)
									self.assertTrue(torch.allclose(lowerCAmelCase      ,				lowerCAmelCase      ,				atol=0.01					)					)










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, apply_forward_hook
from .modeling_utils import ModelMixin
from .vae import Decoder, DecoderOutput, Encoder, VectorQuantizer





@dataclass
class       A    (   A_ ):
							UpperCamelCase_   :							torch.FloatTensor





class       A    (   A_					,				A_ ):


							@register_to_config
							def __init__(self      ,				lowerCAmelCase = 3      ,				lowerCAmelCase = 3      ,				lowerCAmelCase = ("DownEncoderBlock2D",)      ,				lowerCAmelCase = ("UpDecoderBlock2D",)      ,				lowerCAmelCase = (6_4,)      ,				lowerCAmelCase = 1      ,				lowerCAmelCase = "silu"      ,				lowerCAmelCase = 3      ,				lowerCAmelCase = 3_2      ,				lowerCAmelCase = 2_5_6      ,				lowerCAmelCase = 3_2      ,				lowerCAmelCase = None      ,				lowerCAmelCase = 0.1_82_15      ,				lowerCAmelCase = "group"      ,				):
									super().__init__()

									# pass init params to Encoder
									__lowercase=			Encoder(
									    in_channels=lowerCAmelCase      ,				out_channels=lowerCAmelCase      ,				down_block_types=lowerCAmelCase      ,				block_out_channels=lowerCAmelCase      ,				layers_per_block=lowerCAmelCase      ,				act_fn=lowerCAmelCase      ,				norm_num_groups=lowerCAmelCase      ,				double_z=lowerCAmelCase      ,				)

									__lowercase=			vq_embed_dim if vq_embed_dim is not None else latent_channels

									__lowercase=			nn.Convad(lowerCAmelCase      ,				lowerCAmelCase      ,				1					)
									__lowercase=			VectorQuantizer(lowerCAmelCase      ,				lowerCAmelCase      ,				beta=0.25      ,				remap=lowerCAmelCase      ,				sane_index_shape=lowerCAmelCase					)
									__lowercase=			nn.Convad(lowerCAmelCase      ,				lowerCAmelCase      ,				1					)

									# pass init params to Decoder
									__lowercase=			Decoder(
									    in_channels=lowerCAmelCase      ,				out_channels=lowerCAmelCase      ,				up_block_types=lowerCAmelCase      ,				block_out_channels=lowerCAmelCase      ,				layers_per_block=lowerCAmelCase      ,				act_fn=lowerCAmelCase      ,				norm_num_groups=lowerCAmelCase      ,				norm_type=lowerCAmelCase      ,				)


							@apply_forward_hook
							def 			_A       (self      ,				lowerCAmelCase      ,				lowerCAmelCase = True					):
									__lowercase=			self.encoder(lowerCAmelCase					)
									__lowercase=			self.quant_conv(lowerCAmelCase					)

									if not return_dict:
											return (h,)

									return VQEncoderOutput(latents=lowerCAmelCase					)


							@apply_forward_hook
							def 			_A       (self      ,				lowerCAmelCase      ,				lowerCAmelCase = False      ,				lowerCAmelCase = True					):
									# also go through quantization layer
									if not force_not_quantize:
											__lowercase,							__lowercase,							__lowercase=			self.quantize(lowerCAmelCase					)
									else:
											__lowercase=			h
									__lowercase=			self.post_quant_conv(lowerCAmelCase					)
									__lowercase=			self.decoder(lowerCAmelCase      ,				quant if self.config.norm_type == 'spatial' else None					)

									if not return_dict:
											return (dec,)

									return DecoderOutput(sample=lowerCAmelCase					)







							def 			_A       (self      ,				lowerCAmelCase      ,				lowerCAmelCase = True					):
									__lowercase=			sample
									__lowercase=			self.encode(lowerCAmelCase					).latents
									__lowercase=			self.decode(lowerCAmelCase					).sample

									if not return_dict:
											return (dec,)

									return DecoderOutput(sample=lowerCAmelCase					)






