diff --git a/README.md b/README.md index 9ff6006..f6455e8 100644 --- a/README.md +++ b/README.md @@ -117,6 +117,33 @@ When you pull workout details, Tonal gives you remarkably rich per-set data: | `inconsistencyScore` | Rep-to-rep consistency | | `movementSide` | "Both", "Left", or "Right" | +Both the CLI `detail` command and MCP `get_workout_detail` project ROM, +consistency, spotter, eccentric, chains, burnout, drop-set, and duration fields. +The reserved Rest movement (`00000000-0000-0000-0000-000000000005`) is +excluded. Difficulty/consistency sentinels (`-1`) are returned as `null`. +Difficulty, consistency, and suggested-weight placeholder zeros on explicitly +unperformed sets are also `null`, while valid zero scores on performed sets +remain `0`. Actual/count fields stay numeric, and configured mode flags remain +available, when a prescribed set was not performed. + +For catalog movements with `onMachine: false` and `countReps: false`, detail +responses use `measurement_type: "duration"`, expose `duration_sec` and +`prescribed_duration_sec` plus Tonal's separate `duration_based_rep_goal`, and +return resistance-only measurements such as reps, pounds, volume, 1RM, power, +ROM, and difficulty as `null`. Per-movement volume aggregates remain numeric +and use `0` for off-machine work so mixed-workout totals remain safely +summable. Workout `total_volume_lbs` is Tonal's digital-resistance volume; it +does not quantify off-machine bodyweight work. +`measurement_type` describes repetition-versus-duration counting; a timed +on-machine movement can still carry measured resistance fields. + +Raw detail frequently reports `movementSide: "Both"` even for unilateral +movements. Use MCP `get_performance_summary` when you need the formatted +`left` / `right` set objects; `get_workout_detail` does not make a second API +request to join those objects into the raw response. CLI users can get the +same formatted source with `performance ` (`left_side` / +`right_side`). + ## Workout JSON format Workouts are defined as blocks of exercises. Exercises within a block are supersetted. @@ -189,7 +216,7 @@ Add to your Claude Code `settings.json`, Cursor config, or any MCP client: | `get_strength` | Current strength scores | | `get_strength_history` | Strength progression | | `get_profile` | User profile | -| `get_workout_history` | Recent workouts | +| `get_workout_history` | Recent activities, with strength eligibility and optional strength-only filtering | | `get_workout_detail` | Per-set weights, 1RM, power, struggling scores | | `get_performance_summary` | Formatted summary with L/R splits | | `get_exercise_history` | Progressive overload tracking across sessions | @@ -199,6 +226,18 @@ Add to your Claude Code `settings.json`, Cursor config, or any MCP client: | `delete_workout` | Remove custom workout | | `get_volume_report` | Training volume analysis | +`get_workout_history` accepts limits from 1 to 50 and enforces them locally +because Tonal currently returns a fixed 50-row activity page. Its response +includes `returned_count`, `available_in_page`, `requested_limit_satisfied`, +`source_page_exhausted`, and `upstream_page_may_be_truncated`. +`requested_limit_satisfied` means the requested number of matching rows was +returned; `source_page_exhausted` means Tonal returned fewer than 50 source +rows. Neither claims the upstream history is exhaustive. Volume and +exercise-history responses also include completeness metadata; when the flag +is false/true respectively, older activity may exist beyond the upstream page. +Volume reports fail completeness closed and expose +`unparseable_activity_count` when source timestamps are unusable. + ## Token management Tokens expire in ~10 hours. The refresh token lasts much longer. diff --git a/docs/api-reference.md b/docs/api-reference.md index 7f31c29..39d1ef0 100644 --- a/docs/api-reference.md +++ b/docs/api-reference.md @@ -164,7 +164,7 @@ Returns recent workout activities (Tonal sessions, external activities): "activityId": "uuid", "userId": "uuid", "activityTime": "2026-04-11T18:33:21.38Z", - "activityType": "workout", + "activityType": "Internal", "workoutPreview": { "activityId": "uuid", "workoutId": "uuid", @@ -185,12 +185,33 @@ Returns recent workout activities (Tonal sessions, external activities): ] ``` +`activityType` is `Internal` for Tonal strength workouts and `External` for +other logged activities. Only `Internal` activities have workout-detail and +formatted-summary records. + +The live endpoint currently ignores common pagination inputs and returns a +fixed 50-row bare list. The MCP adapter enforces requested history limits +locally and reports page/completeness metadata rather than advertising an +unverified cursor or offset contract. `requested_limit_satisfied` reports +whether the requested number of matching activities was returned, while +`source_page_exhausted` reports whether the observed source page contained +fewer than 50 rows. Volume reports preserve the legacy `days` and `workouts` +fields as aliases, return a stable zero-valued schema, and set `is_complete` +to false when `unparseable_activity_count` is nonzero. + ### Workout Detail (Raw Sets) ``` GET /v6/users/{userId}/workout-activities/{activityId} ``` Returns the full workout with per-set data. This is the richest data source. +The MCP detail tools convert detail-endpoint 404s into structured results: +`no_strength_data` for a known `External` activity, `detail_unavailable` for a +known `Internal` activity, and `activity_not_found` when the ID is absent from a +short or exhausted activity page. If the ID is absent from a full fixed 50-row +page, the result is `activity_type_unknown` with +`activity_lookup_complete: false` and `upstream_page_may_be_truncated: true`. +Each result includes the `activity_id` and HTTP `status`. **Top-level fields:** | Field | Type | Description | @@ -239,12 +260,33 @@ Returns the full workout with per-set data. This is the richest data source. | `beginTime` | string | ISO 8601 | | `endTime` | string | ISO 8601 | +The CLI and MCP detail adapters exclude the reserved Rest movement ID +`00000000-0000-0000-0000-000000000005`. They normalize `-1` difficulty and +consistency sentinels to `null` without discarding legitimate zero scores on +performed sets. For explicitly unperformed sets, placeholder-zero difficulty, +consistency, and suggested-weight values are `null`; actual/count fields remain +numeric zeros and configured mode flags remain available. +Bilateral `StraightBar` load fields use total cable load (2x the upstream +per-arm/base load) consistently in CLI detail, MCP detail, and MCP exercise +history. Volume remains the upstream total and is not doubled. +Off-machine, duration-based movements expose actual `duration_sec`, +`prescribed_duration_sec`, and Tonal's separate `duration_based_rep_goal`; +per-set resistance-only measurements are `null`, and the movement is marked +with `on_machine: false`, `counts_reps: false`, and +`measurement_type: "duration"`. Per-movement volume aggregates stay numeric, +with zero representing no digital-resistance contribution. Top-level volume +also measures Tonal's digital resistance and excludes unmeasured bodyweight +work. `measurement_type` describes repetition-versus-duration counting, so a +timed on-machine movement may still include measured resistance. + ### Formatted Workout Summary ``` GET /v6/formatted/users/{userId}/workout-summaries/{activityId} ``` Returns a pre-aggregated summary with movement names and per-movement totals. Includes left/right side splits for unilateral exercises. +Those split objects exist only on this formatted endpoint. The raw-detail tools +preserve upstream `movementSide` as-is and do not join the two endpoints. **Response structure:** ```json diff --git a/mcp/tonal_mcp.py b/mcp/tonal_mcp.py index 2c2b205..7d9efe6 100644 --- a/mcp/tonal_mcp.py +++ b/mcp/tonal_mcp.py @@ -50,6 +50,9 @@ AUTH0_CLIENT_ID = "ERCyexW-xoVG_Yy3RDe-eV4xsOnRHP6L" GET_TIMEOUT = 15 POST_TIMEOUT = 30 +ACTIVITY_PAGE_SIZE = 50 +REST_MOVEMENT_ID = "00000000-0000-0000-0000-000000000005" +UNPERFORMED_SET_TIME_PREFIX = "0001-01-01T00:00:00" mcp = FastMCP("tonal", instructions="""Tonal smart cable machine integration. Provides muscle readiness, strength tracking, workout history with per-set weights/1RM, @@ -171,6 +174,91 @@ def _movement_map(): return {m["id"]: m for m in movements} +def _load_multiplier(movement): + info = movement.get("onMachineInfo") or {} + return 2 if movement.get("isBilateral") and info.get("accessory") == "StraightBar" else 1 + + +def _number_or_zero(value): + return value if type(value) in (int, float) else 0 + + +def _score_or_none(value): + return round(value, 2) if type(value) in (int, float) and 0 <= value <= 1 else None + + +def _set_was_performed(set_activity): + begin_time = set_activity.get("beginTime") + if isinstance(begin_time, str): + return bool(begin_time) and not begin_time.startswith( + UNPERFORMED_SET_TIME_PREFIX + ) + begin_time_mcb = set_activity.get("beginTimeMCB") + if type(begin_time_mcb) in (int, float): + return begin_time_mcb > 0 + return bool(set_activity.get("repCount")) or bool(set_activity.get("duration")) + + +def _activities(data): + return data if isinstance(data, list) else data.get("activities", data.get("data", [])) + + +def _activity_page(uid): + data = _api_get( + f"/v6/users/{uid}/activities", params={"limit": ACTIVITY_PAGE_SIZE} + ) + return _activities(data) + + +def _activity_datetime(activity): + value = activity.get("activityTime") + if not isinstance(value, str) or not value: + return None + if value.endswith(("Z", "z")): + value = f"{value[:-1]}+00:00" + try: + parsed = datetime.fromisoformat(value) + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + except (TypeError, ValueError): + return None + + +def _is_strength_activity(activity): + return activity.get("activityType") == "Internal" + + +def _strength_activity_data(endpoint, activity_id, uid, activity_type=None): + try: + return _api_get(endpoint) + except ValueError as error: + if str(error).startswith("Tonal API 404:"): + activity_page = None + if activity_type is None: + activity_page = _activity_page(uid) + activity_type = next( + (activity.get("activityType") for activity in activity_page + if activity.get("activityId") == activity_id), + None, + ) + if activity_type == "External": + code = "no_strength_data" + elif activity_type == "Internal": + code = "detail_unavailable" + elif (activity_page is not None + and len(activity_page) == ACTIVITY_PAGE_SIZE): + return { + "error": "activity_type_unknown", + "activity_id": activity_id, + "status": 404, + "activity_lookup_complete": False, + "upstream_page_may_be_truncated": True, + } + else: + code = "activity_not_found" + return {"error": code, "activity_id": activity_id, "status": 404} + raise + + # ── MCP Tools ───────────────────────────────────────────────────────── @mcp.tool() @@ -244,36 +332,91 @@ def get_profile() -> dict: @mcp.tool() -def get_workout_history(limit: int = 10) -> dict: - """Get recent workout history with titles, volume, duration, and target areas.""" +def get_workout_history(limit: int = 10, strength_only: bool = False) -> dict: + """Get recent activity history, optionally limited to Tonal strength workouts.""" + if type(limit) is not int or not 1 <= limit <= ACTIVITY_PAGE_SIZE: + return {"error": "invalid_limit", "min": 1, "max": ACTIVITY_PAGE_SIZE} uid = _uid() - data = _api_get(f"/v6/users/{uid}/activities", params={"limit": limit}) - activities = data if isinstance(data, list) else data.get("activities", data.get("data", [])) + activities = _activity_page(uid) + source_count = len(activities) + if strength_only: + activities = [activity for activity in activities if _is_strength_activity(activity)] + available = len(activities) + activities = activities[:limit] return {"workouts": [{"activity_id": a.get("activityId"), "date": a.get("activityTime","")[:10], + "activity_type": a.get("activityType") or "Unknown", + "has_strength_data": _is_strength_activity(a), "title": a.get("workoutPreview",{}).get("workoutTitle",""), "duration_min": round(a.get("workoutPreview",{}).get("totalDuration",0)/60), "total_volume_lbs": a.get("workoutPreview",{}).get("totalVolume"), "target_area": a.get("workoutPreview",{}).get("targetArea","")} - for a in activities]} + for a in activities], + "returned_count": len(activities), "available_in_page": available, + "source_count": source_count, + "more_available_in_page": available > len(activities), + "requested_limit_satisfied": len(activities) == limit, + "source_page_exhausted": source_count < ACTIVITY_PAGE_SIZE, + "upstream_page_may_be_truncated": source_count == ACTIVITY_PAGE_SIZE} @mcp.tool() def get_workout_detail(activity_id: str) -> dict: - """Get full workout detail with per-set actual weights, reps, 1RM, power, and struggling scores.""" + """Get raw per-set load, ROM, consistency, mode flags, power, and difficulty metrics.""" uid = _uid() - data = _api_get(f"/v6/users/{uid}/workout-activities/{activity_id}") + data = _strength_activity_data( + f"/v6/users/{uid}/workout-activities/{activity_id}", activity_id, uid + ) + if "error" in data: + return data mm = _movement_map() movements = {} for s in data.get("workoutSetActivity", []): mid = s.get("movementId", "") + if mid == REST_MOVEMENT_ID: + continue + movement = mm.get(mid, {}) + on_machine = movement.get("onMachine") is not False + counts_reps = movement.get("countReps") is not False if mid not in movements: - m = mm.get(mid, {}) - movements[mid] = {"name": m.get("name", mid[:8]), "movement_id": mid, "sets": [], "warmup_sets": []} - sd = {"reps": s.get("repCount", 0), "weight_lbs": s.get("baseWeight", 0), - "volume_lbs": s.get("volume", 0), "one_rep_max": round(s.get("oneRepMax", 0)) or None, - "max_power_watts": round(s.get("maxConPower", 0)) or None, - "struggling_score": round(s.get("strugglingScore", 0), 2) if s.get("strugglingScore") else None, - "suggested_weight": round(s.get("suggestedWeight", 0), 1) if s.get("suggestedWeight") else None, + movements[mid] = {"name": movement.get("name", mid[:8]), "movement_id": mid, + "on_machine": on_machine, "counts_reps": counts_reps, + "sets": [], "warmup_sets": []} + multiplier = _load_multiplier(movement) + performed = _set_was_performed(s) + rom = s.get("rom") + inconsistency = s.get("inconsistencyScore") + suggested = s.get("suggestedWeight") + sd = {"reps": s.get("repCount", 0) if counts_reps else None, + "weight_lbs": (_number_or_zero(s.get("baseWeight")) * multiplier + if on_machine else None), + "volume_lbs": s.get("volume", 0) if on_machine else None, + "one_rep_max": (round(s.get("oneRepMax", 0) * multiplier) + if on_machine and s.get("oneRepMax") else None), + "max_power_watts": (round(s.get("maxConPower", 0)) or None + if on_machine else None), + "rom_inches": (round(rom, 1) + if on_machine and rom is not None and rom > 0 else None), + "inconsistency_score": (_score_or_none(inconsistency) + if on_machine and performed else None), + "struggling_score": (_score_or_none(s.get("strugglingScore")) + if on_machine and performed else None), + "suggested_weight": (round(suggested * multiplier, 1) + if (on_machine + and performed + and type(suggested) in (int, float) + and suggested >= 0) + else None), + "spotter": (bool(s["spotter"]) if s.get("spotter") is not None + else s.get("spotterMode", "OFF") not in (None, "", "OFF")), + "eccentric": bool(s.get("eccentric", False)), + "chains": bool(s.get("chains", False)), + "burnout": bool(s.get("burnout", False)), + "drop_set": bool(s.get("dropSet", False)), + "duration_sec": s.get("duration"), + "prescribed_duration_sec": (s.get("prescribedDuration") + if not counts_reps else None), + "duration_based_rep_goal": (s.get("durationBasedRepGoal") + if not counts_reps else None), "side": s.get("movementSide", "Both")} if s.get("warmUp"): movements[mid]["warmup_sets"].append(sd) @@ -283,15 +426,32 @@ def get_workout_detail(activity_id: str) -> dict: summaries = [] for mid, m in movements.items(): ws = m["sets"] - if ws: + warmup_sets = m["warmup_sets"] + if ws or warmup_sets: weights = [s["weight_lbs"] for s in ws if s["weight_lbs"]] + working_volume = sum((s["volume_lbs"] or 0) for s in ws) + warmup_volume = sum((s["volume_lbs"] or 0) for s in warmup_sets) + avg_weight = (round(sum(weights)/len(weights), 1) + if weights else (0 if m["on_machine"] else None)) summaries.append({"name": m["name"], "movement_id": mid, - "working_sets": len(ws), "warmup_sets": len(m["warmup_sets"]), - "avg_weight_lbs": round(sum(weights)/len(weights),1) if weights else 0, - "total_reps": sum(s["reps"] for s in ws), - "total_volume_lbs": sum(s["volume_lbs"] for s in ws), + "on_machine": m["on_machine"], + "counts_reps": m["counts_reps"], + "measurement_type": ("repetitions" + if m["counts_reps"] else "duration"), + "working_sets": len(ws), "warmup_sets": len(warmup_sets), + "avg_weight_lbs": avg_weight, + "total_reps": (sum((s["reps"] or 0) for s in ws) + if m["counts_reps"] else None), + "total_duration_sec": sum( + (s["duration_sec"] or 0) + for s in ws + warmup_sets + ), + "working_volume_lbs": working_volume, + "warmup_volume_lbs": warmup_volume, + "total_volume_lbs": working_volume + warmup_volume, "best_1rm": max((s["one_rep_max"] for s in ws if s["one_rep_max"]), default=None), - "set_details": ws}) + "set_details": ws, + "warmup_set_details": warmup_sets}) return {"activity_id": data.get("id"), "total_duration_min": round(data.get("totalDuration",0)/60), "total_volume_lbs": data.get("totalVolume"), "percent_completed": data.get("percentCompleted"), "movements": summaries} @@ -301,7 +461,11 @@ def get_workout_detail(activity_id: str) -> dict: def get_performance_summary(activity_id: str) -> dict: """Get formatted workout summary with movement names, per-set weights, and L/R side splits.""" uid = _uid() - data = _api_get(f"/v6/formatted/users/{uid}/workout-summaries/{activity_id}") + data = _strength_activity_data( + f"/v6/formatted/users/{uid}/workout-summaries/{activity_id}", activity_id, uid + ) + if "error" in data: + return data result = {"workout_name": data.get("name",""), "coach": data.get("coachName",""), "target_area": data.get("targetArea",""), "date": data.get("localTimestamp","")[:10], "movements": []} @@ -312,11 +476,17 @@ def get_performance_summary(activity_id: str) -> dict: "one_rep_max": s.get("oneRepMax"), "warm_up": s.get("warmUp", False), "suggested_weight_change": s.get("suggestedWeightChange", 0)} if s.get("leftSideMovementSet"): - si["left"] = {"reps": s["leftSideMovementSet"].get("repCount"), - "weight_lbs": s["leftSideMovementSet"].get("weight")} + left = s["leftSideMovementSet"] + si["left"] = {"reps": left.get("repCount"), "weight_lbs": left.get("weight"), + "one_rep_max": left.get("oneRepMax"), + "max_power_watts": left.get("maxConPower"), + "volume_lbs": left.get("totalVolume", 0)} if s.get("rightSideMovementSet"): - si["right"] = {"reps": s["rightSideMovementSet"].get("repCount"), - "weight_lbs": s["rightSideMovementSet"].get("weight")} + right = s["rightSideMovementSet"] + si["right"] = {"reps": right.get("repCount"), "weight_lbs": right.get("weight"), + "one_rep_max": right.get("oneRepMax"), + "max_power_watts": right.get("maxConPower"), + "volume_lbs": right.get("totalVolume", 0)} mov["sets"].append(si) result["movements"].append(mov) return result @@ -354,26 +524,38 @@ def get_exercise_history(exercise_name: str) -> dict: return {"too_many": len(targets), "matches": [{"id": mid, "name": m.get("name")} for mid, m in list(targets.items())[:10]]} - history = _api_get(f"/v6/users/{uid}/activities", params={"limit": 50}) - activities = history if isinstance(history, list) else history.get("activities", history.get("data", [])) + activities = _activity_page(uid) by_date = {} for act in activities: + if not _is_strength_activity(act): + continue aid = act.get("activityId") if not aid: continue - detail = _api_get(f"/v6/users/{uid}/workout-activities/{aid}") - if isinstance(detail, dict) and "error" in detail: + detail = _strength_activity_data( + f"/v6/users/{uid}/workout-activities/{aid}", + aid, + uid, + activity_type=act.get("activityType"), + ) + if "error" in detail: continue for s in detail.get("workoutSetActivity", []): - if s.get("movementId") not in targets or s.get("warmUp"): + movement_id = s.get("movementId") + if movement_id not in targets or s.get("warmUp"): continue + multiplier = _load_multiplier(targets[movement_id]) + base_weight = _number_or_zero(s.get("baseWeight")) + one_rep_max = _number_or_zero(s.get("oneRepMax")) d = act.get("activityTime","")[:10] if d not in by_date: by_date[d] = {"date": d, "workout": act.get("workoutPreview",{}).get("workoutTitle",""), "sets": []} by_date[d]["sets"].append({ - "weight_lbs": s.get("baseWeight", 0), "reps": s.get("repCount", 0), - "volume_lbs": s.get("volume", 0), "one_rep_max": round(s.get("oneRepMax", 0)) or None}) + "weight_lbs": base_weight * multiplier, + "reps": s.get("repCount", 0), + "volume_lbs": s.get("volume", 0), + "one_rep_max": round(one_rep_max * multiplier) or None}) sessions = [] for d in sorted(by_date): @@ -391,7 +573,9 @@ def get_exercise_history(exercise_name: str) -> dict: delta = sessions[-1]["avg_weight_lbs"] - sessions[0]["avg_weight_lbs"] progression = {"weight_change_lbs": round(delta,1), "direction": "increasing" if delta > 0 else "decreasing" if delta < 0 else "flat"} - return {"exercise": name, "sessions_found": len(sessions), "progression": progression, "sessions": sessions} + return {"exercise": name, "sessions_found": len(sessions), "progression": progression, + "sessions": sessions, "source_count": len(activities), + "history_may_be_truncated": len(activities) == ACTIVITY_PAGE_SIZE} @mcp.tool() @@ -479,32 +663,44 @@ def delete_workout(workout_id: str) -> dict: def get_volume_report(days: int = 30) -> dict: """Training volume and frequency analysis over N days, broken down by week and target area.""" uid = _uid() - data = _api_get(f"/v6/users/{uid}/activities", params={"limit": min(days, 100)}) - activities = data if isinstance(data, list) else data.get("activities", data.get("data", [])) - cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat() - recent = [a for a in activities if a.get("activityTime","") >= cutoff] - if not recent: - return {"days": days, "workouts": 0} - - total_vol, total_dur, by_area, by_week = 0, 0, {}, {} - for a in recent: + activities = _activity_page(uid) + cutoff = datetime.now(timezone.utc) - timedelta(days=days) + dated_activities = [(activity, _activity_datetime(activity)) for activity in activities] + activity_times = [activity_time for _, activity_time in dated_activities if activity_time] + unparseable_activity_count = len(dated_activities) - len(activity_times) + oldest = min(activity_times, default=None) + is_complete = (unparseable_activity_count == 0 + and (len(activities) < ACTIVITY_PAGE_SIZE + or bool(oldest and oldest <= cutoff))) + recent = [(activity, activity_time) + for activity, activity_time in dated_activities + if (_is_strength_activity(activity) + and activity_time + and activity_time >= cutoff)] + + total_vol, by_area, by_week = 0, {}, {} + for a, activity_time in recent: p = a.get("workoutPreview", {}) - v, d = p.get("totalVolume", 0), p.get("totalDuration", 0) - total_vol += v; total_dur += d + v = p.get("totalVolume", 0) + total_vol += v area = p.get("targetArea", "OTHER") by_area[area] = by_area.get(area, 0) + 1 - ds = a.get("activityTime","")[:10] - try: - dt = datetime.strptime(ds, "%Y-%m-%d") - ws = (dt - timedelta(days=dt.weekday())).strftime("%Y-%m-%d") - if ws not in by_week: by_week[ws] = {"sessions": 0, "volume_lbs": 0} - by_week[ws]["sessions"] += 1; by_week[ws]["volume_lbs"] += v - except Exception: pass - - return {"period_days": days, "total_workouts": len(recent), - "workouts_per_week": round(len(recent)/(days/7), 1), - "total_volume_lbs": total_vol, "avg_volume_per_session": round(total_vol/len(recent)), - "by_target_area": by_area, "by_week": dict(sorted(by_week.items()))} + week_start = activity_time - timedelta(days=activity_time.weekday()) + ws = week_start.date().isoformat() + if ws not in by_week: by_week[ws] = {"sessions": 0, "volume_lbs": 0} + by_week[ws]["sessions"] += 1; by_week[ws]["volume_lbs"] += v + + return {"period_days": days, "days": days, + "total_workouts": len(recent), "workouts": len(recent), + "workouts_per_week": round(len(recent)/(days/7), 1) if days else 0, + "total_volume_lbs": total_vol, + "avg_volume_per_session": round(total_vol/len(recent)) if recent else 0, + "by_target_area": by_area, "by_week": dict(sorted(by_week.items())), + "source_count": len(activities), + "oldest_activity_date": oldest.date().isoformat() if oldest else None, + "unparseable_activity_count": unparseable_activity_count, + "upstream_page_may_be_truncated": len(activities) == ACTIVITY_PAGE_SIZE, + "is_complete": is_complete} if __name__ == "__main__": diff --git a/tests/test_tonal_mcp.py b/tests/test_tonal_mcp.py new file mode 100644 index 0000000..501000d --- /dev/null +++ b/tests/test_tonal_mcp.py @@ -0,0 +1,1022 @@ +import importlib.util +import sys +import types +import unittest +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import call, patch + + +class FakeFastMCP: + def __init__(self, *args, **kwargs): + pass + + def tool(self): + return lambda function: function + + +fastmcp_module = types.ModuleType("mcp.server.fastmcp") +fastmcp_module.FastMCP = FakeFastMCP +sys.modules.setdefault("mcp", types.ModuleType("mcp")) +sys.modules.setdefault("mcp.server", types.ModuleType("mcp.server")) +sys.modules["mcp.server.fastmcp"] = fastmcp_module + +module_path = Path(__file__).parents[1] / "mcp" / "tonal_mcp.py" +spec = importlib.util.spec_from_file_location("tonal_mcp", module_path) +tonal_mcp = importlib.util.module_from_spec(spec) +spec.loader.exec_module(tonal_mcp) + +cli_path = Path(__file__).parents[1] / "tonal_tool.py" +cli_spec = importlib.util.spec_from_file_location("tonal_tool", cli_path) +tonal_tool = importlib.util.module_from_spec(cli_spec) +cli_spec.loader.exec_module(tonal_tool) + + +class WorkoutDetailTests(unittest.TestCase): + def test_normalizes_straight_bar_load_and_includes_warmup_volume(self): + activity = { + "id": "activity-1", + "totalDuration": 600, + "totalVolume": 250, + "workoutSetActivity": [ + { + "movementId": "straight-bar", + "repCount": 10, + "baseWeight": 24.5, + "volume": 50, + "oneRepMax": 34.8, + "suggestedWeight": 43.8, + "rom": 10.25, + "inconsistencyScore": 0.1, + "spotterMode": "SPOTTER", + "warmUp": True, + }, + { + "movementId": "straight-bar", + "repCount": 6, + "baseWeight": 42, + "volume": 100, + "oneRepMax": 52.36, + "suggestedWeight": 49.39, + "rom": 22.345, + "inconsistencyScore": 0.2345, + "spotter": True, + "eccentric": True, + "chains": False, + "burnout": True, + "dropSet": False, + "duration": 31, + "warmUp": False, + }, + { + "movementId": "handles", + "repCount": 6, + "baseWeight": 30, + "volume": 60, + "oneRepMax": 38, + "suggestedWeight": 32, + "rom": 0, + "inconsistencyScore": -1, + "warmUp": False, + }, + { + "movementId": "warmup-only", + "repCount": 4, + "baseWeight": 10, + "volume": 40, + "oneRepMax": 12, + "suggestedWeight": 11, + "warmUp": True, + }, + ], + } + movement_map = { + "straight-bar": { + "name": "Barbell Bench Press", + "isBilateral": True, + "onMachineInfo": {"accessory": "StraightBar"}, + }, + "handles": { + "name": "Bench Press", + "isBilateral": True, + "onMachineInfo": {"accessory": "Handles"}, + }, + "warmup-only": { + "name": "Activation", + "isBilateral": False, + "onMachineInfo": {"accessory": "Handles"}, + }, + } + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", return_value=activity), + patch.object(tonal_mcp, "_movement_map", return_value=movement_map), + ): + result = tonal_mcp.get_workout_detail("activity-1") + + by_id = {movement["movement_id"]: movement for movement in result["movements"]} + straight_bar = by_id["straight-bar"] + handles = by_id["handles"] + warmup_only = by_id["warmup-only"] + self.assertEqual(straight_bar["set_details"][0]["weight_lbs"], 84) + self.assertEqual(straight_bar["set_details"][0]["one_rep_max"], 105) + self.assertEqual(straight_bar["set_details"][0]["suggested_weight"], 98.8) + self.assertEqual(straight_bar["set_details"][0]["rom_inches"], 22.3) + self.assertEqual(straight_bar["set_details"][0]["inconsistency_score"], 0.23) + self.assertTrue(straight_bar["set_details"][0]["spotter"]) + self.assertTrue(straight_bar["set_details"][0]["eccentric"]) + self.assertFalse(straight_bar["set_details"][0]["chains"]) + self.assertTrue(straight_bar["set_details"][0]["burnout"]) + self.assertFalse(straight_bar["set_details"][0]["drop_set"]) + self.assertEqual(straight_bar["set_details"][0]["duration_sec"], 31) + self.assertEqual(straight_bar["warmup_set_details"][0]["rom_inches"], 10.2) + self.assertEqual( + straight_bar["warmup_set_details"][0]["inconsistency_score"], 0.1 + ) + self.assertTrue(straight_bar["warmup_set_details"][0]["spotter"]) + self.assertEqual(straight_bar["working_volume_lbs"], 100) + self.assertEqual(straight_bar["warmup_volume_lbs"], 50) + self.assertEqual(straight_bar["total_volume_lbs"], 150) + + self.assertEqual(handles["set_details"][0]["weight_lbs"], 30) + self.assertEqual(handles["set_details"][0]["one_rep_max"], 38) + self.assertIsNone(handles["set_details"][0]["rom_inches"]) + self.assertIsNone(handles["set_details"][0]["inconsistency_score"]) + self.assertEqual(handles["total_volume_lbs"], 60) + + self.assertEqual(warmup_only["working_sets"], 0) + self.assertEqual(warmup_only["warmup_sets"], 1) + self.assertEqual(warmup_only["working_volume_lbs"], 0) + self.assertEqual(warmup_only["warmup_volume_lbs"], 40) + self.assertEqual(warmup_only["total_volume_lbs"], 40) + self.assertEqual(warmup_only["set_details"], []) + + self.assertEqual(result["total_volume_lbs"], 250) + self.assertEqual( + sum(movement["total_volume_lbs"] for movement in result["movements"]), + result["total_volume_lbs"], + ) + + def test_load_multiplier_requires_bilateral_straight_bar(self): + self.assertEqual( + tonal_mcp._load_multiplier( + {"isBilateral": True, "onMachineInfo": {"accessory": "StraightBar"}} + ), + 2, + ) + self.assertEqual( + tonal_mcp._load_multiplier( + {"isBilateral": False, "onMachineInfo": {"accessory": "StraightBar"}} + ), + 1, + ) + self.assertEqual( + tonal_mcp._load_multiplier( + {"isBilateral": True, "onMachineInfo": {"accessory": "Handles"}} + ), + 1, + ) + + def test_straight_bar_loads_match_cli_mcp_detail_and_exercise_history(self): + activity = { + "id": "activity-1", + "totalVolume": 100, + "workoutSetActivity": [ + { + "movementId": "straight-bar", + "repCount": 4, + "baseWeight": 25, + "maxWeight": 30, + "minWeight": 20, + "volume": 100, + "oneRepMax": 40, + "suggestedWeight": 27.5, + "beginTime": "2026-07-30T12:00:00Z", + } + ], + } + movement_map = { + "straight-bar": { + "name": "Barbell Bench Press", + "onMachine": True, + "countReps": True, + "isBilateral": True, + "onMachineInfo": {"accessory": "StraightBar"}, + } + } + detail_endpoint = "/v6/users/user-1/workout-activities/activity-1" + activities_endpoint = "/v6/users/user-1/activities" + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", return_value=activity) as mcp_get, + patch.object(tonal_mcp, "_movement_map", return_value=movement_map), + ): + mcp_detail = tonal_mcp.get_workout_detail("activity-1") + mcp_get.assert_called_once_with(detail_endpoint) + + with ( + patch.object(tonal_tool, "get_user_id", return_value="user-1"), + patch.object(tonal_tool, "api_get", return_value=activity) as cli_get, + patch.object( + tonal_tool, "_get_movement_map", return_value=movement_map + ), + ): + cli_detail = tonal_tool.cmd_detail(["activity-1"]) + cli_get.assert_called_once_with(detail_endpoint) + + activity_summary = { + "activityId": "activity-1", + "activityTime": "2026-07-30T12:00:00Z", + "activityType": "Internal", + "workoutPreview": {"workoutTitle": "Strength"}, + } + + def history_api_get(endpoint, params=None): + if endpoint == activities_endpoint: + self.assertEqual( + params, {"limit": tonal_mcp.ACTIVITY_PAGE_SIZE} + ) + return [activity_summary] + if endpoint == detail_endpoint: + self.assertIsNone(params) + return activity + self.fail(f"Unexpected API request: {endpoint}") + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object( + tonal_mcp, "_api_get", side_effect=history_api_get + ) as history_get, + patch.object(tonal_mcp, "_movement_map", return_value=movement_map), + ): + history = tonal_mcp.get_exercise_history("Barbell Bench") + + history_get.assert_has_calls( + [ + call( + activities_endpoint, + params={"limit": tonal_mcp.ACTIVITY_PAGE_SIZE}, + ), + call(detail_endpoint), + ] + ) + self.assertEqual(history_get.call_count, 2) + + mcp_set = mcp_detail["movements"][0]["set_details"][0] + cli_set = cli_detail["movements"][0]["set_details"][0] + for field in ("weight_lbs", "one_rep_max", "suggested_weight"): + self.assertEqual(cli_set[field], mcp_set[field]) + self.assertEqual(mcp_set["weight_lbs"], 50) + self.assertEqual(mcp_set["one_rep_max"], 80) + self.assertEqual(mcp_set["suggested_weight"], 55) + self.assertEqual(cli_set["max_weight_lbs"], 60) + self.assertEqual(cli_set["min_weight_lbs"], 40) + self.assertEqual(cli_detail["movements"][0]["max_weight_lbs"], 50) + + self.assertEqual(history["sessions_found"], 1) + history_session = history["sessions"][0] + self.assertEqual(history_session["avg_weight_lbs"], 50) + self.assertEqual(history_session["best_1rm"], 80) + self.assertEqual(history_session["total_volume_lbs"], 100) + + def test_null_base_weight_is_zero_in_mcp_and_cli_detail(self): + activity = { + "id": "activity-1", + "totalVolume": 0, + "workoutSetActivity": [ + { + "movementId": "machine", + "repCount": 5, + "baseWeight": None, + "volume": 0, + "beginTime": "2026-07-30T12:00:00Z", + } + ], + } + movement_map = { + "machine": { + "name": "Bench Press", + "onMachine": True, + "countReps": True, + } + } + detail_endpoint = "/v6/users/user-1/workout-activities/activity-1" + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", return_value=activity) as mcp_get, + patch.object(tonal_mcp, "_movement_map", return_value=movement_map), + ): + mcp_result = tonal_mcp.get_workout_detail("activity-1") + mcp_get.assert_called_once_with(detail_endpoint) + + with ( + patch.object(tonal_tool, "get_user_id", return_value="user-1"), + patch.object(tonal_tool, "api_get", return_value=activity) as cli_get, + patch.object( + tonal_tool, "_get_movement_map", return_value=movement_map + ), + ): + cli_result = tonal_tool.cmd_detail(["activity-1"]) + cli_get.assert_called_once_with(detail_endpoint) + + self.assertEqual( + mcp_result["movements"][0]["set_details"][0]["weight_lbs"], 0 + ) + self.assertEqual( + cli_result["movements"][0]["set_details"][0]["weight_lbs"], 0 + ) + + def test_detail_filters_rest_and_normalizes_duration_movements(self): + activity = { + "id": "activity-1", + "totalVolume": 200, + "workoutSetActivity": [ + { + "movementId": tonal_mcp.REST_MOVEMENT_ID, + "repCount": 0, + "baseWeight": 0, + "volume": 0, + }, + { + "movementId": "bodyweight", + "repCount": 0, + "baseWeight": 5, + "maxWeight": 7, + "minWeight": 3, + "volume": 0, + "oneRepMax": 5, + "suggestedWeight": 10, + "strugglingScore": 0.5, + "inconsistencyScore": 0.25, + "maxConPower": 100, + "rom": 12, + "duration": 45, + "prescribedDuration": 45, + "durationBasedRepGoal": 13, + }, + { + "movementId": "machine", + "repCount": 5, + "baseWeight": 20, + "volume": 100, + "suggestedWeight": 0, + "strugglingScore": 0, + "inconsistencyScore": 0, + "beginTime": "2026-07-30T12:00:00Z", + "endTime": "2026-07-30T12:00:30Z", + }, + { + "movementId": "machine", + "repCount": 5, + "baseWeight": 20, + "volume": 100, + "suggestedWeight": 15, + "strugglingScore": 0.23, + "inconsistencyScore": 0.45, + "beginTime": "2026-07-30T12:01:00Z", + "endTime": "2026-07-30T12:01:30Z", + }, + { + "movementId": "machine", + "repCount": 1, + "baseWeight": 0, + "volume": 0, + "suggestedWeight": -1, + "strugglingScore": -1, + "inconsistencyScore": -1, + "beginTime": "2026-07-30T12:02:00Z", + "endTime": "2026-07-30T12:02:05Z", + }, + { + "movementId": "machine", + "repCount": 0, + "baseWeight": 0, + "volume": 0, + "duration": 0, + "suggestedWeight": 0, + "strugglingScore": 0, + "inconsistencyScore": 0, + "spotter": True, + "beginTime": "0001-01-01T00:00:00Z", + "endTime": "0001-01-01T00:00:00Z", + }, + ], + } + movement_map = { + tonal_mcp.REST_MOVEMENT_ID: { + "name": "Rest", + "onMachine": False, + "countReps": False, + }, + "bodyweight": { + "name": "Beast Pushup", + "onMachine": False, + "countReps": False, + }, + "machine": { + "name": "Bench Press", + "onMachine": True, + "countReps": True, + }, + } + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", return_value=activity), + patch.object(tonal_mcp, "_movement_map", return_value=movement_map), + ): + mcp_result = tonal_mcp.get_workout_detail("activity-1") + with ( + patch.object(tonal_tool, "get_user_id", return_value="user-1"), + patch.object(tonal_tool, "api_get", return_value=activity), + patch.object(tonal_tool, "_get_movement_map", return_value=movement_map), + ): + cli_result = tonal_tool.cmd_detail(["activity-1"]) + + for result in (mcp_result, cli_result): + by_id = { + movement["movement_id"]: movement + for movement in result["movements"] + } + self.assertNotIn(tonal_mcp.REST_MOVEMENT_ID, by_id) + + bodyweight = by_id["bodyweight"] + self.assertFalse(bodyweight["on_machine"]) + self.assertFalse(bodyweight["counts_reps"]) + self.assertEqual(bodyweight["measurement_type"], "duration") + self.assertIsNone(bodyweight["avg_weight_lbs"]) + self.assertIsNone(bodyweight["total_reps"]) + self.assertEqual(bodyweight["total_duration_sec"], 45) + self.assertEqual(bodyweight["total_volume_lbs"], 0) + bodyweight_set = bodyweight["set_details"][0] + for field in ( + "reps", + "weight_lbs", + "volume_lbs", + "one_rep_max", + "max_power_watts", + "rom_inches", + "suggested_weight", + "struggling_score", + "inconsistency_score", + ): + self.assertIsNone(bodyweight_set[field]) + self.assertEqual(bodyweight_set["duration_sec"], 45) + self.assertEqual(bodyweight_set["prescribed_duration_sec"], 45) + self.assertEqual(bodyweight_set["duration_based_rep_goal"], 13) + if "max_weight_lbs" in bodyweight_set: + self.assertIsNone(bodyweight_set["max_weight_lbs"]) + self.assertIsNone(bodyweight_set["min_weight_lbs"]) + + machine_set = by_id["machine"]["set_details"][0] + self.assertEqual(machine_set["suggested_weight"], 0) + self.assertEqual(machine_set["struggling_score"], 0) + self.assertEqual(machine_set["inconsistency_score"], 0) + sentinel_set = by_id["machine"]["set_details"][2] + self.assertIsNone(sentinel_set["suggested_weight"]) + self.assertIsNone(sentinel_set["struggling_score"]) + self.assertIsNone(sentinel_set["inconsistency_score"]) + unperformed_set = by_id["machine"]["set_details"][3] + self.assertEqual(unperformed_set["reps"], 0) + self.assertEqual(unperformed_set["weight_lbs"], 0) + self.assertEqual(unperformed_set["volume_lbs"], 0) + self.assertEqual(unperformed_set["duration_sec"], 0) + self.assertIsNone(unperformed_set["suggested_weight"]) + self.assertIsNone(unperformed_set["struggling_score"]) + self.assertIsNone(unperformed_set["inconsistency_score"]) + self.assertTrue(unperformed_set["spotter"]) + self.assertEqual( + sum(movement["total_volume_lbs"] + for movement in result["movements"]), + result["total_volume_lbs"], + ) + self.assertEqual(cli_result["movements"][1]["avg_struggling"], 0.12) + + def test_cli_detail_exposes_same_performance_fields(self): + activity = { + "id": "activity-1", + "workoutSetActivity": [ + { + "movementId": "movement-1", + "repCount": 5, + "baseWeight": 20, + "volume": 100, + "rom": 12.345, + "inconsistencyScore": 0.4567, + "spotter": True, + "eccentric": True, + "chains": True, + "burnout": False, + "dropSet": True, + "duration": 25, + }, + { + "movementId": "movement-1", + "repCount": 2, + "baseWeight": 10, + "volume": 20, + "rom": None, + "inconsistencyScore": None, + "spotterMode": "SPOTTER", + "warmUp": True, + } + ], + } + with ( + patch.object(tonal_tool, "get_user_id", return_value="user-1"), + patch.object(tonal_tool, "api_get", return_value=activity), + patch.object( + tonal_tool, + "_get_movement_map", + return_value={"movement-1": {"name": "Bench Press"}}, + ), + ): + result = tonal_tool.cmd_detail(["activity-1"]) + + set_detail = result["movements"][0]["set_details"][0] + self.assertEqual(set_detail["rom_inches"], 12.3) + self.assertEqual(set_detail["inconsistency_score"], 0.46) + self.assertTrue(set_detail["spotter"]) + self.assertTrue(set_detail["eccentric"]) + self.assertTrue(set_detail["chains"]) + self.assertFalse(set_detail["burnout"]) + self.assertTrue(set_detail["drop_set"]) + self.assertEqual(set_detail["duration_sec"], 25) + self.assertEqual(result["movements"][0]["working_volume_lbs"], 100) + self.assertEqual(result["movements"][0]["warmup_volume_lbs"], 20) + self.assertEqual(result["movements"][0]["total_volume_lbs"], 120) + warmup_detail = result["movements"][0]["warmup_set_details"][0] + self.assertIsNone(warmup_detail["rom_inches"]) + self.assertIsNone(warmup_detail["inconsistency_score"]) + self.assertTrue(warmup_detail["spotter"]) + + def test_performance_summary_preserves_left_right_splits(self): + formatted = { + "movementSets": [ + { + "movementName": "Single-Arm Press", + "sets": [ + { + "repCount": 8, + "weight": 20, + "leftSideMovementSet": {"repCount": 8, "weight": 19}, + "rightSideMovementSet": { + "repCount": 7, + "weight": 20, + "oneRepMax": 24, + "maxConPower": 100, + "totalVolume": 140, + }, + } + ], + } + ] + } + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object( + tonal_mcp, "_strength_activity_data", return_value=formatted + ), + ): + result = tonal_mcp.get_performance_summary("activity-1") + + set_detail = result["movements"][0]["sets"][0] + self.assertEqual(set_detail["left"]["reps"], 8) + self.assertEqual(set_detail["left"]["weight_lbs"], 19) + self.assertEqual( + set_detail["right"], + { + "reps": 7, + "weight_lbs": 20, + "one_rep_max": 24, + "max_power_watts": 100, + "volume_lbs": 140, + }, + ) + + +class ActivityTypeTests(unittest.TestCase): + def setUp(self): + self.internal = { + "activityId": "internal-1", + "activityTime": "2026-07-29T12:00:00Z", + "activityType": "Internal", + "workoutPreview": { + "workoutTitle": "Strength", + "totalDuration": 600, + "totalVolume": 100, + "targetArea": "FULL BODY", + }, + } + self.external = { + "activityId": "external-1", + "activityTime": "2026-07-28T12:00:00Z", + "activityType": "External", + "workoutPreview": { + "workoutTitle": "", + "totalDuration": 3600, + "totalVolume": 0, + "targetArea": "", + }, + } + + def test_history_exposes_activity_type_and_filters_strength(self): + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object( + tonal_mcp, "_api_get", return_value=[self.internal, self.external] + ), + ): + all_history = tonal_mcp.get_workout_history() + strength_history = tonal_mcp.get_workout_history(strength_only=True) + + self.assertEqual( + [ + (workout["activity_type"], workout["has_strength_data"]) + for workout in all_history["workouts"] + ], + [("Internal", True), ("External", False)], + ) + self.assertEqual( + [workout["activity_id"] for workout in strength_history["workouts"]], + ["internal-1"], + ) + self.assertEqual(all_history["returned_count"], 2) + self.assertEqual(strength_history["available_in_page"], 1) + self.assertFalse(strength_history["requested_limit_satisfied"]) + self.assertTrue(strength_history["source_page_exhausted"]) + self.assertFalse(all_history["upstream_page_may_be_truncated"]) + + def test_history_enforces_local_limit_and_reports_fixed_page(self): + activities = [ + { + **self.internal, + "activityId": f"internal-{index}", + "activityTime": f"2026-07-{29 - index:02d}T12:00:00Z", + } + for index in range(20) + ] + [ + { + **self.external, + "activityId": f"external-{index}", + "activityTime": f"2026-06-{30 - index:02d}T12:00:00Z", + } + for index in range(30) + ] + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", return_value=activities), + ): + one = tonal_mcp.get_workout_history(limit=1) + five = tonal_mcp.get_workout_history(limit=5) + default = tonal_mcp.get_workout_history() + fifty = tonal_mcp.get_workout_history(limit=50) + strength = tonal_mcp.get_workout_history(limit=10, strength_only=True) + + self.assertEqual(len(one["workouts"]), 1) + self.assertEqual(len(five["workouts"]), 5) + self.assertEqual(len(default["workouts"]), 10) + self.assertEqual(len(fifty["workouts"]), 50) + self.assertEqual(len(strength["workouts"]), 10) + self.assertEqual(strength["available_in_page"], 20) + self.assertTrue(strength["more_available_in_page"]) + self.assertTrue(strength["requested_limit_satisfied"]) + self.assertFalse(strength["source_page_exhausted"]) + self.assertTrue(strength["upstream_page_may_be_truncated"]) + self.assertEqual(strength["source_count"], 50) + + def test_strength_limit_reports_incomplete_when_fixed_page_has_four_matches(self): + activities = [ + {**self.internal, "activityId": f"internal-{index}"} + for index in range(4) + ] + [ + {**self.external, "activityId": f"external-{index}"} + for index in range(46) + ] + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", return_value=activities), + ): + history = tonal_mcp.get_workout_history(limit=10, strength_only=True) + + self.assertEqual(history["returned_count"], 4) + self.assertEqual(history["available_in_page"], 4) + self.assertFalse(history["requested_limit_satisfied"]) + self.assertFalse(history["source_page_exhausted"]) + self.assertTrue(history["upstream_page_may_be_truncated"]) + + def test_strength_limit_distinguishes_short_page_from_satisfied_limit(self): + activities = [ + {**self.internal, "activityId": f"internal-{index}"} + for index in range(4) + ] + [ + {**self.external, "activityId": f"external-{index}"} + for index in range(45) + ] + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", return_value=activities), + ): + history = tonal_mcp.get_workout_history(limit=10, strength_only=True) + + self.assertEqual(history["returned_count"], 4) + self.assertFalse(history["requested_limit_satisfied"]) + self.assertTrue(history["source_page_exhausted"]) + + def test_history_rejects_invalid_limits_without_api_call(self): + with patch.object(tonal_mcp, "_api_get") as api_get: + zero = tonal_mcp.get_workout_history(limit=0) + too_large = tonal_mcp.get_workout_history(limit=51) + + self.assertEqual(zero["error"], "invalid_limit") + self.assertEqual(too_large["error"], "invalid_limit") + api_get.assert_not_called() + + def test_volume_report_counts_only_internal_activities(self): + activity_time = datetime.now(timezone.utc).isoformat() + internal = {**self.internal, "activityTime": activity_time} + external = {**self.external, "activityTime": activity_time} + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", return_value=[internal, external]), + ): + report = tonal_mcp.get_volume_report(days=7) + + self.assertEqual(report["total_workouts"], 1) + self.assertEqual(report["total_volume_lbs"], 100) + self.assertEqual(report["avg_volume_per_session"], 100) + self.assertEqual(report["unparseable_activity_count"], 0) + self.assertTrue(report["is_complete"]) + + def test_volume_report_marks_incomplete_fixed_page(self): + activity_time = datetime.now(timezone.utc).isoformat() + activities = [ + { + **self.internal, + "activityId": f"internal-{index}", + "activityTime": activity_time, + } + for index in range(50) + ] + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", return_value=activities), + ): + report = tonal_mcp.get_volume_report(days=365) + + self.assertEqual(report["source_count"], 50) + self.assertFalse(report["is_complete"]) + + def test_volume_report_marks_full_page_complete_when_cutoff_is_covered(self): + activities = [ + { + **self.internal, + "activityId": f"internal-{index}", + "activityTime": "2024-01-01T12:00:00Z", + } + for index in range(50) + ] + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", return_value=activities), + ): + report = tonal_mcp.get_volume_report(days=365) + + self.assertEqual(report["source_count"], 50) + self.assertTrue(report["is_complete"]) + + def test_volume_report_returns_stable_empty_schema(self): + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", return_value=[]), + ): + report = tonal_mcp.get_volume_report(days=30) + + self.assertEqual(report["days"], 30) + self.assertEqual(report["period_days"], 30) + self.assertEqual(report["workouts"], 0) + self.assertEqual(report["total_workouts"], 0) + self.assertEqual(report["workouts_per_week"], 0) + self.assertEqual(report["total_volume_lbs"], 0) + self.assertEqual(report["avg_volume_per_session"], 0) + self.assertEqual(report["by_target_area"], {}) + self.assertEqual(report["by_week"], {}) + self.assertEqual(report["unparseable_activity_count"], 0) + self.assertTrue(report["is_complete"]) + + def test_volume_report_fails_completeness_closed_for_bad_timestamps(self): + activities = [ + { + **self.internal, + "activityId": f"internal-{index}", + "activityTime": "2024-01-01T12:00:00Z", + } + for index in range(49) + ] + [{**self.internal, "activityId": "bad-time", "activityTime": 123}] + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", return_value=activities), + ): + report = tonal_mcp.get_volume_report(days=365) + + self.assertEqual(report["unparseable_activity_count"], 1) + self.assertFalse(report["is_complete"]) + self.assertTrue(report["upstream_page_may_be_truncated"]) + + def test_activity_datetime_accepts_common_iso_forms_and_rejects_bad_values(self): + expected = datetime(2026, 7, 29, 12, tzinfo=timezone.utc) + self.assertEqual( + tonal_mcp._activity_datetime({"activityTime": "2026-07-29T12:00:00Z"}), + expected, + ) + self.assertEqual( + tonal_mcp._activity_datetime( + {"activityTime": "2026-07-29T08:00:00-04:00"} + ), + expected, + ) + self.assertEqual( + tonal_mcp._activity_datetime({"activityTime": "2026-07-29T12:00:00"}), + expected, + ) + for value in (None, 123, {}, "not-a-date"): + self.assertIsNone(tonal_mcp._activity_datetime({"activityTime": value})) + + def test_exercise_history_marks_fixed_page_as_possibly_truncated(self): + activities = [ + {**self.external, "activityId": f"external-{index}"} + for index in range(50) + ] + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object( + tonal_mcp, + "_movement_map", + return_value={"movement-1": {"name": "Bench Press"}}, + ), + patch.object(tonal_mcp, "_api_get", return_value=activities), + ): + history = tonal_mcp.get_exercise_history("Bench") + + self.assertEqual(history["source_count"], 50) + self.assertTrue(history["history_may_be_truncated"]) + + def test_exercise_history_skips_external_detail_lookup(self): + detail = { + "workoutSetActivity": [ + { + "movementId": "movement-1", + "warmUp": False, + "baseWeight": 20, + "repCount": 5, + "volume": 100, + "oneRepMax": 25, + } + ] + } + + def api_get(endpoint, params=None): + if endpoint.endswith("/activities"): + return [self.external, self.internal] + if endpoint.endswith("/workout-activities/internal-1"): + return detail + self.fail(f"Unexpected detail request: {endpoint}") + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object( + tonal_mcp, + "_movement_map", + return_value={"movement-1": {"name": "Bench Press"}}, + ), + patch.object(tonal_mcp, "_api_get", side_effect=api_get), + ): + history = tonal_mcp.get_exercise_history("Bench") + + self.assertEqual(history["sessions_found"], 1) + self.assertEqual(history["sessions"][0]["total_volume_lbs"], 100) + self.assertFalse(history["history_may_be_truncated"]) + + def test_detail_tools_return_structured_result_for_404(self): + def api_get(endpoint, params=None): + if endpoint.endswith("/activities"): + return [self.external] + raise ValueError("Tonal API 404: Not Found") + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", side_effect=api_get), + ): + raw = tonal_mcp.get_workout_detail("external-1") + formatted = tonal_mcp.get_performance_summary("external-1") + + expected = { + "error": "no_strength_data", + "activity_id": "external-1", + "status": 404, + } + self.assertEqual(raw, expected) + self.assertEqual(formatted, expected) + + def test_unknown_activity_404_is_not_classified_as_external(self): + def api_get(endpoint, params=None): + if endpoint.endswith("/activities"): + return [self.external] + raise ValueError("Tonal API 404: Not Found") + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object(tonal_mcp, "_api_get", side_effect=api_get), + ): + result = tonal_mcp.get_workout_detail("unknown-1") + + self.assertEqual( + result, + {"error": "activity_not_found", "activity_id": "unknown-1", "status": 404}, + ) + + def test_unknown_activity_404_reports_incomplete_full_page_lookup(self): + activities = [ + {**self.external, "activityId": f"external-{index}"} + for index in range(tonal_mcp.ACTIVITY_PAGE_SIZE) + ] + detail_endpoint = "/v6/users/user-1/workout-activities/unknown-1" + activities_endpoint = "/v6/users/user-1/activities" + + def api_get(endpoint, params=None): + if endpoint == detail_endpoint: + self.assertIsNone(params) + raise ValueError("Tonal API 404: Not Found") + if endpoint == activities_endpoint: + self.assertEqual( + params, {"limit": tonal_mcp.ACTIVITY_PAGE_SIZE} + ) + return activities + self.fail(f"Unexpected API request: {endpoint}") + + with ( + patch.object(tonal_mcp, "_uid", return_value="user-1"), + patch.object( + tonal_mcp, "_api_get", side_effect=api_get + ) as mocked_api_get, + ): + result = tonal_mcp.get_workout_detail("unknown-1") + + mocked_api_get.assert_has_calls( + [ + call(detail_endpoint), + call( + activities_endpoint, + params={"limit": tonal_mcp.ACTIVITY_PAGE_SIZE}, + ), + ] + ) + self.assertEqual(mocked_api_get.call_count, 2) + self.assertEqual( + result, + { + "error": "activity_type_unknown", + "activity_id": "unknown-1", + "status": 404, + "activity_lookup_complete": False, + "upstream_page_may_be_truncated": True, + }, + ) + + def test_internal_activity_404_is_temporarily_unavailable(self): + with patch.object( + tonal_mcp, + "_api_get", + side_effect=ValueError("Tonal API 404: Not Found"), + ): + result = tonal_mcp._strength_activity_data( + "/detail", "internal-1", "user-1", activity_type="Internal" + ) + + self.assertEqual( + result, + {"error": "detail_unavailable", "activity_id": "internal-1", "status": 404}, + ) + + def test_strength_detail_propagates_non_404_errors(self): + with patch.object( + tonal_mcp, + "_api_get", + side_effect=ValueError("Tonal API 500: unavailable"), + ): + with self.assertRaisesRegex(ValueError, "500"): + tonal_mcp._strength_activity_data("/detail", "activity-1", "user-1") + + +if __name__ == "__main__": + unittest.main() diff --git a/tonal_tool.py b/tonal_tool.py index 780b7a5..75dacf0 100755 --- a/tonal_tool.py +++ b/tonal_tool.py @@ -59,6 +59,8 @@ # Timeouts GET_TIMEOUT = 15 POST_TIMEOUT = 30 +REST_MOVEMENT_ID = "00000000-0000-0000-0000-000000000005" +UNPERFORMED_SET_TIME_PREFIX = "0001-01-01T00:00:00" # ── Token Management ────────────────────────────────────────────────── @@ -253,6 +255,31 @@ def _movement_name(movement_map, mid): return m.get("name", mid[:8]) if m else mid[:8] +def _load_multiplier(movement): + info = movement.get("onMachineInfo") or {} + return 2 if movement.get("isBilateral") and info.get("accessory") == "StraightBar" else 1 + + +def _scaled_load(value, multiplier): + return value * multiplier if type(value) in (int, float) else None + + +def _score_or_none(value): + return round(value, 2) if type(value) in (int, float) and 0 <= value <= 1 else None + + +def _set_was_performed(set_activity): + begin_time = set_activity.get("beginTime") + if isinstance(begin_time, str): + return bool(begin_time) and not begin_time.startswith( + UNPERFORMED_SET_TIME_PREFIX + ) + begin_time_mcb = set_activity.get("beginTimeMCB") + if type(begin_time_mcb) in (int, float): + return begin_time_mcb > 0 + return bool(set_activity.get("repCount")) or bool(set_activity.get("duration")) + + # ── Auth Commands ───────────────────────────────────────────────────── def cmd_auth(args): @@ -481,7 +508,7 @@ def cmd_history(args): return {"workouts": results, "count": len(results)} def cmd_detail(args): - """Full workout detail with per-set actual weights, reps, 1RM, power, struggling score.""" + """Full workout detail with raw per-set load, ROM, consistency, modes, and power.""" if not args: return {"error": "Usage: tonal_tool.py detail "} uid = get_user_id() @@ -498,29 +525,69 @@ def cmd_detail(args): movements = {} for s in data.get("workoutSetActivity", []): mid = s.get("movementId", "unknown") + if mid == REST_MOVEMENT_ID: + continue + movement = movement_map.get(mid, {}) + on_machine = movement.get("onMachine") is not False + counts_reps = movement.get("countReps") is not False + load_multiplier = _load_multiplier(movement) if mid not in movements: movements[mid] = { "movement_id": mid, "name": _movement_name(movement_map, mid), + "on_machine": on_machine, + "counts_reps": counts_reps, "sets": [], "warm_up_sets": [], } + rom = s.get("rom") + performed = _set_was_performed(s) + inconsistency = s.get("inconsistencyScore") + suggested = s.get("suggestedWeight") + weight = _scaled_load( + s.get("baseWeight", s.get("avgWeight", 0)), load_multiplier + ) + if weight is None: + weight = 0 + max_weight = _scaled_load(s.get("maxWeight"), load_multiplier) + min_weight = _scaled_load(s.get("minWeight"), load_multiplier) + one_rep_max = _scaled_load(s.get("oneRepMax"), load_multiplier) set_data = { - "reps": s.get("repCount", s.get("prescribedReps", 0)), - "weight_lbs": s.get("baseWeight", s.get("avgWeight", 0)), - "max_weight_lbs": s.get("maxWeight"), - "min_weight_lbs": s.get("minWeight"), - "volume_lbs": s.get("volume", s.get("totalVolume", 0)), - "one_rep_max": round(s.get("oneRepMax", 0)) if s.get("oneRepMax") else None, - "max_power_watts": round(s.get("maxConPower", 0)) if s.get("maxConPower") else None, - "rom_inches": round(s.get("rom", 0), 1) if s.get("rom") else None, - "struggling_score": round(s.get("strugglingScore", 0), 2) if s.get("strugglingScore") else None, - "suggested_weight": round(s.get("suggestedWeight", 0), 1) if s.get("suggestedWeight") else None, - "spotter": s.get("spotterMode", "OFF") != "OFF", - "eccentric": s.get("eccentric", False), - "chains": s.get("chains", False), + "reps": (s.get("repCount", s.get("prescribedReps", 0)) + if counts_reps else None), + "weight_lbs": weight if on_machine else None, + "max_weight_lbs": max_weight if on_machine else None, + "min_weight_lbs": min_weight if on_machine else None, + "volume_lbs": (s.get("volume", s.get("totalVolume", 0)) + if on_machine else None), + "one_rep_max": (round(one_rep_max) + if on_machine and one_rep_max else None), + "max_power_watts": (round(s.get("maxConPower", 0)) + if on_machine and s.get("maxConPower") else None), + "rom_inches": (round(rom, 1) + if on_machine and rom is not None and rom > 0 else None), + "inconsistency_score": (_score_or_none(inconsistency) + if on_machine and performed else None), + "struggling_score": (_score_or_none(s.get("strugglingScore")) + if on_machine and performed else None), + "suggested_weight": (round(suggested * load_multiplier, 1) + if (on_machine + and performed + and type(suggested) in (int, float) + and suggested >= 0) + else None), + "spotter": (bool(s["spotter"]) if s.get("spotter") is not None + else s.get("spotterMode", "OFF") not in (None, "", "OFF")), + "eccentric": bool(s.get("eccentric", False)), + "chains": bool(s.get("chains", False)), + "burnout": bool(s.get("burnout", False)), + "drop_set": bool(s.get("dropSet", False)), "side": s.get("movementSide", "Both"), "duration_sec": s.get("duration"), + "prescribed_duration_sec": (s.get("prescribedDuration") + if not counts_reps else None), + "duration_based_rep_goal": (s.get("durationBasedRepGoal") + if not counts_reps else None), } if s.get("warmUp"): movements[mid]["warm_up_sets"].append(set_data) @@ -531,29 +598,47 @@ def cmd_detail(args): movement_summaries = [] for mid, m in movements.items(): working_sets = m["sets"] - if working_sets: + warm_up_sets = m["warm_up_sets"] + if working_sets or warm_up_sets: weights = [s["weight_lbs"] for s in working_sets if s["weight_lbs"]] - reps = [s["reps"] for s in working_sets if s["reps"]] - volumes = [s["volume_lbs"] for s in working_sets if s["volume_lbs"]] + reps = [s["reps"] for s in working_sets if s["reps"] is not None] + volumes = [s["volume_lbs"] for s in working_sets if s["volume_lbs"] is not None] + warm_up_volume = sum((s["volume_lbs"] or 0) for s in warm_up_sets) orms = [s["one_rep_max"] for s in working_sets if s["one_rep_max"]] powers = [s["max_power_watts"] for s in working_sets if s["max_power_watts"]] struggles = [s["struggling_score"] for s in working_sets if s["struggling_score"] is not None] + working_volume = sum(volumes) + avg_weight = (round(sum(weights) / len(weights), 1) + if weights else (0 if m["on_machine"] else None)) + max_weight = max(weights) if weights else (0 if m["on_machine"] else None) summary = { "movement_id": mid, "name": m["name"], + "on_machine": m["on_machine"], + "counts_reps": m["counts_reps"], + "measurement_type": ("repetitions" + if m["counts_reps"] else "duration"), "working_sets": len(working_sets), - "warm_up_sets": len(m["warm_up_sets"]), - "avg_weight_lbs": round(sum(weights) / len(weights), 1) if weights else 0, - "max_weight_lbs": max(weights) if weights else 0, - "total_reps": sum(reps), - "total_volume_lbs": sum(volumes), + "warm_up_sets": len(warm_up_sets), + "avg_weight_lbs": avg_weight, + "max_weight_lbs": max_weight, + "total_reps": sum(reps) if m["counts_reps"] else None, + "total_duration_sec": sum( + (s["duration_sec"] or 0) + for s in working_sets + warm_up_sets + ), + "working_volume_lbs": working_volume, + "warmup_volume_lbs": warm_up_volume, + "total_volume_lbs": working_volume + warm_up_volume, "best_1rm": max(orms) if orms else None, "avg_power_watts": round(sum(powers) / len(powers)) if powers else None, "avg_struggling": round(sum(struggles) / len(struggles), 2) if struggles else None, "set_details": working_sets, + "warmup_set_details": warm_up_sets, } # Include Tonal's suggested next weight if available - suggestions = [s["suggested_weight"] for s in working_sets if s.get("suggested_weight")] + suggestions = [s["suggested_weight"] for s in working_sets + if s.get("suggested_weight") is not None] if suggestions: summary["tonal_suggested_weight"] = round(max(suggestions), 1) movement_summaries.append(summary)