GTFS RT Service Alerts #122

Travigo · Nov 9, 2024 · 8da5b69 · 8da5b69
1 parent 31c95a7
commit 8da5b69
Show file tree

Hide file tree

Showing 9 changed files with 788 additions and 386 deletions.
diff --git a/data/datasources/se-trafiklab.yaml b/data/datasources/se-trafiklab.yaml
@@ -105,3 +105,93 @@ datasets:
     realtimejourneys: true
   linkeddataset: se-trafiklab-gtfs-schedule
   importdestination: realtime-queue
+- identifier: gtfs-realtime-sl-alerts
+  format: gtfs-realtime
+  source: "https://opendata.samtrafiken.se/gtfs-rt-sweden/sl/ServiceAlertsSweden.pb"
+  sourceauthentication:
+    query:
+      key: TRAVIGO_SE_TRAFIKLAB_REALTIME_API_KEY
+  supportedobjects:
+    realtimejourneys: true
+  linkeddataset: se-trafiklab-gtfs-schedule
+  importdestination: realtime-queue
+- identifier: gtfs-realtime-ul-alerts
+  format: gtfs-realtime
+  source: "https://opendata.samtrafiken.se/gtfs-rt-sweden/ul/ServiceAlertsSweden.pb"
+  sourceauthentication:
+    query:
+      key: TRAVIGO_SE_TRAFIKLAB_REALTIME_API_KEY
+  supportedobjects:
+    realtimejourneys: true
+  linkeddataset: se-trafiklab-gtfs-schedule
+  importdestination: realtime-queue
+- identifier: gtfs-realtime-otraf-alerts
+  format: gtfs-realtime
+  source: "https://opendata.samtrafiken.se/gtfs-rt-sweden/otraf/ServiceAlertsSweden.pb"
+  sourceauthentication:
+    query:
+      key: TRAVIGO_SE_TRAFIKLAB_REALTIME_API_KEY
+  supportedobjects:
+    realtimejourneys: true
+  linkeddataset: se-trafiklab-gtfs-schedule
+  importdestination: realtime-queue
+- identifier: gtfs-realtime-klt-alerts
+  format: gtfs-realtime
+  source: "https://opendata.samtrafiken.se/gtfs-rt-sweden/klt/ServiceAlertsSweden.pb"
+  sourceauthentication:
+    query:
+      key: TRAVIGO_SE_TRAFIKLAB_REALTIME_API_KEY
+  supportedobjects:
+    realtimejourneys: true
+  linkeddataset: se-trafiklab-gtfs-schedule
+  importdestination: realtime-queue
+- identifier: gtfs-realtime-skane-alerts
+  format: gtfs-realtime
+  source: "https://opendata.samtrafiken.se/gtfs-rt-sweden/skane/ServiceAlertsSweden.pb"
+  sourceauthentication:
+    query:
+      key: TRAVIGO_SE_TRAFIKLAB_REALTIME_API_KEY
+  supportedobjects:
+    realtimejourneys: true
+  linkeddataset: se-trafiklab-gtfs-schedule
+  importdestination: realtime-queue
+- identifier: gtfs-realtime-dt-alerts
+  format: gtfs-realtime
+  source: "https://opendata.samtrafiken.se/gtfs-rt-sweden/dt/ServiceAlertsSweden.pb"
+  sourceauthentication:
+    query:
+      key: TRAVIGO_SE_TRAFIKLAB_REALTIME_API_KEY
+  supportedobjects:
+    realtimejourneys: true
+  linkeddataset: se-trafiklab-gtfs-schedule
+  importdestination: realtime-queue
+- identifier: gtfs-realtime-varm-alerts
+  format: gtfs-realtime
+  source: "https://opendata.samtrafiken.se/gtfs-rt-sweden/varm/ServiceAlertsSweden.pb"
+  sourceauthentication:
+    query:
+      key: TRAVIGO_SE_TRAFIKLAB_REALTIME_API_KEY
+  supportedobjects:
+    realtimejourneys: true
+  linkeddataset: se-trafiklab-gtfs-schedule
+  importdestination: realtime-queue
+- identifier: gtfs-realtime-xt-alerts
+  format: gtfs-realtime
+  source: "https://opendata.samtrafiken.se/gtfs-rt-sweden/xt/ServiceAlertsSweden.pb"
+  sourceauthentication:
+    query:
+      key: TRAVIGO_SE_TRAFIKLAB_REALTIME_API_KEY
+  supportedobjects:
+    realtimejourneys: true
+  linkeddataset: se-trafiklab-gtfs-schedule
+  importdestination: realtime-queue
+- identifier: gtfs-realtime-vastmanland-alerts
+  format: gtfs-realtime
+  source: "https://opendata.samtrafiken.se/gtfs-rt-sweden/vastmanland/ServiceAlertsSweden.pb"
+  sourceauthentication:
+    query:
+      key: TRAVIGO_SE_TRAFIKLAB_REALTIME_API_KEY
+  supportedobjects:
+    realtimejourneys: true
+  linkeddataset: se-trafiklab-gtfs-schedule
+  importdestination: realtime-queue
diff --git a/deploy/charts/travigo-data-importer/values.yaml b/deploy/charts/travigo-data-importer/values.yaml
@@ -97,24 +97,43 @@ deployments:
     args: ["data-importer", "dataset", "--id", "fr-ilevia-lille-gtfs-realtime", "--repeat-every", "90s"]
   - name: de-gtfs-gtfs-realtime
     args: ["data-importer", "dataset", "--id", "de-gtfs-gtfs-realtime", "--repeat-every", "90s"]
+  # Sweden
   - name: se-trafiklab-gtfs-realtime-sl-trip
-    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-sl-trip", "--repeat-every", "180s"]
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-sl-trip", "--repeat-every", "300s"]
   - name: se-trafiklab-gtfs-realtime-ul-trip
-    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-ul-trip", "--repeat-every", "180s"]
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-ul-trip", "--repeat-every", "300s"]
   - name: se-trafiklab-gtfs-realtime-otraf-trip
-    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-otraf-trip", "--repeat-every", "180s"]
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-otraf-trip", "--repeat-every", "300s"]
   - name: se-trafiklab-gtfs-realtime-klt-trip
-    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-klt-trip", "--repeat-every", "180s"]
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-klt-trip", "--repeat-every", "300s"]
   - name: se-trafiklab-gtfs-realtime-skane-trip
-    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-skane-trip", "--repeat-every", "180s"]
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-skane-trip", "--repeat-every", "300s"]
   - name: se-trafiklab-gtfs-realtime-dt-trip
-    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-dt-trip", "--repeat-every", "180s"]
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-dt-trip", "--repeat-every", "300s"]
   - name: se-trafiklab-gtfs-realtime-varm-trip
-    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-varm-trip", "--repeat-every", "180s"]
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-varm-trip", "--repeat-every", "300s"]
   - name: se-trafiklab-gtfs-realtime-xt-trip
-    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-xt-trip", "--repeat-every", "180s"]
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-xt-trip", "--repeat-every", "300s"]
   - name: se-trafiklab-gtfs-realtime-vastm-trip
-    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-vastmanland-trip", "--repeat-every", "180s"]
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-vastmanland-trip", "--repeat-every", "300s"]
+  - name: se-trafiklab-gtfs-realtime-sl-alerts
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-sl-alerts", "--repeat-every", "900s"]
+  - name: se-trafiklab-gtfs-realtime-ul-alerts
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-ul-alerts", "--repeat-every", "900s"]
+  - name: se-trafiklab-gtfs-realtime-otraf-alerts
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-otraf-alerts", "--repeat-every", "900s"]
+  - name: se-trafiklab-gtfs-realtime-klt-alerts
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-klt-alerts", "--repeat-every", "900s"]
+  - name: se-trafiklab-gtfs-realtime-skane-alerts
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-skane-alerts", "--repeat-every", "900s"]
+  - name: se-trafiklab-gtfs-realtime-dt-alerts
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-dt-alerts", "--repeat-every", "900s"]
+  - name: se-trafiklab-gtfs-realtime-varm-alerts
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-varm-alerts", "--repeat-every", "900s"]
+  - name: se-trafiklab-gtfs-realtime-xt-alerts
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-xt-alerts", "--repeat-every", "900s"]
+  - name: se-trafiklab-gtfs-realtime-vastm-alerts
+    args: ["data-importer", "dataset", "--id", "se-trafiklab-gtfs-realtime-vastmanland-alerts", "--repeat-every", "900s"]
   # - name: us-nyc-subway-realtime-numbers
   #   args: ["data-importer", "dataset", "--id", "us-nyc-subway-relatime-1-2-3-4-5-6-7", "--repeat-every", "60s"]
 

diff --git a/deploy/charts/travigo-stats/values.yaml b/deploy/charts/travigo-stats/values.yaml
@@ -40,11 +40,11 @@ image:
 
 cronjobs:
   - name: basic
-    schedule: "0 8 * * *"
-    args: ["stats", "calculate", "--object", "services,operators,stops,servicealerts"]
-  - name: realtimejourneys
+    schedule: "0 */4 * * *"
+    args: ["stats", "calculate", "--object", "services,operators,stops"]
+  - name: dynamic
     schedule: "*/20 * * * *"
-    args: ["stats", "calculate", "--object", "realtimejourneys"]
+    args: ["stats", "calculate", "--object", "realtimejourneys,servicealerts"]
 
 imagePullSecrets: []
 nameOverride: ""

diff --git a/pkg/api/routes/servicealerts.go b/pkg/api/routes/servicealerts.go
@@ -1,6 +1,8 @@
 package routes
 
 import (
+	"crypto/sha256"
+	"fmt"
 	"strings"
 
 	"github.com/gofiber/fiber/v2"
@@ -14,6 +16,28 @@ func ServiceAlertRouter(router fiber.Router) {
 	router.Get("/stop/:identifier", getStopServiceAlerts)
 }
 
+func filterIdenticalServiceAlerts(serviceAlerts []*ctdf.ServiceAlert) []*ctdf.ServiceAlert {
+	var serviceAlertsFiltered []*ctdf.ServiceAlert
+	uniqueMap := make(map[string]bool)
+
+	for _, serviceAlert := range serviceAlerts {
+		hash := sha256.New()
+
+		hash.Write([]byte(serviceAlert.AlertType))
+		hash.Write([]byte(serviceAlert.Title))
+		hash.Write([]byte(serviceAlert.Text))
+
+		key := fmt.Sprintf("%x", hash.Sum(nil))
+
+		if !uniqueMap[key] {
+			uniqueMap[key] = true
+			serviceAlertsFiltered = append(serviceAlertsFiltered, serviceAlert)
+		}
+	}
+
+	return serviceAlertsFiltered
+}
+
 func getMatchingIdentifierServiceAlerts(c *fiber.Ctx) error {
 	identifier := c.Params("identifier")
 
@@ -22,13 +46,15 @@ func getMatchingIdentifierServiceAlerts(c *fiber.Ctx) error {
 		MatchingIdentifiers: strings.Split(identifier, ","),
 	})
 
+	serviceAlertsFiltered := filterIdenticalServiceAlerts(serviceAlerts)
+
 	if err != nil {
 		c.SendStatus(404)
 		return c.JSON(fiber.Map{
 			"error": err.Error(),
 		})
 	} else {
-		return c.JSON(serviceAlerts)
+		return c.JSON(serviceAlertsFiltered)
 	}
 }
 
@@ -68,12 +94,14 @@ func getStopServiceAlerts(c *fiber.Ctx) error {
 		MatchingIdentifiers: matchingIdentifiers,
 	})
 
+	serviceAlertsFiltered := filterIdenticalServiceAlerts(serviceAlerts)
+
 	if err != nil {
 		c.SendStatus(404)
 		return c.JSON(fiber.Map{
 			"error": err.Error(),
 		})
 	} else {
-		return c.JSON(serviceAlerts)
+		return c.JSON(serviceAlertsFiltered)
 	}
 }
diff --git a/pkg/dataimporter/formats/gtfs/realtime.go b/pkg/dataimporter/formats/gtfs/realtime.go
@@ -14,7 +14,6 @@ import (
 	"github.com/eko/gocache/lib/v4/cache"
 	"github.com/eko/gocache/lib/v4/store"
 	redisstore "github.com/eko/gocache/store/redis/v4"
-	"github.com/kr/pretty"
 	"github.com/rs/zerolog/log"
 	"github.com/travigo/travigo/pkg/ctdf"
 	"github.com/travigo/travigo/pkg/database"
@@ -65,6 +64,7 @@ func (r *Realtime) Import(dataset datasets.DataSet, datasource *ctdf.DataSource)
 	withTripID := 0
 	withLocation := 0
 	withTripUpdate := 0
+	serviceAlertCount := 0
 
 	for _, entity := range feed.Entity {
 		vehiclePosition := entity.GetVehicle()
@@ -93,15 +93,80 @@ func (r *Realtime) Import(dataset datasets.DataSet, datasource *ctdf.DataSource)
 
 		tripID := trip.GetTripId()
 
-		// TODO gtfs-rt alerts
 		if entity.Alert != nil {
-			pretty.Println(entity.GetAlert())
-			collection := database.GetCollection("datadump")
-			collection.InsertOne(context.Background(), bson.M{
-				"type":             "gtfsrt-alert",
-				"creationdatetime": time.Now(),
-				"document":         entity.GetAlert(),
-			})
+			var alertType ctdf.ServiceAlertType
+
+			switch entity.Alert.Effect {
+			case gtfs.Alert_NO_SERVICE.Enum():
+				alertType = ctdf.ServiceAlertTypeServiceSuspended
+			case gtfs.Alert_REDUCED_SERVICE.Enum():
+				alertType = ctdf.ServiceAlertTypeServicePartSuspended
+			case gtfs.Alert_SIGNIFICANT_DELAYS.Enum():
+				alertType = ctdf.ServiceAlertTypeSevereDelays
+			case gtfs.Alert_DETOUR.Enum():
+				alertType = ctdf.ServiceAlertTypeWarning // TODO new type?
+			case gtfs.Alert_ADDITIONAL_SERVICE.Enum():
+				alertType = ctdf.ServiceAlertTypeInformation
+			case gtfs.Alert_MODIFIED_SERVICE.Enum():
+				alertType = ctdf.ServiceAlertTypeWarning
+			case gtfs.Alert_OTHER_EFFECT.Enum():
+				alertType = ctdf.ServiceAlertTypeInformation
+			case gtfs.Alert_UNKNOWN_EFFECT.Enum():
+				alertType = ctdf.ServiceAlertTypeInformation
+			case gtfs.Alert_STOP_MOVED.Enum():
+				alertType = ctdf.ServiceAlertTypeWarning
+			case gtfs.Alert_NO_EFFECT.Enum():
+				alertType = ctdf.ServiceAlertTypeInformation
+			case gtfs.Alert_ACCESSIBILITY_ISSUE.Enum():
+				alertType = ctdf.ServiceAlertTypeInformation
+			default:
+				alertType = ctdf.ServiceAlertTypeInformation
+			}
+
+			// Create one per active period & informed entity
+			for _, informedEntity := range entity.Alert.InformedEntity {
+				for _, activePeriod := range entity.Alert.ActivePeriod {
+					validFromTimestamp := activePeriod.GetStart()
+					validToTimestamp := activePeriod.GetEnd()
+
+					validFrom := time.Unix(int64(validFromTimestamp), 0)
+					validTo := time.Unix(int64(validToTimestamp), 0)
+
+					tripID := informedEntity.GetTrip().GetTripId()
+					routeID := informedEntity.GetRouteId()
+					stopID := informedEntity.GetStopId()
+					agencyID := informedEntity.GetAgencyId()
+
+					updateEvent := vehicletracker.VehicleUpdateEvent{
+						MessageType: vehicletracker.VehicleUpdateEventTypeServiceAlert,
+						LocalID:     fmt.Sprintf("%s-realtime-%d-%d", dataset.Identifier, validFromTimestamp, validToTimestamp),
+						IdentifyingInformation: map[string]string{
+							"TripID":        tripID,
+							"RouteID":       routeID,
+							"StopID":        stopID,
+							"AgencyID":      agencyID,
+							"LinkedDataset": dataset.LinkedDataset,
+						},
+
+						ServiceAlertUpdate: &vehicletracker.ServiceAlertUpdate{
+							Type:        alertType,
+							Title:       *entity.Alert.HeaderText.GetTranslation()[0].Text, // TODO assume we only use the 1 translation
+							Description: *entity.Alert.DescriptionText.GetTranslation()[0].Text,
+							ValidFrom:   validFrom,
+							ValidUntil:  validTo,
+						},
+
+						SourceType: "GTFS-RT",
+						DataSource: datasource,
+						RecordedAt: recordedAtTime,
+					}
+
+					updateEventJson, _ := json.Marshal(updateEvent)
+					r.queue.PublishBytes(updateEventJson)
+
+					serviceAlertCount += 1
+				}
+			}
 		}
 
 		if tripID != "" {
@@ -240,6 +305,7 @@ func (r *Realtime) Import(dataset datasets.DataSet, datasource *ctdf.DataSource)
 		Int("withtrip", withTripID).
 		Int("withlocation", withLocation).
 		Int("withtripupdate", withTripUpdate).
+		Int("servicealert", serviceAlertCount).
 		Int("total", len(feed.Entity)).
 		Msg("Submitted vehicle updates")