From 9f6b8391657d7141fd0471ef7825b726082c075c Mon Sep 17 00:00:00 2001 From: Tobias Karusseit Date: Thu, 8 Jan 2026 17:02:59 +0100 Subject: [PATCH 1/5] update filter to use path dependency --- src/multimodalrouter/graph/dataclasses.py | 2 +- src/multimodalrouter/graph/graph.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/multimodalrouter/graph/dataclasses.py b/src/multimodalrouter/graph/dataclasses.py index 2c1c2bf..babba8d 100644 --- a/src/multimodalrouter/graph/dataclasses.py +++ b/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) diff --git a/src/multimodalrouter/graph/graph.py b/src/multimodalrouter/graph/graph.py index 9838028..1669e7c 100644 --- a/src/multimodalrouter/graph/graph.py +++ b/src/multimodalrouter/graph/graph.py @@ -477,7 +477,7 @@ 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 From d60449a85795de7f2e33c5e0e1a2dbe45fcf3104 Mon Sep 17 00:00:00 2001 From: Tobias Karusseit Date: Thu, 8 Jan 2026 17:03:05 +0100 Subject: [PATCH 2/5] update test --- tests/unit/test_routegraph_public_features.py | 146 ++++++++++++++++++ 1 file changed, 146 insertions(+) diff --git a/tests/unit/test_routegraph_public_features.py b/tests/unit/test_routegraph_public_features.py index 6e059ce..1738301 100644 --- a/tests/unit/test_routegraph_public_features.py +++ b/tests/unit/test_routegraph_public_features.py @@ -517,3 +517,149 @@ 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) \ No newline at end of file From b3014292ff4939338f331557b43cfa9f24b61289 Mon Sep 17 00:00:00 2001 From: Tobias Karusseit Date: Thu, 8 Jan 2026 17:03:12 +0100 Subject: [PATCH 3/5] update version --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 662441f..9d7d37e 100644 --- a/pyproject.toml +++ b/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" } From e875e1073089a98ed8ff9475b3e1153329d2c5eb Mon Sep 17 00:00:00 2001 From: Tobias Karusseit Date: Thu, 8 Jan 2026 21:48:14 +0100 Subject: [PATCH 4/5] lint --- src/multimodalrouter/graph/graph.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/multimodalrouter/graph/graph.py b/src/multimodalrouter/graph/graph.py index 1669e7c..cd2e49b 100644 --- a/src/multimodalrouter/graph/graph.py +++ b/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, path_with_modes): + 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 From 6c8b232640eaa120d0e28ca142390fbca11ab440 Mon Sep 17 00:00:00 2001 From: Tobias Karusseit Date: Thu, 8 Jan 2026 21:51:38 +0100 Subject: [PATCH 5/5] lint tests --- tests/unit/test_routegraph_public_features.py | 37 ++++++++++++------- 1 file changed, 24 insertions(+), 13 deletions(-) diff --git a/tests/unit/test_routegraph_public_features.py b/tests/unit/test_routegraph_public_features.py index 1738301..a29f175 100644 --- a/tests/unit/test_routegraph_public_features.py +++ b/tests/unit/test_routegraph_public_features.py @@ -537,7 +537,7 @@ def test_shortest_path_with_path_aware_filter(self): 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 @@ -550,9 +550,9 @@ def filterEdge(self, edge: EdgeMetadata) -> bool: 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): @@ -561,11 +561,11 @@ def filter(self, start: Hub, end: Hub, edge: EdgeMetadata, current_path: list = 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( @@ -582,11 +582,16 @@ def filter(self, start: Hub, end: Hub, edge: EdgeMetadata, current_path: list = 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)) + 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) @@ -596,7 +601,7 @@ def filter(self, start: Hub, end: Hub, edge: EdgeMetadata, current_path: list = 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) @@ -629,7 +634,7 @@ def filterEdge(self, edge: EdgeMetadata) -> bool: 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): @@ -639,7 +644,7 @@ def filter(self, start: Hub, end: Hub, edge: EdgeMetadata, current_path: list = consecutive += 1 else: break - + return consecutive < self.max_consecutive graph = RouteGraph( @@ -655,11 +660,17 @@ def filter(self, start: Hub, end: Hub, edge: EdgeMetadata, current_path: list = 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)) + 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) \ No newline at end of file + self.assertAlmostEqual(route.totalMetrics.getMetric('distance'), 5, places=5)