feat: add get_activity_map tool for Leaflet-based route visualization and include corresponding unit tests

2026-05-30 12:12:10 +02:00
parent 39b182a87a
commit b4ab185240
2 changed files with 169 additions and 0 deletions
@@ -363,3 +363,95 @@ def register(mcp: FastMCP, strava: StravaClient) -> None:
            return await strava.get_activity_streams(activity_id, key_list)
        except Exception as e:
            return f"Error fetching activity streams: {str(e)}"
+
+    @mcp.tool()
+    async def get_activity_map(ctx: Context, activity_id: int):
+        """
+        Get an interactive map visualization for a specific Strava activity by its ID.
+        :param activity_id: The numeric ID of the Strava activity.
+        Returns a self-contained HTML page loading Leaflet.js and OpenStreetMap to render the route inside Open WebUI.
+        """
+        try:
+            await ctx.info(f"Fetching activity details for map (ID: {activity_id})...")
+            activity = await strava.get_activity(activity_id)
+
+            activity_map = activity.get("map", {})
+            polyline = activity_map.get("polyline") or activity_map.get(
+                "summary_polyline"
+            )
+
+            if not polyline:
+                warning_msg = f"⚠️ Keine GPS-Routen-Daten für diese Aktivität verfügbar (ID: {activity_id})."
+                await ctx.warning(warning_msg)
+                return [TextContent(type="text", text=warning_msg)]
+
+            name = activity.get("name", "Aktivität")
+
+            html_content = f"""<!DOCTYPE html>
+<html>
+<head>
+  <meta charset="utf-8">
+  <title>{name}</title>
+  <link rel="stylesheet" href="https://unpkg.com/leaflet@1.9.4/dist/leaflet.css" />
+  <script src="https://unpkg.com/leaflet@1.9.4/dist/leaflet.js"></script>
+  <style>
+    html, body, #map {{ height: 100%; width: 100%; margin: 0; padding: 0; }}
+  </style>
+</head>
+<body>
+  <div id="map"></div>
+  <script>
+    var map = L.map('map');
+    L.tileLayer('https://{{s}}.tile.openstreetmap.org/{{z}}/{{x}}/{{y}}.png', {{
+      attribution: '&copy; <a href="https://www.openstreetmap.org/copyright">OpenStreetMap</a> contributors'
+    }}).addTo(map);
+
+    function decodePolyline(encoded) {{
+      var points = [];
+      var index = 0, len = encoded.length;
+      var lat = 0, lng = 0;
+      while (index < len) {{
+        var b, shift = 0, result = 0;
+        do {{
+          b = encoded.charCodeAt(index++) - 63;
+          result |= (b & 0x1f) << shift;
+          shift += 5;
+        }} while (b >= 0x20);
+        var dlat = ((result & 1) ? ~(result >> 1) : (result >> 1));
+        lat += dlat;
+        shift = 0;
+        result = 0;
+        do {{
+          b = encoded.charCodeAt(index++) - 63;
+          result |= (b & 0x1f) << shift;
+          shift += 5;
+        }} while (b >= 0x20);
+        var dlng = ((result & 1) ? ~(result >> 1) : (result >> 1));
+        lng += dlng;
+        points.push([lat / 1e5, lng / 1e5]);
+      }}
+      return points;
+    }}
+
+    var encodedPolyline = {json.dumps(polyline)};
+    var coordinates = decodePolyline(encodedPolyline);
+    if (coordinates.length > 0) {{
+      var track = L.polyline(coordinates, {{color: '#fc4c02', weight: 5, opacity: 0.8}}).addTo(map);
+      map.fitBounds(track.getBounds());
+    }} else {{
+      map.setView([0, 0], 2);
+    }}
+  </script>
+</body>
+</html>"""
+            return [
+                TextContent(type="text", text=html_content.strip()),
+                _resource(
+                    f"internal://activities/{activity_id}/map",
+                    {"polyline": polyline},
+                ),
+            ]
+        except Exception as e:
+            error_msg = f"Error rendering activity map: {str(e)}"
+            await ctx.error(error_msg)
+            return [TextContent(type="text", text=error_msg)]
@@ -0,0 +1,77 @@
+from unittest.mock import AsyncMock, MagicMock
+import pytest
+from mcp.server.fastmcp import Context
+from strava_mcp_server.tools.activities import register
+
+
+class MockMCP:
+    def __init__(self):
+        self.tools = {}
+
+    def tool(self):
+        def decorator(func):
+            self.tools[func.__name__] = func
+            return func
+
+        return decorator
+
+
+@pytest.mark.asyncio
+async def test_get_activity_map_success():
+    mcp = MockMCP()
+    strava = MagicMock()
+
+    # Mock detailed activity with a valid polyline
+    strava.get_activity = AsyncMock(
+        return_value={
+            "name": "Evening Ride",
+            "map": {"polyline": "a~l~FqzbwOq}@it@..."},
+        }
+    )
+
+    ctx = AsyncMock(spec=Context)
+
+    # Register tools
+    register(mcp, strava)
+
+    # Retrieve and call get_activity_map tool
+    get_activity_map = mcp.tools["get_activity_map"]
+    result = await get_activity_map(ctx, 12345)
+
+    assert len(result) == 2
+    # First block is the HTML text content
+    assert result[0].type == "text"
+    assert "<!DOCTYPE html>" in result[0].text
+    assert "a~l~FqzbwOq}@it@..." in result[0].text
+    assert "unpkg.com/leaflet" in result[0].text
+
+    # Second block is the internal resource
+    assert result[1].type == "resource"
+    assert "internal://activities/12345/map" in str(result[1].resource.uri)
+
+
+@pytest.mark.asyncio
+async def test_get_activity_map_no_polyline():
+    mcp = MockMCP()
+    strava = MagicMock()
+
+    # Mock activity without GPS map info
+    strava.get_activity = AsyncMock(
+        return_value={
+            "name": "Indoor Trainer",
+            "map": {"polyline": None, "summary_polyline": None},
+        }
+    )
+
+    ctx = AsyncMock(spec=Context)
+
+    # Register tools
+    register(mcp, strava)
+
+    # Retrieve and call get_activity_map tool
+    get_activity_map = mcp.tools["get_activity_map"]
+    result = await get_activity_map(ctx, 67890)
+
+    assert len(result) == 1
+    assert result[0].type == "text"
+    assert "⚠️ Keine GPS-Routen-Daten" in result[0].text