[EXP] Reader iter api experiments

adapters/reader.go

@@ -1,6 +1,8 @@
 package adapters
 
-import "github.com/interline-io/transitland-lib/gtfs"
+import (
+	"github.com/interline-io/transitland-lib/gtfs"
+)
 
 // Reader is the main interface for reading GTFS data
 type Reader interface {

extract/setterfilter.go

@@ -34,7 +34,6 @@ func (tx *SetterFilter) AddValuesFromFile(filename string) error {
 		tx.AddValue(efn, eid, key, val)
 	})
 	return nil
-
 }
 
 // AddValue sets a new value to override.

tlcsv/adapter.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"iter"
 	"net/url"
 	"os"
 	"path/filepath"
@@ -17,6 +18,7 @@ import (
 	"github.com/interline-io/log"
 	"github.com/interline-io/transitland-lib/causes"
 	"github.com/interline-io/transitland-lib/request"
+	"github.com/interline-io/transitland-lib/tt"
 )
 
 // Adapter provides an interface for working with various kinds of GTFS sources: zip, directory, url.
@@ -284,6 +286,67 @@ func (adapter *ZipAdapter) OpenFile(filename string, cb func(io.Reader)) error {
 	return nil
 }
 
+type canFilename interface {
+	Filename() string
+}
+
+func getFilename(entType any) string {
+	if v, ok := entType.(canFilename); ok {
+		return v.Filename()
+	}
+	return ""
+}
+
+func (adapter *ZipAdapter) ReadEntityRows(entType any) (iter.Seq[tt.Row], func() error) {
+	var readErr error
+	errf := func() error { return readErr }
+	return func(yield func(tt.Row) bool) {
+		readErr = adapter.OpenFile(getFilename(entType), func(in io.Reader) {
+			it, errf := ReadRowsIter(in)
+			for row := range it {
+				yield(row)
+			}
+			_ = errf
+		})
+	}, errf
+	// var readErr error
+	// errf := func() error { return readErr }
+	// return func(yield func(tt.Row) bool) {
+	// 	filename := ""
+	// 	if v, ok := entType.(canFilename); ok {
+	// 		filename = v.Filename()
+	// 	}
+	// 	r, err := zip.OpenReader(adapter.path)
+	// 	if err != nil {
+	// 		readErr = err
+	// 		return
+	// 	}
+	// 	defer r.Close()
+	// 	var inFile *zip.File
+	// 	for _, f := range r.File {
+	// 		if f.Name != filepath.Join(adapter.internalPrefix, filename) {
+	// 			continue
+	// 		}
+	// 		inFile = f
+	// 	}
+	// 	if inFile == nil {
+	// 		readErr = causes.NewFileNotPresentError(filename)
+	// 		return
+	// 	}
+	// 	in, err := inFile.Open()
+	// 	if err != nil {
+	// 		return
+	// 	}
+	// 	it, errf := ReadRowsIter(in)
+	// 	for row := range it {
+	// 		yield(row)
+	// 	}
+	// 	readErr = errf()
+	// 	in.Close()
+	// 	r.Close()
+	// }, errf
+}
+
 // ReadRows opens the specified file and runs the callback on each Row. An error is returned if the file cannot be read.
 func (adapter *ZipAdapter) ReadRows(filename string, cb func(Row)) error {
 	return adapter.OpenFile(filename, func(in io.Reader) {

tlcsv/reader.go

@@ -428,10 +428,6 @@ func ReadEntities[T any](reader *Reader, efn string) chan T {
 	return eout
 }
 
-func getFilename(ent tt.Entity) string {
-	return ent.Filename()
-}
-
 // chunkMSI takes a string counter and chunks it into groups of size <= chunkSize
 func chunkMSI(count map[string]int, chunkSize int) s2D {
 	result := s2D{}

tlcsv/row.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 
 	"github.com/dimchansky/utfbom"
+	"github.com/interline-io/transitland-lib/tt"
 )
 
 // Row is a row value with a header.
@@ -43,10 +44,6 @@ func ReadRows(in io.Reader, cb func(Row)) error {
 	// Allow unescaped quotes
 	r.LazyQuotes = true
 	// Go
-	return readRows(r, cb)
-}
-
-func readRows(r *csv.Reader, cb func(Row)) error {
 	// Go for it.
 	firstRow, err := r.Read()
 	if err != nil {
@@ -90,7 +87,7 @@ func readRows(r *csv.Reader, cb func(Row)) error {
 	return nil
 }
 
-func ReadRowsIter(in io.Reader, optFns ...csvOptFn) iter.Seq2[Row, error] {
+func ReadRowsIter(in io.Reader, optFns ...csvOptFn) (iter.Seq[tt.Row], func() error) {
 	// Handle byte-order-marks.
 	r := csv.NewReader(utfbom.SkipOnly(in))
 	// Allow variable columns - very common in GTFS
@@ -105,11 +102,14 @@ func ReadRowsIter(in io.Reader, optFns ...csvOptFn) iter.Seq2[Row, error] {
 	for _, optFn := range optFns {
 		optFn(r)
 	}
-	return func(yield func(Row, error) bool) {
+	var anyValues []any
+	var readErr error
+	errf := func() error { return readErr }
+	return func(yield func(tt.Row) bool) {
 		// Go for it.
-		firstRow, err := r.Read()
-		if err != nil {
-			yield(Row{}, err)
+		firstRow, firstRowErr := r.Read()
+		if firstRowErr != nil {
+			readErr = firstRowErr
 			return
 		}
 		// Copy header, since we will reuse the backing array
@@ -122,34 +122,46 @@ func ReadRowsIter(in io.Reader, optFns ...csvOptFn) iter.Seq2[Row, error] {
 		for k, i := range header {
 			hindex[i] = k
 		}
+		// Reusable slice
+		anyValues = make([]any, len(header))
+		// Read all rows
 		for {
-			row, err := r.Read()
-			if err == nil {
+			row, rowErr := r.Read()
+			if rowErr == nil {
 				// ok
-			} else if err == io.EOF {
+			} else if rowErr == io.EOF {
 				break
-			} else if _, ok := err.(*csv.ParseError); ok {
+			} else if _, ok := rowErr.(*csv.ParseError); ok {
 				// Parse error: clear row, add error to row
-				row = []string{}
+				for i := 0; i < len(anyValues); i++ {
+					anyValues[i] = nil
+				}
 			} else {
 				// Serious error: break and return with error
-				yield(Row{}, err)
+				readErr = rowErr
 				return
 			}
 			// Remove whitespace
 			for i := 0; i < len(row); i++ {
 				v := row[i]
 				// This is dumb but saves substantial time.
+				anyValues[i] = v
 				if len(v) > 0 && (v[0] == ' ' || v[len(v)-1] == ' ' || v[0] == '\t' || v[len(v)-1] == '\t') {
-					row[i] = strings.TrimSpace(v)
+					anyValues[i] = strings.TrimSpace(v)
 				}
 			}
 			// Pass parse errors to row
 			line, _ := r.FieldPos(0)
-			cbrow := Row{Row: row, Line: line, Header: header, Hindex: hindex, Err: err}
-			if !yield(cbrow, nil) {
+			cbrow := tt.Row{
+				Values: anyValues,
+				Line:   line,
+				Header: header,
+				Hindex: hindex,
+				Err:    rowErr,
+			}
+			if !yield(cbrow) {
 				return
 			}
 		}
-	}
+	}, errf
 }

tlcsv/row_test.go

@@ -0,0 +1,26 @@
+package tlcsv
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/interline-io/transitland-lib/gtfs"
+	"github.com/interline-io/transitland-lib/internal/testpath"
+	"github.com/interline-io/transitland-lib/tt"
+)
+
+func TestReadRowsIter(t *testing.T) {
+	adapter := &ZipAdapter{path: testpath.RelPath("testdata/example.zip")}
+	if err := adapter.Open(); err != nil {
+		t.Error(err)
+		return
+	}
+
+	it, errf := tt.ReadEntitiesIter[gtfs.Stop](adapter)
+	for ent := range it {
+		fmt.Println("ent:", ent.StopName)
+	}
+	if err := errf(); err != nil {
+		t.Error(err)
+	}
+}

tt/row.go

@@ -0,0 +1,139 @@
+package tt
+
+import (
+	"fmt"
+	"iter"
+	"reflect"
+
+	"github.com/interline-io/transitland-lib/causes"
+	"github.com/interline-io/transitland-lib/internal/tags"
+	"github.com/jmoiron/sqlx/reflectx"
+)
+
+const bufferSize = 1_000
+
+var MapperCache = tags.NewCache(reflectx.NewMapperFunc("csv", tags.ToSnakeCase))
+
+type Row struct {
+	Header []string
+	Values []any
+	Hindex map[string]int
+	Line   int
+	Err    error
+}
+
+// Get a value from the row.
+func (row *Row) Get(k string) (any, bool) {
+	if i, ok := row.Hindex[k]; ok {
+		if len(row.Values) > i {
+			return row.Values[i], true
+		}
+	}
+	return nil, false
+}
+
+// Get a value from the row as a string.
+func (row *Row) GetString(k string) (string, bool) {
+	if i, ok := row.Hindex[k]; ok {
+		if len(row.Values) > i {
+			return toStrv(row.Values[i]), true
+		}
+	}
+	return "", false
+}
+
+type RowReader interface {
+	ReadEntityRows(any) (iter.Seq[Row], func() error)
+}
+
+func ReadEntities[T any](reader RowReader) chan T {
+	// To get Filename() or TableName()
+	var entType T
+	// Prepare channel
+	eout := make(chan T, bufferSize)
+	go func(c chan T) {
+		it, _ := reader.ReadEntityRows(entType)
+		for row := range it {
+			var e T
+			loadRowReflect(&e, row)
+			c <- e
+		}
+		close(c)
+	}(eout)
+	return eout
+}
+
+func ReadEntitiesIter[T any](reader RowReader) (iter.Seq[T], func() error) {
+	// To get Filename() or TableName()
+	var readErr error
+	var entType *T = new(T)
+	return func(yield func(T) bool) {
+		it, errf := reader.ReadEntityRows(entType)
+		for row := range it {
+			fmt.Println("row:", row)
+			var e T
+			loadRowReflect(&e, row)
+			yield(e)
+		}
+		readErr = errf()
+	}, func() error { return readErr }
+}
+
+// loadRowReflect is the Reflect path
+func loadRowReflect(ent any, row Row) []error {
+	var errs []error
+	// Get the struct tag map
+	fmap := MapperCache.GetStructTagMap(ent)
+	// For each struct tag, set the field value
+	entValue := reflect.ValueOf(ent).Elem()
+	for i := 0; i < len(row.Header); i++ {
+		if i > len(row.Values) {
+			continue
+		}
+		fieldName := row.Header[i]
+		fieldValue := row.Values[i]
+		fieldInfo, ok := fmap[fieldName]
+
+		// Add to extra fields if there's no struct tag
+		if !ok {
+			if extEnt, ok2 := ent.(EntityWithExtra); ok2 {
+				extEnt.SetExtra(fieldName, toStrv(fieldValue))
+			}
+			continue
+		}
+
+		// Skip if empty, special case for strings
+		if fieldValue == nil {
+			continue
+		} else if v, ok := fieldValue.(string); ok && v == "" {
+			continue
+		}
+
+		// Handle different known types
+		entFieldAddr := reflectx.FieldByIndexes(entValue, fieldInfo.Index).Addr().Interface()
+		if v, ok := entFieldAddr.(canScan); ok {
+			if err := v.Scan(fieldValue); err != nil {
+				errs = append(errs, err)
+			}
+		} else if _, scanErr := convertAssign(entFieldAddr, fieldValue); scanErr != nil {
+			errs = append(errs, causes.NewFieldParseError(fieldName, toStrv(fieldValue)))
+		}
+	}
+	if len(errs) > 0 {
+		if extEnt, ok := ent.(EntityWithLoadErrors); ok {
+			for _, err := range errs {
+				extEnt.AddError(err)
+			}
+		}
+	}
+	return errs
+}
+
+func toStrv(value any) string {
+	if v, ok := value.(string); ok {
+		return v
+	}
+	strv := ""
+	convertAssign(&strv, value)
+	return strv
+}