VTA Translator (#27)

* Add VTA translator

gtfs_realtime_translators/__version__.py

@@ -1 +1 @@
-__version__ = '0.9.1'
+__version__ = '1.0.0'

gtfs_realtime_translators/registry/registry.py

@@ -2,7 +2,8 @@
 
 from gtfs_realtime_translators.translators import LaMetroGtfsRealtimeTranslator, \
     SeptaRegionalRailTranslator, MtaSubwayGtfsRealtimeTranslator, NjtRailGtfsRealtimeTranslator, \
-    CtaSubwayGtfsRealtimeTranslator, CtaBusGtfsRealtimeTranslator, PathGtfsRealtimeTranslator
+    CtaSubwayGtfsRealtimeTranslator, CtaBusGtfsRealtimeTranslator, PathGtfsRealtimeTranslator, \
+    VtaGtfsRealtimeTranslator
 
 
 class TranslatorKeyWarning(Warning):
@@ -18,6 +19,7 @@ class TranslatorRegistry:
         'mta-subway': MtaSubwayGtfsRealtimeTranslator,
         'njt-rail': NjtRailGtfsRealtimeTranslator,
         'path': PathGtfsRealtimeTranslator,
+        'vta': VtaGtfsRealtimeTranslator,
     }
 
     @classmethod

gtfs_realtime_translators/translators/__init__.py

@@ -5,3 +5,4 @@
 from .cta_subway import CtaSubwayGtfsRealtimeTranslator
 from .cta_bus import CtaBusGtfsRealtimeTranslator
 from .path_rail import PathGtfsRealtimeTranslator
+from .vta import VtaGtfsRealtimeTranslator

gtfs_realtime_translators/translators/vta.py

@@ -0,0 +1,60 @@
+import json
+import math
+
+import pendulum
+
+from gtfs_realtime_translators.factories import TripUpdate, FeedMessage
+
+
+class VtaGtfsRealtimeTranslator:
+    def __init__(self, stop_id=None):
+        if stop_id is None:
+            raise ValueError('stop_id is required.')
+
+        self.stop_id = stop_id
+
+    def __call__(self, data):
+        json_data = json.loads(data)
+        entities = []
+        for data in json_data["data"]["predictionsData"]:
+            trip_updates = self.__make_trip_updates(data, self.stop_id)
+            entities.extend(trip_updates)
+
+        return FeedMessage.create(entities=entities)
+
+    @classmethod
+    def __make_trip_updates(cls, data, stop_id):
+        trip_updates = []
+        route_id = data.get("routeId")
+
+        ##### Intersection Extensions
+        route_short_name = data.get("routeShortName")
+        route_long_name = data.get("routeName")
+        stop_name = data.get("stopName")
+
+        for destination in data["destinations"]:
+            # Intersection Extension
+            headsign = destination.get("headsign")
+
+            # realtime predictions
+            predictions = enumerate(destination["predictions"])
+            for _idx, arrival in predictions:
+                entity_id = str(_idx + 1)
+                now = int(pendulum.now().timestamp())
+                arrival_or_departure_time = now + math.floor(arrival.get("sec") / 60) * 60
+                trip_id = arrival.get('tripId')
+
+                trip_update = TripUpdate.create(entity_id=entity_id,
+                                                arrival_time=arrival_or_departure_time,
+                                                departure_time=arrival_or_departure_time,
+                                                trip_id=trip_id,
+                                                route_id=route_id,
+                                                route_short_name=route_short_name,
+                                                route_long_name=route_long_name,
+                                                stop_id=stop_id,
+                                                stop_name=stop_name,
+                                                headsign=headsign)
+
+                trip_updates.append(trip_update)
+
+        return trip_updates

test/fixtures/vta_rail.json

@@ -0,0 +1,57 @@
+{
+    "success": true,
+    "route": "/real-time/vta/predictions GET",
+    "data": {
+      "agencyKey": "vta",
+      "predictionsData": [
+        {
+          "routeShortName": "Orange Line",
+          "routeName": "Orange Line - Mountain View - Alum Rock",
+          "routeId": "Ornge",
+          "stopId": "5236",
+          "stopName": "Milpitas Station",
+          "stopCode": 65236,
+          "destinations": [
+            {
+              "directionId": "0",
+              "headsign": "Alum Rock",
+              "predictions": [
+                {
+                  "time": 1600886400,
+                  "sec": 86,
+                  "min": 1,
+                  "departure": true,
+                  "tripId": "2960461",
+                  "vehicleId": "937"
+                },
+                {
+                  "time": 1600887600,
+                  "sec": 1286,
+                  "min": 21,
+                  "departure": true,
+                  "tripId": "2960460",
+                  "vehicleId": "919"
+                },
+                {
+                  "time": 1600888800,
+                  "sec": 2486,
+                  "min": 41,
+                  "departure": true,
+                  "tripId": "2960459",
+                  "vehicleId": "987"
+                },
+                {
+                  "time": 1600890000,
+                  "sec": 3686,
+                  "min": 61,
+                  "departure": true,
+                  "tripId": "2960458",
+                  "vehicleId": "979"
+                }
+              ]
+            }
+          ]
+        }
+      ]
+    }
+  }

test/test_vta.py

@@ -0,0 +1,43 @@
+import pytest
+import pendulum
+
+from gtfs_realtime_translators.translators import VtaGtfsRealtimeTranslator
+from gtfs_realtime_translators.factories import FeedMessage
+from gtfs_realtime_translators.bindings import intersection_pb2 as intersection_gtfs_realtime
+
+
+@pytest.fixture
+def vta_rail():
+    with open('test/fixtures/vta_rail.json') as f:
+        raw = f.read()
+
+    return raw
+
+
+def test_vta_data(vta_rail):
+    translator = VtaGtfsRealtimeTranslator(stop_id='5236')
+    with pendulum.test(pendulum.datetime(2019,2,20,17,0,0)):
+        message = translator(vta_rail)
+
+    entity = message.entity[0]
+    trip_update = entity.trip_update
+    stop_time_update = trip_update.stop_time_update[0]
+
+    assert message.header.gtfs_realtime_version == FeedMessage.VERSION
+
+    assert entity.id == '1'
+
+    assert trip_update.trip.trip_id == '2960461'
+    assert trip_update.trip.route_id == 'Ornge'
+
+    assert stop_time_update.arrival.time == 1550682060
+    assert stop_time_update.departure.time == 1550682060
+    assert stop_time_update.stop_id == '5236'
+
+    # test extensions
+    ixn_stop_time_updates = stop_time_update.Extensions[intersection_gtfs_realtime.intersection_stop_time_update]
+    ixn_trip_updates = trip_update.Extensions[intersection_gtfs_realtime.intersection_trip_update]
+    assert ixn_stop_time_updates.stop_name == "Milpitas Station"
+    assert ixn_trip_updates.headsign == "Alum Rock"
+    assert ixn_trip_updates.route_short_name == "Orange Line"
+    assert ixn_trip_updates.route_long_name == "Orange Line - Mountain View - Alum Rock"