""" ======================================================================= 5.6.e Exploration Data Analysis of the Latent Space: Spherical Data - 2 ======================================================================= In this tutorial, we introduce the problem of conditional sampling, i.e., generating samples from a distribution \( p(y|x) \) using kernel-based models. We use the ConditionnerKernel class from Codpy to perform conditional sampling on synthetic spherical data with two clusters. We define a custom CircleConditionner class that inherits from ConditionerKernel and implements the sampling procedure. """ import numpy as np import pandas as pd from codpy import conditioning import codpy.core from codpy.kernel import Sampler from codpy.core import get_matrix from codpy.plot_utils import multi_plot,plot1D import matplotlib.pyplot as plt from codpy.data_processing import hot_encoder,simple_hot_encoder def sphere_sampling(center, radius, size, epsilon=0.01): from numpy import linalg as la samples = np.random.normal(size=size) noise = np.random.normal(size=size) * epsilon for n in range(samples.shape[0]): samples[n] *= radius / la.norm(samples[n]) samples += noise samples += center return samples def generate_sphere_data(N=500, D=2, centers=[[0, 1], [0, 0.5]], radius=1.0): num_clusters = len(centers) samples_list = [] labels = [] for idx, center in enumerate(centers): size = (N // num_clusters, D) samples = sphere_sampling(center=np.array(center), radius=radius, size=size) samples_list.append(samples) labels.extend([idx] * (N // num_clusters)) X = np.vstack(samples_list) df = pd.DataFrame(X, columns=[f"dim_{i}" for i in range(D)]) labels = pd.Series(labels, name="label") return df, labels def scatter_plot_multiple(dfs, titles, figsize=(14, 3.5)): """ Plots a row of scatter plots, with special label visualization for latent variables. """ fig, axes = plt.subplots(1, len(dfs), figsize=figsize) if len(dfs) == 1: axes = [axes] for ax, df, title in zip(axes, dfs, titles): if title == "Latent Representation": ax.scatter(df["dim_1"].values, df["label"], c=df["label"], cmap="viridis", s=10) ax.set_ylabel("Label values") ax.set_xlabel("Latent values") # # Show label predictions as a 1D classification result # ax.scatter(df["dim_0"], df["dim_1"], c=df["label"], cmap="viridis", s=10) # # plot1D(df.values[:,[1,0]].T) # ax.set_ylabel("Predicted Label") # ax.set_xlabel("Sample Index") else: sc = ax.scatter( df["dim_0"], df["dim_1"], c=df["label"], cmap="viridis", alpha=0.6, edgecolor="k", linewidth=0.2, ) ax.set_xlabel("dim_0") ax.set_ylabel("dim_1") # Optional: show color bar cbar = plt.colorbar(sc, ax=ax, shrink=0.75) cbar.set_label("Label") ax.set_title(title, fontsize=10) plt.tight_layout() plt.show() class CircleConditionner(conditioning.ConditionerKernel): def __init__( self, x, **kwargs, ): """ CircleConditionner is a specific implementation of ConditionerKernel that generates samples from a normal distribution conditioned on a circle in the latent space. Parameters: x (pd.DataFrame): Input data for conditioning. **kwargs: Additional keyword arguments for ConditionerKernel. """ latent_dim_y = 1 x = hot_encoder(pd.DataFrame(x),cat_cols_include={0}) super().__init__( x=x, latent_generator_x=None, latent_generator_y=None, latent_dim_y=latent_dim_y, **kwargs, ) def sample(self, x, n, **kwargs): """ Sample from the conditioned distribution. Parameters: x (pd.DataFrame): Input data for conditioning. n (int): Number of samples to generate. **kwargs: Additional keyword arguments. Returns: np.ndarray: Generated samples. """ # Generate samples from the latent space # Map latent samples to the original space x = simple_hot_encoder(x, num_classes=2) return super().sample(x=x,n=n,**kwargs) def sample_and_plot(samplers, N=500): """ Sample conditionally using provided samplers and plot marginals. """ y_df, y_labels = generate_sphere_data( N=500, D=2, centers=[[0, 1], [0, 0.5]], radius=1.0 ) y = y_df.values labels = y_labels.values results = [] for SamplerClass in samplers: # TODO: remove once conditionner fixed # SamplerClass.sample = dummy_sampler # Initialize and sample model = SamplerClass(x=labels, y=y) # Here should we use different x? How do we chose? x = y_labels sampled = model.sample(x=get_matrix(0.), n=N//2, distance=None).squeeze() sampled = np.concatenate([sampled,model.sample(x=get_matrix(1.), n=N//2).squeeze()]) sampler = model.sampler_y plt.scatter(sampler.get_fx()[:, 0], sampler.get_fx()[:, 1], color='red', label="original distrib.") plt.plot(sampler.get_fx()[:, 0], sampler.get_fx()[:, 1], alpha=0.5,color='black', label="latent connection.") plt.ylabel("y") plt.xlabel("x") plt.title("Parametrization of Original Data") plt.legend() # Latent latent_xy = model.latent_xy latent = pd.DataFrame( latent_xy[:,[0,2]], columns=[f"dim_{i}" for i in range(y.shape[1])] ) latent["label"] = latent_xy[:, 0] # Reconstructed recon = model.sampler_xy(latent_xy) xy_recon = pd.DataFrame(recon[:,2:], columns=[f"dim_{i}" for i in range(y.shape[1])]) xy_recon["label"] = recon[:,0] # New generated samples variate = pd.DataFrame( sampled, columns=[f"dim_{i}" for i in range(y.shape[1])] ) # Attach labels variate["label"] = x original = y_df.copy() original["label"] = labels scatter_plot_multiple( [original, latent, xy_recon, variate], titles=[ "Original Data", "Latent Representation", "Reconstructed", "Generated Samples", ], ) # codpy.core.KerInterface.set_verbose() # test = Sampler(x=np.random.normal(size=[10,1])) # plt.plot(test.x,test.fx,'o') sample_and_plot( samplers=[ CircleConditionner, ] ) plt.show() pass