Merge pull request #4296 from hove-io/manage_odt_information

[jormun]: Update deeplink placeholders in odt_information

Author: azime

source/jormungandr/jormungandr/interfaces/v1/serializer/journey.py

@@ -68,6 +68,7 @@
     SectionType,
     CyclePathType,
     BoardingPosition,
+    OdtInformation,
 )
 import navitiacommon.response_pb2
 from navitiacommon.type_pb2 import RTLevel
@@ -331,6 +332,15 @@ class RidesharingInformationSerializer(PbNestedSerializer):
     seats = SeatsDescriptionSerializer(display_none=False)
 
 
+class OdtInformationSerializer(PbNestedSerializer):
+    name = jsonschema.Field(schema_type=str, display_none=True)
+    url = jsonschema.Field(schema_type=str, display_none=True)
+    conditions = jsonschema.Field(schema_type=str, display_none=True)
+    phone = jsonschema.Field(schema_type=str, display_none=True)
+    deeplink = jsonschema.Field(schema_type=str, display_none=True)
+    applies_on = EnumListField(attr='applies_on', pb_type=OdtInformation.AppliesOn)
+
+
 class SectionSerializer(PbNestedSerializer):
     id = jsonschema.Field(schema_type=str, display_none=True)
     duration = jsonschema.Field(
@@ -421,6 +431,7 @@ def get_ridesharing_journeys(self, obj):
     street_informations = StreetInformationSerializer(
         attr="street_network.street_information", many=True, display_none=False
     )
+    odt_informations = OdtInformationSerializer(display_none=False)
 
 
 class JourneySerializer(PbNestedSerializer):

source/jormungandr/jormungandr/interfaces/v1/test/serializer_tests.py

@@ -31,6 +31,12 @@
 from __future__ import absolute_import, print_function, unicode_literals, division
 
 from jormungandr.interfaces.v1.serializer.base import SortedGenericSerializer
+from jormungandr.interfaces.v1.decorators import get_serializer
+from jormungandr.interfaces.v1.serializer import api
+from jormungandr import app
+import pytz
+from flask import g
+import jormungandr.scenarios.tests.helpers_tests as helpers_tests
 import serpy
 
 
@@ -47,3 +53,29 @@ def sort_key(self, obj):
     assert data[1]['v'] == 2
     assert data[2]['v'] == 3
     assert data[3]['v'] == 4
+
+
+@get_serializer(serpy=api.JourneysSerializer)
+def abcd():
+    deeplink = "https://toto.com?from=from_value&to=to_value"
+    return helpers_tests.get_odt_journey(deeplink=deeplink)
+
+
+def odt_information_serialization_test():
+    with app.app_context():
+        with app.test_request_context():
+            g.timezone = pytz.utc
+            # get journey response in json
+            resp = abcd()
+            assert len(resp.get("journeys", 0)) == 1
+            journey = resp["journeys"][0]
+            assert len(journey.get("sections", 0)) == 3
+            section = journey["sections"][1]
+            odt_information = section.get("odt_informations", None)
+            assert odt_information is not None
+            assert odt_information["url"] == "odt_url_value"
+            assert odt_information["name"] == "odt_name_value"
+            assert odt_information["phone"] == "odt_phone_value"
+            assert odt_information["conditions"] == "odt_conditions_value"
+            assert odt_information["deeplink"] == "https://toto.com?from=from_value&to=to_value"
+            assert odt_information["applies_on"] == ["from"]

source/jormungandr/jormungandr/scenarios/new_default.py

@@ -50,6 +50,7 @@
     get_disruptions_on_poi,
     add_disruptions,
     get_impact_uris_for_poi,
+    update_odt_information_deeplink_in_section,
 )
 from navitiacommon import type_pb2, response_pb2, request_pb2
 from jormungandr.scenarios.qualifier import (
@@ -539,6 +540,20 @@ def update_disruptions_on_pois(instance, pb_resp):
     add_disruptions(pb_resp, poi_disruptions)
 
 
+def update_odt_information_deeplink(pb_resp):
+    """
+    Update placeholders present in sections[i].odt_information.deeplink with their values for each journey
+    for each section of type ON_DEMAND_TRANSPORT
+    """
+    if not pb_resp.journeys:
+        return
+
+    for j in pb_resp.journeys:
+        for s in j.sections:
+            if s.type == response_pb2.ON_DEMAND_TRANSPORT:
+                update_odt_information_deeplink_in_section(s)
+
+
 def update_total_air_pollutants(pb_resp):
     """
     update journey.air_pollutants
@@ -1485,6 +1500,9 @@ def fill_journeys(self, request_type, api_request, instance):
         # Update disruptions on pois
         update_disruptions_on_pois(instance, pb_resp)
 
+        # Update deeplink in odt_information for all sections of type ON_DEMAND_TRANSPORT
+        update_odt_information_deeplink(pb_resp)
+
         self._compute_pagination_links(pb_resp, instance, api_request['clockwise'])
         return pb_resp
 

source/jormungandr/jormungandr/scenarios/tests/helpers_tests.py

@@ -635,3 +635,42 @@ def verify_poi_in_impacted_objects(object, poi_empty=True):
         assert object.poi.name == 'poi_name_from_kraken'
         assert object.poi.coord.lon == 1.0
         assert object.poi.coord.lat == 2.0
+
+
+def get_odt_journey(deeplink):
+    response = response_pb2.Response()
+    journey = response.journeys.add()
+
+    section = journey.sections.add()
+    section.type = response_pb2.STREET_NETWORK
+    section.street_network.mode = response_pb2.Walking
+    section.duration = 20
+    section = journey.sections.add()
+    section.type = response_pb2.ON_DEMAND_TRANSPORT
+    section.duration = 70
+    section.begin_date_time = utils.str_to_time_stamp("20240806T060500")
+    section.origin.uri = 'stop_a'
+    section.origin.embedded_type = type_pb2.STOP_POINT
+    section.origin.stop_point.uri = 'stop_a'
+    section.origin.stop_point.name = 'stop_a_name'
+    section.origin.stop_point.coord.lon = 1.0
+    section.origin.stop_point.coord.lat = 2.0
+    section.destination.uri = 'stop_b'
+    section.destination.embedded_type = type_pb2.STOP_POINT
+    section.destination.stop_point.uri = 'stop_b'
+    section.destination.stop_point.name = 'stop_b_name'
+    section.destination.stop_point.coord.lon = 3.0
+    section.destination.stop_point.coord.lat = 4.0
+    odt_information = section.odt_informations
+    odt_information.name = "odt_name_value"
+    odt_information.deeplink = deeplink
+    odt_information.url = "odt_url_value"
+    odt_information.conditions = "odt_conditions_value"
+    odt_information.phone = "odt_phone_value"
+    odt_information.applies_on.append(response_pb2.OdtInformation.AppliesOn.FROM)
+    section = journey.sections.add()
+    section.type = response_pb2.STREET_NETWORK
+    section.street_network.mode = response_pb2.Walking
+    section.duration = 10
+
+    return response

source/jormungandr/jormungandr/scenarios/tests/new_default_tests.py

@@ -38,6 +38,7 @@
     get_kraken_calls,
     update_best_boarding_positions,
     update_disruptions_on_pois,
+    update_odt_information_deeplink,
 )
 from jormungandr.instance import Instance
 from jormungandr.scenarios.utils import switch_back_to_ridesharing
@@ -831,3 +832,28 @@ def journey_with_disruptions_on_poi_test(mocker):
 
     mock.assert_called_once()
     return
+
+
+def journey_with_odt_information_test():
+    deeplink = (
+        "https://domaine/search?departure-address={from_name}&destination-address={to_name}"
+        "&requested-departure-time={departure_datetime}&from_coord_lat={from_coord_lat}"
+        "&from_coord_lon={from_coord_lon}&not_managed={not_managed}"
+    )
+    response_journey_with_odt = helpers_tests.get_odt_journey(deeplink=deeplink)
+    assert len(response_journey_with_odt.journeys) == 1
+    journey = response_journey_with_odt.journeys[0]
+    assert len(journey.sections) == 3
+    odt_section = journey.sections[1]
+    assert odt_section.type == response_pb2.ON_DEMAND_TRANSPORT
+    assert (
+        odt_section.odt_informations.deeplink
+        == "https://domaine/search?departure-address={from_name}&destination-address={to_name}&requested-departure-time={departure_datetime}&from_coord_lat={from_coord_lat}&from_coord_lon={from_coord_lon}&not_managed={not_managed}"
+    )
+
+    update_odt_information_deeplink(response_journey_with_odt)
+    odt_section = response_journey_with_odt.journeys[0].sections[1]
+    assert (
+        odt_section.odt_informations.deeplink
+        == "https://domaine/search?departure-address=stop_a_name&destination-address=stop_b_name&requested-departure-time=1722924300&from_coord_lat=2.0&from_coord_lon=1.0&not_managed=N/A"
+    )

source/jormungandr/jormungandr/scenarios/tests/utils_tests.py

@@ -29,7 +29,12 @@
 
 from navitiacommon import type_pb2, response_pb2
 import jormungandr.scenarios.tests.helpers_tests as helpers_tests
-from jormungandr.scenarios.utils import fill_disruptions_on_pois, fill_disruptions_on_places_nearby
+from jormungandr.scenarios.utils import (
+    fill_disruptions_on_pois,
+    fill_disruptions_on_places_nearby,
+    update_odt_information_deeplink_in_section,
+)
+
 import pytest
 from pytest_mock import mocker
 
@@ -94,3 +99,61 @@ def update_disruptions_on_pois_for_places_nearby_test(mocker):
 
     mock.assert_called_once()
     return
+
+
+def journey_with_deeplink_in_odt_information_test():
+    instance = lambda: None
+    # Get a response with a section of ODT having odt_information.
+    deeplink = (
+        "https://domaine/search?departure-address={from_name}&destination-address={to_name}"
+        "&requested-departure-time={departure_datetime}&from_coord_lat={from_coord_lat}"
+        "&from_coord_lon={from_coord_lon}&to_coord_lat={to_coord_lat}&to_coord_lon={to_coord_lon}"
+    )
+    response_journey_with_odt = helpers_tests.get_odt_journey(deeplink=deeplink)
+    assert len(response_journey_with_odt.journeys) == 1
+    journey = response_journey_with_odt.journeys[0]
+    assert len(journey.sections) == 3
+    odt_section = journey.sections[1]
+    assert odt_section.type == response_pb2.ON_DEMAND_TRANSPORT
+    odt_information = odt_section.odt_informations
+    assert odt_information.name == "odt_name_value"
+    assert (
+        odt_information.deeplink
+        == "https://domaine/search?departure-address={from_name}&destination-address={to_name}&requested-departure-time={departure_datetime}&from_coord_lat={from_coord_lat}&from_coord_lon={from_coord_lon}&to_coord_lat={to_coord_lat}&to_coord_lon={to_coord_lon}"
+    )
+    assert odt_information.url == "odt_url_value"
+    assert odt_information.conditions == "odt_conditions_value"
+    assert odt_information.phone == "odt_phone_value"
+    update_odt_information_deeplink_in_section(odt_section)
+    assert (
+        odt_information.deeplink
+        == "https://domaine/search?departure-address=stop_a_name&destination-address=stop_b_name&requested-departure-time=1722924300&from_coord_lat=2.0&from_coord_lon=1.0&to_coord_lat=4.0&to_coord_lon=3.0"
+    )
+
+    # Use a deeplink with fewer placeholders
+    deeplink = (
+        "https://domaine/search?departure-address={from_name}&destination-address={to_name}"
+        "&requested-departure-time={departure_datetime}&from_coord_lat={from_coord_lat}&from_coord_lon={from_coord_lon}"
+    )
+    response_journey_with_odt = helpers_tests.get_odt_journey(deeplink=deeplink)
+    odt_section = response_journey_with_odt.journeys[0].sections[1]
+    update_odt_information_deeplink_in_section(odt_section)
+    assert (
+        odt_section.odt_informations.deeplink
+        == "https://domaine/search?departure-address=stop_a_name&destination-address=stop_b_name&requested-departure-time=1722924300&from_coord_lat=2.0&from_coord_lon=1.0"
+    )
+
+    # Add a placeholder which is not predefined in the function to update deeplink
+    # This placeholder will not be replaced(updated)
+    deeplink = (
+        "https://domaine/search?departure-address={from_name}&destination-address={to_name}"
+        "&requested-departure-time={departure_datetime}&from_coord_lat={from_coord_lat}"
+        "&from_coord_lon={from_coord_lon}&toto={toto}"
+    )
+    response_journey_with_odt = helpers_tests.get_odt_journey(deeplink=deeplink)
+    odt_section = response_journey_with_odt.journeys[0].sections[1]
+    update_odt_information_deeplink_in_section(odt_section)
+    assert (
+        odt_section.odt_informations.deeplink
+        == "https://domaine/search?departure-address=stop_a_name&destination-address=stop_b_name&requested-departure-time=1722924300&from_coord_lat=2.0&from_coord_lon=1.0&toto=N/A"
+    )

source/jormungandr/jormungandr/scenarios/utils.py

@@ -33,6 +33,9 @@
 import navitiacommon.request_pb2 as request_pb2
 from future.moves.itertools import zip_longest
 from jormungandr.fallback_modes import FallbackModes
+import re
+from collections import defaultdict
+from string import Formatter
 from copy import deepcopy
 import six
 
@@ -551,3 +554,45 @@ def add_disruptions(pb_resp, pb_disruptions):
     if pb_disruptions is None:
         return
     pb_resp.impacts.extend(pb_disruptions.impacts)
+
+
+def update_odt_information_deeplink_in_section(section):
+    if section.type != response_pb2.ON_DEMAND_TRANSPORT:
+        return
+
+    deeplink = section.odt_informations.deeplink
+    if not deeplink:
+        return
+
+    departure_datetime = section.begin_date_time
+    from_name = section.origin.stop_point.name
+    from_coord_lat = section.origin.stop_point.coord.lat
+    from_coord_lon = section.origin.stop_point.coord.lon
+    to_name = section.destination.stop_point.name
+    to_coord_lat = section.destination.stop_point.coord.lat
+    to_coord_lon = section.destination.stop_point.coord.lon
+
+    # Get all placeholders present in deeplink and match with predefined placeholder variables. value of those
+    # present in deeplink but absent in predefined placeholder variables will be replaced by N/A
+    placeholders = re.findall(r"{(\w+)}", deeplink)
+
+    placeholder_dict = defaultdict(lambda: 'N/A')
+    fmtr = Formatter()
+
+    for p in placeholders:
+        if p == "departure_datetime":
+            placeholder_dict[p] = departure_datetime
+        elif p == "from_name":
+            placeholder_dict[p] = from_name
+        elif p == "from_coord_lat":
+            placeholder_dict[p] = from_coord_lat
+        elif p == "from_coord_lon":
+            placeholder_dict[p] = from_coord_lon
+        elif p == "to_name":
+            placeholder_dict[p] = to_name
+        elif p == "to_coord_lat":
+            placeholder_dict[p] = to_coord_lat
+        elif p == "to_coord_lon":
+            placeholder_dict[p] = to_coord_lon
+
+    section.odt_informations.deeplink = fmtr.vformat(deeplink, (), placeholder_dict)