#!/usr/bin/env python3 from __future__ import annotations import argparse import os import sys from collections.abc import Callable from datetime import date, timedelta from pathlib import Path from typing import Any sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from dotenv import load_dotenv from garmin_coach_clone.auth import account_summary, login from garmin_coach_clone.coach import ( best_workout_for_date, describe_plans, extract_training_plans, find_date_matches, find_workout_like_objects, gc_api_get, get_fbt_adaptive_workout, is_adaptive_plan, plan_id, task_workout_for_date, ) from garmin_coach_clone.dates import parse_date from garmin_coach_clone.io import dump_redacted_json, print_json from garmin_coach_clone.workouts import ( build_dummy_cycling_workout, calendar_entry_date, calendar_entry_id, calendar_entry_name, find_generated_calendar_entry, find_generated_workout, generated_calendar_entries, generated_workouts, summarize_workout, validate_workout_payload, ) load_dotenv() def main() -> int: parser = argparse.ArgumentParser(description="Probe Garmin Connect workout feasibility.") sub = parser.add_subparsers(dest="command", required=True) sub.add_parser("login", help="Authenticate and print account/device summary.") sub.add_parser("plans", help="List training plans and adaptive/Coach candidates.") sub.add_parser("types", help="Dump Garmin Connect workout type metadata.") report = sub.add_parser("report", help="Run read-only feasibility checks in one pass.") report.add_argument("--limit", type=int, default=25, help="Normal workout list limit.") report.add_argument("--dump-json", action="store_true", help="Write debug/probe_report.json.") calendar = sub.add_parser("calendar", help="List calendar entries for a date.") calendar.add_argument("--date", default="today", help="today, tomorrow, or YYYY-MM-DD") calendar.add_argument("--dump-json", action="store_true") coach = sub.add_parser("coach", help="Inspect Coach/adaptive workout data for a date.") coach.add_argument("--date", default="today", help="today, tomorrow, or YYYY-MM-DD") coach.add_argument("--dump-json", action="store_true") coach_endpoints = sub.add_parser( "coach-endpoints", help="Try read-only candidate endpoints for hidden Coach workout details.", ) coach_endpoints.add_argument("--date", default="today", help="today, tomorrow, or YYYY-MM-DD") coach_endpoints.add_argument( "--dump-json", action="store_true", help="Write debug/coach_endpoint_probe_*.json." ) workouts = sub.add_parser("workouts", help="List normal workouts visible via the API.") workouts.add_argument("--limit", type=int, default=25) workouts.add_argument("--dump-json", action="store_true") workout = sub.add_parser("workout", help="Fetch and summarize a normal workout by ID.") workout.add_argument("workout_id", help="Garmin Connect workout ID") workout.add_argument("--dump-json", action="store_true") generated = sub.add_parser("generated", help="List/delete generated probe or clone workouts.") generated.add_argument("--prefix", default=os.getenv("CLONE_PREFIX", "GCClone")) generated.add_argument("--limit", type=int, default=100) generated.add_argument("--delete-id", help="Delete a generated workout template by ID.") generated.add_argument("--confirm", action="store_true", help="Required with --delete-id.") generated_calendar = sub.add_parser( "generated-calendar", help="List/unschedule generated calendar entries for a month." ) generated_calendar.add_argument( "--date", default="today", help="Any date in the month to inspect." ) generated_calendar.add_argument("--prefix", default=os.getenv("CLONE_PREFIX", "GCClone")) generated_calendar.add_argument( "--unschedule-id", help="Unschedule a generated calendar entry by scheduled ID." ) generated_calendar.add_argument( "--dump-json", action="store_true", help="Write the redacted month calendar JSON." ) generated_calendar.add_argument( "--confirm", action="store_true", help="Required with --unschedule-id." ) dummy = sub.add_parser("dummy", help="Create and optionally schedule a dummy cycling workout.") dummy.add_argument("--date", default="tomorrow", help="today, tomorrow, or YYYY-MM-DD") dummy.add_argument("--dry-run", action="store_true", help="Print JSON without uploading.") dummy.add_argument("--schedule", action="store_true", help="Upload and schedule the workout.") args = parser.parse_args() match args.command: case "login": return command_login() case "plans": return command_plans() case "types": return command_types() case "report": return command_report(args.limit, args.dump_json) case "calendar": return command_calendar(parse_date(args.date), args.dump_json) case "coach": return command_coach(parse_date(args.date), args.dump_json) case "coach-endpoints": return command_coach_endpoints(parse_date(args.date), args.dump_json) case "workouts": return command_workouts(args.limit, args.dump_json) case "workout": return command_workout(args.workout_id, args.dump_json) case "generated": return command_generated(args.prefix, args.limit, args.delete_id, args.confirm) case "generated-calendar": return command_generated_calendar( parse_date(args.date), args.prefix, args.unschedule_id, args.dump_json, args.confirm, ) case "dummy": return command_dummy(parse_date(args.date), args.dry_run, args.schedule) return 2 def command_login() -> int: api = login() print("Authenticated.") print_json(account_summary(api)) return 0 def command_plans() -> int: api = login() response = api.get_training_plans() plans = extract_training_plans(response) print(describe_plans(plans)) adaptive = [plan for plan in plans if is_adaptive_plan(plan)] if adaptive: print("\nAdaptive/Coach-looking plans:") print(describe_plans(adaptive)) else: print("\nNo adaptive/Coach-looking plans were identifiable from the plan list.") return 0 def command_types() -> int: api = login() data = api.connectapi("/workout-service/workout/types") out = Path("debug") / "workout_types.json" dump_redacted_json(out, data) print(f"Redacted Garmin workout type metadata written to {out}") print_json(data) return 0 def command_report(limit: int, dump_json: bool) -> int: api = login() today = date.today() tomorrow = today + timedelta(days=1) target_dates = [today, tomorrow] report: dict[str, Any] = {"dates": [day.isoformat() for day in target_dates]} print("Authenticated.") report["account"] = account_summary(api) _print_section("Account") print_json(report["account"]) plans_response = _capture("training_plans", report, api.get_training_plans) plans = extract_training_plans(plans_response) candidates = [plan for plan in plans if is_adaptive_plan(plan)] report["training_plan_summary"] = { "count": len(plans), "adaptive_candidate_count": len(candidates), "adaptive_candidates": [_plan_summary(plan) for plan in candidates], } _print_section("Training Plans") print(f"Plans returned: {len(plans)}") print(f"Adaptive/Coach-looking candidates: {len(candidates)}") if candidates: print(describe_plans(candidates)) types_response = _capture( "workout_types", report, lambda: api.connectapi("/workout-service/workout/types") ) _print_section("Workout Type Metadata") print("Fetched." if not _is_error(types_response) else types_response["error"]) workouts = _capture("workouts", report, lambda: api.get_workouts(limit=limit)) workout_count = len(workouts) if isinstance(workouts, list) else 0 _print_section("Normal Workouts") print(f"Normal workouts returned: {workout_count}") if isinstance(workouts, list): for workout in workouts[:10]: print( " - id={id} sport={sport} name={name}".format( id=workout.get("workoutId"), sport=(workout.get("sportType") or {}).get("sportTypeKey"), name=workout.get("workoutName"), ) ) report["calendar_summary"] = {} _print_section("Calendar") for target_date in target_dates: key = target_date.isoformat() calendar_data = _capture( f"calendar_{key}", report, lambda target_date=target_date: api.get_scheduled_workouts( target_date.year, target_date.month ), ) matches = find_date_matches(calendar_data, target_date) report["calendar_summary"][key] = {"date_match_count": len(matches)} print(f"{key}: {len(matches)} date-containing entries") report["coach_summary"] = {} _print_section("Coach / Adaptive") if not candidates: print("No adaptive/Coach-looking plans found.") for plan in candidates: pid = plan_id(plan) if pid is None: continue category = str(plan.get("trainingPlanCategory", "")).upper() detail = _capture( f"coach_plan_{pid}", report, lambda pid=pid, category=category: api.get_adaptive_training_plan_by_id(pid) if category == "FBT_ADAPTIVE" else api.get_training_plan_by_id(pid), ) plan_summary: dict[str, Any] = { "name": plan.get("name"), "category": plan.get("trainingPlanCategory"), "workout_like_count": len(find_workout_like_objects(detail)), "dates": {}, } for target_date in target_dates: workout = best_workout_for_date(detail, target_date) plan_summary["dates"][target_date.isoformat()] = { "date_match_count": len(find_date_matches(detail, target_date)), "matched_workout": workout is not None, } report["coach_summary"][str(pid)] = plan_summary print( "Plan {pid} {name}: workout-like={count}, today={today_match}, " "tomorrow={tomorrow_match}".format( pid=pid, name=plan.get("name"), count=plan_summary["workout_like_count"], today_match=plan_summary["dates"][today.isoformat()]["matched_workout"], tomorrow_match=plan_summary["dates"][tomorrow.isoformat()]["matched_workout"], ) ) if dump_json: out = Path("debug") / "probe_report.json" dump_redacted_json(out, report) print(f"\nRedacted probe report written to {out}") _print_section("Next Gate") print( "If dummy upload/schedule works on the Edge and Coach matched_workout is true, " "run clone dry-run." ) print("If Coach matched_workout is false, Coach step extraction is still the missing piece.") return 0 def command_calendar(target_date: date, dump_json: bool) -> int: api = login() data = api.get_scheduled_workouts(target_date.year, target_date.month) matches = find_date_matches(data, target_date) print(f"Calendar entries containing {target_date.isoformat()}: {len(matches)}") for idx, item in enumerate(matches, start=1): name = item.get("workoutName") or item.get("name") or item.get("title") or "" item_id = item.get("workoutId") or item.get("scheduledWorkoutId") or item.get("id") print(f" {idx}. id={item_id} name={name}") if dump_json: out = Path("debug") / f"calendar_{target_date.isoformat()}.json" dump_redacted_json(out, data) print(f"Redacted calendar JSON written to {out}") return 0 def command_coach(target_date: date, dump_json: bool) -> int: api = login() plans_response = api.get_training_plans() plans = extract_training_plans(plans_response) candidates = [plan for plan in plans if is_adaptive_plan(plan)] if not candidates: print("No Garmin Coach/adaptive plan was identifiable in get_training_plans().") if dump_json: out = Path("debug") / "training_plans.json" dump_redacted_json(out, plans_response) print(f"Redacted training plan list written to {out}") return 1 any_workout = False for plan in candidates: pid = plan_id(plan) if pid is None: print(f"Skipping plan without numeric ID: {plan.get('name')}") continue category = str(plan.get("trainingPlanCategory", "")).upper() if category == "FBT_ADAPTIVE": detail = api.get_adaptive_training_plan_by_id(pid) else: detail = api.get_training_plan_by_id(pid) if dump_json: out = Path("debug") / f"coach_plan_{pid}_{target_date.isoformat()}.json" dump_redacted_json(out, detail) print(f"Redacted plan detail written to {out}") workout = best_workout_for_date(detail, target_date) if workout is None: workout = _fetch_fbt_workout_for_date(api, detail, target_date) date_matches = find_date_matches(detail, target_date) workout_like = find_workout_like_objects(detail) print(f"\nPlan {pid}: {plan.get('name')}") print(f"Date matches for {target_date.isoformat()}: {len(date_matches)}") print(f"Workout-like objects with workoutSegments/workoutSteps: {len(workout_like)}") if workout: any_workout = True print(summarize_workout(workout)) if dump_json: out = Path("debug") / f"coach_workout_{target_date.isoformat()}.json" dump_redacted_json(out, workout) print(f"Redacted matched workout JSON written to {out}") if target_date == date.today(): today_out = Path("debug") / "coach_workout_today.json" dump_redacted_json(today_out, workout) print(f"Redacted matched workout JSON also written to {today_out}") else: print("No parseable workoutSegments/workoutSteps found for this date.") return 0 if any_workout else 1 def command_coach_endpoints(target_date: date, dump_json: bool) -> int: api = login() plans = extract_training_plans(api.get_training_plans()) candidates = [plan for plan in plans if is_adaptive_plan(plan)] if not candidates: print("No Garmin Coach/adaptive plan was identifiable in get_training_plans().") return 1 report: dict[str, Any] = {"date": target_date.isoformat(), "plans": []} found_workout_like = False for plan in candidates: pid = plan_id(plan) if pid is None: continue detail = ( api.get_adaptive_training_plan_by_id(pid) if str(plan.get("trainingPlanCategory", "")).upper() == "FBT_ADAPTIVE" else api.get_training_plan_by_id(pid) ) workout = task_workout_for_date(detail, target_date) if workout is None: print(f"Plan {pid}: no taskWorkout found for {target_date.isoformat()}.") continue uuid = workout.get("workoutUuid") calendar_id = _calendar_item_id_for_workout(api, target_date, pid, uuid) endpoints = _coach_endpoint_candidates(pid, uuid, calendar_id, target_date) plan_report: dict[str, Any] = { "plan_id": pid, "plan_name": plan.get("name"), "workout_name": workout.get("workoutName"), "workout_uuid": uuid, "calendar_item_id": calendar_id, "endpoints": [], } report["plans"].append(plan_report) print(f"\nPlan {pid}: {plan.get('name')}") print(f"Workout: {workout.get('workoutName')} uuid={uuid} calendar_id={calendar_id}") for endpoint in endpoints: result = _capture_endpoint(api, endpoint) workout_like_count = len(find_workout_like_objects(result)) if workout_like_count: found_workout_like = True plan_report["endpoints"].append( { "endpoint": endpoint, "result": result, "workout_like_count": workout_like_count, } ) status = "error" if _is_error(result) else "ok" keys = ", ".join(result.keys()) if isinstance(result, dict) else type(result).__name__ print(f" - {status}: {endpoint} workout_like={workout_like_count} keys={keys}") if dump_json: out = Path("debug") / f"coach_endpoint_probe_{target_date.isoformat()}.json" dump_redacted_json(out, report) print(f"\nRedacted Coach endpoint probe written to {out}") if found_workout_like: print("\nAt least one candidate endpoint returned workout-like step data.") return 0 print("\nNo candidate endpoint returned workoutSegments/workoutSteps.") return 1 def command_workouts(limit: int, dump_json: bool) -> int: api = login() workouts = api.get_workouts(limit=limit) print(f"Normal workouts returned: {len(workouts)}") for workout in workouts: print( " - id={id} sport={sport} name={name}".format( id=workout.get("workoutId"), sport=(workout.get("sportType") or {}).get("sportTypeKey"), name=workout.get("workoutName"), ) ) if dump_json: out = Path("debug") / "workouts.json" dump_redacted_json(out, workouts) print(f"Redacted workout list written to {out}") return 0 def command_workout(workout_id: str, dump_json: bool) -> int: api = login() workout = api.get_workout_by_id(workout_id) print(summarize_workout(workout)) if dump_json: out = Path("debug") / f"workout_{workout_id}.json" dump_redacted_json(out, workout) print(f"Redacted workout JSON written to {out}") return 0 def command_generated(prefix: str, limit: int, delete_id: str | None, confirm: bool) -> int: api = login() workouts = api.get_workouts(limit=limit) generated = generated_workouts(workouts, prefix) print(f"Generated workouts with prefix {prefix!r}: {len(generated)}") for workout in generated: print( " - id={id} sport={sport} name={name}".format( id=workout.get("workoutId") or workout.get("id"), sport=(workout.get("sportType") or {}).get("sportTypeKey"), name=workout.get("workoutName"), ) ) if delete_id is None: return 0 if not confirm: print("\nRefusing to delete without --confirm.") return 1 selected = find_generated_workout(workouts, delete_id, prefix) if selected is None: print(f"\nRefusing to delete {delete_id}: not found among generated workouts.") return 1 name = selected.get("workoutName") api.delete_workout(delete_id) print(f"\nDeleted generated workout template {delete_id}: {name}") return 0 def command_generated_calendar( target_date: date, prefix: str, unschedule_id: str | None, dump_json: bool, confirm: bool, ) -> int: api = login() calendar_data = api.get_scheduled_workouts(target_date.year, target_date.month) generated = generated_calendar_entries(calendar_data, prefix) print( f"Generated calendar entries with prefix {prefix!r} in " f"{target_date.year:04d}-{target_date.month:02d}: {len(generated)}" ) for entry in generated: print( " - scheduled_id={id} date={date} name={name}".format( id=calendar_entry_id(entry) or "", date=calendar_entry_date(entry) or "", name=calendar_entry_name(entry) or "", ) ) if dump_json: out = Path("debug") / f"generated_calendar_{target_date:%Y_%m}.json" dump_redacted_json(out, calendar_data) print(f"Redacted calendar JSON written to {out}") if unschedule_id is None: return 0 if not confirm: print("\nRefusing to unschedule without --confirm.") return 1 selected = find_generated_calendar_entry(calendar_data, unschedule_id, prefix) if selected is None: print( f"\nRefusing to unschedule {unschedule_id}: " "not found among generated calendar entries." ) return 1 api.unschedule_workout(unschedule_id) print( "\nUnscheduled generated calendar entry {id}: {name}".format( id=unschedule_id, name=calendar_entry_name(selected) or "", ) ) return 0 def command_dummy(target_date: date, dry_run: bool, schedule: bool) -> int: payload = build_dummy_cycling_workout() validation_errors = validate_workout_payload(payload) if validation_errors: print("Dummy workout payload failed local validation:") for error in validation_errors: print(f" - {error}") return 1 print("Dummy workout payload passed local validation.") if dry_run or not schedule: print("Dry run: dummy cycling workout JSON follows. Nothing was uploaded.") print_json(payload) if not schedule: print("\nPass --schedule to upload and schedule this dummy workout.") return 0 api = login() result: dict[str, Any] = api.upload_workout(payload) workout_id = result.get("workoutId") or result.get("id") if workout_id is None: print("Upload returned no workoutId. Raw response:") print_json(result) return 1 schedule_result = api.schedule_workout(workout_id, target_date.isoformat()) print(f"Uploaded dummy workout ID: {workout_id}") print("Schedule response:") print_json(schedule_result) print_edge_check_instructions(target_date) return 0 def print_edge_check_instructions(target_date: date) -> None: print( f""" Next checks on the Edge 1030 after Garmin Connect sync: 1. Sync the Edge 1030 with Garmin Connect / Garmin Express. 2. Check Training > Workouts for the dummy cycling workout. 3. Check Training > Training Plan > calendar icon > {target_date.isoformat()}. 4. Start a normal course/navigation route, then start the workout and verify both can run. """ ) def _capture(label: str, report: dict[str, Any], func: Callable[[], Any]) -> Any: try: result = func() except Exception as exc: # noqa: BLE001 - report should continue through partial failures result = {"error": f"{type(exc).__name__}: {exc}"} report[label] = result return result def _capture_endpoint(api: Any, endpoint: str) -> Any: try: if endpoint.startswith("/gc-api/"): return gc_api_get(api, endpoint) return api.connectapi(endpoint) except Exception as exc: # noqa: BLE001 - endpoint probing should continue return {"error": f"{type(exc).__name__}: {exc}"} def _calendar_item_id_for_workout( api: Any, target_date: date, plan_id_value: int, workout_uuid: Any ) -> str | None: if workout_uuid is None: return None calendar_data = _capture_endpoint( api, f"/calendar-service/year/{target_date.year}/month/{target_date.month - 1}" ) if _is_error(calendar_data): return None for match in find_date_matches(calendar_data, target_date): if ( match.get("itemType") == "fbtAdaptiveWorkout" and str(match.get("trainingPlanId")) == str(plan_id_value) and str(match.get("workoutUuid")) == str(workout_uuid) ): item_id = match.get("id") return str(item_id) if item_id is not None else None return None def _coach_endpoint_candidates( plan_id_value: int, workout_uuid: Any, calendar_item_id: str | None, target_date: date ) -> list[str]: endpoints = [ f"/gc-api/workout-service/fbt-adaptive/{workout_uuid}" if workout_uuid is not None else "", f"/trainingplan-service/trainingplan/fbt-adaptive/{plan_id_value}", f"/trainingplan-service/trainingplan/fbt-adaptive/{plan_id_value}/workouts", f"/trainingplan-service/trainingplan/fbt-adaptive/{plan_id_value}/workouts/{target_date.isoformat()}", f"/trainingplan-service/trainingplan/fbt-adaptive/{plan_id_value}/tasks", f"/trainingplan-service/trainingplan/fbt-adaptive/{plan_id_value}/tasks/{target_date.isoformat()}", ] endpoints = [endpoint for endpoint in endpoints if endpoint] if workout_uuid is not None: uuid = str(workout_uuid) endpoints.extend( [ f"/workout-service/fbt-adaptive/{uuid}", f"/trainingplan-service/trainingplan/fbt-adaptive/{plan_id_value}/workout/{uuid}", f"/trainingplan-service/trainingplan/fbt-adaptive/{plan_id_value}/workouts/{uuid}", f"/trainingplan-service/trainingplan/fbt-adaptive/{plan_id_value}/task/{uuid}", f"/trainingplan-service/trainingplan/fbt-adaptive/{plan_id_value}/tasks/{uuid}", f"/trainingplan-service/trainingplan/fbt-adaptive/workout/{uuid}", f"/trainingplan-service/trainingplan/fbt-adaptive/workouts/{uuid}", f"/trainingplan-service/trainingplan/workout/{uuid}", f"/workout-service/workout/{uuid}", f"/workout-service/workouts/{uuid}", f"/workout-service/workout/uuid/{uuid}", f"/workout-service/workouts/uuid/{uuid}", ] ) if calendar_item_id is not None: endpoints.extend( [ f"/workout-service/schedule/{calendar_item_id}", f"/calendar-service/{calendar_item_id}", f"/calendar-service/calendar/{calendar_item_id}", f"/calendar-service/event/{calendar_item_id}", ] ) return list(dict.fromkeys(endpoints)) def _fetch_fbt_workout_for_date( api: Any, plan_detail: Any, target_date: date ) -> dict[str, Any] | None: workout = task_workout_for_date(plan_detail, target_date) if workout is None: return None workout_uuid = workout.get("workoutUuid") if not workout_uuid: return None try: return get_fbt_adaptive_workout(api, str(workout_uuid)) except Exception as exc: # noqa: BLE001 - shown as probe output print( f"FBT adaptive workout detail fetch failed for {workout_uuid}: " f"{type(exc).__name__}: {exc}" ) return None def _is_error(value: Any) -> bool: return isinstance(value, dict) and isinstance(value.get("error"), str) def _print_section(title: str) -> None: print(f"\n== {title} ==") def _plan_summary(plan: dict[str, Any]) -> dict[str, Any]: return { "id": plan.get("trainingPlanId") or plan.get("planId") or plan.get("id"), "name": plan.get("name"), "category": plan.get("trainingPlanCategory"), } if __name__ == "__main__": raise SystemExit(main())