




'''simple docstring'''

import os
import sys
from pathlib import Path

import pytest

from datasets import Dataset, IterableDataset
from datasets.distributed import split_dataset_by_node

from .utils import execute_subprocess_async, get_torch_dist_unique_port, require_torch




def    snake_case__     (  )			->       Tuple:





			'''simple docstring'''






			snake_case__          =						Dataset.from_dict({"""i""": range(17	)}	)
			snake_case__          =						len(a	)
			snake_case__          =						3
			snake_case__          =						[
			    split_dataset_by_node(a       ,      rank=a       ,      world_size=a	) for rank in range(a	)
			]
			assert sum(len(a	) for ds in datasets_per_rank	) == full_size
			assert len({tuple(x.values()	) for ds in datasets_per_rank for x in ds}	) == full_size




def    snake_case__     (  )			->       List[Any]:





			'''simple docstring'''






			def gen():
						return ({"i": i} for i in range(17	))

			snake_case__          =						3
			snake_case__          =						IterableDataset.from_generator(a	)
			snake_case__          =						len(list(a	)	)
			snake_case__          =						[
			    split_dataset_by_node(a       ,      rank=a       ,      world_size=a	) for rank in range(a	)
			]
			assert sum(len(list(a	)	) for ds in datasets_per_rank	) == full_size
			assert len({tuple(x.values()	) for ds in datasets_per_rank for x in ds}	) == full_size




@pytest.mark.parametrize("""shards_per_node"""       ,      [1, 2, 3]	)
def    snake_case__     (  a	)			->       int:





			'''simple docstring'''






			def gen(a	):
						for shard in shards:
									yield from ({"i": i, "shard": shard} for i in range(17	))

			snake_case__          =						3
			snake_case__          =						shards_per_node * world_size
			snake_case__          =						{"""shards""": [F"""shard_{shard_idx}.txt""" for shard_idx in range(a	)]}
			snake_case__          =						IterableDataset.from_generator(a       ,      gen_kwargs=a	)
			snake_case__          =						len(list(a	)	)
			assert full_ds.n_shards == world_size * shards_per_node
			snake_case__          =						[
			    split_dataset_by_node(a       ,      rank=a       ,      world_size=a	) for rank in range(a	)
			]
			assert [ds.n_shards for ds in datasets_per_rank] == [shards_per_node] * world_size
			assert sum(len(list(a	)	) for ds in datasets_per_rank	) == full_size
			assert len({tuple(x.values()	) for ds in datasets_per_rank for x in ds}	) == full_size




def    snake_case__     (  )			->       List[str]:





			'''simple docstring'''






			def gen():
						return ({"i": i} for i in range(17	))

			snake_case__          =						2
			snake_case__          =						IterableDataset.from_generator(a	)
			snake_case__          =						len(list(a	)	)

			snake_case__          =						split_dataset_by_node(a       ,      rank=0       ,      world_size=a	).shuffle(seed=42	)
			assert len(list(a	)	) == 1 + full_size // world_size
			with pytest.raises(a	):
						split_dataset_by_node(a       ,      rank=0       ,      world_size=a	).shuffle()

			snake_case__          =						split_dataset_by_node(full_ds.shuffle(seed=42	)       ,      rank=0       ,      world_size=a	)
			assert len(list(a	)	) == 1 + full_size // world_size
			with pytest.raises(a	):
						split_dataset_by_node(full_ds.shuffle()       ,      rank=0       ,      world_size=a	)




@pytest.mark.parametrize("""streaming"""       ,      [False, True]	)
@require_torch
@pytest.mark.skipif(os.name == """nt"""       ,      reason="""execute_subprocess_async doesn't support windows"""	)
@pytest.mark.integration
def    snake_case__     (  a	)			->       str:





			'''simple docstring'''






			snake_case__          =						2
			snake_case__          =						get_torch_dist_unique_port()
			snake_case__          =						Path(a	).resolve().parent / """distributed_scripts""" / """launch_torch_distributed.py"""
			snake_case__          =						F"""
        -m torch.distributed.launch
        --nproc_per_node={nproc_per_node}
        --master_port={master_port}
        {test_script}
    """.split()
			snake_case__          =						F"""
        --streaming={streaming}
    """.split()
			snake_case__          =						[sys.executable] + distributed_args + args
			execute_subprocess_async(a       ,      env=os.environ.copy()	)




@pytest.mark.parametrize(
    """nproc_per_node, num_workers"""       ,      [
        (2, 2),  # each node has 2 shards and each worker has 1 shards
        (3, 2),  # each node uses all the shards but skips examples, and each worker has 2 shards
    ]       ,      )
@require_torch
@pytest.mark.skipif(os.name == """nt"""       ,      reason="""execute_subprocess_async doesn't support windows"""	)
@pytest.mark.integration
def    snake_case__     (  a       ,      a	)			->       int:





			'''simple docstring'''






			snake_case__          =						True
			snake_case__          =						get_torch_dist_unique_port()
			snake_case__          =						Path(a	).resolve().parent / """distributed_scripts""" / """launch_torch_distributed.py"""
			snake_case__          =						F"""
        -m torch.distributed.launch
        --nproc_per_node={nproc_per_node}
        --master_port={master_port}
        {test_script}
    """.split()
			snake_case__          =						F"""
        --streaming={streaming}
        --num_workers={num_workers}
    """.split()
			snake_case__          =						[sys.executable] + distributed_args + args
			execute_subprocess_async(a       ,      env=os.environ.copy()	)






import copy

from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ..auto.configuration_auto import AutoConfig
from ..clip.configuration_clip import CLIPVisionConfig


UpperCAmelCase_       : Union[str, Any]							=    logging.get_logger(__name__)





class   lowercase__     (				__A	):
  __UpperCamelCase						  =	"""vision-text-dual-encoder"""
  __UpperCamelCase						  =	True







  def __init__(						self     ,    _lowercase=512     ,    _lowercase=2.6592     ,    **_lowercase    ):
       super().__init__(**_lowercase    )

       if "vision_config" not in kwargs:
            raise ValueError("""`vision_config` can not be `None`."""    )

       if "text_config" not in kwargs:
            raise ValueError("""`text_config` can not be `None`."""    )

       lowerCAmelCase_       :							Dict			 =		kwargs.pop("""vision_config"""    )
       lowerCAmelCase_       :							Optional[int]			 =		kwargs.pop("""text_config"""    )

       lowerCAmelCase_       :							Tuple			 =		vision_config.pop("""model_type"""    )
       lowerCAmelCase_       :							str			 =		text_config.pop("""model_type"""    )

       if vision_model_type == "clip":
            lowerCAmelCase_       :							Dict			 =		AutoConfig.for_model(_lowercase     ,    **_lowercase    ).vision_config
       elif vision_model_type == "clip_vision_model":
            lowerCAmelCase_       :							Any			 =		CLIPVisionConfig(**_lowercase    )
       else:
            lowerCAmelCase_       :							Tuple			 =		AutoConfig.for_model(_lowercase     ,    **_lowercase    )

       lowerCAmelCase_       :							List[Any]			 =		AutoConfig.for_model(_lowercase     ,    **_lowercase    )

       lowerCAmelCase_       :							Union[str, Any]			 =		projection_dim
       lowerCAmelCase_       :							Optional[int]			 =		logit_scale_init_value







  @classmethod
  def 						UpperCAmelCase__		(						cls     ,    _lowercase     ,    _lowercase     ,    **_lowercase    ):
       return cls(vision_config=vision_config.to_dict()     ,    text_config=text_config.to_dict()     ,    **_lowercase    )







  def 						UpperCAmelCase__		(						self    ):
       lowerCAmelCase_       :							int			 =		copy.deepcopy(self.__dict__    )
       lowerCAmelCase_       :							List[str]			 =		self.vision_config.to_dict()
       lowerCAmelCase_       :							List[str]			 =		self.text_config.to_dict()
       lowerCAmelCase_       :							str			 =		self.__class__.model_type
       return output
