
'''simple docstring'''

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')

__UpperCAmelCase												=   "K-MEANS-CLUST/ "



def        SCREAMING_SNAKE_CASE_  (	snake_case_				:	Any   ,     snake_case_				:	List[Any]   ,     snake_case_				:	Dict=None							)       ->	Dict:



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

 # Pick K indices from range [0, N).
 SCREAMING_SNAKE_CASE  :  str								=		np.random.randint(0   ,     snake_case_   ,     snake_case_							)

 # 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.
 SCREAMING_SNAKE_CASE  :  Optional[Any]								=		data[rand_indices, :]

 return centroids



def        SCREAMING_SNAKE_CASE_  (	snake_case_				:	Dict   ,     snake_case_				:	Dict							)       ->	List[str]:
 return pairwise_distances(snake_case_   ,     snake_case_   ,     metric='euclidean'							)



def        SCREAMING_SNAKE_CASE_  (	snake_case_				:	Tuple   ,     snake_case_				:	Any							)       ->	Union[str, Any]:
 SCREAMING_SNAKE_CASE  :  Tuple								=		centroid_pairwise_dist(snake_case_   ,     snake_case_							)

 # Compute cluster assignments for each data point:
 # Fill in the blank (RHS only)
 SCREAMING_SNAKE_CASE  :  Optional[Any]								=		np.argmin(snake_case_   ,     axis=1							)

 return cluster_assignment



def        SCREAMING_SNAKE_CASE_  (	snake_case_				:	Union[str, Any]   ,     snake_case_				:	Union[str, Any]   ,     snake_case_				:	int							)       ->	int:
 SCREAMING_SNAKE_CASE  :  List[str]								=		[]
 for i in range(snake_case_							):
  # Select all data points that belong to cluster i. Fill in the blank (RHS only)
  SCREAMING_SNAKE_CASE  :  Any								=		data[cluster_assignment == i]
  # Compute the mean of the data points. Fill in the blank (RHS only)
  SCREAMING_SNAKE_CASE  :  Union[str, Any]								=		member_data_points.mean(axis=0							)
  new_centroids.append(snake_case_							)
 SCREAMING_SNAKE_CASE  :  Optional[Any]								=		np.array(snake_case_							)

 return new_centroids



def        SCREAMING_SNAKE_CASE_  (	snake_case_				:	List[Any]   ,     snake_case_				:	Optional[Any]   ,     snake_case_				:	Any   ,     snake_case_				:	str							)       ->	str:
 SCREAMING_SNAKE_CASE  :  Optional[int]								=		0.0
 for i in range(snake_case_							):
  # Select all data points that belong to cluster i. Fill in the blank (RHS only)
  SCREAMING_SNAKE_CASE  :  Union[str, Any]								=		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)
   SCREAMING_SNAKE_CASE  :  Optional[int]								=		pairwise_distances(
       snake_case_   ,     [centroids[i]]   ,     metric='euclidean'							)
   SCREAMING_SNAKE_CASE  :  List[Any]								=		distances**2
   heterogeneity += np.sum(snake_case_							)

 return heterogeneity



def        SCREAMING_SNAKE_CASE_  (	snake_case_				:	Any   ,     snake_case_				:	Optional[int]							)       ->	List[Any]:
 plt.figure(figsize=(7, 4)							)
 plt.plot(snake_case_   ,     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_  (	snake_case_				:	Optional[Any]   ,     snake_case_				:	Any   ,     snake_case_				:	List[Any]   ,     snake_case_				:	Tuple=500   ,     snake_case_				:	Any=None   ,     snake_case_				:	Any=False							)       ->	List[Any]:
 SCREAMING_SNAKE_CASE  :  Dict								=		initial_centroids[:]
 SCREAMING_SNAKE_CASE  :  Tuple								=		None

 for itr in range(snake_case_							):
  if verbose:
   print(snake_case_   ,     end=''							)

  # 1. Make cluster assignments using nearest centroids
  SCREAMING_SNAKE_CASE  :  List[str]								=		assign_clusters(snake_case_   ,     snake_case_							)

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

  # 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:
   SCREAMING_SNAKE_CASE  :  Optional[Any]								=		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
   SCREAMING_SNAKE_CASE  :  Any								=		compute_heterogeneity(snake_case_   ,     snake_case_   ,     snake_case_   ,     snake_case_							)
   record_heterogeneity.append(snake_case_							)

  SCREAMING_SNAKE_CASE  :  Optional[int]								=		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

   __UpperCAmelCase												=   ds.load_iris()
   __UpperCAmelCase												=   3
   __UpperCAmelCase												=   []
   __UpperCAmelCase												=   get_initial_centroids(dataset['data'], k, seed=0)
   __UpperCAmelCase												=   kmeans(
       dataset['data'],
       k,
       initial_centroids,
       maxiter=400,
       record_heterogeneity=heterogeneity,
       verbose=True,
   )
   plot_heterogeneity(heterogeneity, k)



def        SCREAMING_SNAKE_CASE_  (	snake_case_				:	pd.DataFrame   ,     snake_case_				:	np.ndarray   ,     snake_case_				:	List[Any]=None							)       ->	str:



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


if __name__ == "__main__":
   import doctest

   doctest.testmod()






from typing import Any, Callable, Dict, List, Optional, Union

import PIL.Image
import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionImgaImgPipeline,
    StableDiffusionInpaintPipelineLegacy,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.configuration_utils import FrozenDict
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import deprecate, logging


__UpperCamelCase							:							Any    							=					logging.get_logger(__name__)  # pylint: disable=invalid-name

class    __magic_name__						(  __lowerCAmelCase):
     A:		Dict								=   ["safety_checker", "feature_extractor"]


     def __init__(   self				:  List[str]				,						lowerCamelCase__				:  AutoencoderKL				,						lowerCamelCase__				:  CLIPTextModel				,						lowerCamelCase__				:  CLIPTokenizer				,						lowerCamelCase__				:  UNetaDConditionModel				,						lowerCamelCase__				:  Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]				,						lowerCamelCase__				:  StableDiffusionSafetyChecker				,						lowerCamelCase__				:  CLIPImageProcessor				,						lowerCamelCase__				:  bool = True				,						)	->   Optional[Any]:





      '''simple docstring'''






      super().__init__()
      if hasattr(scheduler.config				,						'''steps_offset'''      ) and scheduler.config.steps_offset != 1:
       UpperCamelCase__       :     int 							= (
           F"The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`"
           F" should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure "
           '''to update the config accordingly as leaving `steps_offset` might led to incorrect results'''
           ''' in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,'''
           ''' it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`'''
           ''' file'''
       )
       deprecate('''steps_offset!=1'''				,						'''1.0.0'''				,						lowerCamelCase__				,						standard_warn=lowerCamelCase__      )
       UpperCamelCase__       :     int 							= dict(scheduler.config      )
       UpperCamelCase__       :     List[str] 							= 1
       UpperCamelCase__       :     List[str] 							= FrozenDict(lowerCamelCase__      )

      self.register_modules(
          vae=lowerCamelCase__				,						text_encoder=lowerCamelCase__				,						tokenizer=lowerCamelCase__				,						unet=lowerCamelCase__				,						scheduler=lowerCamelCase__				,						safety_checker=lowerCamelCase__				,						feature_extractor=lowerCamelCase__				,						)
      self.register_to_config(requires_safety_checker=lowerCamelCase__      )


     @property
     def   UpperCAmelCase__						(   self				:  str      )	->   Dict[str, Any]:





      '''simple docstring'''






      return {k: getattr(self				,						lowerCamelCase__      ) for k in self.config.keys() if not k.startswith('''_'''      )}


     def   UpperCAmelCase__						(   self				:  str				,						lowerCamelCase__				:  Optional[Union[str, int]] = "auto"      )	->   str:





      '''simple docstring'''






      if slice_size == "auto":
       # half the attention head size is usually a good trade-off between
       # speed and memory
       UpperCamelCase__       :     Dict 							= self.unet.config.attention_head_dim // 2
      self.unet.set_attention_slice(lowerCamelCase__      )


     def   UpperCAmelCase__						(   self				:  int      )	->   Tuple:





      '''simple docstring'''






      self.enable_attention_slicing(lowerCamelCase__      )


     @torch.no_grad()
     def   UpperCAmelCase__						(   self				:  Optional[Any]				,						lowerCamelCase__				:  Union[str, List[str]]				,						lowerCamelCase__				:  Union[torch.FloatTensor, PIL.Image.Image]				,						lowerCamelCase__				:  Union[torch.FloatTensor, PIL.Image.Image]				,						lowerCamelCase__				:  float = 0.8				,						lowerCamelCase__				:  Optional[int] = 50				,						lowerCamelCase__				:  Optional[float] = 7.5				,						lowerCamelCase__				:  Optional[Union[str, List[str]]] = None				,						lowerCamelCase__				:  Optional[int] = 1				,						lowerCamelCase__				:  Optional[float] = 0.0				,						lowerCamelCase__				:  Optional[torch.Generator] = None				,						lowerCamelCase__				:  Optional[str] = "pil"				,						lowerCamelCase__				:  bool = True				,						lowerCamelCase__				:  Optional[Callable[[int, int, torch.FloatTensor], None]] = None				,						lowerCamelCase__				:  int = 1				,						)	->   Optional[Any]:





      '''simple docstring'''






      return StableDiffusionInpaintPipelineLegacy(**self.components      )(
          prompt=lowerCamelCase__				,						image=lowerCamelCase__				,						mask_image=lowerCamelCase__				,						strength=lowerCamelCase__				,						num_inference_steps=lowerCamelCase__				,						guidance_scale=lowerCamelCase__				,						negative_prompt=lowerCamelCase__				,						num_images_per_prompt=lowerCamelCase__				,						eta=lowerCamelCase__				,						generator=lowerCamelCase__				,						output_type=lowerCamelCase__				,						return_dict=lowerCamelCase__				,						callback=lowerCamelCase__				,						)


     @torch.no_grad()
     def   UpperCAmelCase__						(   self				:  Any				,						lowerCamelCase__				:  Union[str, List[str]]				,						lowerCamelCase__				:  Union[torch.FloatTensor, PIL.Image.Image]				,						lowerCamelCase__				:  float = 0.8				,						lowerCamelCase__				:  Optional[int] = 50				,						lowerCamelCase__				:  Optional[float] = 7.5				,						lowerCamelCase__				:  Optional[Union[str, List[str]]] = None				,						lowerCamelCase__				:  Optional[int] = 1				,						lowerCamelCase__				:  Optional[float] = 0.0				,						lowerCamelCase__				:  Optional[torch.Generator] = None				,						lowerCamelCase__				:  Optional[str] = "pil"				,						lowerCamelCase__				:  bool = True				,						lowerCamelCase__				:  Optional[Callable[[int, int, torch.FloatTensor], None]] = None				,						lowerCamelCase__				:  int = 1				,						**lowerCamelCase__				:  str				,						)	->   str:





      '''simple docstring'''






      return StableDiffusionImgaImgPipeline(**self.components      )(
          prompt=lowerCamelCase__				,						image=lowerCamelCase__				,						strength=lowerCamelCase__				,						num_inference_steps=lowerCamelCase__				,						guidance_scale=lowerCamelCase__				,						negative_prompt=lowerCamelCase__				,						num_images_per_prompt=lowerCamelCase__				,						eta=lowerCamelCase__				,						generator=lowerCamelCase__				,						output_type=lowerCamelCase__				,						return_dict=lowerCamelCase__				,						callback=lowerCamelCase__				,						callback_steps=lowerCamelCase__				,						)


     @torch.no_grad()
     def   UpperCAmelCase__						(   self				:  Optional[int]				,						lowerCamelCase__				:  Union[str, List[str]]				,						lowerCamelCase__				:  int = 512				,						lowerCamelCase__				:  int = 512				,						lowerCamelCase__				:  int = 50				,						lowerCamelCase__				:  float = 7.5				,						lowerCamelCase__				:  Optional[Union[str, List[str]]] = None				,						lowerCamelCase__				:  Optional[int] = 1				,						lowerCamelCase__				:  float = 0.0				,						lowerCamelCase__				:  Optional[torch.Generator] = None				,						lowerCamelCase__				:  Optional[torch.FloatTensor] = None				,						lowerCamelCase__				:  Optional[str] = "pil"				,						lowerCamelCase__				:  bool = True				,						lowerCamelCase__				:  Optional[Callable[[int, int, torch.FloatTensor], None]] = None				,						lowerCamelCase__				:  int = 1				,						)	->   Optional[Any]:





      '''simple docstring'''






      return StableDiffusionPipeline(**self.components      )(
          prompt=lowerCamelCase__				,						height=lowerCamelCase__				,						width=lowerCamelCase__				,						num_inference_steps=lowerCamelCase__				,						guidance_scale=lowerCamelCase__				,						negative_prompt=lowerCamelCase__				,						num_images_per_prompt=lowerCamelCase__				,						eta=lowerCamelCase__				,						generator=lowerCamelCase__				,						latents=lowerCamelCase__				,						output_type=lowerCamelCase__				,						return_dict=lowerCamelCase__				,						callback=lowerCamelCase__				,						callback_steps=lowerCamelCase__				,						)


