Merge pull request #9 from UCSC-MedBook/develop

ExpressionVariance filters -> prod

Dockerfile

@@ -24,6 +24,9 @@ RUN easy_install pip
 RUN pip install --upgrade virtualenv
 RUN pip install pymongo
 
+# For expression and variance filters
+RUN pip install numpy
+
 # needed in the outlier analysis
 RUN apt-get install -y bc
 

README.md

@@ -5,9 +5,11 @@
 1. create a feature branch in git
 2. look at https://github.com/UCSC-MedBook/MedBook-JobRunner/blob/f-gsea-new/webapp/server/classes/RunLimmaGSEA.js
 3. create a new class
+4. Add new class and its args to primary-collections repo collections/Jobs.js
 4. add adapters (importers and exporters) to convert from MedBook objects to files that tools understand and store check them into external-tools
-4. add external code to external-tools repo (or mechansim to install it)
-5. add pointers to external code in settings.json
+4. add external code to external-tools repo (or mechanism to install it)
+5. add pointers to external code in your personal settings.json for use while testing
+  - also add these pointers in the MedBook main repo docker-compose.yml METEOR_SETTINGS environment variable
 6. add gui to appropriate MedBook app, that initiates job by inserting into jobs collection
   for example: 
    Jobs.insert({

config/ekephart/run.sh → config/etk/run.sh

@@ -1,4 +1,4 @@
 export MONGO_URL="mongodb://localhost:27017/MedBook"
 export MEDBOOK_FILESTORE=/tmp/filestore
 
-meteor --port 3003 --settings ../config/ekephart/settings.json
+meteor --port 3003 --settings ../config/etk/settings.json

config/etk/settings.json

@@ -0,0 +1,15 @@
+{
+  "sh": "/bin/sh",
+  "rscript": "Rscript",
+  "limma_path": "/Users/Shared/external-tools/limma/limma_ng.R",
+  "outlier_analysis": "/Users/Shared/external-tools/OutlierAnalysis/outlier-analysis.sh",
+  "calculate_outlier_genes": "/Users/Shared/external-tools/OutlierAnalysis/calculate_outlier_genes.R",
+  "genomic_expression_export": "/Users/Shared/external-tools/exporters/genomic_expression_export.py",
+  "gene_set_collection_export": "/Users/Shared/external-tools/exporters/gene_set_collection_export.py",
+  "limma_phenotype_export": "/Users/Shared/external-tools/exporters/limma_phenotype_export.py",
+  "gsea_path": "/Users/Shared/external-tools/gsea/rgGSEA.py",
+  "gsea_jar_path": "/Users/Shared/external-tools/gsea/gsea2-2.2.2.jar",
+  "python": "python",
+  "expression_level_gene_filter": "/Users/Shared/external-tools/ExpressionAndVarianceFilters/filter_out_genes_unexpressed_in_most_samples.py",
+  "variance_gene_filter": "/Users/Shared/external-tools/ExpressionAndVarianceFilters/filter_out_lowest_varying_genes.py"
+}

config/teo/settings.json

@@ -8,5 +8,8 @@
   "gene_set_collection_export": "/Users/mokolodi1/work/medbook/medbook/MedBook-JobRunner/external-tools/exporters/gene_set_collection_export.py",
   "limma_phenotype_export": "/Users/mokolodi1/work/medbook/medbook/MedBook-JobRunner/external-tools/exporters/limma_phenotype_export.py",
   "gsea_path": "/Users/mokolodi1/work/medbook/medbook/MedBook-JobRunner/external-tools/gsea/rgGSEA.py",
-  "gsea_jar_path": "/Users/mokolodi1/work/medbook/medbook/MedBook-JobRunner/external-tools/gsea/gsea2-2.2.2.jar"
+  "gsea_jar_path": "/Users/mokolodi1/work/medbook/medbook/MedBook-JobRunner/external-tools/gsea/gsea2-2.2.2.jar",
+  "python": "python",
+  "expression_level_gene_filter": "/Users/mokolodi1/work/medbook/medbook/MedBook-JobRunner/external-tools/ExpressionAndVarianceFilters/filter_out_genes_unexpressed_in_most_samples.py",
+  "variance_gene_filter": "/Users/mokolodi1/work/medbook/medbook/MedBook-JobRunner/external-tools/ExpressionAndVarianceFilters/filter_out_lowest_varying_genes.py"
 }

webapp/server/classes/ApplyExprAndVarianceFilters.js

@@ -0,0 +1,100 @@
+
+function ApplyExprAndVarianceFilters (job_id) {
+  Job.call(this, job_id);
+}
+
+ApplyExprAndVarianceFilters.prototype = Object.create(Job.prototype);
+ApplyExprAndVarianceFilters.prototype.constructor = ApplyExprAndVarianceFilters;
+
+ApplyExprAndVarianceFilters.prototype.run = function () {
+
+  var workDir = ntemp.mkdirSync("ApplyExprAndVarianceFilters");
+  console.log("workDir: ", workDir);
+
+  var self = this;
+  var sample_group_id = self.job.args.sample_group_id;
+
+  var exportScript = getSetting("genomic_expression_export");
+  var python = getSetting("python");
+  var exprFilterScript = getSetting("expression_level_gene_filter");
+  var varianceFilterScript = getSetting("variance_gene_filter");
+
+  var deferred = Q.defer();
+
+  // exportCommand is a promise for exporting the
+  // sample group data
+  var exportCommand = spawnCommand(exportScript, [
+      "--sample_group_id", sample_group_id,
+    ], workDir)  
+
+  // Export the data
+  exportCommand.then(function(exportResults){ 
+    if(exportResults.exitCode !== 0){
+      throw new Error("Writing file failed (exit code not 0)");
+    }
+    // input & output paths for expression level filter
+    self.sampleGroupPath = exportResults.stdoutPath;
+    self.filteredByExpressionPath = path.join(workDir, "sampleGroup_with_expr_filter_applied.tsv");
+
+      return spawnCommand(python,
+        [
+          exprFilterScript,
+          "--in_file", self.sampleGroupPath,
+          "--proportion_unexpressed", "0.8",
+          "--out_file", self.filteredByExpressionPath,
+        ],
+        workDir);
+
+    }).then(function(exprFilterResults){
+      console.log("expression level filter ran with results", exprFilterResults);
+      if(exprFilterResults.exitCode !== 0){
+        throw new Error("Failed to apply expression-level filter (exit code not 0)");
+      }
+      // set up variance filter
+
+      // End users will see custom path that's set in patientCare when downloading
+      // but internally this path is used.
+      self.fullyFilteredPath = path.join(workDir, "sampleGroup_filteredByExprAndVar.tsv");
+
+      return spawnCommand(python,
+        [
+          varianceFilterScript,
+          "--in_file", self.filteredByExpressionPath,
+          "--filter_level", "0.2",
+          "--out_file", self.fullyFilteredPath,
+        ],
+        workDir); 
+
+    }).then(Meteor.bindEnvironment(function(varianceFilterResults){
+      if(varianceFilterResults.exitCode !== 0){
+        throw new Error("Failed to apply variance filter (exit code not 0)");
+      }
+
+      // Filters were applied; create the output Blob2
+
+      var associated_samplegroup = {
+        collection_name: "SampleGroups",
+        mongo_id: sample_group_id,
+      };
+
+      var createBlob2Sync = Meteor.wrapAsync(Blobs2.create);
+      // Errors from this will be thrown to the catch below
+      var metadata = {"type" : "ExprAndVarFilteredSampleGroupData"}
+      var blob = createBlob2Sync(self.fullyFilteredPath, associated_samplegroup, metadata);
+      var output = {"filtered_samples_blob_id" : blob._id};
+
+      // Everything worked; resolve the promise
+      deferred.resolve(output);
+    },function(err){
+      deferred.reject(err);
+    })).catch(function(error){ 
+      // If we got an error anywhere along the chain,
+      // fail the job
+      deferred.reject(error);
+    });
+  // Will wait for the async code to run and either resolve or reject
+  // before completing the job
+  return deferred.promise;
+};
+
+JobClasses.ApplyExprAndVarianceFilters = ApplyExprAndVarianceFilters ;

webapp/server/classes/RunLimmaGSEA.js

@@ -63,8 +63,6 @@ RunLimmaGSEA.prototype.run = function () {
     };
   });
 
-  console.log("comboSampleGroupDataSets:", comboSampleGroupDataSets);
-
   var comboSampleGroupId = SampleGroups.insert({
     name: "temp - created in RunLimmaGSEA to call an adapter",
     version: 1,

webapp/server/classes/UpDownGenes.js

@@ -12,29 +12,43 @@ UpDownGenes.prototype.run = function () {
   var deferred = Q.defer();
   var self = this;
 
+
+
   // if the first parts of this command have already been run,
   // grab the paths for those output files
   var sample_group_id = self.job.args.sample_group_id;
   var iqr_multiplier = self.job.args.iqr_multiplier;
+
   var associated_object = {
     collection_name: "SampleGroups",
     mongo_id: sample_group_id,
   };
 
-  function getStoragePath(file_name) {
+  // Find a potential blob associated with the object above.
+  // only for median/highthreshold/lowthreshold.tsv blobs which have
+  // the from_filtered_sample_group field
+  // is_filtered: true if it's using the filtered version of a
+  // sample group; null otherwise (NOT false!)
+  function getStoragePath(file_name, is_filtered) {
     var blob = Blobs2.findOne({
       file_name: file_name,
       associated_object: associated_object,
       "metadata.iqr_multiplier": iqr_multiplier,
+      "metadata.from_filtered_sample_group": is_filtered
     });
 
     if (blob) {
       return blob.getFilePath();
     }
   }
-  var medianPath = getStoragePath("median.tsv");
-  var highThresholdPath = getStoragePath("highthreshold.tsv");
-  var lowThresholdPath = getStoragePath("lowthreshold.tsv");
+
+  // Set up to work with getStoragePath; either true or null
+  var usingFilter = self.job.args.use_filtered_sample_group
+  if(! usingFilter){usingFilter = null;}
+
+  var medianPath = getStoragePath("median.tsv", usingFilter);
+  var highThresholdPath = getStoragePath("highthreshold.tsv", usingFilter);
+  var lowThresholdPath = getStoragePath("lowthreshold.tsv", usingFilter);
   var regenerateFiles = false;
 
   // medianPath is the one to check if they exist or not so make sure
@@ -44,6 +58,7 @@ UpDownGenes.prototype.run = function () {
   }
 
   var exportScript = getSetting("genomic_expression_export");
+
   var exportCommands = [
     // single sample data
     spawnCommand(exportScript, [
@@ -54,9 +69,24 @@ UpDownGenes.prototype.run = function () {
 
   // also write out sample group data if necessary
   if (regenerateFiles) {
-    exportCommands.push(spawnCommand(exportScript, [
-      "--sample_group_id", sample_group_id,
-    ], workDir));
+    // If we're using the filtered sample group,
+    // get the blob with filtered data; otherwise, export it from scratch. 
+    if(usingFilter){
+      var filteredBlob = Blobs2.findOne({
+        associated_object:associated_object,
+        "metadata.type":"ExprAndVarFilteredSampleGroupData",
+      });
+      // If we can't find one, just give up instead of generating it here;
+      // We don't want to get into calling a job from another job
+      if(!filteredBlob){
+        throw new Error("Couldn't find the filtered sample group data that was promised.");
+      }
+      var filteredBlobPath = filteredBlob.getFilePath() ;
+    }else{
+      exportCommands.push(spawnCommand(exportScript, [
+        "--sample_group_id", sample_group_id,
+      ], workDir));
+    }
   }
 
   Q.all(exportCommands)
@@ -76,9 +106,14 @@ UpDownGenes.prototype.run = function () {
         // # arg 2: default 1.5
         // /usr/bin/Rscript outlier.R mRNA.NBL.POG.pancan.combat.5.tab 2
 
+        // Get the path for the sample group data
+        // depends on whether we just exported it or are using a filtered blob
+        var sampleGroupPath=(usingFilter)? filteredBlobPath : spawnResults[1].stdoutPath;
+
+        // Calculate the median, high, and low thresholds for the genes
         return spawnCommand("Rscript", [
           getSetting("calculate_outlier_genes"),
-          spawnResults[1].stdoutPath,
+          sampleGroupPath,
           iqr_multiplier,
         ], workDir);
       }
@@ -120,7 +155,10 @@ UpDownGenes.prototype.run = function () {
             console.log("error creating blob:", err);
           }
         }
-        var meta = { iqr_multiplier: iqr_multiplier };
+        // Metadata for median & up downblobs; note if filtered. 
+        var meta = { iqr_multiplier: iqr_multiplier};
+        if(usingFilter){ meta.from_filtered_sample_group = true; }
+
         Blobs2.create(medianPath, associated_object, meta, printError);
         Blobs2.create(highThresholdPath, associated_object, meta, printError);
         Blobs2.create(lowThresholdPath, associated_object, meta, printError);
@@ -141,6 +179,8 @@ UpDownGenes.prototype.run = function () {
       };
       var createBlob2Sync = Meteor.wrapAsync(Blobs2.create);
 
+      // Output files are associated with a job, not the sample group,
+      // so they don't need to be tagged with usingFilter. 
       try{
         var upGenesBlob = createBlob2Sync(upPath, associated_job_object, {});
         var downGenesBlob = createBlob2Sync(downPath, associated_job_object, {});

webapp/server/classes/UpdateCbioSecurity.js

@@ -117,6 +117,8 @@ UpdateCbioSecurity.prototype.run = function () {
 }
 
 Meteor.startup(function () {
+  // NOTE: the job can also be called by any user via Meteor method
+  // in patient-care (refreshCBioPortalAccess)
   var newJobBlueprint = {
     name: "UpdateCbioSecurity",
     user_id: "admin",
@@ -130,7 +132,7 @@ Meteor.startup(function () {
     name: "update-cbio-security",
     schedule: function(parser) {
       // parser is a later.parse object
-      return parser.text('every 12 hours');
+      return parser.text('every 6 hours');
     },
     job: function () {
       Jobs.insert(newJobBlueprint);

webapp/server/startup.js

@@ -74,20 +74,23 @@ function runNextJob () {
   var mustHaveFinished = Jobs.find({
     _id: {$in: mongoJob.prerequisite_job_ids}
   }).fetch();
-  for (var index in mustHaveFinished) {
+  for (var index = 0; index<mustHaveFinished.length; index++) {
     // if there was an error with that one, there's an error with this one
-    if (mustHaveFinished.status === "error") {
+    if (mustHaveFinished[index].status === "error") {
       Jobs.update(mongoJob._id, {
         $set: {
           status: "error",
           error_description: "error in prerequisite job",
         }
       });
-    } else {
+      return;
+    // Prerequisites are still running; retry again later.
+    } else if(mustHaveFinished[index].status !== "done"){
       retryLater("not finished with prerequisite job");
+      return;
     }
-    return;
   }
+  // All prerequisites are "done"; carry on.
 
   // get the job's class
   var jobClass = JobClasses[mongoJob.name];
@@ -132,8 +135,10 @@ function runNextJob () {
 
     var error_description = reason + ""; // convert to string
     console.log("job: rejected - ", reason);
-    if (reason.stack) {
-      console.log("stack trace:", reason.stack);
+    var stackTrace = "";
+    if (reason && reason.stack) {
+      stackTrace = reason.stack
+      console.log("stack trace:", stackTrace);
     }
     if (errorWarningUser) {
       error_description += " Error calling onError: " + errorWarningUser;
@@ -142,7 +147,7 @@ function runNextJob () {
       $set: {
         status: "error",
         error_description: error_description,
-        stack_trace: reason.stack,
+        stack_trace: stackTrace,
       }
     });
   };