diff --git a/scripts/generate_test_map.py b/scripts/generate_test_map.py new file mode 100644 index 0000000..20e3386 --- /dev/null +++ b/scripts/generate_test_map.py @@ -0,0 +1,201 @@ +import asyncio +import os +import sys +from dotenv import load_dotenv + +# Add src directory to path +sys.path.append(os.path.join(os.path.dirname(__file__), "..", "src")) + +from strava_mcp_server.strava_client import StravaClient +from mcp.server.fastmcp import Context +from mcp.types import TextContent + + +class MockContext(Context): + def __init__(self): + # Initialize parent with dummy request / info + pass + + async def info(self, msg): + print(f"ℹ️ [INFO] {msg}") + + async def warning(self, msg): + print(f"⚠️ [WARN] {msg}") + + async def error(self, msg): + print(f"❌ [ERROR] {msg}") + + +async def generate(): + from strava_mcp_server.config import get_config_file + + # Load user config path + config_file = get_config_file() + if os.path.exists(config_file): + load_dotenv(dotenv_path=config_file) + + # Fallback to local .env + load_dotenv() + + client_id = os.getenv("STRAVA_CLIENT_ID") + client_secret = os.getenv("STRAVA_CLIENT_SECRET") + refresh_token = os.getenv("STRAVA_REFRESH_TOKEN") + + if not all([client_id, client_secret, refresh_token]): + print( + "ℹ️ Missing Strava credentials in .env file. Generating a beautiful MOCK route (Alpe d'Huez, France) for testing!" + ) + + # A real encoded polyline representing the famous Alpe d'Huez climb in France! + mock_polyline = ( + "y~yxFn{_e@_AmAcAaBo@sAGo@Gs@H_ANc@jAyB^kALu@D_@Fk@KqBIkACe@DuABsAd@sBHi@@e@KoAKo@eAsC" + "e@oAMa@Bs@Jc@Nk@bAgBv@wAd@{AJc@Ds@Fk@GqAMmAEc@ByA@uAb@sBDi@@e@KoAKo@eAsCe@oAMa@Bs@Jc@Nk" + "@bAgBv@wAd@{AJc@Ds@Fk@GqAMmAEc@ByA@uAb@sBDi@@e@KoAKo@eAsCe@oAMa@Bs@Jc@Nk@bAgBv@wAd@{" + "AJc@Ds@Fk@GqAMmAEc@ByA@oB`@cCDq@@e@IoAKo@eAsCe@oAMi@FkADqAd@sBHi@@e@KoAKo@eAsCe@oAM" + ) + + # Import the tool function + 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 + + mcp = MockMCP() + + # Mock StravaClient that returns our mock activity and streams + class MockClient: + async def get_activity(self, activity_id): + return { + "name": "🏔️ Alpe d'Huez Climb (Mock Test)", + "map": {"polyline": mock_polyline}, + "distance": 14000, + "moving_time": 3600, + "total_elevation_gain": 1110, + "average_speed": 3.88, + "average_heartrate": 165, + "calories": 950, + } + + async def get_activity_streams(self, activity_id, keys): + # Generate Alpe d'Huez climbing curve from 740m to 1850m + return [ + {"type": "distance", "data": [i * 100 for i in range(141)]}, + { + "type": "altitude", + "data": [ + 740 + int(1110 * (i / 140) ** 0.75) for i in range(141) + ], + }, + { + "type": "latlng", + "data": [ + [45.1105 + (i * 0.0001), 6.0375 + (i * 0.0001)] + for i in range(141) + ], + }, + ] + + client = MockClient() + register(mcp, client) + + get_activity_map = mcp.tools["get_activity_map"] + ctx = MockContext() + + print("Generating mock map HTML...") + result = await get_activity_map(ctx, 99999) + + html_text = None + for item in result: + if isinstance(item, TextContent) and "" in item.text: + html_text = item.text + break + + if html_text: + output_path = os.path.join(os.path.dirname(__file__), "..", "test.html") + with open(output_path, "w", encoding="utf-8") as f: + f.write(html_text) + print(f"✅ Success! Mock map written to '{os.path.abspath(output_path)}'") + print( + "👉 Double-click 'test.html' or open it in your browser to view the interactive map!" + ) + return + + client = StravaClient() + + print("Fetching recent activities to find one with GPS data...") + try: + activities = await client.list_activities(limit=20) + gps_activity = None + for a in activities: + map_data = a.get("map", {}) + if map_data.get("summary_polyline") or map_data.get("polyline"): + gps_activity = a + break + + if not gps_activity: + print( + "❌ No activity with GPS/polyline data found in your recent 20 activities." + ) + return + + activity_id = gps_activity["id"] + print(f"Found GPS Activity: '{gps_activity.get('name')}' (ID: {activity_id})") + + # Import the tool function + 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 + + mcp = MockMCP() + register(mcp, client) + + get_activity_map = mcp.tools["get_activity_map"] + ctx = MockContext() + + print("Generating map HTML...") + result = await get_activity_map(ctx, activity_id) + + # Find the TextContent that holds the HTML + html_text = None + for item in result: + if isinstance(item, TextContent) and "" in item.text: + html_text = item.text + break + + if html_text: + output_path = os.path.join(os.path.dirname(__file__), "..", "test.html") + with open(output_path, "w", encoding="utf-8") as f: + f.write(html_text) + print(f"✅ Success! Map written to '{os.path.abspath(output_path)}'") + print( + "👉 Double-click 'test.html' or open it in your browser to view the interactive map!" + ) + else: + # Maybe it returned a warning message instead of HTML + for item in result: + if isinstance(item, TextContent): + print(f"Returned message: {item.text}") + + except Exception as e: + print(f"❌ Failed: {str(e)}") + + +if __name__ == "__main__": + asyncio.run(generate()) diff --git a/src/strava_mcp_server/tools/activities.py b/src/strava_mcp_server/tools/activities.py index b520343..c5c05ec 100644 --- a/src/strava_mcp_server/tools/activities.py +++ b/src/strava_mcp_server/tools/activities.py @@ -386,6 +386,68 @@ def register(mcp: FastMCP, strava: StravaClient) -> None: return [TextContent(type="text", text=warning_msg)] name = activity.get("name", "Aktivität") + sport_type = activity.get("sport_type") or activity.get("type", "Aktivität") + sport_emoji = get_sport_emoji(sport_type) + dist_val = f"{activity.get('distance', 0) / 1000:.2f} km" + time_val = format_duration_markdown(activity.get("moving_time")) + elev_val = f"{activity.get('total_elevation_gain', 0):.0f} m" + speed_val = format_speed_or_pace(activity.get("average_speed"), sport_type) + hr_val = ( + f"{activity.get('average_heartrate', 0):.0f} bpm" + if activity.get("average_heartrate") + else "-" + ) + cal_val = ( + f"{activity.get('calories', 0):.0f} kcal" + if activity.get("calories") + else "-" + ) + + # Fetch streams for elevation chart and hover marker correlation + altitude_data = [] + distance_data = [] + latlng_data = [] + try: + streams = await strava.get_activity_streams( + activity_id, ["distance", "altitude", "latlng"] + ) + if isinstance(streams, dict): + altitude_stream = streams.get("altitude", {}) + altitude_data = altitude_stream.get("data", []) + distance_stream = streams.get("distance", {}) + distance_data = [ + round(d / 1000, 2) for d in distance_stream.get("data", []) + ] + latlng_stream = streams.get("latlng", {}) + latlng_data = latlng_stream.get("data", []) + elif isinstance(streams, list): + for s in streams: + if isinstance(s, dict): + if s.get("type") == "altitude": + altitude_data = s.get("data", []) + elif s.get("type") == "distance": + distance_data = [ + round(d / 1000, 2) for d in s.get("data", []) + ] + elif s.get("type") == "latlng": + latlng_data = s.get("data", []) + except Exception as e: + await ctx.warning( + f"Could not fetch streams for elevation profile: {str(e)}" + ) + + # Downsample to max 150 points for smooth performance + max_points = 150 + if len(altitude_data) > max_points: + step = len(altitude_data) // max_points + altitude_data = altitude_data[::step][:max_points] + distance_data = distance_data[::step][:max_points] + if latlng_data: + latlng_data = latlng_data[::step][:max_points] + + has_elevation_chart = ( + "true" if len(altitude_data) > 0 and len(distance_data) > 0 else "false" + ) html_content = f""" @@ -394,17 +456,155 @@ def register(mcp: FastMCP, strava: StravaClient) -> None: {name} + -
+
+
+ +
+
+ +
+
+
+ 📐 Distanz + {dist_val} +
+
+ ⏱️ Zeit + {time_val} +
+
+ ⚡ Ø Pace/Geschw. + {speed_val} +
+
+ 🏔️ Höhenmeter + {elev_val} +
+
+ ❤️ Ø Herzfrequenz + {hr_val} +
+
+ 🔥 Kalorien + {cal_val} +
+
""" diff --git a/tests/unit/test_map.py b/tests/unit/test_map.py index 67efbe3..c011e60 100644 --- a/tests/unit/test_map.py +++ b/tests/unit/test_map.py @@ -26,6 +26,14 @@ async def test_get_activity_map_success(): return_value={ "name": "Evening Ride", "map": {"polyline": "a~l~FqzbwOq}@it@..."}, + "sport_type": "Ride", + } + ) + strava.get_activity_streams = AsyncMock( + return_value={ + "altitude": {"data": [100.0, 101.0, 102.0]}, + "distance": {"data": [0.0, 100.0, 200.0]}, + "latlng": {"data": [[47.1, 9.1], [47.2, 9.2], [47.3, 9.3]]}, } ) @@ -44,6 +52,11 @@ async def test_get_activity_map_success(): assert "" in result[0].text assert "a~l~FqzbwOq}@it@..." in result[0].text assert "unpkg.com/leaflet" in result[0].text + assert "latlngData" in result[0].text + assert "[47.1, 9.1]" in result[0].text + assert 'sportEmoji = "🚴"' in result[0].text + assert "L.control.layers" in result[0].text + assert "satellite" in result[0].text # Second block is the internal resource assert result[1].type == "resource"