



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 		snake_case_		(				*a__ ,						**a__):
            pass







   def 		lowerCAmelCase__    (					UpperCamelCase_				:       Dict	)->  List[str]:
       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.
_lowercase						  =		(
    "https://huggingface.co/spaces/impira/docquery/resolve/2f6c96314dc84dfda62d40de9da55f2f5165d403/invoice.png"
)
@is_pipeline_test
@require_torch
@require_vision
class     _UpperCAmelCase			(       unittest.TestCase     ):
     UpperCamelCase__		    =				MODEL_FOR_DOCUMENT_QUESTION_ANSWERING_MAPPING




     @require_pytesseract
     @require_vision
     def 		snake_case_		(				self ,						a__ ,						a__ ,						a__):
         A__		 =					pipeline(
             '''document-question-answering''' ,						model=a__ ,						tokenizer=a__ ,						image_processor=a__)

         A__		 =					INVOICE_URL
         A__		 =					list(zip(*apply_tesseract(load_image(a__) ,						a__ ,						'''''')))
         A__		 =					'''What is the placebo?'''
         A__		 =					[
             {
                 '''image''': load_image(a__),
                 '''question''': question,
             },
             {
                 '''image''': image,
                 '''question''': question,
             },
             {
                 '''image''': image,
                 '''question''': question,
                 '''word_boxes''': word_boxes,
             },
         ]
         return dqa_pipeline, examples




     def 		snake_case_		(				self ,						a__ ,						a__):
         A__		 =					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 		snake_case_		(				self):
         A__		 =					pipeline('''document-question-answering''' ,						model='''hf-internal-testing/tiny-random-layoutlmv2''')
         A__		 =					INVOICE_URL
         A__		 =					'''How many cats are there?'''

         A__		 =					[
             {'''score''': 0.0_0_0_1, '''answer''': '''oy 2312/2019''', '''start''': 3_8, '''end''': 3_9},
             {'''score''': 0.0_0_0_1, '''answer''': '''oy 2312/2019 DUE''', '''start''': 3_8, '''end''': 4_0},
         ]
         A__		 =					dqa_pipeline(image=a__ ,						question=a__ ,						top_k=2)
         self.assertEqual(nested_simplify(a__ ,						decimals=4) ,						a__)

         A__		 =					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__		 =					'''./tests/fixtures/tests_samples/COCO/000000039769.png'''
         A__		 =					dqa_pipeline(image=a__ ,						question=a__ ,						top_k=2)
         self.assertEqual(a__ ,						[])

         # We can optionnally pass directly the words and bounding boxes
         A__		 =					'''./tests/fixtures/tests_samples/COCO/000000039769.png'''
         A__		 =					[]
         A__		 =					[]
         A__		 =					dqa_pipeline(image=a__ ,						question=a__ ,						words=a__ ,						boxes=a__ ,						top_k=2)
         self.assertEqual(a__ ,						[])




     @slow
     @require_torch
     @require_detectrona
     @require_pytesseract
     def 		snake_case_		(				self):
         A__		 =					pipeline(
             '''document-question-answering''' ,						model='''tiennvcs/layoutlmv2-base-uncased-finetuned-docvqa''' ,						revision='''9977165''' ,						)
         A__		 =					INVOICE_URL
         A__		 =					'''What is the invoice number?'''

         A__		 =					dqa_pipeline(image=a__ ,						question=a__ ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 {'''score''': 0.9_9_4_4, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                 {'''score''': 0.0_0_0_9, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
             ] ,						)

         A__		 =					dqa_pipeline({'''image''': image, '''question''': question} ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 {'''score''': 0.9_9_4_4, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                 {'''score''': 0.0_0_0_9, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
             ] ,						)

         A__		 =					dqa_pipeline(
             [{'''image''': image, '''question''': question}, {'''image''': image, '''question''': question}] ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 [
                     {'''score''': 0.9_9_4_4, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                     {'''score''': 0.0_0_0_9, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                 ],
             ]
             * 2 ,						)




     @slow
     @require_torch
     @require_detectrona
     @require_pytesseract
     def 		snake_case_		(				self):
         A__		 =					pipeline(
             '''document-question-answering''' ,						model='''tiennvcs/layoutlmv2-base-uncased-finetuned-docvqa''' ,						revision='''9977165''' ,						max_seq_len=5_0 ,						)
         A__		 =					INVOICE_URL
         A__		 =					'''What is the invoice number?'''

         A__		 =					dqa_pipeline(image=a__ ,						question=a__ ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 {'''score''': 0.9_9_7_4, '''answer''': '''1110212019''', '''start''': 2_3, '''end''': 2_3},
                 {'''score''': 0.9_9_4_8, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
             ] ,						)

         A__		 =					dqa_pipeline({'''image''': image, '''question''': question} ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 {'''score''': 0.9_9_7_4, '''answer''': '''1110212019''', '''start''': 2_3, '''end''': 2_3},
                 {'''score''': 0.9_9_4_8, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
             ] ,						)

         A__		 =					dqa_pipeline(
             [{'''image''': image, '''question''': question}, {'''image''': image, '''question''': question}] ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 [
                     {'''score''': 0.9_9_7_4, '''answer''': '''1110212019''', '''start''': 2_3, '''end''': 2_3},
                     {'''score''': 0.9_9_4_8, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                 ]
             ]
             * 2 ,						)




     @slow
     @require_torch
     @require_pytesseract
     @require_vision
     def 		snake_case_		(				self):
         A__		 =					AutoTokenizer.from_pretrained(
             '''impira/layoutlm-document-qa''' ,						revision='''3dc6de3''' ,						add_prefix_space=a__)
         A__		 =					pipeline(
             '''document-question-answering''' ,						model='''impira/layoutlm-document-qa''' ,						tokenizer=a__ ,						revision='''3dc6de3''' ,						)
         A__		 =					INVOICE_URL
         A__		 =					'''What is the invoice number?'''

         A__		 =					dqa_pipeline(image=a__ ,						question=a__ ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 {'''score''': 0.4_2_5_1, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                 {'''score''': 0.0_8_1_9, '''answer''': '''1110212019''', '''start''': 2_3, '''end''': 2_3},
             ] ,						)

         A__		 =					dqa_pipeline({'''image''': image, '''question''': question} ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 {'''score''': 0.4_2_5_1, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                 {'''score''': 0.0_8_1_9, '''answer''': '''1110212019''', '''start''': 2_3, '''end''': 2_3},
             ] ,						)

         A__		 =					dqa_pipeline(
             [{'''image''': image, '''question''': question}, {'''image''': image, '''question''': question}] ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 [
                     {'''score''': 0.4_2_5_1, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                     {'''score''': 0.0_8_1_9, '''answer''': '''1110212019''', '''start''': 2_3, '''end''': 2_3},
                 ]
             ]
             * 2 ,						)

         A__		 =					list(zip(*apply_tesseract(load_image(a__) ,						a__ ,						'''''')))

         # This model should also work if `image` is set to None
         A__		 =					dqa_pipeline({'''image''': None, '''word_boxes''': word_boxes, '''question''': question} ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 {'''score''': 0.4_2_5_1, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                 {'''score''': 0.0_8_1_9, '''answer''': '''1110212019''', '''start''': 2_3, '''end''': 2_3},
             ] ,						)




     @slow
     @require_torch
     @require_pytesseract
     @require_vision
     def 		snake_case_		(				self):
         A__		 =					AutoTokenizer.from_pretrained(
             '''impira/layoutlm-document-qa''' ,						revision='''3dc6de3''' ,						add_prefix_space=a__)
         A__		 =					pipeline(
             '''document-question-answering''' ,						model='''impira/layoutlm-document-qa''' ,						tokenizer=a__ ,						revision='''3dc6de3''' ,						max_seq_len=5_0 ,						)
         A__		 =					INVOICE_URL
         A__		 =					'''What is the invoice number?'''

         A__		 =					dqa_pipeline(image=a__ ,						question=a__ ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 {'''score''': 0.9_9_9_9, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                 {'''score''': 0.9_9_9_8, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
             ] ,						)

         A__		 =					dqa_pipeline(
             [{'''image''': image, '''question''': question}, {'''image''': image, '''question''': question}] ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 [
                     {'''score''': 0.9_9_9_9, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                     {'''score''': 0.9_9_9_8, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                 ]
             ]
             * 2 ,						)

         A__		 =					list(zip(*apply_tesseract(load_image(a__) ,						a__ ,						'''''')))

         # This model should also work if `image` is set to None
         A__		 =					dqa_pipeline({'''image''': None, '''word_boxes''': word_boxes, '''question''': question} ,						top_k=2)
         self.assertEqual(
             nested_simplify(a__ ,						decimals=4) ,						[
                 {'''score''': 0.9_9_9_9, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
                 {'''score''': 0.9_9_9_8, '''answer''': '''us-001''', '''start''': 1_6, '''end''': 1_6},
             ] ,						)




     @slow
     @require_torch
     def 		snake_case_		(				self):
         A__		 =					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__		 =					INVOICE_URL
         A__		 =					'''What is the invoice number?'''
         A__		 =					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 		snake_case_		(				self):
         pass





import torch

from diffusers import DDPMScheduler

from .test_schedulers import SchedulerCommonTest



class     _UpperCAmelCase			(       A__     ):
     UpperCamelCase__		    =				(DDPMScheduler,)




     def 		snake_case_		(				self ,						**a__):
         A__		 =					{
             '''num_train_timesteps''': 1_0_0_0,
             '''beta_start''': 0.0_0_0_1,
             '''beta_end''': 0.0_2,
             '''beta_schedule''': '''linear''',
             '''variance_type''': '''fixed_small''',
             '''clip_sample''': True,
         }

         config.update(**a__)
         return config




     def 		snake_case_		(				self):
         for timesteps in [1, 5, 1_0_0, 1_0_0_0]:
             self.check_over_configs(num_train_timesteps=a__)




     def 		snake_case_		(				self):
         for beta_start, beta_end in zip([0.0_0_0_1, 0.0_0_1, 0.0_1, 0.1] ,						[0.0_0_2, 0.0_2, 0.2, 2]):
             self.check_over_configs(beta_start=a__ ,						beta_end=a__)




     def 		snake_case_		(				self):
         for schedule in ["linear", "squaredcos_cap_v2"]:
             self.check_over_configs(beta_schedule=a__)




     def 		snake_case_		(				self):
         for variance in ["fixed_small", "fixed_large", "other"]:
             self.check_over_configs(variance_type=a__)




     def 		snake_case_		(				self):
         for clip_sample in [True, False]:
             self.check_over_configs(clip_sample=a__)




     def 		snake_case_		(				self):
         self.check_over_configs(thresholding=a__)
         for threshold in [0.5, 1.0, 2.0]:
             for prediction_type in ["epsilon", "sample", "v_prediction"]:
                 self.check_over_configs(
                     thresholding=a__ ,						prediction_type=a__ ,						sample_max_value=a__ ,						)




     def 		snake_case_		(				self):
         for prediction_type in ["epsilon", "sample", "v_prediction"]:
             self.check_over_configs(prediction_type=a__)




     def 		snake_case_		(				self):
         for t in [0, 5_0_0, 9_9_9]:
             self.check_over_forward(time_step=a__)




     def 		snake_case_		(				self):
         A__		 =					self.scheduler_classes[0]
         A__		 =					self.get_scheduler_config()
         A__		 =					scheduler_class(**a__)

         assert torch.sum(torch.abs(scheduler._get_variance(0) - 0.0)) < 1e-5
         assert torch.sum(torch.abs(scheduler._get_variance(4_8_7) - 0.0_0_9_7_9)) < 1e-5
         assert torch.sum(torch.abs(scheduler._get_variance(9_9_9) - 0.0_2)) < 1e-5




     def 		snake_case_		(				self):
         A__		 =					self.scheduler_classes[0]
         A__		 =					self.get_scheduler_config()
         A__		 =					scheduler_class(**a__)

         A__		 =					len(a__)

         A__		 =					self.dummy_model()
         A__		 =					self.dummy_sample_deter
         A__		 =					torch.manual_seed(0)

         for t in reversed(range(a__)):
             # 1. predict noise residual
             A__		 =					model(a__ ,						a__)

             # 2. predict previous mean of sample x_t-1
             A__		 =					scheduler.step(a__ ,						a__ ,						a__ ,						generator=a__).prev_sample

             # if t > 0:
             #     noise = self.dummy_sample_deter
             #     variance = scheduler.get_variance(t) ** (0.5) * noise
             #
             # sample = pred_prev_sample + variance
             A__		 =					pred_prev_sample

         A__		 =					torch.sum(torch.abs(a__))
         A__		 =					torch.mean(torch.abs(a__))

         assert abs(result_sum.item() - 2_5_8.9_6_0_6) < 1e-2
         assert abs(result_mean.item() - 0.3_3_7_2) < 1e-3




     def 		snake_case_		(				self):
         A__		 =					self.scheduler_classes[0]
         A__		 =					self.get_scheduler_config(prediction_type='''v_prediction''')
         A__		 =					scheduler_class(**a__)

         A__		 =					len(a__)

         A__		 =					self.dummy_model()
         A__		 =					self.dummy_sample_deter
         A__		 =					torch.manual_seed(0)

         for t in reversed(range(a__)):
             # 1. predict noise residual
             A__		 =					model(a__ ,						a__)

             # 2. predict previous mean of sample x_t-1
             A__		 =					scheduler.step(a__ ,						a__ ,						a__ ,						generator=a__).prev_sample

             # if t > 0:
             #     noise = self.dummy_sample_deter
             #     variance = scheduler.get_variance(t) ** (0.5) * noise
             #
             # sample = pred_prev_sample + variance
             A__		 =					pred_prev_sample

         A__		 =					torch.sum(torch.abs(a__))
         A__		 =					torch.mean(torch.abs(a__))

         assert abs(result_sum.item() - 2_0_2.0_2_9_6) < 1e-2
         assert abs(result_mean.item() - 0.2_6_3_1) < 1e-3




     def 		snake_case_		(				self):
         A__		 =					self.scheduler_classes[0]
         A__		 =					self.get_scheduler_config()
         A__		 =					scheduler_class(**a__)

         A__		 =					[1_0_0, 8_7, 5_0, 1, 0]

         scheduler.set_timesteps(timesteps=a__)

         A__		 =					scheduler.timesteps

         for i, timestep in enumerate(a__):
             if i == len(a__) - 1:
                 A__		 =					-1
             else:
                 A__		 =					timesteps[i + 1]

             A__		 =					scheduler.previous_timestep(a__)
             A__		 =					prev_t.item()

             self.assertEqual(a__ ,						a__)




     def 		snake_case_		(				self):
         A__		 =					self.scheduler_classes[0]
         A__		 =					self.get_scheduler_config()
         A__		 =					scheduler_class(**a__)

         A__		 =					[1_0_0, 8_7, 5_0, 5_1, 0]

         with self.assertRaises(a__ ,						msg='''`custom_timesteps` must be in descending order.'''):
             scheduler.set_timesteps(timesteps=a__)




     def 		snake_case_		(				self):
         A__		 =					self.scheduler_classes[0]
         A__		 =					self.get_scheduler_config()
         A__		 =					scheduler_class(**a__)

         A__		 =					[1_0_0, 8_7, 5_0, 1, 0]
         A__		 =					len(a__)

         with self.assertRaises(a__ ,						msg='''Can only pass one of `num_inference_steps` or `custom_timesteps`.'''):
             scheduler.set_timesteps(num_inference_steps=a__ ,						timesteps=a__)




     def 		snake_case_		(				self):
         A__		 =					self.scheduler_classes[0]
         A__		 =					self.get_scheduler_config()
         A__		 =					scheduler_class(**a__)

         A__		 =					[scheduler.config.num_train_timesteps]

         with self.assertRaises(
             a__ ,						msg='''`timesteps` must start before `self.config.train_timesteps`: {scheduler.config.num_train_timesteps}}''' ,						):
             scheduler.set_timesteps(timesteps=a__)
