Merge branch '19997-glm'
[lightning.git] / slicenumpy.go
index fb3ae95187c5289e4a525caf9fc58e410838f666..895c3c15fa2d91cfd7b153f65d6b78a9c59ad806 100644 (file)
@@ -8,6 +8,7 @@ import (
        "bufio"
        "bytes"
        "encoding/gob"
+       "encoding/json"
        "errors"
        "flag"
        "fmt"
@@ -52,6 +53,8 @@ type sliceNumpy struct {
        samples         []sampleInfo
        trainingSet     []int // samples index => training set index, or -1 if not in training set
        trainingSetSize int
+       pvalue          func(onehot []bool) float64
+       pvalueCallCount int64
 }
 
 func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) int {
@@ -72,6 +75,7 @@ func (cmd *sliceNumpy) run(prog string, args []string, stdin io.Reader, stdout,
        arvadosVCPUs := flags.Int("arvados-vcpus", 96, "number of VCPUs to request for arvados container")
        projectUUID := flags.String("project", "", "project `UUID` for output data")
        priority := flags.Int("priority", 500, "container request priority")
+       preemptible := flags.Bool("preemptible", true, "request preemptible instance")
        inputDir := flags.String("input-dir", "./in", "input `directory`")
        outputDir := flags.String("output-dir", "./out", "output `directory`")
        ref := flags.String("ref", "", "reference name (if blank, choose last one that appears in input)")
@@ -123,6 +127,7 @@ func (cmd *sliceNumpy) run(prog string, args []string, stdin io.Reader, stdout,
                        Priority:    *priority,
                        KeepCache:   2,
                        APIAccess:   true,
+                       Preemptible: *preemptible,
                }
                err = runner.TranslatePaths(inputDir, regionsFilename, samplesFilename)
                if err != nil {
@@ -280,6 +285,11 @@ func (cmd *sliceNumpy) run(prog string, args []string, stdin io.Reader, stdout,
                                cmd.trainingSet[i] = -1
                        }
                }
+               if cmd.pvalue == nil {
+                       cmd.pvalue = func(onehot []bool) float64 {
+                               return pvalue(onehot, cmd.chi2Cases)
+                       }
+               }
        }
        if cmd.filter.MinCoverage == 1 {
                // In the generic formula below, floating point
@@ -291,39 +301,31 @@ func (cmd *sliceNumpy) run(prog string, args []string, stdin io.Reader, stdout,
                cmd.minCoverage = int(math.Ceil(cmd.filter.MinCoverage * float64(len(cmd.cgnames))))
        }
 
-       {
-               samplesOutFilename := *outputDir + "/samples.csv"
-               log.Infof("writing sample metadata to %s", samplesOutFilename)
-               var f *os.File
-               f, err = os.Create(samplesOutFilename)
-               if err != nil {
-                       return err
-               }
-               defer f.Close()
-               for i, si := range cmd.samples {
-                       var cc, tv string
-                       if si.isCase {
-                               cc = "1"
-                       } else if si.isControl {
-                               cc = "0"
-                       }
-                       if si.isTraining {
-                               tv = "1"
-                       } else {
-                               tv = "0"
-                       }
-                       _, err = fmt.Fprintf(f, "%d,%s,%s,%s\n", i, si.id, cc, tv)
-                       if err != nil {
-                               err = fmt.Errorf("write %s: %w", samplesOutFilename, err)
-                               return err
-                       }
-               }
-               err = f.Close()
+       if len(cmd.samples[0].pcaComponents) > 0 {
+               cmd.pvalue = glmPvalueFunc(cmd.samples, cmd.pcaComponents)
+               // Unfortunately, statsmodel/glm lib logs stuff to
+               // os.Stdout when it panics on an unsolvable
+               // problem. We recover() from the panic in glm.go, but
+               // we also need to commandeer os.Stdout to avoid
+               // producing large quantities of logs.
+               stdoutWas := os.Stdout
+               defer func() { os.Stdout = stdoutWas }()
+               os.Stdout, err = os.Open(os.DevNull)
                if err != nil {
-                       err = fmt.Errorf("close %s: %w", samplesOutFilename, err)
                        return err
                }
-               log.Print("done")
+       }
+
+       // cgnamemap[name]==true for samples that we are including in
+       // output
+       cgnamemap := map[string]bool{}
+       for _, name := range cmd.cgnames {
+               cgnamemap[name] = true
+       }
+
+       err = writeSampleInfo(cmd.samples, *outputDir)
+       if err != nil {
+               return err
        }
 
        log.Info("indexing reference tiles")
@@ -507,7 +509,7 @@ func (cmd *sliceNumpy) run(prog string, args []string, stdin io.Reader, stdout,
                                        if cmd.filter.MaxTag >= 0 && cg.StartTag > tagID(cmd.filter.MaxTag) {
                                                return errSkip
                                        }
-                                       if !matchGenome.MatchString(cg.Name) {
+                                       if !cgnamemap[cg.Name] {
                                                continue
                                        }
                                        // pad to full slice size
@@ -1174,6 +1176,17 @@ func (cmd *sliceNumpy) run(prog string, args []string, stdin io.Reader, stdout,
                        if err != nil {
                                return err
                        }
+                       fnm = fmt.Sprintf("%s/stats.json", *outputDir)
+                       j, err := json.Marshal(map[string]interface{}{
+                               "pvalueCallCount": cmd.pvalueCallCount,
+                       })
+                       if err != nil {
+                               return err
+                       }
+                       err = os.WriteFile(fnm, j, 0777)
+                       if err != nil {
+                               return err
+                       }
                }
                if *onlyPCA {
                        cols := 0
@@ -1245,42 +1258,19 @@ func (cmd *sliceNumpy) run(prog string, args []string, stdin io.Reader, stdout,
                        }
                        log.Print("done")
 
-                       samplesOutFilename := *outputDir + "/samples.csv"
-                       log.Infof("writing sample metadata to %s", samplesOutFilename)
-                       var f *os.File
-                       f, err = os.Create(samplesOutFilename)
-                       if err != nil {
-                               return err
-                       }
-                       defer f.Close()
-                       for i, si := range cmd.samples {
-                               var cc, tv string
-                               if si.isCase {
-                                       cc = "1"
-                               } else if si.isControl {
-                                       cc = "0"
-                               }
-                               if si.isTraining {
-                                       tv = "1"
-                               } else {
-                                       tv = "0"
-                               }
-                               var pcavals string
+                       log.Print("copying pca components to sampleInfo")
+                       for i := range cmd.samples {
+                               cmd.samples[i].pcaComponents = make([]float64, outcols)
                                for c := 0; c < outcols; c++ {
-                                       pcavals += fmt.Sprintf(",%f", pca.At(i, c))
-                               }
-                               _, err = fmt.Fprintf(f, "%d,%s,%s,%s%s\n", i, si.id, cc, tv, pcavals)
-                               if err != nil {
-                                       err = fmt.Errorf("write %s: %w", samplesOutFilename, err)
-                                       return err
+                                       cmd.samples[i].pcaComponents[i] = pca.At(i, c)
                                }
                        }
-                       err = f.Close()
+                       log.Print("done")
+
+                       err = writeSampleInfo(cmd.samples, *outputDir)
                        if err != nil {
-                               err = fmt.Errorf("close %s: %w", samplesOutFilename, err)
                                return err
                        }
-                       log.Print("done")
                }
        }
        if !*mergeOutput && !*onehotChunked && !*onehotSingle && !*onlyPCA {
@@ -1369,13 +1359,60 @@ func loadSampleInfo(samplesFilename string) ([]sampleInfo, error) {
                        isCase:        split[2] == "1",
                        isControl:     split[2] == "0",
                        isTraining:    split[3] == "1",
-                       isValidation:  split[3] == "0",
+                       isValidation:  split[3] == "0" && len(split[2]) > 0, // fix errant 0s in input
                        pcaComponents: pcaComponents,
                })
        }
        return si, nil
 }
 
+func writeSampleInfo(samples []sampleInfo, outputDir string) error {
+       fnm := outputDir + "/samples.csv"
+       log.Infof("writing sample metadata to %s", fnm)
+       f, err := os.Create(fnm)
+       if err != nil {
+               return err
+       }
+       defer f.Close()
+       pcaLabels := ""
+       if len(samples) > 0 {
+               for i := range samples[0].pcaComponents {
+                       pcaLabels += fmt.Sprintf(",PCA%d", i)
+               }
+       }
+       _, err = fmt.Fprintf(f, "Index,SampleID,CaseControl,TrainingValidation%s\n", pcaLabels)
+       if err != nil {
+               return err
+       }
+       for i, si := range samples {
+               var cc, tv string
+               if si.isCase {
+                       cc = "1"
+               } else if si.isControl {
+                       cc = "0"
+               }
+               if si.isTraining {
+                       tv = "1"
+               } else if si.isValidation {
+                       tv = "0"
+               }
+               var pcavals string
+               for _, pcaval := range si.pcaComponents {
+                       pcavals += fmt.Sprintf(",%f", pcaval)
+               }
+               _, err = fmt.Fprintf(f, "%d,%s,%s,%s%s\n", i, si.id, cc, tv, pcavals)
+               if err != nil {
+                       return fmt.Errorf("write %s: %w", fnm, err)
+               }
+       }
+       err = f.Close()
+       if err != nil {
+               return fmt.Errorf("close %s: %w", fnm, err)
+       }
+       log.Print("done")
+       return nil
+}
+
 func (cmd *sliceNumpy) filterHGVScolpair(colpair [2][]int8) bool {
        if cmd.chi2PValue >= 1 {
                return true
@@ -1602,12 +1639,8 @@ func (cmd *sliceNumpy) tv2homhet(cgs map[string]CompactGenome, maxv tileVariantI
                if col < 4 && !cmd.includeVariant1 {
                        continue
                }
-               var p float64
-               if len(cmd.samples[0].pcaComponents) > 0 {
-                       p = pvalueGLM(cmd.samples, obs[col], cmd.pcaComponents)
-               } else {
-                       p = pvalue(obs[col], cmd.chi2Cases)
-               }
+               atomic.AddInt64(&cmd.pvalueCallCount, 1)
+               p := cmd.pvalue(obs[col])
                if cmd.chi2PValue < 1 && !(p < cmd.chi2PValue) {
                        continue
                }