GSoC 2021: Implementing dimensionality reduction techniques that run remotely on a server

README.md

@@ -1,4 +1,5 @@
 # This is git repository for the Cytoscape 3 clustering app: clusterMaker
+
 TODO:
 1) Fix BiClusterView
 2) Add support for BiClustering routines: 
@@ -11,6 +12,34 @@ TODO:
 
 
 
+
+
+7 JUNE - 23 AUGUST 2021 Google Summer of Code by Maija Utriainen
+
+Implementing remote dimensionality reduction techniques using similar approach to last years remote network clusterers (see below).
+  - Dimensionality reduction techniques added: Isomap, Local Linear Embedding, MDS, Spectral, tSNE, UMAP
+  - Changes to ClusterJobExecutionService and RemoteServer: abstracting and replacing code to another class
+  - Subclasses to ClusterJobHandler specific to network clusterers and dimensionality reduction techniques
+  - The new algorithms registered in CyActivator
+
+The added and edited code can be found in clusterMaker2/src/main/java/edu/ucsf/rbvi/clusterMaker2/internal/
+  - utils/remoteUtils
+    - RemoteServer 
+    - ClusterJobExecutionService
+    - NetworkClusterJobHandler
+    - DimensionalityReductionJobHandler
+  - algorithms/dimensionalityRedcution
+    - isomap
+    - linearEmbedding
+    - mds
+    - spectral
+    - tSNERemote
+    - umap
+  - CyActivator 
+
+
+
+
 1 JUNE - 29 AUGUST 2020 changes done as a Google Summer of Code Project by Maija Utriainen
 
 Implementing new algorithms with a new approach that runs the algorithm remotely on a server instead of in clusterMaker/Cytoscape utilizing interfaces in cytoscape.jobs package. 

api/api/algorithms/algorithms.py

@@ -0,0 +1,56 @@
+from api.jobs import Jobs
+from .fastgreedy import FastGreedy
+from .infomap import Infomap
+from .leiden import Leiden
+from .label_propagation import LabelPropagation
+from .leading_eigenvector import LeadingEigenvector
+from .multilevel import Multilevel
+from .dbscan import DBScan
+from .umap import UMAP
+from .tsne import TSNEREMOTE
+from .isomap import IsoMapEmbedding
+from .mds import MDS
+from .spectral import SpectralEmbedding
+#from .lineardisc import LinearDiscriminant
+from .lle import LLE
+import falcon
+import json
+
+class Algorithms(object):
+    jobs = None
+    algorithms = None
+
+    def __init__(self, jobs: Jobs):
+        self.jobs = jobs
+        self.algorithms = {}
+
+        # Initialize the list of algorithms
+        self.algorithms["leiden"] = Leiden(jobs)
+        self.algorithms["fastgreedy"] = FastGreedy(jobs)
+        self.algorithms["infomap"] = Infomap(jobs)
+        self.algorithms["labelpropagation"] = LabelPropagation(jobs)
+        self.algorithms["leadingeigenvector"] = LeadingEigenvector(jobs)
+        self.algorithms["multilevel"] = Multilevel(jobs)
+        self.algorithms["dbscan"] = DBScan(jobs)
+
+        # Dimensionality reduction (manifold) techniques
+        self.algorithms["umap"] = UMAP(jobs)
+        self.algorithms["tsneremote"] = TSNEREMOTE(jobs)
+        self.algorithms["mds"] = MDS(jobs)
+        self.algorithms["isomap"] = IsoMapEmbedding(jobs)
+        self.algorithms["spectral"] = SpectralEmbedding(jobs)
+        self.algorithms["lle"] = LLE(jobs)
+        # self.algorithms["lineardisc"] = LinearDiscriminant(jobs)
+
+    def on_get(self, req: falcon.Request, resp: falcon.Response):
+        resp.code = falcon.HTTP_200
+        resp.text = '{"algorithms":'+json.dumps(list(self.algorithms.keys()))+'}'
+
+
+    def get_algorithms(self) -> list:
+        return self.algorithms.keys()
+
+    def get_algorithm(self, algorithm:str):
+        if algorithm in self.algorithms:
+            return self.algorithms[algorithm]
+        return None

api/api/algorithms/base_algorithm.py

@@ -0,0 +1,108 @@
+"""
+Algorithm base class
+"""
+
+from uuid import UUID
+import json
+
+import falcon
+import logging
+
+from multiprocessing import Manager,Process
+from datetime import datetime
+
+from api.jobs import Jobs
+import api.utils as utils
+
+class BaseAlgorithm():
+    jobs = None
+    myjobs = {}
+    logFile = None
+
+    def __init__(self, jobs: Jobs):
+        self.jobs = jobs
+        self.manager = jobs.get_manager()
+
+    def on_post(self, req: falcon.Request, resp: falcon.Response):
+        """ Get data and arguments """
+        result = self.manager.dict()
+        status = self.manager.dict()
+
+        #self.logFile.write(repr(req)+"\n")
+        self.log(repr(req))
+
+        # Get our parameters
+        args = self.get_args(req)
+        self.log("Arguments: "+repr(args))
+        #self.logFile.write(repr(args))
+
+        # We need to do the load here because we can't pass a stream to our
+        # child process
+        if hasattr(req.get_param('data'),'file'):
+            # Python requests adds an extra dict
+            args['json_data'] = json.load(req.get_param('data').file)
+        else:
+            # Assume it's just straight text
+            args['json_data'] = json.loads(req.get_param('data'))
+
+        #f = open("/var/tmp/wsgi.log", "w")
+        #f.write(str(args['json_data']))
+
+        uuid = self.jobs.create_job(req.path, self)
+        proc = Process(target=self.community_detection, args=(args, status, result))
+        self.myjobs[uuid] = (proc, status, result)
+        proc.start()
+        resp.code = falcon.HTTP_200
+        resp.text = json.dumps({'job_id': str(uuid)})
+
+    def get_status2(self, uid: UUID) -> dict:
+        if uid in self.myjobs:
+            (proc, status, result) = self.myjobs[uid]
+            json_status = {}
+            for key, value in status.items():
+                json_status[key] = value
+
+            return json_status
+        return None
+
+    def get_status(self, uid: UUID) -> str:
+        if uid in self.myjobs:
+            (proc, status, result) = self.myjobs[uid]
+            #self.logFile.write("Getting status for "+repr(uid)+"\n")
+            #self.logFile.write("Status = "+repr(status['status'])+"\n")
+            self.log("Status for "+repr(uid)+" = "+repr(status['status']))
+            return status['status']
+        return None
+
+    def fetch_results(self, uid: UUID, req: falcon.Request, resp: falcon.Response):
+        if uid in self.myjobs:
+            (proc, status, result) = self.myjobs[uid]
+            # Add our response
+            resp.text = utils.get_json_result(status, result)
+            if status['status'] == 'done':
+              resp.code = falcon.HTTP_200
+            elif status['status'] == 'error':
+              resp.code = falcon.HTTP_500
+              self.log("Returning 500. Response = "+repr(resp.text))
+        else:
+            resp.code = falcon.HTTP_400
+            resp.text = str({'error':"no such job: "+str(uid)})
+
+    def terminate(self, uid: UUID, req: falcon.Request, resp: falcon.Response):
+        # Terminate the running process
+        if uid in self.myjobs:
+            (proc, status, result) = self.myjobs[uid]
+            proc.terminate()
+            self.jobs.remove_job(uid)
+            del self.myjobs[uid]
+            resp.text = "Job: "+str(uid)+" terminated"
+            resp.code = falcon.HTTP_200
+        else:
+            resp.code = falcon.HTTP_400
+            resp.text = str({'error':"no such job: "+str(uid)})
+
+    def log(self, message: str):
+        now = datetime.today()
+        logging.info(str(now)+": "+message)
+
+

api/api/algorithms/dbscan.py

@@ -0,0 +1,74 @@
+"""
+DBScan cluster algorithm
+"""
+
+import numpy as np
+import falcon
+import api.utils as utils
+from .base_algorithm import BaseAlgorithm
+from sklearn.preprocessing import StandardScaler
+from sklearn.cluster import DBSCAN
+from scipy.sparse import csgraph
+
+class DBScan(BaseAlgorithm):
+
+    def get_args(self, req: falcon.Request) -> dict:
+        """ Get the arguments """
+
+        # Get our parameters
+        args = {}
+        args['eps'] = utils.get_param_as_float(req, 'eps', 0.5)
+        args['min_samples'] = utils.get_param_as_int(req, 'min_samples', 5)
+        args['metric'] = utils.get_param_as_string(req, 'metric', 'euclidean')
+        args['algorithm'] = utils.get_param_as_string(req, 'algorithm', 'auto')
+        args['leaf_size'] = utils.get_param_as_int(req, 'leaf_size', 30)
+        args['p'] = utils.get_param_as_float(req, 'p', 2)
+
+        return args
+
+    def community_detection(self, args:dict, status:dict, result:dict):
+        status['status'] = 'running'
+
+        # Get our parameters
+        eps = args['eps']
+        min_samples = args['min_samples']
+        metric = args['metric']
+        algorithm = args['algorithm']
+        leaf_size = args['leaf_size']
+        p = args['p']
+        data = args['json_data']
+
+        # We handle two different types of data -- either a square matrix with
+        # weights in the cells or a matrix with columns representing features
+        # If the metric is 'precomputed' then we're dealing with a square matrix
+        # otherwise, we assume we have a standard column matrix
+        if metric == 'precomputed':
+          # Get the graph
+          graph = utils.get_graph(data)
+          # Get the matrix from the graph
+          data = graph.get_adjacency_sparse(attribute="weights")
+        else:
+          df = utils.get_matrix(data)
+          columns = df.columns.to_list()
+
+          # We need to drop any NaNs
+          df = df.dropna()
+
+          data = df[columns[1:]].values # skip over the label and just pull the data
+
+          if (args['scale']):
+            data = StandardScaler().fit_transform(data)
+
+
+        db = DBSCAN(eps=eps, min_samples=min_samples, metric=metric,metric_params=None,
+                    algorithm=algorithm,leaf_size=leaf_size,p=p,n_jobs=None).fit(data) 
+
+        core_samples_mask = np.zeros_like(db.labels_, dtype=bool)
+        core_samples_mask[db.core_sample_indices_] = True
+        # List of labels -- each element refers to a row
+        labels = db.labels_
+
+        result['partitions'] = labels
+
+        status['status'] = 'done'
+

api/api/algorithms/fastgreedy.py

@@ -0,0 +1,38 @@
+"""
+Fast Greedy cluster algorithm
+"""
+
+import falcon
+import logging
+import api.utils as utils
+from api.jobs import Jobs
+from .base_algorithm import BaseAlgorithm
+
+class FastGreedy(BaseAlgorithm):
+
+    def get_args(self, req: falcon.Request) -> dict:
+        return {}
+
+    def community_detection(self, args:dict, status:dict, result:dict):
+        status['status'] = 'running'
+
+        # Get our parameters
+        data = args['json_data']
+
+        # Get our data file
+        graph = utils.get_graph(data)
+
+        try:
+          # FastGreedy doesn't work for multigraphs
+          graph = graph.simplify(multiple=True, loops=True, combine_edges=sum);
+          part = graph.community_fastgreedy(weights="weights")
+        except Exception as e:
+          exc = utils.parse_igraph_exception(repr(e))
+          status['status'] = 'error'
+          status['message'] = exc
+          return
+
+        result['partitions'] = utils.get_vertex_list(graph, part.as_clustering())
+
+        status['status'] = 'done'
+

api/api/algorithms/infomap.py

@@ -0,0 +1,34 @@
+"""
+Infomap cluster algorithm
+"""
+import falcon
+import api.utils as utils
+from .base_algorithm import BaseAlgorithm
+
+class Infomap(BaseAlgorithm):
+    def get_args(self, req: falcon.Request) -> dict:
+        args = {}
+        args['trials'] = utils.get_param_as_int(req, 'trials', 10)
+        return args
+
+    def community_detection(self, args:dict, status:dict, result:dict):
+        status['status'] = 'running'
+
+        # Get our parameters
+        trials = args['trials']
+        data = args['json_data']
+
+        graph = utils.get_graph(data)
+
+        try:
+          part = graph.community_infomap(edge_weights="weights", trials=trials)
+        except Exception as e:
+          exc = utils.parse_igraph_exception(repr(e))
+          status['status'] = 'error'
+          status['message'] = exc
+          return
+
+        result['partitions'] = utils.get_vertex_list(graph, part)
+
+        status['status'] = 'done'
+