Define an abstract class for machine learning models from scikit.

cml/ports/scikit_adapter.py

+from collections import Counter
+from abc import ABCMeta, abstractmethod
+
+import sklearn.cluster
+from numpy import array, linspace, less, greater
+from scipy.signal import argrelextrema
+from sklearn.neighbors.kde import KernelDensity
+
+
+# TODO (dmt): Handle algorithms without cluster initialization!
+SCIKIT_CLUSTERING_TABLE = {
+    sklearn.cluster.KMeans: ("n_clusterss", "labels_"),
+    sklearn.cluster.birch.Birch: ("n_clusters", "labels_"),
+    sklearn.cluster.SpectralClustering: ("n_clusters", "labels_"),
+    sklearn.cluster.FeatureAgglomeration: ("n_clusters", "labels_"),
+    sklearn.cluster.AgglomerativeClustering: ("n_clusters", "labels_")
+}
+
+
+class MachineLearningModel:
+
+    @abstractmethod
+    def train(self, data):
+        pass
+
+
+class ConstructionClusteringMLModel(MachineLearningModel):
+    def __init__(self, model):
+        self.__model = model
+        self._cluster = 2
+
+    def get_labels(self):
+        return self.__model.__getattribute__(
+            SCIKIT_CLUSTERING_TABLE[type(self.__model)][1]
+        )
+
+    @property
+    def cluster(self):
+        return self._cluster
+
+    @cluster.setter
+    def cluster(self, value):
+        self.__model.__setattr__(
+            SCIKIT_CLUSTERING_TABLE[type(self.__model)][0], value
+        )
+
+    @property
+    def cluster_sizes(self):
+        labels = self.__model.__getattribute__(
+            SCIKIT_CLUSTERING_TABLE[type(self.__model)][1]
+        )
+        return Counter(labels)
+
+    @abstractmethod
+    def train(self, data):
+        self.__model.fit(data)
+        return self
+
+
+class KernelDensityEstimator(MachineLearningModel):
+
+    def __init__(self, kernel="gaussian", bandwidth=3, gridsize=256):
+        self.__model = None
+        self.kernel = kernel
+        self.bandwidth = bandwidth
+        self.gridsize = gridsize
+
+    def train(self, data):
+        reshaped_data = array(data).reshape(-1, 1)
+        if not self.__model:
+            self.__model = KernelDensity(kernel=self.kernel,
+                                         bandwidth=self.bandwidth)
+
+        self.__model.fit(reshaped_data)
+        return self
+
+    def density(self):
+        grid = linspace(0, self.gridsize)
+        reshaped_grid = grid.reshape(-1, 1)
+        return self.__model.score_samples(reshaped_grid)
+
+
+def find_relative_extrema(one_dim_data):
+    relative_min_values = argrelextrema(one_dim_data, less)
+    relative_max_values = argrelextrema(one_dim_data, greater)
+    return relative_min_values, relative_max_values