Add departure and arrival datetime to Direction

Author: khamaileon

routingpy/direction.py

@@ -17,6 +17,7 @@
 """
 :class:`.Direction` returns directions results.
 """
+import datetime
 from typing import List, Optional
 
 
@@ -65,7 +66,15 @@ class Direction(object):
     Contains a parsed directions' response. Access via properties ``geometry``, ``duration`` and ``distance``.
     """
 
-    def __init__(self, geometry=None, duration=None, distance=None, raw=None):
+    def __init__(
+        self,
+        geometry: List[List[float]] = None,
+        duration: int = None,
+        distance: int = None,
+        departure_datetime: datetime.datetime = None,
+        arrival_datetime: datetime.datetime = None,
+        raw: dict = None,
+    ):
         """
         Initialize a :class:`Direction` object to hold the properties of a directions request.
 
@@ -78,13 +87,21 @@ def __init__(self, geometry=None, duration=None, distance=None, raw=None):
         :param distance: The distance of the direction in meters.
         :type distance: int
 
+        :param departure_datetime: The departure timezone aware date and time of the direction.
+        :type departure_datetime: datetime.datetime
+
+        :param arrival_datetime: The arrival timezone aware date and time of the direction.
+        :type arrival_datetime: datetime.datetime
+
         :param raw: The raw response of an individual direction (for multiple alternative routes) or the whole direction
             response.
         :type raw: dict
         """
         self._geometry = geometry
         self._duration = duration
         self._distance = distance
+        self._departure_datetime = departure_datetime
+        self._arrival_datetime = arrival_datetime
         self._raw = raw
 
     @property
@@ -114,6 +131,24 @@ def distance(self) -> int:
         """
         return self._distance
 
+    @property
+    def departure_datetime(self) -> Optional[datetime.datetime]:
+        """
+        The departure timezone aware date and time of the direction.
+
+        :rtype: datetime.datetime or None
+        """
+        return self._departure_datetime
+
+    @property
+    def arrival_datetime(self) -> Optional[datetime.datetime]:
+        """
+        The arrival timezone aware date and time of the direction.
+
+        :rtype: datetime.datetime or None
+        """
+        return self._arrival_datetime
+
     @property
     def raw(self) -> Optional[dict]:
         """

routingpy/routers/google.py

@@ -15,9 +15,12 @@
 # the License.
 #
 
+import datetime
 from operator import itemgetter
 from typing import List, Optional, Tuple, Union
 
+import pytz
+
 from .. import convert, utils
 from ..client_base import DEFAULT
 from ..client_default import Client
@@ -319,12 +322,42 @@ def directions(  # noqa: C901
         if transit_routing_preference:
             params["transit_routing_preference"] = transit_routing_preference
 
-        return self.parse_direction_json(
+        return self._parse_direction_json(
             self.client._request("/directions/json", get_params=params, dry_run=dry_run), alternatives
         )
 
-    @staticmethod
-    def parse_direction_json(response, alternatives):
+    def _time_object_to_aware_datetime(self, time_object):
+        timestamp = time_object["value"]
+        dt = datetime.datetime.fromtimestamp(timestamp)
+        timezone = pytz.timezone(time_object["time_zone"])
+        return dt.astimezone(timezone)
+
+    def _parse_legs(self, legs):
+        duration = 0
+        distance = 0
+        geometry = []
+        departure_datetime = None
+        arrival_datetime = None
+
+        for leg in legs:
+            departure_time = leg.get("departure_time")
+            if departure_time:
+                assert len(legs) == 1, "departure_time is only supported for single leg routes"
+                departure_datetime = self._time_object_to_aware_datetime(departure_time)
+
+            arrival_time = leg.get("arrival_time")
+            if arrival_time:
+                assert len(legs) == 1, "arrival_time is only supported for single leg routes"
+                arrival_datetime = self._time_object_to_aware_datetime(arrival_time)
+
+            duration += leg["duration"]["value"]
+            distance += leg["distance"]["value"]
+            for step in leg["steps"]:
+                geometry.extend(utils.decode_polyline5(step["polyline"]["points"]))
+
+        return duration, distance, geometry, departure_datetime, arrival_datetime
+
+    def _parse_direction_json(self, response, alternatives):
         if response is None:  # pragma: no cover
             if alternatives:
                 return Directions()
@@ -345,33 +378,27 @@ def parse_direction_json(response, alternatives):
 
             raise error(STATUS_CODES[status]["code"], STATUS_CODES[status]["message"])
 
-        if alternatives:
-            routes = []
-            for route in response["routes"]:
-                geometry = []
-                duration, distance = 0, 0
-                for leg in route["legs"]:
-                    duration += leg["duration"]["value"]
-                    distance += leg["distance"]["value"]
-                    for step in leg["steps"]:
-                        geometry.extend(utils.decode_polyline5(step["polyline"]["points"]))
-
-                routes.append(
-                    Direction(
-                        geometry=geometry, duration=int(duration), distance=int(distance), raw=route
-                    )
+        directions = []
+        for route in response["routes"]:
+            duration, distance, geometry, departure_datetime, arrival_datetime = self._parse_legs(
+                route["legs"]
+            )
+            directions.append(
+                Direction(
+                    geometry=geometry,
+                    duration=int(duration),
+                    distance=int(distance),
+                    departure_datetime=departure_datetime,
+                    arrival_datetime=arrival_datetime,
+                    raw=route,
                 )
-            return Directions(routes, response)
-        else:
-            geometry = []
-            duration, distance = 0, 0
-            for leg in response["routes"][0]["legs"]:
-                duration += leg["duration"]["value"]
-                distance += leg["distance"]["value"]
-                for step in leg["steps"]:
-                    geometry.extend(utils.decode_polyline5(step["polyline"]["points"]))
-
-            return Direction(geometry=geometry, duration=duration, distance=distance, raw=response)
+            )
+
+        if alternatives:
+            return Directions(directions, raw=response)
+
+        elif directions:
+            return directions[0]
 
     def isochrones(self):  # pragma: no cover
         raise NotImplementedError
@@ -506,12 +533,11 @@ def matrix(  # noqa: C901
         if transit_routing_preference:
             params["transit_routing_preference"] = transit_routing_preference
 
-        return self.parse_matrix_json(
+        return self._parse_matrix_json(
             self.client._request("/distancematrix/json", get_params=params, dry_run=dry_run)
         )
 
-    @staticmethod
-    def parse_matrix_json(response):
+    def _parse_matrix_json(self, response):
         if response is None:  # pragma: no cover
             return Matrix()
 

routingpy/routers/opentripplanner_v2.py

@@ -15,7 +15,7 @@
 # the License.
 #
 import datetime
-from typing import List, Optional  # noqa: F401
+from typing import List, Optional
 
 from .. import convert, utils
 from ..client_base import DEFAULT
@@ -167,30 +167,35 @@ def directions(
         )
         return self._parse_directions_response(response, num_itineraries)
 
+    def _timestamp_to_utc_datetime(self, timestamp):
+        dt = datetime.datetime.fromtimestamp(timestamp / 1000)
+        return dt.astimezone(datetime.timezone.utc)
+
     def _parse_directions_response(self, response, num_itineraries):
         if response is None:  # pragma: no cover
             return Directions() if num_itineraries > 1 else Direction()
 
-        routes = []
+        directions = []
         for itinerary in response["data"]["plan"]["itineraries"]:
-            geometry, distance = self._parse_legs(itinerary["legs"])
-            routes.append(
+            distance, geometry = self._parse_legs(itinerary["legs"])
+            departure_datetime = self._timestamp_to_utc_datetime(itinerary["startTime"])
+            arrival_datetime = self._timestamp_to_utc_datetime(itinerary["endTime"])
+            directions.append(
                 Direction(
                     geometry=geometry,
                     duration=int(itinerary["duration"]),
                     distance=distance,
+                    departure_datetime=departure_datetime,
+                    arrival_datetime=arrival_datetime,
                     raw=itinerary,
                 )
             )
 
         if num_itineraries > 1:
-            return Directions(routes, raw=response)
-
-        elif routes:
-            return routes[0]
+            return Directions(directions, raw=response)
 
-        else:
-            return Direction()
+        elif directions:
+            return directions[0]
 
     def _parse_legs(self, legs):
         distance = 0
@@ -200,7 +205,7 @@ def _parse_legs(self, legs):
             geometry.extend(list(reversed(points)))
             distance += int(leg["distance"])
 
-        return geometry, distance
+        return distance, geometry
 
     def isochrones(
         self,

tests/test_base.py

@@ -84,7 +84,6 @@ def test_skip_api_error(self):
         )
 
         client = ClientMock(base_url="https://httpbin.org", skip_api_error=False)
-        print(client.skip_api_error)
         with self.assertRaises(routingpy.exceptions.RouterApiError):
             client.directions(url="/post", post_params=self.params)
 

tests/test_google.py

@@ -47,7 +47,7 @@ def test_full_directions(self):
             content_type="application/json",
         )
 
-        routes = self.client.directions(**query)
+        directions = self.client.directions(**query)
         self.assertEqual(1, len(responses.calls))
         self.assertURLEqual(
             "https://maps.googleapis.com/maps/api/directions/json?alternatives=true&arrival_time=1567512000&"
@@ -57,12 +57,44 @@ def test_full_directions(self):
             responses.calls[0].request.url,
         )
 
-        self.assertIsInstance(routes, Directions)
-        self.assertIsInstance(routes[0], Direction)
-        self.assertIsInstance(routes[0].geometry, list)
-        self.assertIsInstance(routes[0].distance, int)
-        self.assertIsInstance(routes[0].duration, int)
-        self.assertIsInstance(routes[0].raw, dict)
+        self.assertIsInstance(directions, Directions)
+        self.assertIsInstance(directions[0], Direction)
+        self.assertIsInstance(directions[0].geometry, list)
+        self.assertIsInstance(directions[0].distance, int)
+        self.assertIsInstance(directions[0].duration, int)
+        self.assertIsInstance(directions[0].raw, dict)
+
+    @responses.activate
+    def test_directions_transit(self):
+        query = ENDPOINTS_QUERIES[self.name]["directions_transit"]
+
+        responses.add(
+            responses.GET,
+            "https://maps.googleapis.com/maps/api/directions/json",
+            status=200,
+            json=ENDPOINTS_RESPONSES[self.name]["directions_transit"],
+            content_type="application/json",
+        )
+
+        direction = self.client.directions(**query)
+        self.assertEqual(1, len(responses.calls))
+        self.assertURLEqual(
+            "https://maps.googleapis.com/maps/api/directions/json?arrival_time=1691064000&"
+            "destination=49.415776%2C8.680916&key=sample_key&mode=transit&"
+            "origin=49.420577%2C8.688641",
+            responses.calls[0].request.url,
+        )
+
+        self.assertIsInstance(direction, Direction)
+        self.assertIsInstance(direction, Direction)
+        self.assertIsInstance(direction.geometry, list)
+        self.assertIsInstance(direction.distance, int)
+        self.assertIsInstance(direction.duration, int)
+        self.assertIsInstance(direction.departure_datetime, datetime.datetime)
+        self.assertEqual(direction.departure_datetime.tzinfo.zone, "Europe/Berlin")
+        self.assertIsInstance(direction.arrival_datetime, datetime.datetime)
+        self.assertEqual(direction.arrival_datetime.tzinfo.zone, "Europe/Berlin")
+        self.assertIsInstance(direction.raw, dict)
 
     @responses.activate
     def test_full_directions_no_alternatives(self):
@@ -77,7 +109,7 @@ def test_full_directions_no_alternatives(self):
             content_type="application/json",
         )
 
-        routes = self.client.directions(**query)
+        direction = self.client.directions(**query)
         self.assertEqual(1, len(responses.calls))
         self.assertURLEqual(
             "https://maps.googleapis.com/maps/api/directions/json?alternatives=false&arrival_time=1567512000&"
@@ -87,11 +119,11 @@ def test_full_directions_no_alternatives(self):
             responses.calls[0].request.url,
         )
 
-        self.assertIsInstance(routes, Direction)
-        self.assertIsInstance(routes.geometry, list)
-        self.assertIsInstance(routes.duration, int)
-        self.assertIsInstance(routes.distance, int)
-        self.assertIsInstance(routes.raw, dict)
+        self.assertIsInstance(direction, Direction)
+        self.assertIsInstance(direction.geometry, list)
+        self.assertIsInstance(direction.duration, int)
+        self.assertIsInstance(direction.distance, int)
+        self.assertIsInstance(direction.raw, dict)
 
     @responses.activate
     def test_waypoint_generator_directions(self):
@@ -240,11 +272,11 @@ def test_status_codes(self):
 
         for alternatives in [True, False]:
             with self.assertRaises(RouterApiError):
-                self.client.parse_direction_json(
+                self.client._parse_direction_json(
                     error_responses["ZERO_RESULTS"], alternatives=alternatives
                 )
 
             with self.assertRaises(RouterServerError):
-                self.client.parse_direction_json(
+                self.client._parse_direction_json(
                     error_responses["UNKNOWN_ERROR"], alternatives=alternatives
                 )