diff --git a/.github/workflows/container-build.yml b/.github/workflows/container-build.yml new file mode 100644 index 00000000..b1fd03c4 --- /dev/null +++ b/.github/workflows/container-build.yml @@ -0,0 +1,31 @@ +name: Container build + +on: + pull_request: + +permissions: + contents: read + +jobs: + build: + name: Build multi-architecture image + runs-on: ubuntu-latest + + steps: + - name: Check out repository + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + + - name: Set up QEMU + uses: docker/setup-qemu-action@96fe6ef7f33517b61c61be40b68a1882f3264fb8 # v4.2.0 + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0 + + - name: Build image + uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0 + with: + context: . + platforms: linux/amd64,linux/arm64 + push: false + cache-from: type=gha + cache-to: type=gha,mode=max diff --git a/.github/workflows/container-publish.yml b/.github/workflows/container-publish.yml new file mode 100644 index 00000000..8ecdc3fa --- /dev/null +++ b/.github/workflows/container-publish.yml @@ -0,0 +1,68 @@ +name: Container publish + +on: + push: + branches: + - main + - feat/ghcr-publishing + +permissions: + contents: read + +jobs: + publish: + name: Publish multi-architecture image + runs-on: ubuntu-latest + permissions: + contents: read + packages: write + + steps: + - name: Check out repository + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + + - name: Set up QEMU + uses: docker/setup-qemu-action@96fe6ef7f33517b61c61be40b68a1882f3264fb8 # v4.2.0 + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0 + + - name: Generate image metadata + id: metadata + shell: bash + run: | + image="ghcr.io/${GITHUB_REPOSITORY,,}" + short_sha="${GITHUB_SHA::7}" + if [[ "${GITHUB_REF_NAME}" == "main" ]]; then + release_tag="latest" + else + release_tag="${GITHUB_REF_NAME//\//-}" + fi + + { + echo "image=${image}" + echo "tags<> "${GITHUB_OUTPUT}" + + - name: Log in to GitHub Container Registry + uses: docker/login-action@af1e73f918a031802d376d3c8bbc3fe56130a9b0 # v4.4.0 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Build and publish image + uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0 + with: + context: . + platforms: linux/amd64,linux/arm64 + push: true + tags: ${{ steps.metadata.outputs.tags }} + labels: | + org.opencontainers.image.source=https://github.com/${{ github.repository }} + org.opencontainers.image.revision=${{ github.sha }} + cache-from: type=gha + cache-to: type=gha,mode=max diff --git a/Dockerfile b/Dockerfile index 9bccdda3..dfc1c1d9 100644 --- a/Dockerfile +++ b/Dockerfile @@ -13,16 +13,16 @@ COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv # Set environment variables ENV PYTHONUNBUFFERED=1 \ PYTHONDONTWRITEBYTECODE=1 \ - UV_SYSTEM_PYTHON=1 + PATH="/app/.venv/bin:$PATH" # Copy dependency files and README first for better layer caching -COPY pyproject.toml README.md ./ +COPY pyproject.toml uv.lock README.md ./ # Copy the application source code (needed for editable install) COPY src/ ./src/ -# Install dependencies using uv -RUN uv pip install -e . +# Install the exact locked production dependencies into the project environment. +RUN uv sync --frozen --no-dev --no-editable # Copy test files (optional, for testing in container) COPY tests/ ./tests/ diff --git a/README.md b/README.md index a532b508..0ecc35e1 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,7 @@ Garmin's API is accessed via the awesome [python-garminconnect](https://github.c ### Tool Coverage -This MCP server implements **110+ tools** covering ~90% of the [python-garminconnect](https://github.com/cyberjunky/python-garminconnect) library (v0.3.2): +This MCP server implements **113+ tools** covering ~90% of the [python-garminconnect](https://github.com/cyberjunky/python-garminconnect) library (v0.3.2), plus Garmin's public exercise catalog: - ✅ Activity Management (20 tools) - includes write tools for type, description, event type, perceived effort, and feel - ✅ Health & Wellness (31 tools) - includes custom lightweight summary tools @@ -42,6 +42,7 @@ This MCP server implements **110+ tools** covering ~90% of the [python-garmincon - ✅ Courses (3 tools) - list / upload GPX as course / delete course - ✅ Activity Analysis (2 tools) - FIT file parsing, Power Duration Curve; requires power meter and/or Di2 - ✅ Activity File Downloads (2 tools) - download activity files in FIT, GPX, TCX, or CSV format +- ✅ Exercise Catalog (3 tools) - browse, match, and batch-resolve public Garmin strength-exercise identifiers without account authentication > **Note:** Activity Analysis tools require a compatible power meter (e.g., Garmin Rally, Favero Assioma, PowerTap P1) and/or Shimano Di2 / SRAM eTap electronic shifting. The `fitparse` dependency is installed automatically. @@ -78,7 +79,7 @@ If you need any of these endpoints, please [open an issue](https://github.com/Ta ## Tool Filtering -This server registers 110+ tools by default, which can be a lot of context for +This server registers 113+ tools by default, which can be a lot of context for an LLM to carry in every session. You can expose only the tools you need with two optional environment variables: @@ -99,6 +100,78 @@ Example — expose only sleep, stress, and recent activities: } ``` +## Exercise catalog tools + +The read-only `list_strength_exercises`, `match_strength_exercise`, and +`resolve_strength_exercises` tools use +Garmin's public [exercise catalog](https://connect.garmin.com/web-data/exercises/Exercises.json) +as the canonical source of valid `category` + `exercise_name` pairs. Human-readable +labels come from Garmin's public +[`exercise_types.properties`](https://connect.garmin.com/web-translations/exercise_types/exercise_types.properties) +translation file. These tools do not access an authenticated Garmin account and +never create temporary workouts. + +Browse a category, search labels or identifiers, and paginate the results: + +```json +{ + "category": "CRUNCH", + "search": "reverse", + "limit": 20 +} +``` + +Match a human-friendly description and receive exact identifiers plus alternatives +when the result is uncertain: + +```json +{ + "query": "Reverse Crunch" +} +``` + +The fields to use when creating a workout are the Garmin identifiers, not the +display label: + +```json +{ + "category": "CRUNCH", + "exercise_name": "REVERSE_CRUNCH" +} +``` + +Resolve a complete create-ready batch while preserving sets, reps, rest, and any +other caller metadata: + +```json +{ + "exercises": [ + {"name": "Reverse Crunch", "sets": 3, "reps": 12, "rest_seconds": 60}, + {"category": "CARRY", "exercise_name": "FARMERS_CARRY", "sets": 4, "reps": 1} + ], + "limit": 5 +} +``` + +The batch status is `ready` only when every exercise is exact or confidently +matched. Ambiguous, unknown, conflicting, or invalid inputs return `needs_review` +with per-item alternatives. This resolver never requires Garmin authentication. + +Catalog files are cached in memory and under +`~/.cache/garmin_mcp/exercise_catalog/` for seven days by default. A structurally +valid expired cache remains available as a stale fallback during network outages. +The cache directory, TTL, and both source URLs can be overridden with +`GARMIN_EXERCISE_CACHE_DIR`, `GARMIN_EXERCISE_CACHE_TTL_SECONDS`, +`GARMIN_EXERCISES_URL`, and `GARMIN_EXERCISE_LABELS_URL`. + +A display label such as `Reverse Crunch` is intended for people and search; +`CRUNCH` / `REVERSE_CRUNCH` is the exact Garmin pair. A small conservative alias +table improves searches such as `abdominal invertido`, but aliases are not part +of Garmin's official catalog. Approximate semantic matching cannot guarantee that +two differently named movements are equivalent. When Garmin has no exact entry, +use the scored alternatives to make that decision rather than treating a weak +match as exact. + ## High-level workout tools These builder tools let an LLM create and schedule workouts without writing raw Garmin JSON. @@ -143,7 +216,7 @@ name kept in the step description. The name is also sent as `exerciseName`, but retains that when it matches one of its own exercise keys (e.g. `FARMERS_CARRY`) — any other value is accepted and then stored empty. -`category` is optional and passed straight through. Omit it and the key is left out of the +By default, `category` is optional and passed straight through. Omit it and the key is left out of the payload entirely, which Garmin accepts. Supply it and it must be one of Garmin's exercise categories — anything else, including `OTHER` and `UNASSIGNED`, is rejected with `400 - Invalid category`. The full list is published at @@ -152,6 +225,7 @@ categories — anything else, including `OTHER` and `UNASSIGNED`, is rejected wi ```json { "name": "Full Body A", + "resolve_exercises": true, "exercises": [ {"name": "Sentadillas", "sets": 3, "reps": 12, "rest_seconds": 90}, {"name": "Flexiones", "sets": 3, "reps": 15, "rest_seconds": 60}, @@ -161,6 +235,14 @@ categories — anything else, including `OTHER` and `UNASSIGNED`, is rejected wi } ``` +With `resolve_exercises: true`, the whole batch is checked before JSON construction. +Every item must resolve before anything is uploaded; failures return all unresolved +items and alternatives. A valid supplied `category` + `exercise_name` pair is used +as-is while `name` remains the readable description. If the public catalog and its +cache are both unavailable, creation uses the legacy inputs and the success response +contains an explicit warning. Leaving the flag false keeps the previous behavior and +does not load the catalog. + Returns: `{"status": "success", "workout_id": 1234567890, ...}` ### `schedule_week` diff --git a/pyproject.toml b/pyproject.toml index d56c3c51..4f1492be 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,7 +12,7 @@ dependencies = [ "python-dotenv==1.2.2", "garminconnect==0.3.2", "requests==2.33.0", - "mcp>=1.28.1", + "mcp>=1.28.1,<2", "fitparse>=1.2.0", ] @@ -27,4 +27,3 @@ dev-dependencies = [ "pytest-mock>=3.14.0", "pytest-timeout>=2.3.1", ] - diff --git a/src/garmin_mcp/__init__.py b/src/garmin_mcp/__init__.py index 1fd483e7..12788c2f 100644 --- a/src/garmin_mcp/__init__.py +++ b/src/garmin_mcp/__init__.py @@ -29,6 +29,7 @@ from garmin_mcp import workout_builders from garmin_mcp import courses from garmin_mcp import activity_analysis +from garmin_mcp import exercise_catalog def is_interactive_terminal() -> bool: @@ -389,31 +390,33 @@ def main(): # Initialize Garmin client garmin_client = init_api(email, password) - if not garmin_client: - print("Failed to initialize Garmin Connect client. Exiting.", file=sys.stderr) - return - - print("Garmin Connect client initialized successfully.", file=sys.stderr) - - # Wrap client so runtime auth/rate-limit errors surface as clear messages - garmin_client = _GarminProxy(garmin_client) - - # Configure all modules with the Garmin client - activity_management.configure(garmin_client) - health_wellness.configure(garmin_client) - user_profile.configure(garmin_client) - devices.configure(garmin_client) - gear_management.configure(garmin_client) - weight_management.configure(garmin_client) - challenges.configure(garmin_client) - training.configure(garmin_client) - workouts.configure(garmin_client) - data_management.configure(garmin_client) - womens_health.configure(garmin_client) - nutrition.configure(garmin_client) - workout_builders.configure(garmin_client) - courses.configure(garmin_client) - activity_analysis.configure(garmin_client) + if garmin_client: + print("Garmin Connect client initialized successfully.", file=sys.stderr) + + # Wrap client so runtime auth/rate-limit errors surface as clear messages + garmin_client = _GarminProxy(garmin_client) + + # Configure modules that use the authenticated Garmin client. + activity_management.configure(garmin_client) + health_wellness.configure(garmin_client) + user_profile.configure(garmin_client) + devices.configure(garmin_client) + gear_management.configure(garmin_client) + weight_management.configure(garmin_client) + challenges.configure(garmin_client) + training.configure(garmin_client) + workouts.configure(garmin_client) + data_management.configure(garmin_client) + womens_health.configure(garmin_client) + nutrition.configure(garmin_client) + workout_builders.configure(garmin_client) + courses.configure(garmin_client) + activity_analysis.configure(garmin_client) + else: + print( + "Garmin authentication unavailable; exposing public catalog tools only.", + file=sys.stderr, + ) # Create the MCP app, wrapped so the env-var filter can drop tools. # host/port only matter for the HTTP transports; stdio ignores them. @@ -425,24 +428,27 @@ def main(): print(f"Tool filter: denylist of {len(disabled_tools)} tool(s).", file=sys.stderr) # Register tools from all modules - app = activity_management.register_tools(app) - app = health_wellness.register_tools(app) - app = user_profile.register_tools(app) - app = devices.register_tools(app) - app = gear_management.register_tools(app) - app = weight_management.register_tools(app) - app = challenges.register_tools(app) - app = training.register_tools(app) - app = workouts.register_tools(app) - app = data_management.register_tools(app) - app = womens_health.register_tools(app) - app = nutrition.register_tools(app) - app = workout_builders.register_tools(app) - app = courses.register_tools(app) - app = activity_analysis.register_tools(app) + app = exercise_catalog.register_tools(app) + if garmin_client: + app = activity_management.register_tools(app) + app = health_wellness.register_tools(app) + app = user_profile.register_tools(app) + app = devices.register_tools(app) + app = gear_management.register_tools(app) + app = weight_management.register_tools(app) + app = challenges.register_tools(app) + app = training.register_tools(app) + app = workouts.register_tools(app) + app = data_management.register_tools(app) + app = womens_health.register_tools(app) + app = nutrition.register_tools(app) + app = workout_builders.register_tools(app) + app = courses.register_tools(app) + app = activity_analysis.register_tools(app) # Register resources (workout templates) - app = workout_templates.register_resources(app) + if garmin_client: + app = workout_templates.register_resources(app) # Warn about filter entries that matched no tool (most likely typos) unknown = app.unknown_filter_names() diff --git a/src/garmin_mcp/exercise_catalog.py b/src/garmin_mcp/exercise_catalog.py new file mode 100644 index 00000000..105fb549 --- /dev/null +++ b/src/garmin_mcp/exercise_catalog.py @@ -0,0 +1,616 @@ +"""Public Garmin strength-exercise catalog, search, and MCP tools. + +The canonical identifiers come exclusively from Garmin's ``Exercises.json``. +Labels and the small alias table are presentation/search aids and never create +new valid exercise pairs. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from difflib import SequenceMatcher, get_close_matches +import json +import logging +import os +from pathlib import Path +import re +import tempfile +import time +from typing import Any, Iterable, Mapping +import unicodedata + +import requests + + +EXERCISES_URL = "https://connect.garmin.com/web-data/exercises/Exercises.json" +EXERCISE_LABELS_URL = ( + "https://connect.garmin.com/" + "web-translations/exercise_types/exercise_types.properties" +) +DEFAULT_CACHE_TTL_SECONDS = 7 * 24 * 60 * 60 +REQUEST_TIMEOUT = (5, 20) + +# These are conservative search conveniences, not part of Garmin's catalog. +EXERCISE_ALIASES: dict[str, tuple[str, str]] = { + "reverse crunch": ("CRUNCH", "REVERSE_CRUNCH"), + "abdominal invertido": ("CRUNCH", "REVERSE_CRUNCH"), +} + +logger = logging.getLogger(__name__) + + +class CatalogError(ValueError): + """A safe, client-presentable catalog validation/loading error.""" + + +@dataclass(frozen=True) +class GarminExercise: + category: str + exercise_name: str + display_name: str + category_display_name: str | None + primary_muscles: tuple[str, ...] + secondary_muscles: tuple[str, ...] + + +def humanize(identifier: str) -> str: + return " ".join(part.capitalize() for part in identifier.split("_") if part) + + +def _singularize_token(token: str) -> str: + """Normalize only uncomplicated English plurals used in exercise queries.""" + if len(token) > 4 and token.endswith("ies"): + return token[:-3] + "y" + if len(token) > 2 and token.endswith("s") and not token.endswith( + ("ss", "us", "is") + ): + return token[:-1] + return token + + +def normalize_text(value: str) -> str: + decomposed = unicodedata.normalize("NFKD", str(value).casefold()) + without_marks = "".join(c for c in decomposed if not unicodedata.combining(c)) + separated = re.sub(r"[_\-/\\]+", " ", without_marks) + alphanumeric = re.sub(r"[^\w\s]", " ", separated, flags=re.UNICODE) + return " ".join(_singularize_token(token) for token in alphanumeric.split()) + + +def parse_properties(content: str) -> dict[str, str]: + properties: dict[str, str] = {} + for raw_line in content.splitlines(): + line = raw_line.strip() + if not line or line.startswith(("#", "!")) or "=" not in line: + continue + key, value = line.split("=", 1) + properties[key.strip()] = value.strip() + return properties + + +def _identifier(value: Any) -> str | None: + if not isinstance(value, str) or not value.strip(): + return None + candidate = value.strip().upper() + # Keep Garmin identifiers intact, but reject values that are clearly labels. + if not re.fullmatch(r"[A-Z0-9_]+", candidate): + return None + return candidate + + +def _muscles(metadata: Mapping[str, Any], key: str) -> tuple[str, ...]: + values = metadata.get(key, []) + if not isinstance(values, (list, tuple)): + return () + return tuple( + muscle + for item in values + if (muscle := _identifier(item)) is not None + ) + + +def _exercise_entries(raw: Any) -> Iterable[tuple[Any, Any]]: + if isinstance(raw, dict): + yield from raw.items() + elif isinstance(raw, list): + for item in raw: + if isinstance(item, str): + yield item, {} + elif isinstance(item, dict): + name = item.get("exerciseName") or item.get("key") or item.get("name") + yield name, item + + +def parse_exercises(data: Any, properties: Mapping[str, str] | None = None) -> tuple[GarminExercise, ...]: + if not isinstance(data, dict) or not isinstance(data.get("categories"), dict): + raise CatalogError("Exercise catalog has no valid categories object.") + labels = properties or {} + exercises: list[GarminExercise] = [] + for raw_category, category_data in data["categories"].items(): + category = _identifier(raw_category) + if category is None or not isinstance(category_data, dict): + continue + raw_exercises = category_data.get("exercises") + if not isinstance(raw_exercises, (dict, list)): + continue + category_label = labels.get(f"category_type_{category}") or humanize(category) + for raw_name, raw_metadata in _exercise_entries(raw_exercises): + exercise_name = _identifier(raw_name) + if exercise_name is None: + continue + metadata = raw_metadata if isinstance(raw_metadata, dict) else {} + exercises.append( + GarminExercise( + category=category, + exercise_name=exercise_name, + display_name=labels.get(f"{category}_{exercise_name}") + or humanize(exercise_name), + category_display_name=category_label, + primary_muscles=_muscles(metadata, "primaryMuscles"), + secondary_muscles=_muscles(metadata, "secondaryMuscles"), + ) + ) + if not exercises: + raise CatalogError("Exercise catalog contains no valid exercises.") + unique = {(e.category, e.exercise_name): e for e in exercises} + return tuple(sorted(unique.values(), key=lambda e: (e.category, e.display_name.casefold(), e.exercise_name))) + + +def _without_none(value: Any) -> Any: + if isinstance(value, dict): + return {key: _without_none(item) for key, item in value.items() if item is not None} + if isinstance(value, list): + return [_without_none(item) for item in value] + return value + + +def _serialize(payload: dict[str, Any]) -> str: + return json.dumps(_without_none(payload), indent=2, ensure_ascii=False) + + +class ExerciseCatalog: + def __init__( + self, + exercises: Iterable[GarminExercise], + *, + cache_status: str = "fresh", + aliases: Mapping[str, tuple[str, str]] | None = None, + ) -> None: + self.exercises = tuple(sorted(exercises, key=lambda e: (e.category, e.display_name.casefold(), e.exercise_name))) + self.cache_status = cache_status + self._pairs = {(e.category, e.exercise_name): e for e in self.exercises} + self._categories = {e.category: e.category_display_name for e in self.exercises} + self.aliases: dict[str, tuple[str, str]] = {} + for alias, pair in (aliases if aliases is not None else EXERCISE_ALIASES).items(): + normalized_pair = (str(pair[0]).upper(), str(pair[1]).upper()) + if normalized_pair in self._pairs: + self.aliases[normalize_text(alias)] = normalized_pair + else: + logger.warning("Ignoring exercise alias %r with invalid Garmin pair", alias) + + def validate_pair(self, category: str, exercise_name: str) -> GarminExercise | None: + return self._pairs.get((category.strip().upper(), exercise_name.strip().upper())) + + def resolve_category(self, category: str) -> str | None: + normalized = normalize_text(category) + matches = [ + key for key, label in self._categories.items() + if normalized in {normalize_text(key), normalize_text(label or "")} + ] + return matches[0] if len(matches) == 1 else None + + def category_error(self, category: str) -> dict[str, Any]: + choices = list(self._categories) + normalized_map = {normalize_text(key): key for key in choices} + normalized_map.update({normalize_text(label or ""): key for key, label in self._categories.items()}) + close = get_close_matches(normalize_text(category), list(normalized_map), n=5, cutoff=0.35) + return { + "status": "error", + "error": "unknown_category", + "message": f"Unknown Garmin exercise category: {category}", + "closest_categories": list(dict.fromkeys(normalized_map[item] for item in close)), + } + + @staticmethod + def exercise_dict(exercise: GarminExercise, include_muscles: bool = False) -> dict[str, Any]: + result: dict[str, Any] = { + "category": exercise.category, + "exercise_name": exercise.exercise_name, + "display_name": exercise.display_name, + "category_display_name": exercise.category_display_name, + } + if include_muscles: + result.update(primary_muscles=list(exercise.primary_muscles), secondary_muscles=list(exercise.secondary_muscles)) + return _without_none(result) + + def list(self, category: str | None = None, search: str | None = None) -> tuple[list[GarminExercise] | None, dict[str, Any] | None, str | None]: + resolved = None + candidates = list(self.exercises) + if category is not None: + resolved = self.resolve_category(category) + if resolved is None: + return None, self.category_error(category), None + candidates = [e for e in candidates if e.category == resolved] + if search and (query := normalize_text(search)): + query_tokens = set(query.split()) + filtered = [] + for exercise in candidates: + fields = ( + exercise.exercise_name, + exercise.display_name, + exercise.category, + exercise.category_display_name or "", + *exercise.primary_muscles, + *exercise.secondary_muscles, + ) + combined = normalize_text(" ".join(fields)) + if query in combined or query_tokens.issubset(set(combined.split())): + filtered.append(exercise) + candidates = filtered + return candidates, None, resolved + + def _exact_pair(self, query: str, candidates: list[GarminExercise]) -> GarminExercise | None: + compact = query.strip().upper() + for exercise in candidates: + forms = { + f"{exercise.category}/{exercise.exercise_name}", + f"{exercise.category}:{exercise.exercise_name}", + f"{exercise.category}_{exercise.exercise_name}", + } + if compact in forms: + return exercise + return None + + @staticmethod + def _score(query: str, exercise: GarminExercise) -> float: + candidate_forms = [ + normalize_text(exercise.display_name), + normalize_text(exercise.exercise_name), + normalize_text(f"{exercise.category} {exercise.display_name}"), + normalize_text(f"{exercise.category} {exercise.exercise_name}"), + ] + query_tokens = set(query.split()) + best = 0.0 + for form in candidate_forms: + candidate_tokens = set(form.split()) + union = query_tokens | candidate_tokens + intersection = query_tokens & candidate_tokens + sequence = SequenceMatcher(None, query, form).ratio() + jaccard = len(intersection) / len(union) if union else 0.0 + coverage = len(intersection) / len(query_tokens) if query_tokens else 0.0 + best = max(best, 0.50 * sequence + 0.30 * jaccard + 0.20 * coverage) + return round(best, 4) + + def match(self, query: str, category: str | None = None, limit: int = 5) -> dict[str, Any]: + normalized_query = normalize_text(query) + candidates = list(self.exercises) + if category is not None: + resolved = self.resolve_category(category) + if resolved is None: + return self.category_error(category) + candidates = [e for e in candidates if e.category == resolved] + + pair = self._exact_pair(query, candidates) + if pair: + return self._exact_response(query, pair, 1.0) + + key_matches = [e for e in candidates if normalize_text(e.exercise_name) == normalized_query] + if len(key_matches) == 1: + return self._exact_response(query, key_matches[0], 1.0) + if len(key_matches) > 1: + return self._ambiguous(query, [(e, 1.0) for e in key_matches[:limit]]) + + name_matches = [e for e in candidates if normalize_text(e.display_name) == normalized_query] + if len(name_matches) == 1: + return self._exact_response(query, name_matches[0], 1.0) + if len(name_matches) > 1: + return self._ambiguous(query, [(e, 1.0) for e in name_matches[:limit]]) + + alias_pair = self.aliases.get(normalized_query) + if alias_pair and (alias_match := self._pairs.get(alias_pair)) in candidates: + return self._exact_response(query, alias_match, 0.99) + + ranked = sorted( + ((exercise, self._score(normalized_query, exercise)) for exercise in candidates), + key=lambda item: (-item[1], item[0].category, item[0].display_name.casefold(), item[0].exercise_name), + ) + top_score = ranked[0][1] if ranked else 0.0 + alternatives = ranked[:limit] + margin = top_score - ranked[1][1] if len(ranked) > 1 else top_score + if top_score >= 0.82 and margin >= 0.08: + match = alternatives[0][0] + return { + "status": "matched", "query": query, "confidence": top_score, + "match": self.exercise_dict(match), + "alternatives": [self._scored(e, score) for e, score in alternatives[1:]], + } + if top_score >= 0.65: + return self._ambiguous(query, alternatives) + return { + "status": "no_match", "query": query, "confidence": top_score, + "message": "No confident Garmin exercise match was found.", "alternatives": [], + } + + def _exact_response(self, query: str, exercise: GarminExercise, confidence: float) -> dict[str, Any]: + return {"status": "exact", "query": query, "confidence": confidence, "match": self.exercise_dict(exercise), "alternatives": []} + + def _scored(self, exercise: GarminExercise, score: float) -> dict[str, Any]: + return {**self.exercise_dict(exercise), "score": score} + + def _ambiguous(self, query: str, ranked: list[tuple[GarminExercise, float]]) -> dict[str, Any]: + return { + "status": "ambiguous", "query": query, + "confidence": ranked[0][1] if ranked else 0.0, + "message": "Multiple Garmin exercises are similarly plausible.", + "match": None, + "alternatives": [self._scored(e, score) for e, score in ranked], + } + + +@dataclass(frozen=True) +class _CacheConfig: + directory: Path + ttl_seconds: int + exercises_url: str + labels_url: str + + +_memory_catalogs: dict[_CacheConfig, ExerciseCatalog] = {} + + +def _cache_config() -> _CacheConfig: + ttl_raw = os.getenv("GARMIN_EXERCISE_CACHE_TTL_SECONDS", str(DEFAULT_CACHE_TTL_SECONDS)) + try: + ttl = max(0, int(ttl_raw)) + except ValueError: + ttl = DEFAULT_CACHE_TTL_SECONDS + return _CacheConfig( + Path(os.path.expanduser(os.getenv("GARMIN_EXERCISE_CACHE_DIR", "~/.cache/garmin_mcp/exercise_catalog/"))), + ttl, + os.getenv("GARMIN_EXERCISES_URL", EXERCISES_URL), + os.getenv("GARMIN_EXERCISE_LABELS_URL", EXERCISE_LABELS_URL), + ) + + +def clear_memory_cache() -> None: + """Clear process cache (primarily useful for tests and explicit reloads).""" + _memory_catalogs.clear() + + +def _read_cached(config: _CacheConfig) -> tuple[ExerciseCatalog, float] | None: + exercises_path = config.directory / "Exercises.json" + labels_path = config.directory / "exercise_types.properties" + try: + exercises_text = exercises_path.read_text(encoding="utf-8") + labels_text = labels_path.read_text(encoding="utf-8") + exercises = parse_exercises(json.loads(exercises_text), parse_properties(labels_text)) + age = max(time.time() - exercises_path.stat().st_mtime, time.time() - labels_path.stat().st_mtime) + return ExerciseCatalog(exercises), age + except (OSError, ValueError, json.JSONDecodeError, CatalogError): + return None + + +def _download(config: _CacheConfig) -> tuple[str, str, ExerciseCatalog]: + exercise_response = requests.get(config.exercises_url, timeout=REQUEST_TIMEOUT) + exercise_response.raise_for_status() + labels_response = requests.get(config.labels_url, timeout=REQUEST_TIMEOUT) + labels_response.raise_for_status() + exercises_text = exercise_response.text + labels_text = labels_response.text + exercises = parse_exercises(json.loads(exercises_text), parse_properties(labels_text)) + return exercises_text, labels_text, ExerciseCatalog(exercises) + + +def _atomic_write(path: Path, content: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + handle.write(content) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary, path) + except BaseException: + try: + os.unlink(temporary) + except OSError: + pass + raise + + +def load_catalog() -> ExerciseCatalog: + config = _cache_config() + if config in _memory_catalogs: + return _memory_catalogs[config] + cached = _read_cached(config) + if cached and cached[1] <= config.ttl_seconds: + _memory_catalogs[config] = cached[0] + return cached[0] + try: + exercises_text, labels_text, catalog = _download(config) + _atomic_write(config.directory / "Exercises.json", exercises_text) + _atomic_write(config.directory / "exercise_types.properties", labels_text) + _memory_catalogs[config] = catalog + return catalog + except (requests.RequestException, OSError, ValueError, json.JSONDecodeError, CatalogError) as exc: + if cached: + stale = ExerciseCatalog(cached[0].exercises, cache_status="stale") + _memory_catalogs[config] = stale + logger.warning("Using stale Garmin exercise catalog cache: %s", type(exc).__name__) + return stale + raise CatalogError("Garmin exercise catalog is unavailable and no usable cache exists.") from None + + +def _load_error(exc: Exception) -> str: + return _serialize({"status": "error", "source": "garmin_exercise_catalog", "error": "catalog_unavailable", "message": str(exc)}) + + +def resolve_strength_exercises( + exercises: list[dict[str, Any]], limit: int = 5 +) -> dict[str, Any]: + """Resolve a complete set of caller inputs to canonical Garmin identifiers. + + This is deliberately independent of the authenticated Garmin client. The + catalog is loaded once and each returned exercise retains all caller fields. + """ + if not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 10: + return { + "status": "error", "error": "invalid_limit", + "message": "limit must be between 1 and 10.", + "resolved_exercises": [], "items": [], + } + if not isinstance(exercises, list): + return { + "status": "error", "error": "invalid_exercises", + "message": "exercises must be a list of objects.", + "resolved_exercises": [], "items": [], + } + try: + catalog = load_catalog() + except CatalogError as exc: + return { + "status": "catalog_unavailable", "error": "catalog_unavailable", + "message": str(exc), "resolved_exercises": [], "items": [], + } + + resolved: list[dict[str, Any]] = [] + items: list[dict[str, Any]] = [] + for index, exercise in enumerate(exercises): + if not isinstance(exercise, dict): + items.append({ + "index": index, "status": "invalid", "confidence": 0.0, + "message": "Exercise must be an object.", "alternatives": [], + }) + continue + + category = exercise.get("category") + exercise_name = exercise.get("exercise_name") + name = exercise.get("name") + match: dict[str, Any] + + if category is not None and ( + not isinstance(category, str) or not category.strip() + ): + match = {"status": "invalid", "confidence": 0.0, + "message": "category must be a non-empty string.", "alternatives": []} + elif exercise_name is not None and ( + not isinstance(exercise_name, str) or not exercise_name.strip() + ): + match = {"status": "invalid", "confidence": 0.0, + "message": "exercise_name must be a non-empty string.", "alternatives": []} + elif category is not None and exercise_name is not None: + exact = catalog.validate_pair(category, exercise_name) + if exact is None: + # A supplied pair is authoritative: never silently repair one half. + suggestions = catalog.match(exercise_name, category, limit) + match = { + "status": "invalid_pair", "confidence": 0.0, + "message": "category and exercise_name are not a valid Garmin catalog pair.", + "alternatives": suggestions.get("alternatives", []), + } + else: + match = catalog._exact_response(exercise_name, exact, 1.0) + elif exercise_name is not None: + match = catalog.match(exercise_name, category, limit) + elif isinstance(name, str) and name.strip(): + match = catalog.match(name, category, limit) + else: + match = {"status": "invalid", "confidence": 0.0, + "message": "name or exercise_name is required.", "alternatives": []} + + item = { + "index": index, + "status": match.get("status", "error"), + "confidence": match.get("confidence", 0.0), + "alternatives": match.get("alternatives", []), + } + if match.get("message"): + item["message"] = match["message"] + canonical = match.get("match") + if match.get("status") in {"exact", "matched"} and canonical: + output = dict(exercise) + output["category"] = canonical["category"] + output["exercise_name"] = canonical["exercise_name"] + resolved.append(output) + item["resolved_exercise"] = output + items.append(item) + + ready = len(resolved) == len(exercises) + return { + "status": "ready" if ready else "needs_review", + "source": "garmin_exercise_catalog", "cache_status": catalog.cache_status, + "resolved_exercises": resolved, "items": items, + "unresolved_items": [item for item in items if item["status"] not in {"exact", "matched"}], + } + + +def register_tools(app): + @app.tool( + description=( + "List Garmin strength exercises and their exact category/exerciseName " + "identifiers. Supports category filtering, text search, and pagination. " + "Use the returned category and exercise_name fields when creating a " + "strength workout." + ) + ) + async def list_strength_exercises( + category: str | None = None, + search: str | None = None, + limit: int = 50, + offset: int = 0, + include_muscles: bool = False, + ) -> str: + if not 1 <= limit <= 200: + return _serialize({"status": "error", "error": "invalid_limit", "message": "limit must be between 1 and 200."}) + if offset < 0: + return _serialize({"status": "error", "error": "invalid_offset", "message": "offset must not be negative."}) + try: + catalog = load_catalog() + except CatalogError as exc: + return _load_error(exc) + matches, error, resolved = catalog.list(category, search) + if error: + return _serialize(error) + assert matches is not None + page = matches[offset : offset + limit] + total = len(matches) + return _serialize({ + "status": "success", "source": "garmin_exercise_catalog", + "cache_status": catalog.cache_status, "total": total, "offset": offset, + "limit": limit, "has_more": offset + len(page) < total, + "next_offset": offset + len(page) if offset + len(page) < total else None, + "filters": {"category": resolved, "search": search}, + "exercises": [catalog.exercise_dict(e, include_muscles) for e in page], + }) + + @app.tool( + description=( + "Match a human-friendly exercise description to Garmin's exercise " + "catalog. Returns the exact category and exercise_name identifiers, " + "a confidence score, and alternatives. It never creates or modifies " + "a workout." + ) + ) + async def match_strength_exercise(query: str, category: str | None = None, limit: int = 5) -> str: + if not isinstance(query, str) or not query.strip(): + return _serialize({"status": "error", "error": "invalid_query", "message": "query is required and must not be empty."}) + if not 1 <= limit <= 10: + return _serialize({"status": "error", "error": "invalid_limit", "message": "limit must be between 1 and 10."}) + try: + catalog = load_catalog() + except CatalogError as exc: + return _load_error(exc) + return _serialize(catalog.match(query, category, limit)) + + @app.tool( + name="resolve_strength_exercises", + description=( + "Resolve a batch of strength exercise inputs to exact Garmin category " + "and exercise_name identifiers without creating a workout." + ) + ) + async def resolve_strength_exercises_tool( + exercises: list[dict[str, Any]], limit: int = 5 + ) -> str: + return _serialize(resolve_strength_exercises(exercises, limit)) + + return app diff --git a/src/garmin_mcp/workout_builders.py b/src/garmin_mcp/workout_builders.py index 1f53f773..765c6656 100644 --- a/src/garmin_mcp/workout_builders.py +++ b/src/garmin_mcp/workout_builders.py @@ -7,6 +7,8 @@ import json from typing import Any, Dict, List, Optional +from garmin_mcp.exercise_catalog import humanize + # The garmin_client will be set by the main file garmin_client = None @@ -277,7 +279,12 @@ def build_strength_json( step_order = 1 for ex in exercises: - ex_name = ex.get("name", "Exercise") + identifier = ex.get("exercise_name") + if identifier is not None and (not isinstance(identifier, str) or not identifier.strip()): + raise ValueError("exercise_name must be a non-empty string") + ex_name = ex.get("name") + if ex_name is None: + ex_name = humanize(identifier.strip().upper()) if identifier else "Exercise" sets = int(ex.get("sets", 1)) reps = int(ex.get("reps", 1)) rest_seconds = int(ex.get("rest_seconds", 60)) @@ -291,7 +298,7 @@ def build_strength_json( "endCondition": {"conditionTypeId": 10, "conditionTypeKey": "reps"}, "endConditionValue": float(reps), "targetType": {"workoutTargetTypeId": 1, "workoutTargetTypeKey": "no.target"}, - "exerciseName": ex_name, + "exerciseName": identifier.strip().upper() if identifier else ex_name, } # Only set when the caller asked for it: Garmin rejects values outside its own @@ -484,6 +491,7 @@ async def create_z2_walk_workout( async def create_strength_workout( name: str, exercises: List[Dict[str, Any]], + resolve_exercises: bool = False, ) -> str: """Create a strength workout and upload it to Garmin Connect. @@ -500,9 +508,26 @@ async def create_strength_workout( anything else, including "UNASSIGNED" and "OTHER", is rejected with 400 Invalid category. Full list: https://connect.garmin.com/web-data/exercises/Exercises.json + resolve_exercises: Resolve the whole batch against Garmin's public + catalog before upload. Defaults to false for legacy behavior. """ try: - workout_json = build_strength_json(name=name, exercises=exercises) + catalog_warning = None + upload_exercises = exercises + if resolve_exercises: + from garmin_mcp.exercise_catalog import resolve_strength_exercises + + resolution = resolve_strength_exercises(exercises) + if resolution["status"] == "catalog_unavailable": + catalog_warning = ( + "Garmin exercise catalog unavailable; uploaded using legacy " + "exercise inputs without catalog validation." + ) + elif resolution["status"] != "ready": + return json.dumps(resolution, indent=2, ensure_ascii=False) + else: + upload_exercises = resolution["resolved_exercises"] + workout_json = build_strength_json(name=name, exercises=upload_exercises) result = garmin_client.upload_workout(workout_json) if isinstance(result, dict): @@ -512,6 +537,8 @@ async def create_strength_workout( "name": result.get("workoutName"), "message": "Workout uploaded successfully", } + if catalog_warning: + curated["warning"] = catalog_warning curated = {k: v for k, v in curated.items() if v is not None} return json.dumps(curated, indent=2) return json.dumps(result, indent=2) diff --git a/tests/fixtures/exercise_types_sample.properties b/tests/fixtures/exercise_types_sample.properties new file mode 100644 index 00000000..422f05fe --- /dev/null +++ b/tests/fixtures/exercise_types_sample.properties @@ -0,0 +1,17 @@ +# Garmin exercise labels used by tests +CRUNCH_REVERSE_CRUNCH=Reverse Crunch +CRUNCH_SEATED_LEG_U=Seated Leg U +HIP_STABILITY_DEAD_BUG=Dead Bug +PULL_UP_PULL_UP=Pull-up +PULL_UP_CHIN_UP=Chin-up +BENCH_PRESS_DUMBBELL_BENCH_PRESS=Dumbbell Bench Press +LEG_RAISE_HANGING_LEG_RAISE=Hanging Leg Raise +LEG_RAISE_LYING_LEG_RAISE=Lying Leg Raise +category_type_CRUNCH=Crunch +category_type_HIP_STABILITY=Hip Stability +category_type_PULL_UP=Pull-up +category_type_BENCH_PRESS=Bench Press +category_type_LEG_RAISE=Leg Raise +EMPTY_VALUE= +TEXT_WITH_EQUALS=One=Two +UNICODE_LABEL=L'élévation – contrôlée diff --git a/tests/fixtures/exercises_catalog_sample.json b/tests/fixtures/exercises_catalog_sample.json new file mode 100644 index 00000000..d465c16e --- /dev/null +++ b/tests/fixtures/exercises_catalog_sample.json @@ -0,0 +1,24 @@ +{ + "categories": { + "CRUNCH": { + "exercises": { + "REVERSE_CRUNCH": {"primaryMuscles": ["ABS"], "secondaryMuscles": []}, + "SEATED_LEG_U": {} + } + }, + "HIP_STABILITY": {"exercises": {"DEAD_BUG": {}}}, + "PULL_UP": { + "exercises": { + "PULL_UP": {"primaryMuscles": ["LATS"]}, + "CHIN_UP": {"primaryMuscles": ["LATS"], "secondaryMuscles": ["BICEPS"]} + } + }, + "BENCH_PRESS": {"exercises": {"DUMBBELL_BENCH_PRESS": {}}}, + "LEG_RAISE": { + "exercises": { + "HANGING_LEG_RAISE": {}, + "LYING_LEG_RAISE": {} + } + } + } +} diff --git a/tests/integration/test_exercise_catalog_tools.py b/tests/integration/test_exercise_catalog_tools.py new file mode 100644 index 00000000..9afd7979 --- /dev/null +++ b/tests/integration/test_exercise_catalog_tools.py @@ -0,0 +1,107 @@ +"""FastMCP integration tests for the public exercise catalog tools.""" + +import json +from pathlib import Path + +import pytest +from mcp.server.fastmcp import FastMCP + +from garmin_mcp import _ToolFilter, exercise_catalog + + +FIXTURES = Path(__file__).parents[1] / "fixtures" + + +@pytest.fixture +def sample_catalog(): + exercises = json.loads((FIXTURES / "exercises_catalog_sample.json").read_text()) + labels = exercise_catalog.parse_properties( + (FIXTURES / "exercise_types_sample.properties").read_text() + ) + return exercise_catalog.ExerciseCatalog( + exercise_catalog.parse_exercises(exercises, labels) + ) + + +@pytest.fixture +def catalog_app(monkeypatch, sample_catalog): + monkeypatch.setattr(exercise_catalog, "load_catalog", lambda: sample_catalog) + app = FastMCP("Test Exercise Catalog") + return exercise_catalog.register_tools(app) + + +@pytest.mark.asyncio +async def test_tools_registered_with_expected_schema(catalog_app): + tools = await catalog_app.list_tools() + by_name = {tool.name: tool for tool in tools} + assert {"list_strength_exercises", "match_strength_exercise", "resolve_strength_exercises"} <= set(by_name) + list_properties = by_name["list_strength_exercises"].inputSchema["properties"] + match_properties = by_name["match_strength_exercise"].inputSchema["properties"] + assert {"category", "search", "limit", "offset", "include_muscles"} <= set(list_properties) + assert {"query", "category", "limit"} <= set(match_properties) + assert "query" in by_name["match_strength_exercise"].inputSchema["required"] + resolve_properties = by_name["resolve_strength_exercises"].inputSchema["properties"] + assert {"exercises", "limit"} <= set(resolve_properties) + + +@pytest.mark.asyncio +async def test_list_tool_returns_identifiers_and_valid_json(catalog_app): + result = await catalog_app.call_tool( + "list_strength_exercises", {"search": "reverse crunch"} + ) + payload = json.loads(result[0][0].text) + assert payload["status"] == "success" + assert payload["exercises"][0]["category"] == "CRUNCH" + assert payload["exercises"][0]["exercise_name"] == "REVERSE_CRUNCH" + + +@pytest.mark.asyncio +async def test_match_tool_returns_reverse_crunch(catalog_app): + result = await catalog_app.call_tool( + "match_strength_exercise", {"query": "Reverse Crunch"} + ) + payload = json.loads(result[0][0].text) + assert payload["status"] == "exact" + assert payload["match"]["category"] == "CRUNCH" + assert payload["match"]["exercise_name"] == "REVERSE_CRUNCH" + + +@pytest.mark.asyncio +async def test_validation_responses_are_json(catalog_app): + result = await catalog_app.call_tool("list_strength_exercises", {"limit": 201}) + assert json.loads(result[0][0].text)["status"] == "error" + + +@pytest.mark.asyncio +async def test_list_pagination_and_maximum_limit(catalog_app): + first = await catalog_app.call_tool( + "list_strength_exercises", {"limit": 1, "offset": 0} + ) + page = json.loads(first[0][0].text) + assert len(page["exercises"]) == 1 + assert page["has_more"] is True + assert page["next_offset"] == 1 + + maximum = await catalog_app.call_tool( + "list_strength_exercises", {"limit": 200} + ) + assert json.loads(maximum[0][0].text)["status"] == "success" + + +@pytest.mark.asyncio +async def test_tool_name_filtering(monkeypatch, sample_catalog): + monkeypatch.setattr(exercise_catalog, "load_catalog", lambda: sample_catalog) + inner = FastMCP("Filtered Exercise Catalog") + filtered = _ToolFilter(inner, {"match_strength_exercise"}, set()) + exercise_catalog.register_tools(filtered) + names = {tool.name for tool in await inner.list_tools()} + assert "match_strength_exercise" in names + assert "list_strength_exercises" not in names + + +@pytest.mark.asyncio +async def test_tools_do_not_use_authenticated_client(catalog_app, mock_garmin_client): + await catalog_app.call_tool("list_strength_exercises", {"limit": 1}) + await catalog_app.call_tool("match_strength_exercise", {"query": "Reverse Crunch"}) + await catalog_app.call_tool("resolve_strength_exercises", {"exercises": [{"name": "Reverse Crunch"}]}) + assert mock_garmin_client.mock_calls == [] diff --git a/tests/integration/test_workout_builders_tools.py b/tests/integration/test_workout_builders_tools.py index e45adeff..689897c9 100644 --- a/tests/integration/test_workout_builders_tools.py +++ b/tests/integration/test_workout_builders_tools.py @@ -7,7 +7,7 @@ from mcp.server.fastmcp import FastMCP from unittest.mock import MagicMock -from garmin_mcp import workouts, workout_builders +from garmin_mcp import exercise_catalog, workouts, workout_builders @pytest.fixture @@ -216,3 +216,68 @@ async def test_create_run_workout_exception(app_with_builders, mock_garmin_clien assert result is not None assert "Error" in result[0][0].text assert "Upload failed" in result[0][0].text + + +@pytest.mark.asyncio +async def test_create_strength_resolution_uploads_canonical_pair( + app_with_builders, mock_garmin_client, monkeypatch +): + monkeypatch.setattr(exercise_catalog, "resolve_strength_exercises", lambda exercises: { + "status": "ready", + "resolved_exercises": [{**exercises[0], "category": "CRUNCH", "exercise_name": "REVERSE_CRUNCH"}], + }) + mock_garmin_client.upload_workout.return_value = {"workoutId": 42, "workoutName": "Core"} + result = await app_with_builders.call_tool("create_strength_workout", { + "name": "Core", "resolve_exercises": True, + "exercises": [{"name": "Reverse Crunch", "sets": 3, "reps": 10}], + }) + assert json.loads(result[0][0].text)["status"] == "success" + step = mock_garmin_client.upload_workout.call_args.args[0]["workoutSegments"][0]["workoutSteps"][0] + assert (step["category"], step["exerciseName"]) == ("CRUNCH", "REVERSE_CRUNCH") + assert step["description"].startswith("Reverse Crunch:") + + +@pytest.mark.asyncio +async def test_create_strength_resolution_blocks_entire_upload( + app_with_builders, mock_garmin_client, monkeypatch +): + monkeypatch.setattr(exercise_catalog, "resolve_strength_exercises", lambda exercises: { + "status": "needs_review", "resolved_exercises": [], + "unresolved_items": [{"index": 0, "status": "ambiguous", "alternatives": []}], + }) + result = await app_with_builders.call_tool("create_strength_workout", { + "name": "Core", "resolve_exercises": True, "exercises": [{"name": "raise"}], + }) + assert json.loads(result[0][0].text)["status"] == "needs_review" + mock_garmin_client.upload_workout.assert_not_called() + + +@pytest.mark.asyncio +async def test_create_strength_catalog_outage_uses_legacy_upload( + app_with_builders, mock_garmin_client, monkeypatch +): + monkeypatch.setattr(exercise_catalog, "resolve_strength_exercises", lambda exercises: { + "status": "catalog_unavailable", "resolved_exercises": [], + }) + mock_garmin_client.upload_workout.return_value = {"workoutId": 43, "workoutName": "Legacy"} + result = await app_with_builders.call_tool("create_strength_workout", { + "name": "Legacy", "resolve_exercises": True, + "exercises": [{"name": "Custom movement", "sets": 1, "reps": 2}], + }) + payload = json.loads(result[0][0].text) + assert payload["status"] == "success" and "catalog unavailable" in payload["warning"].lower() + mock_garmin_client.upload_workout.assert_called_once() + + +@pytest.mark.asyncio +async def test_create_strength_default_does_not_resolve( + app_with_builders, mock_garmin_client, monkeypatch +): + resolve = MagicMock(side_effect=AssertionError("catalog should not load")) + monkeypatch.setattr(exercise_catalog, "resolve_strength_exercises", resolve) + mock_garmin_client.upload_workout.return_value = {"workoutId": 44, "workoutName": "Legacy"} + await app_with_builders.call_tool("create_strength_workout", { + "name": "Legacy", "exercises": [{"name": "Custom", "sets": 1, "reps": 2}], + }) + resolve.assert_not_called() + mock_garmin_client.upload_workout.assert_called_once() diff --git a/tests/unit/test_exercise_catalog.py b/tests/unit/test_exercise_catalog.py new file mode 100644 index 00000000..b55fa90e --- /dev/null +++ b/tests/unit/test_exercise_catalog.py @@ -0,0 +1,247 @@ +import json +import os +from pathlib import Path + +import pytest +import requests + +from garmin_mcp import exercise_catalog as module +from garmin_mcp.exercise_catalog import ( + CatalogError, + ExerciseCatalog, + GarminExercise, + normalize_text, + parse_exercises, + parse_properties, + resolve_strength_exercises, +) + + +FIXTURES = Path(__file__).parents[1] / "fixtures" + + +@pytest.fixture +def source_texts(): + return ( + (FIXTURES / "exercises_catalog_sample.json").read_text(), + (FIXTURES / "exercise_types_sample.properties").read_text(), + ) + + +@pytest.fixture +def catalog(source_texts): + raw, labels = source_texts + return ExerciseCatalog(parse_exercises(json.loads(raw), parse_properties(labels))) + + +@pytest.fixture(autouse=True) +def reset_memory_cache(): + module.clear_memory_cache() + yield + module.clear_memory_cache() + + +def test_valid_catalog_parsing_and_uppercase_normalization(): + result = parse_exercises({"categories": {"crunch": {"exercises": {"reverse_crunch": {}}}}}) + assert (result[0].category, result[0].exercise_name) == ("CRUNCH", "REVERSE_CRUNCH") + + +def test_properties_parser_preserves_values(): + result = parse_properties("# c\n! c\n A = L'été = bien \nEMPTY=\nNO_SEPARATOR\n") + assert result == {"A": "L'été = bien", "EMPTY": ""} + + +def test_display_name_fallback_and_missing_muscles(): + exercise = parse_exercises({"categories": {"BENCH_PRESS": {"exercises": ["DUMBBELL_BENCH_PRESS"]}}})[0] + assert exercise.display_name == "Dumbbell Bench Press" + assert exercise.category_display_name == "Bench Press" + assert exercise.primary_muscles == exercise.secondary_muscles == () + + +@pytest.mark.parametrize("bad", [{}, {"categories": []}, {"categories": {"BAD": {}}}, {"categories": {}}]) +def test_invalid_or_empty_catalog_rejected(bad): + with pytest.raises(CatalogError): + parse_exercises(bad) + + +@pytest.mark.parametrize( + ("left", "right"), + [ + ("Pull-up", "pull up"), + ("PULL_UP", "pull_up"), + ("élévation", "elevation"), + ("pull-ups", "pull up"), + ], +) +def test_text_normalization(left, right): + assert normalize_text(left) == normalize_text(right) + + +def test_numbers_are_preserved(): + assert normalize_text("3-way calf raises") == "3 way calf raise" + + +@pytest.mark.parametrize("query", ["CRUNCH/REVERSE_CRUNCH", "CRUNCH:REVERSE_CRUNCH", "CRUNCH_REVERSE_CRUNCH"]) +def test_exact_pair_matching(catalog, query): + result = catalog.match(query) + assert result["status"] == "exact" + assert result["confidence"] == 1.0 + assert result["match"]["exercise_name"] == "REVERSE_CRUNCH" + + +def test_exact_key_and_human_name_matching(catalog): + assert catalog.match("REVERSE_CRUNCH")["match"]["category"] == "CRUNCH" + assert catalog.match("Reverse Crunch")["match"]["exercise_name"] == "REVERSE_CRUNCH" + + +def test_category_filtering_accepts_label(catalog): + exercises, error, resolved = catalog.list("Crunch") + assert error is None and resolved == "CRUNCH" + assert {e.exercise_name for e in exercises} == {"REVERSE_CRUNCH", "SEATED_LEG_U"} + + +def test_unknown_category_has_suggestions(catalog): + result = catalog.match("reverse", "Crnch") + assert result["status"] == "error" + assert "CRUNCH" in result["closest_categories"] + + +def test_ambiguous_and_no_match(catalog): + ambiguous = catalog.match("leg raise") + assert ambiguous["status"] == "ambiguous" + assert len(ambiguous["alternatives"]) >= 2 + assert catalog.match("completely unknown exercise name")["status"] == "no_match" + + +def test_valid_and_invalid_aliases(catalog, caplog): + custom = ExerciseCatalog(catalog.exercises, aliases={"abdominal invertido": ("CRUNCH", "REVERSE_CRUNCH"), "bad": ("NO", "PAIR")}) + assert custom.match("abdominal invertido")["confidence"] == 0.99 + assert "bad" not in custom.aliases + assert "Ignoring exercise alias" in caplog.text + + +def test_search_pagination_and_deterministic_order(catalog): + first, _, _ = catalog.list(search="leg raise") + second, _, _ = catalog.list(search="LEG_RAISE") + assert [(e.category, e.exercise_name) for e in first] == [(e.category, e.exercise_name) for e in second] + assert first == sorted(first, key=lambda e: (e.category, e.display_name.casefold(), e.exercise_name)) + + +def test_batch_resolver_preserves_metadata_and_resolves_once(monkeypatch, catalog, mocker): + load = mocker.Mock(return_value=catalog) + monkeypatch.setattr(module, "load_catalog", load) + result = resolve_strength_exercises([ + {"name": "Reverse Crunch", "sets": 3, "reps": 12, "note": "slow"}, + {"exercise_name": "REVERSE_CRUNCH", "rest_seconds": 30}, + ]) + assert result["status"] == "ready" + assert result["resolved_exercises"][0] == { + "name": "Reverse Crunch", "sets": 3, "reps": 12, "note": "slow", + "category": "CRUNCH", "exercise_name": "REVERSE_CRUNCH", + } + assert all(item["status"] == "exact" for item in result["items"]) + load.assert_called_once_with() + + +def test_batch_resolver_blocks_invalid_pair_and_validates_limit(monkeypatch, catalog): + monkeypatch.setattr(module, "load_catalog", lambda: catalog) + result = resolve_strength_exercises([ + {"name": "Keep description", "category": "CRUNCH", "exercise_name": "NOT_REAL"} + ]) + assert result["status"] == "needs_review" + assert result["items"][0]["status"] == "invalid_pair" + assert resolve_strength_exercises([], 0)["error"] == "invalid_limit" + + +def test_batch_resolver_catalog_failure(monkeypatch): + def unavailable(): + raise CatalogError("unavailable") + monkeypatch.setattr(module, "load_catalog", unavailable) + assert resolve_strength_exercises([{"name": "Squat"}])["status"] == "catalog_unavailable" + + +class Response: + def __init__(self, text, status=200): + self.text = text + self.status_code = status + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.HTTPError(f"HTTP {self.status_code}") + + +def _configure_cache(monkeypatch, tmp_path): + monkeypatch.setenv("GARMIN_EXERCISE_CACHE_DIR", str(tmp_path)) + monkeypatch.setenv("GARMIN_EXERCISE_CACHE_TTL_SECONDS", "60") + + +def _write_cache(tmp_path, source_texts, age=0): + raw, labels = source_texts + (tmp_path / "Exercises.json").write_text(raw) + (tmp_path / "exercise_types.properties").write_text(labels) + timestamp = module.time.time() - age + os.utime(tmp_path / "Exercises.json", (timestamp, timestamp)) + os.utime(tmp_path / "exercise_types.properties", (timestamp, timestamp)) + + +def test_fresh_cache_makes_no_request(monkeypatch, tmp_path, source_texts, mocker): + _configure_cache(monkeypatch, tmp_path) + _write_cache(tmp_path, source_texts) + get = mocker.patch.object(module.requests, "get") + assert module.load_catalog().exercises + get.assert_not_called() + + +def test_expired_cache_is_refreshed(monkeypatch, tmp_path, source_texts, mocker): + _configure_cache(monkeypatch, tmp_path) + _write_cache(tmp_path, source_texts, age=120) + raw, labels = source_texts + get = mocker.patch.object(module.requests, "get", side_effect=[Response(raw), Response(labels)]) + assert module.load_catalog().cache_status == "fresh" + assert get.call_count == 2 + + +@pytest.mark.parametrize("failure", [requests.ConnectionError("offline"), Response("server", 500)]) +def test_network_failure_uses_stale_cache(monkeypatch, tmp_path, source_texts, mocker, failure): + _configure_cache(monkeypatch, tmp_path) + _write_cache(tmp_path, source_texts, age=120) + get = mocker.patch.object(module.requests, "get") + if isinstance(failure, Exception): + get.side_effect = failure + else: + get.return_value = failure + assert module.load_catalog().cache_status == "stale" + + +def test_invalid_json_does_not_replace_valid_cache(monkeypatch, tmp_path, source_texts, mocker): + _configure_cache(monkeypatch, tmp_path) + _write_cache(tmp_path, source_texts, age=120) + original = (tmp_path / "Exercises.json").read_text() + mocker.patch.object(module.requests, "get", side_effect=[Response("bad"), Response("")]) + assert module.load_catalog().cache_status == "stale" + assert (tmp_path / "Exercises.json").read_text() == original + + +def test_no_network_and_no_cache_is_safe_error(monkeypatch, tmp_path, mocker): + _configure_cache(monkeypatch, tmp_path) + mocker.patch.object(module.requests, "get", side_effect=requests.ConnectionError("secret details")) + with pytest.raises(CatalogError, match="unavailable") as exc: + module.load_catalog() + assert "secret details" not in str(exc.value) + + +def test_atomic_write_uses_replace_and_leaves_complete_file(tmp_path, mocker): + replace = mocker.spy(module.os, "replace") + target = tmp_path / "catalog.json" + module._atomic_write(target, "complete") + assert target.read_text() == "complete" + replace.assert_called_once() + assert not list(tmp_path.glob(".catalog.json.*")) + + +def test_two_calls_use_memory_cache(monkeypatch, tmp_path, source_texts, mocker): + _configure_cache(monkeypatch, tmp_path) + raw, labels = source_texts + get = mocker.patch.object(module.requests, "get", side_effect=[Response(raw), Response(labels)]) + assert module.load_catalog() is module.load_catalog() + assert get.call_count == 2 diff --git a/tests/unit/test_workout_builders.py b/tests/unit/test_workout_builders.py index 7e07dd19..ca51ac61 100644 --- a/tests/unit/test_workout_builders.py +++ b/tests/unit/test_workout_builders.py @@ -174,6 +174,23 @@ def test_strength_passes_through_supplied_category(): assert "category" not in second +def test_strength_prefers_identifier_and_preserves_description(): + result = build_strength_json( + name="Resolved", + exercises=[{"name": "My carry", "exercise_name": "FARMERS_CARRY", "category": "CARRY"}], + ) + step = _work_steps(result)[0] + assert step["exerciseName"] == "FARMERS_CARRY" + assert step["description"].startswith("My carry:") + + +def test_strength_humanizes_identifier_only_input(): + result = build_strength_json( + name="Resolved", exercises=[{"exercise_name": "REVERSE_CRUNCH", "category": "CRUNCH"}] + ) + assert _work_steps(result)[0]["description"].startswith("Reverse Crunch:") + + def test_strength_rejects_empty_category(): for bad in ("", " ", 5): with pytest.raises(ValueError): diff --git a/uv.lock b/uv.lock index 81d26ef5..d87b6409 100644 --- a/uv.lock +++ b/uv.lock @@ -358,7 +358,7 @@ dev = [ requires-dist = [ { name = "fitparse", specifier = ">=1.2.0" }, { name = "garminconnect", specifier = "==0.3.2" }, - { name = "mcp", specifier = ">=1.28.1" }, + { name = "mcp", specifier = ">=1.28.1,<2" }, { name = "python-dotenv", specifier = "==1.2.2" }, { name = "requests", specifier = "==2.33.0" }, ]