code: 
  '''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 	_UpperCAmelCase				(  )    ->						Optional[int]:
    _lowerCAmelCase       :  Tuple   				=					Dataset.from_dict({"""i""": range(17		)}		)
    _lowerCAmelCase       :  Optional[int]   				=					len(_A		)
    _lowerCAmelCase       :  List[str]   				=					3
    _lowerCAmelCase       :  Dict   				=					[
        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 	_UpperCAmelCase				(  )    ->						int:
  
  
    def gen():
      return ({"i": i} for i in range(17		))
  
    _lowerCAmelCase       :  Any   				=					3
    _lowerCAmelCase       :  str   				=					IterableDataset.from_generator(_A		)
    _lowerCAmelCase       :  Optional[int]   				=					len(list(_A		)		)
    _lowerCAmelCase       :  List[str]   				=					[
        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 	_UpperCAmelCase				(  _lowerCamelCase						:       Any		)    ->						Union[str, Any]:
  
  
    def gen(_lowerCamelCase						:       Dict		):
      for shard in shards:
        yield from ({"i": i, "shard": shard} for i in range(17		))
  
    _lowerCAmelCase       :  int   				=					3
    _lowerCAmelCase       :  Any   				=					shards_per_node * world_size
    _lowerCAmelCase       :  Optional[int]   				=					{"""shards""": [f'shard_{shard_idx}.txt' for shard_idx in range(_A		)]}
    _lowerCAmelCase       :  Tuple   				=					IterableDataset.from_generator(_A  ,      gen_kwargs=_A		)
    _lowerCAmelCase       :  List[str]   				=					len(list(_A		)		)
    assert full_ds.n_shards == world_size * shards_per_node
    _lowerCAmelCase       :  Tuple   				=					[
        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 	_UpperCAmelCase				(  )    ->						Any:
  
  
    def gen():
      return ({"i": i} for i in range(17		))
  
    _lowerCAmelCase       :  Any   				=					2
    _lowerCAmelCase       :  Optional[Any]   				=					IterableDataset.from_generator(_A		)
    _lowerCAmelCase       :  str   				=					len(list(_A		)		)
  
    _lowerCAmelCase       :  Dict   				=					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()
  
    _lowerCAmelCase       :  Union[str, Any]   				=					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 	_UpperCAmelCase				(  _lowerCamelCase						:       Optional[Any]		)    ->						Any:
    _lowerCAmelCase       :  Any   				=					2
    _lowerCAmelCase       :  Union[str, Any]   				=					get_torch_dist_unique_port()
    _lowerCAmelCase       :  Optional[int]   				=					Path(_A		).resolve().parent / """distributed_scripts""" / """launch_torch_distributed.py"""
    _lowerCAmelCase       :  Dict   				=					f'\n        -m torch.distributed.launch\n        --nproc_per_node={nproc_per_node}\n        --master_port={master_port}\n        {test_script}\n    '.split()
    _lowerCAmelCase       :  int   				=					f'\n        --streaming={streaming}\n    '.split()
    _lowerCAmelCase       :  Optional[Any]   				=					[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 	_UpperCAmelCase				(  _lowerCamelCase						:       Tuple  ,      _lowerCamelCase						:       Optional[int]		)    ->						Any:
    _lowerCAmelCase       :  Optional[Any]   				=					True
    _lowerCAmelCase       :  Dict   				=					get_torch_dist_unique_port()
    _lowerCAmelCase       :  Tuple   				=					Path(_A		).resolve().parent / """distributed_scripts""" / """launch_torch_distributed.py"""
    _lowerCAmelCase       :  Union[str, Any]   				=					f'\n        -m torch.distributed.launch\n        --nproc_per_node={nproc_per_node}\n        --master_port={master_port}\n        {test_script}\n    '.split()
    _lowerCAmelCase       :  Optional[int]   				=					f'\n        --streaming={streaming}\n        --num_workers={num_workers}\n    '.split()
    _lowerCAmelCase       :  Dict   				=					[sys.executable] + distributed_args + args
    execute_subprocess_async(_A  ,      env=os.environ.copy()		)
  
  
  
style_context: 
  '''simple docstring'''
  
  from cva import COLOR_BGR2GRAY, cvtColor, imread, imshow, waitKey
  from numpy import divide, inta, multiply, ravel, sort, zeros_like
  
  
  def 	_UpperCAmelCase				(  _lowerCamelCase						:       str  ,      _lowerCamelCase						:       List[Any]=3		)    ->						Union[str, Any]:
    _lowerCAmelCase       :  str   				=					int(mask / 2		)
    # copy image size
    _lowerCAmelCase       :  Dict   				=					zeros_like(_lowerCamelCase		)
    for i in range(_lowerCamelCase  ,      gray_img.shape[0] - bd		):
      for j in range(_lowerCamelCase  ,      gray_img.shape[1] - bd		):
        # get mask according with mask
        _lowerCAmelCase       :  Union[str, Any]   				=					ravel(gray_img[i - bd : i + bd + 1, j - bd : j + bd + 1]		)
        # calculate mask median
        _lowerCAmelCase       :  str   				=					sort(_lowerCamelCase		)[inta(divide((multiply(_lowerCamelCase  ,      _lowerCamelCase		))  ,      2		) + 1		)]
        _lowerCAmelCase       :  Dict   				=					median
    return median_img
  
  
  if __name__ == "__main__":
       # read original image
       UpperCamelCase_     			=       imread("""../image_data/lena.jpg""")
       # turn image in gray scale value
       UpperCamelCase_     			=       cvtColor(img, COLOR_BGR2GRAY)
  
       # get values with two different mask size
       UpperCamelCase_     			=       median_filter(gray, 3)
       UpperCamelCase_     			=       median_filter(gray, 5)
  
       # show result images
       imshow("""median filter with 3x3 mask""", medianaxa)
       imshow("""median filter with 5x5 mask""", medianaxa)
       waitKey(0)
  
  
  