






"""simple docstring"""







import json
import os
from pathlib import Path

import pytest

from datasets.download.download_config import DownloadConfig
from datasets.download.download_manager import DownloadManager
from datasets.utils.file_utils import hash_url_to_filename


lowerCAmelCase_	:    Any            =      '''http://www.mocksite.com/file1.txt'''
lowerCAmelCase_	:    Dict            =      '''"text": ["foo", "foo"]'''
lowerCAmelCase_	:    List[str]            =      '''6d8ce9aa78a471c7477201efbeabd3bb01ac2e7d100a6dc024ba1608361f90a8'''





class   UpperCamelCase_ :
  _A       :					Dict				=	200
  _A       :					Dict				=	{'Content-Length': '100'}
  _A       :					Any				=	{}



  def 		UpperCamelCase_				(      self ,		**snake_case__				)			->			List[Any]:



         """simple docstring"""





         return [bytes(_a ,		"""utf-8"""				)]






def       _lowerCAmelCase       (      *lowerCAmelCase	,      **lowerCAmelCase		):







       '''simple docstring'''
       return MockResponse()






@pytest.mark.parametrize("""urls_type"""	,      [str, list, dict]		)
def       _lowerCAmelCase       (      lowerCAmelCase	,      lowerCAmelCase	,      lowerCAmelCase		):







       '''simple docstring'''
       import requests

       monkeypatch.setattr(lowerCAmelCase	,      """request"""	,      lowerCAmelCase		)

       UpperCAmelCase	    =			URL
       if issubclass(lowerCAmelCase	,      lowerCAmelCase		):
              UpperCAmelCase	    =			url
       elif issubclass(lowerCAmelCase	,      lowerCAmelCase		):
              UpperCAmelCase	    =			[url]
       elif issubclass(lowerCAmelCase	,      lowerCAmelCase		):
              UpperCAmelCase	    =			{"""train""": url}
       UpperCAmelCase	    =			"""dummy"""
       UpperCAmelCase	    =			"""downloads"""
       UpperCAmelCase	    =			tmp_path
       UpperCAmelCase	    =			DownloadConfig(
           cache_dir=os.path.join(lowerCAmelCase	,      lowerCAmelCase		)	,      use_etag=lowerCAmelCase	,      )
       UpperCAmelCase	    =			DownloadManager(dataset_name=lowerCAmelCase	,      download_config=lowerCAmelCase		)
       UpperCAmelCase	    =			dl_manager.download(lowerCAmelCase		)
       UpperCAmelCase	    =			urls
       for downloaded_paths in [downloaded_paths]:
              if isinstance(lowerCAmelCase	,      lowerCAmelCase		):
                     UpperCAmelCase	    =			[downloaded_paths]
                     UpperCAmelCase	    =			[urls]
              elif isinstance(lowerCAmelCase	,      lowerCAmelCase		):
                     assert "train" in downloaded_paths.keys()
                     UpperCAmelCase	    =			downloaded_paths.values()
                     UpperCAmelCase	    =			urls.values()
              assert downloaded_paths
              for downloaded_path, input_url in zip(lowerCAmelCase	,      lowerCAmelCase		):
                     assert downloaded_path == dl_manager.downloaded_paths[input_url]
                     UpperCAmelCase	    =			Path(lowerCAmelCase		)
                     UpperCAmelCase	    =			downloaded_path.parts
                     assert parts[-1] == HASH
                     assert parts[-2] == cache_subdir
                     assert downloaded_path.exists()
                     UpperCAmelCase	    =			downloaded_path.read_text()
                     assert content == CONTENT
                     UpperCAmelCase	    =			downloaded_path.with_suffix(""".json"""		)
                     assert metadata_downloaded_path.exists()
                     UpperCAmelCase	    =			json.loads(metadata_downloaded_path.read_text()		)
                     assert metadata_content == {"url": URL, "etag": None}






@pytest.mark.parametrize("""paths_type"""	,      [str, list, dict]		)
def       _lowerCAmelCase       (      lowerCAmelCase	,      lowerCAmelCase	,      lowerCAmelCase		):







       '''simple docstring'''
       UpperCAmelCase	    =			str(lowerCAmelCase		)
       if issubclass(lowerCAmelCase	,      lowerCAmelCase		):
              UpperCAmelCase	    =			filename
       elif issubclass(lowerCAmelCase	,      lowerCAmelCase		):
              UpperCAmelCase	    =			[filename]
       elif issubclass(lowerCAmelCase	,      lowerCAmelCase		):
              UpperCAmelCase	    =			{"""train""": filename}
       UpperCAmelCase	    =			"""dummy"""
       UpperCAmelCase	    =			xz_file.parent
       UpperCAmelCase	    =			"""extracted"""
       UpperCAmelCase	    =			DownloadConfig(
           cache_dir=lowerCAmelCase	,      use_etag=lowerCAmelCase	,      )
       UpperCAmelCase	    =			DownloadManager(dataset_name=lowerCAmelCase	,      download_config=lowerCAmelCase		)
       UpperCAmelCase	    =			dl_manager.extract(lowerCAmelCase		)
       UpperCAmelCase	    =			paths
       for extracted_paths in [extracted_paths]:
              if isinstance(lowerCAmelCase	,      lowerCAmelCase		):
                     UpperCAmelCase	    =			[extracted_paths]
                     UpperCAmelCase	    =			[paths]
              elif isinstance(lowerCAmelCase	,      lowerCAmelCase		):
                     assert "train" in extracted_paths.keys()
                     UpperCAmelCase	    =			extracted_paths.values()
                     UpperCAmelCase	    =			paths.values()
              assert extracted_paths
              for extracted_path, input_path in zip(lowerCAmelCase	,      lowerCAmelCase		):
                     assert extracted_path == dl_manager.extracted_paths[input_path]
                     UpperCAmelCase	    =			Path(lowerCAmelCase		)
                     UpperCAmelCase	    =			extracted_path.parts
                     assert parts[-1] == hash_url_to_filename(lowerCAmelCase	,      etag=lowerCAmelCase		)
                     assert parts[-2] == extracted_subdir
                     assert extracted_path.exists()
                     UpperCAmelCase	    =			extracted_path.read_text()
                     UpperCAmelCase	    =			text_file.read_text()
                     assert extracted_file_content == expected_file_content






def       _lowerCAmelCase       (      lowerCAmelCase	,      lowerCAmelCase		):







       '''simple docstring'''
       assert path.endswith(""".jsonl"""		)
       for num_items, line in enumerate(lowerCAmelCase	,      start=1		):
              UpperCAmelCase	    =			json.loads(line.decode("""utf-8"""		)		)
              assert item.keys() == {"col_1", "col_2", "col_3"}
       assert num_items == 4






@pytest.mark.parametrize("""archive_jsonl"""	,      ["""tar_jsonl_path""", """zip_jsonl_path"""]		)
def       _lowerCAmelCase       (      lowerCAmelCase	,      lowerCAmelCase		):







       '''simple docstring'''
       UpperCAmelCase	    =			request.getfixturevalue(lowerCAmelCase		)
       UpperCAmelCase	    =			DownloadManager()
       for num_jsonl, (path, file) in enumerate(dl_manager.iter_archive(lowerCAmelCase		)	,      start=1		):
              _test_jsonl(lowerCAmelCase	,      lowerCAmelCase		)
       assert num_jsonl == 2






@pytest.mark.parametrize("""archive_nested_jsonl"""	,      ["""tar_nested_jsonl_path""", """zip_nested_jsonl_path"""]		)
def       _lowerCAmelCase       (      lowerCAmelCase	,      lowerCAmelCase		):







       '''simple docstring'''
       UpperCAmelCase	    =			request.getfixturevalue(lowerCAmelCase		)
       UpperCAmelCase	    =			DownloadManager()
       for num_tar, (path, file) in enumerate(dl_manager.iter_archive(lowerCAmelCase		)	,      start=1		):
              for num_jsonl, (subpath, subfile) in enumerate(dl_manager.iter_archive(lowerCAmelCase		)	,      start=1		):
                     _test_jsonl(lowerCAmelCase	,      lowerCAmelCase		)
       assert num_tar == 1
       assert num_jsonl == 2






def       _lowerCAmelCase       (      lowerCAmelCase		):







       '''simple docstring'''
       UpperCAmelCase	    =			DownloadManager()
       for num_file, file in enumerate(dl_manager.iter_files(lowerCAmelCase		)	,      start=1		):
              assert os.path.basename(lowerCAmelCase		) == ("test.txt" if num_file == 1 else "train.txt")
       assert num_file == 2








"""simple docstring"""







import tempfile

import numpy as np
import torch
from transformers import AutoTokenizer, TaEncoderModel

from diffusers import DDPMScheduler, UNetaDConditionModel
from diffusers.models.attention_processor import AttnAddedKVProcessor
from diffusers.pipelines.deepfloyd_if import IFWatermarker
from diffusers.utils.testing_utils import torch_device

from ..test_pipelines_common import to_np





class   UpperCamelCase_ :



  def 		UpperCamelCase_				(      self				)			->			Any:



         """simple docstring"""





         torch.manual_seed(0				)
         UpperCAmelCase	    =			TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5"""				)

         torch.manual_seed(0				)
         UpperCAmelCase	    =			AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5"""				)

         torch.manual_seed(0				)
         UpperCAmelCase	    =			UNetaDConditionModel(
             sample_size=32 ,		layers_per_block=1 ,		block_out_channels=[32, 64] ,		down_block_types=[
                 """ResnetDownsampleBlock2D""",
                 """SimpleCrossAttnDownBlock2D""",
             ] ,		mid_block_type="""UNetMidBlock2DSimpleCrossAttn""" ,		up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""] ,		in_channels=3 ,		out_channels=6 ,		cross_attention_dim=32 ,		encoder_hid_dim=32 ,		attention_head_dim=8 ,		addition_embed_type="""text""" ,		addition_embed_type_num_heads=2 ,		cross_attention_norm="""group_norm""" ,		resnet_time_scale_shift="""scale_shift""" ,		act_fn="""gelu""" ,		)
         unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

         torch.manual_seed(0				)
         UpperCAmelCase	    =			DDPMScheduler(
             num_train_timesteps=10_00 ,		beta_schedule="""squaredcos_cap_v2""" ,		beta_start=0.0_001 ,		beta_end=0.02 ,		thresholding=snake_case__ ,		dynamic_thresholding_ratio=0.95 ,		sample_max_value=1.0 ,		prediction_type="""epsilon""" ,		variance_type="""learned_range""" ,		)

         torch.manual_seed(0				)
         UpperCAmelCase	    =			IFWatermarker()

         return {
             "text_encoder": text_encoder,
             "tokenizer": tokenizer,
             "unet": unet,
             "scheduler": scheduler,
             "watermarker": watermarker,
             "safety_checker": None,
             "feature_extractor": None,
         }



  def 		UpperCamelCase_				(      self				)			->			int:



         """simple docstring"""





         torch.manual_seed(0				)
         UpperCAmelCase	    =			TaEncoderModel.from_pretrained("""hf-internal-testing/tiny-random-t5"""				)

         torch.manual_seed(0				)
         UpperCAmelCase	    =			AutoTokenizer.from_pretrained("""hf-internal-testing/tiny-random-t5"""				)

         torch.manual_seed(0				)
         UpperCAmelCase	    =			UNetaDConditionModel(
             sample_size=32 ,		layers_per_block=[1, 2] ,		block_out_channels=[32, 64] ,		down_block_types=[
                 """ResnetDownsampleBlock2D""",
                 """SimpleCrossAttnDownBlock2D""",
             ] ,		mid_block_type="""UNetMidBlock2DSimpleCrossAttn""" ,		up_block_types=["""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""] ,		in_channels=6 ,		out_channels=6 ,		cross_attention_dim=32 ,		encoder_hid_dim=32 ,		attention_head_dim=8 ,		addition_embed_type="""text""" ,		addition_embed_type_num_heads=2 ,		cross_attention_norm="""group_norm""" ,		resnet_time_scale_shift="""scale_shift""" ,		act_fn="""gelu""" ,		class_embed_type="""timestep""" ,		mid_block_scale_factor=1.414 ,		time_embedding_act_fn="""gelu""" ,		time_embedding_dim=32 ,		)
         unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

         torch.manual_seed(0				)
         UpperCAmelCase	    =			DDPMScheduler(
             num_train_timesteps=10_00 ,		beta_schedule="""squaredcos_cap_v2""" ,		beta_start=0.0_001 ,		beta_end=0.02 ,		thresholding=snake_case__ ,		dynamic_thresholding_ratio=0.95 ,		sample_max_value=1.0 ,		prediction_type="""epsilon""" ,		variance_type="""learned_range""" ,		)

         torch.manual_seed(0				)
         UpperCAmelCase	    =			DDPMScheduler(
             num_train_timesteps=10_00 ,		beta_schedule="""squaredcos_cap_v2""" ,		beta_start=0.0_001 ,		beta_end=0.02 ,		)

         torch.manual_seed(0				)
         UpperCAmelCase	    =			IFWatermarker()

         return {
             "text_encoder": text_encoder,
             "tokenizer": tokenizer,
             "unet": unet,
             "scheduler": scheduler,
             "image_noising_scheduler": image_noising_scheduler,
             "watermarker": watermarker,
             "safety_checker": None,
             "feature_extractor": None,
         }



  def 		UpperCamelCase_				(      self				)			->			Tuple:



         """simple docstring"""





         UpperCAmelCase	    =			self.get_dummy_components()
         UpperCAmelCase	    =			self.pipeline_class(**snake_case__				)
         pipe.to(snake_case__				)
         pipe.set_progress_bar_config(disable=snake_case__				)

         UpperCAmelCase	    =			self.get_dummy_inputs(snake_case__				)

         UpperCAmelCase	    =			inputs["""prompt"""]
         UpperCAmelCase	    =			inputs["""generator"""]
         UpperCAmelCase	    =			inputs["""num_inference_steps"""]
         UpperCAmelCase	    =			inputs["""output_type"""]

         if "image" in inputs:
                UpperCAmelCase	    =			inputs["""image"""]
         else:
                UpperCAmelCase	    =			None

         if "mask_image" in inputs:
                UpperCAmelCase	    =			inputs["""mask_image"""]
         else:
                UpperCAmelCase	    =			None

         if "original_image" in inputs:
                UpperCAmelCase	    =			inputs["""original_image"""]
         else:
                UpperCAmelCase	    =			None

         UpperCAmelCase    ,       UpperCAmelCase	    =			pipe.encode_prompt(snake_case__				)

         # inputs with prompt converted to embeddings
         UpperCAmelCase	    =			{
             """prompt_embeds""": prompt_embeds,
             """negative_prompt_embeds""": negative_prompt_embeds,
             """generator""": generator,
             """num_inference_steps""": num_inference_steps,
             """output_type""": output_type,
         }

         if image is not None:
                UpperCAmelCase	    =			image

         if mask_image is not None:
                UpperCAmelCase	    =			mask_image

         if original_image is not None:
                UpperCAmelCase	    =			original_image

         # set all optional components to None
         for optional_component in pipe._optional_components:
                setattr(snake_case__ ,		snake_case__ ,		snake_case__				)

         UpperCAmelCase	    =			pipe(**snake_case__				)[0]

         with tempfile.TemporaryDirectory() as tmpdir:
                pipe.save_pretrained(snake_case__				)
                UpperCAmelCase	    =			self.pipeline_class.from_pretrained(snake_case__				)
                pipe_loaded.to(snake_case__				)
                pipe_loaded.set_progress_bar_config(disable=snake_case__				)

         pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

         for optional_component in pipe._optional_components:
                self.assertTrue(
                    getattr(snake_case__ ,		snake_case__				) is None ,		f'''`{optional_component}` did not stay set to None after loading.''' ,		)

         UpperCAmelCase	    =			self.get_dummy_inputs(snake_case__				)

         UpperCAmelCase	    =			inputs["""generator"""]
         UpperCAmelCase	    =			inputs["""num_inference_steps"""]
         UpperCAmelCase	    =			inputs["""output_type"""]

         # inputs with prompt converted to embeddings
         UpperCAmelCase	    =			{
             """prompt_embeds""": prompt_embeds,
             """negative_prompt_embeds""": negative_prompt_embeds,
             """generator""": generator,
             """num_inference_steps""": num_inference_steps,
             """output_type""": output_type,
         }

         if image is not None:
                UpperCAmelCase	    =			image

         if mask_image is not None:
                UpperCAmelCase	    =			mask_image

         if original_image is not None:
                UpperCAmelCase	    =			original_image

         UpperCAmelCase	    =			pipe_loaded(**snake_case__				)[0]

         UpperCAmelCase	    =			np.abs(to_np(snake_case__				) - to_np(snake_case__				)				).max()
         self.assertLess(snake_case__ ,		1e-4				)



  def 		UpperCamelCase_				(      self				)			->			int:



         """simple docstring"""





         UpperCAmelCase	    =			self.get_dummy_components()
         UpperCAmelCase	    =			self.pipeline_class(**snake_case__				)
         pipe.to(snake_case__				)
         pipe.set_progress_bar_config(disable=snake_case__				)

         UpperCAmelCase	    =			self.get_dummy_inputs(snake_case__				)
         UpperCAmelCase	    =			pipe(**snake_case__				)[0]

         with tempfile.TemporaryDirectory() as tmpdir:
                pipe.save_pretrained(snake_case__				)
                UpperCAmelCase	    =			self.pipeline_class.from_pretrained(snake_case__				)
                pipe_loaded.to(snake_case__				)
                pipe_loaded.set_progress_bar_config(disable=snake_case__				)

         pipe_loaded.unet.set_attn_processor(AttnAddedKVProcessor()				)  # For reproducibility tests

         UpperCAmelCase	    =			self.get_dummy_inputs(snake_case__				)
         UpperCAmelCase	    =			pipe_loaded(**snake_case__				)[0]

         UpperCAmelCase	    =			np.abs(to_np(snake_case__				) - to_np(snake_case__				)				).max()
         self.assertLess(snake_case__ ,		1e-4				)
