trustVis/sampeling.py

"""The Sampeling class serve as a helper module for retriving subject model data"""
from abc import ABC, abstractmethod

import os
import gc
import time

import numpy as np
from sklearn.neighbors import NearestNeighbors
from scipy.special import gamma
import math
from pynndescent import NNDescent
from sklearn.cluster import KMeans


"""
DataContainder module
1. calculate information entropy for singel sample and subset
2. sample informative subset
"""
class SampelingAbstractClass(ABC):
    
    def __init__(self, data_provider, epoch):
        self.mode = "abstract"
        self.data_provider = data_provider
        # self.model = model
        self.epoch = epoch
        
    @abstractmethod
    def info_calculator(self):
        pass

class Sampleing(SampelingAbstractClass):
    def __init__(self,data_provider, epoch, device):
        self.data_provider = data_provider
        # self.model = model
        self.epoch = epoch
        self.DEVICE = device
       

    def probability_density_cal(self,X,dim,k):
        """
        calculate the probability of each sample
        :param data: numpy.ndarray
        :param k: nearest neibour number
        :return: probability, numpy.ndarray
        """
        ## p(xi) = k / (n * V * σ)
        # step one: calculate volumes:  V = π^(d/2) / Γ(d/2 + 1) * r^d 
        print("start calculate the nbrs")
        # nbrs = NearestNeighbors(n_neighbors=k, algorithm='kd_tree').fit(X)
        # distances, indicates = nbrs.kneighbors(X)

        """
        construct a vietoris-rips complex
        """
        # number of trees in random projection forest
        n_trees = min(64, 5 + int(round((X.shape[0]) ** 0.5 / 20.0)))
        # max number of nearest neighbor iters to perform
        n_iters = max(5, int(round(np.log2(X.shape[0]))))
        # distance metric
        metric = "euclidean"
        # get nearest neighbors
        
        nnd = NNDescent(
            X,
            n_neighbors=k,
            metric=metric,
            n_trees=n_trees,
            n_iters=n_iters,
            max_candidates=60,
            verbose=True
        )
        indicates, distances = nnd.neighbor_graph
        print("finish calculate the nbrs")
        pred = self.data_provider.get_pred(self.epoch,X)
        d = dim  # dimensional
        volumes = []
        variances = []
        r_col = []
        print("start calculate the volumes and variances")
        for i in range(len(X)):
            r = distances[i, -1]  # farest neibour 's distance as r
            V = (math.pi**(d/2) / gamma(d/2 + 1)) * (r**d) +1e-8  # calculate the volumes
            volumes.append(V)
            r_col.append(r)
            # calculate prediction varians: σ
            neighbor_indices = indicates[i]
            neighbor_preds = pred[neighbor_indices]  # get predictions of k neighbors
            # Flatten the neighbor_preds array to a 1D array
            flatten_preds = neighbor_preds.flatten()

            # Calculate the variance for all predictions
            variance = np.var(flatten_preds)

            # variance = np.var(neighbor_preds, axis=0) # calculate variance for each class
            variances.append(variance)

        print("finsih calculate the volumes and variances")
        # step three: calculate the probability: p(xi) = k / (n * V * σ)
        n = len(X)
        probabilities = []
        for i in range(len(X)):
            # p = k / (n * volumes[i] * variances[i])
            p = k / (n * (r_col[i]) * (variances[i]))
            probabilities.append(p)
        
        return probabilities,volumes,variances,r_col


    def info_calculator(self):
        data = self.data_provider.train_representation(self.epoch)
        # data_nbrs = NearestNeighbors(n_neibour = )
    


    def clustering(self,data, n_clusters):

        kmeans = KMeans(n_clusters=n_clusters, random_state=0)
        # k-means
        kmeans.fit(data)

        # label
        labels = kmeans.labels_
        # print(labels)
        # clustering center
        centers = kmeans.cluster_centers_
        # print(centers)
        return labels,centers
    
    def space_split(self, data):
        self.pred = self.data_provider.get_pred(self.epoch, data)
        cluster_idx = 10
        print("clustering....")
        labels, centers = self.clustering(data, n_clusters=cluster_idx)
        print("clustering finfished")
   
        for i in range(10):
            subset_indices = np.where(labels == i)[0]
            subset = data[subset_indices]
            info = self.subset_info_cal(subset, centers[i])
        
        # 如果信息量大于0，继续分割
        print("info",info)
        if info > 0:
            labels, new_cluster_idx = self.split(data,subset_indices, labels, cluster_idx)
            cluster_idx = new_cluster_idx

        return labels
    
    def split(self, org_data,indices, labels, cluster_idx, m=1.5, n=1.5):
        data = org_data[indices]  # get the data subset from global data using indices
        print("data.shape",data.shape)
    
        # divide clustering
        sub_labels, centers = self.clustering(data, n_clusters=2)

        # calculate the information of the subset's  clustering
        info = []
        for i in range(2):
            subset_indices = indices[sub_labels == i]
            subset = data[sub_labels == i]
            info_i = self.subset_info_cal(subset, centers[i], m, n)
            info.append(info_i)

        # if information > 0
        for i in range(2):
            if info[i] > 0:
                subset_indices = indices[sub_labels == i]
                labels, cluster_idx = self.split(org_data, subset_indices, labels, cluster_idx+1, m, n)
            else:
                subset_indices = indices[sub_labels == i]
                labels[subset_indices] = cluster_idx
                cluster_idx += 1

        return labels, cluster_idx


    def subset_info_cal(self,data,center_data,m=1.5,n=1.5):
        """
        use infomration theroy quintify the information of each subset
        information = - log(p(d < m)) - log(p(a < n)) 
        """
        # calculat the center samples to each sample's distance
        dists = np.sqrt(np.sum((data - center_data)**2, axis=1))
        preds = self.data_provider.get_pred(self.epoch, data)
        pred_i = self.data_provider.get_pred(self.epoch, np.array([center_data]))
        # 计算预测值的误差，如果误差小于n，认为是True
        diffs = np.abs(preds - pred_i[0])

        # 计算满足条件的概率
        p_d = np.mean(dists < m) + 1e-8
        
        p_a = np.mean(diffs < n) + 1e-8
       
        print("p_d",p_d, "p_a",p_a)

        info = -np.log(p_d) - np.log(p_a)
        return info
    
    def sample_data(self, data, sample_ratio=0.2):
        all_indices = []  # store the selected indices
        labels = self.space_split(data)
        unique_labels = np.unique(labels)

        for label in unique_labels:
            indices = np.where(labels == label)[0]  # indices of data in the current cluster
            sample_size = int(len(indices) * sample_ratio)  # number of samples to select
            if sample_size == 0 and len(indices) > 0:  # in case sample_size is zero for small clusters, select at least one data point
                sample_size = len(indices)
            sampled_indices = np.random.choice(indices, size=sample_size, replace=False)  # select samples without replacement
            all_indices.append(sampled_indices)

        return np.concatenate(all_indices)  #