Bump slice-numpy memory.
[lightning.git] / slicenumpy.go
index d0b06397587d9e379519432f42c3b035204c5557..2cb67960b60aa253a5cc8d8512557412e4de530f 100644 (file)
@@ -10,6 +10,7 @@ import (
        "flag"
        "fmt"
        "io"
+       "io/ioutil"
        "net/http"
        _ "net/http/pprof"
        "os"
@@ -72,7 +73,7 @@ func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
                        Name:        "lightning slice-numpy",
                        Client:      arvados.NewClientFromEnv(),
                        ProjectUUID: *projectUUID,
-                       RAM:         650000000000,
+                       RAM:         750000000000,
                        VCPUs:       96,
                        Priority:    *priority,
                        KeepCache:   2,
@@ -307,6 +308,12 @@ func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
                                go func() {
                                        defer throttleCPU.Release()
                                        count := make(map[[blake2b.Size256]byte]int, len(variants))
+
+                                       rt := reftile[tag]
+                                       if rt != nil {
+                                               count[blake2b.Sum256(rt.tiledata)] = 0
+                                       }
+
                                        for _, cg := range cgs {
                                                idx := int(tag-tagstart) * 2
                                                if idx < len(cg.Variants) {
@@ -348,6 +355,9 @@ func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
                                                remap[i] = rank[tv.Blake2b]
                                        }
                                        variantRemap[tag-tagstart] = remap
+                                       if rt != nil {
+                                               rt.variant = rank[blake2b.Sum256(rt.tiledata)]
+                                       }
                                }()
                        }
                        throttleCPU.Wait()
@@ -373,25 +383,24 @@ func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
                                        // mention it in annotations?)
                                        continue
                                }
-                               variants, ok := seq[tag]
-                               if !ok {
-                                       outcol++
-                                       continue
-                               }
+                               fmt.Fprintf(annow, "%d,%d,%d,=,%s,%d,,,\n", tag, outcol, rt.variant, rt.seqname, rt.pos)
+                               variants := seq[tag]
                                reftilestr := strings.ToUpper(string(rt.tiledata))
                                remap := variantRemap[tag-tagstart]
                                done := make([]bool, len(variants))
                                for v, tv := range variants {
                                        v := remap[v]
-                                       if done[v] {
+                                       if v == rt.variant || done[v] {
                                                continue
                                        } else {
                                                done[v] = true
                                        }
                                        if len(tv.Sequence) < taglen || !bytes.HasSuffix(rt.tiledata, tv.Sequence[len(tv.Sequence)-taglen:]) {
+                                               fmt.Fprintf(annow, "%d,%d,%d,,%s,%d,,,\n", tag, outcol, v, rt.seqname, rt.pos)
                                                continue
                                        }
                                        if lendiff := len(rt.tiledata) - len(tv.Sequence); lendiff < -1000 || lendiff > 1000 {
+                                               fmt.Fprintf(annow, "%d,%d,%d,,%s,%d,,,\n", tag, outcol, v, rt.seqname, rt.pos)
                                                continue
                                        }
                                        diffs, _ := hgvs.Diff(reftilestr, strings.ToUpper(string(tv.Sequence)), 0)
@@ -468,6 +477,7 @@ func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
                        cols += len(chunk) / rows
                }
                out := make([]int16, rows*cols)
+               hgvsCols := map[string][2][]int16{} // hgvs -> [[g0,g1,g2,...], [g0,g1,g2,...]] (slice of genomes for each phase)
                startcol := 0
                for outIdx, chunk := range toMerge {
                        chunkcols := len(chunk) / rows
@@ -490,9 +500,71 @@ func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
                                if len(line) == 0 {
                                        continue
                                }
-                               fields := bytes.SplitN(line, []byte{','}, 3)
+                               fields := bytes.SplitN(line, []byte{','}, 9)
+                               tag, _ := strconv.Atoi(string(fields[0]))
                                incol, _ := strconv.Atoi(string(fields[1]))
-                               fmt.Fprintf(annow, "%s,%d,%s\n", fields[0], incol+startcol/2, fields[2])
+                               tileVariant, _ := strconv.Atoi(string(fields[2]))
+                               hgvsID := string(fields[3])
+                               seqname := string(fields[4])
+                               pos, _ := strconv.Atoi(string(fields[5]))
+                               refseq := fields[6]
+                               if hgvsID == "" {
+                                       // Null entry for un-diffable
+                                       // tile variant
+                                       continue
+                               }
+                               if hgvsID == "=" {
+                                       // Null entry for ref tile
+                                       continue
+                               }
+                               if mask != nil && !mask.Check(strings.TrimPrefix(seqname, "chr"), pos, pos+len(refseq)) {
+                                       // The tile intersects one of
+                                       // the selected regions, but
+                                       // this particular HGVS
+                                       // variant does not.
+                                       continue
+                               }
+                               hgvsColPair := hgvsCols[hgvsID]
+                               if hgvsColPair[0] == nil {
+                                       // values in new columns start
+                                       // out as -1 ("no data yet")
+                                       // or 0 ("=ref") here, may
+                                       // change to 1 ("hgvs variant
+                                       // present") below, either on
+                                       // this line or a future line.
+                                       hgvsColPair = [2][]int16{make([]int16, len(cgnames)), make([]int16, len(cgnames))}
+                                       rt, ok := reftile[tagID(tag)]
+                                       if !ok {
+                                               err = fmt.Errorf("bug: seeing annotations for tag %d, but it has no reftile entry", tag)
+                                               return 1
+                                       }
+                                       for ph := 0; ph < 2; ph++ {
+                                               for row := 0; row < rows; row++ {
+                                                       v := chunk[row*chunkcols+incol*2+ph]
+                                                       if tileVariantID(v) == rt.variant {
+                                                               hgvsColPair[ph][row] = 0
+                                                       } else {
+                                                               hgvsColPair[ph][row] = -1
+                                                       }
+                                               }
+                                       }
+                                       hgvsCols[hgvsID] = hgvsColPair
+                                       hgvsref := hgvs.Variant{
+                                               Position: pos,
+                                               Ref:      string(refseq),
+                                               New:      string(refseq),
+                                       }
+                                       fmt.Fprintf(annow, "%d,%d,%d,%s:g.%s,%s,%d,%s,%s,%s\n", tag, incol+startcol/2, rt.variant, seqname, hgvsref.String(), seqname, pos, refseq, refseq, fields[8])
+                               }
+                               fmt.Fprintf(annow, "%d,%d,%d,%s,%s,%d,%s,%s,%s\n", tag, incol+startcol/2, tileVariant, hgvsID, seqname, pos, refseq, fields[7], fields[8])
+                               for ph := 0; ph < 2; ph++ {
+                                       for row := 0; row < rows; row++ {
+                                               v := chunk[row*chunkcols+incol*2+ph]
+                                               if int(v) == tileVariant {
+                                                       hgvsColPair[ph][row] = 1
+                                               }
+                                       }
+                               }
                        }
 
                        startcol += chunkcols
@@ -509,6 +581,37 @@ func (cmd *sliceNumpy) RunCommand(prog string, args []string, stdin io.Reader, s
                if err != nil {
                        return 1
                }
+               out = nil
+
+               cols = len(hgvsCols) * 2
+               log.Printf("building hgvs-based matrix: %d rows x %d cols", rows, cols)
+               out = make([]int16, rows*cols)
+               hgvsIDs := make([]string, 0, len(hgvsCols))
+               for hgvsID := range hgvsCols {
+                       hgvsIDs = append(hgvsIDs, hgvsID)
+               }
+               sort.Strings(hgvsIDs)
+               var hgvsLabels bytes.Buffer
+               for idx, hgvsID := range hgvsIDs {
+                       fmt.Fprintf(&hgvsLabels, "%d,%s\n", idx, hgvsID)
+                       for ph := 0; ph < 2; ph++ {
+                               hgvscol := hgvsCols[hgvsID][ph]
+                               for row, val := range hgvscol {
+                                       out[row*cols+idx*2+ph] = val
+                               }
+                       }
+               }
+               err = writeNumpyInt16(fmt.Sprintf("%s/hgvs.npy", *outputDir), out, rows, cols)
+               if err != nil {
+                       return 1
+               }
+
+               fnm := fmt.Sprintf("%s/hgvs.annotations.csv", *outputDir)
+               log.Printf("writing hgvs labels: %s", fnm)
+               err = ioutil.WriteFile(fnm, hgvsLabels.Bytes(), 0777)
+               if err != nil {
+                       return 1
+               }
        }
        return 0
 }