236 lines
8.0 KiB
Python
236 lines
8.0 KiB
Python
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, EmbeddedResource
|
||
|
||
|
||
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):
|
||
if "<!DOCTYPE html>" in item.text:
|
||
html_text = item.text
|
||
break
|
||
elif '<iframe srcdoc="' in item.text:
|
||
prefix = '<iframe srcdoc="'
|
||
suffix = '" style="'
|
||
start = item.text.find(prefix)
|
||
if start != -1:
|
||
start += len(prefix)
|
||
end = item.text.find(suffix, start)
|
||
if end != -1:
|
||
html_text = item.text[start:end].replace(""", '"')
|
||
break
|
||
elif (
|
||
isinstance(item, EmbeddedResource)
|
||
and item.resource.mimeType == "text/html"
|
||
):
|
||
html_text = item.resource.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):
|
||
if "<!DOCTYPE html>" in item.text:
|
||
html_text = item.text
|
||
break
|
||
elif '<iframe srcdoc="' in item.text:
|
||
prefix = '<iframe srcdoc="'
|
||
suffix = '" style="'
|
||
start = item.text.find(prefix)
|
||
if start != -1:
|
||
start += len(prefix)
|
||
end = item.text.find(suffix, start)
|
||
if end != -1:
|
||
html_text = item.text[start:end].replace(""", '"')
|
||
break
|
||
elif (
|
||
isinstance(item, EmbeddedResource)
|
||
and item.resource.mimeType == "text/html"
|
||
):
|
||
html_text = item.resource.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())
|