





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

lowerCAmelCase      :  Union[str, Any]      							='''K-MEANS-CLUST/ '''
def 							UpperCAmelCase_						(		__lowerCamelCase      :      str			,__lowerCamelCase      :      List[Any]			,__lowerCamelCase      :      Tuple=None				):
   if seed is not None:  # useful for obtaining consistent results
      np.random.seed(__lowerCamelCase				)
   lowercase_    :Union[str, Any]					    =				data.shape[0]  # number of data points

   # Pick K indices from range [0, N).
   lowercase_    :List[Any]					    =				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_    :List[Any]					    =				data[rand_indices, :]

   return centroids
def 							UpperCAmelCase_						(		__lowerCamelCase      :      str			,__lowerCamelCase      :      Union[str, Any]				):
   return pairwise_distances(__lowerCamelCase			,__lowerCamelCase			,metric="euclidean"				)
def 							UpperCAmelCase_						(		__lowerCamelCase      :      List[Any]			,__lowerCamelCase      :      Optional[Any]				):
   # Compute distances between each data point and the set of centroids:
   # Fill in the blank (RHS only)
   lowercase_    :int					    =				centroid_pairwise_dist(__lowerCamelCase			,__lowerCamelCase				)

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

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

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

   return heterogeneity
def 							UpperCAmelCase_						(		__lowerCamelCase      :      Tuple			,__lowerCamelCase      :      List[str]				):
   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 							UpperCAmelCase_						(		__lowerCamelCase      :      Tuple			,__lowerCamelCase      :      List[Any]			,__lowerCamelCase      :      List[Any]			,__lowerCamelCase      :      Optional[Any]=5_00			,__lowerCamelCase      :      int=None			,__lowerCamelCase      :      Optional[int]=False				):
   lowercase_    :Tuple					    =				initial_centroids[:]
   lowercase_    :Optional[Any]					    =				None

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

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

      # 2. Compute a new centroid for each of the k clusters, averaging all data
      #    points assigned to that cluster.
      lowercase_    :Optional[int]					    =				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_    :Optional[int]					    =				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_    :List[Any]					    =				compute_heterogeneity(__lowerCamelCase			,__lowerCamelCase			,__lowerCamelCase			,__lowerCamelCase				)
         record_heterogeneity.append(__lowerCamelCase				)

      lowercase_    :Optional[Any]					    =				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

 lowerCAmelCase      :  List[Any]      							=ds.load_iris()
 lowerCAmelCase      :  Any      							=3
 lowerCAmelCase      :  Dict      							=[]
 lowerCAmelCase      :  Union[str, Any]      							=get_initial_centroids(dataset['''data'''], k, seed=0)
 lowerCAmelCase							,	lowerCAmelCase      :  Optional[Any]      							=kmeans(
     dataset['''data'''],
     k,
     initial_centroids,
     maxiter=400,
     record_heterogeneity=heterogeneity,
     verbose=True,
 )
 plot_heterogeneity(heterogeneity, k)
def 							UpperCAmelCase_						(		__lowerCamelCase      :      pd.DataFrame			,__lowerCamelCase      :      np.ndarray			,__lowerCamelCase      :      Union[str, Any]=None				):
   # Fill missing values with given rules
   if fill_missing_report:
      lowercase_    :Optional[int]					    =				df.fillna(value=__lowerCamelCase				)
   lowercase_    :Any					    =				1
   lowercase_    :Optional[Any]					    =				df.select_dtypes(np.number				).columns
   lowercase_    :int					    =				(
       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 / 1_00				) : round(len(__lowerCamelCase				) * 75 / 1_00				)
                           ]				)				),
               ),
               ("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_    :str					    =				report[
       (report["Features"] == "dummy") & (report["Type"] == "count")
   ].copy()  # avoid SettingWithCopyWarning
   lowercase_    :List[str]					    =				(
       "ClusterSize"  # rename created cluster df to match report column names
   )
   lowercase_    :List[Any]					    =				"# of Customers"
   lowercase_    :Union[str, Any]					    =				pd.DataFrame(
       clustersize.iloc[:, 2:].values
       / clustersize.iloc[:, 2:].values.sum()				)
   lowercase_    :Union[str, Any]					    =				"% of Customers"  # rename created cluster df to match report column names
   lowercase_    :str					    =				"ClusterProportion"
   lowercase_    :Tuple					    =				clusterproportion.columns.tolist()
   lowercase_    :Any					    =				cols[-2:] + cols[:-2]
   lowercase_    :Any					    =				clusterproportion[cols]  # rearrange columns to match report
   lowercase_    :str					    =				report.columns
   lowercase_    :Dict					    =				pd.DataFrame(
       abs(
           report[report["Type"] == "count"].iloc[:, 2:].values
           - clustersize.iloc[:, 2:].values				)				)  # generating df with count of nan values
   lowercase_    :Dict					    =				0
   lowercase_    :Optional[int]					    =				"# of nan"
   lowercase_    :Any					    =				report[
       report["Type"] == "count"
   ].Features.tolist()  # filling values in order to match report
   lowercase_    :List[Any]					    =				a.columns.tolist()
   lowercase_    :Optional[Any]					    =				cols[-2:] + cols[:-2]
   lowercase_    :int					    =				a[cols]  # rearrange columns to match report
   lowercase_    :str					    =				report.columns  # rename columns to match report
   lowercase_    :Any					    =				report.drop(
       report[report.Type == "count"].index				)  # drop count values except cluster size
   lowercase_    :Tuple					    =				pd.concat(
       [report, a, clustersize, clusterproportion]			,axis=0				)  # concat report with clustert size and nan values
   lowercase_    :str					    =				report["Features"].isin(__lowerCamelCase				)
   lowercase_    :Union[str, Any]					    =				report.columns.tolist()
   lowercase_    :List[Any]					    =				cols[0:2] + cols[-1:] + cols[2:-1]
   lowercase_    :Any					    =				report[cols]
   lowercase_    :Dict					    =				{
       "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_    :Dict					    =				(
       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_    :Union[str, Any]					    =				""
   lowercase_    :int					    =				report.reset_index()
   lowercase_    :List[str]					    =				report.drop(columns=["index"]				)
   return report


if __name__ == "__main__":
 import doctest

 doctest.testmod()









'''simple docstring'''


class    a_     :





     def __init__(				self	:      str			,   lowercase	:      List[Any]		):





        """simple docstring"""




        lowercase_    :List[str]					    =				val
        # Number of nodes in left subtree
        lowercase_    :Optional[int]					    =				0
        lowercase_    :List[str]					    =				None
        lowercase_    :List[Any]					    =				None
        lowercase_    :Optional[Any]					    =				None







     def 							lowercase__   (				self	:      Union[str, Any]			,   lowercase	:      Dict		):





        """simple docstring"""




        assert self.left_tree_size == other.left_tree_size, "Unequal Sizes of Blocks"

        if self.val < other.val:
           lowercase_    :List[Any]					    =				self.right
           lowercase_    :str					    =				None
           if self.right:
              lowercase_    :Optional[int]					    =				other
           lowercase_    :Optional[Any]					    =				other
           lowercase_    :str					    =				self.left_tree_size * 2 + 1
           return self
        else:
           lowercase_    :Any					    =				other.right
           lowercase_    :Any					    =				None
           if other.right:
              lowercase_    :List[Any]					    =				self
           lowercase_    :Union[str, Any]					    =				self
           lowercase_    :Optional[int]					    =				other.left_tree_size * 2 + 1
           return other







class    a_     :





     def __init__(				self	:      Optional[Any]			,   lowercase	:      Any=None			,   lowercase	:      Optional[Any]=None			,   lowercase	:      int=0		):





        """simple docstring"""




        lowercase_    :Optional[int]					    =				heap_size
        lowercase_    :List[str]					    =				bottom_root
        lowercase_    :Dict					    =				min_node





     def 							lowercase__   (				self	:      List[str]			,   lowercase	:      List[Any]		):





        """simple docstring"""




        if other.size == 0:
           return None
        if self.size == 0:
           lowercase_    :Dict					    =				other.size
           lowercase_    :int					    =				other.bottom_root
           lowercase_    :Tuple					    =				other.min_node
           return None
        # Update size
        lowercase_    :Optional[int]					    =				self.size + other.size

        # Update min.node
        if self.min_node.val > other.min_node.val:
           lowercase_    :int					    =				other.min_node
        # Merge

        # Order roots by left_subtree_size
        lowercase_    :Any					    =				[]
        lowercase_    ,							lowercase_    :List[str]					    =				self.bottom_root, other.bottom_root
        while i or j:
           if i and ((not j) or i.left_tree_size < j.left_tree_size):
              combined_roots_list.append((i, True)		)
              lowercase_    :str					    =				i.parent
           else:
              combined_roots_list.append((j, False)		)
              lowercase_    :Dict					    =				j.parent
        # Insert links between them
        for i in range(len(lowercase		) - 1		):
           if combined_roots_list[i][1] != combined_roots_list[i + 1][1]:
              lowercase_    :Dict					    =				combined_roots_list[i + 1][0]
              lowercase_    :str					    =				combined_roots_list[i][0]
        # Consecutively merge roots with same left_tree_size
        lowercase_    :int					    =				combined_roots_list[0][0]
        while i.parent:
           if (
               (i.left_tree_size == i.parent.left_tree_size) and (not i.parent.parent)
           ) or (
               i.left_tree_size == i.parent.left_tree_size
               and i.left_tree_size != i.parent.parent.left_tree_size
           ):
              # Neighbouring Nodes
              lowercase_    :List[Any]					    =				i.left
              lowercase_    :Any					    =				i.parent.parent

              # Merging trees
              lowercase_    :Any					    =				i.merge_trees(i.parent		)

              # Updating links
              lowercase_    :List[Any]					    =				previous_node
              lowercase_    :int					    =				next_node
              if previous_node:
                 lowercase_    :int					    =				i
              if next_node:
                 lowercase_    :str					    =				i
           else:
              lowercase_    :str					    =				i.parent
        # Updating self.bottom_root
        while i.left:
           lowercase_    :List[str]					    =				i.left
        lowercase_    :List[str]					    =				i

        # Update other
        lowercase_    :Any					    =				self.size
        lowercase_    :Optional[Any]					    =				self.bottom_root
        lowercase_    :int					    =				self.min_node

        # Return the merged heap
        return self





     def 							lowercase__   (				self	:      int			,   lowercase	:      List[Any]		):





        """simple docstring"""




        if self.size == 0:
           lowercase_    :Dict					    =				Node(lowercase		)
           lowercase_    :Optional[Any]					    =				1
           lowercase_    :Optional[Any]					    =				self.bottom_root
        else:
           # Create new node
           lowercase_    :Union[str, Any]					    =				Node(lowercase		)

           # Update size
           self.size += 1

           # update min_node
           if val < self.min_node.val:
              lowercase_    :Union[str, Any]					    =				new_node
           # Put new_node as a bottom_root in heap
           lowercase_    :int					    =				new_node
           lowercase_    :List[str]					    =				self.bottom_root
           lowercase_    :Any					    =				new_node

           # Consecutively merge roots with same left_tree_size
           while (
               self.bottom_root.parent
               and self.bottom_root.left_tree_size
               == self.bottom_root.parent.left_tree_size
           ):
              # Next node
              lowercase_    :int					    =				self.bottom_root.parent.parent

              # Merge
              lowercase_    :Union[str, Any]					    =				self.bottom_root.merge_trees(self.bottom_root.parent		)

              # Update Links
              lowercase_    :Optional[Any]					    =				next_node
              lowercase_    :Any					    =				None
              if next_node:
                 lowercase_    :Union[str, Any]					    =				self.bottom_root





     def 							lowercase__   (				self	:      str		):





        """simple docstring"""




        return self.min_node.val





     def 							lowercase__   (				self	:      Optional[int]		):





        """simple docstring"""




        return self.size == 0





     def 							lowercase__   (				self	:      str		):





        """simple docstring"""




        lowercase_    :str					    =				self.min_node.val

        # Last element in heap corner case
        if self.size == 1:
           # Update size
           lowercase_    :Dict					    =				0

           # Update bottom root
           lowercase_    :int					    =				None

           # Update min_node
           lowercase_    :Tuple					    =				None

           return min_value
        # No right subtree corner case
        # The structure of the tree implies that this should be the bottom root
        # and there is at least one other root
        if self.min_node.right is None:
           # Update size
           self.size -= 1

           # Update bottom root
           lowercase_    :Optional[Any]					    =				self.bottom_root.parent
           lowercase_    :Tuple					    =				None

           # Update min_node
           lowercase_    :Dict					    =				self.bottom_root
           lowercase_    :List[str]					    =				self.bottom_root.parent
           while i:
              if i.val < self.min_node.val:
                 lowercase_    :Union[str, Any]					    =				i
              lowercase_    :Dict					    =				i.parent
           return min_value
        # General case
        # Find the BinomialHeap of the right subtree of min_node
        lowercase_    :List[str]					    =				self.min_node.right
        lowercase_    :Dict					    =				None
        lowercase_    :str					    =				bottom_of_new
        lowercase_    :Optional[int]					    =				1

        # Size, min_node and bottom_root
        while bottom_of_new.left:
           lowercase_    :List[Any]					    =				size_of_new * 2 + 1
           lowercase_    :Union[str, Any]					    =				bottom_of_new.left
           if bottom_of_new.val < min_of_new.val:
              lowercase_    :Optional[int]					    =				bottom_of_new
        # Corner case of single root on top left path
        if (not self.min_node.left) and (not self.min_node.parent):
           lowercase_    :Optional[int]					    =				size_of_new
           lowercase_    :Union[str, Any]					    =				bottom_of_new
           lowercase_    :Dict					    =				min_of_new
           # print("Single root, multiple nodes case")
           return min_value
        # Remaining cases
        # Construct heap of right subtree
        lowercase_    :Tuple					    =				BinomialHeap(
            bottom_root=lowercase			,   min_node=lowercase			,   heap_size=lowercase		)

        # Update size
        lowercase_    :Dict					    =				self.size - 1 - size_of_new

        # Neighbour nodes
        lowercase_    :Optional[Any]					    =				self.min_node.left
        lowercase_    :str					    =				self.min_node.parent

        # Initialize new bottom_root and min_node
        lowercase_    :int					    =				previous_node or next_node
        lowercase_    :Any					    =				next_node

        # Update links of previous_node and search below for new min_node and
        # bottom_root
        if previous_node:
           lowercase_    :int					    =				next_node

           # Update bottom_root and search for min_node below
           lowercase_    :str					    =				previous_node
           lowercase_    :List[Any]					    =				previous_node
           while self.bottom_root.left:
              lowercase_    :List[str]					    =				self.bottom_root.left
              if self.bottom_root.val < self.min_node.val:
                 lowercase_    :List[Any]					    =				self.bottom_root
        if next_node:
           lowercase_    :List[Any]					    =				previous_node

           # Search for new min_node above min_node
           lowercase_    :Dict					    =				next_node
           while i:
              if i.val < self.min_node.val:
                 lowercase_    :Optional[Any]					    =				i
              lowercase_    :str					    =				i.parent
        # Merge heaps
        self.merge_heaps(lowercase		)

        return min_value





     def 							lowercase__   (				self	:      Dict		):





        """simple docstring"""




        lowercase_    :Optional[Any]					    =				self.bottom_root
        while top_root.parent:
           lowercase_    :int					    =				top_root.parent
        # preorder
        lowercase_    :List[str]					    =				[]
        self.__traversal(lowercase			,   lowercase		)
        return heap_pre_order





     def 							lowercase__   (				self	:      Union[str, Any]			,   lowercase	:      List[Any]			,   lowercase	:      List[str]			,   lowercase	:      Dict=0		):





        """simple docstring"""




        if curr_node:
           preorder.append((curr_node.val, level)		)
           self.__traversal(curr_node.left			,   lowercase			,   level + 1		)
           self.__traversal(curr_node.right			,   lowercase			,   level + 1		)
        else:
           preorder.append(("#", level)		)







     def __str__(				self	:      Optional[int]		):





        """simple docstring"""




        if self.is_empty():
           return ""
        lowercase_    :int					    =				self.pre_order()

        return "\n".join(("-" * level + str(lowercase		)) for value, level in preorder_heap		)


# Unit Tests
if __name__ == "__main__":
 import doctest

 doctest.testmod()


