






'''simple docstring'''







import inspect
import unittest

import requests

from transformers import MgpstrConfig
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
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
			import torch
			from torch import nn

			from transformers import MgpstrForSceneTextRecognition


if is_vision_available():
			from PIL import Image

			from transformers import MgpstrProcessor

class _lowercase      :




							def __init__(				self:							Optional[Any]     ,				UpperCamelCase__:							str     ,				UpperCamelCase__:							str=False     ,				UpperCamelCase__:							List[Any]=13     ,				UpperCamelCase__:							List[Any]=(32, 128)     ,				UpperCamelCase__:							List[Any]=4     ,				UpperCamelCase__:							List[str]=3     ,				UpperCamelCase__:							Dict=27     ,				UpperCamelCase__:							List[str]=38     ,				UpperCamelCase__:							int=99     ,				UpperCamelCase__:							Optional[int]=99     ,				UpperCamelCase__:							List[Any]=32     ,				UpperCamelCase__:							Union[str, Any]=5     ,				UpperCamelCase__:							Union[str, Any]=4     ,				UpperCamelCase__:							Optional[Any]=4.0     ,				UpperCamelCase__:							List[str]=257     ,				UpperCamelCase__:							Optional[int]=None     ,				):
														lowerCamelCase__   :      str			    =			parent
														lowerCamelCase__   :      List[Any]			    =			is_training
														lowerCamelCase__   :      List[Any]			    =			batch_size
														lowerCamelCase__   :      Optional[int]			    =			image_size
														lowerCamelCase__   :      Optional[Any]			    =			patch_size
														lowerCamelCase__   :      Optional[Any]			    =			num_channels
														lowerCamelCase__   :      str			    =			max_token_length
														lowerCamelCase__   :      int			    =			num_character_labels
														lowerCamelCase__   :      Union[str, Any]			    =			num_bpe_labels
														lowerCamelCase__   :      Any			    =			num_wordpiece_labels
														lowerCamelCase__   :      Optional[Any]			    =			hidden_size
														lowerCamelCase__   :      int			    =			num_hidden_layers
														lowerCamelCase__   :      Dict			    =			num_attention_heads
														lowerCamelCase__   :      List[Any]			    =			mlp_ratio
														lowerCamelCase__   :      str			    =			patch_embeds_hidden_size
														lowerCamelCase__   :      Dict			    =			output_hidden_states




							def 						lowerCamelCase_							(				self:							Optional[int]	):
														lowerCamelCase__   :      List[Any]			    =			floats_tensor([self.batch_size, self.num_channels, self.image_size[0], self.image_size[1]]	)
														lowerCamelCase__   :      List[Any]			    =			self.get_config()
														return config, pixel_values




							def 						lowerCamelCase_							(				self:							List[str]	):
														return MgpstrConfig(
														    image_size=self.image_size     ,				patch_size=self.patch_size     ,				num_channels=self.num_channels     ,				max_token_length=self.max_token_length     ,				num_character_labels=self.num_character_labels     ,				num_bpe_labels=self.num_bpe_labels     ,				num_wordpiece_labels=self.num_wordpiece_labels     ,				hidden_size=self.hidden_size     ,				num_hidden_layers=self.num_hidden_layers     ,				num_attention_heads=self.num_attention_heads     ,				mlp_ratio=self.mlp_ratio     ,				output_hidden_states=self.output_hidden_states     ,				)




							def 						lowerCamelCase_							(				self:							Any     ,				UpperCamelCase__:							List[str]     ,				UpperCamelCase__:							Dict	):
														lowerCamelCase__   :      Any			    =			MgpstrForSceneTextRecognition(UpperCamelCase__	)
														model.to(UpperCamelCase__	)
														model.eval()
														with torch.no_grad():
																					lowerCamelCase__   :      Any			    =			model(UpperCamelCase__	)
														self.parent.assertEqual(
														    generated_ids[0][0].shape     ,				(self.batch_size, self.max_token_length, self.num_character_labels)	)




							def 						lowerCamelCase_							(				self:							List[Any]	):
														lowerCamelCase__   :      Union[str, Any]			    =			self.prepare_config_and_inputs()
														lowerCamelCase__    ,						lowerCamelCase__   :      Union[str, Any]			    =			config_and_inputs
														lowerCamelCase__   :      Optional[Any]			    =			{"""pixel_values""": pixel_values}
														return config, inputs_dict



@require_torch
class _lowercase      (							_lowercase							, _lowercase							, unittest.TestCase						):
							a							 =			(MgpstrForSceneTextRecognition,) if is_torch_available() else ()
							a							 =			{"""feature-extraction""": MgpstrForSceneTextRecognition} if is_torch_available() else {}
							a							 =			False

							a							 =			False
							a							 =			False
							a							 =			False
							a							 =			False




							def 						lowerCamelCase_							(				self:							List[Any]	):
														lowerCamelCase__   :      Tuple			    =			MgpstrModelTester(self	)
														lowerCamelCase__   :      int			    =			ConfigTester(self     ,				config_class=UpperCamelCase__     ,				has_text_modality=UpperCamelCase__	)




							def 						lowerCamelCase_							(				self:							Tuple	):
														self.config_tester.run_common_tests()




							def 						lowerCamelCase_							(				self:							int	):
														lowerCamelCase__   :      List[Any]			    =			self.model_tester.prepare_config_and_inputs()
														self.model_tester.create_and_check_model(*UpperCamelCase__	)




							@unittest.skip(reason="""MgpstrModel does not use inputs_embeds"""	)
							def 						lowerCamelCase_							(				self:							int	):
														pass




							def 						lowerCamelCase_							(				self:							Optional[int]	):
														lowerCamelCase__    ,						lowerCamelCase__   :      Optional[Any]			    =			self.model_tester.prepare_config_and_inputs_for_common()

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




							def 						lowerCamelCase_							(				self:							List[str]	):
														lowerCamelCase__    ,						lowerCamelCase__   :      Tuple			    =			self.model_tester.prepare_config_and_inputs_for_common()

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

																					lowerCamelCase__   :      Optional[Any]			    =			["""pixel_values"""]
																					self.assertListEqual(arg_names[:1]     ,				UpperCamelCase__	)




							@unittest.skip(reason="""MgpstrModel does not support feedforward chunking"""	)
							def 						lowerCamelCase_							(				self:							List[Any]	):
														pass




							def 						lowerCamelCase_							(				self:							int	):
														lowerCamelCase__    ,						lowerCamelCase__   :      int			    =			self.model_tester.prepare_config_and_inputs_for_common()

														for model_class in self.all_model_classes:
																					if not model_class.supports_gradient_checkpointing:
																												continue

																					lowerCamelCase__   :      Optional[int]			    =			True
																					lowerCamelCase__   :      Tuple			    =			model_class(UpperCamelCase__	)
																					self.assertTrue(model.is_gradient_checkpointing	)




							def 						lowerCamelCase_							(				self:							Tuple	):
														def check_hidden_states_output(UpperCamelCase__:							Optional[int]     ,				UpperCamelCase__:							Any     ,				UpperCamelCase__:							Any	):
																					lowerCamelCase__   :      List[str]			    =			model_class(UpperCamelCase__	)
																					model.to(UpperCamelCase__	)
																					model.eval()

																					with torch.no_grad():
																												lowerCamelCase__   :      Any			    =			model(**self._prepare_for_class(UpperCamelCase__     ,				UpperCamelCase__	)	)

																					lowerCamelCase__   :      List[str]			    =			outputs.hidden_states

																					lowerCamelCase__   :      Any			    =			getattr(
																					    self.model_tester     ,				"""expected_num_hidden_layers"""     ,				self.model_tester.num_hidden_layers + 1	)
																					self.assertEqual(len(UpperCamelCase__	)     ,				UpperCamelCase__	)

																					self.assertListEqual(
																					    list(hidden_states[0].shape[-2:]	)     ,				[self.model_tester.patch_embeds_hidden_size, self.model_tester.hidden_size]     ,				)

														lowerCamelCase__    ,						lowerCamelCase__   :      Dict			    =			self.model_tester.prepare_config_and_inputs_for_common()

														for model_class in self.all_model_classes:
																					lowerCamelCase__   :      Dict			    =			True
																					check_hidden_states_output(UpperCamelCase__     ,				UpperCamelCase__     ,				UpperCamelCase__	)

																					# check that output_hidden_states also work using config
																					del inputs_dict["output_hidden_states"]
																					lowerCamelCase__   :      Union[str, Any]			    =			True

																					check_hidden_states_output(UpperCamelCase__     ,				UpperCamelCase__     ,				UpperCamelCase__	)




							def 						lowerCamelCase_							(				self:							List[str]	):
														lowerCamelCase__    ,						lowerCamelCase__   :      List[Any]			    =			self.model_tester.prepare_config_and_inputs_for_common()

														lowerCamelCase__   :      Dict			    =			_config_zero_init(UpperCamelCase__	)
														for model_class in self.all_model_classes:
																					lowerCamelCase__   :      Any			    =			model_class(config=UpperCamelCase__	)
																					for name, param in model.named_parameters():
																												if isinstance(UpperCamelCase__     ,				(nn.Linear, nn.Convad, nn.LayerNorm)	):
																																			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'''     ,				)




							@unittest.skip(reason="""Retain_grad is tested in individual model tests"""	)
							def 						lowerCamelCase_							(				self:							Optional[int]	):
														pass





def 					SCREAMING_SNAKE_CASE_			() ->			List[str]:
							lowerCamelCase__   :      Any			    =			"""https://i.postimg.cc/ZKwLg2Gw/367-14.png"""
							lowerCamelCase__   :      Any			    =			Image.open(requests.get(UpperCamelCase							,     stream=UpperCamelCase ).raw ).convert("""RGB""" )
							return im



@require_vision
@require_torch
class _lowercase      (							unittest.TestCase						):




							@slow
							def 						lowerCamelCase_							(				self:							Dict	):
														lowerCamelCase__   :      int			    =			"""alibaba-damo/mgp-str-base"""
														lowerCamelCase__   :      Dict			    =			MgpstrForSceneTextRecognition.from_pretrained(UpperCamelCase__	).to(UpperCamelCase__	)
														lowerCamelCase__   :      Any			    =			MgpstrProcessor.from_pretrained(UpperCamelCase__	)

														lowerCamelCase__   :      Tuple			    =			prepare_img()
														lowerCamelCase__   :      str			    =			processor(images=UpperCamelCase__     ,				return_tensors="""pt"""	).pixel_values.to(UpperCamelCase__	)

														# forward pass
														with torch.no_grad():
																					lowerCamelCase__   :      Any			    =			model(UpperCamelCase__	)

														# verify the logits
														self.assertEqual(outputs.logits[0].shape     ,				torch.Size((1, 27, 38)	)	)

														lowerCamelCase__   :      Union[str, Any]			    =			processor.batch_decode(outputs.logits	)
														lowerCamelCase__   :      str			    =			"""ticket"""

														self.assertEqual(out_strs["""generated_text"""][0]     ,				UpperCamelCase__	)

														lowerCamelCase__   :      Dict			    =			torch.tensor(
														    [[[-39.5_397, -44.4_024, -36.1_844], [-61.4_709, -63.8_639, -58.3_454], [-74.0_225, -68.5_494, -71.2_164]]]     ,				device=UpperCamelCase__     ,				)

														self.assertTrue(torch.allclose(outputs.logits[0][:, 1:4, 1:4]     ,				UpperCamelCase__     ,				atol=1e-4	)	)









'''simple docstring'''







import csv
import os

import pytest

from datasets import Dataset, DatasetDict, Features, NamedSplit, Value
from datasets.io.csv import CsvDatasetReader, CsvDatasetWriter

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases





def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase							,     UpperCamelCase ) ->			Dict:
							assert isinstance(UpperCamelCase							,     UpperCamelCase )
							assert dataset.num_rows == 4
							assert dataset.num_columns == 3
							assert dataset.column_names == ["col_1", "col_2", "col_3"]
							for feature, expected_dtype in expected_features.items():
														assert dataset.features[feature].dtype == expected_dtype





@pytest.mark.parametrize("""keep_in_memory"""							,     [False, True] )
def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase							,     UpperCamelCase							,     UpperCamelCase ) ->			Any:
							lowerCamelCase__   :      Any			    =			tmp_path / """cache"""
							lowerCamelCase__   :      int			    =			{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
							with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
														lowerCamelCase__   :      Any			    =			CsvDatasetReader(UpperCamelCase							,     cache_dir=UpperCamelCase							,     keep_in_memory=UpperCamelCase ).read()
							_check_csv_dataset(UpperCamelCase							,     UpperCamelCase )





@pytest.mark.parametrize(
    """features"""							,     [
        None,
        {"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""},
        {"""col_1""": """string""", """col_2""": """string""", """col_3""": """string"""},
        {"""col_1""": """int32""", """col_2""": """int32""", """col_3""": """int32"""},
        {"""col_1""": """float32""", """col_2""": """float32""", """col_3""": """float32"""},
    ]							,     )
def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase							,     UpperCamelCase							,     UpperCamelCase ) ->			Any:
							lowerCamelCase__   :      Optional[int]			    =			tmp_path / """cache"""
							# CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
							lowerCamelCase__   :      int			    =			{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
							lowerCamelCase__   :      int			    =			features.copy() if features else default_expected_features
							lowerCamelCase__   :      List[Any]			    =			(
							    Features({feature: Value(UpperCamelCase ) for feature, dtype in features.items()} ) if features is not None else None
							)
							lowerCamelCase__   :      Union[str, Any]			    =			CsvDatasetReader(UpperCamelCase							,     features=UpperCamelCase							,     cache_dir=UpperCamelCase ).read()
							_check_csv_dataset(UpperCamelCase							,     UpperCamelCase )





@pytest.mark.parametrize("""split"""							,     [None, NamedSplit("""train""" ), """train""", """test"""] )
def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase							,     UpperCamelCase							,     UpperCamelCase ) ->			Union[str, Any]:
							lowerCamelCase__   :      Any			    =			tmp_path / """cache"""
							lowerCamelCase__   :      Union[str, Any]			    =			{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
							lowerCamelCase__   :      List[Any]			    =			CsvDatasetReader(UpperCamelCase							,     cache_dir=UpperCamelCase							,     split=UpperCamelCase ).read()
							_check_csv_dataset(UpperCamelCase							,     UpperCamelCase )
							assert dataset.split == split if split else "train"





@pytest.mark.parametrize("""path_type"""							,     [str, list] )
def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase							,     UpperCamelCase							,     UpperCamelCase ) ->			Union[str, Any]:
							if issubclass(UpperCamelCase							,     UpperCamelCase ):
														lowerCamelCase__   :      int			    =			csv_path
							elif issubclass(UpperCamelCase							,     UpperCamelCase ):
														lowerCamelCase__   :      Tuple			    =			[csv_path]
							lowerCamelCase__   :      Optional[int]			    =			tmp_path / """cache"""
							lowerCamelCase__   :      str			    =			{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
							lowerCamelCase__   :      Dict			    =			CsvDatasetReader(UpperCamelCase							,     cache_dir=UpperCamelCase ).read()
							_check_csv_dataset(UpperCamelCase							,     UpperCamelCase )





def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase							,     UpperCamelCase							,     UpperCamelCase=("train",) ) ->			int:
							assert isinstance(UpperCamelCase							,     UpperCamelCase )
							for split in splits:
														lowerCamelCase__   :      Optional[Any]			    =			dataset_dict[split]
														assert dataset.num_rows == 4
														assert dataset.num_columns == 3
														assert dataset.column_names == ["col_1", "col_2", "col_3"]
														for feature, expected_dtype in expected_features.items():
																					assert dataset.features[feature].dtype == expected_dtype





@pytest.mark.parametrize("""keep_in_memory"""							,     [False, True] )
def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase							,     UpperCamelCase							,     UpperCamelCase ) ->			Optional[Any]:
							lowerCamelCase__   :      List[Any]			    =			tmp_path / """cache"""
							lowerCamelCase__   :      Dict			    =			{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
							with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
														lowerCamelCase__   :      Any			    =			CsvDatasetReader({"""train""": csv_path}							,     cache_dir=UpperCamelCase							,     keep_in_memory=UpperCamelCase ).read()
							_check_csv_datasetdict(UpperCamelCase							,     UpperCamelCase )





@pytest.mark.parametrize(
    """features"""							,     [
        None,
        {"""col_1""": """string""", """col_2""": """int64""", """col_3""": """float64"""},
        {"""col_1""": """string""", """col_2""": """string""", """col_3""": """string"""},
        {"""col_1""": """int32""", """col_2""": """int32""", """col_3""": """int32"""},
        {"""col_1""": """float32""", """col_2""": """float32""", """col_3""": """float32"""},
    ]							,     )
def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase							,     UpperCamelCase							,     UpperCamelCase ) ->			int:
							lowerCamelCase__   :      int			    =			tmp_path / """cache"""
							# CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
							lowerCamelCase__   :      Optional[Any]			    =			{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
							lowerCamelCase__   :      List[str]			    =			features.copy() if features else default_expected_features
							lowerCamelCase__   :      Tuple			    =			(
							    Features({feature: Value(UpperCamelCase ) for feature, dtype in features.items()} ) if features is not None else None
							)
							lowerCamelCase__   :      str			    =			CsvDatasetReader({"""train""": csv_path}							,     features=UpperCamelCase							,     cache_dir=UpperCamelCase ).read()
							_check_csv_datasetdict(UpperCamelCase							,     UpperCamelCase )





@pytest.mark.parametrize("""split"""							,     [None, NamedSplit("""train""" ), """train""", """test"""] )
def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase							,     UpperCamelCase							,     UpperCamelCase ) ->			Any:
							if split:
														lowerCamelCase__   :      Tuple			    =			{split: csv_path}
							else:
														lowerCamelCase__   :      Union[str, Any]			    =			{"""train""": csv_path, """test""": csv_path}
							lowerCamelCase__   :      Dict			    =			tmp_path / """cache"""
							lowerCamelCase__   :      Union[str, Any]			    =			{"""col_1""": """int64""", """col_2""": """int64""", """col_3""": """float64"""}
							lowerCamelCase__   :      Tuple			    =			CsvDatasetReader(UpperCamelCase							,     cache_dir=UpperCamelCase ).read()
							_check_csv_datasetdict(UpperCamelCase							,     UpperCamelCase							,     splits=list(path.keys() ) )
							assert all(dataset[split].split == split for split in path.keys() )





def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase ) ->			Tuple:
							with open(UpperCamelCase							,     encoding="""utf-8""" ) as csvfile:
														yield from csv.reader(UpperCamelCase )





def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase							,     UpperCamelCase ) ->			Optional[int]:
							lowerCamelCase__   :      Union[str, Any]			    =			tmp_path / """cache"""
							lowerCamelCase__   :      List[str]			    =			os.path.join(UpperCamelCase							,     """tmp.csv""" )
							lowerCamelCase__   :      Optional[Any]			    =			CsvDatasetReader({"""train""": csv_path}							,     cache_dir=UpperCamelCase ).read()
							CsvDatasetWriter(dataset["""train"""]							,     UpperCamelCase							,     num_proc=1 ).write()

							lowerCamelCase__   :      Optional[Any]			    =			iter_csv_file(UpperCamelCase )
							lowerCamelCase__   :      str			    =			iter_csv_file(UpperCamelCase )

							for rowa, rowa in zip(UpperCamelCase							,     UpperCamelCase ):
														assert rowa == rowa





def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase							,     UpperCamelCase ) ->			List[str]:
							lowerCamelCase__   :      Any			    =			tmp_path / """cache"""
							lowerCamelCase__   :      List[str]			    =			os.path.join(UpperCamelCase							,     """tmp.csv""" )
							lowerCamelCase__   :      Optional[int]			    =			CsvDatasetReader({"""train""": csv_path}							,     cache_dir=UpperCamelCase ).read()
							CsvDatasetWriter(dataset["""train"""]							,     UpperCamelCase							,     num_proc=2 ).write()

							lowerCamelCase__   :      Tuple			    =			iter_csv_file(UpperCamelCase )
							lowerCamelCase__   :      Union[str, Any]			    =			iter_csv_file(UpperCamelCase )

							for rowa, rowa in zip(UpperCamelCase							,     UpperCamelCase ):
														assert rowa == rowa





def 					SCREAMING_SNAKE_CASE_			(UpperCamelCase							,     UpperCamelCase ) ->			Optional[int]:
							lowerCamelCase__   :      List[Any]			    =			tmp_path / """cache"""
							lowerCamelCase__   :      List[Any]			    =			os.path.join(UpperCamelCase							,     """tmp.csv""" )
							lowerCamelCase__   :      Dict			    =			CsvDatasetReader({"""train""": csv_path}							,     cache_dir=UpperCamelCase ).read()
							with pytest.raises(UpperCamelCase ):
														CsvDatasetWriter(dataset["""train"""]							,     UpperCamelCase							,     num_proc=0 )

