func (r Record) Data() []byte {
	var b bytes.Buffer
	b.WriteString(strings.Join([]string(r), ","))
	b.WriteString("\n")
	return b.Bytes()
}
Data returns the underlying []string in a Record as a []byte
package ais
import (
	"bufio"
	"bytes"
	"encoding/csv"
	"errors"
	"fmt"
	"hash"
	"hash/fnv"
	"io"
	"os"
	"reflect"
	"sort"
	"strconv"
	"strings"
	"text/tabwriter"
	"time"
	"github.com/FATHOM5/haversine"
	"github.com/mmcloughlin/geohash"
)
const TimeLayout = `2006-01-02T15:04:05`
const flushThreshold = 250000
var ErrEmptySet = errors.New("ErrEmptySet")
type Matching interface {
	Match(*Record) (bool, error)
}
type Vessel struct {
	MMSI       string
	VesselName string
}
type VesselSet map[Vessel]int
type Field string
type Generator interface {
	Generate(rec Record, index ...int) (Field, error)
}
type Geohasher RecordSet
func NewGeohasher(rs *RecordSet) *Geohasher {
	g := Geohasher(*rs)
	return &g
}
func (g *Geohasher) Generate(rec Record, index ...int) (Field, error) {
	if len(index) != 2 {
		return "", fmt.Errorf("geohash: generate: len(index) must equal" +
			" 2 where the first int is the index of `LAT` and the second int is the index of `LON`")
	}
	indexLat, indexLon := index[0], index[1]
	lat, err := rec.ParseFloat(indexLat)
	if err != nil {
		return "", fmt.Errorf("geohash: unable to parse lat")
	}
	lon, err := rec.ParseFloat(indexLon)
	if err != nil {
		return "", fmt.Errorf("geohash: unable to parse lon")
	}
	hash := geohash.EncodeIntWithPrecision(lat, lon, uint(22))
	return Field(fmt.Sprintf("%#x", hash)), nil
}
type RecordSet struct {
	r     *csv.Reader    
	w     *csv.Writer    
	h     Headers        
	data  io.ReadWriter  
	first *Record        
	stash *Record        
}
func NewRecordSet() *RecordSet {
	rs := new(RecordSet)
	buf := bytes.Buffer{}
	rs.data = &buf
	rs.r = csv.NewReader(&buf)
	rs.w = csv.NewWriter(&buf)
	rs.r.LazyQuotes = true
	rs.r.Comment = '#'
	return rs
}
func OpenRecordSet(filename string) (*RecordSet, error) {
	rs := NewRecordSet()
	f, err := os.OpenFile(filename, os.O_RDWR, 0666)  
	if err != nil {
		return nil, fmt.Errorf("open recordset: %v", err)
	}
	rs.data = f
	rs.r = csv.NewReader(f)
	rs.r.LazyQuotes = true
	rs.r.Comment = '#'
	rs.w = csv.NewWriter(f)
	var h Headers
	h.Fields, err = rs.r.Read()
	if err != nil {
		return nil, fmt.Errorf("open recordset: %v", err)
	}
	rs.h = h
	return rs, nil
}
func (rs *RecordSet) SetHeaders(h Headers) {
	rs.h = h
}
func (rs *RecordSet) Read() (*Record, error) {
	if rec := rs.first; rec != nil {
		rs.first = nil
		return rec, nil
	}
	if rec := rs.stash; rec != nil {
		rs.stash = nil
		return rec, nil
	}
	r, err := rs.r.Read()
	if err == io.EOF {
		return nil, err
	}
	if err != nil {
		return nil, fmt.Errorf("recordset read: %v", err)
	}
	rec := Record(r)
	return &rec, nil
}
func (rs *RecordSet) readFirst() (*Record, error) {
	if rs.first != nil {
		return rs.first, nil
	}
	r, err := rs.r.Read()
	if err != nil {
		return nil, err
	}
	rec := Record(r)
	rs.first = &rec
	return rs.first, nil
}
func (rs *RecordSet) Stash(rec *Record) {
	rs.stash = rec
}
func (rs *RecordSet) Write(rec Record) error {
	err := rs.w.Write(rec)
	return err
}
func (rs *RecordSet) Flush() error {
	rs.w.Flush()
	err := rs.w.Error()
	return err
}
func (rs *RecordSet) AppendField(newField string, requiredHeaders []string, gen Generator) (*RecordSet, error) {
	rs2 := NewRecordSet()
	h := rs.Headers()
	h.Fields = append(h.Fields, newField)
	rs2.SetHeaders(h)
	var indices []int
	for _, target := range requiredHeaders {
		index, ok := rs.Headers().Contains(target)
		if !ok {
			return nil, fmt.Errorf("append: headers does not contain %s", target)
		}
		indices = append(indices, index)
	}
	written := 0
	for {
		var rec Record
		rec, err := rs.r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("append: read error on csv file: %v", err)
		}
		field, err := gen.Generate(rec, indices...)
		if err != nil {
			return nil, fmt.Errorf("appendfield: generate: %v", err)
		}
		rec = append(rec, string(field))
		err = rs2.Write(rec)
		if err != nil {
			return nil, fmt.Errorf("appendfield: csv write error: %v", err)
		}
		written++
		if written%flushThreshold == 0 {
			err := rs2.Flush()
			if err != nil {
				return nil, fmt.Errorf("appendfield: csv flush error: %v", err)
			}
		}
	}
	err := rs2.Flush()
	if err != nil {
		return nil, fmt.Errorf("appendfield: csv flush error: %v", err)
	}
	return rs2, nil
}
func (rs *RecordSet) Close() error {
	if rs.data == nil {
		return nil
	}
	closerType := reflect.TypeOf((*io.Closer)(nil)).Elem()
	if reflect.TypeOf(rs.data).Implements(closerType) {
		v := reflect.ValueOf(rs.data).MethodByName("Close").Call(nil)
		if v[0].Interface() == (error)(nil) {
			return nil
		}
		err := v[0].Interface().(error)
		return fmt.Errorf("recordset close: %v", err)
	}
	return nil
}
func (rs *RecordSet) Headers() Headers { return rs.h }
func (rs *RecordSet) Save(name string) error {
	var err error
	rs.data, err = os.Create(name)
	if err != nil {
		return fmt.Errorf("recordset save: %v", err)
	}
	rs.w = csv.NewWriter(rs.data)  
	rs.Write(rs.h.Fields)
	for {
		rec, err := rs.r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return fmt.Errorf("recordset save: read error on csv file: %v", err)
		}
		rs.Write(rec)
	}
	err = rs.Flush()
	if err != nil {
		return fmt.Errorf("recordset save: flush error: %v", err)
	}
	return nil
}
func (rs *RecordSet) SubsetLimit(m Matching, n int, multipass bool) (*RecordSet, error) {
	rs2 := NewRecordSet()
	rs2.SetHeaders(rs.Headers())
	copyBuf := &bytes.Buffer{}
	copyWriter := bufio.NewWriter(copyBuf)
	recordsLeftToWrite := n
	for recordsLeftToWrite != 0 {
		var rec *Record
		rec, err := rs.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("subset: read error on csv file: %v", err)
		}
		if multipass {
			copyWriter.Write(rec.Data())
		}
		match, err := m.Match(rec)
		if err != nil {
			return nil, err
		}
		if match {
			err := rs2.Write(*rec)
			if err != nil {
				return nil, fmt.Errorf("subset: csv write error: %v", err)
			}
			recordsLeftToWrite--
			if recordsLeftToWrite%flushThreshold == 0 {
				err := rs2.Flush()
				if err != nil {
					return nil, fmt.Errorf("subset: csv flush error: %v", err)
				}
			}
		}
	}
	err := rs2.Flush()
	if err != nil {
		return nil, fmt.Errorf("subset: csv flush error: %v", err)
	}
	if recordsLeftToWrite == n {  
		return rs2, ErrEmptySet
	}
	if multipass {
		copyWriter.Flush()
		rs.r = csv.NewReader(copyBuf)
	}
	return rs2, nil
}
func (rs *RecordSet) Subset(m Matching) (*RecordSet, error) {
	return rs.SubsetLimit(m, -1, false)
}
func (rs *RecordSet) UniqueVessels() (VesselSet, error) {
	return rs.UniqueVesselsMulti(false)
}
func (rs *RecordSet) UniqueVesselsMulti(multipass bool) (VesselSet, error) {
	vs := make(VesselSet)
	var defaultVesselName = "no VesselName header"
	mmsiIndex, ok := rs.Headers().Contains("MMSI")
	if !ok {
		return nil, fmt.Errorf("unique vessels: recordset does not contain MMSI header")
	}
	vesselNameIndex, okVesselName := rs.Headers().Contains("VesselName")
	var rec *Record
	var err error
	copyBuf := &bytes.Buffer{}
	copyWriter := bufio.NewWriter(copyBuf)
	for {
		rec, err = rs.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, fmt.Errorf("unique vessel: read error on csv file: %v", err)
		}
		if multipass {
			copyWriter.Write(rec.Data())
		}
		if okVesselName {
			vs[Vessel{MMSI: (*rec)[mmsiIndex], VesselName: (*rec)[vesselNameIndex]}]++
		} else {
			vs[Vessel{MMSI: (*rec)[mmsiIndex], VesselName: defaultVesselName}]++
		}
	}
	if multipass {
		copyWriter.Flush()
		rs.r = csv.NewReader(copyBuf)
	}
	return vs, nil
}
func (rs *RecordSet) SortByTime() (*RecordSet, error) {
	rs2 := NewRecordSet()
	rs2.SetHeaders(rs.Headers())
	bt, err := NewByTimestamp(rs)
	if err != nil {
		return nil, fmt.Errorf("sortbytime: %v", err)
	}
	sort.Sort(bt)
	written := 0
	for _, rec := range *bt.data {
		rs2.Write(rec)
		written++
		if written%flushThreshold == 0 {
			err := rs2.Flush()
			if err != nil {
				return nil, fmt.Errorf("sortbytime: flush error writing to new recordset: %v", err)
			}
		}
	}
	err = rs2.Flush()
	if err != nil {
		return nil, fmt.Errorf("sortbytime: flush error writing to new recordset: %v", err)
	}
	return rs2, nil
}
type Box struct {
	MinLat, MaxLat, MinLon, MaxLon float64
	LatIndex, LonIndex             int
}
func (b *Box) Match(rec *Record) (bool, error) {
	lat, err := rec.ParseFloat(b.LatIndex)
	if err != nil {
		return false, fmt.Errorf("unable to parse %v", (*rec)[b.LatIndex])
	}
	lon, err := rec.ParseFloat(b.LonIndex)
	if err != nil {
		return false, fmt.Errorf("unable to parse %v", (*rec)[b.LonIndex])
	}
	return lat >= b.MinLat && lat <= b.MaxLat && lon >= b.MinLon && lon <= b.MaxLon, nil
}
type ByTimestamp struct {
	h    Headers
	data *[]Record
}
func NewByTimestamp(rs *RecordSet) (*ByTimestamp, error) {
	bt := new(ByTimestamp)
	bt.h = rs.Headers()
	var err error
	bt.data, err = rs.loadRecords()
	if err != nil {
		return nil, fmt.Errorf("new bytimestamp: unable to load data: %v", err)
	}
	return bt, nil
}
func (bt ByTimestamp) Len() int { return len(*bt.data) }
func (bt ByTimestamp) Swap(i, j int) {
	(*bt.data)[i], (*bt.data)[j] = (*bt.data)[j], (*bt.data)[i]
}
func (bt ByTimestamp) Less(i, j int) bool {
	timeIndex, ok := bt.h.Contains("BaseDateTime")
	if !ok {
		panic("bytimestamp: less: headers does not contain BaseDateTime")
	}
	t1, err := time.Parse(TimeLayout, (*bt.data)[i][timeIndex])
	if err != nil {
		panic(err)
	}
	t2, err := time.Parse(TimeLayout, (*bt.data)[j][timeIndex])
	if err != nil {
		panic(err)
	}
	return t1.Before(t2)
}
func (rs *RecordSet) loadRecords() (*[]Record, error) {
	recs := new([]Record)
	record := new(Record)
	for {
		var err error
		record, err = rs.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return nil, err
		}
		*recs = append(*recs, *record)
	}
	return recs, nil
}
type Headers struct {
	Fields []string
}
func (h Headers) Contains(field string) (i int, ok bool) {
	for i, s := range h.Fields {
		if s == field {
			return i, true
		}
	}
	return 0, false
}
type HeaderMap struct {
	Present bool
	Idx     int
}
func (h Headers) ContainsMulti(fields ...string) (idxMap map[string]HeaderMap, ok bool) {
	idxMap = make(map[string]HeaderMap)
	for _, f := range fields {  
		idx, ok := h.Contains(f)
		if ok {
			idxMap[f] = HeaderMap{Present: true, Idx: idx}
		} else {
			return nil, false
		}
	}
	return idxMap, true
}
func (h Headers) String() string {
	const pad = ' '  
	b := new(bytes.Buffer)
	w := tabwriter.NewWriter(b, 0, 0, 2, pad, 0)
	fmt.Fprintf(w, "Index\tHeader\n")
	for i, header := range h.Fields {
		header = strings.TrimSpace(header)
		fmt.Fprintf(w, "%d\t%s\n", i, header)
	}
	w.Flush()
	return b.String()
}
func (h Headers) Equals(h2 Headers) bool {
	if (h.Fields == nil) != (h2.Fields == nil) {
		return false
	}
	if len(h.Fields) != len(h2.Fields) {
		return false
	}
	for i, f := range h.Fields {
		if f != h2.Fields[i] {
			return false
		}
	}
	return true
}
type Record []string
func (r Record) Hash() uint64 {
	var h64 hash.Hash64
	h64 = fnv.New64a()
	h64.Write(r.Data())
	return h64.Sum64()
}