feat: add interactive activity map visualization with elevation charts and update test suite to support local map generation.
This commit is contained in:
@@ -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 "<!DOCTYPE html>" 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 "<!DOCTYPE html>" 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())
|
||||
Reference in New Issue
Block a user