szczanieckiej/traffic/convert.go

package traffic

// todo(BAF10) direction (0|1) to const (TO|BACK)
// fixme converting lines takes much time (graphs)

import (
	"apiote.xyz/p/szczanieckiej/config"
	"apiote.xyz/p/szczanieckiej/file"

	"bufio"
	"encoding/csv"
	"errors"
	"fmt"
	"image/color"
	"io"
	"io/fs"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"sort"
	"syscall"
	"time"

	gott2 "apiote.xyz/p/gott/v2"
	"git.sr.ht/~sircmpwn/go-bare"
	"notabug.org/apiote/gott"
)

type ErrEmpty struct{}

func (ErrEmpty) Error() string {
	return ""
}

type result struct {
	config             config.Config
	pid                int
	tmpPath            string
	feed               Feed
	feedName           string
	location           *time.Location
	tmpFeedPath        string
	homeFeedPath       string
	downloadedVersions []Version
	allVersions        []Version
	gtfsFilenames      []string
	missingVersions    []Version
	updatesFile        *os.File
	updates            map[string]string
}

type feedConverter struct {
	TmpFeedPath  string
	GtfsFilename string
	Feed         Feed
	HomeFeedPath string

	TrafficCalendarFile *os.File
	Departures          map[string][]Departure
	TripsThroughStop    map[string][]StopOrder
	LineNames           map[string]string
	TripsOffsets        map[string]uint
	TripChangeOpts      map[string]ChangeOption
	StopsCodeIndex      CodeIndex
	StopsNameIndex      map[string][]uint
	Stops               map[string]Stop
	LineGraphs          map[string]map[uint]LineGraph
	LineIndex           map[string][]uint
}

// helper functions

func hex2colour(hex string) color.RGBA {
	if hex[0] == '#' {
		hex = hex[1:]
	}
	colour := color.RGBA{
		A: 0xff,
	}
	hexToByte := func(b byte) byte {
		switch {
		case b >= '0' && b <= '9':
			return b - '0'
		case b >= 'a' && b <= 'f':
			return b - 'a' + 10
		case b >= 'A' && b <= 'F':
			return b - 'A' + 10
		default:
			return 0
		}
	}
	colour.R = hexToByte(hex[0])<<4 + hexToByte(hex[1])
	colour.G = hexToByte(hex[2])<<4 + hexToByte(hex[3])
	colour.B = hexToByte(hex[4])<<4 + hexToByte(hex[5])
	return colour
}

func contains(array []Version, item Version) bool {
	for _, i := range array {
		if item.ValidFrom.Equal(i.ValidFrom) && i.ValidTill.Equal(i.ValidTill) {
			return true
		}
	}
	return false
}

// converting functions

func createTmpPath(input ...interface{}) (interface{}, error) {
	args := input[0].(result)
	p := filepath.Join(args.tmpPath, args.feedName)
	err := os.MkdirAll(p, 0755)
	args.tmpFeedPath = p
	return gott.Tuple{args}, err
}

func createFeedHome(input ...interface{}) (interface{}, error) {
	args := input[0].(result)
	p := filepath.Join(args.config.FeedsPath, args.feedName)
	err := os.MkdirAll(p, 0755)
	args.homeFeedPath = p
	return gott.Tuple{args}, err
}

func listDownloadedVersions(input ...interface{}) (interface{}, error) {
	args := input[0].(result)
	v, err := ListVersions(args.config, args.feed)
	args.downloadedVersions = v
	return gott.Tuple{args}, err
}

func getAllVersions(input ...interface{}) (interface{}, error) {
	args := input[0].(result)
	v, err := args.feed.GetVersions(time.Now().In(args.location))
	args.allVersions = v
	return gott.Tuple{args}, err
}

func findValidVersions(input ...interface{}) interface{} {
	args := input[0].(result)
	dateValidVersions := []Version{}
	now := time.Now().In(args.location)
	for _, version := range args.allVersions {
		if version.ValidFrom.Before(now) && now.Before(version.ValidTill) {
			dateValidVersions = append(dateValidVersions, version)
		}
	}
	sort.Slice(dateValidVersions, func(i, j int) bool {
		return dateValidVersions[i].ValidFrom.Before(dateValidVersions[j].ValidFrom)
	})
	sort.Slice(args.downloadedVersions, func(i, j int) bool {
		return args.downloadedVersions[i].ValidFrom.Before(args.downloadedVersions[j].ValidFrom)
	})
	validVersions := FindValidVersions(dateValidVersions, now)

	missingVersions := []Version{}
	for _, version := range validVersions {
		if !contains(args.downloadedVersions, version) {
			missingVersions = append(missingVersions, version)
		}
	}
	args.missingVersions = missingVersions
	return gott.Tuple{args}
}

func getGtfsFiles(input ...interface{}) (interface{}, error) {
	args := input[0].(result)
	names := []string{}
	for _, version := range args.missingVersions {
		name := version.String() + ".zip"
		zipPath := filepath.Join(args.tmpFeedPath, name)
		url := version.Link
		response, err := http.Get(url)
		if err != nil {
			return gott.Tuple{args}, fmt.Errorf("while downloading gtfs %s %w\n", name, err)
		}
		file, err := os.Create(zipPath)
		if err != nil {
			return gott.Tuple{args}, fmt.Errorf("while downloading gtfs %s %w\n", name, err)
		}
		defer file.Close()
		_, err = io.Copy(file, response.Body)
		if err != nil {
			return gott.Tuple{args}, fmt.Errorf("while downloading gtfs %s %w\n", name, err)
		}
		names = append(names, name)
	}
	args.gtfsFilenames = names
	return gott.Tuple{args}, nil
}

func unzipGtfs(c feedConverter) error {
	return file.UnzipGtfs(c.TmpFeedPath, c.GtfsFilename)
}

func convertVehicles(c feedConverter) error {
	return c.Feed.ConvertVehicles(c.TmpFeedPath)
}

func createTrafficCalendarFile(c feedConverter) (feedConverter, error) {
	path := c.TmpFeedPath
	var err error
	c.TrafficCalendarFile, err = os.Create(filepath.Join(path, "calendar.bare"))
	return c, err
}

func recoverCalendar(c feedConverter, e error) (feedConverter, error) {
	var pathError *os.PathError
	if errors.As(e, &pathError) && errors.Is(pathError, fs.ErrNotExist) {
		return c, nil
	}
	return c, e
}

func convertCalendar(c feedConverter) (feedConverter, error) {
	path := c.TmpFeedPath
	resultFile := c.TrafficCalendarFile
	calendarFile, err := os.Open(filepath.Join(path, "calendar.txt"))
	if err != nil {
		return c, fmt.Errorf("while opening file: %w", err)
	}
	defer calendarFile.Close()

	r := csv.NewReader(bufio.NewReader(calendarFile))
	header, err := r.Read()
	if err != nil {
		return c, fmt.Errorf("while reading header: %w", err)
	}
	fields := map[string]int{}
	for i, headerField := range header {
		fields[headerField] = i
	}

	for {
		schedule := Schedule{}
		record, err := r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return c, fmt.Errorf("while reading a record: %w", err)
		}
		schedule.ScheduleID = record[fields["service_id"]]
		startDate := record[fields["start_date"]]
		endDate := record[fields["end_date"]]
		schedule.StartDate = startDate[:4] + "-" + startDate[4:6] + "-" + startDate[6:]
		schedule.EndDate = endDate[:4] + "-" + endDate[4:6] + "-" + endDate[6:]
		if record[fields["monday"]] == "1" {
			schedule.Weekdays |= (1 << 1)
		}
		if record[fields["tuesday"]] == "1" {
			schedule.Weekdays |= (1 << 2)
		}
		if record[fields["wednesday"]] == "1" {
			schedule.Weekdays |= (1 << 3)
		}
		if record[fields["thursday"]] == "1" {
			schedule.Weekdays |= (1 << 4)
		}
		if record[fields["friday"]] == "1" {
			schedule.Weekdays |= (1 << 5)
		}
		if record[fields["saturday"]] == "1" {
			schedule.Weekdays |= (1 << 6)
		}
		if record[fields["sunday"]] == "1" {
			schedule.Weekdays |= (1 << 0)
			schedule.Weekdays |= (1 << 7)
		}
		bytes, err := bare.Marshal(&schedule)
		if err != nil {
			return c, fmt.Errorf("while marshalling: %w", err)
		}
		_, err = resultFile.Write(bytes)
		if err != nil {
			return c, fmt.Errorf("while writing: %w", err)
		}
	}
	return c, nil
}

func convertCalendarDates(c feedConverter) (feedConverter, error) {
	path := c.TmpFeedPath
	resultFile := c.TrafficCalendarFile
	datesFile, err := os.Open(filepath.Join(path, "calendar_dates.txt"))
	if err != nil {
		return c, fmt.Errorf("while opening file: %w", err)
	}
	defer datesFile.Close()

	r := csv.NewReader(bufio.NewReader(datesFile))
	header, err := r.Read()
	if err != nil {
		return c, fmt.Errorf("while reading header: %w", err)
	}
	fields := map[string]int{}
	for i, headerField := range header {
		fields[headerField] = i
	}

	for {
		schedule := Schedule{}
		record, err := r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return c, fmt.Errorf("while reading a record: %w", err)
		}
		if record[fields["exception_type"]] == "1" {
			date := record[fields["date"]]
			schedule.ScheduleID = record[fields["service_id"]]
			schedule.StartDate = date[:4] + "-" + date[4:6] + "-" + date[6:]
			schedule.EndDate = date[:4] + "-" + date[4:6] + "-" + date[6:]
			schedule.Weekdays = 0xff
		} else {
			continue
		}
		bytes, err := bare.Marshal(&schedule)
		if err != nil {
			return c, fmt.Errorf("while marshalling: %w", err)
		}
		_, err = resultFile.Write(bytes)
		if err != nil {
			return c, fmt.Errorf("while writing: %w", err)
		}
	}
	return c, nil
}

func checkAnyCalendarConverted(c feedConverter) error {
	info, err := c.TrafficCalendarFile.Stat()
	if err == nil && info.Size() == 0 {
		return fmt.Errorf("no calendar converted")
	}
	return err
}

func closeTrafficCalendarFile(c feedConverter, e error) (feedConverter, error) {
	if c.TrafficCalendarFile != nil {
		c.TrafficCalendarFile.Close()
	}
	return c, e
}

func convertDepartures(c feedConverter) (feedConverter, error) { // O(n:stop_times) ; ( -- departures:map[tripID][]departure, tripsThroughStop:map[stopID][]{tripID,order} >> )
	path := c.TmpFeedPath

	file, err := os.Open(filepath.Join(path, "stop_times.txt"))
	if err != nil {
		return c, fmt.Errorf("while opening file: %w", err)
	}
	defer file.Close()

	departures := map[string][]Departure{}

	r := csv.NewReader(bufio.NewReader(file))
	header, err := r.Read()
	if err != nil {
		return c, fmt.Errorf("while reading header: %w", err)
	}
	fields := map[string]int{}
	for i, headerField := range header {
		fields[headerField] = i
	}

	tripsThroughStop := map[string][]StopOrder{}

	for {
		departure := Departure{}
		record, err := r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return c, fmt.Errorf("while reading a record: %w", err)
		}

		stopID := record[fields["stop_id"]]

		tripID := record[fields["trip_id"]]
		fmt.Sscanf(record[fields["stop_sequence"]], "%d", &departure.StopSeq)
		fmt.Sscanf(record[fields["pickup_type"]], "%d", &departure.Pickup)
		fmt.Sscanf(record[fields["drop_off_type"]], "%d", &departure.Dropoff)

		tripsThroughStop[stopID] = append(tripsThroughStop[stopID], StopOrder{
			TripID: tripID,
			Order:  departure.StopSeq,
		})

		var hours, minutes uint
		fmt.Sscanf(record[fields["arrival_time"]], "%d:%d", &hours, &minutes)
		departure.Time = hours*60 + minutes

		departures[tripID] = append(departures[tripID], departure)
	}

	c.Departures = departures
	c.TripsThroughStop = tripsThroughStop
	return c, nil
}

func getLineNames(c feedConverter) (feedConverter, error) { // O(n:routes) ; ( -- lineNames:map[routeID]lineName >> )
	path := c.TmpFeedPath

	file, err := os.Open(filepath.Join(path, "routes.txt"))
	if err != nil {
		return c, fmt.Errorf("while opening file: %w", err)
	}
	defer file.Close()
	r := csv.NewReader(bufio.NewReader(file))
	header, err := r.Read()
	if err != nil {
		return c, fmt.Errorf("while reading header: %w", err)
	}
	fields := map[string]int{}
	for i, headerField := range header {
		fields[headerField] = i
	}

	names := map[string]string{}

	for {
		record, err := r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return c, fmt.Errorf("while reading a record: %w", err)
		}

		routeID := record[fields["route_id"]]
		names[routeID] = record[fields["route_short_name"]]
	}

	c.LineNames = names
	return c, nil
}

func convertTrips(c feedConverter) (feedConverter, error) { // O(n:trips) ; (departures, lineNames -- tripsOffsets:map[tripID]offset, tripsChangeOpts:map[tripID]{lineID,headsign} >> trips)
	path := c.TmpFeedPath
	departures := c.Departures
	lineNames := c.LineNames

	file, err := os.Open(filepath.Join(path, "trips.txt"))
	if err != nil {
		return c, fmt.Errorf("while opening file: %w", err)
	}
	defer file.Close()

	result, err := os.Create(filepath.Join(path, "trips.bare"))
	if err != nil {
		return c, fmt.Errorf("while creating file: %w", err)
	}
	defer result.Close()

	r := csv.NewReader(bufio.NewReader(file))
	header, err := r.Read()
	if err != nil {
		return c, fmt.Errorf("while reading header: %w", err)
	}
	fields := map[string]int{}
	for i, headerField := range header {
		fields[headerField] = i
	}

	var offset uint = 0
	tripsOffsets := map[string]uint{}

	tripChangeOpts := map[string]ChangeOption{}

	for {
		trip := Trip{}
		record, err := r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return c, fmt.Errorf("while reading a record: %w", err)
		}

		trip.ID = record[fields["trip_id"]]
		trip.Headsign = record[fields["trip_headsign"]]
		trip.Departures = departures[trip.ID]
		trip.ScheduleID = record[fields["service_id"]]
		trip.LineName = lineNames[record[fields["route_id"]]]
		fmt.Sscanf(record[fields["direction_id"]], "%d", &trip.Direction)

		tripChangeOpts[trip.ID] = ChangeOption{
			LineID:   record[fields["route_id"]],
			Headsign: trip.Headsign,
		}

		bytes, err := bare.Marshal(&trip)
		if err != nil {
			return c, fmt.Errorf("while marshalling: %w", err)
		}
		b, err := result.Write(bytes)
		if err != nil {
			return c, fmt.Errorf("while writing: %w", err)
		}
		tripsOffsets[trip.ID] = offset
		offset += uint(b)
	}

	c.TripsOffsets = tripsOffsets
	c.TripChangeOpts = tripChangeOpts
	return c, nil
}

func convertStops(c feedConverter) (feedConverter, error) { // O(n:stops) ; (tripsThroughStop, lineNames, tripChangeOpts, tripOffsets -- stopsOffsetsByCode:CodeIndex, stopsOffsetsByName:map[name][]offsets >> stops)
	path := c.TmpFeedPath
	tripsThroughStop := c.TripsThroughStop
	lineNames := c.LineNames
	tripChangeOpts := c.TripChangeOpts
	tripsOffsets := c.TripsOffsets

	file, err := os.Open(filepath.Join(path, "stops.txt"))
	if err != nil {
		return c, fmt.Errorf("while opening file: %w", err)
	}
	defer file.Close()

	result, err := os.Create(filepath.Join(path, "stops.bare"))
	if err != nil {
		return c, fmt.Errorf("while creating file: %w", err)
	}
	defer result.Close()

	r := csv.NewReader(bufio.NewReader(file))
	header, err := r.Read()
	if err != nil {
		return c, fmt.Errorf("while reading header: %w", err)
	}
	fields := map[string]int{}
	for i, headerField := range header {
		fields[headerField] = i
	}

	var offset uint = 0
	stopsOffsetsByName := map[string][]uint{}
	stopsOffsetsByCode := CodeIndex{}
	stops := map[string]Stop{}

	for {
		stop := Stop{}
		record, err := r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return c, fmt.Errorf("while reading a record: %w", err)
		}

		stopID := record[fields["stop_id"]]

		stopTrips := tripsThroughStop[stopID]

		stop.ID = stopID
		stop.Code = record[fields["stop_code"]]
		if stop.Code == "" {
			stop.Code = stopID
		}
		stop.Name = record[fields["stop_name"]]
		stop.Zone = record[fields["zone_id"]]

		stops[record[fields["stop_id"]]] = stop

		if field, ok := fields["stop_timezone"]; ok {
			stop.Timezone = record[field]
		}

		var lat, lon float64
		fmt.Sscanf(record[fields["stop_lat"]], "%f", &lat)
		fmt.Sscanf(record[fields["stop_lon"]], "%f", &lon)
		stop.Coordinates = Position{lat, lon}

		changeOptionMap := map[string]ChangeOption{}
		stop.ChangeOptions = []ChangeOption{}
		stop.Order = []StopOrder{}
		for _, stopTrip := range stopTrips {
			changeOption := ChangeOption{
				LineName: lineNames[tripChangeOpts[stopTrip.TripID].LineID],
				Headsign: tripChangeOpts[stopTrip.TripID].Headsign,
			}
			stopOrder := StopOrder{
				TripOffset: tripsOffsets[stopTrip.TripID],
				Order:      stopTrip.Order,
			}
			stop.Order = append(stop.Order, stopOrder)
			changeOptionMap[changeOption.LineName+"->"+changeOption.Headsign] = changeOption
		}
		for _, option := range changeOptionMap {
			stop.ChangeOptions = append(stop.ChangeOptions, option)
		}
		sort.Slice(stop.ChangeOptions, func(i, j int) bool {
			var num1, num2 int
			_, err1 := fmt.Sscanf(stop.ChangeOptions[i].LineName, "%d", &num1)
			_, err2 := fmt.Sscanf(stop.ChangeOptions[j].LineName, "%d", &num2)
			if err1 != nil && err2 != nil {
				return stop.ChangeOptions[i].LineName < stop.ChangeOptions[j].LineName
			} else if err1 != nil {
				return false
			} else if err2 != nil {
				return true
			} else {
				return num1 < num2
			}
		})

		bytes, err := bare.Marshal(&stop)
		if err != nil {
			return c, fmt.Errorf("while marshalling: %w", err)
		}
		b, err := result.Write(bytes)
		if err != nil {
			return c, fmt.Errorf("while writing: %w", err)
		}
		stopsOffsetsByName[stop.Name] = append(stopsOffsetsByName[stop.Name], offset)
		stopsOffsetsByCode[ID(stop.Code)] = offset
		offset += uint(b)
	}

	c.StopsCodeIndex = stopsOffsetsByCode
	c.StopsNameIndex = stopsOffsetsByName
	c.Stops = stops
	return c, nil
}

func convertLineGraphs(c feedConverter) (feedConverter, error) { // O(n:stop_times) ; (tripsOffsets, stops -- lineGrapsh:map[lineName]map[direction]graph >> )
	path := c.TmpFeedPath
	tripsOffsets := c.TripsOffsets
	stops := c.Stops

	file, err := os.Open(filepath.Join(path, "stop_times.txt"))
	if err != nil {
		return c, fmt.Errorf("while opening stop_times: %w", err)
	}
	defer file.Close()

	trips, err := os.Open(filepath.Join(path, "trips.bare"))
	if err != nil {
		return c, fmt.Errorf("while opening trips: %w", err)
	}
	defer trips.Close()

	r := csv.NewReader(bufio.NewReader(file))
	header, err := r.Read()
	if err != nil {
		return c, fmt.Errorf("while reading header: %w", err)
	}
	fields := map[string]int{}
	for i, headerField := range header {
		fields[headerField] = i
	}

	//            lineNa     dire
	graphs := map[string]map[uint]LineGraph{}

	previousTripID := ""
	previous := -1
	previousTrip := Trip{}

	for {
		record, err := r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return c, fmt.Errorf("while reading a record: %w", err)
		}
		tripID := record[fields["trip_id"]]
		stop := stops[record[fields["stop_id"]]]
		tripOffset := tripsOffsets[tripID]
		_, err = trips.Seek(int64(tripOffset), 0)
		if err != nil {
			return c, fmt.Errorf("while seeking: %w", err)
		}
		trip := unmarshalTripFromFile(trips)

		if _, ok := graphs[trip.LineName]; !ok {
			graphs[trip.LineName] = map[uint]LineGraph{}
			graphs[trip.LineName][0] = LineGraph{
				NextNodes: map[int][]int{},
				PrevNodes: map[int][]int{},
			}
			graphs[trip.LineName][1] = LineGraph{
				NextNodes: map[int][]int{},
				PrevNodes: map[int][]int{},
			}
		}

		if previousTripID != tripID {
			if previousTripID != "" {
				// last of previous trip
				previousGraph := graphs[previousTrip.LineName][previousTrip.Direction]
				connectionDone := false
				for _, n := range previousGraph.NextNodes[previous] {
					if n == -1 {
						connectionDone = true
					}
				}
				if !connectionDone {
					previousGraph.NextNodes[previous] = append(previousGraph.NextNodes[previous], -1)
				}
				graphs[previousTrip.LineName][previousTrip.Direction] = previousGraph
			}
		}

		graph := graphs[trip.LineName][trip.Direction]
		current := -1
		for i, code := range graph.StopCodes {
			if code == stop.Code {
				current = i
				break
			}
		}
		if current == -1 {
			current = len(graph.StopCodes)
			graph.StopCodes = append(graph.StopCodes, stop.Code)
		}
		if previousTripID != tripID {
			// first of current trip
			connectionDone := false
			for _, n := range graph.PrevNodes[current] {
				if n == -1 {
					connectionDone = true
				}
			}
			if !connectionDone {
				graph.PrevNodes[current] = append(graph.PrevNodes[current], -1)
			}
		} else {
			// second <- first to last <- penultimate of current trip
			connectionDone := false
			for _, n := range graph.NextNodes[previous] {
				if n == current {
					connectionDone = true
				}
			}
			if !connectionDone {
				graph.NextNodes[previous] = append(graph.NextNodes[previous], current)
				graph.PrevNodes[current] = append(graph.PrevNodes[current], previous)
			}
		}

		previous = current
		previousTripID = tripID
		previousTrip = trip

		graphs[trip.LineName][trip.Direction] = graph
	}

	g := graphs[previousTrip.LineName][previousTrip.Direction]
	connectionDone := false
	for _, n := range g.NextNodes[previous] {
		if n == -1 {
			connectionDone = true
		}
	}
	if !connectionDone {
		g.NextNodes[previous] = append(g.NextNodes[previous], -1)
	}

	c.LineGraphs = graphs
	return c, nil
}

func convertLines(c feedConverter) (feedConverter, error) { // O(n:routes) ; (lineGraphs -- lineIndex:map[lineName][]offsets >> lines)
	path := c.TmpFeedPath
	feed := c.Feed

	file, err := os.Open(filepath.Join(path, "routes.txt"))
	if err != nil {
		return c, fmt.Errorf("while opening file: %w", err)
	}
	defer file.Close()

	result, err := os.Create(filepath.Join(path, "lines.bare"))
	if err != nil {
		return c, fmt.Errorf("while creating file: %w", err)
	}
	defer result.Close()

	r := csv.NewReader(bufio.NewReader(file))
	header, err := r.Read()
	if err != nil {
		return c, fmt.Errorf("while reading header: %w", err)
	}
	fields := map[string]int{}
	for i, headerField := range header {
		fields[headerField] = i
	}

	names := map[string]string{}

	var offset uint = 0
	index := map[string][]uint{}

	for {
		record, err := r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return c, fmt.Errorf("while reading a record: %w", err)
		}

		routeID := record[fields["route_id"]]
		names[routeID] = record[fields["route_short_name"]]
		lineName := record[fields["route_short_name"]]
		var kind uint
		fmt.Sscanf(record[fields["route_type"]], "%d", &kind)
		var (
			colour string = record[fields["route_color"]]
		)

		if colour == "" {
			colour = "ffffff"
		}

		line := Line{
			ID:         routeID,
			Name:       lineName,
			Colour:     hex2colour(colour),
			Type:       kind,
			GraphThere: c.LineGraphs[lineName][0],
			GraphBack:  c.LineGraphs[lineName][1],
		}

		if field, present := fields["agency_id"]; present {
			line.AgencyID = ID(record[field])
		}

		bytes, err := bare.Marshal(&line)
		if err != nil {
			return c, fmt.Errorf("while marshalling: %w", err)
		}
		b, err := result.Write(bytes)
		if err != nil {
			return c, fmt.Errorf("while writing: %w", err)
		}
		cleanQuery, err := CleanQuery(line.Name, feed)
		if err != nil {
			return c, fmt.Errorf("while cleaning line name: %w", err)
		}
		index[cleanQuery] = []uint{offset}
		offset += uint(b)
	}

	c.LineIndex = index
	return c, nil
}

func convertFeedInfo(c feedConverter) error { // O(1:feed_info) ; ( -- >> feed_info)
	path := c.TmpFeedPath

	result, err := os.Create(filepath.Join(path, "feed_info.bare"))
	if err != nil {
		return fmt.Errorf("while creating file: %w", err)
	}
	defer result.Close()

	file, err := os.Open(filepath.Join(path, "feed_info.txt"))
	if err != nil {
		if errors.Is(err, fs.ErrNotExist) {
			log.Println("[WARN] no feed_info.txt")
			return nil
		}
		return fmt.Errorf("while opening file: %w", err)
	}
	defer file.Close()

	r := csv.NewReader(bufio.NewReader(file))
	header, err := r.Read()
	if err != nil {
		return fmt.Errorf("while reading header: %w", err)
	}
	fields := map[string]int{}
	for i, headerField := range header {
		fields[headerField] = i
	}

	feedInfo := FeedInfo{}
	record, err := r.Read()
	if err != nil {
		return fmt.Errorf("while reading a record: %w", err)
	}

	feedInfo.Name = record[fields["feed_publisher_name"]]
	feedInfo.Website = record[fields["feed_publisher_url"]]
	feedInfo.Language = record[fields["feed_lang"]]

	bytes, err := bare.Marshal(&feedInfo)
	if err != nil {
		return fmt.Errorf("while marshalling: %w", err)
	}
	_, err = result.Write(bytes)
	if err != nil {
		return fmt.Errorf("while writing: %w", err)
	}

	return nil
}

func convertAgencies(c feedConverter) error { // O(n:agency) ; ( -- >> agencies)
	path := c.TmpFeedPath

	file, err := os.Open(filepath.Join(path, "agency.txt"))
	if err != nil {
		return fmt.Errorf("while opening file: %w", err)
	}
	defer file.Close()

	result, err := os.Create(filepath.Join(path, "agencies.bare"))
	if err != nil {
		return fmt.Errorf("while creating file: %w", err)
	}
	defer file.Close()

	r := csv.NewReader(bufio.NewReader(file))
	header, err := r.Read()
	if err != nil {
		return fmt.Errorf("while reading header: %w", err)
	}
	fields := map[string]int{}
	for i, headerField := range header {
		fields[headerField] = i
	}

	for {
		record, err := r.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			return fmt.Errorf("while reading a record: %w", err)
		}

		agency := Agency{
			ID:       ID(record[fields["agency_id"]]),
			Name:     record[fields["agency_name"]],
			Website:  record[fields["agency_url"]],
			Timezone: record[fields["agency_timezone"]],
		}
		if field, present := fields["agency_lang"]; present {
			agency.Language = record[field]
		}
		if field, present := fields["agency_phone"]; present {
			agency.PhoneNumber = record[field]
		}
		if field, present := fields["agency_fare_url"]; present {
			agency.FareWebsite = record[field]
		}
		if field, present := fields["agency_email"]; present {
			agency.Email = record[field]
		}

		bytes, err := bare.Marshal(&agency)
		if err != nil {
			return fmt.Errorf("while marshalling: %w", err)
		}
		_, err = result.Write(bytes)
		if err != nil {
			return fmt.Errorf("while writing: %w", err)
		}
	}
	return nil
}

func writeNameIndex(c feedConverter, index map[string][]uint, filename string, raw bool) error {
	path := c.TmpFeedPath
	feed := c.Feed

	result, err := os.Create(filepath.Join(path, filename))
	if err != nil {
		return fmt.Errorf("while creating file: %w", err)
	}
	defer result.Close()

	for name, offsets := range index {
		cleanQuery := name
		if !raw {
			cleanQuery, err = CleanQuery(name, feed)
			if err != nil {
				return fmt.Errorf("while cleaning name %s: %w", name, err)
			}
		}
		stopOffset := NameOffset{
			Name:    cleanQuery,
			Offsets: offsets,
		}
		bytes, err := bare.Marshal(&stopOffset)
		if err != nil {
			return fmt.Errorf("while marshalling: %w", err)
		}
		_, err = result.Write(bytes)
		if err != nil {
			return fmt.Errorf("while writing: %w", err)
		}
	}
	return nil
}

func writeStopNameIndex(c feedConverter) error {
	return writeNameIndex(c, c.StopsNameIndex, "ix_stop_names.bare", false)
}

func writeLineIndex(c feedConverter) error {
	return writeNameIndex(c, c.LineIndex, "ix_lines.bare", false)
}

func writeTripIndex(c feedConverter) error {
	tripIndex := map[string][]uint{}
	for trip, offset := range c.TripsOffsets {
		tripIndex[trip] = []uint{offset}
	}
	return writeNameIndex(c, tripIndex, "ix_trips.bare", true)
}

func writeStopCodeIndex(c feedConverter) error {
	path := c.TmpFeedPath
	stopsOffsetsByCode := c.StopsCodeIndex

	result, err := os.Create(filepath.Join(path, "ix_stop_codes.bare"))
	if err != nil {
		return fmt.Errorf("while creating file: %w", err)
	}
	defer result.Close()
	bytes, err := bare.Marshal(&stopsOffsetsByCode)
	if err != nil {
		return fmt.Errorf("while marshalling: %w", err)
	}
	_, err = result.Write(bytes)
	if err != nil {
		return fmt.Errorf("while writing: %w", err)
	}
	return nil
}

func deleteTxtFiles(c feedConverter) error {
	return file.DeleteTxtFiles(c.TmpFeedPath, c.GtfsFilename)
}

func compressTraffic(c feedConverter) error {
	return file.CompressBare(c.TmpFeedPath, c.GtfsFilename)
}

func deleteBareFiles(c feedConverter) error {
	return file.DeleteBareFiles(c.TmpFeedPath)
}

func moveTraffic(c feedConverter) error {
	// todo(BAF25) filename take from calendar
	return file.MoveTraffic(c.GtfsFilename, c.TmpFeedPath, c.HomeFeedPath)
}

func convert(input ...interface{}) (interface{}, error) {
	args := input[0].(result)
	for _, gtfsFile := range args.gtfsFilenames {
		r := gott2.R[feedConverter]{
			S: feedConverter{
				TmpFeedPath:  args.tmpFeedPath,
				GtfsFilename: gtfsFile,
				Feed:         args.feed,
				HomeFeedPath: args.homeFeedPath,
			},
			LogLevel: gott2.Debug,
		}
		r = r.
			Tee(unzipGtfs).
			Tee(convertVehicles).
			Bind(createTrafficCalendarFile).
			Bind(convertCalendar).
			Recover(recoverCalendar).
			Bind(convertCalendarDates).
			Recover(recoverCalendar).
			Tee(checkAnyCalendarConverted).
			Recover(closeTrafficCalendarFile).
			Bind(convertDepartures).
			Bind(getLineNames).
			Bind(convertTrips).
			Bind(convertStops).
			Bind(convertLineGraphs).
			Bind(convertLines).
			Tee(convertFeedInfo).
			Tee(convertAgencies).
			Tee(writeStopNameIndex).
			Tee(writeLineIndex).
			Tee(writeStopCodeIndex).
			Tee(writeTripIndex).
			Tee(deleteTxtFiles).
			Tee(compressTraffic).
			Tee(deleteBareFiles).
			Tee(moveTraffic)

		if r.E != nil {
			return gott.Tuple{args}, r.E
		}
	}
	return gott.Tuple{args}, nil
}

func signal(input ...interface{}) (interface{}, error) {
	args := input[0].(result)
	if len(args.gtfsFilenames) > 0 && args.pid > 0 {
		process, err := os.FindProcess(args.pid)
		if err != nil {
			return gott.Tuple{args}, err
		}
		err = process.Signal(syscall.SIGUSR1)
		if err != nil {
			return gott.Tuple{args}, err
		}
	}
	return gott.Tuple{args}, nil
}

func openLastUpdated(input ...interface{}) (interface{}, error) {
	args := input[0].(result)
	updatesFilename := filepath.Join(args.config.FeedsPath, "updated.bare")
	var err error
	args.updatesFile, err = os.OpenFile(updatesFilename, os.O_RDWR|os.O_CREATE, 0644)
	return gott.Tuple{args}, err
}

func isEmpty(input ...interface{}) error {
	args := input[0].(result)
	stat, err := os.Stat(args.updatesFile.Name())
	if err != nil {
		return err
	}
	if stat.Size() == 0 {
		return ErrEmpty{}
	}
	return nil
}

func unmarshalLastUpdated(input ...interface{}) (interface{}, error) {
	args := input[0].(result)
	var lastUpdated map[string]string
	err := bare.UnmarshalReader(args.updatesFile, &lastUpdated)
	args.updates = lastUpdated
	return gott.Tuple{args}, err
}

func recoverEmpty(input ...interface{}) (interface{}, error) {
	args := input[0].(result)
	err := input[1].(error)
	var emptyError ErrEmpty
	if errors.As(err, &emptyError) {
		return gott.Tuple{args}, nil
	} else {
		return gott.Tuple{args}, err
	}
}

func lastUpdated(input ...interface{}) interface{} {
	args := input[0].(result)
	args.updates[args.feed.String()] = time.Now().Format(time.RFC3339)
	return gott.Tuple{args}
}

func marshalLastUpdated(input ...interface{}) error {
	args := input[0].(result)
	err := bare.MarshalWriter(bare.NewWriter(args.updatesFile), &args.updates)
	return err
}

func Prepare(cfg config.Config, t Traffic, bimbaPid int) error { // todo(BAF18) remove pid
	for _, feed := range t.Feeds {
		r := gott.Tuple{result{
			config:   cfg,
			pid:      bimbaPid,
			tmpPath:  os.TempDir(),
			feed:     feed,
			feedName: feed.String(),
			location: feed.GetLocation(),
			updates:  map[string]string{},
		}}
		_, err := gott.NewResult(r).
			SetLevelLog(gott.Debug).
			Bind(createTmpPath).
			Bind(createFeedHome).
			Bind(listDownloadedVersions).
			Bind(getAllVersions).
			Map(findValidVersions).
			Bind(getGtfsFiles).
			Bind(convert).
			Bind(signal).
			Bind(openLastUpdated).
			Tee(isEmpty).
			Bind(unmarshalLastUpdated).
			Recover(recoverEmpty).
			Map(lastUpdated).
			Tee(marshalLastUpdated).
			Finish()
		if err != nil {
			log.Printf("Error converting %s: %v\n", feed.String(), err)
		}
	}
	return nil
}