

import inspect
import unittest

from transformers import SwinaSRConfig
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, _config_zero_init, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
	import torch
	from torch import nn

	from transformers import SwinaSRForImageSuperResolution, SwinaSRModel
	from transformers.models.swinasr.modeling_swinasr import SWIN2SR_PRETRAINED_MODEL_ARCHIVE_LIST

if is_vision_available():
	from PIL import Image

	from transformers import SwinaSRImageProcessor



class 					A__:
						"""simple docstring"""






						def __init__(						self     ,			_lowercase     ,			_lowercase=13     ,			_lowercase=32     ,			_lowercase=1     ,			_lowercase=3     ,			_lowercase=16     ,			_lowercase=[1, 2, 1]     ,			_lowercase=[2, 2, 4]     ,			_lowercase=2     ,			_lowercase=2.0     ,			_lowercase=True     ,			_lowercase=0.0     ,			_lowercase=0.0     ,			_lowercase=0.1     ,			_lowercase="gelu"     ,			_lowercase=False     ,			_lowercase=True     ,			_lowercase=0.0_2     ,			_lowercase=1e-5     ,			_lowercase=True     ,			_lowercase=None     ,			_lowercase=False     ,			_lowercase=2     ,			)    ->   int:
									a_	:	List[Any]          =   parent
									a_	:	List[str]          =   batch_size
									a_	:	int          =   image_size
									a_	:	Tuple          =   patch_size
									a_	:	List[Any]          =   num_channels
									a_	:	str          =   embed_dim
									a_	:	Optional[int]          =   depths
									a_	:	Any          =   num_heads
									a_	:	List[str]          =   window_size
									a_	:	List[Any]          =   mlp_ratio
									a_	:	int          =   qkv_bias
									a_	:	Union[str, Any]          =   hidden_dropout_prob
									a_	:	int          =   attention_probs_dropout_prob
									a_	:	Union[str, Any]          =   drop_path_rate
									a_	:	Optional[int]          =   hidden_act
									a_	:	List[str]          =   use_absolute_embeddings
									a_	:	Any          =   patch_norm
									a_	:	List[Any]          =   layer_norm_eps
									a_	:	Dict          =   initializer_range
									a_	:	Optional[Any]          =   is_training
									a_	:	Tuple          =   scope
									a_	:	str          =   use_labels
									a_	:	List[str]          =   upscale

									# here we set some attributes to make tests pass
									a_	:	Any          =   len(lowercase_						)
									a_	:	str          =   embed_dim
									a_	:	Optional[int]          =   (image_size // patch_size) ** 2





						def  UpperCamelCase__				(						self						)    ->   Optional[Any]:
									a_	:	List[str]          =   floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]						)

									a_	:	Optional[int]          =   None
									if self.use_labels:
												a_	:	Dict          =   ids_tensor([self.batch_size]     ,			self.type_sequence_label_size						)

									a_	:	Optional[Any]          =   self.get_config()

									return config, pixel_values, labels





						def  UpperCamelCase__				(						self						)    ->   Tuple:
									return SwinaSRConfig(
									    image_size=self.image_size     ,			patch_size=self.patch_size     ,			num_channels=self.num_channels     ,			embed_dim=self.embed_dim     ,			depths=self.depths     ,			num_heads=self.num_heads     ,			window_size=self.window_size     ,			mlp_ratio=self.mlp_ratio     ,			qkv_bias=self.qkv_bias     ,			hidden_dropout_prob=self.hidden_dropout_prob     ,			attention_probs_dropout_prob=self.attention_probs_dropout_prob     ,			drop_path_rate=self.drop_path_rate     ,			hidden_act=self.hidden_act     ,			use_absolute_embeddings=self.use_absolute_embeddings     ,			path_norm=self.patch_norm     ,			layer_norm_eps=self.layer_norm_eps     ,			initializer_range=self.initializer_range     ,			upscale=self.upscale     ,			)





						def  UpperCamelCase__				(						self     ,			_lowercase     ,			_lowercase     ,			_lowercase						)    ->   List[str]:
									a_	:	int          =   SwinaSRModel(config=lowercase_						)
									model.to(lowercase_						)
									model.eval()
									a_	:	Union[str, Any]          =   model(lowercase_						)

									self.parent.assertEqual(
									    result.last_hidden_state.shape     ,			(self.batch_size, self.embed_dim, self.image_size, self.image_size)						)





						def  UpperCamelCase__				(						self     ,			_lowercase     ,			_lowercase     ,			_lowercase						)    ->   List[Any]:
									a_	:	Union[str, Any]          =   SwinaSRForImageSuperResolution(lowercase_						)
									model.to(lowercase_						)
									model.eval()
									a_	:	str          =   model(lowercase_						)

									a_	:	Any          =   self.image_size * self.upscale
									self.parent.assertEqual(
									    result.reconstruction.shape     ,			(self.batch_size, self.num_channels, expected_image_size, expected_image_size)						)





						def  UpperCamelCase__				(						self						)    ->   List[str]:
									a_	:	Any          =   self.prepare_config_and_inputs()
									a_       ,				a_       ,				a_	:	Dict          =   config_and_inputs
									a_	:	Dict          =   {"""pixel_values""": pixel_values}
									return config, inputs_dict







@require_torch
class 					A__(lowerCamelCase__,     lowerCamelCase__,     unittest.TestCase       ):
						"""simple docstring"""






						_A						:							Dict									= (SwinaSRModel, SwinaSRForImageSuperResolution) if is_torch_available() else ()
						_A						:							Optional[int]									= {'feature-extraction': SwinaSRModel} if is_torch_available() else {}

						_A						:							str									= False
						_A						:							Any									= False
						_A						:							Dict									= False
						_A						:							int									= False
						_A						:							Any									= False





						def  UpperCamelCase__				(						self						)    ->   List[Any]:
									a_	:	Dict          =   SwinaSRModelTester(self						)
									a_	:	List[Any]          =   ConfigTester(self     ,			config_class=lowercase_     ,			embed_dim=37						)





						def  UpperCamelCase__				(						self						)    ->   Any:
									self.config_tester.create_and_test_config_to_json_string()
									self.config_tester.create_and_test_config_to_json_file()
									self.config_tester.create_and_test_config_from_and_save_pretrained()
									self.config_tester.create_and_test_config_with_num_labels()
									self.config_tester.check_config_can_be_init_without_params()
									self.config_tester.check_config_arguments_init()





						def  UpperCamelCase__				(						self						)    ->   Any:
									a_	:	Dict          =   self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_model(*lowercase_						)





						def  UpperCamelCase__				(						self						)    ->   Optional[int]:
									a_	:	str          =   self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_image_super_resolution(*lowercase_						)





						@unittest.skip(reason="""Got `CUDA error: misaligned address` with PyTorch 2.0.0."""						)
						def  UpperCamelCase__				(						self						)    ->   Optional[Any]:
									pass





						@unittest.skip(reason="""Swin2SR does not use inputs_embeds"""						)
						def  UpperCamelCase__				(						self						)    ->   Dict:
									pass





						@unittest.skip(reason="""Swin2SR does not support training yet"""						)
						def  UpperCamelCase__				(						self						)    ->   List[Any]:
									pass





						@unittest.skip(reason="""Swin2SR does not support training yet"""						)
						def  UpperCamelCase__				(						self						)    ->   Union[str, Any]:
									pass





						def  UpperCamelCase__				(						self						)    ->   Union[str, Any]:
									a_       ,				a_	:	str          =   self.model_tester.prepare_config_and_inputs_for_common()

									for model_class in self.all_model_classes:
												a_	:	str          =   model_class(lowercase_						)
												self.assertIsInstance(model.get_input_embeddings()     ,			(nn.Module)						)
												a_	:	Optional[int]          =   model.get_output_embeddings()
												self.assertTrue(x is None or isinstance(lowercase_     ,			nn.Linear						)						)





						def  UpperCamelCase__				(						self						)    ->   str:
									a_       ,				a_	:	int          =   self.model_tester.prepare_config_and_inputs_for_common()

									for model_class in self.all_model_classes:
												a_	:	Tuple          =   model_class(lowercase_						)
												a_	:	List[str]          =   inspect.signature(model.forward						)
												# signature.parameters is an OrderedDict => so arg_names order is deterministic
												a_	:	Any          =   [*signature.parameters.keys()]

												a_	:	Optional[int]          =   ["""pixel_values"""]
												self.assertListEqual(arg_names[:1]     ,			lowercase_						)





						@slow
						def  UpperCamelCase__				(						self						)    ->   Optional[int]:






									for model_name in SWIN2SR_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
												a_	:	Optional[int]          =   SwinaSRModel.from_pretrained(lowercase_						)
												self.assertIsNotNone(lowercase_						)





						def  UpperCamelCase__				(						self						)    ->   Dict:
									a_       ,				a_	:	Optional[Any]          =   self.model_tester.prepare_config_and_inputs_for_common()

									a_	:	Union[str, Any]          =   _config_zero_init(lowercase_						)
									for model_class in self.all_model_classes:
												a_	:	Optional[Any]          =   model_class(config=lowercase_						)
												for name, param in model.named_parameters():
															if "logit_scale" in name:
																		continue
															if param.requires_grad:
																		self.assertIn(
																		    ((param.data.mean() * 1e9).round() / 1e9).item()     ,			[0.0, 1.0]     ,			msg=F'''Parameter {name} of model {model_class} seems not properly initialized'''     ,			)





						def  UpperCamelCase__				(						self						)    ->   Dict:
									a_       ,				a_	:	Optional[Any]          =   self.model_tester.prepare_config_and_inputs_for_common()
									a_	:	List[Any]          =   True

									for model_class in self.all_model_classes:
												a_	:	Any          =   True
												a_	:	Dict          =   False
												a_	:	Dict          =   True
												a_	:	Optional[Any]          =   model_class(lowercase_						)
												model.to(lowercase_						)
												model.eval()
												with torch.no_grad():
															a_	:	Optional[int]          =   model(**self._prepare_for_class(lowercase_     ,			lowercase_						)						)
												a_	:	str          =   outputs.attentions
												a_	:	Optional[Any]          =   len(self.model_tester.depths						)
												self.assertEqual(len(lowercase_						)     ,			lowercase_						)

												# check that output_attentions also work using config
												del inputs_dict["output_attentions"]
												a_	:	Any          =   True
												a_	:	Optional[Any]          =   config.window_size**2
												a_	:	Dict          =   model_class(lowercase_						)
												model.to(lowercase_						)
												model.eval()
												with torch.no_grad():
															a_	:	List[Any]          =   model(**self._prepare_for_class(lowercase_     ,			lowercase_						)						)
												a_	:	int          =   outputs.attentions
												self.assertEqual(len(lowercase_						)     ,			lowercase_						)

												self.assertListEqual(
												    list(attentions[0].shape[-3:]						)     ,			[self.model_tester.num_heads[0], window_size_squared, window_size_squared]     ,			)
												a_	:	Optional[Any]          =   len(lowercase_						)

												# Check attention is always last and order is fine
												a_	:	int          =   True
												a_	:	Union[str, Any]          =   True
												a_	:	Union[str, Any]          =   model_class(lowercase_						)
												model.to(lowercase_						)
												model.eval()
												with torch.no_grad():
															a_	:	Optional[int]          =   model(**self._prepare_for_class(lowercase_     ,			lowercase_						)						)

												self.assertEqual(out_len + 1     ,			len(lowercase_						)						)

												a_	:	str          =   outputs.attentions

												self.assertEqual(len(lowercase_						)     ,			lowercase_						)

												self.assertListEqual(
												    list(self_attentions[0].shape[-3:]						)     ,			[self.model_tester.num_heads[0], window_size_squared, window_size_squared]     ,			)







@require_vision
@require_torch
@slow
class 					A__(unittest.TestCase       ):
						"""simple docstring"""






						def  UpperCamelCase__				(						self						)    ->   int:
									a_	:	int          =   SwinaSRImageProcessor()
									a_	:	Optional[Any]          =   SwinaSRForImageSuperResolution.from_pretrained("""caidas/swin2SR-classical-sr-x2-64"""						).to(lowercase_						)

									a_	:	Optional[Any]          =   Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""						)
									a_	:	Union[str, Any]          =   processor(images=lowercase_     ,			return_tensors="""pt"""						).to(lowercase_						)

									# forward pass
									with torch.no_grad():
												a_	:	Any          =   model(**lowercase_						)

									# verify the logits
									a_	:	Optional[Any]          =   torch.Size([1, 3, 976, 1_296]						)
									self.assertEqual(outputs.reconstruction.shape     ,			lowercase_						)
									a_	:	Optional[int]          =   torch.tensor(
									    [[0.5_4_5_8, 0.5_5_4_6, 0.5_6_3_8], [0.5_5_2_6, 0.5_5_6_5, 0.5_6_5_1], [0.5_3_9_6, 0.5_4_2_6, 0.5_6_2_1]]						).to(lowercase_						)
									self.assertTrue(torch.allclose(outputs.reconstruction[0, 0, :3, :3]     ,			lowercase_     ,			atol=1e-4						)						)





import os
from typing import Optional, Union

import numpy as np
from huggingface_hub import hf_hub_download

from ... import AutoTokenizer
from ...utils import logging


__snake_case							:			Optional[int]      		= """block_records.npy"""


__snake_case							:			List[Any]      		= logging.get_logger(__name__)


def 				_UpperCAmelCase							(	a__     ,  a__):

			'''simple docstring'''



			import tensorflow.compat.va as tf

			a_	:	List[Any]          =   tf.data.TFRecordDataset(a__     ,  buffer_size=5_1_2 * 1_0_2_4 * 1_0_2_4)
			a_	:	Optional[int]          =   blocks_dataset.batch(a__     ,  drop_remainder=a__)
			a_	:	Tuple          =   next(blocks_dataset.take(1).as_numpy_iterator())

			return np_record



class 					A__:
						"""simple docstring"""






						def __init__(						self     ,			_lowercase     ,			_lowercase     ,			_lowercase=2     ,			_lowercase=1_000     ,			_lowercase=100     ,			_lowercase=100_000     ,			)    ->   Optional[Any]:
									from scann.scann_ops.py.scann_ops_pybind import builder as Builder

									a_	:	Optional[int]          =   Builder(db=_lowercase     ,			num_neighbors=_lowercase     ,			distance_measure="""dot_product"""						)
									a_	:	Dict          =   builder.tree(
									    num_leaves=_lowercase     ,			num_leaves_to_search=_lowercase     ,			training_sample_size=_lowercase						)
									a_	:	Optional[Any]          =   builder.score_ah(dimensions_per_block=_lowercase						)

									a_	:	Optional[int]          =   builder.build()





						def  UpperCamelCase__				(						self     ,			_lowercase						)    ->   str:
									a_       ,				a_	:	Union[str, Any]          =   self.searcher.search_batched(question_projection.detach().cpu()						)
									return retrieved_block_ids.astype("""int64"""						)







class 					A__:
						"""simple docstring"""






						def __init__(						self     ,			_lowercase     ,			_lowercase						)    ->   Optional[Any]:
									super().__init__()
									a_	:	List[Any]          =   block_records
									a_	:	Tuple          =   tokenizer





						def __call__(						self     ,			_lowercase     ,			_lowercase     ,			_lowercase     ,			_lowercase=None     ,			_lowercase="pt"						)    ->   Optional[Any]:
									a_	:	List[str]          =   np.take(self.block_records     ,			indices=_lowercase     ,			axis=0						)

									a_	:	Optional[Any]          =   self.tokenizer.decode(question_input_ids[0]     ,			skip_special_tokens=_lowercase						)

									a_	:	str          =   []
									a_	:	List[Any]          =   []
									for retrieved_block in retrieved_blocks:
												text.append(_lowercase						)
												text_pair.append(retrieved_block.decode()						)

									a_	:	str          =   self.tokenizer(
									    _lowercase     ,			_lowercase     ,			padding=_lowercase     ,			truncation=_lowercase     ,			return_special_tokens_mask=_lowercase     ,			max_length=_lowercase						)
									a_	:	Any          =   concat_inputs.convert_to_tensors(_lowercase						)

									if answer_ids is not None:
												return self.block_has_answer(_lowercase     ,			_lowercase						) + (concat_inputs_tensors,)
									else:
												return (None, None, None, concat_inputs_tensors)





						@classmethod
						def  UpperCamelCase__				(						cls     ,			_lowercase     ,			*_lowercase     ,			**_lowercase						)    ->   Tuple:
									if os.path.isdir(_lowercase						):
												a_	:	Optional[Any]          =   os.path.join(_lowercase     ,			_REALM_BLOCK_RECORDS_FILENAME						)
									else:
												a_	:	Tuple          =   hf_hub_download(
												    repo_id=_lowercase     ,			filename=_REALM_BLOCK_RECORDS_FILENAME     ,			**_lowercase						)
									a_	:	int          =   np.load(_lowercase     ,			allow_pickle=_lowercase						)

									a_	:	Tuple          =   AutoTokenizer.from_pretrained(_lowercase     ,			*_lowercase     ,			**_lowercase						)

									return cls(_lowercase     ,			_lowercase						)





						def  UpperCamelCase__				(						self     ,			_lowercase						)    ->   List[Any]:
									# save block records
									np.save(os.path.join(_lowercase     ,			_REALM_BLOCK_RECORDS_FILENAME						)     ,			self.block_records						)
									# save tokenizer
									self.tokenizer.save_pretrained(_lowercase						)





						def  UpperCamelCase__				(						self     ,			_lowercase     ,			_lowercase						)    ->   Any:
									a_	:	List[Any]          =   []
									a_	:	str          =   []
									a_	:	List[str]          =   []
									a_	:	Optional[int]          =   0

									for input_id in concat_inputs.input_ids:
												a_	:	Optional[Any]          =   input_id.tolist()
												# Check answers between two [SEP] tokens
												a_	:	Tuple          =   input_id_list.index(self.tokenizer.sep_token_id						)
												a_	:	Optional[Any]          =   first_sep_idx + 1 + input_id_list[first_sep_idx + 1 :].index(self.tokenizer.sep_token_id						)

												start_pos.append([]						)
												end_pos.append([]						)
												for answer in answer_ids:
															for idx in range(first_sep_idx + 1     ,			_lowercase						):
																		if answer[0] == input_id_list[idx]:
																					if input_id_list[idx : idx + len(_lowercase						)] == answer:
																								start_pos[-1].append(_lowercase						)
																								end_pos[-1].append(idx + len(_lowercase						) - 1						)

												if len(start_pos[-1]						) == 0:
															has_answers.append(_lowercase						)
												else:
															has_answers.append(_lowercase						)
															if len(start_pos[-1]						) > max_answers:
																		a_	:	List[str]          =   len(start_pos[-1]						)

        # Pad -1 to max_answers
									for start_pos_, end_pos_ in zip(_lowercase     ,			_lowercase						):
												if len(start_pos_						) < max_answers:
															a_	:	Tuple          =   [-1] * (max_answers - len(start_pos_						))
															start_pos_ += padded
															end_pos_ += padded
									return has_answers, start_pos, end_pos


