convert.go 46 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683
  1. // SPDX-FileCopyrightText: Adam Evyčędo
  2. //
  3. // SPDX-License-Identifier: AGPL-3.0-or-later
  4. package traffic
  5. // TODO(BAF10) direction (0|1) to const (TO|BACK)
  6. // TODO Agency.language, FeedInfo.language -> IETF language tag
  7. // TODO Agency.phoneNumber -> E.123 format
  8. import (
  9. "slices"
  10. "text/template"
  11. "apiote.xyz/p/szczanieckiej/config"
  12. "apiote.xyz/p/szczanieckiej/file"
  13. "bufio"
  14. "embed"
  15. "encoding/csv"
  16. "errors"
  17. "fmt"
  18. "io"
  19. "io/fs"
  20. "log"
  21. "net/http"
  22. "os"
  23. "path/filepath"
  24. "sort"
  25. "strings"
  26. "syscall"
  27. "time"
  28. "gopkg.in/yaml.v3"
  29. gott2 "apiote.xyz/p/gott/v2"
  30. "git.sr.ht/~sircmpwn/go-bare"
  31. "notabug.org/apiote/gott"
  32. )
  33. //go:embed realtime_lua
  34. var luaScripts embed.FS
  35. type _LineGraph struct {
  36. StopCodesArray []string
  37. StopCodes map[string]int
  38. NextNodes map[int]map[int]struct{}
  39. }
  40. type ErrEmpty struct{}
  41. func (ErrEmpty) Error() string {
  42. return ""
  43. }
  44. type result struct {
  45. config config.Config
  46. pid int
  47. tmpPath string
  48. feed Feed
  49. feedName string
  50. location *time.Location
  51. tmpFeedPath string
  52. homeFeedPath string
  53. downloadedVersions []Version
  54. allVersions []Version
  55. gtfsFilenames []string
  56. missingVersions []Version
  57. updatesFile *os.File
  58. updates map[string]string
  59. etags map[string]string
  60. newEtags map[string]string
  61. feedTranslations embed.FS
  62. }
  63. type feedConverter struct {
  64. TmpFeedPath string
  65. GtfsFilename string
  66. Feed Feed
  67. HomeFeedPath string
  68. feedTranslations embed.FS
  69. config config.Config
  70. Timezone *time.Location
  71. TrafficCalendarFile *os.File
  72. tripsInputIndex map[string]int64
  73. routesInputIndex map[string]int64
  74. stopsInputIndex map[string]int64
  75. tripsOffsets map[string]uint
  76. Headways map[string][]Headway
  77. StopsCodeIndex CodeIndex
  78. StopsNameIndex map[string][]uint
  79. Stops map[string]string
  80. LineGraphs map[string]map[uint]LineGraph
  81. lineHeadsigns map[string]map[uint][]string
  82. LineIndex map[string][]uint
  83. LineIdIndex CodeIndex
  84. ValidFrom time.Time
  85. ValidFromError []error
  86. ValidTill time.Time
  87. ValidTillError []error
  88. feedInfo FeedInfo
  89. defaultLanguage string
  90. translations map[string]map[string]string
  91. schedules map[string]Schedule
  92. trips map[string]Trip
  93. }
  94. // helper functions
  95. func translateFieldDefault(key, feedLanguage, defaultLanguage string, translations map[string]map[string]string) string {
  96. if feedLanguage == "mul" {
  97. if value, ok := translations[key][defaultLanguage]; !ok {
  98. return key
  99. } else {
  100. return value
  101. }
  102. }
  103. return key
  104. }
  105. func translateField(key, feedLanguage, defaultLanguage string, translations map[string]map[string]string) []Translation {
  106. var result []Translation
  107. if feedLanguage == "mul" {
  108. if value, ok := translations[key][defaultLanguage]; !ok {
  109. result = []Translation{{Language: defaultLanguage, Value: key}}
  110. } else {
  111. result = []Translation{{Language: defaultLanguage, Value: value}}
  112. }
  113. }
  114. for language, value := range translations[key] {
  115. if language == defaultLanguage {
  116. continue
  117. }
  118. result = append(result, Translation{Language: language, Value: value})
  119. }
  120. return result
  121. }
  122. func hex2colour(hex string) Colour {
  123. if hex[0] == '#' {
  124. hex = hex[1:]
  125. }
  126. colour := Colour{
  127. A: 0xff,
  128. }
  129. hexToByte := func(b byte) byte {
  130. switch {
  131. case b >= '0' && b <= '9':
  132. return b - '0'
  133. case b >= 'a' && b <= 'f':
  134. return b - 'a' + 10
  135. case b >= 'A' && b <= 'F':
  136. return b - 'A' + 10
  137. default:
  138. return 0
  139. }
  140. }
  141. switch len(hex) {
  142. case 6:
  143. colour.R = hexToByte(hex[0])<<4 + hexToByte(hex[1])
  144. colour.G = hexToByte(hex[2])<<4 + hexToByte(hex[3])
  145. colour.B = hexToByte(hex[4])<<4 + hexToByte(hex[5])
  146. case 3:
  147. colour.R = hexToByte(hex[0])<<4 + hexToByte(hex[0])
  148. colour.G = hexToByte(hex[1])<<4 + hexToByte(hex[1])
  149. colour.B = hexToByte(hex[2])<<4 + hexToByte(hex[2])
  150. }
  151. return colour
  152. }
  153. func readEtags(cfg config.Config) (map[string]string, error) {
  154. etagsFilename := filepath.Join(cfg.FeedsPath, "etags.bare")
  155. etagsFile, err := os.Open(etagsFilename)
  156. if err != nil {
  157. var pathError *os.PathError
  158. if errors.As(err, &pathError) && errors.Is(pathError, fs.ErrNotExist) {
  159. return map[string]string{}, nil
  160. }
  161. return nil, fmt.Errorf("while opening file: %w", err)
  162. }
  163. defer etagsFile.Close()
  164. var etags map[string]string
  165. err = bare.UnmarshalReader(etagsFile, &etags)
  166. if err != nil {
  167. return nil, fmt.Errorf("while unmarshalling: %w", err)
  168. }
  169. return etags, nil
  170. }
  171. func saveEtags(cfg config.Config, etags map[string]string) error {
  172. etagsFilename := filepath.Join(cfg.FeedsPath, "etags.bare")
  173. etagsFile, err := os.OpenFile(etagsFilename, os.O_RDWR|os.O_CREATE, 0644)
  174. if err != nil {
  175. return fmt.Errorf("while opening: %w", err)
  176. }
  177. defer etagsFile.Close()
  178. bytes, err := bare.Marshal(&etags)
  179. if err != nil {
  180. return fmt.Errorf("while marshalling: %w", err)
  181. }
  182. _, err = etagsFile.Write(bytes)
  183. if err != nil {
  184. return fmt.Errorf("while writing: %w", err)
  185. }
  186. return nil
  187. }
  188. // converting functions
  189. func createTmpPath(input ...interface{}) (interface{}, error) {
  190. args := input[0].(result)
  191. p := filepath.Join(args.tmpPath, args.feedName)
  192. err := os.MkdirAll(p, 0755)
  193. args.tmpFeedPath = p
  194. return gott.Tuple{args}, err
  195. }
  196. func createFeedHome(input ...interface{}) (interface{}, error) {
  197. args := input[0].(result)
  198. p := filepath.Join(args.config.FeedsPath, args.feedName)
  199. err := os.MkdirAll(p, 0755)
  200. args.homeFeedPath = p
  201. return gott.Tuple{args}, err
  202. }
  203. func listDownloadedVersions(input ...interface{}) (interface{}, error) {
  204. args := input[0].(result)
  205. v, err := ListVersionsTimezone(args.config, args.feed, args.feed.getTimezone())
  206. args.downloadedVersions = v
  207. return gott.Tuple{args}, err
  208. }
  209. func getAllVersions(input ...interface{}) (interface{}, error) {
  210. args := input[0].(result)
  211. v, err := args.feed.GetVersions(time.Now().In(args.location), args.location)
  212. args.allVersions = v
  213. return gott.Tuple{args}, err
  214. }
  215. func findValidVersions(input ...interface{}) interface{} {
  216. args := input[0].(result)
  217. now := time.Now().In(args.location)
  218. validVersions := FindValidVersions(args.allVersions, now)
  219. downloadedVersions := map[string]struct{}{}
  220. for _, downloadedVersion := range args.downloadedVersions {
  221. downloadedVersions[downloadedVersion.String()] = struct{}{}
  222. }
  223. missingVersions := []Version{}
  224. for _, version := range validVersions {
  225. if _, ok := downloadedVersions[version.String()]; !ok {
  226. missingVersions = append(missingVersions, version)
  227. }
  228. }
  229. // log.Println("all", args.allVersions)
  230. // log.Println("valid", validVersions)
  231. // log.Println("downloaded", downloadedVersions)
  232. // log.Println("missing", missingVersions)
  233. args.missingVersions = missingVersions
  234. return gott.Tuple{args}
  235. }
  236. func getGtfsFiles(input ...interface{}) (interface{}, error) {
  237. args := input[0].(result)
  238. names := []string{}
  239. for i, version := range args.missingVersions {
  240. name := fmt.Sprintf("%s_%d.zip", version.String(), i)
  241. zipPath := filepath.Join(args.tmpFeedPath, name)
  242. url := version.Link
  243. request, err := http.NewRequest("GET", url, nil)
  244. if err != nil {
  245. return gott.Tuple{args}, fmt.Errorf("while creating request for %s: %w", url, err)
  246. }
  247. request.Header.Add("If-None-Match", args.etags[url])
  248. client := http.Client{} // todo timeout
  249. response, err := client.Do(request)
  250. if err != nil {
  251. return gott.Tuple{args}, fmt.Errorf("while downloading gtfs %s %w", name, err)
  252. }
  253. if response.StatusCode != http.StatusOK && response.StatusCode != http.StatusNotModified {
  254. return gott.Tuple{args}, fmt.Errorf("wrong response code %d for %s: %w", response.StatusCode, url, err)
  255. }
  256. if response.StatusCode == 200 {
  257. args.newEtags[url] = response.Header.Get("etag")
  258. } else {
  259. args.newEtags[url] = args.etags[url]
  260. continue
  261. }
  262. file, err := os.Create(zipPath)
  263. if err != nil {
  264. return gott.Tuple{args}, fmt.Errorf("while creating zip for %s %w", name, err)
  265. }
  266. defer file.Close()
  267. _, err = io.Copy(file, response.Body)
  268. if err != nil {
  269. return gott.Tuple{args}, fmt.Errorf("while copying gtfs %s %w", name, err)
  270. }
  271. names = append(names, name)
  272. }
  273. args.gtfsFilenames = names
  274. return gott.Tuple{args}, nil
  275. }
  276. func unzipGtfs(c feedConverter) error {
  277. return file.UnzipGtfs(c.TmpFeedPath, c.GtfsFilename)
  278. }
  279. func convertVehicles(c feedConverter) error { // ( -- >> vehicles.bare)
  280. result, err := os.Create(filepath.Join(c.TmpFeedPath, "vehicles.bare"))
  281. if err != nil {
  282. return fmt.Errorf("while creating file: %w", err)
  283. }
  284. defer result.Close()
  285. vehicles, err := c.Feed.ConvertVehicles()
  286. for _, vehicle := range vehicles {
  287. bytes, err := bare.Marshal(&vehicle)
  288. if err != nil {
  289. return fmt.Errorf("while marshalling: %w", err)
  290. }
  291. _, err = result.Write(bytes)
  292. if err != nil {
  293. return fmt.Errorf("while writing to file: %w", err)
  294. }
  295. }
  296. return nil
  297. }
  298. func prepareFeedGtfs(c feedConverter) error {
  299. return c.Feed.FeedPrepareZip(c.TmpFeedPath)
  300. }
  301. func createTrafficCalendarFile(c feedConverter) (feedConverter, error) {
  302. path := c.TmpFeedPath
  303. var err error
  304. c.TrafficCalendarFile, err = os.Create(filepath.Join(path, "calendar.bare"))
  305. return c, err
  306. }
  307. func recoverCalendar(c feedConverter, e error) (feedConverter, error) {
  308. var pathError *os.PathError
  309. if errors.As(e, &pathError) && errors.Is(pathError, fs.ErrNotExist) {
  310. return c, nil
  311. }
  312. return c, e
  313. }
  314. func convertCalendar(c feedConverter) (feedConverter, error) { // ( feedInfo -- schedules >> )
  315. c.schedules = map[string]Schedule{}
  316. path := c.TmpFeedPath
  317. calendarFile, err := os.Open(filepath.Join(path, "calendar.txt"))
  318. if err != nil {
  319. return c, fmt.Errorf("while opening file: %w", err)
  320. }
  321. defer calendarFile.Close()
  322. r := csv.NewReader(bufio.NewReader(calendarFile))
  323. header, err := r.Read()
  324. if err != nil {
  325. return c, fmt.Errorf("while reading header: %w", err)
  326. }
  327. fields := map[string]int{}
  328. for i, headerField := range header {
  329. fields[headerField] = i
  330. }
  331. for {
  332. schedule := Schedule{}
  333. record, err := r.Read()
  334. if err == io.EOF {
  335. break
  336. }
  337. if err != nil {
  338. return c, fmt.Errorf("while reading a record: %w", err)
  339. }
  340. schedule.Id = record[fields["service_id"]]
  341. startDate := record[fields["start_date"]]
  342. endDate := record[fields["end_date"]]
  343. schedule.DateRanges = []DateRange{
  344. DateRange{
  345. Start: startDate[:4] + "-" + startDate[4:6] + "-" + startDate[6:],
  346. End: endDate[:4] + "-" + endDate[4:6] + "-" + endDate[6:],
  347. },
  348. }
  349. if record[fields["monday"]] == "1" {
  350. schedule.DateRanges[0].Weekdays |= (1 << 1)
  351. }
  352. if record[fields["tuesday"]] == "1" {
  353. schedule.DateRanges[0].Weekdays |= (1 << 2)
  354. }
  355. if record[fields["wednesday"]] == "1" {
  356. schedule.DateRanges[0].Weekdays |= (1 << 3)
  357. }
  358. if record[fields["thursday"]] == "1" {
  359. schedule.DateRanges[0].Weekdays |= (1 << 4)
  360. }
  361. if record[fields["friday"]] == "1" {
  362. schedule.DateRanges[0].Weekdays |= (1 << 5)
  363. }
  364. if record[fields["saturday"]] == "1" {
  365. schedule.DateRanges[0].Weekdays |= (1 << 6)
  366. }
  367. if record[fields["sunday"]] == "1" {
  368. schedule.DateRanges[0].Weekdays |= (1 << 0)
  369. schedule.DateRanges[0].Weekdays |= (1 << 7)
  370. }
  371. c.schedules[schedule.Id] = schedule
  372. scheduleStart, err := time.ParseInLocation(DateFormat, schedule.DateRanges[0].Start, c.Timezone)
  373. if err != nil {
  374. c.ValidFromError = append(c.ValidFromError, err)
  375. }
  376. if err == nil && (c.ValidFrom.IsZero() || scheduleStart.Before(c.ValidFrom)) {
  377. c.ValidFrom = scheduleStart
  378. c.feedInfo.ValidSince = scheduleStart.Format(ValidityFormat)
  379. }
  380. scheduleEnd, err := time.ParseInLocation(DateFormat, schedule.DateRanges[0].End, c.Timezone)
  381. if err != nil {
  382. c.ValidTillError = append(c.ValidTillError, err)
  383. }
  384. if err == nil && (c.ValidTill.IsZero() || scheduleEnd.After(c.ValidTill)) {
  385. c.ValidTill = scheduleEnd
  386. c.feedInfo.ValidTill = scheduleEnd.Format(ValidityFormat)
  387. }
  388. }
  389. return c, nil
  390. }
  391. func convertCalendarDates(c feedConverter) (feedConverter, error) { // ( feedInfo -- schedules >> )
  392. path := c.TmpFeedPath
  393. datesFile, err := os.Open(filepath.Join(path, "calendar_dates.txt"))
  394. if err != nil {
  395. return c, fmt.Errorf("while opening file: %w", err)
  396. }
  397. defer datesFile.Close()
  398. r := csv.NewReader(bufio.NewReader(datesFile))
  399. header, err := r.Read()
  400. if err != nil {
  401. return c, fmt.Errorf("while reading header: %w", err)
  402. }
  403. fields := map[string]int{}
  404. for i, headerField := range header {
  405. fields[headerField] = i
  406. }
  407. for {
  408. record, err := r.Read()
  409. if err == io.EOF {
  410. break
  411. }
  412. if err != nil {
  413. return c, fmt.Errorf("while reading a record: %w", err)
  414. }
  415. if record[fields["exception_type"]] == "1" {
  416. id := record[fields["service_id"]]
  417. schedule := c.schedules[id]
  418. date := record[fields["date"]]
  419. dateRange := DateRange{
  420. Start: date[:4] + "-" + date[4:6] + "-" + date[6:],
  421. End: date[:4] + "-" + date[4:6] + "-" + date[6:],
  422. Weekdays: 0xff,
  423. }
  424. if len(schedule.DateRanges) == 0 {
  425. schedule.Id = id
  426. schedule.DateRanges = []DateRange{dateRange}
  427. } else {
  428. schedule.DateRanges = append(schedule.DateRanges, dateRange)
  429. sort.Slice(schedule.DateRanges, func(i, j int) bool {
  430. return schedule.DateRanges[i].Start < schedule.DateRanges[j].Start
  431. })
  432. }
  433. c.schedules[schedule.Id] = schedule
  434. } else {
  435. date := record[fields["date"]]
  436. formatedDate := date[:4] + "-" + date[4:6] + "-" + date[6:]
  437. scheduleToEdit := c.schedules[record[fields["service_id"]]]
  438. newDateRanges := []DateRange{}
  439. for i := 0; i < len(scheduleToEdit.DateRanges); i++ {
  440. dateRange := scheduleToEdit.DateRanges[i]
  441. if dateRange.Start == formatedDate {
  442. d, _ := time.ParseInLocation(DateFormat, dateRange.Start, c.Timezone)
  443. dateRange.Start = d.AddDate(0, 0, 1).Format(DateFormat)
  444. if dateRange.Start <= dateRange.End {
  445. newDateRanges = append(newDateRanges, dateRange)
  446. }
  447. continue
  448. }
  449. if dateRange.Start < formatedDate && formatedDate < dateRange.End {
  450. d, _ := time.ParseInLocation(DateFormat, formatedDate, c.Timezone)
  451. range1 := DateRange{dateRange.Start, d.AddDate(0, 0, -1).Format(DateFormat), dateRange.Weekdays}
  452. range2 := DateRange{d.AddDate(0, 0, 1).Format(DateFormat), dateRange.End, dateRange.Weekdays}
  453. newDateRanges = append(newDateRanges, range1)
  454. newDateRanges = append(newDateRanges, range2)
  455. continue
  456. }
  457. if formatedDate == dateRange.End {
  458. d, _ := time.ParseInLocation(DateFormat, dateRange.End, c.Timezone)
  459. dateRange.End = d.AddDate(0, 0, -1).Format(DateFormat)
  460. newDateRanges = append(newDateRanges, dateRange)
  461. continue
  462. }
  463. newDateRanges = append(newDateRanges, dateRange)
  464. }
  465. scheduleToEdit.DateRanges = newDateRanges
  466. c.schedules[record[fields["service_id"]]] = scheduleToEdit
  467. }
  468. }
  469. for _, schedule := range c.schedules {
  470. lastDateRange := len(schedule.DateRanges) - 1
  471. scheduleStart, err := time.ParseInLocation(DateFormat, schedule.DateRanges[0].Start, c.Timezone)
  472. if err != nil {
  473. c.ValidFromError = append(c.ValidFromError, err)
  474. }
  475. if err == nil && (c.ValidFrom.IsZero() || scheduleStart.Before(c.ValidFrom)) {
  476. c.ValidFrom = scheduleStart
  477. c.feedInfo.ValidSince = scheduleStart.Format(ValidityFormat)
  478. }
  479. scheduleEnd, err := time.ParseInLocation(DateFormat, schedule.DateRanges[lastDateRange].End, c.Timezone)
  480. if err != nil {
  481. c.ValidTillError = append(c.ValidTillError, err)
  482. }
  483. if err == nil && (c.ValidTill.IsZero() || scheduleEnd.After(c.ValidTill)) {
  484. c.ValidTill = scheduleEnd
  485. c.feedInfo.ValidTill = scheduleEnd.Format(ValidityFormat)
  486. }
  487. }
  488. return c, nil
  489. }
  490. func checkAnyCalendarConverted(c feedConverter) error {
  491. if len(c.schedules) == 0 {
  492. return fmt.Errorf("no calendar converted")
  493. }
  494. return nil
  495. }
  496. func saveSchedules(c feedConverter) error {
  497. resultFile := c.TrafficCalendarFile
  498. schedulesArray := make([]Schedule, len(c.schedules))
  499. i := 0
  500. for _, schedule := range c.schedules {
  501. schedulesArray[i] = schedule
  502. i++
  503. }
  504. sort.Slice(schedulesArray, func(i, j int) bool {
  505. return schedulesArray[i].DateRanges[0].Start < schedulesArray[j].DateRanges[0].Start
  506. })
  507. for _, schedule := range schedulesArray {
  508. bytes, err := bare.Marshal(&schedule)
  509. if err != nil {
  510. return fmt.Errorf("while marshalling: %w", err)
  511. }
  512. _, err = resultFile.Write(bytes)
  513. if err != nil {
  514. return fmt.Errorf("while writing: %w", err)
  515. }
  516. }
  517. c.schedules = map[string]Schedule{}
  518. return nil
  519. }
  520. func saveFeedInfo(c feedConverter) error {
  521. path := c.TmpFeedPath
  522. result, err := os.Create(filepath.Join(path, "feed_info.bare"))
  523. if err != nil {
  524. return fmt.Errorf("while creating file: %w", err)
  525. }
  526. defer result.Close()
  527. bytes, err := bare.Marshal(&c.feedInfo)
  528. if err != nil {
  529. return fmt.Errorf("while marshalling: %w", err)
  530. }
  531. _, err = result.Write(bytes)
  532. if err != nil {
  533. return fmt.Errorf("while writing: %w", err)
  534. }
  535. log.Printf("timetable is valid: %s to %s\n", c.feedInfo.ValidSince, c.feedInfo.ValidTill)
  536. c.feedInfo = FeedInfo{}
  537. return nil
  538. }
  539. func closeTrafficCalendarFile(c feedConverter, e error) (feedConverter, error) {
  540. if c.TrafficCalendarFile != nil {
  541. c.TrafficCalendarFile.Close()
  542. }
  543. return c, e
  544. }
  545. func interpolateDepartures(c feedConverter) error { // O(n:stop_times) ; ( -- >> stop_times.txt)
  546. path := c.TmpFeedPath
  547. stopTimesFile, err := os.Open(filepath.Join(path, "stop_times.txt"))
  548. if err != nil {
  549. return fmt.Errorf("while opening stop_times file: %w", err)
  550. }
  551. defer stopTimesFile.Close()
  552. stopTimes2File, err := os.OpenFile(filepath.Join(path, "stop_times2.txt"), os.O_RDWR|os.O_CREATE, 0644)
  553. if err != nil {
  554. return fmt.Errorf("while opening stop_times2 file: %w", err)
  555. }
  556. defer stopTimes2File.Close()
  557. r := csv.NewReader(bufio.NewReader(stopTimesFile))
  558. w := csv.NewWriter(stopTimes2File)
  559. header, err := r.Read()
  560. if err != nil {
  561. return fmt.Errorf("while reading stop_times header: %w", err)
  562. }
  563. fields := map[string]int{}
  564. for i, headerField := range header {
  565. fields[headerField] = i
  566. }
  567. err = w.Write(header)
  568. if err != nil {
  569. return fmt.Errorf("while writing stop_times header: %w", err)
  570. }
  571. lastTripID := ""
  572. var lastKnownTime uint = 0
  573. collectedRecords := [][]string{}
  574. for {
  575. record, err := r.Read()
  576. if err == io.EOF {
  577. break
  578. }
  579. if err != nil {
  580. return fmt.Errorf("while reading a stop_times record: %w", err)
  581. }
  582. tripID := record[fields["trip_id"]]
  583. arrivalTime := record[fields["arrival_time"]]
  584. if tripID != lastTripID && arrivalTime == "" {
  585. return fmt.Errorf("no arrival time at the beginning of the trip ‘%s’", tripID)
  586. }
  587. if ix, ok := fields["timepoint"]; !ok || record[ix] == "1" || arrivalTime != "" {
  588. dT, err := parseDepartureTime(arrivalTime)
  589. if err != nil {
  590. return fmt.Errorf("incorrect end departure time %s: %w", arrivalTime, err)
  591. }
  592. if len(collectedRecords) > 0 {
  593. if tripID != lastTripID && lastTripID != "" {
  594. return fmt.Errorf("no arrival time at the end of the trip ‘%s’", lastTripID)
  595. }
  596. singleDuration := (dT - lastKnownTime) / uint(len(collectedRecords)+1)
  597. for i, collectedRecord := range collectedRecords {
  598. collectedRecord[fields["arrival_time"]] = formatDepartureTime(lastKnownTime + (uint(i+1) * singleDuration))
  599. err = w.Write(collectedRecord)
  600. if err != nil {
  601. return fmt.Errorf("while writing a stop_times collected record %d: %w", i, err)
  602. }
  603. }
  604. }
  605. err = w.Write(record)
  606. if err != nil {
  607. return fmt.Errorf("while writing a stop_times record: %w", err)
  608. }
  609. collectedRecords = [][]string{}
  610. lastKnownTime = dT
  611. } else {
  612. collectedRecords = append(collectedRecords, record)
  613. }
  614. lastTripID = tripID
  615. }
  616. w.Flush()
  617. err = os.Remove(filepath.Join(path, "stop_times.txt"))
  618. if err != nil {
  619. return fmt.Errorf("while removing stop_times: %w", err)
  620. }
  621. err = os.Rename(filepath.Join(path, "stop_times2.txt"), filepath.Join(path, "stop_times.txt"))
  622. if err != nil {
  623. return fmt.Errorf("while renaming stop_times: %w", err)
  624. }
  625. return nil
  626. }
  627. func readFrequencies(c feedConverter) (feedConverter, error) { // O(n:frequencies) ; ( -- headways:map[tripID][]headways >> )
  628. path := c.TmpFeedPath
  629. file, err := os.Open(filepath.Join(path, "frequencies.txt"))
  630. if err != nil {
  631. if errors.Is(err, fs.ErrNotExist) {
  632. log.Println("[WARN] no frequencies.txt")
  633. return c, nil
  634. } else {
  635. return c, fmt.Errorf("while opening file: %w", err)
  636. }
  637. }
  638. defer file.Close()
  639. headways := map[string][]Headway{}
  640. r := csv.NewReader(bufio.NewReader(file))
  641. header, err := r.Read()
  642. if err != nil {
  643. return c, fmt.Errorf("while reading header: %w", err)
  644. }
  645. fields := map[string]int{}
  646. for i, headerField := range header {
  647. fields[headerField] = i
  648. }
  649. for {
  650. headway := Headway{}
  651. record, err := r.Read()
  652. if err == io.EOF {
  653. break
  654. }
  655. if err != nil {
  656. return c, fmt.Errorf("while reading a record: %w", err)
  657. }
  658. tripID := record[fields["trip_id"]]
  659. startTime, err := parseDepartureTime(record[fields["start_time"]])
  660. if err != nil {
  661. return c, fmt.Errorf("while parsing start_time: %w", err)
  662. }
  663. endTime, err := parseDepartureTime(record[fields["end_time"]])
  664. if err != nil {
  665. return c, fmt.Errorf("while parsing start_time: %w", err)
  666. }
  667. headway.StartTime = uint(startTime)
  668. headway.EndTime = uint(endTime)
  669. fmt.Sscanf(record[fields["headway_secs"]], "%d", &headway.Interval)
  670. if ix, ok := fields["exact_times"]; ok && record[ix] == "1" {
  671. headway.Exact = true
  672. }
  673. headways[tripID] = append(headways[tripID], headway)
  674. }
  675. sortedHeadways := map[string][]Headway{}
  676. for tripID, h := range headways {
  677. slices.SortFunc(h, func(a, b Headway) int {
  678. if a.StartTime < b.StartTime {
  679. return -1
  680. } else if a.StartTime > b.StartTime {
  681. return 1
  682. } else {
  683. return 0
  684. }
  685. })
  686. sortedHeadways[tripID] = h
  687. }
  688. c.Headways = sortedHeadways
  689. return c, nil
  690. }
  691. func forEachRow(filename string, f func(int64, map[string]int, []string) error) error {
  692. return forEachRowWithHeader(filename, func(_ []string) error {
  693. return nil
  694. }, f)
  695. }
  696. func forEachRowWithHeader(filename string, h func([]string) error, f func(int64, map[string]int, []string) error) error {
  697. file, err := os.Open(filename)
  698. if err != nil {
  699. return fmt.Errorf("while opening file: %w", err)
  700. }
  701. defer file.Close()
  702. r := csv.NewReader(file)
  703. header, err := r.Read()
  704. if err != nil {
  705. return fmt.Errorf("while reading header: %w", err)
  706. }
  707. err = h(header)
  708. if err != nil {
  709. return fmt.Errorf("while performing function on header: %w", err)
  710. }
  711. fields := map[string]int{}
  712. for i, headerField := range header {
  713. fields[headerField] = i
  714. }
  715. for {
  716. offset := r.InputOffset()
  717. record, err := r.Read()
  718. if err == io.EOF {
  719. break
  720. }
  721. if err != nil {
  722. return fmt.Errorf("while reading a record: %w", err)
  723. }
  724. err = f(offset, fields, record)
  725. if err != nil {
  726. return fmt.Errorf("while performing function: %w", err)
  727. }
  728. }
  729. return nil
  730. }
  731. func clearStops(c feedConverter) feedConverter {
  732. c.Stops = map[string]string{}
  733. return c
  734. }
  735. func clearTripOffsets(c feedConverter) feedConverter {
  736. c.tripsOffsets = map[string]uint{}
  737. return c
  738. }
  739. func clearLineGraphs(c feedConverter) feedConverter {
  740. c.LineGraphs = map[string]map[uint]LineGraph{}
  741. return c
  742. }
  743. func clearLineHeadsigns(c feedConverter) feedConverter {
  744. c.lineHeadsigns = map[string]map[uint][]string{}
  745. return c
  746. }
  747. func getTrips(c feedConverter) (feedConverter, error) {
  748. file, err := os.Open(filepath.Join(c.TmpFeedPath, "trips.bare"))
  749. if err != nil {
  750. return c, fmt.Errorf("while opening trips: %w", err)
  751. }
  752. trips := map[string]Trip{}
  753. for {
  754. var trip Trip
  755. err := bare.UnmarshalReader(file, &trip)
  756. trip.Departures = []Departure{}
  757. trips[trip.Id] = trip
  758. if err != nil {
  759. if err == io.EOF {
  760. break
  761. } else {
  762. return c, fmt.Errorf("while unmarshaling: %w", err)
  763. }
  764. }
  765. }
  766. c.trips = trips
  767. return c, nil
  768. }
  769. func convertLineGraphs(c feedConverter) (feedConverter, error) { // O(n:stop_times) ; (trips, stops -- lineGrapsh:map[lineID]map[direction]graph, lineHeadsigns:map[lineID]map[direction][]headsigns >> )
  770. path := c.TmpFeedPath
  771. trips := c.trips
  772. stops := c.Stops
  773. // lineID dire headsi
  774. lineHeadsignsMap := map[string]map[uint]map[string]struct{}{}
  775. // lineID dire headsi
  776. lineHeadsigns := map[string]map[uint][]string{}
  777. // lineNa dire
  778. graphs := map[string]map[uint]_LineGraph{}
  779. file, err := os.Open(filepath.Join(path, "stop_times.txt"))
  780. if err != nil {
  781. return c, fmt.Errorf("while opening stop_times: %w", err)
  782. }
  783. defer file.Close()
  784. r := csv.NewReader(bufio.NewReader(file))
  785. header, err := r.Read()
  786. if err != nil {
  787. return c, fmt.Errorf("while reading header: %w", err)
  788. }
  789. fields := map[string]int{}
  790. for i, headerField := range header {
  791. fields[headerField] = i
  792. }
  793. previousTripID := ""
  794. previous := -1
  795. previousTrip := Trip{}
  796. for {
  797. record, err := r.Read()
  798. if err == io.EOF {
  799. break
  800. }
  801. if err != nil {
  802. return c, fmt.Errorf("while reading a record: %w", err)
  803. }
  804. tripID := record[fields["trip_id"]]
  805. stop := stops[record[fields["stop_id"]]]
  806. trip := trips[tripID]
  807. if _, ok := lineHeadsignsMap[trip.LineID]; !ok {
  808. lineHeadsignsMap[trip.LineID] = map[uint]map[string]struct{}{}
  809. lineHeadsigns[trip.LineID] = map[uint][]string{}
  810. }
  811. if _, ok := lineHeadsignsMap[trip.LineID][trip.Direction.Value()]; !ok {
  812. lineHeadsignsMap[trip.LineID][trip.Direction.Value()] = map[string]struct{}{}
  813. lineHeadsigns[trip.LineID][trip.Direction.Value()] = []string{}
  814. }
  815. lineHeadsignsMap[trip.LineID][trip.Direction.Value()][trip.Headsign] = struct{}{}
  816. if _, ok := graphs[trip.LineID]; !ok {
  817. graphs[trip.LineID] = map[uint]_LineGraph{}
  818. }
  819. if previousTripID != tripID && previousTripID != "" {
  820. // last of previous trip
  821. graph := graphs[previousTrip.LineID][previousTrip.Direction.Value()]
  822. if graph.NextNodes == nil {
  823. graph.NextNodes = map[int]map[int]struct{}{}
  824. }
  825. if graph.NextNodes[previous] == nil {
  826. graph.NextNodes[previous] = map[int]struct{}{}
  827. }
  828. graphs[previousTrip.LineID][previousTrip.Direction.Value()] = graph
  829. graphs[previousTrip.LineID][previousTrip.Direction.Value()].NextNodes[previous][-1] = struct{}{}
  830. }
  831. graph := graphs[trip.LineID][trip.Direction.Value()]
  832. if graph.StopCodes == nil {
  833. graph.StopCodes = map[string]int{}
  834. }
  835. if graph.NextNodes == nil {
  836. graph.NextNodes = map[int]map[int]struct{}{}
  837. }
  838. current := -1
  839. current, ok := graph.StopCodes[stop]
  840. if !ok {
  841. current = len(graph.StopCodesArray)
  842. graph.StopCodesArray = append(graph.StopCodesArray, stop)
  843. graph.StopCodes[stop] = current
  844. }
  845. if previousTripID != tripID {
  846. // first of current trip
  847. if graph.NextNodes[-1] == nil {
  848. graph.NextNodes[-1] = map[int]struct{}{}
  849. }
  850. if _, ok := graph.NextNodes[-1][current]; !ok {
  851. graph.NextNodes[-1][current] = struct{}{}
  852. }
  853. } else {
  854. // second <- first to last <- penultimate of current trip
  855. if graph.NextNodes[previous] == nil {
  856. graph.NextNodes[previous] = map[int]struct{}{}
  857. }
  858. if _, ok := graph.NextNodes[previous][current]; !ok {
  859. graph.NextNodes[previous][current] = struct{}{}
  860. }
  861. }
  862. previous = current
  863. previousTripID = tripID
  864. previousTrip = trip
  865. graphs[trip.LineID][trip.Direction.Value()] = graph
  866. }
  867. g := graphs[previousTrip.LineID][previousTrip.Direction.Value()]
  868. if g.NextNodes[previous] == nil {
  869. g.NextNodes[previous] = map[int]struct{}{}
  870. }
  871. if _, ok := g.NextNodes[previous][-1]; !ok {
  872. g.NextNodes[previous][-1] = struct{}{}
  873. }
  874. for lineID, directions := range lineHeadsignsMap {
  875. for direction, headsigns := range directions {
  876. for headsign := range headsigns {
  877. lineHeadsigns[lineID][direction] = append(lineHeadsigns[lineID][direction], headsign)
  878. }
  879. }
  880. }
  881. c.LineGraphs = map[string]map[uint]LineGraph{}
  882. for lineID, graphByDirection := range graphs {
  883. c.LineGraphs[lineID] = map[uint]LineGraph{}
  884. for direction, graph := range graphByDirection {
  885. c.LineGraphs[lineID][direction] = LineGraph{
  886. StopCodes: graph.StopCodesArray,
  887. NextNodes: map[int][]int{},
  888. }
  889. for from, tos := range graph.NextNodes {
  890. for to := range tos {
  891. c.LineGraphs[lineID][direction].NextNodes[from] = append(c.LineGraphs[lineID][direction].NextNodes[from], to)
  892. }
  893. }
  894. }
  895. }
  896. c.lineHeadsigns = lineHeadsigns
  897. return c, nil
  898. }
  899. func convertLines(c feedConverter) (feedConverter, error) { // O(n:routes) ; (lineGraphs, lineHeadsigns -- lineIndex:map[lineName][]offsets, lineIdIndex:CodeIndex >> lines)
  900. path := c.TmpFeedPath
  901. feed := c.Feed
  902. file, err := os.Open(filepath.Join(path, "routes.txt"))
  903. if err != nil {
  904. return c, fmt.Errorf("while opening file: %w", err)
  905. }
  906. defer file.Close()
  907. result, err := os.Create(filepath.Join(path, "lines.bare"))
  908. if err != nil {
  909. return c, fmt.Errorf("while creating file: %w", err)
  910. }
  911. defer result.Close()
  912. r := csv.NewReader(bufio.NewReader(file))
  913. header, err := r.Read()
  914. if err != nil {
  915. return c, fmt.Errorf("while reading header: %w", err)
  916. }
  917. fields := map[string]int{}
  918. for i, headerField := range header {
  919. fields[headerField] = i
  920. }
  921. var offset uint = 0
  922. index := map[string][]uint{}
  923. idIndex := CodeIndex{}
  924. for {
  925. record, err := r.Read()
  926. if err == io.EOF {
  927. break
  928. }
  929. if err != nil {
  930. return c, fmt.Errorf("while reading a record: %w", err)
  931. }
  932. routeID := record[fields["route_id"]]
  933. lineName := c.Feed.Flags().LineName
  934. for _, template := range []string{"route_short_name", "route_long_name"} {
  935. lineName = strings.Replace(lineName, "{{"+template+"}}", record[fields[template]], -1)
  936. }
  937. var kind uint
  938. fmt.Sscanf(record[fields["route_type"]], "%d", &kind)
  939. colour := "ffffff"
  940. if colourIx, ok := fields["route_color"]; ok && record[colourIx] != "" {
  941. colour = record[colourIx]
  942. }
  943. directions := []uint{}
  944. for direction := range c.lineHeadsigns[routeID] {
  945. directions = append(directions, direction)
  946. }
  947. sort.Slice(directions, func(i, j int) bool {
  948. return directions[i] < directions[j]
  949. })
  950. headsigns := [][]string{}
  951. translatedHeadsigns := [][][]Translation{}
  952. for _, direction := range directions {
  953. dirHeadsigns := c.lineHeadsigns[routeID][direction]
  954. headsigns = append(headsigns, dirHeadsigns)
  955. translatedHeadsign := [][]Translation{}
  956. for _, headsign := range dirHeadsigns {
  957. translatedHeadsign = append(translatedHeadsign, translateField(headsign, c.feedInfo.Language, c.defaultLanguage, c.translations))
  958. }
  959. translatedHeadsigns = append(translatedHeadsigns, translatedHeadsign)
  960. }
  961. graphs := []LineGraph{}
  962. for _, direction := range directions {
  963. graphs = append(graphs, c.LineGraphs[routeID][direction])
  964. }
  965. line := Line{
  966. Id: routeID,
  967. Name: lineName,
  968. Colour: hex2colour(colour),
  969. Kind: LineType(kind),
  970. Graphs: graphs,
  971. Headsigns: headsigns,
  972. }
  973. if field, present := fields["agency_id"]; present {
  974. line.AgencyID = record[field]
  975. }
  976. bytes, err := bare.Marshal(&line)
  977. if err != nil {
  978. return c, fmt.Errorf("while marshalling: %w", err)
  979. }
  980. b, err := result.Write(bytes)
  981. if err != nil {
  982. return c, fmt.Errorf("while writing: %w", err)
  983. }
  984. cleanQuery, err := CleanQuery(line.Name, feed)
  985. if err != nil {
  986. return c, fmt.Errorf("while cleaning line name: %w", err)
  987. }
  988. index[cleanQuery] = append(index[cleanQuery], offset)
  989. idIndex[routeID] = offset
  990. offset += uint(b)
  991. }
  992. c.LineIdIndex = idIndex
  993. c.LineIndex = index
  994. return c, nil
  995. }
  996. func convertFeedInfo(c feedConverter) (feedConverter, error) { // O(1:feed_info) ; ( -- feed_info >> )
  997. path := c.TmpFeedPath
  998. feedInfo := FeedInfo{}
  999. file, err := os.Open(filepath.Join(path, "feed_info.txt"))
  1000. if err != nil {
  1001. if errors.Is(err, fs.ErrNotExist) {
  1002. log.Println("[WARN] no feed_info.txt")
  1003. file = nil
  1004. } else {
  1005. return c, fmt.Errorf("while opening file: %w", err)
  1006. }
  1007. }
  1008. if file != nil {
  1009. defer file.Close()
  1010. r := csv.NewReader(bufio.NewReader(file))
  1011. header, err := r.Read()
  1012. if err != nil {
  1013. return c, fmt.Errorf("while reading header: %w", err)
  1014. }
  1015. fields := map[string]int{}
  1016. for i, headerField := range header {
  1017. fields[headerField] = i
  1018. }
  1019. record, err := r.Read()
  1020. if err != nil {
  1021. return c, fmt.Errorf("while reading a record: %w", err)
  1022. }
  1023. feedInfo.Website = record[fields["feed_publisher_url"]]
  1024. feedInfo.Language = record[fields["feed_lang"]]
  1025. if defaultLanguageIndex, ok := fields["default_lang"]; ok {
  1026. c.defaultLanguage = record[defaultLanguageIndex]
  1027. }
  1028. if ix, ok := fields["feed_start_date"]; ok {
  1029. c.ValidFrom, err = time.ParseInLocation("20060102", record[ix], c.Timezone)
  1030. if err != nil {
  1031. c.ValidFromError = append(c.ValidFromError, err)
  1032. }
  1033. feedInfo.ValidSince = record[ix]
  1034. }
  1035. if ix, ok := fields["feed_end_date"]; ok {
  1036. c.ValidTill, err = time.ParseInLocation("20060102", record[ix], c.Timezone)
  1037. if err != nil {
  1038. c.ValidTillError = append(c.ValidTillError, err)
  1039. }
  1040. feedInfo.ValidTill = record[ix]
  1041. }
  1042. }
  1043. feedInfo.Timezone = c.Timezone.String()
  1044. feedInfo.RealtimeFeeds = c.Feed.RealtimeFeeds()
  1045. feedInfo.QrHost, feedInfo.QrLocation, feedInfo.QrSelector = c.Feed.QRInfo()
  1046. feedInfo.Attributions, feedInfo.Descriptions, err = getAttrDesc(c.Feed.String(), c.feedTranslations)
  1047. feedInfo.Name = c.Feed.Name()
  1048. c.feedInfo = feedInfo
  1049. return c, err
  1050. }
  1051. func convertLuaScripts(c feedConverter) error { // O(1) ; ( -- >> updates.lua, alerts.lua, vehicles.lua )
  1052. filenames := []string{"updates", "vehicles", "alerts"}
  1053. for _, filename := range filenames {
  1054. t, err := template.ParseFS(luaScripts, "realtime_lua/"+c.Feed.String()+"_"+filename+".lua")
  1055. if err != nil {
  1056. if strings.Contains(err.Error(), "pattern matches no files") {
  1057. log.Printf("%s.lua for this feed does not exist, ignoring\n", filename)
  1058. continue
  1059. }
  1060. return fmt.Errorf("while parsing template %s: %w", filename, err)
  1061. }
  1062. path := c.TmpFeedPath
  1063. writeFile, err := os.Create(filepath.Join(path, filename+".lua"))
  1064. if err != nil {
  1065. return fmt.Errorf("while creating %s: %w", filename, err)
  1066. }
  1067. defer writeFile.Close()
  1068. err = t.Execute(writeFile, c.config.Auth[c.Feed.String()])
  1069. if err != nil {
  1070. return fmt.Errorf("while executing template %s: %w", filename, err)
  1071. }
  1072. }
  1073. return nil
  1074. }
  1075. func getAttrDesc(feedID string, feedTranslations embed.FS) (map[string]string, map[string]string, error) {
  1076. attributions := map[string]string{}
  1077. descriptions := map[string]string{}
  1078. dir, err := feedTranslations.ReadDir("translations")
  1079. if err != nil {
  1080. return attributions, descriptions, err
  1081. }
  1082. for _, f := range dir {
  1083. translation := map[string]string{}
  1084. name := f.Name()
  1085. lang := strings.Split(name, ".")[1]
  1086. fileContent, err := feedTranslations.ReadFile("translations/" + name)
  1087. if err != nil {
  1088. log.Printf("error reading translation %s\n", name)
  1089. continue
  1090. }
  1091. yaml.Unmarshal(fileContent, &translation)
  1092. attributions[lang] = translation[feedID+"_attribution"]
  1093. descriptions[lang] = translation[feedID+"_description"]
  1094. if lang == "en" {
  1095. attributions["und"] = translation[feedID+"_attribution"]
  1096. descriptions["und"] = translation[feedID+"_description"]
  1097. }
  1098. }
  1099. return attributions, descriptions, nil
  1100. }
  1101. func readTranslations(c feedConverter) (feedConverter, error) { // O(n:translations) ; ( -- translations >>)
  1102. path := c.TmpFeedPath
  1103. file, err := os.Open(filepath.Join(path, "translations.txt"))
  1104. if err != nil {
  1105. return c, fmt.Errorf("while opening file: %w", err)
  1106. }
  1107. defer file.Close()
  1108. translations := map[string]map[string]string{}
  1109. r := csv.NewReader(bufio.NewReader(file))
  1110. header, err := r.Read()
  1111. if err != nil {
  1112. return c, fmt.Errorf("while reading header: %w", err)
  1113. }
  1114. fields := map[string]int{}
  1115. for i, headerField := range header {
  1116. fields[headerField] = i
  1117. }
  1118. for {
  1119. record, err := r.Read()
  1120. if err == io.EOF {
  1121. break
  1122. }
  1123. if err != nil {
  1124. return c, fmt.Errorf("while reading a record: %w", err)
  1125. }
  1126. key := record[fields["field_value"]]
  1127. language := record[fields["language"]]
  1128. translation := record[fields["translation"]]
  1129. if _, ok := translations[key]; !ok {
  1130. translations[key] = map[string]string{}
  1131. }
  1132. translations[key][language] = translation
  1133. }
  1134. c.translations = translations
  1135. return c, nil
  1136. }
  1137. func recoverTranslations(c feedConverter, e error) (feedConverter, error) {
  1138. var pathError *os.PathError
  1139. if errors.As(e, &pathError) && errors.Is(pathError, fs.ErrNotExist) {
  1140. return c, nil
  1141. }
  1142. return c, e
  1143. }
  1144. func convertAgencies(c feedConverter) (feedConverter, error) { // O(n:agency) ; ( -- >> agencies)
  1145. path := c.TmpFeedPath
  1146. file, err := os.Open(filepath.Join(path, "agency.txt"))
  1147. if err != nil {
  1148. return c, fmt.Errorf("while opening file: %w", err)
  1149. }
  1150. defer file.Close()
  1151. result, err := os.Create(filepath.Join(path, "agencies.bare"))
  1152. if err != nil {
  1153. return c, fmt.Errorf("while creating file: %w", err)
  1154. }
  1155. defer file.Close()
  1156. r := csv.NewReader(bufio.NewReader(file))
  1157. header, err := r.Read()
  1158. if err != nil {
  1159. return c, fmt.Errorf("while reading header: %w", err)
  1160. }
  1161. fields := map[string]int{}
  1162. for i, headerField := range header {
  1163. fields[headerField] = i
  1164. }
  1165. for {
  1166. record, err := r.Read()
  1167. if err == io.EOF {
  1168. break
  1169. }
  1170. if err != nil {
  1171. return c, fmt.Errorf("while reading a record: %w", err)
  1172. }
  1173. agency := Agency{
  1174. Id: record[fields["agency_id"]],
  1175. Name: record[fields["agency_name"]],
  1176. TranslatedNames: translateField(record[fields["agency_name"]], c.feedInfo.Language, c.defaultLanguage, c.translations),
  1177. Website: record[fields["agency_url"]],
  1178. TranslatedWebsites: translateField(record[fields["agency_url"]], c.feedInfo.Language, c.defaultLanguage, c.translations),
  1179. Timezone: record[fields["agency_timezone"]],
  1180. }
  1181. c.Timezone, _ = time.LoadLocation(agency.Timezone)
  1182. if field, present := fields["agency_lang"]; present {
  1183. agency.Language = record[field]
  1184. }
  1185. if field, present := fields["agency_phone"]; present {
  1186. agency.PhoneNumber = record[field]
  1187. agency.TranslatedPhoneNumbers = translateField(record[field], c.feedInfo.Language, c.defaultLanguage, c.translations)
  1188. }
  1189. if field, present := fields["agency_fare_url"]; present {
  1190. agency.FareWebsite = record[field]
  1191. agency.TranslatedFareWebsites = translateField(record[field], c.feedInfo.Language, c.defaultLanguage, c.translations)
  1192. }
  1193. if field, present := fields["agency_email"]; present {
  1194. agency.Email = record[field]
  1195. agency.TranslatedEmails = translateField(record[field], c.feedInfo.Language, c.defaultLanguage, c.translations)
  1196. }
  1197. bytes, err := bare.Marshal(&agency)
  1198. if err != nil {
  1199. return c, fmt.Errorf("while marshalling: %w", err)
  1200. }
  1201. _, err = result.Write(bytes)
  1202. if err != nil {
  1203. return c, fmt.Errorf("while writing: %w", err)
  1204. }
  1205. }
  1206. return c, nil
  1207. }
  1208. func writeNameIndex(c feedConverter, index map[string][]uint, filename string, raw bool) error {
  1209. path := c.TmpFeedPath
  1210. feed := c.Feed
  1211. result, err := os.Create(filepath.Join(path, filename))
  1212. if err != nil {
  1213. return fmt.Errorf("while creating file: %w", err)
  1214. }
  1215. defer result.Close()
  1216. for name, offsets := range index {
  1217. cleanQuery := name
  1218. if !raw {
  1219. cleanQuery, err = CleanQuery(name, feed)
  1220. if err != nil {
  1221. return fmt.Errorf("while cleaning name %s: %w", name, err)
  1222. }
  1223. }
  1224. stopOffset := NameOffset{
  1225. Name: cleanQuery,
  1226. Offsets: offsets,
  1227. }
  1228. bytes, err := bare.Marshal(&stopOffset)
  1229. if err != nil {
  1230. return fmt.Errorf("while marshalling: %w", err)
  1231. }
  1232. _, err = result.Write(bytes)
  1233. if err != nil {
  1234. return fmt.Errorf("while writing: %w", err)
  1235. }
  1236. }
  1237. return nil
  1238. }
  1239. func writeStopNameIndex(c feedConverter) error {
  1240. err := writeNameIndex(c, c.StopsNameIndex, "ix_stop_names.bare", false)
  1241. c.StopsNameIndex = map[string][]uint{}
  1242. return err
  1243. }
  1244. func writeLineIndex(c feedConverter) error {
  1245. err := writeNameIndex(c, c.LineIndex, "ix_lines.bare", false)
  1246. c.LineIndex = map[string][]uint{}
  1247. return err
  1248. }
  1249. func writeLineIdIndex(c feedConverter) error {
  1250. err := writeCodeIndex(c, c.LineIdIndex, "ix_line_codes.bare")
  1251. c.LineIndex = map[string][]uint{}
  1252. return err
  1253. }
  1254. func writeTripIndex(c feedConverter) error {
  1255. tripIndex := map[string][]uint{}
  1256. for trip, offset := range c.tripsOffsets {
  1257. tripIndex[trip] = []uint{offset}
  1258. }
  1259. err := writeNameIndex(c, tripIndex, "ix_trips.bare", true)
  1260. c.tripsOffsets = map[string]uint{}
  1261. return err
  1262. }
  1263. func writeStopCodeIndex(c feedConverter) error {
  1264. err := writeCodeIndex(c, c.StopsCodeIndex, "ix_stop_codes.bare")
  1265. c.StopsCodeIndex = CodeIndex{}
  1266. return err
  1267. }
  1268. func writeCodeIndex(c feedConverter, i CodeIndex, filename string) error {
  1269. path := c.TmpFeedPath
  1270. result, err := os.Create(filepath.Join(path, filename))
  1271. if err != nil {
  1272. return fmt.Errorf("while creating file: %w", err)
  1273. }
  1274. defer result.Close()
  1275. bytes, err := bare.Marshal(&i)
  1276. if err != nil {
  1277. return fmt.Errorf("while marshalling: %w", err)
  1278. }
  1279. _, err = result.Write(bytes)
  1280. if err != nil {
  1281. return fmt.Errorf("while writing: %w", err)
  1282. }
  1283. return nil
  1284. }
  1285. func deleteTxtFiles(c feedConverter) error {
  1286. return file.DeleteTxtFiles(c.TmpFeedPath, c.GtfsFilename)
  1287. }
  1288. func compressTraffic(c feedConverter) error {
  1289. return file.CompressBare(c.TmpFeedPath, c.GtfsFilename)
  1290. }
  1291. func deleteBareFiles(c feedConverter) error {
  1292. return file.DeleteBareFiles(c.TmpFeedPath)
  1293. }
  1294. func moveTraffic(c feedConverter) error {
  1295. if err := append(c.ValidFromError, c.ValidTillError...); len(err) != 0 {
  1296. return errors.Join(err...)
  1297. }
  1298. return file.MoveTraffic(c.GtfsFilename, c.ValidFrom.Format("20060102")+"_"+c.ValidTill.Format("20060102")+".txz", c.TmpFeedPath, c.HomeFeedPath)
  1299. }
  1300. func convert(input ...interface{}) (interface{}, error) {
  1301. allErrors := []error{}
  1302. args := input[0].(result)
  1303. for _, gtfsFile := range args.gtfsFilenames {
  1304. log.Printf("converting feed %s/%s\n", args.feed.Name(), gtfsFile)
  1305. r := gott2.R[feedConverter]{
  1306. S: feedConverter{
  1307. TmpFeedPath: args.tmpFeedPath,
  1308. GtfsFilename: gtfsFile,
  1309. Feed: args.feed,
  1310. HomeFeedPath: args.homeFeedPath,
  1311. feedTranslations: args.feedTranslations,
  1312. config: args.config,
  1313. },
  1314. LogLevel: gott2.Debug,
  1315. }
  1316. r = r.
  1317. Tee(unzipGtfs).
  1318. Tee(prepareFeedGtfs).
  1319. Tee(convertVehicles).
  1320. Bind(convertAgencies).
  1321. Bind(convertFeedInfo).
  1322. Tee(convertLuaScripts).
  1323. Bind(readTranslations).
  1324. Recover(recoverTranslations).
  1325. Bind(createTrafficCalendarFile).
  1326. Bind(convertCalendar).
  1327. Recover(recoverCalendar).
  1328. Bind(convertCalendarDates).
  1329. Recover(recoverCalendar).
  1330. Tee(checkAnyCalendarConverted).
  1331. Tee(saveSchedules).
  1332. Tee(saveFeedInfo).
  1333. Recover(closeTrafficCalendarFile).
  1334. Tee(interpolateDepartures).
  1335. // ---
  1336. Bind(readInputTripsIndex).
  1337. Bind(readInputStopsIndex).
  1338. Bind(readInputRoutesIndex).
  1339. Bind(convertDepartures).
  1340. Map(dropInputRoutesIndex).
  1341. Map(dropInputStopsIndex).
  1342. Map(dropInputTripsIndex).
  1343. // ---
  1344. Tee(sortTripsThroughStop).
  1345. Bind(readTripsThroughStopsIndex).
  1346. Bind(convertStops).
  1347. Map(dropInputRoutesIndex).
  1348. Map(dropInputStopsIndex).
  1349. Tee(writeTripIndex).
  1350. Map(clearTripOffsets).
  1351. Tee(writeStopNameIndex).
  1352. Tee(writeStopCodeIndex).
  1353. // ---
  1354. Bind(getTrips).
  1355. Bind(convertLineGraphs).
  1356. Map(clearStops).
  1357. Bind(convertLines).
  1358. Tee(writeLineIndex).
  1359. Tee(writeLineIdIndex).
  1360. Map(clearLineGraphs).
  1361. Map(clearLineHeadsigns).
  1362. Tee(deleteTxtFiles).
  1363. Tee(compressTraffic).
  1364. Tee(deleteBareFiles).
  1365. Tee(moveTraffic)
  1366. if r.E != nil {
  1367. log.Printf("Error converting %s: %v\n", args.feed.Name(), r.E)
  1368. allErrors = append(allErrors, r.E)
  1369. }
  1370. if err := append(r.S.ValidFromError, r.S.ValidTillError...); len(err) != 0 {
  1371. allErrors = append(allErrors, err...)
  1372. log.Printf("Error converting %s: %v\n", args.feed.Name(), errors.Join(err...))
  1373. }
  1374. }
  1375. if len(allErrors) > 0 {
  1376. return gott.Tuple{args}, errors.Join(allErrors...)
  1377. }
  1378. return gott.Tuple{args}, nil
  1379. }
  1380. func signal(input ...interface{}) (interface{}, error) {
  1381. args := input[0].(result)
  1382. if len(args.gtfsFilenames) > 0 && args.pid > 0 {
  1383. process, err := os.FindProcess(args.pid)
  1384. if err != nil {
  1385. return gott.Tuple{args}, err
  1386. }
  1387. err = process.Signal(syscall.SIGUSR1)
  1388. if err != nil {
  1389. return gott.Tuple{args}, err
  1390. }
  1391. }
  1392. return gott.Tuple{args}, nil
  1393. }
  1394. func openLastUpdated(input ...interface{}) (interface{}, error) {
  1395. args := input[0].(result)
  1396. updatesFilename := filepath.Join(args.config.FeedsPath, "updated.bare")
  1397. var err error
  1398. args.updatesFile, err = os.OpenFile(updatesFilename, os.O_RDWR|os.O_CREATE, 0644)
  1399. return gott.Tuple{args}, err
  1400. }
  1401. func isEmpty(input ...interface{}) error {
  1402. args := input[0].(result)
  1403. stat, err := os.Stat(args.updatesFile.Name())
  1404. if err != nil {
  1405. return err
  1406. }
  1407. if stat.Size() == 0 {
  1408. return ErrEmpty{}
  1409. }
  1410. return nil
  1411. }
  1412. func unmarshalLastUpdated(input ...interface{}) (interface{}, error) {
  1413. args := input[0].(result)
  1414. var lastUpdated map[string]string
  1415. err := bare.UnmarshalReader(args.updatesFile, &lastUpdated)
  1416. args.updates = lastUpdated
  1417. return gott.Tuple{args}, err
  1418. }
  1419. func recoverEmpty(input ...interface{}) (interface{}, error) {
  1420. args := input[0].(result)
  1421. err := input[1].(error)
  1422. var emptyError ErrEmpty
  1423. if errors.As(err, &emptyError) {
  1424. return gott.Tuple{args}, nil
  1425. } else {
  1426. return gott.Tuple{args}, err
  1427. }
  1428. }
  1429. func lastUpdated(input ...interface{}) interface{} {
  1430. args := input[0].(result)
  1431. args.updates[args.feed.String()] = time.Now().Format(time.RFC3339)
  1432. return gott.Tuple{args}
  1433. }
  1434. func seekLastUpdated(input ...interface{}) (interface{}, error) {
  1435. args := input[0].(result)
  1436. _, err := args.updatesFile.Seek(0, 0)
  1437. return gott.Tuple{args}, err
  1438. }
  1439. func marshalLastUpdated(input ...interface{}) error {
  1440. args := input[0].(result)
  1441. err := bare.MarshalWriter(bare.NewWriter(args.updatesFile), &args.updates)
  1442. args.updatesFile.Close()
  1443. return err
  1444. }
  1445. func Prepare(cfg config.Config, t Traffic, bimbaPid int, feedTranslations embed.FS) error { // todo(BAF18) remove pid
  1446. etags, err := readEtags(cfg)
  1447. if err != nil {
  1448. return fmt.Errorf("while reading etags: %w", err)
  1449. }
  1450. newEtags := map[string]string{}
  1451. for _, feed := range t.Feeds {
  1452. log.Printf("converting %s\n", feed.Name())
  1453. r := gott.Tuple{result{
  1454. config: cfg,
  1455. pid: bimbaPid,
  1456. tmpPath: os.TempDir(),
  1457. feed: feed,
  1458. feedName: feed.String(),
  1459. location: feed.getTimezone(),
  1460. updates: map[string]string{},
  1461. etags: etags,
  1462. newEtags: newEtags,
  1463. feedTranslations: feedTranslations,
  1464. }}
  1465. s, err := gott.NewResult(r).
  1466. SetLevelLog(gott.Debug).
  1467. Bind(createTmpPath).
  1468. Bind(createFeedHome).
  1469. Bind(listDownloadedVersions).
  1470. Bind(getAllVersions).
  1471. Map(findValidVersions).
  1472. Bind(getGtfsFiles).
  1473. Bind(convert).
  1474. Bind(signal).
  1475. Bind(openLastUpdated).
  1476. Tee(isEmpty).
  1477. Bind(unmarshalLastUpdated).
  1478. Recover(recoverEmpty).
  1479. Map(lastUpdated).
  1480. Bind(seekLastUpdated).
  1481. Tee(marshalLastUpdated).
  1482. Finish()
  1483. if err != nil {
  1484. log.Printf("Error converting %s: %v\n", feed.String(), err)
  1485. } else {
  1486. etags = s.(gott.Tuple)[0].(result).etags
  1487. newEtags = s.(gott.Tuple)[0].(result).newEtags
  1488. }
  1489. }
  1490. return saveEtags(cfg, newEtags)
  1491. }