26 type importer struct {
33 func (cmd *importer) RunCommand(prog string, args []string, stdin io.Reader, stdout, stderr io.Writer) int {
37 fmt.Fprintf(stderr, "%s\n", err)
40 flags := flag.NewFlagSet("", flag.ContinueOnError)
41 flags.SetOutput(stderr)
42 flags.StringVar(&cmd.tagLibraryFile, "tag-library", "", "tag library fasta `file`")
43 flags.StringVar(&cmd.refFile, "ref", "", "reference fasta `file`")
44 flags.StringVar(&cmd.outputFile, "o", "", "output `file`")
45 pprof := flags.String("pprof", "", "serve Go profile data at http://`[addr]:port`")
46 err = flags.Parse(args)
47 if err == flag.ErrHelp {
50 } else if err != nil {
52 } else if cmd.tagLibraryFile == "" {
53 fmt.Fprintln(os.Stderr, "cannot import without -tag-library argument")
55 } else if flags.NArg() == 0 {
62 log.Println(http.ListenAndServe(*pprof, nil))
66 infiles, err := listInputFiles(flags.Args())
71 tilelib, err := cmd.loadTileLibrary()
76 for range time.Tick(10 * time.Second) {
77 log.Printf("tilelib.Len() == %d", tilelib.Len())
83 if cmd.outputFile == "" {
84 w = bufio.NewWriter(stdout)
86 outfile, err = os.OpenFile(cmd.outputFile, os.O_CREATE|os.O_WRONLY, 0777)
91 w = bufio.NewWriter(outfile)
93 cmd.encoder = gob.NewEncoder(w)
95 err = cmd.tileInputs(tilelib, infiles)
104 err = outfile.Close()
112 func (cmd *importer) tileFasta(tilelib *tileLibrary, infile string) (tileSeq, error) {
113 var input io.ReadCloser
114 input, err := os.Open(infile)
119 if strings.HasSuffix(infile, ".gz") {
120 input, err = gzip.NewReader(input)
126 return tilelib.TileFasta(infile, input)
129 func (cmd *importer) loadTileLibrary() (*tileLibrary, error) {
130 log.Printf("tag library %s load starting", cmd.tagLibraryFile)
131 f, err := os.Open(cmd.tagLibraryFile)
136 var rdr io.ReadCloser = f
137 if strings.HasSuffix(cmd.tagLibraryFile, ".gz") {
138 rdr, err = gzip.NewReader(f)
140 return nil, fmt.Errorf("%s: gzip: %s", cmd.tagLibraryFile, err)
144 var taglib tagLibrary
145 err = taglib.Load(rdr)
149 if taglib.Len() < 1 {
150 return nil, fmt.Errorf("cannot tile: tag library is empty")
152 log.Printf("tag library %s load done", cmd.tagLibraryFile)
153 return &tileLibrary{taglib: &taglib}, nil
156 func listInputFiles(paths []string) (files []string, err error) {
157 for _, path := range paths {
158 if fi, err := os.Stat(path); err != nil {
159 return nil, fmt.Errorf("%s: stat failed: %s", path, err)
160 } else if !fi.IsDir() {
161 if !strings.HasSuffix(path, ".2.fasta") || strings.HasSuffix(path, ".2.fasta.gz") {
162 files = append(files, path)
166 d, err := os.Open(path)
168 return nil, fmt.Errorf("%s: open failed: %s", path, err)
171 names, err := d.Readdirnames(0)
173 return nil, fmt.Errorf("%s: readdir failed: %s", path, err)
176 for _, name := range names {
177 if strings.HasSuffix(name, ".vcf") || strings.HasSuffix(name, ".vcf.gz") {
178 files = append(files, filepath.Join(path, name))
179 } else if strings.HasSuffix(name, ".1.fasta") || strings.HasSuffix(name, ".1.fasta.gz") {
180 files = append(files, filepath.Join(path, name))
185 for _, file := range files {
186 if strings.HasSuffix(file, ".1.fasta") || strings.HasSuffix(file, ".1.fasta.gz") {
188 } else if _, err := os.Stat(file + ".csi"); err == nil {
190 } else if _, err = os.Stat(file + ".tbi"); err == nil {
193 return nil, fmt.Errorf("%s: cannot read without .tbi or .csi index file", file)
199 func (cmd *importer) tileInputs(tilelib *tileLibrary, infiles []string) error {
200 starttime := time.Now()
201 errs := make(chan error, 1)
202 todo := make(chan func() error, len(infiles)*2)
203 var encodeJobs sync.WaitGroup
204 for _, infile := range infiles {
206 var phases sync.WaitGroup
208 variants := make([][]tileVariantID, 2)
209 if strings.HasSuffix(infile, ".1.fasta") || strings.HasSuffix(infile, ".1.fasta.gz") {
210 todo <- func() error {
212 log.Printf("%s starting", infile)
213 defer log.Printf("%s done", infile)
214 tseqs, err := cmd.tileFasta(tilelib, infile)
215 variants[0] = tseqs.Variants()
218 infile2 := regexp.MustCompile(`\.1\.fasta(\.gz)?$`).ReplaceAllString(infile, `.2.fasta$1`)
219 todo <- func() error {
221 log.Printf("%s starting", infile2)
222 defer log.Printf("%s done", infile2)
223 tseqs, err := cmd.tileFasta(tilelib, infile2)
224 variants[1] = tseqs.Variants()
228 for phase := 0; phase < 2; phase++ {
230 todo <- func() error {
232 log.Printf("%s phase %d starting", infile, phase+1)
233 defer log.Printf("%s phase %d done", infile, phase+1)
234 tseqs, err := cmd.tileGVCF(tilelib, infile, phase)
235 variants[phase] = tseqs.Variants()
242 defer encodeJobs.Done()
247 ntags := len(variants[0])
248 if ntags < len(variants[1]) {
249 ntags = len(variants[1])
251 flat := make([]tileVariantID, ntags*2)
252 for i := 0; i < ntags; i++ {
253 flat[i*2] = variants[0][i]
254 flat[i*2+1] = variants[1][i]
256 err := cmd.encoder.Encode(LibraryEntry{
257 CompactGenomes: []CompactGenome{{Name: infile, Variants: flat}},
268 var tileJobs sync.WaitGroup
269 running := int64(runtime.NumCPU())
270 for i := 0; i < runtime.NumCPU(); i++ {
273 defer tileJobs.Done()
274 defer atomic.AddInt64(&running, -1)
275 for fn := range todo {
286 remain := len(todo) + int(atomic.LoadInt64(&running)) - 1
287 ttl := time.Now().Sub(starttime) * time.Duration(remain) / time.Duration(cap(todo)-remain)
288 eta := time.Now().Add(ttl)
289 log.Printf("progress %d/%d, eta %v (%v)", cap(todo)-remain, cap(todo), eta, ttl)
299 func (cmd *importer) tileGVCF(tilelib *tileLibrary, infile string, phase int) (tileseq tileSeq, err error) {
300 if cmd.refFile == "" {
301 err = errors.New("cannot import vcf: reference data (-ref) not specified")
304 args := []string{"bcftools", "consensus", "--fasta-ref", cmd.refFile, "-H", fmt.Sprint(phase + 1), infile}
305 indexsuffix := ".tbi"
306 if _, err := os.Stat(infile + ".csi"); err == nil {
309 if out, err := exec.Command("docker", "image", "ls", "-q", "lightning-runtime").Output(); err == nil && len(out) > 0 {
310 args = append([]string{
311 "docker", "run", "--rm",
313 "--volume=" + infile + ":" + infile + ":ro",
314 "--volume=" + infile + indexsuffix + ":" + infile + indexsuffix + ":ro",
315 "--volume=" + cmd.refFile + ":" + cmd.refFile + ":ro",
319 consensus := exec.Command(args[0], args[1:]...)
320 consensus.Stderr = os.Stderr
321 stdout, err := consensus.StdoutPipe()
326 err = consensus.Start()
330 defer consensus.Wait()
331 tileseq, err = tilelib.TileFasta(fmt.Sprintf("%s phase %d", infile, phase+1), stdout)
339 err = consensus.Wait()
341 err = fmt.Errorf("%s phase %d: bcftools: %s", infile, phase, err)