



'''simple docstring'''

from dataclasses import dataclass
from typing import Optional

import numpy as np
import torch
import torch.nn as nn

from ..utils import BaseOutput, is_torch_version, randn_tensor
from .attention_processor import SpatialNorm
from .unet_ad_blocks import UNetMidBlockaD, get_down_block, get_up_block

@dataclass
class       __magic_name__		(       _UpperCamelCase  ):
							lowerCAmelCase			:       torch.FloatTensor

class       __magic_name__		(       nn.Module  ):







							def __init__(				self   :			Union[str, Any]      ,_UpperCAmelCase   :			int=3      ,_UpperCAmelCase   :			Tuple=3      ,_UpperCAmelCase   :			int=("DownEncoderBlock2D",)      ,_UpperCAmelCase   :			Dict=(64,)      ,_UpperCAmelCase   :			List[str]=2      ,_UpperCAmelCase   :			List[Any]=32      ,_UpperCAmelCase   :			int="silu"      ,_UpperCAmelCase   :			List[Any]=True      ,):
									super().__init__()
									_a	: List[Any]								=				layers_per_block

									_a	: List[Any]								=				torch.nn.Convad(
									    _UpperCAmelCase      ,block_out_channels[0]      ,kernel_size=3      ,stride=1      ,padding=1      ,)

									_a	: Union[str, Any]								=				None
									_a	: str								=				nn.ModuleList([]						)

									# down
									_a	: str								=				block_out_channels[0]
									for i, down_block_type in enumerate(_UpperCAmelCase						):
											_a	: Dict								=				output_channel
											_a	: List[Any]								=				block_out_channels[i]
											_a	: Any								=				i == len(_UpperCAmelCase						) - 1

											_a	: Tuple								=				get_down_block(
											    _UpperCAmelCase      ,num_layers=self.layers_per_block      ,in_channels=_UpperCAmelCase      ,out_channels=_UpperCAmelCase      ,add_downsample=not is_final_block      ,resnet_eps=1E-6      ,downsample_padding=0      ,resnet_act_fn=_UpperCAmelCase      ,resnet_groups=_UpperCAmelCase      ,attention_head_dim=_UpperCAmelCase      ,temb_channels=_UpperCAmelCase      ,)
											self.down_blocks.append(_UpperCAmelCase						)

									# mid
									_a	: List[Any]								=				UNetMidBlockaD(
									    in_channels=block_out_channels[-1]      ,resnet_eps=1E-6      ,resnet_act_fn=_UpperCAmelCase      ,output_scale_factor=1      ,resnet_time_scale_shift='default'      ,attention_head_dim=block_out_channels[-1]      ,resnet_groups=_UpperCAmelCase      ,temb_channels=_UpperCAmelCase      ,)

									# out
									_a	: str								=				nn.GroupNorm(num_channels=block_out_channels[-1]      ,num_groups=_UpperCAmelCase      ,eps=1E-6						)
									_a	: Union[str, Any]								=				nn.SiLU()

									_a	: Any								=				2 * out_channels if double_z else out_channels
									_a	: Optional[Any]								=				nn.Convad(block_out_channels[-1]      ,_UpperCAmelCase      ,3      ,padding=1						)

									_a	: Tuple								=				False







							def 				__lowercase     (				self   :			str      ,_UpperCAmelCase   :			List[Any]						):
									_a	: Union[str, Any]								=				x
									_a	: Optional[int]								=				self.conv_in(_UpperCAmelCase						)

									if self.training and self.gradient_checkpointing:

											def create_custom_forward(_UpperCAmelCase   :			Any						):
													def custom_forward(*_UpperCAmelCase   :			Union[str, Any]						):
															return module(*_UpperCAmelCase						)

													return custom_forward

											# down
											if is_torch_version('>='      ,'1.11.0'						):
													for down_block in self.down_blocks:
															_a	: int								=				torch.utils.checkpoint.checkpoint(
															    create_custom_forward(_UpperCAmelCase						)      ,_UpperCAmelCase      ,use_reentrant=_UpperCAmelCase						)
													# middle
													_a	: Union[str, Any]								=				torch.utils.checkpoint.checkpoint(
													    create_custom_forward(self.mid_block						)      ,_UpperCAmelCase      ,use_reentrant=_UpperCAmelCase						)
											else:
													for down_block in self.down_blocks:
															_a	: List[str]								=				torch.utils.checkpoint.checkpoint(create_custom_forward(_UpperCAmelCase						)      ,_UpperCAmelCase						)
													# middle
													_a	: Union[str, Any]								=				torch.utils.checkpoint.checkpoint(create_custom_forward(self.mid_block						)      ,_UpperCAmelCase						)

									else:
											# down
											for down_block in self.down_blocks:
													_a	: str								=				down_block(_UpperCAmelCase						)

											# middle
											_a	: int								=				self.mid_block(_UpperCAmelCase						)

									# post-process
									_a	: str								=				self.conv_norm_out(_UpperCAmelCase						)
									_a	: List[Any]								=				self.conv_act(_UpperCAmelCase						)
									_a	: Union[str, Any]								=				self.conv_out(_UpperCAmelCase						)

									return sample

class       __magic_name__		(       nn.Module  ):







							def __init__(				self   :			Optional[int]      ,_UpperCAmelCase   :			Tuple=3      ,_UpperCAmelCase   :			List[str]=3      ,_UpperCAmelCase   :			Optional[int]=("UpDecoderBlock2D",)      ,_UpperCAmelCase   :			Dict=(64,)      ,_UpperCAmelCase   :			Tuple=2      ,_UpperCAmelCase   :			Any=32      ,_UpperCAmelCase   :			Optional[Any]="silu"      ,_UpperCAmelCase   :			Any="group"      ,):
									super().__init__()
									_a	: int								=				layers_per_block

									_a	: Optional[int]								=				nn.Convad(
									    _UpperCAmelCase      ,block_out_channels[-1]      ,kernel_size=3      ,stride=1      ,padding=1      ,)

									_a	: List[str]								=				None
									_a	: int								=				nn.ModuleList([]						)

									_a	: List[str]								=				in_channels if norm_type == 'spatial' else None

									# mid
									_a	: Tuple								=				UNetMidBlockaD(
									    in_channels=block_out_channels[-1]      ,resnet_eps=1E-6      ,resnet_act_fn=_UpperCAmelCase      ,output_scale_factor=1      ,resnet_time_scale_shift='default' if norm_type == 'group' else norm_type      ,attention_head_dim=block_out_channels[-1]      ,resnet_groups=_UpperCAmelCase      ,temb_channels=_UpperCAmelCase      ,)

									# up
									_a	: Optional[int]								=				list(reversed(_UpperCAmelCase						)						)
									_a	: str								=				reversed_block_out_channels[0]
									for i, up_block_type in enumerate(_UpperCAmelCase						):
											_a	: List[Any]								=				output_channel
											_a	: str								=				reversed_block_out_channels[i]

											_a	: Dict								=				i == len(_UpperCAmelCase						) - 1

											_a	: List[Any]								=				get_up_block(
											    _UpperCAmelCase      ,num_layers=self.layers_per_block + 1      ,in_channels=_UpperCAmelCase      ,out_channels=_UpperCAmelCase      ,prev_output_channel=_UpperCAmelCase      ,add_upsample=not is_final_block      ,resnet_eps=1E-6      ,resnet_act_fn=_UpperCAmelCase      ,resnet_groups=_UpperCAmelCase      ,attention_head_dim=_UpperCAmelCase      ,temb_channels=_UpperCAmelCase      ,resnet_time_scale_shift=_UpperCAmelCase      ,)
											self.up_blocks.append(_UpperCAmelCase						)
											_a	: Tuple								=				output_channel

									# out
									if norm_type == "spatial":
											_a	: Optional[int]								=				SpatialNorm(block_out_channels[0]      ,_UpperCAmelCase						)
									else:
											_a	: List[str]								=				nn.GroupNorm(num_channels=block_out_channels[0]      ,num_groups=_UpperCAmelCase      ,eps=1E-6						)
									_a	: Tuple								=				nn.SiLU()
									_a	: str								=				nn.Convad(block_out_channels[0]      ,_UpperCAmelCase      ,3      ,padding=1						)

									_a	: Any								=				False







							def 				__lowercase     (				self   :			int      ,_UpperCAmelCase   :			Dict      ,_UpperCAmelCase   :			Tuple=None						):
									_a	: List[Any]								=				z
									_a	: List[str]								=				self.conv_in(_UpperCAmelCase						)

									_a	: Optional[Any]								=				next(iter(self.up_blocks.parameters()						)						).dtype
									if self.training and self.gradient_checkpointing:

											def create_custom_forward(_UpperCAmelCase   :			List[str]						):
													def custom_forward(*_UpperCAmelCase   :			List[str]						):
															return module(*_UpperCAmelCase						)

													return custom_forward

											if is_torch_version('>='      ,'1.11.0'						):
													# middle
													_a	: Optional[int]								=				torch.utils.checkpoint.checkpoint(
													    create_custom_forward(self.mid_block						)      ,_UpperCAmelCase      ,_UpperCAmelCase      ,use_reentrant=_UpperCAmelCase						)
													_a	: Optional[int]								=				sample.to(_UpperCAmelCase						)

													# up
													for up_block in self.up_blocks:
															_a	: Any								=				torch.utils.checkpoint.checkpoint(
															    create_custom_forward(_UpperCAmelCase						)      ,_UpperCAmelCase      ,_UpperCAmelCase      ,use_reentrant=_UpperCAmelCase						)
											else:
													# middle
													_a	: Dict								=				torch.utils.checkpoint.checkpoint(
													    create_custom_forward(self.mid_block						)      ,_UpperCAmelCase      ,_UpperCAmelCase						)
													_a	: Union[str, Any]								=				sample.to(_UpperCAmelCase						)

													# up
													for up_block in self.up_blocks:
															_a	: Union[str, Any]								=				torch.utils.checkpoint.checkpoint(create_custom_forward(_UpperCAmelCase						)      ,_UpperCAmelCase      ,_UpperCAmelCase						)
									else:
											# middle
											_a	: Union[str, Any]								=				self.mid_block(_UpperCAmelCase      ,_UpperCAmelCase						)
											_a	: Dict								=				sample.to(_UpperCAmelCase						)

											# up
											for up_block in self.up_blocks:
													_a	: Optional[Any]								=				up_block(_UpperCAmelCase      ,_UpperCAmelCase						)

        # post-process
									if latent_embeds is None:
											_a	: str								=				self.conv_norm_out(_UpperCAmelCase						)
									else:
											_a	: Dict								=				self.conv_norm_out(_UpperCAmelCase      ,_UpperCAmelCase						)
									_a	: Any								=				self.conv_act(_UpperCAmelCase						)
									_a	: Optional[int]								=				self.conv_out(_UpperCAmelCase						)

									return sample

class       __magic_name__		(       nn.Module  ):







							def __init__(				self   :			Optional[Any]      ,_UpperCAmelCase   :			Optional[Any]      ,_UpperCAmelCase   :			List[str]      ,_UpperCAmelCase   :			Dict      ,_UpperCAmelCase   :			Optional[int]=None      ,_UpperCAmelCase   :			int="random"      ,_UpperCAmelCase   :			Tuple=False      ,_UpperCAmelCase   :			Any=True						):
									super().__init__()
									_a	: Union[str, Any]								=				n_e
									_a	: List[str]								=				vq_embed_dim
									_a	: Dict								=				beta
									_a	: Optional[Any]								=				legacy

									_a	: Union[str, Any]								=				nn.Embedding(self.n_e      ,self.vq_embed_dim						)
									self.embedding.weight.data.uniform_(-1.0 / self.n_e      ,1.0 / self.n_e						)

									_a	: Tuple								=				remap
									if self.remap is not None:
											self.register_buffer('used'      ,torch.tensor(np.load(self.remap						)						)						)
											_a	: List[Any]								=				self.used.shape[0]
											_a	: List[Any]								=				unknown_index  # "random" or "extra" or integer
											if self.unknown_index == "extra":
													_a	: Tuple								=				self.re_embed
													_a	: Tuple								=				self.re_embed + 1
											print(
											    F"""Remapping {self.n_e} indices to {self.re_embed} indices. """
											    F"""Using {self.unknown_index} for unknown indices."""						)
									else:
											_a	: str								=				n_e

									_a	: Optional[Any]								=				sane_index_shape







							def 				__lowercase     (				self   :			Optional[int]      ,_UpperCAmelCase   :			int						):
									_a	: List[str]								=				inds.shape
									assert len(_UpperCAmelCase						) > 1
									_a	: List[str]								=				inds.reshape(ishape[0]      ,-1						)
									_a	: Optional[int]								=				self.used.to(_UpperCAmelCase						)
									_a	: str								=				(inds[:, :, None] == used[None, None, ...]).long()
									_a	: str								=				match.argmax(-1						)
									_a	: Optional[int]								=				match.sum(2						) < 1
									if self.unknown_index == "random":
											_a	: Any								=				torch.randint(0      ,self.re_embed      ,size=new[unknown].shape						).to(device=new.device						)
									else:
											_a	: int								=				self.unknown_index
									return new.reshape(_UpperCAmelCase						)







							def 				__lowercase     (				self   :			List[str]      ,_UpperCAmelCase   :			Union[str, Any]						):
									_a	: List[str]								=				inds.shape
									assert len(_UpperCAmelCase						) > 1
									_a	: Dict								=				inds.reshape(ishape[0]      ,-1						)
									_a	: Optional[Any]								=				self.used.to(_UpperCAmelCase						)
									if self.re_embed > self.used.shape[0]:  # extra token
											_a	: Dict								=				0  # simply set to zero
									_a	: Dict								=				torch.gather(used[None, :][inds.shape[0] * [0], :]      ,1      ,_UpperCAmelCase						)
									return back.reshape(_UpperCAmelCase						)







							def 				__lowercase     (				self   :			Optional[int]      ,_UpperCAmelCase   :			List[str]						):
									# reshape z -> (batch, height, width, channel) and flatten
									_a	: List[Any]								=				z.permute(0      ,2      ,3      ,1						).contiguous()
									_a	: str								=				z.view(-1      ,self.vq_embed_dim						)

									# distances from z to embeddings e_j (z - e)^2 = z^2 + e^2 - 2 e * z
									_a	: Any								=				torch.argmin(torch.cdist(_UpperCAmelCase      ,self.embedding.weight						)      ,dim=1						)

									_a	: List[str]								=				self.embedding(_UpperCAmelCase						).view(z.shape						)
									_a	: str								=				None
									_a	: Dict								=				None

									# compute loss for embedding
									if not self.legacy:
											_a	: int								=				self.beta * torch.mean((z_q.detach() - z) ** 2						) + torch.mean((z_q - z.detach()) ** 2						)
									else:
											_a	: List[Any]								=				torch.mean((z_q.detach() - z) ** 2						) + self.beta * torch.mean((z_q - z.detach()) ** 2						)

									# preserve gradients
									_a	: Optional[Any]								=				z + (z_q - z).detach()

									# reshape back to match original input shape
									_a	: int								=				z_q.permute(0      ,3      ,1      ,2						).contiguous()

									if self.remap is not None:
											_a	: int								=				min_encoding_indices.reshape(z.shape[0]      ,-1						)  # add batch axis
											_a	: int								=				self.remap_to_used(_UpperCAmelCase						)
											_a	: Any								=				min_encoding_indices.reshape(-1      ,1						)  # flatten

									if self.sane_index_shape:
											_a	: Union[str, Any]								=				min_encoding_indices.reshape(z_q.shape[0]      ,z_q.shape[2]      ,z_q.shape[3]						)

									return z_q, loss, (perplexity, min_encodings, min_encoding_indices)







							def 				__lowercase     (				self   :			str      ,_UpperCAmelCase   :			Optional[int]      ,_UpperCAmelCase   :			int						):
									# shape specifying (batch, height, width, channel)
									if self.remap is not None:
											_a	: Union[str, Any]								=				indices.reshape(shape[0]      ,-1						)  # add batch axis
											_a	: int								=				self.unmap_to_all(_UpperCAmelCase						)
											_a	: Dict								=				indices.reshape(-1						)  # flatten again

									# get quantized latent vectors
									_a	: Optional[int]								=				self.embedding(_UpperCAmelCase						)

									if shape is not None:
											_a	: Any								=				z_q.view(_UpperCAmelCase						)
											# reshape back to match original input shape
											_a	: List[str]								=				z_q.permute(0      ,3      ,1      ,2						).contiguous()

									return z_q

class       __magic_name__		(       _UpperCamelCase  ):







							def __init__(				self   :			Tuple      ,_UpperCAmelCase   :			str      ,_UpperCAmelCase   :			Dict=False						):
									_a	: Union[str, Any]								=				parameters
									_a			,   _a	: List[str]								=				torch.chunk(_UpperCAmelCase      ,2      ,dim=1						)
									_a	: Tuple								=				torch.clamp(self.logvar      ,-30.0      ,20.0						)
									_a	: List[Any]								=				deterministic
									_a	: Any								=				torch.exp(0.5 * self.logvar						)
									_a	: Dict								=				torch.exp(self.logvar						)
									if self.deterministic:
											_a	: Dict								=				torch.zeros_like(
											    self.mean      ,device=self.parameters.device      ,dtype=self.parameters.dtype						)







							def 				__lowercase     (				self   :			List[Any]      ,_UpperCAmelCase   :			Optional[torch.Generator] = None						):
									# make sure sample is on the same device as the parameters and has same dtype
									_a	: Tuple								=				randn_tensor(
									    self.mean.shape      ,generator=_UpperCAmelCase      ,device=self.parameters.device      ,dtype=self.parameters.dtype						)
									_a	: int								=				self.mean + self.std * sample
									return x







							def 				__lowercase     (				self   :			Union[str, Any]      ,_UpperCAmelCase   :			List[Any]=None						):
									if self.deterministic:
											return torch.Tensor([0.0]						)
									else:
											if other is None:
													return 0.5 * torch.sum(torch.pow(self.mean      ,2						) + self.var - 1.0 - self.logvar      ,dim=[1, 2, 3]						)
											else:
													return 0.5 * torch.sum(
													    torch.pow(self.mean - other.mean      ,2						) / other.var
													    + self.var / other.var
													    - 1.0
													    - self.logvar
													    + other.logvar      ,dim=[1, 2, 3]      ,)







							def 				__lowercase     (				self   :			List[str]      ,_UpperCAmelCase   :			List[Any]      ,_UpperCAmelCase   :			Any=[1, 2, 3]						):
									if self.deterministic:
											return torch.Tensor([0.0]						)
									_a	: Optional[int]								=				np.log(2.0 * np.pi						)
									return 0.5 * torch.sum(logtwopi + self.logvar + torch.pow(sample - self.mean      ,2						) / self.var      ,dim=_UpperCAmelCase						)







							def 				__lowercase     (				self   :			Optional[int]						):
									return self.mean









'''simple docstring'''

# Lint as: python3
import os
import re
import urllib.parse
from pathlib import Path
from typing import Callable, List, Optional, Union
from zipfile import ZipFile

from ..utils.file_utils import cached_path, hf_github_url
from ..utils.logging import get_logger
from ..utils.version import Version


__lowerCAmelCase		     =  get_logger(__name__)

class       __magic_name__		:
							lowerCAmelCase			:       Optional[Any]  							=   'dummy_data'
							lowerCAmelCase			:       List[str]  							=   'datasets'
							lowerCAmelCase			:       str  							=   False







							def __init__(				self   :			Tuple      ,_UpperCAmelCase   :			str      ,_UpperCAmelCase   :			str      ,_UpperCAmelCase   :			Union[Version, str]      ,_UpperCAmelCase   :			Optional[str] = None      ,_UpperCAmelCase   :			bool = False      ,_UpperCAmelCase   :			bool = True      ,_UpperCAmelCase   :			Optional[List[Callable]] = None      ,):
									_a	: int								=				0
									_a	: str								=				dataset_name
									_a	: Optional[int]								=				cache_dir
									_a	: int								=				use_local_dummy_data
									_a	: Optional[int]								=				config
									# download_callbacks take a single url as input
									_a	: List[Callable]								=				download_callbacks or []
									# if False, it doesn't load existing files and it returns the paths of the dummy files relative
									# to the dummy_data zip file root
									_a	: str								=				load_existing_dummy_data

									# TODO(PVP, QL) might need to make this more general
									_a	: Optional[int]								=				str(_UpperCAmelCase						)
									# to be downloaded
									_a	: Union[str, Any]								=				None
									_a	: Dict								=				None







							@property
							def 				__lowercase     (				self   :			Any						):
									if self._dummy_file is None:
											_a	: Dict								=				self.download_dummy_data()
									return self._dummy_file







							@property
							def 				__lowercase     (				self   :			Optional[int]						):
									if self.config is not None:
											# structure is dummy / config_name / version_name
											return os.path.join('dummy'      ,self.config.name      ,self.version_name						)
									# structure is dummy / version_name
									return os.path.join('dummy'      ,self.version_name						)







							@property
							def 				__lowercase     (				self   :			Any						):
									return os.path.join(self.dummy_data_folder      ,'dummy_data.zip'						)







							def 				__lowercase     (				self   :			List[str]						):
									_a	: Dict								=				(
									    self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
									)

									_a	: Dict								=				cached_path(
									    _UpperCAmelCase      ,cache_dir=self.cache_dir      ,extract_compressed_file=_UpperCAmelCase      ,force_extract=_UpperCAmelCase						)

									return os.path.join(_UpperCAmelCase      ,self.dummy_file_name						)







							@property
							def 				__lowercase     (				self   :			List[str]						):
									return os.path.join(self.datasets_scripts_dir      ,self.dataset_name      ,self.dummy_zip_file						)







							@property
							def 				__lowercase     (				self   :			Optional[Any]						):
									if self._bucket_url is None:
											_a	: Any								=				hf_github_url(self.dataset_name      ,self.dummy_zip_file.replace(os.sep      ,'/'						)						)
									return self._bucket_url







							@property
							def 				__lowercase     (				self   :			int						):
									# return full path if its a dir
									if os.path.isdir(self.dummy_file						):
											return self.dummy_file
									# else cut off path to file -> example `xsum`.
									return "/".join(self.dummy_file.replace(os.sep      ,'/'						).split('/'						)[:-1]						)







							def 				__lowercase     (				self   :			Dict      ,_UpperCAmelCase   :			str      ,*_UpperCAmelCase   :			int						):
									if self.load_existing_dummy_data:
											# dummy data is downloaded and tested
											_a	: List[Any]								=				self.dummy_file
									else:
											# dummy data cannot be downloaded and only the path to dummy file is returned
											_a	: str								=				self.dummy_file_name

									# special case when data_url is a dict
									if isinstance(_UpperCAmelCase      ,_UpperCAmelCase						):
											return self.create_dummy_data_dict(_UpperCAmelCase      ,_UpperCAmelCase						)
									elif isinstance(_UpperCAmelCase      ,(list, tuple)						):
											return self.create_dummy_data_list(_UpperCAmelCase      ,_UpperCAmelCase						)
									else:
											return self.create_dummy_data_single(_UpperCAmelCase      ,_UpperCAmelCase						)







							def 				__lowercase     (				self   :			List[str]      ,_UpperCAmelCase   :			Optional[int]      ,*_UpperCAmelCase   :			Optional[int]						):
									return self.download_and_extract(_UpperCAmelCase						)







							def 				__lowercase     (				self   :			Tuple      ,_UpperCAmelCase   :			List[Any]      ,_UpperCAmelCase   :			Optional[Any]						):
									return self.download_and_extract(_UpperCAmelCase						)







							def 				__lowercase     (				self   :			List[str]      ,_UpperCAmelCase   :			List[str]      ,*_UpperCAmelCase   :			int      ,**_UpperCAmelCase   :			Dict						):
									return path







							def 				__lowercase     (				self   :			Optional[Any]						):
									return {}







							def 				__lowercase     (				self   :			Optional[Any]      ,_UpperCAmelCase   :			List[Any]      ,_UpperCAmelCase   :			str						):
									_a	: int								=				{}
									for key, single_urls in data_url.items():
											for download_callback in self.download_callbacks:
													if isinstance(_UpperCAmelCase      ,_UpperCAmelCase						):
															for single_url in single_urls:
																	download_callback(_UpperCAmelCase						)
													else:
															_a	: Union[str, Any]								=				single_urls
															download_callback(_UpperCAmelCase						)
            # we force the name of each key to be the last file / folder name of the url path
            # if the url has arguments, we need to encode them with urllib.parse.quote_plus
											if isinstance(_UpperCAmelCase      ,_UpperCAmelCase						):
													_a	: Optional[int]								=				[os.path.join(_UpperCAmelCase      ,urllib.parse.quote_plus(Path(_UpperCAmelCase						).name						)						) for x in single_urls]
											else:
													_a	: Tuple								=				single_urls
													_a	: Optional[Any]								=				os.path.join(_UpperCAmelCase      ,urllib.parse.quote_plus(Path(_UpperCAmelCase						).name						)						)
											_a	: Union[str, Any]								=				value

									# make sure that values are unique
									if all(isinstance(_UpperCAmelCase      ,_UpperCAmelCase						) for i in dummy_data_dict.values()						) and len(set(dummy_data_dict.values()						)						) < len(
									    dummy_data_dict.values()						):
											# append key to value to make its name unique
											_a	: Dict								=				{key: value + key for key, value in dummy_data_dict.items()}

									return dummy_data_dict







							def 				__lowercase     (				self   :			List[Any]      ,_UpperCAmelCase   :			List[str]      ,_UpperCAmelCase   :			List[str]						):
									_a	: str								=				[]
									# trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
									_a	: Tuple								=				all(bool(re.findall('[0-9]{3,}-of-[0-9]{3,}'      ,_UpperCAmelCase						)						) for url in data_url						)
									_a	: str								=				all(
									    url.startswith('https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/pubmed'						) for url in data_url						)
									if data_url and (is_tf_records or is_pubmed_records):
											_a	: str								=				[data_url[0]] * len(_UpperCAmelCase						)
									for single_url in data_url:
											for download_callback in self.download_callbacks:
													download_callback(_UpperCAmelCase						)
											# we force the name of each key to be the last file / folder name of the url path
											# if the url has arguments, we need to encode them with urllib.parse.quote_plus
											_a	: List[Any]								=				os.path.join(_UpperCAmelCase      ,urllib.parse.quote_plus(single_url.split('/'						)[-1]						)						)
											dummy_data_list.append(_UpperCAmelCase						)
									return dummy_data_list







							def 				__lowercase     (				self   :			List[Any]      ,_UpperCAmelCase   :			Union[str, Any]      ,_UpperCAmelCase   :			List[str]						):
									for download_callback in self.download_callbacks:
											download_callback(_UpperCAmelCase						)
									# we force the name of each key to be the last file / folder name of the url path
									# if the url has arguments, we need to encode them with urllib.parse.quote_plus
									_a	: List[Any]								=				os.path.join(_UpperCAmelCase      ,urllib.parse.quote_plus(data_url.split('/'						)[-1]						)						)
									if os.path.exists(_UpperCAmelCase						) or not self.load_existing_dummy_data:
											return value
									else:
											# Backward compatibility, maybe deprecate at one point.
											# For many datasets with single url calls to dl_manager.download_and_extract,
											# the dummy_data.zip file is actually the zipped downloaded file
											# while now we expected the dummy_data.zip file to be a directory containing
											# the downloaded file.
											return path_to_dummy_data







							def 				__lowercase     (				self   :			Tuple						):
									pass







							def 				__lowercase     (				self   :			Optional[int]						):
									pass







							def 				__lowercase     (				self   :			List[str]      ,_UpperCAmelCase   :			Dict						):
									def _iter_archive_members(_UpperCAmelCase   :			List[Any]						):
											# this preserves the order of the members inside the ZIP archive
											_a	: int								=				Path(self.dummy_file						).parent
											_a	: List[str]								=				path.relative_to(_UpperCAmelCase						)
											with ZipFile(self.local_path_to_dummy_data						) as zip_file:
													_a	: List[Any]								=				zip_file.namelist()
											for member in members:
													if member.startswith(relative_path.as_posix()						):
															yield dummy_parent_path.joinpath(_UpperCAmelCase						)

									_a	: Dict								=				Path(_UpperCAmelCase						)
									_a	: Dict								=				_iter_archive_members(_UpperCAmelCase						) if self.use_local_dummy_data else path.rglob('*'						)
									for file_path in file_paths:
											if file_path.is_file() and not file_path.name.startswith(('.', '__')						):
													yield file_path.relative_to(_UpperCAmelCase						).as_posix(), file_path.open('rb'						)







							def 				__lowercase     (				self   :			Any      ,_UpperCAmelCase   :			Union[str, Any]						):
									if not isinstance(_UpperCAmelCase      ,_UpperCAmelCase						):
											_a	: str								=				[paths]
									for path in paths:
											if os.path.isfile(_UpperCAmelCase						):
													if os.path.basename(_UpperCAmelCase						).startswith(('.', '__')						):
															return
													yield path
											else:
													for dirpath, dirnames, filenames in os.walk(_UpperCAmelCase						):
															if os.path.basename(_UpperCAmelCase						).startswith(('.', '__')						):
																	continue
															dirnames.sort()
															for filename in sorted(_UpperCAmelCase						):
																	if filename.startswith(('.', '__')						):
																			continue
																	yield os.path.join(_UpperCAmelCase      ,_UpperCAmelCase						)




