

import csv
import json
import logging
import os
import re
import subprocess
import tempfile
import unittest
import zipfile
from pathlib import Path
from typing import Optional
from unittest import mock

import torch

# We use TF to parse the logs
from accelerate import Accelerator
from accelerate.test_utils.testing import (
    MockingTestCase,
    TempDirTestCase,
    require_comet_ml,
    require_tensorboard,
    require_wandb,
    skip,
)
from accelerate.tracking import CometMLTracker, GeneralTracker
from accelerate.utils import ProjectConfiguration, is_comet_ml_available, is_tensorboard_available


if is_comet_ml_available():
  from comet_ml import OfflineExperiment

if is_tensorboard_available():
  import struct

  import tensorboard.compat.proto.event_pba as event_pba

A__										=      logging.getLogger(__name__)







@require_tensorboard
class  UpperCamelCase_      (	unittest.TestCase      ):


  """simple docstring"""



  def   _UpperCamelCase     (						self     :					str						)       ->			int:
      SCREAMING_SNAKE_CASE__         =    "test_project_with_config"
      with tempfile.TemporaryDirectory() as dirpath:
          SCREAMING_SNAKE_CASE__         =    Accelerator(log_with="tensorboard"  ,	project_dir=A_						)
          SCREAMING_SNAKE_CASE__         =    {"num_iterations": 12, "learning_rate": 1E-2, "some_boolean": False, "some_string": "some_value"}
          accelerator.init_trackers(A_  ,	A_						)
          accelerator.end_training()
          for child in Path(f"{dirpath}/{project_name}"						).glob("*/**"						):
              SCREAMING_SNAKE_CASE__         =    list(filter(lambda A_						: x.is_file()  ,	child.iterdir()						)						)[0]
          self.assertNotEqual(str(A_						)  ,	""						)



  def   _UpperCamelCase     (						self     :					Tuple						)       ->			Dict:
      SCREAMING_SNAKE_CASE__         =    "test_project_with_log"
      with tempfile.TemporaryDirectory() as dirpath:
          SCREAMING_SNAKE_CASE__         =    Accelerator(log_with="tensorboard"  ,	project_dir=A_						)
          accelerator.init_trackers(A_						)
          SCREAMING_SNAKE_CASE__         =    {"total_loss": 0.1, "iteration": 1, "my_text": "some_value"}
          accelerator.log(A_  ,	step=0						)
          accelerator.end_training()
          # Logged values are stored in the outermost-tfevents file and can be read in as a TFRecord
          # Names are randomly generated each time
          SCREAMING_SNAKE_CASE__         =    list(filter(lambda A_						: x.is_file()  ,	Path(f"{dirpath}/{project_name}"						).iterdir()						)						)[0]
          self.assertNotEqual(str(A_						)  ,	""						)



  def   _UpperCamelCase     (						self     :					Union[str, Any]						)       ->			Dict:
      SCREAMING_SNAKE_CASE__         =    "test_project_with_log"
      with tempfile.TemporaryDirectory() as dirpath:
          SCREAMING_SNAKE_CASE__         =    Accelerator(log_with="tensorboard"  ,	project_dir=A_						)
          accelerator.init_trackers(A_						)
          SCREAMING_SNAKE_CASE__         =    {"tensor": torch.tensor(1						)}
          accelerator.log(A_  ,	step=0						)
          accelerator.end_training()
          # Logged values are stored in the outermost-tfevents file and can be read in as a TFRecord
          # Names are randomly generated each time
          SCREAMING_SNAKE_CASE__         =    list(filter(lambda A_						: x.is_file()  ,	Path(f"{dirpath}/{project_name}"						).iterdir()						)						)[0]
          # Reading implementation based on https://github.com/pytorch/pytorch/issues/45327#issuecomment-703757685
          with open(A_  ,	"rb"						) as f:
              SCREAMING_SNAKE_CASE__         =    f.read()
          SCREAMING_SNAKE_CASE__         =    False
          while data:
              SCREAMING_SNAKE_CASE__         =    struct.unpack("Q"  ,	data[:8]						)

              SCREAMING_SNAKE_CASE__         =    data[12 : 12 + int(header[0]						)]  # 8+4
              SCREAMING_SNAKE_CASE__         =    data[12 + int(header[0]						) + 4 :]
              SCREAMING_SNAKE_CASE__         =    event_pba.Event()

              event.ParseFromString(A_						)
              if event.HasField("summary"						):
                  for value in event.summary.value:
                      if value.simple_value == 1.0 and value.tag == "tensor":
                          SCREAMING_SNAKE_CASE__         =    True
          self.assertTrue(A_  ,	"Converted tensor was not found in the log file!"						)



  def   _UpperCamelCase     (						self     :					Union[str, Any]						)       ->			Dict:
      with self.assertRaisesRegex(A_  ,	"Logging with `tensorboard` requires a `logging_dir`"						):
          SCREAMING_SNAKE_CASE__         =    Accelerator(log_with="tensorboard"						)
      with tempfile.TemporaryDirectory() as dirpath:
          SCREAMING_SNAKE_CASE__         =    Accelerator(log_with="tensorboard"  ,	project_dir=A_						)







  def   _UpperCamelCase     (						self     :					Any						)       ->			List[Any]:
      SCREAMING_SNAKE_CASE__         =    ProjectConfiguration(total_limit=30						)
      with tempfile.TemporaryDirectory() as dirpath:
          SCREAMING_SNAKE_CASE__         =    Accelerator(log_with="tensorboard"  ,	project_dir=A_  ,	project_config=A_						)







@require_wandb
@mock.patch.dict(os.environ		,  {"""WANDB_MODE""": """offline"""}      )
class  UpperCamelCase_      (	_snake_case		,  _snake_case      ):


  """simple docstring"""



  def   _UpperCamelCase     (						self     :					Dict						)       ->			Optional[Any]:
      super().setUp()
      # wandb let's us override where logs are stored to via the WANDB_DIR env var
      self.add_mocks(mock.patch.dict(os.environ  ,	{"WANDB_DIR": self.tmpdir}						)						)



  @staticmethod
  def   _UpperCamelCase     (						A_     :					str  ,	A_     :					str  ,	A_     :					bool = True						)       ->			int:
      SCREAMING_SNAKE_CASE__         =    rf"{section} ([\S\s]*?)\n\n"
      if record:
          SCREAMING_SNAKE_CASE__         =    rf"Record: {pattern}"
      SCREAMING_SNAKE_CASE__         =    re.findall(A_  ,	A_						)[0]
      # A config
      if section == "config" or section == "history":
          SCREAMING_SNAKE_CASE__         =    re.findall(r"\"([a-zA-Z0-9_.,]+)"  ,	A_						)
          return {key: val for key, val in zip(cleaned_record[0::2]  ,	cleaned_record[1::2]						)}
      # Everything else
      else:
          return dict(re.findall(r"(\w+): \"([^\s]+)\""  ,	A_						)						)







  @skip
  def   _UpperCamelCase     (						self     :					str						)       ->			Any:
      SCREAMING_SNAKE_CASE__         =    "test_project_with_config"
      SCREAMING_SNAKE_CASE__         =    Accelerator(log_with="wandb"						)
      SCREAMING_SNAKE_CASE__         =    {"num_iterations": 12, "learning_rate": 1E-2, "some_boolean": False, "some_string": "some_value"}
      SCREAMING_SNAKE_CASE__         =    {"wandb": {"tags": ["my_tag"]}}
      accelerator.init_trackers(A_  ,	A_  ,	A_						)
      SCREAMING_SNAKE_CASE__         =    {"total_loss": 0.1, "iteration": 1, "my_text": "some_value"}
      accelerator.log(A_  ,	step=0						)
      accelerator.end_training()
      # The latest offline log is stored at wandb/latest-run/*.wandb
      for child in Path(f"{self.tmpdir}/wandb/latest-run"						).glob("*"						):
          if child.is_file() and child.suffix == ".wandb":
              SCREAMING_SNAKE_CASE__         =    subprocess.check_output(
                  ["wandb", "sync", "--view", "--verbose", str(A_						)]  ,	env=os.environ.copy()						).decode("utf8"  ,	"ignore"						)
              break

        # Check HPS through careful parsing and cleaning
      SCREAMING_SNAKE_CASE__         =    self.parse_log(A_  ,	"config"						)
      self.assertEqual(logged_items["num_iterations"]  ,	"12"						)
      self.assertEqual(logged_items["learning_rate"]  ,	"0.01"						)
      self.assertEqual(logged_items["some_boolean"]  ,	"false"						)
      self.assertEqual(logged_items["some_string"]  ,	"some_value"						)
      self.assertEqual(logged_items["some_string"]  ,	"some_value"						)

      # Run tags
      SCREAMING_SNAKE_CASE__         =    self.parse_log(A_  ,	"run"  ,	A_						)
      self.assertEqual(logged_items["tags"]  ,	"my_tag"						)

      # Actual logging
      SCREAMING_SNAKE_CASE__         =    self.parse_log(A_  ,	"history"						)
      self.assertEqual(logged_items["total_loss"]  ,	"0.1"						)
      self.assertEqual(logged_items["iteration"]  ,	"1"						)
      self.assertEqual(logged_items["my_text"]  ,	"some_value"						)
      self.assertEqual(logged_items["_step"]  ,	"0"						)





def 				__a							(       self			,							__A			,							__A						):
    SCREAMING_SNAKE_CASE__         =    run_name
    SCREAMING_SNAKE_CASE__         =    OfflineExperiment(project_name=__A			,							offline_directory=__A						)
    logger.info(f"Initialized offline CometML project {self.run_name}"						)
    logger.info("Make sure to log any initial configurations with `self.store_init_configuration` before training!"						)







@require_comet_ml
@mock.patch.object(_snake_case		,  """__init__"""		,  _snake_case      )
class  UpperCamelCase_      (	unittest.TestCase      ):


  """simple docstring"""



  @staticmethod
  def   _UpperCamelCase     (						A_     :					Optional[Any]  ,	A_     :					str  ,	A_     :					bool = False						)       ->			List[str]:
      for log in log_list:
          SCREAMING_SNAKE_CASE__         =    json.loads(A_						)["payload"]
          if is_param and "param" in j.keys():
              if j["param"]["paramName"] == key:
                  return j["param"]["paramValue"]
          if "log_other" in j.keys():
              if j["log_other"]["key"] == key:
                  return j["log_other"]["val"]
          if "metric" in j.keys():
              if j["metric"]["metricName"] == key:
                  return j["metric"]["metricValue"]



  def   _UpperCamelCase     (						self     :					Union[str, Any]						)       ->			str:
      with tempfile.TemporaryDirectory() as d:
          SCREAMING_SNAKE_CASE__         =    CometMLTracker("test_project_with_config"  ,	A_						)
          SCREAMING_SNAKE_CASE__         =    Accelerator(log_with=A_						)
          SCREAMING_SNAKE_CASE__         =    {"num_iterations": 12, "learning_rate": 1E-2, "some_boolean": False, "some_string": "some_value"}
          accelerator.init_trackers(A_  ,	A_						)
          accelerator.end_training()
          SCREAMING_SNAKE_CASE__         =    os.listdir(A_						)[0]  # Comet is nice, it's just a zip file here
          # We parse the raw logs
          SCREAMING_SNAKE_CASE__         =    os.path.join(A_  ,	A_						)
          SCREAMING_SNAKE_CASE__         =    zipfile.ZipFile(A_  ,	"r"						)
          SCREAMING_SNAKE_CASE__         =    archive.open("messages.json"						).read().decode("utf-8"						)
      SCREAMING_SNAKE_CASE__         =    log.split("\n"						)[:-1]
      self.assertEqual(self.get_value_from_key(A_  ,	"num_iterations"  ,	A_						)  ,	12						)
      self.assertEqual(self.get_value_from_key(A_  ,	"learning_rate"  ,	A_						)  ,	0.01						)
      self.assertEqual(self.get_value_from_key(A_  ,	"some_boolean"  ,	A_						)  ,	A_						)
      self.assertEqual(self.get_value_from_key(A_  ,	"some_string"  ,	A_						)  ,	"some_value"						)







  def   _UpperCamelCase     (						self     :					int						)       ->			str:
      with tempfile.TemporaryDirectory() as d:
          SCREAMING_SNAKE_CASE__         =    CometMLTracker("test_project_with_config"  ,	A_						)
          SCREAMING_SNAKE_CASE__         =    Accelerator(log_with=A_						)
          accelerator.init_trackers(A_						)
          SCREAMING_SNAKE_CASE__         =    {"total_loss": 0.1, "iteration": 1, "my_text": "some_value"}
          accelerator.log(A_  ,	step=0						)
          accelerator.end_training()
          SCREAMING_SNAKE_CASE__         =    os.listdir(A_						)[0]  # Comet is nice, it's just a zip file here
          # We parse the raw logs
          SCREAMING_SNAKE_CASE__         =    os.path.join(A_  ,	A_						)
          SCREAMING_SNAKE_CASE__         =    zipfile.ZipFile(A_  ,	"r"						)
          SCREAMING_SNAKE_CASE__         =    archive.open("messages.json"						).read().decode("utf-8"						)
      SCREAMING_SNAKE_CASE__         =    log.split("\n"						)[:-1]
      self.assertEqual(self.get_value_from_key(A_  ,	"curr_step"  ,	A_						)  ,	0						)
      self.assertEqual(self.get_value_from_key(A_  ,	"total_loss"						)  ,	0.1						)
      self.assertEqual(self.get_value_from_key(A_  ,	"iteration"						)  ,	1						)
      self.assertEqual(self.get_value_from_key(A_  ,	"my_text"						)  ,	"some_value"						)







class  UpperCamelCase_      (	_snake_case      ):


  """simple docstring"""
  SCREAMING_SNAKE_CASE__					:			List[Any]					    =   [
      """total_loss""",
      """iteration""",
      """my_text""",
      """learning_rate""",
      """num_iterations""",
      """some_boolean""",
      """some_string""",
  ]

  SCREAMING_SNAKE_CASE__					:			Any					    =   """my_custom_tracker"""
  SCREAMING_SNAKE_CASE__					:			Optional[Any]					    =   False



  def __init__(						self     :					Any  ,	A_     :					str						)       ->			List[Any]:
      SCREAMING_SNAKE_CASE__         =    open(f"{dir}/log.csv"  ,	"w+"						)
      SCREAMING_SNAKE_CASE__         =    csv.DictWriter(self.f  ,	fieldnames=self._col_names						)
      self.writer.writeheader()



  @property
  def   _UpperCamelCase     (						self     :					str						)       ->			Optional[Any]:
      return self.writer



  def   _UpperCamelCase     (						self     :					Optional[Any]  ,	A_     :					dict						)       ->			Optional[int]:
      logger.info("Call init"						)
      self.writer.writerow(A_						)



  def   _UpperCamelCase     (						self     :					Any  ,	A_     :					dict  ,	A_     :					Optional[int]						)       ->			List[Any]:
      logger.info("Call log"						)
      self.writer.writerow(A_						)







  def   _UpperCamelCase     (						self     :					Dict						)       ->			List[str]:
      self.f.close()







class  UpperCamelCase_      (	unittest.TestCase      ):


  """simple docstring"""



  def   _UpperCamelCase     (						self     :					Tuple						)       ->			str:
      with tempfile.TemporaryDirectory() as d:
          SCREAMING_SNAKE_CASE__         =    MyCustomTracker(A_						)
          SCREAMING_SNAKE_CASE__         =    Accelerator(log_with=A_						)
          SCREAMING_SNAKE_CASE__         =    {"num_iterations": 12, "learning_rate": 1E-2, "some_boolean": False, "some_string": "some_value"}
          accelerator.init_trackers("Some name"  ,	A_						)
          accelerator.end_training()
          with open(f"{d}/log.csv"  ,	"r"						) as f:
              SCREAMING_SNAKE_CASE__         =    csv.DictReader(A_						)
              SCREAMING_SNAKE_CASE__         =    next(A_						)
              SCREAMING_SNAKE_CASE__         =    {
                  "total_loss": "",
                  "iteration": "",
                  "my_text": "",
                  "learning_rate": "0.01",
                  "num_iterations": "12",
                  "some_boolean": "False",
                  "some_string": "some_value",
              }
              self.assertDictEqual(A_  ,	A_						)







  def   _UpperCamelCase     (						self     :					Dict						)       ->			Tuple:
      with tempfile.TemporaryDirectory() as d:
          SCREAMING_SNAKE_CASE__         =    MyCustomTracker(A_						)
          SCREAMING_SNAKE_CASE__         =    Accelerator(log_with=A_						)
          accelerator.init_trackers("Some name"						)
          SCREAMING_SNAKE_CASE__         =    {"total_loss": 0.1, "iteration": 1, "my_text": "some_value"}
          accelerator.log(A_  ,	step=0						)
          accelerator.end_training()
          with open(f"{d}/log.csv"  ,	"r"						) as f:
              SCREAMING_SNAKE_CASE__         =    csv.DictReader(A_						)
              SCREAMING_SNAKE_CASE__         =    next(A_						)
              SCREAMING_SNAKE_CASE__         =    {
                  "total_loss": "0.1",
                  "iteration": "1",
                  "my_text": "some_value",
                  "learning_rate": "",
                  "num_iterations": "",
                  "some_boolean": "",
                  "some_string": "",
              }
              self.assertDictEqual(A_  ,	A_						)





import unittest

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

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


if is_torch_available():
  import torch

  from transformers import (
      LayoutLMForMaskedLM,
      LayoutLMForQuestionAnswering,
      LayoutLMForSequenceClassification,
      LayoutLMForTokenClassification,
      LayoutLMModel,
  )







class  UpperCamelCase_      :


  """simple docstring"""



  def __init__(						self     :					Tuple  ,	A_     :					List[str]  ,	A_     :					List[Any]=13  ,	A_     :					Optional[int]=7  ,	A_     :					List[Any]=True  ,	A_     :					str=True  ,	A_     :					str=True  ,	A_     :					str=True  ,	A_     :					List[Any]=99  ,	A_     :					str=32  ,	A_     :					Dict=5  ,	A_     :					Tuple=4  ,	A_     :					Optional[Any]=37  ,	A_     :					List[Any]="gelu"  ,	A_     :					Any=0.1  ,	A_     :					Union[str, Any]=0.1  ,	A_     :					List[Any]=512  ,	A_     :					Dict=16  ,	A_     :					List[Any]=2  ,	A_     :					Optional[Any]=0.02  ,	A_     :					List[Any]=3  ,	A_     :					Optional[Any]=4  ,	A_     :					Dict=None  ,	A_     :					str=1000  ,	)       ->			str:
      SCREAMING_SNAKE_CASE__         =    parent
      SCREAMING_SNAKE_CASE__         =    batch_size
      SCREAMING_SNAKE_CASE__         =    seq_length
      SCREAMING_SNAKE_CASE__         =    is_training
      SCREAMING_SNAKE_CASE__         =    use_input_mask
      SCREAMING_SNAKE_CASE__         =    use_token_type_ids
      SCREAMING_SNAKE_CASE__         =    use_labels
      SCREAMING_SNAKE_CASE__         =    vocab_size
      SCREAMING_SNAKE_CASE__         =    hidden_size
      SCREAMING_SNAKE_CASE__         =    num_hidden_layers
      SCREAMING_SNAKE_CASE__         =    num_attention_heads
      SCREAMING_SNAKE_CASE__         =    intermediate_size
      SCREAMING_SNAKE_CASE__         =    hidden_act
      SCREAMING_SNAKE_CASE__         =    hidden_dropout_prob
      SCREAMING_SNAKE_CASE__         =    attention_probs_dropout_prob
      SCREAMING_SNAKE_CASE__         =    max_position_embeddings
      SCREAMING_SNAKE_CASE__         =    type_vocab_size
      SCREAMING_SNAKE_CASE__         =    type_sequence_label_size
      SCREAMING_SNAKE_CASE__         =    initializer_range
      SCREAMING_SNAKE_CASE__         =    num_labels
      SCREAMING_SNAKE_CASE__         =    num_choices
      SCREAMING_SNAKE_CASE__         =    scope
      SCREAMING_SNAKE_CASE__         =    range_bbox



  def   _UpperCamelCase     (						self     :					List[str]						)       ->			int:
      SCREAMING_SNAKE_CASE__         =    ids_tensor([self.batch_size, self.seq_length]  ,	self.vocab_size						)

      SCREAMING_SNAKE_CASE__         =    ids_tensor([self.batch_size, self.seq_length, 4]  ,	self.range_bbox						)
      # Ensure that bbox is legal
      for i in range(bbox.shape[0]						):
          for j in range(bbox.shape[1]						):
              if bbox[i, j, 3] < bbox[i, j, 1]:
                  SCREAMING_SNAKE_CASE__         =    bbox[i, j, 3]
                  SCREAMING_SNAKE_CASE__         =    bbox[i, j, 1]
                  SCREAMING_SNAKE_CASE__         =    t
              if bbox[i, j, 2] < bbox[i, j, 0]:
                  SCREAMING_SNAKE_CASE__         =    bbox[i, j, 2]
                  SCREAMING_SNAKE_CASE__         =    bbox[i, j, 0]
                  SCREAMING_SNAKE_CASE__         =    t

      SCREAMING_SNAKE_CASE__         =    None
      if self.use_input_mask:
          SCREAMING_SNAKE_CASE__         =    ids_tensor([self.batch_size, self.seq_length]  ,	vocab_size=2						)

      SCREAMING_SNAKE_CASE__         =    None
      if self.use_token_type_ids:
          SCREAMING_SNAKE_CASE__         =    ids_tensor([self.batch_size, self.seq_length]  ,	self.type_vocab_size						)

      SCREAMING_SNAKE_CASE__         =    None
      SCREAMING_SNAKE_CASE__         =    None
      SCREAMING_SNAKE_CASE__         =    None
      if self.use_labels:
          SCREAMING_SNAKE_CASE__         =    ids_tensor([self.batch_size]  ,	self.type_sequence_label_size						)
          SCREAMING_SNAKE_CASE__         =    ids_tensor([self.batch_size, self.seq_length]  ,	self.num_labels						)
          SCREAMING_SNAKE_CASE__         =    ids_tensor([self.batch_size]  ,	self.num_choices						)

      SCREAMING_SNAKE_CASE__         =    self.get_config()

      return config, input_ids, bbox, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels



  def   _UpperCamelCase     (						self     :					List[str]						)       ->			int:
      return LayoutLMConfig(
          vocab_size=self.vocab_size  ,	hidden_size=self.hidden_size  ,	num_hidden_layers=self.num_hidden_layers  ,	num_attention_heads=self.num_attention_heads  ,	intermediate_size=self.intermediate_size  ,	hidden_act=self.hidden_act  ,	hidden_dropout_prob=self.hidden_dropout_prob  ,	attention_probs_dropout_prob=self.attention_probs_dropout_prob  ,	max_position_embeddings=self.max_position_embeddings  ,	type_vocab_size=self.type_vocab_size  ,	initializer_range=self.initializer_range  ,	)



  def   _UpperCamelCase     (						self     :					List[Any]  ,	A_     :					Any  ,	A_     :					List[str]  ,	A_     :					Optional[int]  ,	A_     :					Dict  ,	A_     :					int  ,	A_     :					List[str]  ,	A_     :					Any  ,	A_     :					Tuple						)       ->			List[str]:
      SCREAMING_SNAKE_CASE__         =    LayoutLMModel(config=A_						)
      model.to(A_						)
      model.eval()
      SCREAMING_SNAKE_CASE__         =    model(A_  ,	A_  ,	attention_mask=A_  ,	token_type_ids=A_						)
      SCREAMING_SNAKE_CASE__         =    model(A_  ,	A_  ,	token_type_ids=A_						)
      SCREAMING_SNAKE_CASE__         =    model(A_  ,	A_						)
      self.parent.assertEqual(result.last_hidden_state.shape  ,	(self.batch_size, self.seq_length, self.hidden_size)						)
      self.parent.assertEqual(result.pooler_output.shape  ,	(self.batch_size, self.hidden_size)						)



  def   _UpperCamelCase     (						self     :					int  ,	A_     :					int  ,	A_     :					Any  ,	A_     :					Tuple  ,	A_     :					Dict  ,	A_     :					int  ,	A_     :					Any  ,	A_     :					int  ,	A_     :					Union[str, Any]						)       ->			Tuple:
      SCREAMING_SNAKE_CASE__         =    LayoutLMForMaskedLM(config=A_						)
      model.to(A_						)
      model.eval()
      SCREAMING_SNAKE_CASE__         =    model(A_  ,	A_  ,	attention_mask=A_  ,	token_type_ids=A_  ,	labels=A_						)
      self.parent.assertEqual(result.logits.shape  ,	(self.batch_size, self.seq_length, self.vocab_size)						)



  def   _UpperCamelCase     (						self     :					Dict  ,	A_     :					Union[str, Any]  ,	A_     :					Union[str, Any]  ,	A_     :					Dict  ,	A_     :					Optional[int]  ,	A_     :					Tuple  ,	A_     :					Optional[int]  ,	A_     :					Optional[int]  ,	A_     :					Optional[Any]						)       ->			Optional[Any]:
      SCREAMING_SNAKE_CASE__         =    self.num_labels
      SCREAMING_SNAKE_CASE__         =    LayoutLMForSequenceClassification(A_						)
      model.to(A_						)
      model.eval()
      SCREAMING_SNAKE_CASE__         =    model(
          A_  ,	A_  ,	attention_mask=A_  ,	token_type_ids=A_  ,	labels=A_						)
      self.parent.assertEqual(result.logits.shape  ,	(self.batch_size, self.num_labels)						)



  def   _UpperCamelCase     (						self     :					Any  ,	A_     :					Any  ,	A_     :					Any  ,	A_     :					str  ,	A_     :					Any  ,	A_     :					Optional[Any]  ,	A_     :					Tuple  ,	A_     :					Tuple  ,	A_     :					Any						)       ->			Any:
      SCREAMING_SNAKE_CASE__         =    self.num_labels
      SCREAMING_SNAKE_CASE__         =    LayoutLMForTokenClassification(config=A_						)
      model.to(A_						)
      model.eval()
      SCREAMING_SNAKE_CASE__         =    model(A_  ,	A_  ,	attention_mask=A_  ,	token_type_ids=A_  ,	labels=A_						)
      self.parent.assertEqual(result.logits.shape  ,	(self.batch_size, self.seq_length, self.num_labels)						)



  def   _UpperCamelCase     (						self     :					Optional[int]  ,	A_     :					str  ,	A_     :					Optional[Any]  ,	A_     :					int  ,	A_     :					str  ,	A_     :					Optional[Any]  ,	A_     :					List[str]  ,	A_     :					Union[str, Any]  ,	A_     :					Optional[Any]						)       ->			Any:
      SCREAMING_SNAKE_CASE__         =    LayoutLMForQuestionAnswering(config=A_						)
      model.to(A_						)
      model.eval()
      SCREAMING_SNAKE_CASE__         =    model(
          A_  ,	bbox=A_  ,	attention_mask=A_  ,	token_type_ids=A_  ,	start_positions=A_  ,	end_positions=A_  ,	)
      self.parent.assertEqual(result.start_logits.shape  ,	(self.batch_size, self.seq_length)						)
      self.parent.assertEqual(result.end_logits.shape  ,	(self.batch_size, self.seq_length)						)







  def   _UpperCamelCase     (						self     :					Optional[Any]						)       ->			List[str]:
      SCREAMING_SNAKE_CASE__         =    self.prepare_config_and_inputs()
      (
          (
          SCREAMING_SNAKE_CASE__
      )  ,		(
          SCREAMING_SNAKE_CASE__
      )  ,		(
          SCREAMING_SNAKE_CASE__
      )  ,		(
          SCREAMING_SNAKE_CASE__
      )  ,		(
          SCREAMING_SNAKE_CASE__
      )  ,		(
          SCREAMING_SNAKE_CASE__
      )  ,		(
          SCREAMING_SNAKE_CASE__
      )  ,		(
          SCREAMING_SNAKE_CASE__
      )  ,		
      )         =    config_and_inputs
      SCREAMING_SNAKE_CASE__         =    {
          "input_ids": input_ids,
          "bbox": bbox,
          "token_type_ids": token_type_ids,
          "attention_mask": input_mask,
      }
      return config, inputs_dict







@require_torch
class  UpperCamelCase_      (	_snake_case		,  _snake_case		,  unittest.TestCase      ):


  """simple docstring"""
  SCREAMING_SNAKE_CASE__					:			Optional[int]					    =   (
      (
          LayoutLMModel,
          LayoutLMForMaskedLM,
          LayoutLMForSequenceClassification,
          LayoutLMForTokenClassification,
          LayoutLMForQuestionAnswering,
      )
      if is_torch_available()
      else None
  )
  SCREAMING_SNAKE_CASE__					:			Union[str, Any]					    =   (
      {
          """document-question-answering""": LayoutLMForQuestionAnswering,
          """feature-extraction""": LayoutLMModel,
          """fill-mask""": LayoutLMForMaskedLM,
          """text-classification""": LayoutLMForSequenceClassification,
          """token-classification""": LayoutLMForTokenClassification,
          """zero-shot""": LayoutLMForSequenceClassification,
      }
      if is_torch_available()
      else {}
  )
  SCREAMING_SNAKE_CASE__					:			Any					    =   True



  def   _UpperCamelCase     (						self     :					int						)       ->			Optional[Any]:
      SCREAMING_SNAKE_CASE__         =    LayoutLMModelTester(self						)
      SCREAMING_SNAKE_CASE__         =    ConfigTester(self  ,	config_class=A_  ,	hidden_size=37						)



  def   _UpperCamelCase     (						self     :					Tuple						)       ->			Tuple:
      self.config_tester.run_common_tests()



  def   _UpperCamelCase     (						self     :					Dict						)       ->			Tuple:
      SCREAMING_SNAKE_CASE__         =    self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_model(*A_						)



  def   _UpperCamelCase     (						self     :					Optional[Any]						)       ->			Dict:
      SCREAMING_SNAKE_CASE__         =    self.model_tester.prepare_config_and_inputs()
      for type in ["absolute", "relative_key", "relative_key_query"]:
          SCREAMING_SNAKE_CASE__         =    type
          self.model_tester.create_and_check_model(*A_						)



  def   _UpperCamelCase     (						self     :					Tuple						)       ->			Tuple:
      SCREAMING_SNAKE_CASE__         =    self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_masked_lm(*A_						)



  def   _UpperCamelCase     (						self     :					Dict						)       ->			Optional[Any]:
      SCREAMING_SNAKE_CASE__         =    self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_sequence_classification(*A_						)



  def   _UpperCamelCase     (						self     :					List[str]						)       ->			List[str]:
      SCREAMING_SNAKE_CASE__         =    self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_token_classification(*A_						)



  def   _UpperCamelCase     (						self     :					int						)       ->			Optional[Any]:
      SCREAMING_SNAKE_CASE__         =    self.model_tester.prepare_config_and_inputs()
      self.model_tester.create_and_check_for_question_answering(*A_						)







  @unittest.skip("Will be fixed soon by reducing the size of the model used for common tests."						)
  def   _UpperCamelCase     (						self     :					List[Any]						)       ->			str:
      pass





def 				__a							(       ):
    # Here we prepare a batch of 2 sequences to test a LayoutLM forward pass on:
    # fmt: off
    SCREAMING_SNAKE_CASE__         =    torch.tensor([[101,1019,1014,1016,1037,12849,4747,1004,14246,2278,5439,4524,5002,2930,2193,2930,4341,3208,1005,1055,2171,2848,11300,3531,102],[101,4070,4034,7020,1024,3058,1015,1013,2861,1013,6070,19274,2772,6205,27814,16147,16147,4343,2047,10283,10969,14389,1012,2338,102]]			,							device=__A						)  # noqa: E231
    SCREAMING_SNAKE_CASE__         =    torch.tensor([[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],]			,							device=__A						)  # noqa: E231
    SCREAMING_SNAKE_CASE__         =    torch.tensor([[[0,0,0,0],[423,237,440,251],[427,272,441,287],[419,115,437,129],[961,885,992,912],[256,38,330,58],[256,38,330,58],[336,42,353,57],[360,39,401,56],[360,39,401,56],[411,39,471,59],[479,41,528,59],[533,39,630,60],[67,113,134,131],[141,115,209,132],[68,149,133,166],[141,149,187,164],[195,148,287,165],[195,148,287,165],[195,148,287,165],[295,148,349,165],[441,149,492,166],[497,149,546,164],[64,201,125,218],[1000,1000,1000,1000]],[[0,0,0,0],[662,150,754,166],[665,199,742,211],[519,213,554,228],[519,213,554,228],[134,433,187,454],[130,467,204,480],[130,467,204,480],[130,467,204,480],[130,467,204,480],[130,467,204,480],[314,469,376,482],[504,684,582,706],[941,825,973,900],[941,825,973,900],[941,825,973,900],[941,825,973,900],[610,749,652,765],[130,659,168,672],[176,657,237,672],[238,657,312,672],[443,653,628,672],[443,653,628,672],[716,301,825,317],[1000,1000,1000,1000]]]			,							device=__A						)  # noqa: E231
    SCREAMING_SNAKE_CASE__         =    torch.tensor([[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0],[0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0]]			,							device=__A						)  # noqa: E231
    # these are sequence labels (i.e. at the token level)
    SCREAMING_SNAKE_CASE__         =    torch.tensor([[-100,10,10,10,9,1,-100,7,7,-100,7,7,4,2,5,2,8,8,-100,-100,5,0,3,2,-100],[-100,12,12,12,-100,12,10,-100,-100,-100,-100,10,12,9,-100,-100,-100,10,10,10,9,12,-100,10,-100]]			,							device=__A						)  # noqa: E231
    # fmt: on

    return input_ids, attention_mask, bbox, token_type_ids, labels







@require_torch
class  UpperCamelCase_      (	unittest.TestCase      ):


  """simple docstring"""



  @slow
  def   _UpperCamelCase     (						self     :					List[str]						)       ->			List[Any]:
      SCREAMING_SNAKE_CASE__         =    LayoutLMModel.from_pretrained("microsoft/layoutlm-base-uncased"						).to(A_						)

      SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__         =    prepare_layoutlm_batch_inputs()

      # forward pass
      SCREAMING_SNAKE_CASE__         =    model(input_ids=A_  ,	bbox=A_  ,	attention_mask=A_  ,	token_type_ids=A_						)

      # test the sequence output on [0, :3, :3]
      SCREAMING_SNAKE_CASE__         =    torch.tensor(
          [[0.1_785, -0.1_947, -0.0_425], [-0.3_254, -0.2_807, 0.2_553], [-0.5_391, -0.3_322, 0.3_364]]  ,	device=A_  ,	)

      self.assertTrue(torch.allclose(outputs.last_hidden_state[0, :3, :3]  ,	A_  ,	atol=1E-3						)						)

      # test the pooled output on [1, :3]
      SCREAMING_SNAKE_CASE__         =    torch.tensor([-0.6_580, -0.0_214, 0.8_552]  ,	device=A_						)

      self.assertTrue(torch.allclose(outputs.pooler_output[1, :3]  ,	A_  ,	atol=1E-3						)						)



  @slow
  def   _UpperCamelCase     (						self     :					str						)       ->			str:
      # initialize model with randomly initialized sequence classification head
      SCREAMING_SNAKE_CASE__         =    LayoutLMForSequenceClassification.from_pretrained("microsoft/layoutlm-base-uncased"  ,	num_labels=2						).to(
          A_						)

      SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__         =    prepare_layoutlm_batch_inputs()

      # forward pass
      SCREAMING_SNAKE_CASE__         =    model(
          input_ids=A_  ,	bbox=A_  ,	attention_mask=A_  ,	token_type_ids=A_  ,	labels=torch.tensor([1, 1]  ,	device=A_						)  ,	)

      # test whether we get a loss as a scalar
      SCREAMING_SNAKE_CASE__         =    outputs.loss
      SCREAMING_SNAKE_CASE__         =    torch.Size([]						)
      self.assertEqual(loss.shape  ,	A_						)

      # test the shape of the logits
      SCREAMING_SNAKE_CASE__         =    outputs.logits
      SCREAMING_SNAKE_CASE__         =    torch.Size((2, 2)						)
      self.assertEqual(logits.shape  ,	A_						)



  @slow
  def   _UpperCamelCase     (						self     :					List[str]						)       ->			List[Any]:
      # initialize model with randomly initialized token classification head
      SCREAMING_SNAKE_CASE__         =    LayoutLMForTokenClassification.from_pretrained("microsoft/layoutlm-base-uncased"  ,	num_labels=13						).to(
          A_						)

      SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__         =    prepare_layoutlm_batch_inputs()

      # forward pass
      SCREAMING_SNAKE_CASE__         =    model(
          input_ids=A_  ,	bbox=A_  ,	attention_mask=A_  ,	token_type_ids=A_  ,	labels=A_						)

      # test the loss calculation to be around 2.65
      # expected_loss = torch.tensor(2.65, device=torch_device)

      # The loss is currently somewhat random and can vary between 0.1-0.3 atol.
      # self.assertTrue(torch.allclose(outputs.loss, expected_loss, atol=0.1))

      # test the shape of the logits
      SCREAMING_SNAKE_CASE__         =    outputs.logits
      SCREAMING_SNAKE_CASE__         =    torch.Size((2, 25, 13)						)
      self.assertEqual(logits.shape  ,	A_						)







  @slow
  def   _UpperCamelCase     (						self     :					Dict						)       ->			List[Any]:
      # initialize model with randomly initialized token classification head
      SCREAMING_SNAKE_CASE__         =    LayoutLMForQuestionAnswering.from_pretrained("microsoft/layoutlm-base-uncased"						).to(A_						)

      SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__  ,		SCREAMING_SNAKE_CASE__         =    prepare_layoutlm_batch_inputs()

      # forward pass
      SCREAMING_SNAKE_CASE__         =    model(input_ids=A_  ,	bbox=A_  ,	attention_mask=A_  ,	token_type_ids=A_						)

      # test the shape of the logits
      SCREAMING_SNAKE_CASE__         =    torch.Size((2, 25)						)
      self.assertEqual(outputs.start_logits.shape  ,	A_						)
      self.assertEqual(outputs.end_logits.shape  ,	A_						)


