



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  __lowerCAmelCase			:




				def __init__(							self						,    _snake_case						,    _snake_case=13						,    _snake_case=32						,    _snake_case=1						,    _snake_case=3						,    _snake_case=16						,    _snake_case=[1, 2, 1]						,    _snake_case=[2, 2, 4]						,    _snake_case=2						,    _snake_case=2.0						,    _snake_case=True						,    _snake_case=0.0						,    _snake_case=0.0						,    _snake_case=0.1						,    _snake_case="gelu"						,    _snake_case=False						,    _snake_case=True						,    _snake_case=0.02						,    _snake_case=1e-5						,    _snake_case=True						,    _snake_case=None						,    _snake_case=False						,    _snake_case=2						,    ):



									"""simple docstring"""





									_lowerCAmelCase      =   parent
									_lowerCAmelCase      =   batch_size
									_lowerCAmelCase      =   image_size
									_lowerCAmelCase      =   patch_size
									_lowerCAmelCase      =   num_channels
									_lowerCAmelCase      =   embed_dim
									_lowerCAmelCase      =   depths
									_lowerCAmelCase      =   num_heads
									_lowerCAmelCase      =   window_size
									_lowerCAmelCase      =   mlp_ratio
									_lowerCAmelCase      =   qkv_bias
									_lowerCAmelCase      =   hidden_dropout_prob
									_lowerCAmelCase      =   attention_probs_dropout_prob
									_lowerCAmelCase      =   drop_path_rate
									_lowerCAmelCase      =   hidden_act
									_lowerCAmelCase      =   use_absolute_embeddings
									_lowerCAmelCase      =   patch_norm
									_lowerCAmelCase      =   layer_norm_eps
									_lowerCAmelCase      =   initializer_range
									_lowerCAmelCase      =   is_training
									_lowerCAmelCase      =   scope
									_lowerCAmelCase      =   use_labels
									_lowerCAmelCase      =   upscale

									# here we set some attributes to make tests pass
									_lowerCAmelCase      =   len(_snake_case		)
									_lowerCAmelCase      =   embed_dim
									_lowerCAmelCase      =   (image_size // patch_size) ** 2




				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]		)

									_lowerCAmelCase      =   None
									if self.use_labels:
														_lowerCAmelCase      =   ids_tensor([self.batch_size]						,    self.type_sequence_label_size		)

									_lowerCAmelCase      =   self.get_config()

									return config, pixel_values, labels




				def 							snake_case       (							self		):



									"""simple docstring"""





									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 							snake_case       (							self						,    _snake_case						,    _snake_case						,    _snake_case		):



									"""simple docstring"""





									_lowerCAmelCase      =   SwinaSRModel(config=_snake_case		)
									model.to(_snake_case		)
									model.eval()
									_lowerCAmelCase      =   model(_snake_case		)

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




				def 							snake_case       (							self						,    _snake_case						,    _snake_case						,    _snake_case		):



									"""simple docstring"""





									_lowerCAmelCase      =   SwinaSRForImageSuperResolution(_snake_case		)
									model.to(_snake_case		)
									model.eval()
									_lowerCAmelCase      =   model(_snake_case		)

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




				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   self.prepare_config_and_inputs()
									_lowerCAmelCase     ,     _lowerCAmelCase     ,     _lowerCAmelCase      =   config_and_inputs
									_lowerCAmelCase      =   {"""pixel_values""": pixel_values}
									return config, inputs_dict





@require_torch
class  __lowerCAmelCase			(					lowerCamelCase__	,						lowerCamelCase__	,						unittest.TestCase	):
				__lowerCamelCase			=    (SwinaSRModel, SwinaSRForImageSuperResolution) if is_torch_available() else ()
				__lowerCamelCase			=    {'''feature-extraction''': SwinaSRModel} if is_torch_available() else {}

				__lowerCamelCase			=    False
				__lowerCamelCase			=    False
				__lowerCamelCase			=    False
				__lowerCamelCase			=    False
				__lowerCamelCase			=    False




				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   SwinaSRModelTester(self		)
									_lowerCAmelCase      =   ConfigTester(self						,    config_class=_snake_case						,    embed_dim=37		)




				def 							snake_case       (							self		):



									"""simple docstring"""





									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 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_model(*_snake_case		)




				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   self.model_tester.prepare_config_and_inputs()
									self.model_tester.create_and_check_for_image_super_resolution(*_snake_case		)




				@unittest.skip(reason="""Got `CUDA error: misaligned address` with PyTorch 2.0.0."""		)
				def 							snake_case       (							self		):



									"""simple docstring"""





									pass




				@unittest.skip(reason="""Swin2SR does not use inputs_embeds"""		)
				def 							snake_case       (							self		):



									"""simple docstring"""





									pass




				@unittest.skip(reason="""Swin2SR does not support training yet"""		)
				def 							snake_case       (							self		):



									"""simple docstring"""





									pass




				@unittest.skip(reason="""Swin2SR does not support training yet"""		)
				def 							snake_case       (							self		):



									"""simple docstring"""





									pass




				def 							snake_case       (							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(_snake_case		)
														self.assertIsInstance(model.get_input_embeddings()						,    (nn.Module)		)
														_lowerCAmelCase      =   model.get_output_embeddings()
														self.assertTrue(x is None or isinstance(_snake_case						,    nn.Linear		)		)




				def 							snake_case       (							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(_snake_case		)
														_lowerCAmelCase      =   inspect.signature(model.forward		)
														# signature.parameters is an OrderedDict => so arg_names order is deterministic
														_lowerCAmelCase      =   [*signature.parameters.keys()]

														_lowerCAmelCase      =   ["""pixel_values"""]
														self.assertListEqual(arg_names[:1]						,    _snake_case		)




				@slow
				def 							snake_case       (							self		):



									"""simple docstring"""





									for model_name in SWIN2SR_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														_lowerCAmelCase      =   SwinaSRModel.from_pretrained(_snake_case		)
														self.assertIsNotNone(_snake_case		)




				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase     ,     _lowerCAmelCase      =   self.model_tester.prepare_config_and_inputs_for_common()

									_lowerCAmelCase      =   _config_zero_init(_snake_case		)
									for model_class in self.all_model_classes:
														_lowerCAmelCase      =   model_class(config=_snake_case		)
														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 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase     ,     _lowerCAmelCase      =   self.model_tester.prepare_config_and_inputs_for_common()
									_lowerCAmelCase      =   True

									for model_class in self.all_model_classes:
														_lowerCAmelCase      =   True
														_lowerCAmelCase      =   False
														_lowerCAmelCase      =   True
														_lowerCAmelCase      =   model_class(_snake_case		)
														model.to(_snake_case		)
														model.eval()
														with torch.no_grad():
																			_lowerCAmelCase      =   model(**self._prepare_for_class(_snake_case						,    _snake_case		)		)
														_lowerCAmelCase      =   outputs.attentions
														_lowerCAmelCase      =   len(self.model_tester.depths		)
														self.assertEqual(len(_snake_case		)						,    _snake_case		)

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

														self.assertListEqual(
														    list(attentions[0].shape[-3:]		)						,    [self.model_tester.num_heads[0], window_size_squared, window_size_squared]						,    )
														_lowerCAmelCase      =   len(_snake_case		)

														# Check attention is always last and order is fine
														_lowerCAmelCase      =   True
														_lowerCAmelCase      =   True
														_lowerCAmelCase      =   model_class(_snake_case		)
														model.to(_snake_case		)
														model.eval()
														with torch.no_grad():
																			_lowerCAmelCase      =   model(**self._prepare_for_class(_snake_case						,    _snake_case		)		)

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

														_lowerCAmelCase      =   outputs.attentions

														self.assertEqual(len(_snake_case		)						,    _snake_case		)

														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  __lowerCAmelCase			(					unittest.TestCase	):




				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   SwinaSRImageProcessor()
									_lowerCAmelCase      =   SwinaSRForImageSuperResolution.from_pretrained("""caidas/swin2SR-classical-sr-x2-64"""		).to(_snake_case		)

									_lowerCAmelCase      =   Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""		)
									_lowerCAmelCase      =   processor(images=_snake_case						,    return_tensors="""pt"""		).to(_snake_case		)

									# forward pass
									with torch.no_grad():
														_lowerCAmelCase      =   model(**_snake_case		)

									# verify the logits
									_lowerCAmelCase      =   torch.Size([1, 3, 976, 1296]		)
									self.assertEqual(outputs.reconstruction.shape						,    _snake_case		)
									_lowerCAmelCase      =   torch.tensor(
									    [[0.5458, 0.5546, 0.5638], [0.5526, 0.5565, 0.5651], [0.5396, 0.5426, 0.5621]]		).to(_snake_case		)
									self.assertTrue(torch.allclose(outputs.reconstruction[0, 0, :3, :3]						,    _snake_case						,    atol=1e-4		)		)








import json
import os
import subprocess
import sys
import warnings
from argparse import ArgumentParser
from contextlib import AbstractContextManager
from typing import Dict, List, Optional

import requests

from ..utils import logging
from . import BaseTransformersCLICommand


A__     				=	logging.get_logger(__name__)  # pylint: disable=invalid-name


A__     				=	"""lfs-multipart-upload"""





class  __lowerCAmelCase			(					lowerCamelCase__	):




				@staticmethod
				def 							snake_case       (							_snake_case		):



									"""simple docstring"""





									_lowerCAmelCase      =   parser.add_parser(
									    """lfs-enable-largefiles"""						,    help=(
									        """Deprecated: use `huggingface-cli` instead. Configure your repository to enable upload of files > 5GB."""
									    )						,    )
									enable_parser.add_argument("""path"""						,    type=_snake_case						,    help="""Local path to repository you want to configure."""		)
									enable_parser.set_defaults(func=lambda _snake_case		: LfsEnableCommand(_snake_case		)		)

									_lowerCAmelCase      =   parser.add_parser(
									    _snake_case						,    help=(
									        """Deprecated: use `huggingface-cli` instead. """
									        """Command will get called by git-lfs, do not call it directly."""
									    )						,    )
									upload_parser.set_defaults(func=lambda _snake_case		: LfsUploadCommand(_snake_case		)		)





class  __lowerCAmelCase			:




				def __init__(							self						,    _snake_case		):



									"""simple docstring"""





									_lowerCAmelCase      =   args




				def 							snake_case       (							self		):



									"""simple docstring"""





									warnings.warn(
									    """Managing repositories through transformers-cli is deprecated. Please use `huggingface-cli` instead."""		)
									_lowerCAmelCase      =   os.path.abspath(self.args.path		)
									if not os.path.isdir(_snake_case		):
														print("""This does not look like a valid git repo."""		)
														exit(1		)
									subprocess.run(
									    """git config lfs.customtransfer.multipart.path transformers-cli""".split()						,    check=_snake_case						,    cwd=_snake_case		)
									subprocess.run(
									    F'git config lfs.customtransfer.multipart.args {LFS_MULTIPART_UPLOAD_COMMAND}'.split()						,    check=_snake_case						,    cwd=_snake_case						,    )
									print("""Local repo set up for largefiles"""		)


def 		_UpperCAmelCase    (				snake_case      ):





					"""simple docstring"""

					_lowerCAmelCase      =   json.dumps(snake_case      ) + """\n"""
					sys.stdout.write(snake_case      )
					sys.stdout.flush()


def 		_UpperCAmelCase    (				):





					"""simple docstring"""

					_lowerCAmelCase      =   json.loads(sys.stdin.readline().strip()      )

					if "terminate" in (msg.get("""type"""      ), msg.get("""event"""      )):
										# terminate message received
										return None

					if msg.get("""event"""      ) not in ("download", "upload"):
										logger.critical("""Received unexpected message"""      )
										sys.exit(1      )

					return msg





class  __lowerCAmelCase			(					lowerCamelCase__	):




				def __init__(							self						,    _snake_case						,    _snake_case						,    _snake_case		):



									"""simple docstring"""





									_lowerCAmelCase      =   filepath
									_lowerCAmelCase      =   seek_from
									_lowerCAmelCase      =   read_limit
									_lowerCAmelCase      =   0




				def __enter__(							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   open(self.filepath						,    """rb"""		)
									self.f.seek(self.seek_from		)
									return self




				def __len__(							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   os.fstat(self.f.fileno()		).st_size
									return min(self.read_limit						,    total_length - self.seek_from		)




				def 							snake_case       (							self						,    _snake_case=-1		):



									"""simple docstring"""





									if self.n_seen >= self.read_limit:
														return b""
									_lowerCAmelCase      =   self.read_limit - self.n_seen
									_lowerCAmelCase      =   self.f.read(remaining_amount if n < 0 else min(_snake_case						,    _snake_case		)		)
									self.n_seen += len(_snake_case		)
									return data




				def __iter__(							self		):



									"""simple docstring"""





									yield self.read(n=4 * 1024 * 1024		)




				def __exit__(							self						,    *_snake_case		):



									"""simple docstring"""





									self.f.close()





class  __lowerCAmelCase			:




				def __init__(							self						,    _snake_case		):



									"""simple docstring"""





									_lowerCAmelCase      =   args




				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   json.loads(sys.stdin.readline().strip()		)
									if not (init_msg.get("""event"""		) == "init" and init_msg.get("""operation"""		) == "upload"):
														write_msg({"""error""": {"""code""": 32, """message""": """Wrong lfs init operation"""}}		)
														sys.exit(1		)

									# The transfer process should use the information it needs from the
									# initiation structure, and also perform any one-off setup tasks it
									# needs to do. It should then respond on stdout with a simple empty
									# confirmation structure, as follows:
									write_msg({}		)

									# After the initiation exchange, git-lfs will send any number of
									# transfer requests to the stdin of the transfer process, in a serial sequence.
									while True:
														_lowerCAmelCase      =   read_msg()
														if msg is None:
																			# When all transfers have been processed, git-lfs will send
																			# a terminate event to the stdin of the transfer process.
																			# On receiving this message the transfer process should
																			# clean up and terminate. No response is expected.
																			sys.exit(0		)

														_lowerCAmelCase      =   msg["""oid"""]
														_lowerCAmelCase      =   msg["""path"""]
														_lowerCAmelCase      =   msg["""action"""]["""href"""]
														_lowerCAmelCase      =   msg["""action"""]["""header"""]
														_lowerCAmelCase      =   int(header.pop("""chunk_size"""		)		)
														_lowerCAmelCase      =   list(header.values()		)

														_lowerCAmelCase      =   []
														for i, presigned_url in enumerate(_snake_case		):
																			with FileSlice(_snake_case						,    seek_from=i * chunk_size						,    read_limit=_snake_case		) as data:
																								_lowerCAmelCase      =   requests.put(_snake_case						,    data=_snake_case		)
																								r.raise_for_status()
																								parts.append(
																								    {
																								        """etag""": r.headers.get("""etag"""		),
																								        """partNumber""": i + 1,
																								    }		)
																								# In order to support progress reporting while data is uploading / downloading,
																								# the transfer process should post messages to stdout
																								write_msg(
																								    {
																								        """event""": """progress""",
																								        """oid""": oid,
																								        """bytesSoFar""": (i + 1) * chunk_size,
																								        """bytesSinceLast""": chunk_size,
																								    }		)
																								# Not precise but that's ok.

														_lowerCAmelCase      =   requests.post(
														    _snake_case						,    json={
														        """oid""": oid,
														        """parts""": parts,
														    }						,    )
														r.raise_for_status()

														write_msg({"""event""": """complete""", """oid""": oid}		)



