diff --git a/.gitea/workflows/release.yaml b/.gitea/workflows/release.yaml index 7325786..60f45bd 100644 --- a/.gitea/workflows/release.yaml +++ b/.gitea/workflows/release.yaml @@ -85,7 +85,12 @@ jobs: run: | v="${{ steps.bump.outputs.version }}" # Body = this version's section from CHANGELOG.md (fallback to a stub). - body=$(awk "/^## \\[$v\\]/{f=1;next} /^## \\[/{f=0} f" CHANGELOG.md) + # Commitizen writes headings as "## vX.Y.Z (date)", so match that form + # (not the Keep-a-Changelog "## [X.Y.Z]" brackets) and stop at the next. + # Escape dots and anchor on the trailing space so the start pattern is an + # exact version match ("## v0.3.0 " won't re-arm on "## v0.3.01 ..."). + ve=$(printf '%s' "$v" | sed 's/\./\\./g') + body=$(awk "/^## v$ve /{f=1;next} /^## v/{f=0} f" CHANGELOG.md) [ -z "$body" ] && body="Release v$v" jq -n --arg tag "v$v" --arg name "v$v" --arg body "$body" \ '{tag_name:$tag, name:$name, body:$body, draft:false, prerelease:false}' \ diff --git a/src/intervals_mcp_server/api/client.py b/src/intervals_mcp_server/api/client.py index 685d973..94334f2 100644 --- a/src/intervals_mcp_server/api/client.py +++ b/src/intervals_mcp_server/api/client.py @@ -154,7 +154,7 @@ async def make_intervals_request( api_key: str | None = None, params: dict[str, Any] | None = None, method: str = "GET", - data: dict[str, Any] | None = None, + data: dict[str, Any] | list[Any] | None = None, ) -> dict[str, Any] | list[dict[str, Any]]: """ Make a request to the Intervals.icu API with proper error handling. diff --git a/src/intervals_mcp_server/server.py b/src/intervals_mcp_server/server.py index 5efe59a..e8ee07f 100644 --- a/src/intervals_mcp_server/server.py +++ b/src/intervals_mcp_server/server.py @@ -71,6 +71,9 @@ config = get_config() # Import tool modules to register them (tools register themselves via @mcp.tool() decorators) # Import tool functions for re-export from intervals_mcp_server.tools.activities import ( # pylint: disable=wrong-import-position # noqa: E402 + get_activity_best_efforts, + get_activity_interval_stats, + search_activities, add_activity_message, get_activities, get_activity_details, @@ -87,9 +90,22 @@ from intervals_mcp_server.tools.events import ( # pylint: disable=wrong-import- ) from intervals_mcp_server.tools.gear import get_gear_list # pylint: disable=wrong-import-position # noqa: E402 from intervals_mcp_server.tools.wellness import ( # pylint: disable=wrong-import-position # noqa: E402 + get_training_readiness, get_wellness_data, update_wellness, + update_wellness_bulk, ) +from intervals_mcp_server.tools.athlete import ( # pylint: disable=wrong-import-position # noqa: E402 + get_athlete_profile, + get_athlete_summary, + get_sport_settings, + update_sport_settings, +) +from intervals_mcp_server.tools.workouts import ( # pylint: disable=wrong-import-position # noqa: E402 + get_workout, + get_workouts, +) + from intervals_mcp_server.tools.power_curves import get_athlete_power_curves # pylint: disable=wrong-import-position # noqa: E402 from intervals_mcp_server.tools.custom_items import ( # pylint: disable=wrong-import-position # noqa: E402 create_custom_item, @@ -110,6 +126,9 @@ __all__ = [ "get_activity_intervals", "get_activity_messages", "get_activity_streams", + "search_activities", + "get_activity_best_efforts", + "get_activity_interval_stats", "get_events", "get_event_by_id", "delete_event", @@ -117,6 +136,14 @@ __all__ = [ "add_or_update_event", "get_wellness_data", "update_wellness", + "update_wellness_bulk", + "get_training_readiness", + "get_athlete_profile", + "get_sport_settings", + "get_athlete_summary", + "update_sport_settings", + "get_workouts", + "get_workout", "get_gear_list", "get_athlete_power_curves", "get_custom_items", diff --git a/src/intervals_mcp_server/tools/__init__.py b/src/intervals_mcp_server/tools/__init__.py index a6b3128..8eadbfe 100644 --- a/src/intervals_mcp_server/tools/__init__.py +++ b/src/intervals_mcp_server/tools/__init__.py @@ -10,9 +10,12 @@ from mcp.server.fastmcp import FastMCP # pylint: disable=import-error # Note: Tools register themselves via @mcp.tool() decorators when imported from intervals_mcp_server.tools.activities import ( # noqa: F401 get_activities, + get_activity_best_efforts, get_activity_details, + get_activity_interval_stats, get_activity_intervals, get_activity_streams, + search_activities, ) from intervals_mcp_server.tools.events import ( # noqa: F401 add_or_update_event, @@ -33,9 +36,18 @@ from intervals_mcp_server.tools.power_curves import ( # noqa: F401 ) from intervals_mcp_server.tools.gear import get_gear_list # noqa: F401 from intervals_mcp_server.tools.wellness import ( # noqa: F401 + get_training_readiness, get_wellness_data, update_wellness, + update_wellness_bulk, ) +from intervals_mcp_server.tools.athlete import ( # noqa: F401 + get_athlete_profile, + get_athlete_summary, + get_sport_settings, + update_sport_settings, +) +from intervals_mcp_server.tools.workouts import get_workout, get_workouts # noqa: F401 def register_tools(mcp_instance: FastMCP) -> None: @@ -60,6 +72,9 @@ __all__ = [ "get_activity_details", "get_activity_intervals", "get_activity_streams", + "search_activities", + "get_activity_best_efforts", + "get_activity_interval_stats", "get_events", "get_event_by_id", "delete_event", @@ -74,4 +89,12 @@ __all__ = [ "get_gear_list", "get_wellness_data", "update_wellness", + "update_wellness_bulk", + "get_training_readiness", + "get_athlete_profile", + "get_sport_settings", + "get_athlete_summary", + "update_sport_settings", + "get_workouts", + "get_workout", ] diff --git a/src/intervals_mcp_server/tools/activities.py b/src/intervals_mcp_server/tools/activities.py index 5a54af6..52cb531 100644 --- a/src/intervals_mcp_server/tools/activities.py +++ b/src/intervals_mcp_server/tools/activities.py @@ -14,7 +14,14 @@ from intervals_mcp_server.tools.gear import ( resolve_gear_for_activity, resolve_gear_for_activities, ) -from intervals_mcp_server.utils.formatting import format_activity_message, format_activity_summary, format_intervals +from intervals_mcp_server.utils.formatting import ( + format_activity_message, + format_activity_search_results, + format_activity_summary, + format_best_efforts, + format_interval_stats, + format_intervals, +) from intervals_mcp_server.utils.validation import resolve_date_params # Import mcp instance from shared module for tool registration @@ -405,3 +412,110 @@ async def add_activity_message( if msg_id is not None: return f"Successfully added message (ID: {msg_id}) to activity {activity_id}." return f"Message appears to have been added to activity {activity_id}, but no ID was returned. Please verify manually." + + +@mcp.tool() +async def search_activities(query: str, limit: int = 20) -> str: + """Search the athlete's activities by name/keyword. + + Args: + query: Search text matched against activity name/description (required). + limit: Maximum number of results to return (default 20). + """ + try: + athlete_id_to_use, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + if not query or not query.strip(): + return "Error: a non-empty search query is required." + + params: dict[str, Any] = {"q": query.strip(), "limit": limit} + result = await make_intervals_request( + url=f"/athlete/{athlete_id_to_use}/activities/search", api_key=api_key, params=params + ) + + if isinstance(result, dict) and "error" in result: + return f"Error searching activities: {result.get('message')}" + + results = [r for r in result if isinstance(r, dict)] if isinstance(result, list) else [] + if not results: + return f"No activities found matching '{query}'." + return format_activity_search_results(results) + + +@mcp.tool() +async def get_activity_best_efforts( + activity_id: str, + stream: str = "watts", + duration: int | None = None, + distance: float | None = None, + count: int | None = None, +) -> str: + """Get the best efforts (peak values over windows) for an activity. + + Args: + activity_id: The Intervals.icu activity ID. + stream: Data stream to analyze — e.g. "watts", "heartrate", "pace" (default "watts"). + duration: Optional window duration in seconds to target. + distance: Optional window distance in meters to target. + count: Optional maximum number of efforts to return. + """ + try: + _athlete_id, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + params: dict[str, Any] = {"stream": stream} + if duration is not None: + params["duration"] = duration + if distance is not None: + params["distance"] = distance + if count is not None: + params["count"] = count + + result = await make_intervals_request( + url=f"/activity/{activity_id}/best-efforts", api_key=api_key, params=params + ) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching best efforts: {result.get('message')}" + + efforts = result.get("efforts") if isinstance(result, dict) else None + if not efforts: + return f"No best-effort data found for activity {activity_id} (stream: {stream})." + return format_best_efforts([e for e in efforts if isinstance(e, dict)], stream) + + +@mcp.tool() +async def get_activity_interval_stats(activity_id: str, start_index: int, end_index: int) -> str: + """Compute aggregate stats for an index range of an activity's data streams. + + start_index/end_index are positions in the activity's streams (as seen in the + streams or interval output). This computes metrics for that slice — it does NOT + list the activity's own intervals (use get_activity_intervals for that). + + Args: + activity_id: The Intervals.icu activity ID. + start_index: Start position in the activity streams (required). + end_index: End position in the activity streams (required, > start_index). + """ + try: + _athlete_id, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + if start_index < 0 or end_index <= start_index: + return "Error: end_index must be greater than start_index and both non-negative." + + params: dict[str, Any] = {"start_index": start_index, "end_index": end_index} + result = await make_intervals_request( + url=f"/activity/{activity_id}/interval-stats", api_key=api_key, params=params + ) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching interval stats: {result.get('message')}" + + if not isinstance(result, dict) or not result: + return f"No interval stats found for activity {activity_id} ({start_index}-{end_index})." + return format_interval_stats(result) diff --git a/src/intervals_mcp_server/tools/athlete.py b/src/intervals_mcp_server/tools/athlete.py new file mode 100644 index 0000000..e06256a --- /dev/null +++ b/src/intervals_mcp_server/tools/athlete.py @@ -0,0 +1,260 @@ +""" +Athlete-profile and configuration MCP tools for Intervals.icu. + +Read tools exposing the athlete's profile, per-sport training settings +(FTP / zones / thresholds), and training-load summaries — the context an AI +coach needs to reason about intensity and readiness. +""" + +import logging +from typing import Any + +from mcp.server.fastmcp import Context # pylint: disable=import-error +from pydantic import BaseModel # pylint: disable=import-error + +from intervals_mcp_server import credentials +from intervals_mcp_server.api.client import make_intervals_request +from intervals_mcp_server.credentials import CredentialError +from intervals_mcp_server.utils.formatting import ( + format_athlete_profile, + format_athlete_summary, + format_sport_settings, +) +from intervals_mcp_server.utils.validation import resolve_date_params + +# Import mcp instance from shared module for tool registration +from intervals_mcp_server.mcp_instance import mcp # noqa: F401 + +logger = logging.getLogger("intervals_icu_mcp_server") + + +class _ConfirmThresholdChange(BaseModel): + """Elicitation schema: the user confirms (or not) a threshold change.""" + + confirm: bool = False + + +@mcp.tool() +async def get_athlete_profile() -> str: + """Get the signed-in athlete's profile from Intervals.icu. + + Returns identity and physiology basics (name, sex, weight, resting HR, + timezone, units, location). For per-sport FTP / zones / thresholds use + get_sport_settings instead. + """ + try: + athlete_id, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + result = await make_intervals_request(url=f"/athlete/{athlete_id}", api_key=api_key) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching athlete profile: {result.get('message')}" + if not isinstance(result, dict): + return "No athlete profile found." + return format_athlete_profile(result) + + +@mcp.tool() +async def get_sport_settings(sport: str | None = None) -> str: + """Get the athlete's per-sport training settings (FTP, zones, thresholds). + + These are the values that drive load, intensity and zone calculations across + Intervals.icu. Each record's "Settings ID" is the identifier update_sport_settings + uses to target a specific sport. + + Args: + sport: Optional sport type to filter by (e.g. "Ride", "Run"). Matches the + record's sport types case-insensitively. If omitted, all sports are returned. + """ + try: + athlete_id, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + result = await make_intervals_request( + url=f"/athlete/{athlete_id}/sport-settings", api_key=api_key + ) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching sport settings: {result.get('message')}" + + records = [r for r in result if isinstance(r, dict)] if isinstance(result, list) else [] + if not records: + return "No sport settings found." + + if sport: + want = sport.strip().lower() + records = [ + r for r in records if any(want == str(t).lower() for t in (r.get("types") or [])) + ] + if not records: + return f"No sport settings found for sport '{sport}'." + + return "\n\n".join(format_sport_settings(r) for r in records) + + +@mcp.tool() +async def get_athlete_summary( + start_date: str | None = None, + end_date: str | None = None, +) -> str: + """Get a training-load summary (fitness/fatigue/form and totals) over a date range. + + Note: the underlying endpoint's ``tags`` parameter filters *athletes* (a + coach-facing feature), not activities, so no tag filter is offered here. + + Args: + start_date: Start date in YYYY-MM-DD format (optional, defaults to 30 days ago). + end_date: End date in YYYY-MM-DD format (optional, defaults to today). + """ + try: + athlete_id, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + start_date, end_date = resolve_date_params(start_date, end_date) + params: dict[str, Any] = {"start": start_date, "end": end_date} + + result = await make_intervals_request( + url=f"/athlete/{athlete_id}/athlete-summary", api_key=api_key, params=params + ) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching athlete summary: {result.get('message')}" + + if isinstance(result, list): + summaries = [s for s in result if isinstance(s, dict)] + elif isinstance(result, dict): + summaries = [result] + else: + summaries = [] + + if not summaries: + return "No summary data found for the specified date range." + + header = f"Athlete Summary ({start_date} to {end_date}):\n\n" + return header + "\n\n".join(format_athlete_summary(s) for s in summaries) + + +@mcp.tool() +async def update_sport_settings( # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals,too-many-return-statements + settings_id: int, + ftp: int | None = None, + indoor_ftp: int | None = None, + w_prime: int | None = None, + lthr: int | None = None, + max_hr: int | None = None, + threshold_pace: float | None = None, + recalc_hr_zones: bool = False, + confirm: bool = False, + ctx: Context | None = None, +) -> str: + """⚠️ Change the athlete's training thresholds (FTP, LTHR, pace) for one sport. + + These values drive ALL future load, intensity and zone calculations across + Intervals.icu. Do NOT call this speculatively — show the athlete the exact + old→new values and get their explicit approval first. + + This is a confirmed write. On clients that support MCP elicitation you will be + prompted to approve the change; otherwise you MUST pass ``confirm=True`` after the + athlete has agreed. Without confirmation the tool refuses and returns the diff. + + Args: + settings_id: The sport-settings record ID (from get_sport_settings). + ftp: New FTP in watts. + indoor_ftp: New indoor FTP in watts. + w_prime: New W' in joules. + lthr: New lactate-threshold HR in bpm. + max_hr: New max HR in bpm. + threshold_pace: New threshold pace (in the sport's pace units). + recalc_hr_zones: If True, ask Intervals.icu to recompute HR zones from the new LTHR/max HR. + confirm: Set True to confirm the change on clients without elicitation support. + """ + try: + athlete_id, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + proposed = { + "ftp": ftp, + "indoor_ftp": indoor_ftp, + "w_prime": w_prime, + "lthr": lthr, + "max_hr": max_hr, + "threshold_pace": threshold_pace, + } + if all(v is None for v in proposed.values()): + return "No settings provided. Pass at least one threshold to change." + + current_list = await make_intervals_request( + url=f"/athlete/{athlete_id}/sport-settings", api_key=api_key + ) + if isinstance(current_list, dict) and "error" in current_list: + return f"Error fetching current sport settings: {current_list.get('message')}" + records = [r for r in current_list if isinstance(r, dict)] if isinstance(current_list, list) else [] + current = next((r for r in records if r.get("id") == settings_id), None) + if current is None: + return ( + f"No sport settings found with ID {settings_id}. " + "Use get_sport_settings to list valid IDs." + ) + + changed: dict[str, Any] = {} + diff_lines: list[str] = [] + for key, value in proposed.items(): + if value is not None and current.get(key) != value: + changed[key] = value + diff_lines.append(f" {key}: {current.get(key)} -> {value}") + if not changed: + return "No changes — the provided values already match the current settings." + + sport = ", ".join(str(t) for t in (current.get("types") or [])) or f"settings {settings_id}" + diff = "\n".join(diff_lines) + + # Guardrail. If the client answers an elicitation prompt, that answer is + # authoritative: anything short of accept-with-confirm is a refusal and we + # stop WITHOUT emitting the confirm=true fallback instructions (an agentic + # client could otherwise use them to bypass the refusal it just received). + # Only when elicitation is unavailable (no ctx, or the request itself fails) + # do we fall back to requiring the explicit confirm flag. + approved = False + elicitation_answered = False + if ctx is not None: + try: + elicited = await ctx.elicit( + message=f"Update {sport} thresholds?\n{diff}", schema=_ConfirmThresholdChange + ) + elicitation_answered = True + action = getattr(elicited, "action", None) + data = getattr(elicited, "data", None) + approved = action == "accept" and bool(getattr(data, "confirm", False)) + except Exception as exc: # noqa: BLE001 - capability absent or elicitation failed + logger.warning("Elicitation unavailable, falling back to confirm flag: %s", exc) + + if elicitation_answered and not approved: + return "Sport settings unchanged — you did not confirm the change." + + if not approved and not confirm: + return ( + f"⚠️ This will change your {sport} thresholds:\n{diff}\n\n" + "These drive ALL future load / intensity / zone calculations. " + "If the athlete confirms, re-run with confirm=true." + ) + + updated = dict(current) + updated.update(changed) + result = await make_intervals_request( + url=f"/athlete/{athlete_id}/sport-settings/{settings_id}", + api_key=api_key, + method="PUT", + params={"recalcHrZones": recalc_hr_zones}, + data=updated, + ) + if isinstance(result, dict) and "error" in result: + return f"Error updating sport settings: {result.get('message')}" + + # An empty-body 200 parses to {}; render the merged record in that case. + body = result if isinstance(result, dict) and result else updated + return f"Updated {sport} settings:\n\n" + format_sport_settings(body) diff --git a/src/intervals_mcp_server/tools/wellness.py b/src/intervals_mcp_server/tools/wellness.py index bd2199f..9c80b65 100644 --- a/src/intervals_mcp_server/tools/wellness.py +++ b/src/intervals_mcp_server/tools/wellness.py @@ -4,18 +4,60 @@ Wellness-related MCP tools for Intervals.icu. This module contains tools for retrieving athlete wellness data. """ -from datetime import datetime +from datetime import datetime, timedelta from typing import Any from intervals_mcp_server import credentials from intervals_mcp_server.api.client import make_intervals_request from intervals_mcp_server.credentials import CredentialError from intervals_mcp_server.utils.formatting import format_wellness_entry +from intervals_mcp_server.utils.readiness import assess_readiness, render_readiness from intervals_mcp_server.utils.validation import resolve_date_params, validate_date # Import mcp instance from shared module for tool registration from intervals_mcp_server.mcp_instance import mcp # noqa: F401 +# snake_case tool param -> Intervals.icu camelCase wellness field. `sleep_hours` +# is handled separately (converted to sleepSecs). Shared by the single-day and +# bulk write tools so their field mapping can never drift. +_WELLNESS_FIELD_MAP: list[tuple[str, str]] = [ + ("weight", "weight"), + ("resting_hr", "restingHR"), + ("hrv", "hrv"), + ("sleep_quality", "sleepQuality"), + ("calories_consumed", "kcalConsumed"), + ("carbohydrates", "carbohydrates"), + ("protein", "protein"), + ("fat", "fatTotal"), + ("hydration_volume", "hydrationVolume"), + ("hydration_score", "hydration"), + ("soreness", "soreness"), + ("fatigue", "fatigue"), + ("stress", "stress"), + ("mood", "mood"), + ("motivation", "motivation"), + ("injury", "injury"), + ("comments", "comments"), + ("locked", "locked"), +] + + +def _wellness_payload(fields: dict[str, Any]) -> dict[str, Any]: + """Map snake_case wellness fields to the Intervals.icu camelCase payload. + + Only non-None values are included. ``sleep_hours`` becomes ``sleepSecs`` (with + -1 passing through unscaled as the clear sentinel). + """ + payload: dict[str, Any] = {} + for snake, camel in _WELLNESS_FIELD_MAP: + value = fields.get(snake) + if value is not None: + payload[camel] = value + sleep_hours = fields.get("sleep_hours") + if sleep_hours is not None: + payload["sleepSecs"] = -1 if sleep_hours == -1 else int(sleep_hours * 3600) + return payload + @mcp.tool() async def get_wellness_data( @@ -136,35 +178,29 @@ async def update_wellness( # pylint: disable=too-many-arguments,too-many-positi except ValueError as exc: return f"Error: {exc}" - # Sleep is passed in hours but stored as seconds; -1 is the clear sentinel and - # must pass through unscaled. - sleep_secs: int | None = None - if sleep_hours is not None: - sleep_secs = -1 if sleep_hours == -1 else int(sleep_hours * 3600) - - # Map snake_case tool params to the Intervals.icu camelCase wellness fields. - field_map: list[tuple[str, Any]] = [ - ("weight", weight), - ("restingHR", resting_hr), - ("hrv", hrv), - ("sleepSecs", sleep_secs), - ("sleepQuality", sleep_quality), - ("kcalConsumed", calories_consumed), - ("carbohydrates", carbohydrates), - ("protein", protein), - ("fatTotal", fat), - ("hydrationVolume", hydration_volume), - ("hydration", hydration_score), - ("soreness", soreness), - ("fatigue", fatigue), - ("stress", stress), - ("mood", mood), - ("motivation", motivation), - ("injury", injury), - ("comments", comments), - ("locked", locked), - ] - payload: dict[str, Any] = {k: v for k, v in field_map if v is not None} + payload = _wellness_payload( + { + "weight": weight, + "resting_hr": resting_hr, + "hrv": hrv, + "sleep_hours": sleep_hours, + "sleep_quality": sleep_quality, + "calories_consumed": calories_consumed, + "carbohydrates": carbohydrates, + "protein": protein, + "fat": fat, + "hydration_volume": hydration_volume, + "hydration_score": hydration_score, + "soreness": soreness, + "fatigue": fatigue, + "stress": stress, + "mood": mood, + "motivation": motivation, + "injury": injury, + "comments": comments, + "locked": locked, + } + ) if not payload: return "No wellness fields provided. Pass at least one field to update." @@ -187,3 +223,121 @@ async def update_wellness( # pylint: disable=too-many-arguments,too-many-positi result["date"] = date return f"Updated wellness for {date}:\n\n" + format_wellness_entry(result) return f"Updated wellness for {date}." + + +@mcp.tool() +async def update_wellness_bulk(entries: list[dict[str, Any]]) -> str: + """Create or update multiple days of wellness data in a single call. + + Writes to PUT /athlete/{id}/wellness-bulk. Each entry is a dict with a ``date`` + (YYYY-MM-DD) plus any of the same fields as update_wellness: weight, resting_hr, + hrv, sleep_hours, sleep_quality, calories_consumed, carbohydrates, protein, fat, + hydration_volume, hydration_score, soreness, fatigue, stress, mood, motivation, + injury, comments, locked. Pass -1 to clear a numeric field. Every date is + validated up front — if any entry is invalid the whole batch is rejected, so + there are no partial writes. + + Args: + entries: List of per-day wellness dicts, each with a ``date`` and one or more fields. + """ + try: + athlete_id_to_use, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + if not entries: + return "No entries provided. Pass at least one day to update." + if len(entries) > 92: + return f"Too many entries ({len(entries)}). Limit a bulk update to 92 days." + + # Recognized entry keys: the shared snake_case field names plus date/sleep_hours. + # Anything else (e.g. API-style camelCase like "restingHR") is rejected rather + # than silently dropped — otherwise values the caller asked to record would be + # lost behind a success message. + allowed_keys = {snake for snake, _ in _WELLNESS_FIELD_MAP} | {"date", "sleep_hours"} + + records: list[dict[str, Any]] = [] + summaries: list[str] = [] + for i, entry in enumerate(entries): + if not isinstance(entry, dict): + return f"Error: entry {i} is not an object." + raw_date = entry.get("date") + if not raw_date: + return f"Error: entry {i} is missing a 'date'." + try: + date = validate_date(str(raw_date)) + except ValueError as exc: + return f"Error in entry {i}: {exc}" + + unknown = sorted(set(entry) - allowed_keys) + if unknown: + return ( + f"Error: entry {i} ({date}) has unrecognized field(s): {', '.join(unknown)}. " + f"Valid fields: {', '.join(sorted(allowed_keys - {'date'}))}." + ) + + payload = _wellness_payload(entry) + if not payload: + return f"Error: entry {i} ({date}) has no wellness fields to update." + payload["id"] = date + records.append(payload) + summaries.append(f"{date}: {', '.join(k for k in payload if k != 'id')}") + + result = await make_intervals_request( + url=f"/athlete/{athlete_id_to_use}/wellness-bulk", + api_key=api_key, + method="PUT", + data=records, + ) + + if isinstance(result, dict) and "error" in result: + return f"Error updating wellness data: {result.get('message')}" + + return f"Updated {len(records)} day(s):\n" + "\n".join(summaries) + + +@mcp.tool() +async def get_training_readiness(days: int = 45) -> str: + """Assess training readiness from recent wellness data. + + Synthesizes the athlete's recent wellness history into a readiness read: + HRV-guided (7-day rolling lnRMSSD vs baseline +/- smallest worthwhile change), + resting-HR and sleep trends, and subjective inputs (soreness/fatigue/stress/ + mood/motivation). When there is too little data — notably fewer than ~2 weeks + of HRV — the verdict is withheld rather than guessed, and the report lists which + signals it could and could not use. + + Args: + days: How many days of history to analyze (default 45; minimum 14 is enforced). + """ + try: + athlete_id_to_use, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + end = datetime.now() + start = end - timedelta(days=max(days, 14)) + params = {"oldest": start.strftime("%Y-%m-%d"), "newest": end.strftime("%Y-%m-%d")} + + result = await make_intervals_request( + url=f"/athlete/{athlete_id_to_use}/wellness", api_key=api_key, params=params + ) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching wellness data: {result.get('message')}" + + records: list[dict[str, Any]] = [] + if isinstance(result, dict): + for date_str, data in result.items(): + if isinstance(data, dict): + data.setdefault("id", date_str) + records.append(data) + elif isinstance(result, list): + records = [r for r in result if isinstance(r, dict)] + + if not records: + return "No wellness data found to assess readiness." + + # Anchor the calendar windows on today so weeks-old data reads as "no recent + # data" rather than being presented as the athlete's current state. + return render_readiness(assess_readiness(records, reference_date=end.strftime("%Y-%m-%d"))) diff --git a/src/intervals_mcp_server/tools/workouts.py b/src/intervals_mcp_server/tools/workouts.py new file mode 100644 index 0000000..a33a5a8 --- /dev/null +++ b/src/intervals_mcp_server/tools/workouts.py @@ -0,0 +1,72 @@ +""" +Workout-library MCP tools for Intervals.icu. + +Read tools exposing the athlete's reusable workout library (distinct from the +calendar *events* handled in tools/events.py). +""" + +from intervals_mcp_server import credentials +from intervals_mcp_server.api.client import make_intervals_request +from intervals_mcp_server.credentials import CredentialError +from intervals_mcp_server.utils.formatting import format_workout_details, format_workout_summary + +# Import mcp instance from shared module for tool registration +from intervals_mcp_server.mcp_instance import mcp # noqa: F401 + + +@mcp.tool() +async def get_workouts(folder_id: int | None = None, sport_type: str | None = None) -> str: + """List the athlete's reusable workout library. + + Filtering is applied client-side (the API returns the full library). + + Args: + folder_id: Optional folder ID to restrict results to one folder. + sport_type: Optional sport type to filter by (e.g. "Ride", "Run"). + """ + try: + athlete_id, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + result = await make_intervals_request(url=f"/athlete/{athlete_id}/workouts", api_key=api_key) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching workouts: {result.get('message')}" + + workouts = [w for w in result if isinstance(w, dict)] if isinstance(result, list) else [] + if folder_id is not None: + workouts = [w for w in workouts if w.get("folder_id") == folder_id] + if sport_type: + want = sport_type.strip().lower() + workouts = [w for w in workouts if str(w.get("type", "")).lower() == want] + + if not workouts: + return "No workouts found." + + lines = [f"Workout Library ({len(workouts)}):", ""] + lines.extend(format_workout_summary(w) for w in workouts) + return "\n".join(lines) + + +@mcp.tool() +async def get_workout(workout_id: int) -> str: + """Get a single library workout's full structure (steps and targets). + + Args: + workout_id: The Intervals.icu workout ID. + """ + try: + athlete_id, api_key = await credentials.resolve_caller_credentials() + except CredentialError as exc: + return str(exc) + + result = await make_intervals_request( + url=f"/athlete/{athlete_id}/workouts/{workout_id}", api_key=api_key + ) + + if isinstance(result, dict) and "error" in result: + return f"Error fetching workout: {result.get('message')}" + if not isinstance(result, dict) or not result: + return f"No workout found with ID {workout_id}." + return format_workout_details(result) diff --git a/src/intervals_mcp_server/utils/formatting.py b/src/intervals_mcp_server/utils/formatting.py index 3e93efe..3833bb8 100644 --- a/src/intervals_mcp_server/utils/formatting.py +++ b/src/intervals_mcp_server/utils/formatting.py @@ -420,6 +420,146 @@ def format_wellness_entry(entries: dict[str, Any], include_all_fields: bool = Fa return "\n".join(lines) +def format_athlete_profile(athlete: dict[str, Any]) -> str: + """Format an athlete profile into a readable string. + + Renders identity/physiology basics. The embedded per-sport settings blob is + only summarised (a count) — get_sport_settings renders the detail. + """ + name = athlete.get("name") or " ".join( + p for p in [athlete.get("firstname"), athlete.get("lastname")] if p + ) or "Unknown" + lines = ["Athlete Profile:", "", f"Name: {name}", f"ID: {athlete.get('id', 'N/A')}"] + + weight = athlete.get("weight") + if weight is None: + weight = athlete.get("icu_weight") + for label, value, unit in [ + ("Sex", athlete.get("sex"), ""), + ("Date of Birth", athlete.get("icu_date_of_birth"), ""), + ("Weight", weight, "kg"), + ("Resting HR", athlete.get("icu_resting_hr"), "bpm"), + ("Timezone", athlete.get("timezone"), ""), + ("Units", athlete.get("measurement_preference"), ""), + ]: + if value is not None and value != "": + lines.append(f"{label}: {value}{(' ' + unit) if unit else ''}") + + location = ", ".join( + p for p in [athlete.get("city"), athlete.get("state"), athlete.get("country")] if p + ) + if location: + lines.append(f"Location: {location}") + if athlete.get("icu_coach"): + lines.append("Role: Coach") + if athlete.get("bio"): + lines.append(f"Bio: {athlete['bio']}") + + sport_settings = athlete.get("sportSettings") or athlete.get("icu_type_settings") + if isinstance(sport_settings, list) and sport_settings: + lines.append( + f"Sport Settings: {len(sport_settings)} sport(s) configured " + "(use get_sport_settings for FTP/zones/thresholds)" + ) + return "\n".join(lines) + + +def _format_zone_line(label: str, zones: Any, names: Any, unit: str = "") -> str | None: + """Render a zone-boundary array, pairing with names when they line up.""" + if not isinstance(zones, list) or not zones: + return None + if isinstance(names, list) and len(names) == len(zones): + parts = [f"{n}: {z}{unit}" for n, z in zip(names, zones, strict=True)] + else: + parts = [f"{z}{unit}" for z in zones] + return f"{label}: " + ", ".join(parts) + + +def format_sport_settings(settings: dict[str, Any]) -> str: + """Format one per-sport settings record (FTP, zones, thresholds) into text. + + The record ``id`` is always shown — it is the identifier update_sport_settings + needs to target a specific sport's settings. + """ + types = settings.get("types") or [] + sport = ", ".join(str(t) for t in types) if types else "Unknown" + lines = [f"Sport Settings — {sport}:", f"Settings ID: {settings.get('id', 'N/A')}"] + + power_bits = [] + for key, label, unit in [ + ("ftp", "FTP", "W"), + ("indoor_ftp", "Indoor FTP", "W"), + ("w_prime", "W'", "J"), + ("p_max", "Pmax", "W"), + ]: + if settings.get(key) is not None: + power_bits.append(f"{label}: {settings[key]}{unit}") + if power_bits: + lines += ["", "Power:"] + [f"- {b}" for b in power_bits] + zone_line = _format_zone_line("Zones", settings.get("power_zones"), settings.get("power_zone_names")) + if zone_line: + lines.append(f"- {zone_line}") + + hr_bits = [] + for key, label in [("lthr", "LTHR"), ("max_hr", "Max HR")]: + if settings.get(key) is not None: + hr_bits.append(f"{label}: {settings[key]} bpm") + if hr_bits: + lines += ["", "Heart Rate:"] + [f"- {b}" for b in hr_bits] + zone_line = _format_zone_line("Zones", settings.get("hr_zones"), settings.get("hr_zone_names")) + if zone_line: + lines.append(f"- {zone_line}") + + if settings.get("threshold_pace") is not None: + units = settings.get("pace_units", "") + lines += ["", "Pace:", f"- Threshold: {settings['threshold_pace']} {units}".rstrip()] + zone_line = _format_zone_line("Zones", settings.get("pace_zones"), settings.get("pace_zone_names")) + if zone_line: + lines.append(f"- {zone_line}") + + defaults = [] + for key, label in [("warmup_time", "Warmup"), ("cooldown_time", "Cooldown")]: + if settings.get(key) is not None: + defaults.append(f"{label}: {settings[key]}s") + if defaults: + lines += ["", "Defaults: " + ", ".join(defaults)] + return "\n".join(lines) + + +def format_athlete_summary(summary: dict[str, Any]) -> str: + """Format a training-load summary (fitness/fatigue/form + totals) into text.""" + lines: list[str] = [] + if summary.get("date"): + lines.append(f"Period ending {summary['date']}:") + for key, label, unit in [ + ("count", "Activities", ""), + ("moving_time", "Moving Time", "s"), + ("distance", "Distance", "m"), + ("total_elevation_gain", "Elevation Gain", "m"), + ("training_load", "Training Load", ""), + ("calories", "Calories", "kcal"), + ("fitness", "Fitness (CTL)", ""), + ("fatigue", "Fatigue (ATL)", ""), + ("form", "Form (TSB)", ""), + ("rampRate", "Ramp Rate", ""), + ("eftp", "eFTP", "W"), + ]: + if summary.get(key) is not None: + lines.append(f"- {label}: {summary[key]}{(' ' + unit) if unit else ''}") + + categories = summary.get("byCategory") + if isinstance(categories, list) and categories: + lines.append("By category:") + for cat in categories: + if not isinstance(cat, dict): + continue + lines.append( + f" - {cat.get('category', '?')}: {cat.get('count', 0)} activities, " + f"load {cat.get('training_load', 'N/A')}, {cat.get('moving_time', 'N/A')}s" + ) + return "\n".join(lines) if lines else "No summary metrics available." + + def format_event_summary(event: dict[str, Any]) -> str: """Format a basic event summary into a readable string.""" @@ -676,3 +816,163 @@ def format_power_curves( lines.append("") return "\n".join(lines) + + +def format_activity_search_results(results: list[dict[str, Any]]) -> str: + """Format activity search hits into a compact one-line-per-result list.""" + lines = [f"Found {len(results)} activit{'y' if len(results) == 1 else 'ies'}:", ""] + for r in results: + date = r.get("start_date_local", "") + if isinstance(date, str) and len(date) > 10: + date = date[:10] + extra = [] + if r.get("distance") is not None: + extra.append(f"{r['distance']}m") + if r.get("moving_time") is not None: + extra.append(f"{r['moving_time']}s") + if r.get("race"): + extra.append("RACE") + line = " | ".join([date or "?", str(r.get("type", "?")), str(r.get("name", "Unnamed"))]) + if extra: + line += " (" + ", ".join(extra) + ")" + line += f" [id: {r.get('id', 'N/A')}]" + lines.append(line) + return "\n".join(lines) + + +def format_best_efforts(efforts: list[dict[str, Any]], stream: str) -> str: + """Format best-effort windows for a stream (power/hr/pace) into text.""" + lines = [f"Best Efforts ({stream}):", ""] + for e in efforts: + parts = [] + if e.get("duration") is not None: + parts.append(_format_duration_label(int(e["duration"]))) + if e.get("distance") is not None: + parts.append(f"{e['distance']}m") + label = " / ".join(parts) if parts else "effort" + idx = f"[idx {e.get('start_index', '?')}-{e.get('end_index', '?')}]" + lines.append(f"- {label}: avg {e.get('average', 'N/A')} {idx}") + return "\n".join(lines) + + +def format_interval_stats(interval: dict[str, Any]) -> str: + """Format a computed interval-stats block (a single Interval) into text.""" + lines = ["Interval Stats:", ""] + for key, label, unit in [ + ("moving_time", "Moving Time", "s"), + ("distance", "Distance", "m"), + ("average_watts", "Avg Power", "W"), + ("weighted_average_watts", "Weighted Avg Power", "W"), + ("max_watts", "Max Power", "W"), + ("average_watts_kg", "Avg Power", "W/kg"), + ("intensity", "Intensity", ""), + ("training_load", "Training Load", ""), + ("joules", "Work", "J"), + ("decoupling", "Decoupling", "%"), + ("average_heartrate", "Avg HR", "bpm"), + ("max_heartrate", "Max HR", "bpm"), + ("average_cadence", "Avg Cadence", "rpm"), + ("average_speed", "Avg Speed", "m/s"), + ("gap", "GAP", "m/s"), + ]: + if interval.get(key) is not None: + lines.append(f"- {label}: {interval[key]}{(' ' + unit) if unit else ''}") + return "\n".join(lines) + + +def _format_step_intensity(step: dict[str, Any]) -> str: + """Render a workout step's intensity target(s) from raw workout_doc JSON.""" + bits = [] + for key, label in [("power", ""), ("hr", "HR"), ("pace", "Pace"), ("cadence", "Cad")]: + v = step.get(key) + if not isinstance(v, dict): + continue + units = v.get("units", "") + if v.get("start") is not None and v.get("end") is not None: + val = f"{v['start']}-{v['end']}" + elif v.get("value") is not None: + val = f"{v['value']}" + else: + continue + bits.append(f"{(label + ' ') if label else ''}{val}{units}") + return ", ".join(bits) + + +def _format_workout_step(step: dict[str, Any], depth: int = 0) -> list[str]: + """Recursively render one workout_doc step (handles repeat blocks). Depth-capped.""" + indent = " " * (depth + 1) + if depth > 6: + return [f"{indent}- ...(nested too deep)"] + reps = step.get("reps") + substeps = step.get("steps") + if reps and isinstance(substeps, list): + lines = [f"{indent}{reps}x:"] + for sub in substeps: + if isinstance(sub, dict): + lines.extend(_format_workout_step(sub, depth + 1)) + return lines + parts = [] + if step.get("duration") is not None: + parts.append(_format_duration_label(int(step["duration"]))) + if step.get("distance") is not None: + parts.append(f"{step['distance']}m") + tag = "" + if step.get("warmup"): + tag = " (warmup)" + elif step.get("cooldown"): + tag = " (cooldown)" + elif step.get("freeride"): + tag = " (free ride)" + intensity = _format_step_intensity(step) + if step.get("ramp") and intensity: + intensity = "ramp " + intensity + label = " ".join(parts) if parts else "step" + detail = f" @ {intensity}" if intensity else "" + text = step.get("text") + return [f"{indent}- {label}{detail}{tag}{(' — ' + text) if text else ''}"] + + +def format_workout_summary(workout: dict[str, Any]) -> str: + """Format one library workout as a compact one-line list entry.""" + line = " | ".join([str(workout.get("name", "Unnamed")), str(workout.get("type", "?"))]) + extra = [] + if workout.get("icu_training_load") is not None: + extra.append(f"load {workout['icu_training_load']}") + if workout.get("moving_time") is not None: + extra.append(f"{workout['moving_time']}s") + if workout.get("folder_id") is not None: + extra.append(f"folder {workout['folder_id']}") + if extra: + line += " (" + ", ".join(extra) + ")" + return line + f" [id: {workout.get('id', 'N/A')}]" + + +def format_workout_details(workout: dict[str, Any]) -> str: + """Format a library workout in full, including its structured steps.""" + lines = [f"Workout: {workout.get('name', 'Unnamed')}", f"ID: {workout.get('id', 'N/A')}"] + for key, label, unit in [ + ("type", "Type", ""), + ("sub_type", "Sub-type", ""), + ("indoor", "Indoor", ""), + ("moving_time", "Duration", "s"), + ("distance", "Distance", "m"), + ("icu_training_load", "Training Load", ""), + ("icu_intensity", "Intensity", ""), + ("carbs_per_hour", "Carbs", "g/hr"), + ("folder_id", "Folder", ""), + ]: + if workout.get(key) is not None: + lines.append(f"{label}: {workout[key]}{(' ' + unit) if unit else ''}") + if workout.get("description"): + lines.append(f"Description: {workout['description']}") + tags = workout.get("tags") + if isinstance(tags, list) and tags: + lines.append("Tags: " + ", ".join(str(t) for t in tags)) + + doc = workout.get("workout_doc") + if isinstance(doc, dict) and isinstance(doc.get("steps"), list) and doc["steps"]: + lines += ["", "Steps:"] + for step in doc["steps"]: + if isinstance(step, dict): + lines.extend(_format_workout_step(step)) + return "\n".join(lines) diff --git a/src/intervals_mcp_server/utils/readiness.py b/src/intervals_mcp_server/utils/readiness.py new file mode 100644 index 0000000..dbc1264 --- /dev/null +++ b/src/intervals_mcp_server/utils/readiness.py @@ -0,0 +1,330 @@ +""" +Training-readiness computation for Intervals.icu wellness data. + +Pure functions (no I/O) so they can be unit-tested on fixtures. The HRV method +follows Plews & Laursen: a rolling mean of ``ln(rMSSD)`` over the last 7 calendar +days compared to a baseline from the preceding ~30 days, with a "normal" band of +baseline mean +/- the smallest worthwhile change (SWC = 0.5 x baseline SD, with a +floor so a near-constant baseline can't produce a zero-width band). Resting HR, +sleep and subjective inputs are each compared to their own recent baseline. + +All windows are **calendar-based**, anchored on ``reference_date`` (callers should +pass today): a metric whose samples are older than the window reports "no recent +data" instead of silently treating stale samples as current. Nothing is +fabricated: a metric with too little data in its window reports "no data" rather +than defaulting, and the overall verdict is withheld (not guessed) when the +objective signals are too sparse to be meaningful. Subjective fields use the +conventional Intervals.icu direction (soreness/fatigue/stress/injury: higher is +worse; mood/motivation: higher is better) and only ever contribute a soft +warning, never a hard alert. +""" + +from __future__ import annotations + +import math +import statistics +from datetime import date, timedelta +from typing import Any + +_RECENT_DAYS = 7 +_BASELINE_DAYS = 30 +_MIN_HRV_RECENT = 4 # samples needed inside the 7-day window +_MIN_HRV_BASELINE = 7 +_MIN_RHR_RECENT = 4 +_MIN_RHR_BASELINE = 5 +_MIN_SLEEP_BASELINE = 5 +_MIN_SUBJ_BASELINE = 5 +_SUBJ_LATEST_MAX_AGE = 3 # days; older subjective entries aren't "current" feelings + +# Floor for the HRV smallest-worthwhile-change band, in ln(rMSSD) units. A +# near-constant baseline (coarsely-rounded device output, very steady athlete) +# would otherwise give SWC ~= 0 and flag trivial fluctuations as alerts. 0.05 ln +# units is ~5% in rMSSD — on the order of normal day-to-day variation. +_SWC_FLOOR = 0.05 + +_SUBJ_WORSE_HIGH = ("soreness", "fatigue", "stress", "injury") +_SUBJ_WORSE_LOW = ("motivation", "mood") + + +def _parse_date(value: Any) -> date | None: + try: + return date.fromisoformat(str(value)[:10]) + except (ValueError, TypeError): + return None + + +def _dated_series( + records: list[dict[str, Any]], key: str, positive: bool = False +) -> list[tuple[date, float]]: + """Date-sorted ``(date, value)`` pairs for ``key``; undated/non-numeric skipped.""" + out: list[tuple[date, float]] = [] + for r in records: + if not isinstance(r, dict): + continue + d = _parse_date(r.get("id") or r.get("date")) + v = r.get(key) + if d is None or not isinstance(v, (int, float)) or isinstance(v, bool): + continue + if positive and v <= 0: + continue + out.append((d, float(v))) + out.sort(key=lambda p: p[0]) + return out + + +def _windows( + pairs: list[tuple[date, float]], ref: date +) -> tuple[list[float], list[float]]: + """Split values into recent (last 7 calendar days) and baseline (30 before that).""" + recent_start = ref - timedelta(days=_RECENT_DAYS) + baseline_start = recent_start - timedelta(days=_BASELINE_DAYS) + recent = [v for d, v in pairs if recent_start < d <= ref] + baseline = [v for d, v in pairs if baseline_start < d <= recent_start] + return recent, baseline + + +def _newest_date(records: list[dict[str, Any]]) -> date | None: + dates = [ + d + for d in (_parse_date(r.get("id") or r.get("date")) for r in records if isinstance(r, dict)) + if d is not None + ] + return max(dates) if dates else None + + +def _resolve_ref(records: list[dict[str, Any]], reference_date: str | None) -> date | None: + return _parse_date(reference_date) if reference_date else _newest_date(records) + + +def hrv_signal(records: list[dict[str, Any]], reference_date: str | None = None) -> dict[str, Any]: + """HRV readiness via 7-day rolling lnRMSSD vs baseline band (mean +/- SWC).""" + pairs = _dated_series(records, "hrv", positive=True) + ref = _resolve_ref(records, reference_date) + if ref is None or not pairs: + return {"name": "HRV", "level": "nodata", "detail": "no HRV data"} + recent_vals, baseline_vals = _windows(pairs, ref) + if len(recent_vals) < _MIN_HRV_RECENT: + return { + "name": "HRV", + "level": "nodata", + "detail": f"only {len(recent_vals)} HRV sample(s) in the last {_RECENT_DAYS} days", + } + if len(baseline_vals) < _MIN_HRV_BASELINE: + return { + "name": "HRV", + "level": "nodata", + "detail": f"only {len(baseline_vals)} baseline day(s) — need >= {_MIN_HRV_BASELINE}", + } + recent_mean = statistics.mean(math.log(v) for v in recent_vals) + ln_base = [math.log(v) for v in baseline_vals] + base_mean = statistics.mean(ln_base) + swc = max(0.5 * statistics.pstdev(ln_base), _SWC_FLOOR) + if recent_mean < base_mean - swc: + return { + "name": "HRV", + "level": "alert", + "detail": "7-day lnHRV below baseline band — parasympathetic suppression", + } + if recent_mean > base_mean + swc: + return { + "name": "HRV", + "level": "warn", + "detail": "7-day lnHRV above baseline band — super-compensation, " + "or saturation if resting HR is also elevated", + } + return {"name": "HRV", "level": "ok", "detail": "7-day lnHRV within normal band"} + + +def rhr_signal(records: list[dict[str, Any]], reference_date: str | None = None) -> dict[str, Any]: + """Resting-HR readiness: 7-day mean vs a disjoint 30-day baseline, flag if >5% above.""" + pairs = _dated_series(records, "restingHR", positive=True) + ref = _resolve_ref(records, reference_date) + if ref is None or not pairs: + return {"name": "Resting HR", "level": "nodata", "detail": "no resting-HR data"} + recent_vals, baseline_vals = _windows(pairs, ref) + if len(recent_vals) < _MIN_RHR_RECENT: + return { + "name": "Resting HR", + "level": "nodata", + "detail": f"only {len(recent_vals)} RHR sample(s) in the last {_RECENT_DAYS} days", + } + if len(baseline_vals) < _MIN_RHR_BASELINE: + return { + "name": "Resting HR", + "level": "nodata", + "detail": f"only {len(baseline_vals)} baseline day(s) of RHR — need >= {_MIN_RHR_BASELINE}", + } + recent = statistics.mean(recent_vals) + base = statistics.mean(baseline_vals) + if base > 0 and (recent - base) / base > 0.05: + return { + "name": "Resting HR", + "level": "warn", + "detail": f"7-day RHR {recent:.0f} is >5% above baseline {base:.0f}", + } + return { + "name": "Resting HR", + "level": "ok", + "detail": f"7-day RHR {recent:.0f} near baseline {base:.0f}", + } + + +def sleep_signal(records: list[dict[str, Any]], reference_date: str | None = None) -> dict[str, Any]: + """Sleep readiness: last night (dated within a day of reference) vs baseline mean.""" + pairs = _dated_series(records, "sleepSecs", positive=True) + ref = _resolve_ref(records, reference_date) + if ref is None or not pairs: + return {"name": "Sleep", "level": "nodata", "detail": "no sleep data"} + last_date, last_secs = pairs[-1] + if (ref - last_date).days > 1: + return { + "name": "Sleep", + "level": "nodata", + "detail": f"no sleep logged since {last_date.isoformat()}", + } + baseline = [ + v / 3600 + for d, v in pairs + if d != last_date and ref - timedelta(days=_BASELINE_DAYS) < d <= ref + ] + if len(baseline) < _MIN_SLEEP_BASELINE: + return {"name": "Sleep", "level": "nodata", "detail": "not enough sleep data"} + last = last_secs / 3600 + mean = statistics.mean(baseline) + if mean > 0 and last < 0.85 * mean: + return { + "name": "Sleep", + "level": "warn", + "detail": f"last night {last:.1f}h below baseline {mean:.1f}h", + } + return { + "name": "Sleep", + "level": "ok", + "detail": f"last night {last:.1f}h near baseline {mean:.1f}h", + } + + +def subjective_signals( + records: list[dict[str, Any]], reference_date: str | None = None +) -> list[dict[str, Any]]: + """Soft warnings when a *current* subjective field has moved off baseline for the worse.""" + signals: list[dict[str, Any]] = [] + ref = _resolve_ref(records, reference_date) + if ref is None: + return signals + fields = [(f, True) for f in _SUBJ_WORSE_HIGH] + [(f, False) for f in _SUBJ_WORSE_LOW] + for field, worse_high in fields: + pairs = _dated_series(records, field) + if not pairs: + continue + latest_date, latest = pairs[-1] + if (ref - latest_date).days > _SUBJ_LATEST_MAX_AGE: + continue # stale entries aren't current feelings + baseline = [ + v + for d, v in pairs + if d != latest_date and ref - timedelta(days=_BASELINE_DAYS) < d <= ref + ] + if len(baseline) < _MIN_SUBJ_BASELINE: + continue + mean = statistics.mean(baseline) + sd = statistics.pstdev(baseline) if len(baseline) > 1 else 0.0 + threshold = max(sd, 0.5) # require a meaningful move, not noise + worse = (latest - mean > threshold) if worse_high else (mean - latest > threshold) + if worse: + direction = "elevated" if worse_high else "low" + signals.append( + { + "name": field.capitalize(), + "level": "warn", + "detail": f"{field} {direction} vs baseline ({latest:g} vs {mean:.1f})", + } + ) + return signals + + +def form_context(records: list[dict[str, Any]]) -> dict[str, Any] | None: + """Latest Form (TSB = CTL - ATL) if both components are present.""" + dated = sorted( + [r for r in records if isinstance(r, dict)], + key=lambda r: str(r.get("id") or r.get("date") or ""), + ) + if not dated: + return None + latest = dated[-1] + ctl, atl = latest.get("ctl"), latest.get("atl") + if isinstance(ctl, (int, float)) and isinstance(atl, (int, float)): + return {"form": round(ctl - atl, 1), "ctl": ctl, "atl": atl} + return None + + +def assess_readiness( + records: list[dict[str, Any]], reference_date: str | None = None +) -> dict[str, Any]: + """Produce a structured readiness assessment from wellness records. + + ``reference_date`` (YYYY-MM-DD) anchors the calendar windows — pass today so + stale data reads as "no recent data" instead of masquerading as current. If + omitted, the newest record's date is used (fixture-friendly, but blind to + how old that record is). + """ + core = [ + hrv_signal(records, reference_date), + rhr_signal(records, reference_date), + sleep_signal(records, reference_date), + ] + signals = core + subjective_signals(records, reference_date) + + alerts = [s for s in signals if s["level"] == "alert"] + warns = [s for s in signals if s["level"] == "warn"] + core_with_data = [s for s in core if s["level"] != "nodata"] + + if core[0]["level"] == "nodata" and len(core_with_data) < 2: + verdict = "insufficient" + elif alerts or len(warns) >= 3: + verdict = "red" + elif warns: + verdict = "amber" + else: + verdict = "green" + + return { + "verdict": verdict, + "signals": signals, + "form": form_context(records), + "days": len(records), + } + + +_VERDICT_LABEL = { + "green": "🟢 Ready — signals within normal range", + "amber": "🟡 Caution — one or more signals off baseline", + "red": "🔴 Compromised — strong or multiple negative signals", + "insufficient": "⚪ Verdict withheld — not enough data to judge", +} +_LEVEL_ICON = {"ok": "✓", "warn": "!", "alert": "‼", "nodata": "·"} + + +def render_readiness(assessment: dict[str, Any]) -> str: + """Render a readiness assessment into a plain-language report.""" + lines = [ + "Training Readiness:", + "", + _VERDICT_LABEL.get(assessment["verdict"], assessment["verdict"]), + f"(based on {assessment['days']} day(s) of wellness data)", + "", + "Signals:", + ] + for s in assessment["signals"]: + lines.append(f" {_LEVEL_ICON.get(s['level'], '-')} {s['name']}: {s['detail']}") + + form = assessment["form"] + if form: + lines += ["", f"Form (TSB): {form['form']} (CTL {form['ctl']} / ATL {form['atl']})"] + + if assessment["verdict"] == "insufficient": + lines += [ + "", + "Log daily HRV (and resting HR) for ~2+ weeks to enable a readiness verdict.", + ] + return "\n".join(lines) diff --git a/tests/test_activity_analytics.py b/tests/test_activity_analytics.py new file mode 100644 index 0000000..ef79866 --- /dev/null +++ b/tests/test_activity_analytics.py @@ -0,0 +1,154 @@ +""" +Tests for the 0.3.0 activity search + analytics tools in +intervals_mcp_server.tools.activities: search_activities, +get_activity_best_efforts, get_activity_interval_stats. + +HTTP is stubbed at the module level; the autouse conftest fixture supplies the +caller credentials (athlete ``i1``). +""" + +import asyncio + +from intervals_mcp_server.tools import activities + + +def _patch_request(monkeypatch, result): + calls: list[dict] = [] + + async def fake(**kwargs): + calls.append(kwargs) + return result + + monkeypatch.setattr(activities, "make_intervals_request", fake) + return calls + + +# --------------------------------------------------------------------------- # +# search_activities +# --------------------------------------------------------------------------- # +SEARCH_HITS = [ + { + "id": "a1", + "name": "Threshold intervals", + "start_date_local": "2026-07-18T07:00:00", + "type": "Ride", + "distance": 42000, + "moving_time": 5400, + "race": False, + }, + {"id": "a2", "name": "Local crit", "start_date_local": "2026-07-15", "type": "Ride", "race": True}, +] + + +def test_search_activities_success(monkeypatch): + calls = _patch_request(monkeypatch, SEARCH_HITS) + out = asyncio.run(activities.search_activities("threshold", limit=5)) + assert calls[0]["url"] == "/athlete/i1/activities/search" + assert calls[0]["params"] == {"q": "threshold", "limit": 5} + assert "Found 2 activities" in out + assert "2026-07-18 | Ride | Threshold intervals" in out + assert "[id: a1]" in out + assert "RACE" in out # the crit + + +def test_search_activities_empty_query(monkeypatch): + calls = _patch_request(monkeypatch, SEARCH_HITS) + out = asyncio.run(activities.search_activities(" ")) + assert "non-empty search query is required" in out + assert calls == [] # no request made + + +def test_search_activities_no_results(monkeypatch): + _patch_request(monkeypatch, []) + assert "No activities found matching 'zzz'" in asyncio.run(activities.search_activities("zzz")) + + +def test_search_activities_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "boom"}) + assert "Error searching activities: boom" in asyncio.run(activities.search_activities("x")) + + +# --------------------------------------------------------------------------- # +# get_activity_best_efforts +# --------------------------------------------------------------------------- # +BEST_EFFORTS = { + "efforts": [ + {"duration": 300, "average": 320, "start_index": 100, "end_index": 400}, + {"distance": 1000, "average": 305, "start_index": 500, "end_index": 700}, + ] +} + + +def test_best_efforts_success_and_param_passthrough(monkeypatch): + calls = _patch_request(monkeypatch, BEST_EFFORTS) + out = asyncio.run( + activities.get_activity_best_efforts("a1", stream="watts", duration=300, count=5) + ) + params = calls[0]["params"] + assert calls[0]["url"] == "/activity/a1/best-efforts" + assert params["stream"] == "watts" + assert params["duration"] == 300 + assert params["count"] == 5 + assert "distance" not in params # None omitted + assert "Best Efforts (watts)" in out + assert "5m: avg 320" in out + assert "1000m: avg 305" in out + + +def test_best_efforts_empty(monkeypatch): + _patch_request(monkeypatch, {"efforts": []}) + out = asyncio.run(activities.get_activity_best_efforts("a1")) + assert "No best-effort data found" in out + + +def test_best_efforts_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "nope"}) + assert "Error fetching best efforts: nope" in asyncio.run( + activities.get_activity_best_efforts("a1") + ) + + +# --------------------------------------------------------------------------- # +# get_activity_interval_stats +# --------------------------------------------------------------------------- # +INTERVAL_STATS = { + "moving_time": 1200, + "average_watts": 265, + "weighted_average_watts": 272, + "max_watts": 410, + "intensity": 0.88, + "training_load": 45, + "average_heartrate": 158, + "decoupling": 3.2, +} + + +def test_interval_stats_success(monkeypatch): + calls = _patch_request(monkeypatch, INTERVAL_STATS) + out = asyncio.run(activities.get_activity_interval_stats("a1", 100, 500)) + assert calls[0]["url"] == "/activity/a1/interval-stats" + assert calls[0]["params"] == {"start_index": 100, "end_index": 500} + assert "Interval Stats:" in out + assert "Avg Power: 265 W" in out + assert "Weighted Avg Power: 272 W" in out + assert "Decoupling: 3.2 %" in out + + +def test_interval_stats_bad_indices(monkeypatch): + calls = _patch_request(monkeypatch, INTERVAL_STATS) + out = asyncio.run(activities.get_activity_interval_stats("a1", 500, 100)) + assert "end_index must be greater than start_index" in out + assert calls == [] # no request + + +def test_interval_stats_empty(monkeypatch): + _patch_request(monkeypatch, {}) + out = asyncio.run(activities.get_activity_interval_stats("a1", 0, 100)) + assert "No interval stats found" in out + + +def test_interval_stats_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "bad range"}) + assert "Error fetching interval stats: bad range" in asyncio.run( + activities.get_activity_interval_stats("a1", 0, 100) + ) diff --git a/tests/test_athlete.py b/tests/test_athlete.py new file mode 100644 index 0000000..0d3a6db --- /dev/null +++ b/tests/test_athlete.py @@ -0,0 +1,327 @@ +""" +Tests for intervals_mcp_server.tools.athlete. + +Covers the athlete-context read tools (get_athlete_profile, get_sport_settings, +get_athlete_summary): request shape, sport filtering, formatting of realistic +fixtures, and the empty / error / credential branches. Default caller credentials +come from the autouse fixture in conftest (athlete ``i1``). +""" + +import asyncio + +from intervals_mcp_server import credentials +from intervals_mcp_server.credentials import CredentialError +from intervals_mcp_server.tools import athlete + +PROFILE = { + "id": "i1", + "name": "Test Athlete", + "sex": "M", + "weight": 72.5, + "icu_resting_hr": 48, + "timezone": "Europe/Madrid", + "measurement_preference": "meters", + "city": "Girona", + "country": "Spain", + "icu_coach": True, + "icu_type_settings": [{"id": 1}, {"id": 2}], +} + +SPORT_SETTINGS = [ + { + "id": 100, + "types": ["Ride", "VirtualRide"], + "ftp": 280, + "indoor_ftp": 275, + "w_prime": 22000, + "power_zones": [55, 75, 90, 105, 120], + "power_zone_names": ["Z1", "Z2", "Z3", "Z4", "Z5"], + "lthr": 165, + "max_hr": 190, + "hr_zones": [120, 145, 160, 175], + "threshold_pace": 4.2, + "pace_units": "MINS_KM", + "pace_zones": [3.5, 4.0, 4.5], + "warmup_time": 600, + "cooldown_time": 300, + }, + {"id": 101, "types": ["Run"], "threshold_pace": 3.8, "pace_units": "MINS_KM"}, +] + +SUMMARY = [ + { + "date": "2026-07-20", + "count": 12, + "moving_time": 43200, + "distance": 320000, + "training_load": 640, + "fitness": 78.5, + "fatigue": 71.0, + "form": 7.5, + "eftp": 285, + "byCategory": [{"category": "Ride", "count": 8, "training_load": 500, "moving_time": 32400}], + } +] + + +def _patch_request(monkeypatch, result): + calls: list[dict] = [] + + async def fake(**kwargs): + calls.append(kwargs) + return result + + monkeypatch.setattr(athlete, "make_intervals_request", fake) + return calls + + +def _patch_seq(monkeypatch, results): + """Patch make_intervals_request to return queued results, one per call.""" + calls: list[dict] = [] + seq = iter(results) + + async def fake(**kwargs): + calls.append(kwargs) + return next(seq) + + monkeypatch.setattr(athlete, "make_intervals_request", fake) + return calls + + +class _StubCtx: + """Minimal stand-in for FastMCP Context.elicit used by the guardrail tests.""" + + def __init__(self, action="accept", confirm=True, raise_exc=False): + self._action = action + self._confirm = confirm + self._raise = raise_exc + self.elicit_calls = 0 + + async def elicit(self, message, schema): # noqa: ARG002 - signature parity + self.elicit_calls += 1 + if self._raise: + raise RuntimeError("client does not support elicitation") + data = type("Data", (), {"confirm": self._confirm})() + return type("Result", (), {"action": self._action, "data": data})() + + +# --------------------------------------------------------------------------- # +# get_athlete_profile +# --------------------------------------------------------------------------- # +def test_get_athlete_profile_success(monkeypatch): + calls = _patch_request(monkeypatch, PROFILE) + out = asyncio.run(athlete.get_athlete_profile()) + assert calls[0]["url"] == "/athlete/i1" + assert "Name: Test Athlete" in out + assert "Weight: 72.5 kg" in out + assert "Resting HR: 48 bpm" in out + assert "Location: Girona, Spain" in out + assert "Role: Coach" in out + assert "2 sport(s) configured" in out + + +def test_get_athlete_profile_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "nope"}) + assert "Error fetching athlete profile: nope" in asyncio.run(athlete.get_athlete_profile()) + + +def test_get_athlete_profile_non_dict(monkeypatch): + _patch_request(monkeypatch, []) + assert "No athlete profile found" in asyncio.run(athlete.get_athlete_profile()) + + +def test_get_athlete_profile_credential_error(monkeypatch): + async def _deny(): + raise CredentialError("not approved") + + monkeypatch.setattr(credentials, "resolve_caller_credentials", _deny) + assert "not approved" in asyncio.run(athlete.get_athlete_profile()) + + +# --------------------------------------------------------------------------- # +# get_sport_settings +# --------------------------------------------------------------------------- # +def test_get_sport_settings_all(monkeypatch): + calls = _patch_request(monkeypatch, SPORT_SETTINGS) + out = asyncio.run(athlete.get_sport_settings()) + assert calls[0]["url"] == "/athlete/i1/sport-settings" + assert "Sport Settings — Ride, VirtualRide" in out + assert "Settings ID: 100" in out + assert "FTP: 280W" in out + assert "Z1: 55, Z2: 75" in out # power zones paired with names + assert "LTHR: 165 bpm" in out + assert "Threshold: 4.2 MINS_KM" in out + assert "Settings ID: 101" in out # second record rendered too + + +def test_get_sport_settings_filter_hit(monkeypatch): + _patch_request(monkeypatch, SPORT_SETTINGS) + out = asyncio.run(athlete.get_sport_settings(sport="run")) # case-insensitive + assert "Settings ID: 101" in out + assert "Settings ID: 100" not in out + + +def test_get_sport_settings_filter_miss(monkeypatch): + _patch_request(monkeypatch, SPORT_SETTINGS) + out = asyncio.run(athlete.get_sport_settings(sport="Swim")) + assert "No sport settings found for sport 'Swim'" in out + + +def test_get_sport_settings_empty(monkeypatch): + _patch_request(monkeypatch, []) + assert "No sport settings found" in asyncio.run(athlete.get_sport_settings()) + + +def test_get_sport_settings_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "boom"}) + assert "Error fetching sport settings: boom" in asyncio.run(athlete.get_sport_settings()) + + +# --------------------------------------------------------------------------- # +# get_athlete_summary +# --------------------------------------------------------------------------- # +def test_get_athlete_summary_success(monkeypatch): + calls = _patch_request(monkeypatch, SUMMARY) + out = asyncio.run(athlete.get_athlete_summary(start_date="2026-06-20", end_date="2026-07-20")) + call = calls[0] + assert call["url"] == "/athlete/i1/athlete-summary" + assert call["params"]["start"] == "2026-06-20" + assert call["params"]["end"] == "2026-07-20" + assert "Fitness (CTL): 78.5" in out + assert "Form (TSB): 7.5" in out + assert "By category:" in out + assert "Ride: 8 activities" in out + + +def test_get_athlete_summary_defaults_dates(monkeypatch): + calls = _patch_request(monkeypatch, SUMMARY) + asyncio.run(athlete.get_athlete_summary()) + # resolve_date_params fills both ends with YYYY-MM-DD + assert len(calls[0]["params"]["start"]) == 10 + assert len(calls[0]["params"]["end"]) == 10 + + +def test_get_athlete_summary_empty(monkeypatch): + _patch_request(monkeypatch, []) + assert "No summary data found" in asyncio.run(athlete.get_athlete_summary()) + + +def test_get_athlete_summary_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "bad"}) + assert "Error fetching athlete summary: bad" in asyncio.run(athlete.get_athlete_summary()) + + +# --------------------------------------------------------------------------- # +# update_sport_settings (dual-guardrail write) +# --------------------------------------------------------------------------- # +CURRENT_SS = [{"id": 100, "types": ["Ride"], "ftp": 280, "lthr": 165}] + + +def test_update_sport_settings_no_fields(monkeypatch): + calls = _patch_seq(monkeypatch, []) + out = asyncio.run(athlete.update_sport_settings(settings_id=100)) + assert "No settings provided" in out + assert calls == [] # returns before any fetch + + +def test_update_sport_settings_refuses_without_confirm_or_ctx(monkeypatch): + calls = _patch_seq(monkeypatch, [CURRENT_SS]) # only the GET happens + out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=300)) + assert "⚠️ This will change your Ride thresholds" in out + assert "ftp: 280 -> 300" in out + assert "re-run with confirm=true" in out + assert len(calls) == 1 and calls[0]["url"] == "/athlete/i1/sport-settings" # no PUT + + +def test_update_sport_settings_confirm_true_writes(monkeypatch): + calls = _patch_seq(monkeypatch, [CURRENT_SS, {"id": 100, "types": ["Ride"], "ftp": 300}]) + out = asyncio.run( + athlete.update_sport_settings(settings_id=100, ftp=300, recalc_hr_zones=True, confirm=True) + ) + put = calls[1] + assert put["method"] == "PUT" + assert put["url"] == "/athlete/i1/sport-settings/100" + assert put["params"] == {"recalcHrZones": True} + assert put["data"]["ftp"] == 300 # merged into the full record + assert "Updated Ride settings" in out + + +def test_update_sport_settings_elicit_accept_writes(monkeypatch): + calls = _patch_seq(monkeypatch, [CURRENT_SS, {"id": 100, "types": ["Ride"], "ftp": 300}]) + ctx = _StubCtx(action="accept", confirm=True) + out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=300, ctx=ctx)) + assert ctx.elicit_calls == 1 + assert len(calls) == 2 and calls[1]["method"] == "PUT" + assert "Updated Ride settings" in out + + +def test_update_sport_settings_elicit_decline(monkeypatch): + calls = _patch_seq(monkeypatch, [CURRENT_SS]) + ctx = _StubCtx(action="decline") + out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=300, ctx=ctx)) + assert "did not confirm" in out + assert "confirm=true" not in out # no bypass instructions after a refusal + assert len(calls) == 1 # no PUT + + +def test_update_sport_settings_elicit_cancel(monkeypatch): + _patch_seq(monkeypatch, [CURRENT_SS]) + ctx = _StubCtx(action="cancel") + out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=300, ctx=ctx)) + assert "did not confirm" in out + + +def test_update_sport_settings_accept_without_confirm_refuses_hard(monkeypatch): + # Submitting the elicitation without ticking confirm is a refusal: the tool + # must stop and must NOT emit the confirm=true bypass instructions — and an + # explicit confirm=True param must not override the answered elicitation. + calls = _patch_seq(monkeypatch, [CURRENT_SS]) + ctx = _StubCtx(action="accept", confirm=False) + out = asyncio.run( + athlete.update_sport_settings(settings_id=100, ftp=300, confirm=True, ctx=ctx) + ) + assert "did not confirm" in out + assert "confirm=true" not in out + assert len(calls) == 1 # no PUT + + +def test_update_sport_settings_elicit_unsupported_falls_back(monkeypatch): + calls = _patch_seq(monkeypatch, [CURRENT_SS]) + ctx = _StubCtx(raise_exc=True) # client without elicitation capability + out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=300, ctx=ctx)) + assert "re-run with confirm=true" in out + assert len(calls) == 1 # refused, no PUT + + +def test_update_sport_settings_unknown_id(monkeypatch): + _patch_seq(monkeypatch, [CURRENT_SS]) + out = asyncio.run(athlete.update_sport_settings(settings_id=999, ftp=300, confirm=True)) + assert "No sport settings found with ID 999" in out + + +def test_update_sport_settings_no_op(monkeypatch): + _patch_seq(monkeypatch, [CURRENT_SS]) + out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=280, confirm=True)) + assert "No changes" in out + + +def test_update_sport_settings_fetch_error(monkeypatch): + _patch_seq(monkeypatch, [{"error": True, "message": "down"}]) + out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=300, confirm=True)) + assert "Error fetching current sport settings: down" in out + + +def test_update_sport_settings_put_error(monkeypatch): + _patch_seq(monkeypatch, [CURRENT_SS, {"error": True, "message": "rejected"}]) + out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=300, confirm=True)) + assert "Error updating sport settings: rejected" in out + + +def test_update_sport_settings_empty_echo_renders_merged(monkeypatch): + # An empty-body 200 parses to {}; the confirmation must render the merged + # record (with the new FTP), not format_sport_settings({}). + calls = _patch_seq(monkeypatch, [CURRENT_SS, {}]) + out = asyncio.run(athlete.update_sport_settings(settings_id=100, ftp=300, confirm=True)) + assert len(calls) == 2 + assert "FTP: 300W" in out + assert "Settings ID: 100" in out diff --git a/tests/test_readiness.py b/tests/test_readiness.py new file mode 100644 index 0000000..a2f97d1 --- /dev/null +++ b/tests/test_readiness.py @@ -0,0 +1,208 @@ +""" +Tests for the training-readiness feature. + +The pure compute layer (utils/readiness.py) is exercised directly on deterministic +fixtures; one integration test drives the get_training_readiness tool with the HTTP +layer stubbed. Fixtures are built so verdicts are unambiguous. +""" + +import asyncio +from datetime import date, timedelta + +from intervals_mcp_server.tools import wellness +from intervals_mcp_server.utils import readiness + + +def _days(specs: list[dict]) -> list[dict]: + """Build wellness records with sequential dates from a list of field dicts.""" + return [{"id": f"2026-06-{i + 1:02d}", **spec} for i, spec in enumerate(specs)] + + +def _days_ending_today(specs: list[dict]) -> list[dict]: + """Like _days, but the last record is dated today (for tool-level tests).""" + start = date.today() - timedelta(days=len(specs) - 1) + return [ + {"id": (start + timedelta(days=i)).isoformat(), **spec} for i, spec in enumerate(specs) + ] + + +def _stable(n: int, **fields) -> list[dict]: + return _days([dict(fields) for _ in range(n)]) + + +# --------------------------------------------------------------------------- # +# HRV signal +# --------------------------------------------------------------------------- # +def test_hrv_insufficient_data(): + recs = _stable(10, hrv=50) + sig = readiness.hrv_signal(recs) + assert sig["level"] == "nodata" + + +def test_hrv_normal_band(): + # 23 stable baseline days + 7 stable recent days -> within band + recs = _days([{"hrv": 50 + (i % 3)} for i in range(30)]) + assert readiness.hrv_signal(recs)["level"] == "ok" + + +def test_hrv_suppressed_alert(): + baseline = [{"hrv": 50 + (i % 3)} for i in range(23)] + recent = [{"hrv": 34} for _ in range(7)] + assert readiness.hrv_signal(_days(baseline + recent))["level"] == "alert" + + +def test_hrv_elevated_warn(): + baseline = [{"hrv": 50 + (i % 3)} for i in range(23)] + recent = [{"hrv": 75} for _ in range(7)] + assert readiness.hrv_signal(_days(baseline + recent))["level"] == "warn" + + +def test_hrv_constant_baseline_small_dip_is_not_alert(): + # A near-constant baseline gives SWC ~ 0; the floor must keep a trivial + # 50 -> 49 fluctuation from producing a false "Compromised" alert. + recs = _days([{"hrv": 50} for _ in range(23)] + [{"hrv": 49} for _ in range(7)]) + assert readiness.hrv_signal(recs)["level"] == "ok" + + +# --------------------------------------------------------------------------- # +# RHR / sleep signals +# --------------------------------------------------------------------------- # +def test_rhr_elevated_warn(): + recs = _days([{"restingHR": 48} for _ in range(23)] + [{"restingHR": 56} for _ in range(7)]) + assert readiness.rhr_signal(recs)["level"] == "warn" + + +def test_rhr_normal_ok(): + assert readiness.rhr_signal(_stable(20, restingHR=48))["level"] == "ok" + + +def test_rhr_minimum_days_is_nodata_not_self_baseline(): + # With only 7 samples there is no disjoint baseline; a uniformly-elevated + # (ill) week must NOT read "ok" from being compared against itself. + sig = readiness.rhr_signal(_stable(7, restingHR=58)) + assert sig["level"] == "nodata" + + +def test_sleep_short_warn(): + recs = _days([{"sleepSecs": 28800} for _ in range(10)] + [{"sleepSecs": 18000}]) + assert readiness.sleep_signal(recs)["level"] == "warn" + + +def test_sleep_nodata(): + assert readiness.sleep_signal(_stable(3, sleepSecs=28800))["level"] == "nodata" + + +# --------------------------------------------------------------------------- # +# subjective signals (conventional direction) +# --------------------------------------------------------------------------- # +def test_subjective_fatigue_elevated_warns(): + recs = _days([{"fatigue": 2} for _ in range(10)] + [{"fatigue": 4}]) + sigs = readiness.subjective_signals(recs) + assert any(s["name"] == "Fatigue" and s["level"] == "warn" for s in sigs) + + +def test_subjective_stable_no_warning(): + assert readiness.subjective_signals(_stable(10, fatigue=2, mood=3)) == [] + + +# --------------------------------------------------------------------------- # +# overall verdict +# --------------------------------------------------------------------------- # +def test_verdict_green_all_stable(): + recs = _days( + [{"hrv": 50 + (i % 3), "restingHR": 48, "sleepSecs": 28800} for i in range(30)] + ) + assert readiness.assess_readiness(recs)["verdict"] == "green" + + +def test_verdict_red_on_hrv_suppression(): + recs = _days( + [{"hrv": 50 + (i % 3), "restingHR": 48, "sleepSecs": 28800} for i in range(23)] + + [{"hrv": 33, "restingHR": 57, "sleepSecs": 28800} for _ in range(7)] + ) + assert readiness.assess_readiness(recs)["verdict"] == "red" + + +def test_verdict_insufficient_when_hrv_sparse_and_little_else(): + # Only 3 days total, no HRV baseline and <2 other core signals with data. + recs = _stable(3, restingHR=48) + out = readiness.assess_readiness(recs) + assert out["verdict"] == "insufficient" + + +def test_verdict_uses_rhr_and_sleep_when_hrv_missing(): + # No HRV, but RHR + sleep both have data -> a verdict is still produced (green here). + recs = _days([{"restingHR": 48, "sleepSecs": 28800} for _ in range(20)]) + out = readiness.assess_readiness(recs) + assert out["verdict"] == "green" + assert any(s["name"] == "HRV" and s["level"] == "nodata" for s in out["signals"]) + + +def test_form_context_computed(): + recs = _days([{"ctl": 60, "atl": 70}]) + assert readiness.form_context(recs) == {"form": -10.0, "ctl": 60, "atl": 70} + + +# --------------------------------------------------------------------------- # +# render + tool integration +# --------------------------------------------------------------------------- # +def test_stale_data_withholds_verdict(): + # Daily logging that STOPPED 3 weeks ago must not produce a current verdict: + # with today as the reference date every calendar window is empty. + old = _days([{"hrv": 50 + (i % 3), "restingHR": 48, "sleepSecs": 28800} for i in range(30)]) + out = readiness.assess_readiness(old, reference_date=date.today().isoformat()) + assert out["verdict"] == "insufficient" + assert all(s["level"] == "nodata" for s in out["signals"]) + + +def test_sleep_not_logged_recently_is_nodata(): + recs = _days([{"sleepSecs": 28800} for _ in range(10)]) + sig = readiness.sleep_signal(recs, reference_date="2026-07-01") # 3 weeks later + assert sig["level"] == "nodata" + assert "no sleep logged since" in sig["detail"] + + +def test_render_insufficient_mentions_logging(): + out = readiness.render_readiness(readiness.assess_readiness(_stable(3, restingHR=48))) + assert "Verdict withheld" in out + assert "Log daily HRV" in out + + +def test_get_training_readiness_tool(monkeypatch): + # Wellness API returns a date-keyed dict; the tool must normalize and assess it. + start = date.today() - timedelta(days=29) + records = { + (start + timedelta(days=i)).isoformat(): { + "hrv": 50 + (i % 3), + "restingHR": 48, + "sleepSecs": 28800, + } + for i in range(30) + } + calls: list[dict] = [] + + async def fake(**kwargs): + calls.append(kwargs) + return records + + monkeypatch.setattr(wellness, "make_intervals_request", fake) + out = asyncio.run(wellness.get_training_readiness(days=45)) + assert calls[0]["url"] == "/athlete/i1/wellness" + assert "Training Readiness:" in out + assert "🟢 Ready" in out + + +def test_get_training_readiness_no_data(monkeypatch): + async def fake(**kwargs): + return {} + + monkeypatch.setattr(wellness, "make_intervals_request", fake) + assert "No wellness data found" in asyncio.run(wellness.get_training_readiness()) + + +def test_get_training_readiness_error(monkeypatch): + async def fake(**kwargs): + return {"error": True, "message": "down"} + + monkeypatch.setattr(wellness, "make_intervals_request", fake) + assert "Error fetching wellness data: down" in asyncio.run(wellness.get_training_readiness()) diff --git a/tests/test_tool_auth.py b/tests/test_tool_auth.py index 37a8e21..a572e12 100644 --- a/tests/test_tool_auth.py +++ b/tests/test_tool_auth.py @@ -11,16 +11,28 @@ import pytest from intervals_mcp_server import credentials from intervals_mcp_server.credentials import CredentialError -from intervals_mcp_server.tools import activities, custom_items, events, gear, power_curves, wellness +from intervals_mcp_server.tools import ( + activities, + athlete, + custom_items, + events, + gear, + power_curves, + wellness, + workouts, +) # (tool callable, minimal required positional args) -TOOL_CALLS = [ +TOOL_CALLS: list[tuple] = [ (activities.get_activities, ()), (activities.get_activity_details, ("1",)), (activities.get_activity_intervals, ("1",)), (activities.get_activity_streams, ("1",)), (activities.get_activity_messages, ("1",)), (activities.add_activity_message, ("1", "hi")), + (activities.search_activities, ("ride",)), + (activities.get_activity_best_efforts, ("1",)), + (activities.get_activity_interval_stats, ("1", 0, 100)), (events.get_events, ()), (events.get_event_by_id, ("e1",)), (events.delete_event, ("e1",)), @@ -28,6 +40,15 @@ TOOL_CALLS = [ (events.add_or_update_event, ("Ride", "Name")), (events.add_or_update_note, ("Name", "desc")), (wellness.get_wellness_data, ()), + (wellness.update_wellness, ()), + (wellness.update_wellness_bulk, ([],)), + (wellness.get_training_readiness, ()), + (athlete.get_athlete_profile, ()), + (athlete.get_sport_settings, ()), + (athlete.get_athlete_summary, ()), + (athlete.update_sport_settings, (1,)), + (workouts.get_workouts, ()), + (workouts.get_workout, (1,)), (power_curves.get_athlete_power_curves, ()), (gear.get_gear_list, ()), (custom_items.get_custom_items, ()), @@ -50,6 +71,17 @@ def test_tool_returns_message_when_unauthorized(monkeypatch, func, args): assert result == "ACCOUNT NOT APPROVED" -def test_all_20_tools_covered(): - """Guard: if a tool is added, add it here so its auth gate is tested.""" - assert len(TOOL_CALLS) == 20 +def test_all_tools_covered(): + """Guard: every registered MCP tool must appear in TOOL_CALLS. + + Compares against the live tool registry instead of a hand-maintained count, + so adding a tool without adding its auth-gate test fails loudly here. + """ + from intervals_mcp_server.mcp_instance import mcp + + registered = {t.name for t in asyncio.run(mcp.list_tools())} + covered = {f.__name__ for f, _ in TOOL_CALLS} + assert covered == registered, ( + f"auth-gate matrix out of sync: missing={sorted(registered - covered)} " + f"extra={sorted(covered - registered)}" + ) diff --git a/tests/test_wellness.py b/tests/test_wellness.py index 398fb06..87415cd 100644 --- a/tests/test_wellness.py +++ b/tests/test_wellness.py @@ -134,3 +134,102 @@ def test_update_wellness_echo_without_date_shows_written_date(monkeypatch): out = asyncio.run(wellness.update_wellness(date="2025-05-24", weight=80)) assert "Date: 2025-05-24" in out assert "Date: N/A" not in out + + +# --------------------------------------------------------------------------- # +# update_wellness_bulk +# --------------------------------------------------------------------------- # +def test_update_wellness_bulk_success(monkeypatch): + calls = _patch_request(monkeypatch, [{"id": "2026-07-18"}, {"id": "2026-07-19"}]) + out = asyncio.run( + wellness.update_wellness_bulk( + [ + {"date": "2026-07-18", "weight": 80, "carbohydrates": 300}, + {"date": "2026-07-19", "sleep_hours": 8}, + ] + ) + ) + call = calls[0] + assert call["method"] == "PUT" + assert call["url"] == "/athlete/i1/wellness-bulk" + body = call["data"] + assert isinstance(body, list) and len(body) == 2 + assert body[0]["id"] == "2026-07-18" + assert body[0]["weight"] == 80 + assert body[0]["carbohydrates"] == 300 # camelCase mapping shared with update_wellness + assert body[1]["sleepSecs"] == 8 * 3600 + assert "Updated 2 day(s)" in out + + +def test_update_wellness_bulk_mapping_matches_single(monkeypatch): + # The bulk payload for a day must equal the single-day payload for the same fields + # (plus the id) — proving the shared _wellness_payload helper prevents drift. + fields = {"weight": 78, "resting_hr": 50, "sleep_hours": 7.5, "fat": 60, "locked": True} + from intervals_mcp_server.tools.wellness import _wellness_payload + + single = _wellness_payload(fields) + calls = _patch_request(monkeypatch, [{}]) + asyncio.run(wellness.update_wellness_bulk([{"date": "2026-07-18", **fields}])) + bulk_entry = {k: v for k, v in calls[0]["data"][0].items() if k != "id"} + assert bulk_entry == single + + +def test_update_wellness_bulk_invalid_date_rejects_whole_batch(monkeypatch): + calls = _patch_request(monkeypatch, [{}]) + out = asyncio.run( + wellness.update_wellness_bulk( + [{"date": "2026-07-18", "weight": 80}, {"date": "not-a-date", "weight": 81}] + ) + ) + assert "Error in entry 1" in out + assert calls == [] # no partial write + + +def test_update_wellness_bulk_missing_date(monkeypatch): + calls = _patch_request(monkeypatch, [{}]) + out = asyncio.run(wellness.update_wellness_bulk([{"weight": 80}])) + assert "entry 0 is missing a 'date'" in out + assert calls == [] + + +def test_update_wellness_bulk_entry_no_fields(monkeypatch): + calls = _patch_request(monkeypatch, [{}]) + out = asyncio.run(wellness.update_wellness_bulk([{"date": "2026-07-18"}])) + assert "has no wellness fields" in out + assert calls == [] + + +def test_update_wellness_bulk_empty(monkeypatch): + calls = _patch_request(monkeypatch, [{}]) + out = asyncio.run(wellness.update_wellness_bulk([])) + assert "No entries provided" in out + assert calls == [] + + +def test_update_wellness_bulk_too_many(monkeypatch): + calls = _patch_request(monkeypatch, [{}]) + out = asyncio.run( + wellness.update_wellness_bulk([{"date": "2026-01-01", "weight": 80}] * 93) + ) + assert "Too many entries" in out + assert calls == [] + + +def test_update_wellness_bulk_rejects_unknown_keys(monkeypatch): + # camelCase/API-style names must be rejected, not silently dropped: the value + # the caller asked to record would otherwise be lost behind a success message. + calls = _patch_request(monkeypatch, [{}]) + out = asyncio.run( + wellness.update_wellness_bulk( + [{"date": "2026-07-18", "weight": 80, "restingHR": 50}] + ) + ) + assert "unrecognized field(s): restingHR" in out + assert "resting_hr" in out # the error names the valid fields + assert calls == [] # whole batch rejected, nothing written + + +def test_update_wellness_bulk_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "boom"}) + out = asyncio.run(wellness.update_wellness_bulk([{"date": "2026-07-18", "weight": 80}])) + assert "Error updating wellness data: boom" in out diff --git a/tests/test_workouts.py b/tests/test_workouts.py new file mode 100644 index 0000000..92c7052 --- /dev/null +++ b/tests/test_workouts.py @@ -0,0 +1,131 @@ +""" +Tests for intervals_mcp_server.tools.workouts (0.3.0 workout library). + +Covers get_workouts (list + client-side filters) and get_workout (full detail +with a nested workout_doc), plus empty / error / credential branches. +""" + +import asyncio + +from intervals_mcp_server import credentials +from intervals_mcp_server.credentials import CredentialError +from intervals_mcp_server.tools import workouts + +LIBRARY = [ + {"id": 10, "name": "VO2 5x5", "type": "Ride", "icu_training_load": 95, "moving_time": 3600, "folder_id": 1}, + {"id": 11, "name": "Easy run", "type": "Run", "moving_time": 2400, "folder_id": 2}, +] + +WORKOUT_DETAIL = { + "id": 10, + "name": "VO2 5x5", + "type": "Ride", + "indoor": True, + "moving_time": 3600, + "icu_training_load": 95, + "description": "VO2max builder", + "tags": ["vo2", "key"], + "workout_doc": { + "steps": [ + {"duration": 900, "power": {"value": 60, "units": "%ftp"}, "warmup": True}, + { + "reps": 5, + "steps": [ + {"duration": 300, "power": {"value": 115, "units": "%ftp"}, "text": "hard"}, + {"duration": 300, "power": {"value": 50, "units": "%ftp"}, "text": "easy"}, + ], + }, + {"duration": 600, "power": {"start": 60, "end": 40, "units": "%ftp"}, "cooldown": True}, + ] + }, +} + + +def _patch_request(monkeypatch, result): + calls: list[dict] = [] + + async def fake(**kwargs): + calls.append(kwargs) + return result + + monkeypatch.setattr(workouts, "make_intervals_request", fake) + return calls + + +def test_get_workouts_all(monkeypatch): + calls = _patch_request(monkeypatch, LIBRARY) + out = asyncio.run(workouts.get_workouts()) + assert calls[0]["url"] == "/athlete/i1/workouts" + assert "Workout Library (2)" in out + assert "VO2 5x5 | Ride (load 95, 3600s, folder 1) [id: 10]" in out + + +def test_get_workouts_filter_folder(monkeypatch): + _patch_request(monkeypatch, LIBRARY) + out = asyncio.run(workouts.get_workouts(folder_id=2)) + assert "Easy run" in out + assert "VO2 5x5" not in out + + +def test_get_workouts_filter_sport(monkeypatch): + _patch_request(monkeypatch, LIBRARY) + out = asyncio.run(workouts.get_workouts(sport_type="ride")) + assert "VO2 5x5" in out + assert "Easy run" not in out + + +def test_get_workouts_empty(monkeypatch): + _patch_request(monkeypatch, []) + assert "No workouts found" in asyncio.run(workouts.get_workouts()) + + +def test_get_workouts_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "boom"}) + assert "Error fetching workouts: boom" in asyncio.run(workouts.get_workouts()) + + +def test_get_workout_detail_with_nested_doc(monkeypatch): + calls = _patch_request(monkeypatch, WORKOUT_DETAIL) + out = asyncio.run(workouts.get_workout(10)) + assert calls[0]["url"] == "/athlete/i1/workouts/10" + assert "Workout: VO2 5x5" in out + assert "Training Load: 95" in out + assert "Tags: vo2, key" in out + assert "Steps:" in out + assert "15m @ 60%ftp (warmup)" in out + assert "5x:" in out # repeat block rendered + assert "5m @ 115%ftp — hard" in out # nested step + assert "10m @ 60-40%ftp (cooldown)" in out # power range (no ramp flag set) + + +def test_get_workout_ramp_step(monkeypatch): + _patch_request( + monkeypatch, + { + "id": 12, + "name": "Ramp test", + "workout_doc": { + "steps": [{"ramp": True, "power": {"start": 100, "end": 300, "units": "w"}}] + }, + }, + ) + out = asyncio.run(workouts.get_workout(12)) + assert "ramp 100-300w" in out + + +def test_get_workout_not_found(monkeypatch): + _patch_request(monkeypatch, {}) + assert "No workout found with ID 99" in asyncio.run(workouts.get_workout(99)) + + +def test_get_workout_error(monkeypatch): + _patch_request(monkeypatch, {"error": True, "message": "nope"}) + assert "Error fetching workout: nope" in asyncio.run(workouts.get_workout(10)) + + +def test_get_workout_credential_error(monkeypatch): + async def _deny(): + raise CredentialError("not approved") + + monkeypatch.setattr(credentials, "resolve_caller_credentials", _deny) + assert "not approved" in asyncio.run(workouts.get_workout(10))