


import warnings
from typing import List, Optional, Union

import numpy as np
import PIL
import torch
from PIL import Image

from .configuration_utils import ConfigMixin, register_to_config
from .utils import CONFIG_NAME, PIL_INTERPOLATION, deprecate







class     __lowerCamelCase     (  snake_case_     ):


							"""simple docstring"""

							lowerCAmelCase__          =				CONFIG_NAME




							@register_to_config
							def __init__(  self						,       UpperCAmelCase = True						,       UpperCAmelCase = 8						,       UpperCAmelCase = "lanczos"						,       UpperCAmelCase = True						,       UpperCAmelCase = False						,       )     ->       Optional[int]:






								'''simple docstring'''






								super().__init__()




							@staticmethod
							def 				A__      (  UpperCAmelCase  )     ->       PIL.Image.Image:






								'''simple docstring'''






								if images.ndim == 3:
									lowercase_			 = images[None, ...]
								lowercase_			 = (images * 255).round().astype("uint8"  )
								if images.shape[-1] == 1:
									# special case for grayscale (single channel) images
									lowercase_			 = [Image.fromarray(image.squeeze()						,       mode="L"  ) for image in images]
								else:
									lowercase_			 = [Image.fromarray(UpperCAmelCase  ) for image in images]

								return pil_images




							@staticmethod
							def 				A__      (  UpperCAmelCase  )     ->       np.ndarray:






								'''simple docstring'''






								if not isinstance(UpperCAmelCase						,       UpperCAmelCase  ):
									lowercase_			 = [images]
								lowercase_			 = [np.array(UpperCAmelCase  ).astype(np.floataa  ) / 255.0 for image in images]
								lowercase_			 = np.stack(UpperCAmelCase						,       axis=0  )

								return images




							@staticmethod
							def 				A__      (  UpperCAmelCase  )     ->       torch.FloatTensor:






								'''simple docstring'''






								if images.ndim == 3:
									lowercase_			 = images[..., None]

								lowercase_			 = torch.from_numpy(images.transpose(0						,       3						,       1						,       2  )  )
								return images




							@staticmethod
							def 				A__      (  UpperCAmelCase  )     ->       np.ndarray:






								'''simple docstring'''






								lowercase_			 = images.cpu().permute(0						,       2						,       3						,       1  ).float().numpy()
								return images




							@staticmethod
							def 				A__      (  UpperCAmelCase  )     ->       Tuple:






								'''simple docstring'''






								return 2.0 * images - 1.0




							@staticmethod
							def 				A__      (  UpperCAmelCase  )     ->       Union[str, Any]:






								'''simple docstring'''






								return (images / 2 + 0.5).clamp(0						,       1  )




							@staticmethod
							def 				A__      (  UpperCAmelCase  )     ->       PIL.Image.Image:






								'''simple docstring'''






								lowercase_			 = image.convert("RGB"  )
								return image




							def 				A__      (  self						,       UpperCAmelCase						,       UpperCAmelCase = None						,       UpperCAmelCase = None						,       )     ->       PIL.Image.Image:






								'''simple docstring'''






								if height is None:
									lowercase_			 = image.height
								if width is None:
									lowercase_			 = image.width

								lowercase_ ,	lowercase_			 = (
								    x - x % self.config.vae_scale_factor for x in (width, height)
								)  # resize to integer multiple of vae_scale_factor
								lowercase_			 = image.resize((width, height)						,       resample=PIL_INTERPOLATION[self.config.resample]  )
								return image




							def 				A__      (  self						,       UpperCAmelCase						,       UpperCAmelCase = None						,       UpperCAmelCase = None						,       )     ->       torch.Tensor:






								'''simple docstring'''






								lowercase_			 = (PIL.Image.Image, np.ndarray, torch.Tensor)
								if isinstance(UpperCAmelCase						,       UpperCAmelCase  ):
									lowercase_			 = [image]
								elif not (isinstance(UpperCAmelCase						,       UpperCAmelCase  ) and all(isinstance(UpperCAmelCase						,       UpperCAmelCase  ) for i in image  )):
									raise ValueError(
									    F'Input is in incorrect format: {[type(UpperCAmelCase  ) for i in image]}. Currently, we only support {", ".join(UpperCAmelCase  )}'  )

								if isinstance(image[0]						,       PIL.Image.Image  ):
									if self.config.do_convert_rgb:
										lowercase_			 = [self.convert_to_rgb(UpperCAmelCase  ) for i in image]
									if self.config.do_resize:
										lowercase_			 = [self.resize(UpperCAmelCase						,       UpperCAmelCase						,       UpperCAmelCase  ) for i in image]
									lowercase_			 = self.pil_to_numpy(UpperCAmelCase  )  # to np
									lowercase_			 = self.numpy_to_pt(UpperCAmelCase  )  # to pt

								elif isinstance(image[0]						,       np.ndarray  ):
									lowercase_			 = np.concatenate(UpperCAmelCase						,       axis=0  ) if image[0].ndim == 4 else np.stack(UpperCAmelCase						,       axis=0  )
									lowercase_			 = self.numpy_to_pt(UpperCAmelCase  )
									lowercase_ ,	lowercase_ ,	lowercase_ ,	lowercase_			 = image.shape
									if self.config.do_resize and (
									    height % self.config.vae_scale_factor != 0 or width % self.config.vae_scale_factor != 0
									):
										raise ValueError(
										    F'Currently we only support resizing for PIL image - please resize your numpy array to be divisible by {self.config.vae_scale_factor}'
										    F'currently the sizes are {height} and {width}. You can also pass a PIL image instead to use resize option in VAEImageProcessor'  )

								elif isinstance(image[0]						,       torch.Tensor  ):
									lowercase_			 = torch.cat(UpperCAmelCase						,       axis=0  ) if image[0].ndim == 4 else torch.stack(UpperCAmelCase						,       axis=0  )
									lowercase_ ,	lowercase_ ,	lowercase_ ,	lowercase_			 = image.shape

									# don't need any preprocess if the image is latents
									if channel == 4:
										return image

									if self.config.do_resize and (
									    height % self.config.vae_scale_factor != 0 or width % self.config.vae_scale_factor != 0
									):
										raise ValueError(
										    F'Currently we only support resizing for PIL image - please resize your pytorch tensor to be divisible by {self.config.vae_scale_factor}'
										    F'currently the sizes are {height} and {width}. You can also pass a PIL image instead to use resize option in VAEImageProcessor'  )

        # expected range [0,1], normalize to [-1,1]
								lowercase_			 = self.config.do_normalize
								if image.min() < 0:
									warnings.warn(
									    "Passing `image` as torch tensor with value range in [-1,1] is deprecated. The expected value range for image tensor is [0,1] "
									    F'when passing as pytorch tensor or numpy Array. You passed `image` with value range [{image.min()},{image.max()}]'						,       UpperCAmelCase						,       )
									lowercase_			 = False

								if do_normalize:
									lowercase_			 = self.normalize(UpperCAmelCase  )

								return image




							def 				A__      (  self						,       UpperCAmelCase						,       UpperCAmelCase = "pil"						,       UpperCAmelCase = None						,       )     ->       Dict:






								'''simple docstring'''






								if not isinstance(UpperCAmelCase						,       torch.Tensor  ):
									raise ValueError(
									    F'Input for postprocessing is in incorrect format: {type(UpperCAmelCase  )}. We only support pytorch tensor'  )
								if output_type not in ["latent", "pt", "np", "pil"]:
									lowercase_			 = (
									    F'the output_type {output_type} is outdated and has been set to `np`. Please make sure to set it to one of these instead: '
									    "`pil`, `np`, `pt`, `latent`"
									)
									deprecate("Unsupported output_type"						,       "1.0.0"						,       UpperCAmelCase						,       standard_warn=UpperCAmelCase  )
									lowercase_			 = "np"

								if output_type == "latent":
									return image

								if do_denormalize is None:
									lowercase_			 = [self.config.do_normalize] * image.shape[0]

								lowercase_			 = torch.stack(
								    [self.denormalize(image[i]  ) if do_denormalize[i] else image[i] for i in range(image.shape[0]  )]  )

								if output_type == "pt":
									return image

								lowercase_			 = self.pt_to_numpy(UpperCAmelCase  )

								if output_type == "np":
									return image

								if output_type == "pil":
									return self.numpy_to_pil(UpperCAmelCase  )







class     __lowerCamelCase     (  snake_case_     ):


							"""simple docstring"""

							lowerCAmelCase__          =				CONFIG_NAME




							@register_to_config
							def __init__(  self						,       UpperCAmelCase = True						,       UpperCAmelCase = 8						,       UpperCAmelCase = "lanczos"						,       UpperCAmelCase = True						,       )     ->       Dict:






								'''simple docstring'''






								super().__init__()




							@staticmethod
							def 				A__      (  UpperCAmelCase  )     ->       Any:






								'''simple docstring'''






								if images.ndim == 3:
									lowercase_			 = images[None, ...]
								lowercase_			 = (images * 255).round().astype("uint8"  )
								if images.shape[-1] == 1:
									# special case for grayscale (single channel) images
									lowercase_			 = [Image.fromarray(image.squeeze()						,       mode="L"  ) for image in images]
								else:
									lowercase_			 = [Image.fromarray(image[:, :, :3]  ) for image in images]

								return pil_images




							@staticmethod
							def 				A__      (  UpperCAmelCase  )     ->       Dict:






								'''simple docstring'''






								return image[:, :, 1] * 2**8 + image[:, :, 2]




							def 				A__      (  self						,       UpperCAmelCase  )     ->       Union[str, Any]:






								'''simple docstring'''






								if images.ndim == 3:
									lowercase_			 = images[None, ...]
								lowercase_			 = images[:, :, :, 3:]
								if images.shape[-1] == 6:
									lowercase_			 = (images_depth * 255).round().astype("uint8"  )
									lowercase_			 = [
									    Image.fromarray(self.rgblike_to_depthmap(UpperCAmelCase  )						,       mode="I;16"  ) for image_depth in images_depth
									]
								elif images.shape[-1] == 4:
									lowercase_			 = (images_depth * 65535.0).astype(np.uintaa  )
									lowercase_			 = [Image.fromarray(UpperCAmelCase						,       mode="I;16"  ) for image_depth in images_depth]
								else:
									raise Exception("Not supported"  )

								return pil_images




							def 				A__      (  self						,       UpperCAmelCase						,       UpperCAmelCase = "pil"						,       UpperCAmelCase = None						,       )     ->       int:






								'''simple docstring'''






								if not isinstance(UpperCAmelCase						,       torch.Tensor  ):
									raise ValueError(
									    F'Input for postprocessing is in incorrect format: {type(UpperCAmelCase  )}. We only support pytorch tensor'  )
								if output_type not in ["latent", "pt", "np", "pil"]:
									lowercase_			 = (
									    F'the output_type {output_type} is outdated and has been set to `np`. Please make sure to set it to one of these instead: '
									    "`pil`, `np`, `pt`, `latent`"
									)
									deprecate("Unsupported output_type"						,       "1.0.0"						,       UpperCAmelCase						,       standard_warn=UpperCAmelCase  )
									lowercase_			 = "np"

								if do_denormalize is None:
									lowercase_			 = [self.config.do_normalize] * image.shape[0]

								lowercase_			 = torch.stack(
								    [self.denormalize(image[i]  ) if do_denormalize[i] else image[i] for i in range(image.shape[0]  )]  )

								lowercase_			 = self.pt_to_numpy(UpperCAmelCase  )

								if output_type == "np":
									if image.shape[-1] == 6:
										lowercase_			 = np.stack([self.rgblike_to_depthmap(im[:, :, 3:]  ) for im in image]						,       axis=0  )
									else:
										lowercase_			 = image[:, :, :, 3:]
									return image[:, :, :, :3], image_depth

								if output_type == "pil":
									return self.numpy_to_pil(UpperCAmelCase  ), self.numpy_to_depth(UpperCAmelCase  )
								else:
									raise Exception(F'This type {output_type} is not supported'  )





import warnings

import numpy as np
import pandas as pd
from matplotlib import pyplot as plt
from sklearn.metrics import pairwise_distances

warnings.filterwarnings("""ignore""")

SCREAMING_SNAKE_CASE__       			=				"""K-MEANS-CLUST/ """




def SCREAMING_SNAKE_CASE_  (				__lowerCamelCase:		List[str]  ,			__lowerCamelCase:		List[Any]  ,			__lowerCamelCase:		Tuple=None  ):







	'''simple docstring'''



	if seed is not None:  # useful for obtaining consistent results
		np.random.seed(__lowerCamelCase  )
	lowercase_			 = data.shape[0]  # number of data points

	# Pick K indices from range [0, N).
	lowercase_			 = np.random.randint(0  ,			__lowerCamelCase  ,			__lowerCamelCase  )

	# Keep centroids as dense format, as many entries will be nonzero due to averaging.
	# As long as at least one document in a cluster contains a word,
	# it will carry a nonzero weight in the TF-IDF vector of the centroid.
	lowercase_			 = data[rand_indices, :]

	return centroids




def SCREAMING_SNAKE_CASE_  (				__lowerCamelCase:		Union[str, Any]  ,			__lowerCamelCase:		Union[str, Any]  ):







	'''simple docstring'''



	return pairwise_distances(__lowerCamelCase  ,			__lowerCamelCase  ,			metric="euclidean"  )




def SCREAMING_SNAKE_CASE_  (				__lowerCamelCase:		List[str]  ,			__lowerCamelCase:		int  ):







	'''simple docstring'''



	lowercase_			 = centroid_pairwise_dist(__lowerCamelCase  ,			__lowerCamelCase  )

	# Compute cluster assignments for each data point:
	# Fill in the blank (RHS only)
	lowercase_			 = np.argmin(__lowerCamelCase  ,			axis=1  )

	return cluster_assignment




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







	'''simple docstring'''



	lowercase_			 = []
	for i in range(__lowerCamelCase  ):
		# Select all data points that belong to cluster i. Fill in the blank (RHS only)
		lowercase_			 = data[cluster_assignment == i]
		# Compute the mean of the data points. Fill in the blank (RHS only)
		lowercase_			 = member_data_points.mean(axis=0  )
		new_centroids.append(__lowerCamelCase  )
	lowercase_			 = np.array(__lowerCamelCase  )

	return new_centroids




def SCREAMING_SNAKE_CASE_  (				__lowerCamelCase:		str  ,			__lowerCamelCase:		str  ,			__lowerCamelCase:		List[Any]  ,			__lowerCamelCase:		Optional[Any]  ):







	'''simple docstring'''



	lowercase_			 = 0.0
	for i in range(__lowerCamelCase  ):
		# Select all data points that belong to cluster i. Fill in the blank (RHS only)
		lowercase_			 = data[cluster_assignment == i, :]

		if member_data_points.shape[0] > 0:  # check if i-th cluster is non-empty
			# Compute distances from centroid to data points (RHS only)
			lowercase_			 = pairwise_distances(
			    __lowerCamelCase  ,			[centroids[i]]  ,			metric="euclidean"  )
			lowercase_			 = distances**2
			heterogeneity += np.sum(__lowerCamelCase  )

	return heterogeneity




def SCREAMING_SNAKE_CASE_  (				__lowerCamelCase:		List[Any]  ,			__lowerCamelCase:		Union[str, Any]  ):







	'''simple docstring'''



	plt.figure(figsize=(7, 4)  )
	plt.plot(__lowerCamelCase  ,			linewidth=4  )
	plt.xlabel("# Iterations"  )
	plt.ylabel("Heterogeneity"  )
	plt.title(F'Heterogeneity of clustering over time, K={k:d}'  )
	plt.rcParams.update({"font.size": 16}  )
	plt.show()




def SCREAMING_SNAKE_CASE_  (				__lowerCamelCase:		int  ,			__lowerCamelCase:		Dict  ,			__lowerCamelCase:		Any  ,			__lowerCamelCase:		str=500  ,			__lowerCamelCase:		Union[str, Any]=None  ,			__lowerCamelCase:		Optional[int]=False  ):







	'''simple docstring'''



	lowercase_			 = initial_centroids[:]
	lowercase_			 = None

	for itr in range(__lowerCamelCase  ):
		if verbose:
			print(__lowerCamelCase  ,			end=""  )

		# 1. Make cluster assignments using nearest centroids
		lowercase_			 = assign_clusters(__lowerCamelCase  ,			__lowerCamelCase  )

		# 2. Compute a new centroid for each of the k clusters, averaging all data
		#    points assigned to that cluster.
		lowercase_			 = revise_centroids(__lowerCamelCase  ,			__lowerCamelCase  ,			__lowerCamelCase  )

		# Check for convergence: if none of the assignments changed, stop
		if (
		    prev_cluster_assignment is not None
		    and (prev_cluster_assignment == cluster_assignment).all()
		):
			break

		# Print number of new assignments
		if prev_cluster_assignment is not None:
			lowercase_			 = np.sum(prev_cluster_assignment != cluster_assignment  )
			if verbose:
				print(
				    F'    {num_changed:5d} elements changed their cluster assignment.'  )

        # Record heterogeneity convergence metric
		if record_heterogeneity is not None:
			# YOUR CODE HERE
			lowercase_			 = compute_heterogeneity(__lowerCamelCase  ,			__lowerCamelCase  ,			__lowerCamelCase  ,			__lowerCamelCase  )
			record_heterogeneity.append(__lowerCamelCase  )

		lowercase_			 = cluster_assignment[:]

	return centroids, cluster_assignment


# Mock test below
if False:  # change to true to run this test case.
							from sklearn import datasets as ds

							SCREAMING_SNAKE_CASE__       			=				ds.load_iris()
							SCREAMING_SNAKE_CASE__       			=				3
							SCREAMING_SNAKE_CASE__       			=				[]
							SCREAMING_SNAKE_CASE__       			=				get_initial_centroids(dataset["""data"""], k, seed=0)
							SCREAMING_SNAKE_CASE__			,      SCREAMING_SNAKE_CASE__       			=				kmeans(
							    dataset["""data"""],
							    k,
							    initial_centroids,
							    maxiter=4_0_0,
							    record_heterogeneity=heterogeneity,
							    verbose=True,
							)
							plot_heterogeneity(heterogeneity, k)




def SCREAMING_SNAKE_CASE_  (				__lowerCamelCase:		pd.DataFrame  ,			__lowerCamelCase:		np.ndarray  ,			__lowerCamelCase:		str=None  ):







	'''simple docstring'''



	if fill_missing_report:
		lowercase_			 = df.fillna(value=__lowerCamelCase  )
	lowercase_			 = 1
	lowercase_			 = df.select_dtypes(np.number  ).columns
	lowercase_			 = (
	    df.groupby(["Cluster"]  )[  # construct report dataframe
	        numeric_cols
	    ]  # group by cluster number
	    .agg(
	        [
	            ("sum", np.sum),
	            ("mean_with_zeros", lambda __lowerCamelCase  : np.mean(np.nan_to_num(__lowerCamelCase  )  )),
	            ("mean_without_zeros", lambda __lowerCamelCase  : x.replace(0  ,			np.NaN  ).mean()),
	            (
	                "mean_25-75",
	                lambda __lowerCamelCase  : np.mean(
	                    np.nan_to_num(
	                        sorted(__lowerCamelCase  )[
	                            round(len(__lowerCamelCase  ) * 25 / 100  ) : round(len(__lowerCamelCase  ) * 75 / 100  )
	                        ]  )  ),
	            ),
	            ("mean_with_na", np.mean),
	            ("min", lambda __lowerCamelCase  : x.min()),
	            ("5%", lambda __lowerCamelCase  : x.quantile(0.05  )),
	            ("25%", lambda __lowerCamelCase  : x.quantile(0.25  )),
	            ("50%", lambda __lowerCamelCase  : x.quantile(0.50  )),
	            ("75%", lambda __lowerCamelCase  : x.quantile(0.75  )),
	            ("95%", lambda __lowerCamelCase  : x.quantile(0.95  )),
	            ("max", lambda __lowerCamelCase  : x.max()),
	            ("count", lambda __lowerCamelCase  : x.count()),
	            ("stdev", lambda __lowerCamelCase  : x.std()),
	            ("mode", lambda __lowerCamelCase  : x.mode()[0]),
	            ("median", lambda __lowerCamelCase  : x.median()),
	            ("# > 0", lambda __lowerCamelCase  : (x > 0).sum()),
	        ]  )
	    .T.reset_index()
	    .rename(index=__lowerCamelCase  ,			columns={"level_0": "Features", "level_1": "Type"}  )
	)  # rename columns
	# calculate the size of cluster(count of clientID's)
	lowercase_			 = report[
	    (report["Features"] == "dummy") & (report["Type"] == "count")
	].copy()  # avoid SettingWithCopyWarning
	lowercase_			 = (
	    "ClusterSize"  # rename created cluster df to match report column names
	)
	lowercase_			 = "# of Customers"
	lowercase_			 = pd.DataFrame(
	    clustersize.iloc[:, 2:].values
	    / clustersize.iloc[:, 2:].values.sum()  )
	lowercase_			 = "% of Customers"  # rename created cluster df to match report column names
	lowercase_			 = "ClusterProportion"
	lowercase_			 = clusterproportion.columns.tolist()
	lowercase_			 = cols[-2:] + cols[:-2]
	lowercase_			 = clusterproportion[cols]  # rearrange columns to match report
	lowercase_			 = report.columns
	lowercase_			 = pd.DataFrame(
	    abs(
	        report[report["Type"] == "count"].iloc[:, 2:].values
	        - clustersize.iloc[:, 2:].values  )  )  # generating df with count of nan values
	lowercase_			 = 0
	lowercase_			 = "# of nan"
	lowercase_			 = report[
	    report["Type"] == "count"
	].Features.tolist()  # filling values in order to match report
	lowercase_			 = a.columns.tolist()
	lowercase_			 = cols[-2:] + cols[:-2]
	lowercase_			 = a[cols]  # rearrange columns to match report
	lowercase_			 = report.columns  # rename columns to match report
	lowercase_			 = report.drop(
	    report[report.Type == "count"].index  )  # drop count values except cluster size
	lowercase_			 = pd.concat(
	    [report, a, clustersize, clusterproportion]  ,			axis=0  )  # concat report with clustert size and nan values
	lowercase_			 = report["Features"].isin(__lowerCamelCase  )
	lowercase_			 = report.columns.tolist()
	lowercase_			 = cols[0:2] + cols[-1:] + cols[2:-1]
	lowercase_			 = report[cols]
	lowercase_			 = {
	    "ClusterSize": 9,
	    "ClusterProportion": 8,
	    "mean_with_zeros": 7,
	    "mean_with_na": 6,
	    "max": 5,
	    "50%": 4,
	    "min": 3,
	    "25%": 2,
	    "75%": 1,
	    "# of nan": 0,
	    "# > 0": -1,
	    "sum_with_na": -2,
	}
	lowercase_			 = (
	    report.assign(
	        Sortera=lambda __lowerCamelCase  : x.Type.map(__lowerCamelCase  )  ,			Sortera=lambda __lowerCamelCase  : list(reversed(range(len(__lowerCamelCase  )  )  )  )  ,			)
	    .sort_values(["Sorter1", "Mark", "Sorter2"]  ,			ascending=__lowerCamelCase  )
	    .drop(["Sorter1", "Sorter2"]  ,			axis=1  )
	)
	lowercase_			 = ""
	lowercase_			 = report.reset_index()
	lowercase_			 = report.drop(columns=["index"]  )
	return report


if __name__ == "__main__":
							import doctest

							doctest.testmod()

