Add path to filter

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "multimodalrouter"
-version = "0.1.8"
+version = "0.1.9"
 description = "A graph-based routing library for dynamic routing."
 readme = "README.md"
 license = { file = "LICENSE.md" }

src/multimodalrouter/graph/dataclasses.py

@@ -195,5 +195,5 @@ def filterHub(self, hub: Hub) -> bool:
         """
         pass
 
-    def filter(self, start: Hub, end: Hub, edge: EdgeMetadata) -> bool:
+    def filter(self, start: Hub, end: Hub, edge: EdgeMetadata, current_path: list = None) -> bool:
         return self.filterHub(start) and self.filterHub(end) and self.filterEdge(edge)

src/multimodalrouter/graph/graph.py

@@ -477,7 +477,10 @@ def find_shortest_path(
                             raise ValueError(
                                 f"Hub with ID '{next_hub_id}' not found in graph! But it is connected to hub '{current_hub_id}' via mode '{mode}'." # noqa: E501
                             )
-                        if custom_filter is not None and not custom_filter.filter(current_hub, next_hub, connection_metrics):
+                        if (
+                            custom_filter is not None and
+                            not custom_filter.filter(current_hub, next_hub, connection_metrics, path_with_modes)
+                        ):
                             continue
 
                         # get the selected metric alue for this connection

tests/unit/test_routegraph_public_features.py

@@ -517,3 +517,160 @@ def filterEdge(self, edge: EdgeMetadata) -> bool:
         reachableIds = [hub.id for _, hub in reachable]
         self.assertIn('B', reachableIds)
         self.assertIn('A', reachableIds)
+
+    def test_shortest_path_with_path_aware_filter(self):
+        """Test that filter can access and use the current path to limit consecutive segments."""
+        testDf = pd.DataFrame(
+            columns=['source', 'destination', 'distance', 'source_lat', 'source_lng', 'destination_lat', 'destination_lng'],
+            data=[
+                ('A', 'B', 1, 1, 1, 1, 2),
+                ('B', 'C', 1, 1, 2, 1, 3),
+                ('C', 'D', 1, 1, 3, 1, 4),
+                ('D', 'E', 1, 1, 4, 1, 5),
+                ('E', 'F', 1, 1, 5, 1, 6),
+                ('A', 'F', 10, 1, 1, 1, 6),  # Direct route is longer
+            ]
+        )
+
+        temp_path = os.path.join(self.temp_dir.name, 'temp_path_aware.csv')
+        testDf.to_csv(temp_path, index=False)
+
+        class PathAwareFilter(Filter):
+            """Filter that limits consecutive segments with the same transport mode."""
+
+            def __init__(self, max_consecutive_segments: int = 2):
+                self.max_consecutive_segments = max_consecutive_segments
+
+            def filterHub(self, hub: Hub) -> bool:
+                return True
+
+            def filterEdge(self, edge: EdgeMetadata) -> bool:
+                return True
+
+            def filter(self, start: Hub, end: Hub, edge: EdgeMetadata, current_path: list = None) -> bool:
+                if current_path is None or len(current_path) == 0:
+                    return True
+
+                mode = edge.transportMode
+
+                # count consecutive segments with the same mode
+                consecutive_count = 0
+                for i in range(len(current_path) - 1, -1, -1):
+                    path_mode = current_path[i][1] if len(current_path[i]) > 1 else ""
+                    if path_mode == mode:
+                        consecutive_count += 1
+                    else:
+                        break
+
+                # if more than max consecutive segments, return False
+                if consecutive_count >= self.max_consecutive_segments:
+                    return False
+
+                return True
+
+        graph = RouteGraph(
+            maxDistance=50,
+            transportModes={'H': 'mv'},
+            dataPaths={'H': temp_path},
+            compressed=False,
+            extraMetricsKeys=[],
+            drivingEnabled=False
+        )
+
+        f = io.StringIO()
+        with contextlib.redirect_stdout(f), contextlib.redirect_stderr(f):
+            graph.build()
+
+        # search route that only ever has two consecutive segments of one mode
+        route = graph.find_shortest_path(
+            'A',
+            'F',
+            allowed_modes=['mv'],
+            custom_filter=PathAwareFilter(max_consecutive_segments=2)
+        )
+        self.assertIsNotNone(route)
+        path = route.path
+        starts = [p[0] for p in path]
+
+        # should take direct route A -> F (10 distance) instead of A -> B -> C (would be 3 segments)
+        self.assertEqual(starts, ['A', 'F'])
+        self.assertAlmostEqual(route.totalMetrics.getMetric('distance'), 10, places=5)
+
+        # without filter, should take the shorter multi-hop route
+        route_no_filter = graph.find_shortest_path('A', 'F', allowed_modes=['mv'])
+        self.assertIsNotNone(route_no_filter)
+        path_no_filter = route_no_filter.path
+        starts_no_filter = [p[0] for p in path_no_filter]
+
+        # should be A -> B -> C -> D -> E -> F (5 distance total)
+        self.assertEqual(starts_no_filter, ['A', 'B', 'C', 'D', 'E', 'F'])
+        self.assertAlmostEqual(route_no_filter.totalMetrics.getMetric('distance'), 5, places=5)
+
+    def test_shortest_path_with_path_aware_filter_verbose(self):
+        """Test path-aware filter works with verbose mode."""
+        testDf = pd.DataFrame(
+            columns=['source', 'destination', 'distance', 'source_lat', 'source_lng', 'destination_lat', 'destination_lng'],
+            data=[
+                ('A', 'B', 1, 1, 1, 1, 2),
+                ('B', 'C', 1, 1, 2, 1, 3),
+                ('C', 'D', 1, 1, 3, 1, 4),
+                ('A', 'D', 5, 1, 1, 1, 4),
+            ]
+        )
+
+        temp_path = os.path.join(self.temp_dir.name, 'temp_path_aware_verbose.csv')
+        testDf.to_csv(temp_path, index=False)
+
+        class PathAwareFilter(Filter):
+            def __init__(self, max_consecutive: int = 1):
+                self.max_consecutive = max_consecutive
+
+            def filterHub(self, hub: Hub) -> bool:
+                return True
+
+            def filterEdge(self, edge: EdgeMetadata) -> bool:
+                return True
+
+            def filter(self, start: Hub, end: Hub, edge: EdgeMetadata, current_path: list = None) -> bool:
+                if current_path is None or len(current_path) == 0:
+                    return True
+
+                mode = edge.transportMode
+                consecutive = 0
+                for i in range(len(current_path) - 1, -1, -1):
+                    # Handle both verbose and non-verbose path formats
+                    path_mode = current_path[i][1] if len(current_path[i]) > 1 else ""
+                    if path_mode == mode:
+                        consecutive += 1
+                    else:
+                        break
+
+                return consecutive < self.max_consecutive
+
+        graph = RouteGraph(
+            maxDistance=50,
+            transportModes={'H': 'mv'},
+            dataPaths={'H': temp_path},
+            compressed=False,
+            extraMetricsKeys=[],
+            drivingEnabled=False
+        )
+
+        f = io.StringIO()
+        with contextlib.redirect_stdout(f), contextlib.redirect_stderr(f):
+            graph.build()
+
+        route = graph.find_shortest_path(
+            'A',
+            'D',
+            allowed_modes=['mv'],
+            verbose=True,
+            custom_filter=PathAwareFilter(max_consecutive=1)
+        )
+        self.assertIsNotNone(route)
+        path = route.path
+        starts = [p[0] for p in path]
+
+        # max 1 consecutive, forced to take direct route
+        self.assertEqual(starts, ['A', 'D'])
+        self.assertAlmostEqual(route.totalMetrics.getMetric('distance'), 5, places=5)