





import unittest

from transformers import MODEL_FOR_DOCUMENT_QUESTION_ANSWERING_MAPPING, AutoTokenizer, is_vision_available
from transformers.pipelines import pipeline
from transformers.pipelines.document_question_answering import apply_tesseract
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_detectrona,
    require_pytesseract,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
      from PIL import Image

      from transformers.image_utils import load_image
else:

      class    _UpperCAmelCase							:
            @staticmethod
            def 		_lowerCamelCase      (		*a__      ,       **a__       ):
               pass






      def     _lowerCAmelCase		(  _lowerCAmelCase     ):
         '''simple docstring'''


         return None


# This is a pinned image from a specific revision of a document question answering space, hosted by HuggingFace,
# so we can expect it to be available.
_lowerCAmelCase 				=     (
    'https://huggingface.co/spaces/impira/docquery/resolve/2f6c96314dc84dfda62d40de9da55f2f5165d403/invoice.png'
)

@is_pipeline_test
@require_torch
@require_vision
class    _UpperCAmelCase							(		unittest.TestCase     ):
      a						       =							MODEL_FOR_DOCUMENT_QUESTION_ANSWERING_MAPPING
      @require_pytesseract
      @require_vision
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__      ,       a__       ):
         A_     :				Optional[Any]			      =	pipeline(
             """document-question-answering"""      ,       model=__a      ,       tokenizer=__a      ,       image_processor=__a       )

         A_     :				List[Any]			      =	INVOICE_URL
         A_     :				str			      =	list(zip(*apply_tesseract(load_image(__a       )      ,       __a      ,       """"""       )       )       )
         A_     :				Optional[Any]			      =	"""What is the placebo?"""
         A_     :				Tuple			      =	[
             {
                 """image""": load_image(__a       ),
                 """question""": question,
             },
             {
                 """image""": image,
                 """question""": question,
             },
             {
                 """image""": image,
                 """question""": question,
                 """word_boxes""": word_boxes,
             },
         ]
         return dqa_pipeline, examples
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__       ):
         A_     :				Optional[int]			      =	dqa_pipeline(__a      ,       top_k=2       )
         self.assertEqual(
             __a      ,       [
                 [
                     {"""score""": ANY(__a       ), """answer""": ANY(__a       ), """start""": ANY(__a       ), """end""": ANY(__a       )},
                     {"""score""": ANY(__a       ), """answer""": ANY(__a       ), """start""": ANY(__a       ), """end""": ANY(__a       )},
                 ]
             ]
             * 3      ,       )
      @require_torch
      @require_detectrona
      @require_pytesseract
      def 		_lowerCamelCase      (		self       ):
         A_     :				List[str]			      =	pipeline("""document-question-answering"""      ,       model="""hf-internal-testing/tiny-random-layoutlmv2"""       )
         A_     :				Optional[int]			      =	INVOICE_URL
         A_     :				Optional[int]			      =	"""How many cats are there?"""

         A_     :				Any			      =	[
             {"""score""": 0.0001, """answer""": """oy 2312/2019""", """start""": 38, """end""": 39},
             {"""score""": 0.0001, """answer""": """oy 2312/2019 DUE""", """start""": 38, """end""": 40},
         ]
         A_     :				Optional[int]			      =	dqa_pipeline(image=__a      ,       question=__a      ,       top_k=2       )
         self.assertEqual(nested_simplify(__a      ,       decimals=4       )      ,       __a       )

         A_     :				Union[str, Any]			      =	dqa_pipeline({"""image""": image, """question""": question}      ,       top_k=2       )
         self.assertEqual(nested_simplify(__a      ,       decimals=4       )      ,       __a       )

         # This image does not detect ANY text in it, meaning layoutlmv2 should fail.
         # Empty answer probably
         A_     :				Union[str, Any]			      =	"""./tests/fixtures/tests_samples/COCO/000000039769.png"""
         A_     :				Tuple			      =	dqa_pipeline(image=__a      ,       question=__a      ,       top_k=2       )
         self.assertEqual(__a      ,       []       )

         # We can optionnally pass directly the words and bounding boxes
         A_     :				List[str]			      =	"""./tests/fixtures/tests_samples/COCO/000000039769.png"""
         A_     :				str			      =	[]
         A_     :				int			      =	[]
         A_     :				Dict			      =	dqa_pipeline(image=__a      ,       question=__a      ,       words=__a      ,       boxes=__a      ,       top_k=2       )
         self.assertEqual(__a      ,       []       )
      @slow
      @require_torch
      @require_detectrona
      @require_pytesseract
      def 		_lowerCamelCase      (		self       ):
         A_     :				Tuple			      =	pipeline(
             """document-question-answering"""      ,       model="""tiennvcs/layoutlmv2-base-uncased-finetuned-docvqa"""      ,       revision="""9977165"""      ,       )
         A_     :				Optional[Any]			      =	INVOICE_URL
         A_     :				Union[str, Any]			      =	"""What is the invoice number?"""

         A_     :				Dict			      =	dqa_pipeline(image=__a      ,       question=__a      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 {"""score""": 0.9944, """answer""": """us-001""", """start""": 16, """end""": 16},
                 {"""score""": 0.0009, """answer""": """us-001""", """start""": 16, """end""": 16},
             ]      ,       )

         A_     :				Optional[Any]			      =	dqa_pipeline({"""image""": image, """question""": question}      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 {"""score""": 0.9944, """answer""": """us-001""", """start""": 16, """end""": 16},
                 {"""score""": 0.0009, """answer""": """us-001""", """start""": 16, """end""": 16},
             ]      ,       )

         A_     :				Tuple			      =	dqa_pipeline(
             [{"""image""": image, """question""": question}, {"""image""": image, """question""": question}]      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 [
                     {"""score""": 0.9944, """answer""": """us-001""", """start""": 16, """end""": 16},
                     {"""score""": 0.0009, """answer""": """us-001""", """start""": 16, """end""": 16},
                 ],
             ]
             * 2      ,       )
      @slow
      @require_torch
      @require_detectrona
      @require_pytesseract
      def 		_lowerCamelCase      (		self       ):
         A_     :				Optional[Any]			      =	pipeline(
             """document-question-answering"""      ,       model="""tiennvcs/layoutlmv2-base-uncased-finetuned-docvqa"""      ,       revision="""9977165"""      ,       max_seq_len=50      ,       )
         A_     :				Union[str, Any]			      =	INVOICE_URL
         A_     :				Any			      =	"""What is the invoice number?"""

         A_     :				Any			      =	dqa_pipeline(image=__a      ,       question=__a      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 {"""score""": 0.9974, """answer""": """1110212019""", """start""": 23, """end""": 23},
                 {"""score""": 0.9948, """answer""": """us-001""", """start""": 16, """end""": 16},
             ]      ,       )

         A_     :				int			      =	dqa_pipeline({"""image""": image, """question""": question}      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 {"""score""": 0.9974, """answer""": """1110212019""", """start""": 23, """end""": 23},
                 {"""score""": 0.9948, """answer""": """us-001""", """start""": 16, """end""": 16},
             ]      ,       )

         A_     :				Optional[Any]			      =	dqa_pipeline(
             [{"""image""": image, """question""": question}, {"""image""": image, """question""": question}]      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 [
                     {"""score""": 0.9974, """answer""": """1110212019""", """start""": 23, """end""": 23},
                     {"""score""": 0.9948, """answer""": """us-001""", """start""": 16, """end""": 16},
                 ]
             ]
             * 2      ,       )
      @slow
      @require_torch
      @require_pytesseract
      @require_vision
      def 		_lowerCamelCase      (		self       ):
         A_     :				str			      =	AutoTokenizer.from_pretrained(
             """impira/layoutlm-document-qa"""      ,       revision="""3dc6de3"""      ,       add_prefix_space=__a       )
         A_     :				str			      =	pipeline(
             """document-question-answering"""      ,       model="""impira/layoutlm-document-qa"""      ,       tokenizer=__a      ,       revision="""3dc6de3"""      ,       )
         A_     :				Tuple			      =	INVOICE_URL
         A_     :				int			      =	"""What is the invoice number?"""

         A_     :				List[str]			      =	dqa_pipeline(image=__a      ,       question=__a      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 {"""score""": 0.4251, """answer""": """us-001""", """start""": 16, """end""": 16},
                 {"""score""": 0.0819, """answer""": """1110212019""", """start""": 23, """end""": 23},
             ]      ,       )

         A_     :				Optional[int]			      =	dqa_pipeline({"""image""": image, """question""": question}      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 {"""score""": 0.4251, """answer""": """us-001""", """start""": 16, """end""": 16},
                 {"""score""": 0.0819, """answer""": """1110212019""", """start""": 23, """end""": 23},
             ]      ,       )

         A_     :				Union[str, Any]			      =	dqa_pipeline(
             [{"""image""": image, """question""": question}, {"""image""": image, """question""": question}]      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 [
                     {"""score""": 0.4251, """answer""": """us-001""", """start""": 16, """end""": 16},
                     {"""score""": 0.0819, """answer""": """1110212019""", """start""": 23, """end""": 23},
                 ]
             ]
             * 2      ,       )

         A_     :				str			      =	list(zip(*apply_tesseract(load_image(__a       )      ,       __a      ,       """"""       )       )       )

         # This model should also work if `image` is set to None
         A_     :				Optional[Any]			      =	dqa_pipeline({"""image""": None, """word_boxes""": word_boxes, """question""": question}      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 {"""score""": 0.4251, """answer""": """us-001""", """start""": 16, """end""": 16},
                 {"""score""": 0.0819, """answer""": """1110212019""", """start""": 23, """end""": 23},
             ]      ,       )
      @slow
      @require_torch
      @require_pytesseract
      @require_vision
      def 		_lowerCamelCase      (		self       ):
         A_     :				int			      =	AutoTokenizer.from_pretrained(
             """impira/layoutlm-document-qa"""      ,       revision="""3dc6de3"""      ,       add_prefix_space=__a       )
         A_     :				int			      =	pipeline(
             """document-question-answering"""      ,       model="""impira/layoutlm-document-qa"""      ,       tokenizer=__a      ,       revision="""3dc6de3"""      ,       max_seq_len=50      ,       )
         A_     :				int			      =	INVOICE_URL
         A_     :				int			      =	"""What is the invoice number?"""

         A_     :				List[Any]			      =	dqa_pipeline(image=__a      ,       question=__a      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 {"""score""": 0.9999, """answer""": """us-001""", """start""": 16, """end""": 16},
                 {"""score""": 0.9998, """answer""": """us-001""", """start""": 16, """end""": 16},
             ]      ,       )

         A_     :				str			      =	dqa_pipeline(
             [{"""image""": image, """question""": question}, {"""image""": image, """question""": question}]      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 [
                     {"""score""": 0.9999, """answer""": """us-001""", """start""": 16, """end""": 16},
                     {"""score""": 0.9998, """answer""": """us-001""", """start""": 16, """end""": 16},
                 ]
             ]
             * 2      ,       )

         A_     :				List[str]			      =	list(zip(*apply_tesseract(load_image(__a       )      ,       __a      ,       """"""       )       )       )

         # This model should also work if `image` is set to None
         A_     :				Dict			      =	dqa_pipeline({"""image""": None, """word_boxes""": word_boxes, """question""": question}      ,       top_k=2       )
         self.assertEqual(
             nested_simplify(__a      ,       decimals=4       )      ,       [
                 {"""score""": 0.9999, """answer""": """us-001""", """start""": 16, """end""": 16},
                 {"""score""": 0.9998, """answer""": """us-001""", """start""": 16, """end""": 16},
             ]      ,       )
      @slow
      @require_torch
      def 		_lowerCamelCase      (		self       ):
         A_     :				Tuple			      =	pipeline(
             """document-question-answering"""      ,       model="""naver-clova-ix/donut-base-finetuned-docvqa"""      ,       tokenizer=AutoTokenizer.from_pretrained("""naver-clova-ix/donut-base-finetuned-docvqa"""       )      ,       feature_extractor="""naver-clova-ix/donut-base-finetuned-docvqa"""      ,       )

         A_     :				str			      =	INVOICE_URL
         A_     :				Any			      =	"""What is the invoice number?"""
         A_     :				Tuple			      =	dqa_pipeline(image=__a      ,       question=__a      ,       top_k=2       )
         self.assertEqual(nested_simplify(__a      ,       decimals=4       )      ,       [{"""answer""": """us-001"""}]       )






      @require_tf
      @unittest.skip("""Document question answering not implemented in TF"""       )
      def 		_lowerCamelCase      (		self       ):
         pass









import baseaa
import io
import json
import os
from copy import deepcopy

from ..optimizer import AcceleratedOptimizer
from ..scheduler import AcceleratedScheduler

class    _UpperCAmelCase							:
      def __init__(		self      ,       a__       ):
         if isinstance(a__      ,       a__       ):
            # Don't modify user's data should they want to reuse it (e.g. in tests), because once we
            # modified it, it will not be accepted here again, since `auto` values would have been overridden
            A_     :				Optional[Any]			      =	deepcopy(a__       )
         elif os.path.exists(a__       ):
            with io.open(a__      ,       """r"""      ,       encoding="""utf-8"""       ) as f:
               A_     :				str			      =	json.load(a__       )
         else:
            try:
               A_     :				Dict			      =	baseaa.urlsafe_baadecode(a__       ).decode("""utf-8"""       )
               A_     :				List[Any]			      =	json.loads(a__       )
            except (UnicodeDecodeError, AttributeError, ValueError):
               raise ValueError(
                   F"""Expected a string path to an existing deepspeed config, or a dictionary, or a base64 encoded string. Received: {config_file_or_dict}"""       )

         A_     :				Any			      =	config

         self.set_stage_and_offload()
      def 		_lowerCamelCase      (		self       ):
         # zero stage - this is done as early as possible, before model is created, to allow
         # ``is_deepspeed_zero3_enabled`` query and getting to the early deepspeed config object
         # during ``zero.Init()`` which needs to know the dtype, and some other hparams.
         A_     :				List[str]			      =	self.get_value("""zero_optimization.stage"""      ,       -1       )

         # offload
         A_     :				Any			      =	False
         if self.is_zeroa() or self.is_zeroa():
            A_     :				Optional[int]			      =	set(["""cpu""", """nvme"""]       )
            A_     :				Dict			      =	set(
                [
                    self.get_value("""zero_optimization.offload_optimizer.device"""       ),
                    self.get_value("""zero_optimization.offload_param.device"""       ),
                ]       )
            if len(offload_devices & offload_devices_valid       ) > 0:
               A_     :				Tuple			      =	True
      def 		_lowerCamelCase      (		self      ,       a__       ):
         A_     :				List[Any]			      =	self.config

         # find the config node of interest if it exists
         A_     :				Optional[Any]			      =	ds_key_long.split("""."""       )
         A_     :				Union[str, Any]			      =	nodes.pop()
         for node in nodes:
            A_     :				List[str]			      =	config.get(a__       )
            if config is None:
               return None, ds_key

         return config, ds_key
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__=None       ):
         A_      ,				A_     :				Union[str, Any]			      =	self.find_config_node(a__       )
         if config is None:
            return default
         return config.get(a__      ,       a__       )
      def 		_lowerCamelCase      (		self      ,       a__      ,       a__=False       ):
         A_     :				Union[str, Any]			      =	self.config

         # find the config node of interest if it exists
         A_     :				str			      =	ds_key_long.split("""."""       )
         for node in nodes:
            A_     :				int			      =	config
            A_     :				int			      =	config.get(a__       )
            if config is None:
               if must_exist:
                  raise ValueError(F"""Can't find {ds_key_long} entry in the config: {self.config}"""       )
               else:
                  return

        # if found remove it
         if parent_config is not None:
            parent_config.pop(a__       )
      def 		_lowerCamelCase      (		self      ,       a__       ):
         A_     :				Optional[Any]			      =	self.get_value(a__       )
         return False if value is None else bool(a__       )
      def 		_lowerCamelCase      (		self      ,       a__       ):
         A_     :				Optional[Any]			      =	self.get_value(a__       )
         return False if value is None else not bool(a__       )
      def 		_lowerCamelCase      (		self       ):
         return self._stage == 2
      def 		_lowerCamelCase      (		self       ):
         return self._stage == 3






      def 		_lowerCamelCase      (		self       ):
         return self._offload






class    _UpperCAmelCase							:
      def __init__(		self      ,       a__       ):
         A_     :				Any			      =	engine






      def 		_lowerCamelCase      (		self      ,       a__      ,       **a__       ):
         # runs backpropagation and handles mixed precision
         self.engine.backward(a__      ,       **a__       )

         # Deepspeed's `engine.step` performs the following operations:
         # - gradient accumulation check
         # - gradient clipping
         # - optimizer step
         # - zero grad
         # - checking overflow
         # - lr_scheduler step (only if engine.lr_scheduler is not None)
         self.engine.step()
         # and this plugin overrides the above calls with no-ops when Accelerate runs under
         # Deepspeed, but allows normal functionality for non-Deepspeed cases thus enabling a simple
         # training loop that works transparently under many training regimes.






class    _UpperCAmelCase							(		_lowerCamelCase     ):
      def __init__(		self      ,       a__       ):
         super().__init__(a__      ,       device_placement=a__      ,       scaler=a__       )
         A_     :				Dict			      =	hasattr(self.optimizer      ,       """overflow"""       )
      def 		_lowerCamelCase      (		self      ,       a__=None       ):
         pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed
      def 		_lowerCamelCase      (		self       ):
         pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed






      @property
      def 		_lowerCamelCase      (		self       ):
         if self.__has_overflow__:
            return self.optimizer.overflow
         return False






class    _UpperCAmelCase							(		_lowerCamelCase     ):
      def __init__(		self      ,       a__      ,       a__       ):
         super().__init__(a__      ,       a__       )






      def 		_lowerCamelCase      (		self       ):
         pass  # `accelerator.backward(loss)` is doing that automatically. Therefore, its implementation is not needed






class    _UpperCAmelCase							:
      def __init__(		self      ,       a__      ,       a__=0.001      ,       a__=0      ,       **a__       ):
         A_     :				List[str]			      =	params
         A_     :				Any			      =	lr
         A_     :				int			      =	weight_decay
         A_     :				Optional[int]			      =	kwargs






class    _UpperCAmelCase							:
      def __init__(		self      ,       a__      ,       a__=None      ,       a__=0      ,       **a__       ):
         A_     :				Union[str, Any]			      =	optimizer
         A_     :				int			      =	total_num_steps
         A_     :				Any			      =	warmup_num_steps
         A_     :				int			      =	kwargs






