Add departure and arrival datetime to Direction

Author: khamaileon

poetry.lock

@@ -24,6 +24,7 @@ matplotlib = {version = "^3.4.1", optional = true}
 contextily = {version = "^1.1.0", optional = true}
 geopandas = {version = "^0.8.2", optional = true}
 descartes = {version = "^1.0.0", optional = true}
+pytz = "^2023.3"
 
 [tool.poetry.extras]
 notebooks = ["shapely", "ipykernel", "geopandas", "contextily", "matplotlib", "descartes"]

pyproject.toml

@@ -1,5 +1,6 @@
 certifi==2023.7.22 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"
 charset-normalizer==3.2.0 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"
 idna==3.4 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"
+pytz==2023.3 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"
 requests==2.31.0 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"
 urllib3==2.0.4 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"

requirements.txt

@@ -25,7 +25,7 @@ pluggy==1.2.0 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"
 pre-commit==2.21.0 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"
 pygments==2.15.1 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"
 pytest==7.4.0 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"
-pytz==2023.3 ; python_full_version >= "3.8.0" and python_version < "3.9"
+pytz==2023.3 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"
 pyyaml==6.0.1 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"
 requests==2.31.0 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"
 responses==0.10.16 ; python_full_version >= "3.8.0" and python_full_version < "4.0.0"

requirements_dev.txt

@@ -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/direction.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/google.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,

routingpy/routers/opentripplanner_v2.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)