Source code for meld.meld

# Copyright (C) 2020 Krishnaswamy Lab, Yale University

import numpy as np
import pandas as pd
import graphtools

from . import utils
from . import filter
from graphtools.estimator import GraphEstimator, attribute
from functools import partial

[docs]class MELD(GraphEstimator): """MELD operator for filtering signals over a graph. Parameters ---------- beta : int, optional, Default: 60 Amount of smoothing to apply. Default value of 60 determined through analysis of simulated data using Splatter. offset: float, optional, Default: 0 Amount to shift the MELD filter in the eigenvalue spectrum. Recommend using an eigenvalue from the graph based on the spectral distribution. Should be in interval [0,1] order: int, optional, Default: 1 Falloff and smoothness of the filter. High order leads to square-like filters. filter: str, optional, Default: 'heat' Filter type to use. Should be in ['heat', 'laplacian'] solver : string, optional, Default: 'chebyshev' Method to solve convex problem. 'chebyshev' uses a chebyshev polynomial approximation of the corresponding filter. 'exact' uses the eigenvalue solution to the problem chebyshev_order : int, optional, Default: 50 Order of chebyshev approximation to use. lap_type : ('combinatorial', 'normalized'), Default: 'combinatorial' The kind of Laplacian to calculate sample_normalize : boolean, optional, Default: True If True, the sample indicator vectors are column normalized to sum to 1 """ # parameters beta = attribute( "beta", doc="Amount of smoothing to apply. Default value of 60 determined through" "analysis of simulated data using Splatter", default=40, on_set=graphtools.utils.check_positive, ) offset = attribute( "offset", doc="Amount to shift the MELD filter in the eigenvalue spectrum." "Recommend using an eigenvalue from the graph based on the" "spectral distribution. Should be in interval [0,1]", default=0, ) order = attribute( "order", doc="Falloff and smoothness of the filter." "High order leads to square-like filters.", default=1, ) filter = attribute( "filter", default="heat", doc="Filter type to use. Should be in ['heat', 'laplacian']", on_set=partial(graphtools.utils.check_in, ["heat", "laplacian"]), ) solver = attribute( "solver", default="chebyshev", doc="Method to solve convex problem." "'chebyshev' uses a chebyshev polynomial approximation of the corresponding" "filter. 'exact' uses the eigenvalue solution to the problem", on_set=partial(graphtools.utils.check_in, ["chebyshev", "exact"]), ) chebyshev_order = attribute( "chebyshev_order", default=30, doc="Order of chebyshev approximation to use.", on_set=[graphtools.utils.check_int, graphtools.utils.check_positive], ) lap_type = attribute( "lap_type", default="combinatorial", doc="The kind of Laplacian to calculate", on_set=partial(graphtools.utils.check_in, ["combinatorial", "normalized"]), ) # stored attributes sample_densities = attribute( "sample_densities", doc="Density associated with each sample" ) def __init__( self, beta=60, offset=0, order=1, filter="heat", solver="chebyshev", chebyshev_order=50, lap_type="combinatorial", sample_normalize=True, anisotropy=1, n_landmark=None, **kwargs ): self.beta = beta self.offset = offset self.order = order self.solver = solver self.chebyshev_order = chebyshev_order self.lap_type = lap_type self.filter = filter self.sample_normalize = sample_normalize kwargs["use_pygsp"] = True super().__init__(anisotropy=anisotropy, n_landmark=n_landmark, **kwargs) def _reset_graph(self): self._reset_filter() def _reset_filter(self): self.filt = None self.sample_densities = None def set_params(self, **params): for p in [ "beta", "offset", "order", "solver", "chebyshev_order", "lap_type", "filter", ]: if p in params and params[p] != getattr(self, p): self._reset_filter() setattr(self, p, params[p]) del params[p] super().set_params(**params) def _create_sample_indicators(self, sample_labels): """ Helper function to take an array-like of non-numerics and produce a collection of sample indicator vectors. """ self.sample_labels_ = sample_labels self.samples = np.unique(sample_labels) try: labels = sample_labels.values except AttributeError: labels = self.sample_labels_ if len(labels.shape) > 1: # If you have a 2D array if labels.shape[1] == 1: # If it's just a column-vector, reshape it labels = labels.reshape(-1) else: # If its got multiple-columns, raise Error raise ValueError( "sample_labels must be a single column. Got" "shape={}".format(labels.shape) ) if self.samples.shape[0] == 2: # When there's two samples (i.e. [A, A, B, B]) # LabelBinarizer doesn't work nicely with only two labels # This creates a two-column dataframe using the sample labels df = pd.DataFrame( [labels == self.samples[0], labels == self.samples[1]], columns=self._labels_index, ).astype(int) df.index = self.samples self.sample_indicators = df.T else: # We have more than two samples, use label binarizer. import sklearn self._LB = sklearn.preprocessing.LabelBinarizer() sample_indicators = self._LB.fit_transform(self.sample_labels_) self.sample_indicators = pd.DataFrame( sample_indicators, columns=self._LB.classes_ ) return self.sample_indicators
[docs] def transform(self, sample_labels): """Filters a collection of sample_indicators over the data graph. Parameters ---------- sample_indicators : ndarray [n, p] 1- or 2-dimensional sample indicator array to filter. Returns ------- sample_densities: ndarray [n, p] A density estimate for each sample. """ self.graph = utils._check_pygsp_graph(self.graph) self._sample_labels = sample_labels if sample_labels.shape[0] != self.graph.N: raise ValueError( "Input data ({}) and input graph ({}) " "are not of the same size".format(sample_labels.shape, self.graph.N) ) if len(np.unique(sample_labels)) == 1: raise ValueError( "Found only one unqiue sample label. Cannot estimate density " "of a single sample." ) # self._label_cls = type(sample_labels) if hasattr(sample_labels, "index"): self._labels_index = sample_labels.index else: self._labels_index = None self._create_sample_indicators(sample_labels) if self.sample_normalize: self.sample_indicators = ( self.sample_indicators / self.sample_indicators.sum(axis=0) ) # apply filter densities = filter.filter( signal=self.sample_indicators, graph=self.graph, filter=self.filter, beta=self.beta, offset=self.offset, order=self.order, solver=self.solver, chebyshev_order=self.chebyshev_order, ) self.sample_densities = pd.DataFrame( densities, index=self._labels_index, columns=self.sample_indicators.columns ) return self.sample_densities
[docs] def fit_transform(self, X, sample_labels, **kwargs): """Builds the MELD filter over a graph built on data `X` and estimates density of each sample in `sample_labels` Parameters ---------- X : array-like, shape=[n_samples, m_features] Data on which to build graph to perform data smoothing over. sample_labels : array-like, shape=[n_samples, p_signals] 1- or 2-dimensional array of non-numerics indicating the sample origin for each cell. kwargs : additional arguments for graphtools.Graph Returns ------- sample_densities : ndarray, shape=[n_samples, p_signals] Density estimate for each sample over a graph built from X """, **kwargs) return self.transform(sample_labels)