wsjtx-dashboards/cmd/alltxt2csv/main.go

package main

import (
  "fmt"
  "github.com/jnovack/flag"
  log "github.com/sirupsen/logrus"
//  "strings"
//  "strconv"
//  "time"
  "os"
  "bufio"
  "runtime"
  "sync"
//  "github.com/mmcloughlin/geohash"
//  "github.com/tzneal/ham-go/dxcc"
  "github.com/denzs/wsjtx-dashboards/shared/wsjtx"
)

var station     string
var pathin      string
var pathout     string
var trace       bool

func usage() {
    fmt.Printf("Usage of %s:\n", os.Args[0])
    flag.PrintDefaults()
}

func init() {
  flag.StringVar(&pathin, "in", "", "path to wsjt-x ALL.txt")
  flag.StringVar(&pathout, "out", "", "path to csv outfile")
  flag.StringVar(&station, "station", "localstation", "your callsign or wsjtx instance identifier")
  flag.BoolVar(&trace, "trace", false, "log every line... yes really ;)")
  flag.Parse()

  if pathin == "" || pathout == "" {
    usage()
    os.Exit(1)
  }

  formatter := &log.TextFormatter{
    FullTimestamp: true,
  }
  log.SetFormatter(formatter)

  if trace {
    log.SetLevel(log.TraceLevel)
    log.Info("trace logging enabled")
  } else {
    log.Info("normal logging enabled")
  }

  _ , err := os.Stat(pathout)
  if !os.IsNotExist(err) {
    log.Fatalf("file %s already exists..", pathout)
  }

}

func eatline(id int, lines chan string, results chan wsjtx.Result, wg *sync.WaitGroup) {
  defer wg.Done()
  log.Infof("worker %d.. starting..", id)
  loop:
  for {
    select {
      case line, more := <- lines :
        if more {
          result, parsed := wsjtx.ScanLine(line)
          if parsed {
            results <- result
          }
        } else {
          log.Infof("eatline worker[%d]: resultchan seems down.. going home..", id)
          break loop
        }
    }
  }
  log.Infof("worker %d.. done!", id)
  return
}

func eatfile(results chan wsjtx.Result, done chan bool) {
  var wg sync.WaitGroup
  log.Info("starting eating file, please wait..")

  filein, err := os.Open(pathin)
  if err != nil {
      log.Fatal(err)
  }
  scanner := bufio.NewScanner(filein)

  lines := make(chan string,runtime.NumCPU())

  for w := 0; w <= runtime.NumCPU(); w++ {
    wg.Add(1)
    go eatline(w, lines, results, &wg)
  }

  i := 0
  for scanner.Scan() {
    i++
    if i % 1000000 == 0 {
      log.Infof("%d lines read..", i)
    }
    log.Info("line: ", scanner.Text())
    lines <- scanner.Text()
  }

  log.Info("sending children to bed..")
  close(lines)
  wg.Wait()
  log.Info("children are in bed now! ;)")

  filein.Close()

  done <- true

  log.Info("done.. eatfile")
  return
}

func main(){
  fileout, err := os.OpenFile(pathout,os.O_CREATE|os.O_RDWR, 0666)
  if err != nil {
    log.Fatal(err)
  }

  writer := bufio.NewWriter(fileout)

  results := make(chan wsjtx.Result,runtime.NumCPU())
  done := make(chan bool)

  go eatfile(results, done)

  loop:
  for {
    select {
      case result := <- results :
        _, err := writer.WriteString(fmt.Sprintf("\"%s\",\"%s\",\"%s\",\"%s\",\"%s\",\"%s\",\"%s\",\"%s\",\"%d\",\"%d\",\n", result.Timestamp.Format("2006-01-02 15:04:05"), station, result.Call, result.Band, result.Ent.Continent, result.Mode, result.Ent.Entity, result.GeoHash, result.Signal, result.Rx))
        if err != nil {
          log.Warn(err)
        }
     case <- done :
      break loop
    }
  }
  writer.Flush()
  fileout.Close()

  log.Info("done.. main")
}