


import importlib
import inspect
import json
import os
import re
import shutil
import sys
from pathlib import Path
from typing import Dict, Optional, Union
from urllib import request

from huggingface_hub import HfFolder, cached_download, hf_hub_download, model_info
from packaging import version

from .. import __version__
from . import DIFFUSERS_DYNAMIC_MODULE_NAME, HF_MODULES_CACHE, logging


UpperCAmelCase   :    str     =		(
    '''https://raw.githubusercontent.com/huggingface/diffusers/{revision}/examples/community/{pipeline}.py'''
)


UpperCAmelCase   :    str     =		logging.get_logger(__name__)  # pylint: disable=invalid-name






def 	__lowerCamelCase						(							):





			'''simple docstring'''




			lowerCamelCase    		=  """https://pypi.org/pypi/diffusers/json"""
			lowerCamelCase    		=  json.loads(request.urlopen(_lowerCamelCase      ).read()      )["""releases"""].keys()
			return sorted(_lowerCamelCase  ,							key=lambda lowerCamelCase__      : version.Version(_lowerCamelCase      )      )






def 	__lowerCamelCase						(							):





			'''simple docstring'''




			if HF_MODULES_CACHE in sys.path:
						return

			sys.path.append(_lowerCamelCase      )
			os.makedirs(_lowerCamelCase  ,							exist_ok=_lowerCamelCase      )
			lowerCamelCase    		=  Path(_lowerCamelCase      ) / """__init__.py"""
			if not init_path.exists():
						init_path.touch()






def 	__lowerCamelCase						(							lowerCamelCase__		:						Union[str, os.PathLike]      ):





			'''simple docstring'''




			init_hf_modules()
			lowerCamelCase    		=  Path(_lowerCamelCase      ) / name
			# If the parent module does not exist yet, recursively create it.
			if not dynamic_module_path.parent.exists():
						create_dynamic_module(dynamic_module_path.parent      )
			os.makedirs(_lowerCamelCase  ,							exist_ok=_lowerCamelCase      )
			lowerCamelCase    		=  dynamic_module_path / """__init__.py"""
			if not init_path.exists():
						init_path.touch()






def 	__lowerCamelCase						(							lowerCamelCase__		:						Dict      ):





			'''simple docstring'''




			with open(_lowerCamelCase  ,							"""r"""  ,							encoding="""utf-8"""      ) as f:
						lowerCamelCase    		=  f.read()

			# Imports of the form `import .xxx`
			lowerCamelCase    		=  re.findall("""^\s*import\s+\.(\S+)\s*$"""  ,							_lowerCamelCase  ,							flags=re.MULTILINE      )
			# Imports of the form `from .xxx import yyy`
			relative_imports += re.findall("""^\s*from\s+\.(\S+)\s+import"""  ,							_lowerCamelCase  ,							flags=re.MULTILINE      )
			# Unique-ify
			return list(set(_lowerCamelCase      )      )






def 	__lowerCamelCase						(							lowerCamelCase__		:						Dict      ):





			'''simple docstring'''




			lowerCamelCase    		=  False
			lowerCamelCase    		=  [module_file]
			lowerCamelCase    		=  []

			# Let's recurse through all relative imports
			while not no_change:
						lowerCamelCase    		=  []
						for f in files_to_check:
									new_imports.extend(get_relative_imports(_lowerCamelCase      )      )

						lowerCamelCase    		=  Path(_lowerCamelCase      ).parent
						lowerCamelCase    		=  [str(module_path / m      ) for m in new_imports]
						lowerCamelCase    		=  [f for f in new_import_files if f not in all_relative_imports]
						lowerCamelCase    		=  [f'{f}.py' for f in new_import_files]

						lowerCamelCase    		=  len(_lowerCamelCase      ) == 0
						all_relative_imports.extend(_lowerCamelCase      )

			return all_relative_imports






def 	__lowerCamelCase						(							lowerCamelCase__		:						Optional[Any]      ):





			'''simple docstring'''




			with open(_lowerCamelCase  ,							"""r"""  ,							encoding="""utf-8"""      ) as f:
						lowerCamelCase    		=  f.read()

			# Imports of the form `import xxx`
			lowerCamelCase    		=  re.findall("""^\s*import\s+(\S+)\s*$"""  ,							_lowerCamelCase  ,							flags=re.MULTILINE      )
			# Imports of the form `from xxx import yyy`
			imports += re.findall("""^\s*from\s+(\S+)\s+import"""  ,							_lowerCamelCase  ,							flags=re.MULTILINE      )
			# Only keep the top-level module
			lowerCamelCase    		=  [imp.split("""."""      )[0] for imp in imports if not imp.startswith("""."""      )]

			# Unique-ify and test we got them all
			lowerCamelCase    		=  list(set(_lowerCamelCase      )      )
			lowerCamelCase    		=  []
			for imp in imports:
						try:
									importlib.import_module(_lowerCamelCase      )
						except ImportError:
									missing_packages.append(_lowerCamelCase      )

			if len(_lowerCamelCase      ) > 0:
						raise ImportError(
						    """This modeling file requires the following packages that were not found in your environment: """
						    f'{", ".join(_lowerCamelCase      )}. Run `pip install {" ".join(_lowerCamelCase      )}`'      )

			return get_relative_imports(_lowerCamelCase      )






def 	__lowerCamelCase						(							lowerCamelCase__		:						str  ,							lowerCamelCase__		:						Dict      ):





			'''simple docstring'''




			lowerCamelCase    		=  module_path.replace(os.path.sep  ,							"""."""      )
			lowerCamelCase    		=  importlib.import_module(_lowerCamelCase      )

			if class_name is None:
						return find_pipeline_class(_lowerCamelCase      )
			return getattr(_lowerCamelCase  ,							_lowerCamelCase      )






def 	__lowerCamelCase						(							lowerCamelCase__		:						List[str]      ):





			'''simple docstring'''




			from ..pipelines import DiffusionPipeline

			lowerCamelCase    		=  dict(inspect.getmembers(_lowerCamelCase  ,							inspect.isclass      )      )

			lowerCamelCase    		=  None
			for cls_name, cls in cls_members.items():
						if (
						    cls_name != DiffusionPipeline.__name__
						    and issubclass(cls  ,							_lowerCamelCase      )
						    and cls.__module__.split("""."""      )[0] != "diffusers"
						):
									if pipeline_class is not None:
												raise ValueError(
												    f'Multiple classes that inherit from {DiffusionPipeline.__name__} have been found:'
												    f' {pipeline_class.__name__}, and {cls_name}. Please make sure to define only one in'
												    f' {loaded_module}.'      )
									lowerCamelCase    		=  cls

			return pipeline_class






def 	__lowerCamelCase						(							lowerCamelCase__		:						Union[str, os.PathLike]  ,							lowerCamelCase__		:						str  ,							lowerCamelCase__		:						Optional[Union[str, os.PathLike]] = None  ,							lowerCamelCase__		:						bool = False  ,							lowerCamelCase__		:						bool = False  ,							lowerCamelCase__		:						Optional[Dict[str, str]] = None  ,							lowerCamelCase__		:						Optional[Union[bool, str]] = None  ,							lowerCamelCase__		:						Optional[str] = None  ,							lowerCamelCase__		:						bool = False  ,							):





			'''simple docstring'''




			lowerCamelCase    		=  str(_lowerCamelCase      )

			lowerCamelCase    		=  os.path.join(_lowerCamelCase  ,							_lowerCamelCase      )

			if os.path.isfile(_lowerCamelCase      ):
						lowerCamelCase    		=  module_file_or_url
						lowerCamelCase    		=  """local"""
			elif pretrained_model_name_or_path.count("""/"""      ) == 0:
						lowerCamelCase    		=  get_diffusers_versions()
						# cut ".dev0"
						lowerCamelCase    		=  """v""" + """.""".join(__version__.split("""."""      )[:3]      )

						# retrieve github version that matches
						if revision is None:
									lowerCamelCase    		=  latest_version if latest_version[1:] in available_versions else """main"""
									logger.info(f'Defaulting to latest_version: {revision}.'      )
						elif revision in available_versions:
									lowerCamelCase    		=  f'v{revision}'
						elif revision == "main":
									lowerCamelCase    		=  revision
						else:
									raise ValueError(
									    f'`custom_revision`: {revision} does not exist. Please make sure to choose one of'
									    f' {", ".join(available_versions + ["main"]      )}.'      )

						# community pipeline on GitHub
						lowerCamelCase    		=  COMMUNITY_PIPELINES_URL.format(revision=_lowerCamelCase  ,							pipeline=_lowerCamelCase      )
						try:
									lowerCamelCase    		=  cached_download(
									    _lowerCamelCase  ,							cache_dir=_lowerCamelCase  ,							force_download=_lowerCamelCase  ,							proxies=_lowerCamelCase  ,							resume_download=_lowerCamelCase  ,							local_files_only=_lowerCamelCase  ,							use_auth_token=_lowerCamelCase  ,							)
									lowerCamelCase    		=  """git"""
									lowerCamelCase    		=  pretrained_model_name_or_path + """.py"""
						except EnvironmentError:
									logger.error(f'Could not locate the {module_file} inside {pretrained_model_name_or_path}.'      )
									raise
			else:
						try:
									# Load from URL or cache if already cached
									lowerCamelCase    		=  hf_hub_download(
									    _lowerCamelCase  ,							_lowerCamelCase  ,							cache_dir=_lowerCamelCase  ,							force_download=_lowerCamelCase  ,							proxies=_lowerCamelCase  ,							resume_download=_lowerCamelCase  ,							local_files_only=_lowerCamelCase  ,							use_auth_token=_lowerCamelCase  ,							)
									lowerCamelCase    		=  os.path.join("""local"""  ,							"""--""".join(pretrained_model_name_or_path.split("""/"""      )      )      )
						except EnvironmentError:
									logger.error(f'Could not locate the {module_file} inside {pretrained_model_name_or_path}.'      )
									raise

    # Check we have all the requirements in our environment
			lowerCamelCase    		=  check_imports(_lowerCamelCase      )

			# Now we move the module inside our cached dynamic modules.
			lowerCamelCase    		=  DIFFUSERS_DYNAMIC_MODULE_NAME + os.path.sep + submodule
			create_dynamic_module(_lowerCamelCase      )
			lowerCamelCase    		=  Path(_lowerCamelCase      ) / full_submodule
			if submodule == "local" or submodule == "git":
						# We always copy local files (we could hash the file to see if there was a change, and give them the name of
						# that hash, to only copy when there is a modification but it seems overkill for now).
						# The only reason we do the copy is to avoid putting too many folders in sys.path.
						shutil.copy(_lowerCamelCase  ,							submodule_path / module_file      )
						for module_needed in modules_needed:
									lowerCamelCase    		=  f'{module_needed}.py'
									shutil.copy(os.path.join(_lowerCamelCase  ,							_lowerCamelCase      )  ,							submodule_path / module_needed      )
			else:
						# Get the commit hash
						# TODO: we will get this info in the etag soon, so retrieve it from there and not here.
						if isinstance(_lowerCamelCase  ,							_lowerCamelCase      ):
									lowerCamelCase    		=  use_auth_token
						elif use_auth_token is True:
									lowerCamelCase    		=  HfFolder.get_token()
						else:
									lowerCamelCase    		=  None

						lowerCamelCase    		=  model_info(_lowerCamelCase  ,							revision=_lowerCamelCase  ,							token=_lowerCamelCase      ).sha

						# The module file will end up being placed in a subfolder with the git hash of the repo. This way we get the
						# benefit of versioning.
						lowerCamelCase    		=  submodule_path / commit_hash
						lowerCamelCase    		=  full_submodule + os.path.sep + commit_hash
						create_dynamic_module(_lowerCamelCase      )

						if not (submodule_path / module_file).exists():
									shutil.copy(_lowerCamelCase  ,							submodule_path / module_file      )
						# Make sure we also have every file with relative
						for module_needed in modules_needed:
									if not (submodule_path / module_needed).exists():
												get_cached_module_file(
												    _lowerCamelCase  ,							f'{module_needed}.py'  ,							cache_dir=_lowerCamelCase  ,							force_download=_lowerCamelCase  ,							resume_download=_lowerCamelCase  ,							proxies=_lowerCamelCase  ,							use_auth_token=_lowerCamelCase  ,							revision=_lowerCamelCase  ,							local_files_only=_lowerCamelCase  ,							)
			return os.path.join(_lowerCamelCase  ,							_lowerCamelCase      )






def 	__lowerCamelCase						(							lowerCamelCase__		:						Union[str, os.PathLike]  ,							lowerCamelCase__		:						str  ,							lowerCamelCase__		:						Optional[str] = None  ,							lowerCamelCase__		:						Optional[Union[str, os.PathLike]] = None  ,							lowerCamelCase__		:						bool = False  ,							lowerCamelCase__		:						bool = False  ,							lowerCamelCase__		:						Optional[Dict[str, str]] = None  ,							lowerCamelCase__		:						Optional[Union[bool, str]] = None  ,							lowerCamelCase__		:						Optional[str] = None  ,							lowerCamelCase__		:						bool = False  ,							**lowerCamelCase__		:						Optional[Any]  ,							):





			'''simple docstring'''




			lowerCamelCase    		=  get_cached_module_file(
			    _lowerCamelCase  ,							_lowerCamelCase  ,							cache_dir=_lowerCamelCase  ,							force_download=_lowerCamelCase  ,							resume_download=_lowerCamelCase  ,							proxies=_lowerCamelCase  ,							use_auth_token=_lowerCamelCase  ,							revision=_lowerCamelCase  ,							local_files_only=_lowerCamelCase  ,							)
			return get_class_in_module(_lowerCamelCase  ,							final_module.replace(""".py"""  ,							""""""      )      )







'''simple docstring'''



from unittest.mock import patch

import pyspark

from datasets.packaged_modules.spark.spark import (
    Spark,
    SparkExamplesIterable,
    _generate_iterable_examples,
)

from ..utils import (
    require_dill_gt_0_3_2,
    require_not_windows,
)







def       _UpperCamelCase			(_lowerCamelCase				:  Union[str, Any] ,						_lowerCamelCase				:  List[Any]						)-> Optional[Any]:




     '''simple docstring'''





     __snake_case            =	[]
     for part_id in partition_order:
          __snake_case            =	df.where(f'''SPARK_PARTITION_ID() = {part_id}'''						).collect()
          for row_idx, row in enumerate(_lowerCamelCase						):
               expected_row_ids_and_row_dicts.append((f'''{part_id}_{row_idx}''', row.asDict())						)
     return expected_row_ids_and_row_dicts







@require_not_windows
@require_dill_gt_0_3_2
def       _UpperCamelCase			()-> Any:




     '''simple docstring'''





     __snake_case            =	pyspark.sql.SparkSession.builder.master('''local[*]'''						).appName('''pyspark'''						).getOrCreate()
     __snake_case            =	spark.range(1_00						).repartition(1						)
     __snake_case            =	Spark(_lowerCamelCase						)
     # The id ints will be converted to Pyarrow int64s, so each row will be 8 bytes. Setting a max_shard_size of 16 means
     # that each partition can hold 2 rows.
     spark_builder._repartition_df_if_needed(max_shard_size=16						)
     # Given that the dataframe has 100 rows and each partition has 2 rows, we expect 50 partitions.
     assert spark_builder.df.rdd.getNumPartitions() == 50







@require_not_windows
@require_dill_gt_0_3_2
def       _UpperCamelCase			()-> Tuple:




     '''simple docstring'''





     __snake_case            =	pyspark.sql.SparkSession.builder.master('''local[*]'''						).appName('''pyspark'''						).getOrCreate()
     __snake_case            =	spark.range(10						).repartition(2						)
     __snake_case            =	[1, 0]
     __snake_case            =	_generate_iterable_examples(_lowerCamelCase ,						_lowerCamelCase						)  # Reverse the partitions.
     __snake_case            =	_get_expected_row_ids_and_row_dicts_for_partition_order(_lowerCamelCase ,						_lowerCamelCase						)

     for i, (row_id, row_dict) in enumerate(generate_fn()						):
          __snake_case       ,     __snake_case            =	expected_row_ids_and_row_dicts[i]
          assert row_id == expected_row_id
          assert row_dict == expected_row_dict







@require_not_windows
@require_dill_gt_0_3_2
def       _UpperCamelCase			()-> int:




     '''simple docstring'''





     __snake_case            =	pyspark.sql.SparkSession.builder.master('''local[*]'''						).appName('''pyspark'''						).getOrCreate()
     __snake_case            =	spark.range(10						).repartition(1						)
     __snake_case            =	SparkExamplesIterable(_lowerCamelCase						)
     assert it.n_shards == 1
     for i, (row_id, row_dict) in enumerate(_lowerCamelCase						):
          assert row_id == f'''0_{i}'''
          assert row_dict == {"id": i}







@require_not_windows
@require_dill_gt_0_3_2
def       _UpperCamelCase			()-> Union[str, Any]:




     '''simple docstring'''





     __snake_case            =	pyspark.sql.SparkSession.builder.master('''local[*]'''						).appName('''pyspark'''						).getOrCreate()
     __snake_case            =	spark.range(30						).repartition(3						)
     # Mock the generator so that shuffle reverses the partition indices.
     with patch('''numpy.random.Generator'''						) as generator_mock:
          __snake_case            =	lambda _lowerCamelCase						: x.reverse()
          __snake_case            =	_get_expected_row_ids_and_row_dicts_for_partition_order(_lowerCamelCase ,						[2, 1, 0]						)

          __snake_case            =	SparkExamplesIterable(_lowerCamelCase						).shuffle_data_sources(_lowerCamelCase						)
          assert shuffled_it.n_shards == 3
          for i, (row_id, row_dict) in enumerate(_lowerCamelCase						):
               __snake_case       ,     __snake_case            =	expected_row_ids_and_row_dicts[i]
               assert row_id == expected_row_id
               assert row_dict == expected_row_dict







@require_not_windows
@require_dill_gt_0_3_2
def       _UpperCamelCase			()-> Tuple:




     '''simple docstring'''





     __snake_case            =	pyspark.sql.SparkSession.builder.master('''local[*]'''						).appName('''pyspark'''						).getOrCreate()
     __snake_case            =	spark.range(20						).repartition(4						)

     # Partitions 0 and 2
     __snake_case            =	SparkExamplesIterable(_lowerCamelCase						).shard_data_sources(worker_id=0 ,						num_workers=2						)
     assert shard_it_a.n_shards == 2
     __snake_case            =	_get_expected_row_ids_and_row_dicts_for_partition_order(_lowerCamelCase ,						[0, 2]						)
     for i, (row_id, row_dict) in enumerate(_lowerCamelCase						):
          __snake_case       ,     __snake_case            =	expected_row_ids_and_row_dicts_a[i]
          assert row_id == expected_row_id
          assert row_dict == expected_row_dict

     # Partitions 1 and 3
     __snake_case            =	SparkExamplesIterable(_lowerCamelCase						).shard_data_sources(worker_id=1 ,						num_workers=2						)
     assert shard_it_a.n_shards == 2
     __snake_case            =	_get_expected_row_ids_and_row_dicts_for_partition_order(_lowerCamelCase ,						[1, 3]						)
     for i, (row_id, row_dict) in enumerate(_lowerCamelCase						):
          __snake_case       ,     __snake_case            =	expected_row_ids_and_row_dicts_a[i]
          assert row_id == expected_row_id
          assert row_dict == expected_row_dict







@require_not_windows
@require_dill_gt_0_3_2
def       _UpperCamelCase			()-> Optional[int]:




     '''simple docstring'''





     __snake_case            =	pyspark.sql.SparkSession.builder.master('''local[*]'''						).appName('''pyspark'''						).getOrCreate()
     __snake_case            =	spark.range(1_00						).repartition(1						)
     __snake_case            =	Spark(_lowerCamelCase						)
     # Choose a small max_shard_size for maximum partitioning.
     spark_builder._repartition_df_if_needed(max_shard_size=1						)
     # The new number of partitions should not be greater than the number of rows.
     assert spark_builder.df.rdd.getNumPartitions() == 1_00






