Merge pull request #4089 from hove-io/jormungandr_implement_pt_journe…

…y_fare

[Jormungandr] Implement pt journey fare

source/jormungandr/jormungandr/instance.py

@@ -73,6 +73,7 @@
     str_to_time_stamp,
 )
 from jormungandr import pt_planners_manager, transient_socket
+from jormungandr.pt_journey_fare import PtJourneyFareBackendManager
 import os
 
 type_to_pttype = {
@@ -162,6 +163,7 @@ def __init__(
         streetnetwork_backend_manager,
         external_service_provider_configurations,
         pt_planners_configurations,
+        pt_journey_fare_configurations,
         ghost_words=None,
         instance_db=None,
         best_boarding_positions_dir=None,
@@ -310,6 +312,11 @@ def __init__(
             file_path = os.path.join(stop_points_attractivities_dir, "{}.csv".format(self.name))
             self.stop_points_attractivities = read_stop_points_attractivities(file_path)
 
+        # TODO: use db
+        self._pt_journey_fare_backend_manager = PtJourneyFareBackendManager(
+            self, pt_journey_fare_configurations, None
+        )
+
     def get_providers_from_db(self):
         """
         :return: a callable query of equipment providers associated to the current instance in db
@@ -825,12 +832,20 @@ def additional_parameters(self):
     default_pt_planner = _make_property_getter('default_pt_planner')
     pt_planners_configurations = _make_property_getter('pt_planners_configurations')
 
+    loki_pt_journey_fare = _make_property_getter('loki_pt_journey_fare')
+    loki_compute_pt_journey_fare = _make_property_getter('loki_compute_pt_journey_fare')
+    loki_pt_journey_fare_configurations = _make_property_getter('loki_pt_journey_fare_configurations')
+
     filter_odt_journeys = _make_property_getter('filter_odt_journeys')
 
     def get_pt_planner(self, pt_planner_id=None):
         pt_planner_id = pt_planner_id or self.default_pt_planner
         return self._pt_planner_manager.get_pt_planner(pt_planner_id)
 
+    def get_pt_journey_fare(self, loki_pt_journey_fare_id=None):
+        pt_journey_fare_id = loki_pt_journey_fare_id or self.loki_pt_journey_fare
+        return self._pt_journey_fare_backend_manager.get_pt_journey_fare(pt_journey_fare_id)
+
     @property
     def places_proximity_radius(self):
         # type: () -> int

source/jormungandr/jormungandr/instance_manager.py

@@ -121,6 +121,7 @@ def register_instance(self, config):
             self._streetnetwork_backend_manager,
             config.get('external_services_providers', []),
             config.get('pt_planners', {}),
+            config.get('pt_journey_fares', {}),
             config.get('ghost_words', []),
             best_boarding_positions_dir=app.config.get(str('BEST_BOARDING_POSITIONS_DIR'), None),
             olympics_forbidden_uris=config.get('olympics_forbidden_uris', None),

source/jormungandr/jormungandr/interfaces/v1/Journeys.py

@@ -683,6 +683,24 @@ def __init__(self):
             help="used to adjust the search range in Asgard when computing matrix",
         )
 
+        parser_get.add_argument(
+            "_loki_pt_journey_fare",
+            type=OptionValue(['kraken']),
+            default='kraken',
+            hidden=True,
+            help="only works when loki is selected as pt journey engine, "
+            "choose which PT-fare engine to compute the journey's PT-fares",
+        )
+
+        parser_get.add_argument(
+            "_loki_compute_pt_journey_fare",
+            type=BooleanType(),
+            default=False,
+            hidden=True,
+            help="only works when loki is selected as pt journey engine, "
+            "whether to use external engine to compute pt journey fare",
+        )
+
     @add_tad_links()
     @add_debug_info()
     @add_fare_links()
@@ -814,6 +832,12 @@ def _set_specific_params(mod):
             if args.get('_filter_odt_journeys') is None:
                 args['_filter_odt_journeys'] = mod.filter_odt_journeys
 
+            if args.get('_loki_pt_journey_fare') is None:
+                args['_loki_pt_journey_fare'] = mod.loki_pt_journey_fare
+
+            if args.get('_loki_compute_pt_journey_fare') is None:
+                args['_loki_compute_pt_journey_fare'] = mod.loki_compute_pt_journey_fare
+
         # When computing 'same_journey_schedules'(is_journey_schedules=True), some parameters need to be overridden
         # because they are contradictory to the request
         if args.get("is_journey_schedules"):

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

@@ -153,6 +153,7 @@ def __init__(
             autocomplete_type='kraken',
             streetnetwork_backend_manager=StreetNetworkBackendManager(),
             external_service_provider_configurations=[],
+            pt_journey_fare_configurations={},
         )
         self.disable_database = disable_database
         self.equipment_provider_manager = EquipmentProviderManager(equipment_details_config)

source/jormungandr/jormungandr/pt_journey_fare/__init__.py

@@ -0,0 +1,29 @@
+# Copyright (c) 2001-2023, Hove and/or its affiliates. All rights reserved.
+#
+# This file is part of Navitia,
+#     the software to build cool stuff with public transport.
+#
+# Hope you'll enjoy and contribute to this project,
+#     powered by Hove (www.hove.com).
+# Help us simplify mobility and open public transport:
+#     a non ending quest to the responsive locomotion way of traveling!
+#
+# LICENCE: This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+# Stay tuned using
+# twitter @navitia
+# channel `#navitia` on riot https://riot.im/app/#/room/#navitia:matrix.org
+# https://groups.google.com/d/forum/navitia
+# www.navitia.io
+from .pt_journey_fare_backend_manager import PtJourneyFareBackendManager

source/jormungandr/jormungandr/pt_journey_fare/kraken.py

@@ -0,0 +1,67 @@
+# Copyright (c) 2001-2023, Hove and/or its affiliates. All rights reserved.
+#
+# This file is part of Navitia,
+#     the software to build cool stuff with public transport.
+#
+# Hope you'll enjoy and contribute to this project,
+#     powered by Hove (www.hove.com).
+# Help us simplify mobility and open public transport:
+#     a non ending quest to the responsive locomotion way of traveling!
+#
+# LICENCE: This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+# Stay tuned using
+# twitter @navitia
+# channel `#navitia` on riot https://riot.im/app/#/room/#navitia:matrix.org
+# https://groups.google.com/d/forum/navitia
+# www.navitia.io
+from jormungandr.pt_journey_fare.pt_journey_fare import AbstractPtJourneyFare
+from navitiacommon import response_pb2, request_pb2, type_pb2
+
+
+class Kraken(AbstractPtJourneyFare):
+    def __init__(self, instance):
+        self.instance = instance
+
+    @staticmethod
+    def _pt_sections(journey):
+        return [s for s in journey.sections if s.type == response_pb2.PUBLIC_TRANSPORT]
+
+    def create_fare_request(self, pt_journeys):
+        request = request_pb2.Request()
+        request.requested_api = type_pb2.pt_fares
+        pt_fare_request = request.pt_fares
+        for journey in pt_journeys:
+            pt_sections = self._pt_sections(journey)
+            if not pt_sections:
+                continue
+            pt_journey_request = pt_fare_request.pt_journeys.add(id=journey.internal_id)
+            for s in pt_sections:
+                pt_journey_request.pt_sections.add(
+                    id=s.id,
+                    network_uri=s.uris.network,
+                    start_stop_area_uri=s.origin.stop_point.stop_area.uri,
+                    end_stop_area_uri=s.destination.stop_point.stop_area.uri,
+                    line_uri=s.uris.line,
+                    begin_date_time=s.begin_date_time,
+                    end_date_time=s.end_date_time,
+                    first_stop_point_uri=s.origin.stop_point.uri,
+                    last_stop_point_uri=s.destination.stop_point.uri,
+                    physical_mode=s.uris.physical_mode,
+                )
+        return request
+
+    def get_pt_journeys_fares(self, pt_journeys, request_id):
+        fare_request = self.create_fare_request(pt_journeys)
+        return self.instance.send_and_receive(fare_request, request_id="{}_fare".format(request_id))

source/jormungandr/jormungandr/pt_journey_fare/pt_journey_fare.py

@@ -0,0 +1,40 @@
+# Copyright (c) 2001-2023, Hove and/or its affiliates. All rights reserved.
+#
+# This file is part of Navitia,
+#     the software to build cool stuff with public transport.
+#
+# Hope you'll enjoy and contribute to this project,
+#     powered by Hove (www.hove.com).
+# Help us simplify mobility and open public transport:
+#     a non ending quest to the responsive locomotion way of traveling!
+#
+# LICENCE: This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+# Stay tuned using
+# twitter @navitia
+# channel `#navitia` on riot https://riot.im/app/#/room/#navitia:matrix.org
+# https://groups.google.com/d/forum/navitia
+# www.navitia.io
+import abc
+
+
+# Using abc.ABCMeta in a way it is compatible both with Python 2.7 and Python 3.x
+# http://stackoverflow.com/a/38668373/1614576
+ABC = abc.ABCMeta(str("ABC"), (object,), {})
+
+
+class AbstractPtJourneyFare(ABC):  # type: ignore
+    @abc.abstractmethod
+    def get_pt_journeys_fares(self, pt_journeys, request_id):
+        raise NotImplementedError("Method not implemented")

source/jormungandr/jormungandr/pt_journey_fare/pt_journey_fare_backend_manager.py

@@ -0,0 +1,63 @@
+# Copyright (c) 2001-2023, Hove and/or its affiliates. All rights reserved.
+#
+# This file is part of Navitia,
+#     the software to build cool stuff with public transport.
+#
+# Hope you'll enjoy and contribute to this project,
+#     powered by Hove (www.hove.com).
+# Help us simplify mobility and open public transport:
+#     a non ending quest to the responsive locomotion way of traveling!
+#
+# LICENCE: This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+# Stay tuned using
+# twitter @navitia
+# channel `#navitia` on riot https://riot.im/app/#/room/#navitia:matrix.org
+# https://groups.google.com/d/forum/navitia
+# www.navitia.io
+from jormungandr import utils
+
+
+class NoRequestedPtJourneyFare(Exception):
+    pass
+
+
+class PtJourneyFareBackendManager(object):
+    def __init__(self, instance, configs, db_configs_getter=None):
+        # TODO: read from db
+        self._db_configs_getter = db_configs_getter
+
+        self.instance = instance
+        kraken_default_config = {
+            "class": "jormungandr.pt_journey_fare.kraken.Kraken",
+            "args": {"instance": self.instance},
+        }
+        backends_configs = {
+            "kraken": kraken_default_config,
+        }
+        backends_configs.update(configs or {})
+
+        self.backends = {}
+
+        self.init(backends_configs)
+
+    def init(self, backends_configs):
+        for k, v in backends_configs.items():
+            self.backends[k] = utils.create_object(v)
+
+    def get_pt_journey_fare(self, backend_id):
+        backend = self.backends.get(backend_id)
+        if backend:
+            return backend
+        raise NoRequestedPtJourneyFare("no requested pt_journey_fare backend: {}".format(backend_id))

source/jormungandr/jormungandr/scenarios/distributed.py

@@ -48,7 +48,6 @@
 from jormungandr.street_network.utils import crowfly_distance_between
 from jormungandr.scenarios.utils import (
     fill_uris,
-    switch_back_to_ridesharing,
     updated_common_journey_request_with_default,
 )
 from jormungandr.new_relic import record_custom_parameter
@@ -322,10 +321,18 @@ def finalise_journeys(self, future_manager, request, responses, context, instanc
             future_manager=future_manager, instance=instance, request=request, request_id=request_id
         )
 
+        pt_journey_fare_pool = PtJourneyFarePool(
+            future_manager=future_manager, instance=instance, request=request, request_id=request_id
+        )
+
         if request['_transfer_path'] is True:
             for journey in journeys_to_complete:
                 transfer_pool.async_compute_transfer(journey.pt_journeys.sections)
 
+        if request['_loki_compute_pt_journey_fare'] is True:
+            for response in responses:
+                pt_journey_fare_pool.async_compute_fare(response, request_id)
+
         wait_and_complete_pt_journey(
             requested_orig_obj=context.requested_orig_obj,
             requested_dest_obj=context.requested_dest_obj,
@@ -339,6 +346,10 @@ def finalise_journeys(self, future_manager, request, responses, context, instanc
             journeys=journeys_to_complete,
             request_id="{}_complete_pt_journey".format(request_id),
         )
+        if request['_loki_compute_pt_journey_fare'] is True:
+            wait_and_complete_pt_journey_fare(
+                pt_elements=journeys_to_complete, pt_journey_fare_pool=pt_journey_fare_pool
+            )
 
     def _compute_isochrone_common(
         self,

source/jormungandr/jormungandr/scenarios/helper_classes/__init__.py

@@ -34,7 +34,8 @@
 from .pt_journey import PtJourneyPool
 from .proximities_by_crowfly import ProximitiesByCrowflyPool
 from .transfer import TransferPool
-from .complete_pt_journey import wait_and_complete_pt_journey
+from .pt_journey_fare import PtJourneyFarePool
+from .complete_pt_journey import wait_and_complete_pt_journey, wait_and_complete_pt_journey_fare
 from .helper_exceptions import PtException, EntryPointException, FinaliseException, StreetNetworkException
 from .helper_utils import get_entry_point_or_raise, check_final_results_or_raise
 from .helper_future import FutureManager

source/jormungandr/jormungandr/scenarios/helper_classes/complete_pt_journey.py

@@ -144,7 +144,7 @@ def get_journeys_to_complete(responses, context, is_debug):
         if r is None:
             continue
         for j in r.journeys:
-            if is_debug == False and "to_delete" in j.tags:
+            if is_debug is False and "to_delete" in j.tags:
                 continue
             if j.internal_id in context.journeys_to_modes:
                 journey_modes = context.journeys_to_modes[j.internal_id]
@@ -217,3 +217,19 @@ def wait_and_complete_pt_journey(
                     pt_journey=pt_element.pt_journeys,
                     transfer_pool=transfer_pool,
                 )
+
+
+def wait_and_complete_pt_journey_fare(pt_elements, pt_journey_fare_pool):
+    journeys_map = {j.pt_journeys.internal_id: j.pt_journeys for j in pt_elements}
+    for response, fare_response in pt_journey_fare_pool.wait_and_generate():
+        response.tickets.extend(fare_response.tickets)
+        for f in fare_response.pt_journey_fares:
+            journey = journeys_map.get(f.journey_id)
+            if journey is None:
+                logging.getLogger(__name__).warning(
+                    "something wrong has occurred when completing journey fare: journey {} can't be found".format(
+                        f.journey_id
+                    )
+                )
+                continue
+            journey.fare.CopyFrom(f.fare)