GTFS-Flex: database, copier, validation support

adapters/direct/reader.go

@@ -11,34 +11,38 @@ var bufferSize = 1000
 
 // Reader is a mocked up Reader used for testing.
 type Reader struct {
-	AgencyList           []gtfs.Agency
-	RouteList            []gtfs.Route
-	TripList             []gtfs.Trip
-	StopList             []gtfs.Stop
-	StopTimeList         []gtfs.StopTime
-	ShapeList            []gtfs.Shape
-	CalendarList         []gtfs.Calendar
-	CalendarDateList     []gtfs.CalendarDate
-	FeedInfoList         []gtfs.FeedInfo
-	FareRuleList         []gtfs.FareRule
-	FareAttributeList    []gtfs.FareAttribute
-	FrequencyList        []gtfs.Frequency
-	TransferList         []gtfs.Transfer
-	LevelList            []gtfs.Level
-	PathwayList          []gtfs.Pathway
-	AttributionList      []gtfs.Attribution
-	TranslationList      []gtfs.Translation
-	AreaList             []gtfs.Area
-	StopAreaList         []gtfs.StopArea
-	FareLegRuleList      []gtfs.FareLegRule
-	FareTransferRuleList []gtfs.FareTransferRule
-	FareMediaList        []gtfs.FareMedia
-	FareProductList      []gtfs.FareProduct
-	RiderCategoryList    []gtfs.RiderCategory
-	TimeframeList        []gtfs.Timeframe
-	NetworkList          []gtfs.Network
-	RouteNetworkList     []gtfs.RouteNetwork
-	OtherList            []tt.Entity
+	AgencyList            []gtfs.Agency
+	RouteList             []gtfs.Route
+	TripList              []gtfs.Trip
+	StopList              []gtfs.Stop
+	StopTimeList          []gtfs.StopTime
+	ShapeList             []gtfs.Shape
+	CalendarList          []gtfs.Calendar
+	CalendarDateList      []gtfs.CalendarDate
+	FeedInfoList          []gtfs.FeedInfo
+	FareRuleList          []gtfs.FareRule
+	FareAttributeList     []gtfs.FareAttribute
+	FrequencyList         []gtfs.Frequency
+	TransferList          []gtfs.Transfer
+	LevelList             []gtfs.Level
+	PathwayList           []gtfs.Pathway
+	AttributionList       []gtfs.Attribution
+	TranslationList       []gtfs.Translation
+	AreaList              []gtfs.Area
+	StopAreaList          []gtfs.StopArea
+	FareLegRuleList       []gtfs.FareLegRule
+	FareTransferRuleList  []gtfs.FareTransferRule
+	FareMediaList         []gtfs.FareMedia
+	FareProductList       []gtfs.FareProduct
+	RiderCategoryList     []gtfs.RiderCategory
+	TimeframeList         []gtfs.Timeframe
+	NetworkList           []gtfs.Network
+	RouteNetworkList      []gtfs.RouteNetwork
+	LocationGroupList     []gtfs.LocationGroup
+	LocationGroupStopList []gtfs.LocationGroupStop
+	BookingRuleList       []gtfs.BookingRule
+	LocationList          []gtfs.Location
+	OtherList             []tt.Entity
 }
 
 // NewReader returns a new Reader.
@@ -432,3 +436,47 @@ func (mr *Reader) RouteNetworks() chan gtfs.RouteNetwork {
 	}()
 	return out
 }
+
+func (mr *Reader) LocationGroups() chan gtfs.LocationGroup {
+	out := make(chan gtfs.LocationGroup, bufferSize)
+	go func() {
+		for _, ent := range mr.LocationGroupList {
+			out <- ent
+		}
+		close(out)
+	}()
+	return out
+}
+
+func (mr *Reader) LocationGroupStops() chan gtfs.LocationGroupStop {
+	out := make(chan gtfs.LocationGroupStop, bufferSize)
+	go func() {
+		for _, ent := range mr.LocationGroupStopList {
+			out <- ent
+		}
+		close(out)
+	}()
+	return out
+}
+
+func (mr *Reader) BookingRules() chan gtfs.BookingRule {
+	out := make(chan gtfs.BookingRule, bufferSize)
+	go func() {
+		for _, ent := range mr.BookingRuleList {
+			out <- ent
+		}
+		close(out)
+	}()
+	return out
+}
+
+func (mr *Reader) Locations() chan gtfs.Location {
+	out := make(chan gtfs.Location, bufferSize)
+	go func() {
+		for _, ent := range mr.LocationList {
+			out <- ent
+		}
+		close(out)
+	}()
+	return out
+}

adapters/empty/reader.go

@@ -139,6 +139,34 @@ func (mr *Reader) RiderCategories() chan gtfs.RiderCategory {
 	return readNullEntities[gtfs.RiderCategory](mr)
 }
 
+func (mr *Reader) Timeframes() chan gtfs.Timeframe {
+	return readNullEntities[gtfs.Timeframe](mr)
+}
+
+func (mr *Reader) Networks() chan gtfs.Network {
+	return readNullEntities[gtfs.Network](mr)
+}
+
+func (mr *Reader) RouteNetworks() chan gtfs.RouteNetwork {
+	return readNullEntities[gtfs.RouteNetwork](mr)
+}
+
+func (mr *Reader) LocationGroups() chan gtfs.LocationGroup {
+	return readNullEntities[gtfs.LocationGroup](mr)
+}
+
+func (mr *Reader) LocationGroupStops() chan gtfs.LocationGroupStop {
+	return readNullEntities[gtfs.LocationGroupStop](mr)
+}
+
+func (mr *Reader) BookingRules() chan gtfs.BookingRule {
+	return readNullEntities[gtfs.BookingRule](mr)
+}
+
+func (mr *Reader) Locations() chan gtfs.Location {
+	return readNullEntities[gtfs.Location](mr)
+}
+
 func readNullEntities[T any](reader *Reader) chan T {
 	out := make(chan T, bufferSize)
 	go func() {

adapters/multireader/reader.go

@@ -176,6 +176,22 @@ func (mr *Reader) RouteNetworks() chan gtfs.RouteNetwork {
 	return readEntities(mr, func(r adapters.Reader) chan gtfs.RouteNetwork { return r.RouteNetworks() }, setFv[*gtfs.RouteNetwork])
 }
 
+func (mr *Reader) LocationGroups() chan gtfs.LocationGroup {
+	return readEntities(mr, func(r adapters.Reader) chan gtfs.LocationGroup { return r.LocationGroups() }, setFv[*gtfs.LocationGroup])
+}
+
+func (mr *Reader) LocationGroupStops() chan gtfs.LocationGroupStop {
+	return readEntities(mr, func(r adapters.Reader) chan gtfs.LocationGroupStop { return r.LocationGroupStops() }, setFv[*gtfs.LocationGroupStop])
+}
+
+func (mr *Reader) BookingRules() chan gtfs.BookingRule {
+	return readEntities(mr, func(r adapters.Reader) chan gtfs.BookingRule { return r.BookingRules() }, setFv[*gtfs.BookingRule])
+}
+
+func (mr *Reader) Locations() chan gtfs.Location {
+	return readEntities(mr, func(r adapters.Reader) chan gtfs.Location { return r.Locations() }, setFv[*gtfs.Location])
+}
+
 type canSetFV interface {
 	SetFeedVersionID(int)
 }

cmds/merge_cmd_test.go

@@ -5,16 +5,16 @@ import (
 	"testing"
 
 	"github.com/interline-io/transitland-lib/ext"
-	"github.com/interline-io/transitland-lib/internal/testutil"
+	"github.com/interline-io/transitland-lib/internal/testreader"
 	"github.com/interline-io/transitland-lib/tt"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestMerge(t *testing.T) {
 	ctx := context.TODO()
 	t.Run("merge", func(t *testing.T) {
-		f1 := testutil.ExampleFeedBART
-		f2 := testutil.ExampleFeedCaltrain
+		f1 := testreader.ExampleFeedBART
+		f2 := testreader.ExampleFeedCaltrain
 		cmd := MergeCommand{}
 		tdir := t.TempDir()
 		if err := cmd.Parse([]string{tdir, f1.URL, f2.URL}); err != nil {
@@ -28,7 +28,7 @@ func TestMerge(t *testing.T) {
 			t.Fatal(err)
 		}
 		entCount := map[string]int{}
-		testutil.AllEntities(outReader, func(ent tt.Entity) {
+		testreader.AllEntities(outReader, func(ent tt.Entity) {
 			entCount[ent.Filename()] += 1
 		})
 		expectCount := map[string]int{}

copier/copier.go

@@ -291,6 +291,11 @@ func NewCopier(ctx context.Context, reader adapters.Reader, writer adapters.Writ
 			&rules.CalendarDuplicateDates{},
 			&rules.FareProductRiderCategoryDefaultCheck{},
 			&rules.TransferStopLocationTypeCheck{},
+			// GTFS-Flex validators
+			&rules.FlexGeographyIDUniqueCheck{},
+			&rules.FlexStopLocationTypeCheck{},
+			&rules.FlexLocationGeometryCheck{},
+			&rules.FlexZoneIDConditionalCheck{},
 		)
 	}
 
@@ -612,6 +617,10 @@ func (copier *Copier) Copy(ctx context.Context) (*Result, error) {
 			)
 		},
 		copier.copyCalendars,
+		func() error { return batchCopy(copier, batchChan(r.BookingRules(), bs, nil)) },
+		func() error { return batchCopy(copier, batchChan(r.Locations(), bs, nil)) },
+		func() error { return batchCopy(copier, batchChan(r.LocationGroups(), bs, nil)) },
+		func() error { return batchCopy(copier, batchChan(r.LocationGroupStops(), bs, nil)) },
 		copier.copyTripsAndStopTimes,
 		func() error { return batchCopy(copier, batchChan(r.Pathways(), bs, nil)) },
 		func() error { return batchCopy(copier, batchChan(r.FareAttributes(), bs, nil)) },
@@ -1016,6 +1025,9 @@ func (copier *Copier) logCount(ent tt.Entity) {
 func (copier *Copier) createMissingShape(shapeID string, stoptimes []gtfs.StopTime) (string, error) {
 	stopids := []string{}
 	for _, st := range stoptimes {
+		if !st.StopID.Valid {
+			continue
+		}
 		stopids = append(stopids, st.StopID.Val)
 	}
 	line, dists, err := copier.geomCache.MakeShape(stopids...)

copier/stoppattern.go

@@ -9,9 +9,11 @@ import (
 )
 
 func stopPatternKey(stoptimes []gtfs.StopTime) string {
-	key := make([]string, len(stoptimes))
+	key := make([]string, 0, len(stoptimes))
 	for i := 0; i < len(stoptimes); i++ {
-		key[i] = stoptimes[i].StopID.Val
+		if stoptimes[i].StopID.Valid {
+			key = append(key, stoptimes[i].StopID.Val)
+		}
 	}
 	return strings.Join(key, string(byte(0)))
 }
@@ -34,11 +36,15 @@ func journeyPatternKey(trip *gtfs.Trip) string {
 	)))
 	for i := 0; i < len(trip.StopTimes); i++ {
 		st := trip.StopTimes[i]
+		stopID := ""
+		if st.StopID.Valid {
+			stopID = st.StopID.Val
+		}
 		m.Write([]byte(fmt.Sprintf(
 			"%d-%d-%s-%s-%d-%d-%d",
 			st.ArrivalTime.Val-a.Val,
 			st.DepartureTime.Val-b.Val,
-			st.StopID.Val,
+			stopID,
 			st.StopHeadsign.Val,
 			st.PickupType.Val,
 			st.DropOffType.Val,

copier/stoppattern_test.go

@@ -11,7 +11,7 @@ import (
 func Benchmark_stopPatternKey(b *testing.B) {
 	stoptimes := []gtfs.StopTime{}
 	for i := 0; i < 50; i++ {
-		stoptimes = append(stoptimes, gtfs.StopTime{StopID: tt.NewString(fmt.Sprintf("%d", i*100))})
+		stoptimes = append(stoptimes, gtfs.StopTime{StopID: tt.NewKey(fmt.Sprintf("%d", i*100))})
 	}
 	m := map[string]int{}
 	b.ResetTimer()

dmfr/feed_version_tables.go

@@ -87,6 +87,7 @@ func GetFeedVersionTables() FeedVersionTables {
 			"gtfs_fare_rules",
 			"gtfs_attributions",
 			"gtfs_translations",
+			"gtfs_location_group_stops",
 		},
 		GtfsEntityTables: []string{
 			"gtfs_fare_media",
@@ -102,6 +103,9 @@ func GetFeedVersionTables() FeedVersionTables {
 			"gtfs_fare_attributes",
 			"gtfs_trips",
 			"gtfs_shapes",
+			"gtfs_booking_rules",
+			"gtfs_location_groups",
+			"gtfs_locations",
 			"gtfs_calendars",
 			"gtfs_stops",
 			"gtfs_levels",

ext/builders/agency_place_builder.go

@@ -62,6 +62,9 @@ func (pp *AgencyPlaceBuilder) AfterWrite(eid string, ent tt.Entity, emap *tt.Ent
 	case *gtfs.Trip:
 		pp.tripAgency[eid] = pp.routeAgency[v.RouteID.Val]
 	case *gtfs.StopTime:
+		if !v.StopID.Valid {
+			return nil
+		}
 		aid := pp.tripAgency[v.TripID.Val]
 		if sg, ok := pp.stops[v.StopID.Val]; ok {
 			if v, ok := pp.agencyStops[aid]; ok {

ext/builders/convex_hull_builder.go

@@ -73,6 +73,9 @@ func (pp *ConvexHullBuilder) AfterWrite(eid string, ent tt.Entity, emap *tt.Enti
 	case *gtfs.Trip:
 		pp.tripRoutes[eid] = v.RouteID.Val
 	case *gtfs.StopTime:
+		if !v.StopID.Valid {
+			return nil
+		}
 		r, ok := pp.routeStopGeoms[pp.tripRoutes[v.TripID.Val]]
 		if !ok {
 			// log.For(ctx).Debug().Msgf("no route:", v.TripID, pp.tripRoutes[v.TripID])

ext/builders/convex_hull_builder_test.go

@@ -4,7 +4,7 @@ import (
 	"testing"
 
 	"github.com/interline-io/transitland-lib/internal/testpath"
-	"github.com/interline-io/transitland-lib/internal/testutil"
+	"github.com/interline-io/transitland-lib/internal/testreader"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -19,7 +19,7 @@ func TestConvexHullBuilder(t *testing.T) {
 	}
 	groups := map[string]testgroup{
 		"ExampleFeed": {
-			testutil.ExampleZip.URL,
+			testreader.ExampleZip.URL,
 			[]testcase{
 				{
 					FeedVersionGeometry: []float64{-117.133162, 36.425288, -116.81797, 36.88108, -116.76821, 36.914893, -116.751677, 36.915682, -116.40094, 36.641496, -117.133162, 36.425288},
@@ -30,7 +30,7 @@ func TestConvexHullBuilder(t *testing.T) {
 			},
 		},
 		"Caltrain": {
-			testutil.ExampleFeedCaltrain.URL,
+			testreader.ExampleFeedCaltrain.URL,
 			[]testcase{{
 				FeedVersionGeometry: []float64{-121.566225, 37.003485, -122.232, 37.486101, -122.386832, 37.599797, -122.412076, 37.631108, -122.394992, 37.77639, -122.394935, 37.776348, -121.650244, 37.129363, -121.610936, 37.086653, -121.610049, 37.085225, -121.566088, 37.003538, -121.566225, 37.003485},
 				AgencyGeoms: map[string][]float64{
@@ -39,7 +39,7 @@ func TestConvexHullBuilder(t *testing.T) {
 			}},
 		},
 		"BART": {
-			testutil.ExampleFeedBART.URL,
+			testreader.ExampleFeedBART.URL,
 			[]testcase{
 				{
 					FeedVersionGeometry: []float64{-121.939313, 37.502171, -122.386702, 37.600271, -122.466233, 37.684638, -122.469081, 37.706121, -122.353099, 37.936853, -122.024653, 38.003193, -121.945154, 38.018914, -121.889457, 38.016941, -121.78042, 37.995388, -121.939313, 37.502171},

ext/builders/onestop_id_builder.go

@@ -94,6 +94,9 @@ func (pp *OnestopIDBuilder) AfterWrite(eid string, ent tt.Entity, emap *tt.Entit
 	case *gtfs.Trip:
 		pp.tripRoutes[eid] = v.RouteID.Val
 	case *gtfs.StopTime:
+		if !v.StopID.Valid {
+			return nil
+		}
 		r, ok := pp.routeStopGeoms[pp.tripRoutes[v.TripID.Val]]
 		if !ok {
 			// log.For(ctx).Debug().Msgf("OnestopIDBuilder no route:", v.TripID, pp.tripRoutes[v.TripID])

ext/builders/onestop_id_builder_test.go

@@ -3,7 +3,7 @@ package builders
 import (
 	"testing"
 
-	"github.com/interline-io/transitland-lib/internal/testutil"
+	"github.com/interline-io/transitland-lib/internal/testreader"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -19,7 +19,7 @@ func TestOnestopIDBuilder(t *testing.T) {
 	}
 	groups := map[string]testgroup{
 		"ExampleFeed": {
-			testutil.ExampleZip.URL,
+			testreader.ExampleZip.URL,
 			[]testcase{
 				{"o-9qs-demotransitauthority", hw{"o-9qs-demotransitauthority": []string{"DTA"}}},
 				{"r-9qsb-20", hw{"r-9qsb-20": []string{"BFC"}}},
@@ -39,7 +39,7 @@ func TestOnestopIDBuilder(t *testing.T) {
 			},
 		},
 		"Caltrain": {
-			testutil.ExampleFeedCaltrain.URL,
+			testreader.ExampleFeedCaltrain.URL,
 			[]testcase{
 				{"o-9q9-caltrain", hw{"o-9q9-caltrain": []string{"caltrain-ca-us"}}},
 				{"r-9q9-limited", hw{"r-9q9-limited": []string{"Li-130"}}},
@@ -57,7 +57,7 @@ func TestOnestopIDBuilder(t *testing.T) {
 			},
 		},
 		"BART": {
-			testutil.ExampleFeedBART.URL,
+			testreader.ExampleFeedBART.URL,
 			[]testcase{
 				{"o-9q9-bayarearapidtransit", hw{"o-9q9-bayarearapidtransit": []string{"BART"}}},
 				{"r-9q8y-richmond~dalycity~millbrae", hw{"r-9q8y-richmond~dalycity~millbrae": []string{"07"}}},