import io import math import pickle import imageio import numpy as np import matplotlib.pyplot as plt from tqdm import tqdm from moviepy.editor import ImageSequenceClip class ClusterSOM: def __init__(self): self.hdbscan_model = None self.som_models = {} self.sigma_values = {} self.mean_values = {} self.cluster_mapping = {} self.embedding = None self.dim_red_op = None def load(self, file_path): """ Load a ClusterSOM model from a file. """ with open(file_path, "rb") as f: model_data = pickle.load(f) self.hdbscan_model, self.som_models, self.mean_values, self.sigma_values, self.cluster_mapping = model_data[:5] if len(model_data) > 5: self.label_centroids, self.label_encodings = model_data[5:] def predict(self, data, sigma_factor=1.5): """ Predict the cluster and BMU SOM coordinate for each sample in the data if it's inside the sigma value. Also, predict the label and distance to the center of the label if labels are trained. """ results = [] for sample in data: min_distance = float('inf') nearest_cluster_idx = None nearest_node = None for i, som in self.som_models.items(): x, y = som.winner(sample) node = som.get_weights()[x, y] distance = np.linalg.norm(sample - node) if distance < min_distance: min_distance = distance nearest_cluster_idx = i nearest_node = (x, y) # Check if the nearest node is within the sigma value if min_distance <= self.mean_values[nearest_cluster_idx][nearest_node] * 1.5: # * self.sigma_values[nearest_cluster_idx][nearest_node] * sigma_factor: if hasattr(self, 'label_centroids'): # Predict the label and distance to the center of the label label_idx = self.label_encodings.inverse_transform([nearest_cluster_idx - 1])[0] label_distance = np.linalg.norm(sample - self.label_centroids[label_idx]) results.append((nearest_cluster_idx, nearest_node, label_idx, label_distance)) else: results.append((nearest_cluster_idx, nearest_node)) else: results.append((-1, None)) # Noise return results # rearranging the subplots in the closest square format def rearrange_subplots(self, num_subplots): # Calculate the number of rows and columns for the subplot grid num_rows = math.isqrt(num_subplots) num_cols = math.ceil(num_subplots / num_rows) # Create the figure and subplots fig, axes = plt.subplots(num_rows, num_cols, sharex=True, sharey=True) # Flatten the axes array if it is multidimensional if isinstance(axes, np.ndarray): axes = axes.flatten() # Hide any empty subplots for i in range(num_subplots, len(axes)): axes[i].axis('off') return fig, axes def plot_activation(self, data, start=None, end=None): """ Generate a GIF visualization of the prediction output using the activation maps of individual SOMs. """ if len(self.som_models) == 0: raise ValueError("SOM models not trained yet.") if start is None: start = 0 if end is None: end = len(data) images = [] for sample in tqdm(data[start:end], desc="Visualizing prediction output"): prediction = self.predict([sample])[0] fig, axes = self.rearrange_subplots(len(self.som_models)) # fig, axes = plt.subplots(1, len(self.som_models), figsize=(20, 5), sharex=True, sharey=True) fig.suptitle(f"Activation map for SOM {prediction[0]}, node {prediction[1]}", fontsize=16) for idx, (som_key, som) in enumerate(self.som_models.items()): ax = axes[idx] activation_map = np.zeros(som._weights.shape[:2]) for x in range(som._weights.shape[0]): for y in range(som._weights.shape[1]): activation_map[x, y] = np.linalg.norm(sample - som._weights[x, y]) winner = som.winner(sample) # Find the BMU for this SOM activation_map[winner] = 0 # Set the BMU's value to 0 so it will be red in the colormap if som_key == prediction[0]: # Active SOM im_active = ax.imshow(activation_map, cmap='viridis', origin='lower', interpolation='none') ax.plot(winner[1], winner[0], 'r+') # Mark the BMU with a red plus sign ax.set_title(f"A {som_key}", color='blue', fontweight='bold', fontsize=10) if hasattr(self, 'label_centroids'): label_idx = self.label_encodings.inverse_transform([som_key - 1])[0] ax.set_xlabel(f"Label: {label_idx}", fontsize=12) else: # Inactive SOM im_inactive = ax.imshow(activation_map, cmap='gray', origin='lower', interpolation='none') ax.set_title(f"A {som_key}", fontsize=10) ax.set_xticks([]) ax.set_yticks([]) ax.grid(True, linestyle='-', linewidth=0.5) # Create a colorbar for each frame plt.tight_layout() fig.subplots_adjust(wspace=0, hspace=0) # Save the plot to a buffer buf = io.BytesIO() plt.savefig(buf, format='png') buf.seek(0) img = imageio.imread(buf) images.append(img) plt.close() # Create the video using moviepy and save it as a mp4 file video = ImageSequenceClip(images, fps=1) return video def plot_activation_v2(self, data, slice_select): """ Generate a GIF visualization of the prediction output using the activation maps of individual SOMs. """ if len(self.som_models) == 0: raise ValueError("SOM models not trained yet.") try: prediction = self.predict([data[int(slice_select)-1]])[0] except: prediction = self.predict([data[int(slice_select)-2]])[0] fig, axes = plt.subplots(1, len(self.som_models), figsize=(20, 5), sharex=True, sharey=True) fig.suptitle(f"Activation map for SOM {prediction[0]}, node {prediction[1]}", fontsize=16) for idx, (som_key, som) in enumerate(self.som_models.items()): ax = axes[idx] activation_map = np.zeros(som._weights.shape[:2]) for x in range(som._weights.shape[0]): for y in range(som._weights.shape[1]): activation_map[x, y] = np.linalg.norm(data[int(slice_select)-1] - som._weights[x, y]) winner = som.winner(data[int(slice_select)-1]) # Find the BMU for this SOM activation_map[winner] = 0 # Set the BMU's value to 0 so it will be red in the colormap if som_key == prediction[0]: # Active SOM im_active = ax.imshow(activation_map, cmap='viridis', origin='lower', interpolation='none') ax.plot(winner[1], winner[0], 'r+') # Mark the BMU with a red plus sign ax.set_title(f"SOM {som_key}", color='blue', fontweight='bold') if hasattr(self, 'label_centroids'): label_idx = self.label_encodings.inverse_transform([som_key - 1])[0] ax.set_xlabel(f"Label: {label_idx}", fontsize=12) else: # Inactive SOM im_inactive = ax.imshow(activation_map, cmap='gray', origin='lower', interpolation='none') ax.set_title(f"SOM {som_key}") ax.set_xticks(range(activation_map.shape[1])) ax.set_yticks(range(activation_map.shape[0])) ax.grid(True, linestyle='-', linewidth=0.5) plt.tight_layout() return fig