Yandrik/garmin-coach-to-cal-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: retention days limit and clone name templating and useful descr.

Browse Source
This commit is contained in:
Yandrik
2026-07-28 16:01:12 +02:00
parent 9141a09b56
commit 6fab8fe4a5
13 changed files with 273 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.env.example
View File

@ -9,8 +9,10 @@ GARMIN_PASSWORD=
GARMIN_TOKEN_STORE=.garmin-tokens
GARMIN_TOKEN_DIR=/data/garmin-tokens
CLONE_PREFIX=GCClone
CLONE_NAME_TEMPLATE={type} {date} ({clone})
OVERWRITE_EXISTING=true
DELETE_OLD_CLONES=false
CLONE_RETENTION_DAYS=5
CHANGE_DETECTION_INTERVAL_MINUTES=30
CHANGE_DETECTION_ACTIVE_WINDOW=05:00-22:00
CHANGE_DETECTION_FIXED_TIMES=05:15,06:15,07:15,08:15,09:15,10:15,11:15,12:15,13:15

2
README.md
View File

@ -67,8 +67,10 @@ APP_SECRET_KEY=replace-with-a-long-random-secret
SYNC_ENABLED=true
SYNC_DAYS_AHEAD=1
CLONE_PREFIX=GCClone
CLONE_NAME_TEMPLATE={type} {date} ({clone})
OVERWRITE_EXISTING=true
DELETE_OLD_CLONES=false
CLONE_RETENTION_DAYS=5
CHANGE_DETECTION_INTERVAL_MINUTES=30
CHANGE_DETECTION_ACTIVE_WINDOW=05:00-22:00
CHANGE_DETECTION_FIXED_TIMES=05:15,06:15,07:15,08:15,09:15,10:15,11:15,12:15,13:15

2
docker-compose.yml
View File

@ -14,8 +14,10 @@ services:
SYNC_ENABLED: "${SYNC_ENABLED:-true}"
SYNC_DAYS_AHEAD: "${SYNC_DAYS_AHEAD:-1}"
CLONE_PREFIX: "${CLONE_PREFIX:-GCClone}"
CLONE_NAME_TEMPLATE: "${CLONE_NAME_TEMPLATE:-{type} {date} ({clone})}"
OVERWRITE_EXISTING: "${OVERWRITE_EXISTING:-true}"
DELETE_OLD_CLONES: "${DELETE_OLD_CLONES:-false}"
CLONE_RETENTION_DAYS: "${CLONE_RETENTION_DAYS:-5}"
CHANGE_DETECTION_INTERVAL_MINUTES: "${CHANGE_DETECTION_INTERVAL_MINUTES:-30}"
CHANGE_DETECTION_ACTIVE_WINDOW: "${CHANGE_DETECTION_ACTIVE_WINDOW:-05:00-22:00}"
CHANGE_DETECTION_FIXED_TIMES: "${CHANGE_DETECTION_FIXED_TIMES:-05:15,06:15,07:15,08:15,09:15,10:15,11:15,12:15,13:15}"

2
src/garmin_coach_clone/app.py
View File

@ -249,6 +249,7 @@ def create_app() -> FastAPI:
active_window: str = Form(...),
fixed_times: str = Form(...),
days_ahead: int = Form(...),
clone_name_template: str = Form(...),
) -> Any:
config = ScheduleConfig(
enabled=enabled == "on",
@ -256,6 +257,7 @@ def create_app() -> FastAPI:
active_window=active_window.strip(),
fixed_times=[item.strip() for item in fixed_times.split(",") if item.strip()],
days_ahead=days_ahead,
clone_name_template=clone_name_template.strip(),
)
try:
validate_schedule_config(config)

3
src/garmin_coach_clone/config.py
View File

@ -17,6 +17,7 @@ DEFAULT_FIXED_TIMES = [
]
DEFAULT_ACTIVE_WINDOW = "05:00-22:00"
DEFAULT_INTERVAL_MINUTES = 30
DEFAULT_CLONE_NAME_TEMPLATE = "{type} {date} ({clone})"
def _project_root() -> Path:
@ -37,6 +38,7 @@ class Settings:
app_secret_key: str
timezone: str
clone_prefix: str
clone_name_template: str
sync_enabled: bool
sync_days_ahead: int
overwrite_existing: bool
@ -72,6 +74,7 @@ def load_settings() -> Settings:
app_secret_key=os.getenv("APP_SECRET_KEY", "replace-with-long-random-secret"),
timezone=os.getenv("TZ", "Europe/Berlin"),
clone_prefix=os.getenv("CLONE_PREFIX", "GCClone"),
clone_name_template=os.getenv("CLONE_NAME_TEMPLATE", DEFAULT_CLONE_NAME_TEMPLATE),
sync_enabled=_bool_env("SYNC_ENABLED", True),
sync_days_ahead=max(0, int(os.getenv("SYNC_DAYS_AHEAD", "1"))),
overwrite_existing=_bool_env("OVERWRITE_EXISTING", True),

28
src/garmin_coach_clone/repository.py
View File

@ -7,6 +7,7 @@ from typing import Any
from .config import (
DEFAULT_ACTIVE_WINDOW,
DEFAULT_CLONE_NAME_TEMPLATE,
DEFAULT_FIXED_TIMES,
DEFAULT_INTERVAL_MINUTES,
Settings,
@ -23,6 +24,7 @@ class ScheduleConfig:
active_window: str
fixed_times: list[str]
days_ahead: int
clone_name_template: str = DEFAULT_CLONE_NAME_TEMPLATE
class Repository:
@ -88,6 +90,8 @@ class Repository:
self.get_setting("change_fixed_times"), self.settings.change_fixed_times
),
days_ahead=int(self.get_setting("sync_days_ahead") or self.settings.sync_days_ahead),
clone_name_template=self.get_setting("clone_name_template")
or self.settings.clone_name_template,
)
def save_schedule_config(self, config: ScheduleConfig) -> None:
@ -97,6 +101,7 @@ class Repository:
self.set_setting("change_active_window", config.active_window)
self.set_setting("change_fixed_times", ",".join(config.fixed_times))
self.set_setting("sync_days_ahead", str(config.days_ahead))
self.set_setting("clone_name_template", config.clone_name_template)
def restore_default_schedule(self) -> ScheduleConfig:
config = ScheduleConfig(
@ -105,6 +110,7 @@ class Repository:
active_window=DEFAULT_ACTIVE_WINDOW,
fixed_times=DEFAULT_FIXED_TIMES.copy(),
days_ahead=self.settings.sync_days_ahead,
clone_name_template=self.settings.clone_name_template,
)
self.save_schedule_config(config)
return config
@ -315,6 +321,28 @@ def validate_schedule_config(config: ScheduleConfig) -> None:
raise ValueError(f"invalid fixed time: {value}")
if config.days_ahead < 0 or config.days_ahead > 14:
raise ValueError("days ahead must be between 0 and 14")
_validate_clone_name_template(config.clone_name_template)
def _validate_clone_name_template(value: str) -> None:
if not value.strip():
raise ValueError("clone name template cannot be blank")
allowed = {"type", "date", "clone"}
import string
try:
fields = {
field_name
for _, field_name, _, _ in string.Formatter().parse(value)
if field_name
}
except ValueError as exc:
raise ValueError(f"invalid clone name template: {exc}") from exc
unknown = fields - allowed
if unknown:
raise ValueError("unknown clone name template token(s): " + ", ".join(sorted(unknown)))
if "clone" not in fields:
raise ValueError("clone name template must include {clone}")
def _split_window(value: str) -> tuple[str, str]:

56
src/garmin_coach_clone/sync_service.py
View File

@ -159,7 +159,12 @@ class SyncService:
source_name = str(
source.get("workoutName") or task_workout.get("workoutName") or "Coach Workout"
)
payload = clone_workout_payload(source, target_date, self.settings.clone_prefix)
payload = clone_workout_payload(
source,
target_date,
self.settings.clone_prefix,
self.repo.schedule_config().clone_name_template,
)
errors = validate_workout_payload(payload)
if errors:
return self._trace(
@ -187,6 +192,8 @@ class SyncService:
target_date,
payload,
)
if current_workout and current_workout.get("workoutName"):
clone_name = str(current_workout["workoutName"])
scheduled_ids = self._find_scheduled_ids(client, clone_name, target_date)
if current_workout is not None and scheduled_ids:
removed = self._repair_current_schedule(
@ -223,7 +230,11 @@ class SyncService:
else:
action = "create" if mapping is None else "replace_changed"
existing = existing_clone_workouts(
existing = [
workout
for workout in client.get_workouts(limit=100)
if str(workout.get("workoutName") or "") == str(payload["workoutName"])
] or existing_clone_workouts(
client.get_workouts(limit=100), target_date, self.settings.clone_prefix
)
if not mapping and existing:
@ -272,7 +283,7 @@ class SyncService:
elif action == "recreate_missing":
for scheduled_id in stale_scheduled_ids:
client.unschedule_workout(scheduled_id)
self._delete_date_marker_workouts(client, target_date)
self._delete_date_marker_workouts(client, target_date, str(payload["workoutName"]))
upload_result = client.upload_workout(payload)
workout_id = upload_result.get("workoutId") or upload_result.get("id")
@ -355,6 +366,25 @@ class SyncService:
target_date: date,
payload: dict[str, Any],
) -> dict[str, Any] | None:
mapped_id = mapping.get("clone_workout_id")
if mapped_id:
for workout in workouts:
if _workout_id(workout) == str(mapped_id):
detail = self._workout_detail(client, workout)
if detail is not None and workout_steps_equal(detail, payload):
return detail
expected_names = {
str(mapping.get("clone_workout_name") or ""),
str(payload.get("workoutName") or ""),
}
for workout in workouts:
if str(workout.get("workoutName") or "") not in expected_names:
continue
detail = self._workout_detail(client, workout)
if detail is not None and workout_steps_equal(detail, payload):
return detail
# Existing installations used a date-prefixed name. Continue recognizing those
# clones even if the user has since switched to a title template.
for workout in existing_clone_workouts(workouts, target_date, self.settings.clone_prefix):
detail = self._workout_detail(client, workout)
if detail is not None and workout_steps_equal(detail, payload):
@ -389,11 +419,16 @@ class SyncService:
if self.settings.delete_old_clones and mapping.get("clone_workout_id"):
client.delete_workout(str(mapping["clone_workout_id"]))
def _delete_date_marker_workouts(self, client: Any, target_date: date) -> list[str]:
def _delete_date_marker_workouts(
self, client: Any, target_date: date, expected_name: str
) -> list[str]:
deleted: list[str] = []
for workout in existing_clone_workouts(
client.get_workouts(limit=100), target_date, self.settings.clone_prefix
):
for workout in client.get_workouts(limit=100):
name = str(workout.get("workoutName") or "")
if name != expected_name and workout not in existing_clone_workouts(
[workout], target_date, self.settings.clone_prefix
):
continue
workout_id = _workout_id(workout)
if workout_id is None:
continue
@ -405,23 +440,24 @@ class SyncService:
self, client: Any, clone_name: str, target_date: date
) -> list[str]:
date_s = target_date.isoformat()
marker = f"{self.settings.clone_prefix} {date_s}"
calendar = client.get_scheduled_workouts(target_date.year, target_date.month)
ids: list[str] = []
for entry in generated_calendar_entries(calendar, self.settings.clone_prefix):
for entry in generated_calendar_entries(calendar):
entry_date = calendar_entry_date(entry)
if entry_date is not None and entry_date != date_s:
continue
name = calendar_entry_name(entry)
if (
name is not None
and name.startswith(marker)
and name == clone_name
and calendar_entry_id(entry) is not None
):
ids.append(str(calendar_entry_id(entry)))
return ids
def _cleanup_old_clones(self, run_id: int, client: Any) -> dict[str, int]:
if not self.settings.delete_old_clones:
return {"warnings": 0}
retention_days = self.settings.clone_retention_days
cutoff = _today(self.settings) - timedelta(days=retention_days)
warnings = 0

141
src/garmin_coach_clone/workouts.py
View File

@ -16,6 +16,8 @@ from garminconnect.workout import (
create_warmup_step,
)
from .config import DEFAULT_CLONE_NAME_TEMPLATE
CYCLING_SPORT = {"sportTypeId": 2, "sportTypeKey": "cycling", "displayOrder": 2}
CLONE_ID_FIELDS = {
"author",
@ -66,7 +68,10 @@ def build_dummy_cycling_workout(name: str | None = None) -> dict[str, Any]:
def clone_workout_payload(
source: dict[str, Any], scheduled_date: date, prefix: str
source: dict[str, Any],
scheduled_date: date,
prefix: str,
name_template: str = DEFAULT_CLONE_NAME_TEMPLATE,
) -> dict[str, Any]:
if "workoutSegments" not in source:
raise ValueError("source object does not contain workoutSegments")
@ -79,11 +84,10 @@ def clone_workout_payload(
or source.get("title")
or "Garmin Coach Workout"
)
payload["workoutName"] = f"{prefix} {scheduled_date.isoformat()} {original_name}"[:120]
payload["description"] = (
"Generated by garmin-coach-to-cal-sync probe from a Garmin Coach/adaptive "
"workout-like object. Verify targets on the Edge before relying on it."
payload["workoutName"] = clone_workout_name(
original_name, scheduled_date, prefix, name_template
)
payload["description"] = clone_workout_description(source)
payload["sportType"] = CYCLING_SPORT
for idx, segment in enumerate(payload.get("workoutSegments", []), start=1):
if isinstance(segment, dict):
@ -93,6 +97,35 @@ def clone_workout_payload(
return payload
def clone_workout_name(
original_name: str, scheduled_date: date, prefix: str, template: str
) -> str:
rendered_date = f"{scheduled_date.strftime('%B')} {scheduled_date.day}, {scheduled_date.year}"
try:
name = template.format(
type=original_name,
date=rendered_date,
clone=prefix.lower(),
)
except (KeyError, ValueError) as exc:
raise ValueError(f"invalid clone name template: {exc}") from exc
if not name.strip():
raise ValueError("clone name template rendered an empty name")
return name[:120]
def clone_workout_description(source: dict[str, Any]) -> str:
original_description = str(
source.get("description") or source.get("workoutDescription") or ""
).strip()
intervals = _interval_summary(source)
lines = [line for line in (original_description, intervals) if line]
lines.append(
"GCClone — cloned from Garmin Coach; completing it may not update the Coach plan."
)
return "\n".join(lines)
def validate_workout_payload(workout: dict[str, Any]) -> list[str]:
errors: list[str] = []
name = workout.get("workoutName")
@ -218,13 +251,19 @@ def find_generated_workout(
return None
def generated_calendar_entries(calendar_data: Any, prefix: str) -> list[dict[str, Any]]:
def generated_calendar_entries(
calendar_data: Any, prefix: str | None = None
) -> list[dict[str, Any]]:
entries: list[dict[str, Any]] = []
def walk(node: Any) -> None:
if isinstance(node, dict):
name = calendar_entry_name(node)
if name is not None and name.startswith(prefix) and calendar_entry_id(node) is not None:
if (
name is not None
and (prefix is None or name.startswith(prefix))
and calendar_entry_id(node) is not None
):
entries.append(node)
for child in node.values():
walk(child)
@ -290,6 +329,94 @@ def summarize_workout(workout: dict[str, Any]) -> str:
return "\n".join(lines)
def _interval_summary(workout: dict[str, Any]) -> str:
entries: list[str] = []
for segment in workout.get("workoutSegments", []):
if isinstance(segment, dict):
entries.extend(_interval_entries(segment.get("workoutSteps", [])))
return f"Intervals: {' | '.join(entries)}" if entries else ""
def _interval_entries(steps: Any) -> list[str]:
if not isinstance(steps, list):
return []
entries: list[str] = []
for step in steps:
if not isinstance(step, dict):
continue
step_type = _key(step.get("stepType")) or str(step.get("type") or "step")
if step_type == "repeat" or str(step.get("type")) == "RepeatGroupDTO":
nested = _interval_entries(step.get("workoutSteps", []))
if nested:
iterations = _number(step.get("numberOfIterations")) or _number(
step.get("endConditionValue")
)
count = _format_number(iterations) if iterations is not None else "?"
entries.append(f"{count}× ({' | '.join(nested)})")
continue
duration = _step_duration(step)
if duration is None:
continue
label = _step_label(step_type)
target = _step_target(step)
entries.append(f"{label} {duration}" + (f" @ {target}" if target else ""))
return entries
def _step_duration(step: dict[str, Any]) -> str | None:
if _key(step.get("endCondition")) != "time":
return None
seconds = _number(step.get("endConditionValue"))
if seconds is None or seconds < 0:
return None
total = round(seconds)
hours, remainder = divmod(total, 3600)
minutes, secs = divmod(remainder, 60)
return f"{hours}:{minutes:02}:{secs:02}" if hours else f"{minutes}:{secs:02}"
def _step_label(step_type: str) -> str:
labels = {
"warmup": "Warm up",
"cooldown": "Cool down",
"interval": "Interval",
"recovery": "Recover",
"rest": "Rest",
"main": "Main",
"other": "Other",
}
return labels.get(step_type, step_type.replace("_", " ").title())
def _step_target(step: dict[str, Any]) -> str:
target_type = _key(step.get("targetType")) or ""
one = _number(step.get("targetValueOne"))
two = _number(step.get("targetValueTwo"))
if target_type in {"", "no.target"}:
return ""
if target_type == "instruction":
return str(step.get("description") or "instruction")
unit = str(step.get("targetValueUnit") or "").strip()
if not unit:
if "power" in target_type:
unit = "W"
elif "heart.rate" in target_type:
unit = "bpm"
elif "cadence" in target_type:
unit = "rpm"
values = [value for value in (one, two) if value is not None]
if not values:
return target_type.replace(".", " ")
value_text = "–".join(_format_number(value) for value in values)
return f"{value_text} {unit}".rstrip()
def _format_number(value: float | None) -> str:
if value is None:
return "?"
return str(int(value)) if value.is_integer() else f"{value:g}"
def _summarize_steps(lines: list[str], steps: Any, indent: str) -> None:
if not isinstance(steps, list):
return

7
templates/schedule.html
View File

@ -36,6 +36,13 @@
<label>Fixed check times
<textarea name="fixed_times" rows="3" required>{{ schedule.fixed_times | join(",") }}</textarea>
</label>
<fieldset>
<legend>Clone presentation</legend>
<label>Clone name template
<input name="clone_name_template" value="{{ schedule.clone_name_template }}" required>
</label>
<p class="muted">Use <code>{type}</code>, <code>{date}</code>, and <code>{clone}</code>. The template must include <code>{clone}</code> so generated workouts remain identifiable.</p>
</fieldset>
<button type="submit">Save schedule</button>
</form>
</section>

3
tests/test_analyze_dump.py
View File

@ -52,5 +52,4 @@ def test_analyze_dump_finds_and_clones_fixture(tmp_path: Path) -> None:
assert result.returncode == 0
assert "Local clone payload passed validation." in result.stdout
assert "GCClone 2026-06-16 Coach Ride" in result.stdout
assert "Coach Ride June 16, 2026 (gcclone)" in result.stdout

22
tests/test_repository.py
View File

@ -2,7 +2,11 @@ from __future__ import annotations
import pytest
from garmin_coach_clone.config import DEFAULT_FIXED_TIMES, load_settings
from garmin_coach_clone.config import (
DEFAULT_CLONE_NAME_TEMPLATE,
DEFAULT_FIXED_TIMES,
load_settings,
)
from garmin_coach_clone.db import Database
from garmin_coach_clone.repository import Repository, ScheduleConfig, validate_schedule_config
@ -17,6 +21,7 @@ def test_schedule_restore_defaults(tmp_path, monkeypatch) -> None:
restored = repo.restore_default_schedule()
assert restored.fixed_times == DEFAULT_FIXED_TIMES
assert restored.clone_name_template == DEFAULT_CLONE_NAME_TEMPLATE
assert repo.schedule_config().active_window == "05:00-22:00"
@ -31,3 +36,18 @@ def test_schedule_validation_rejects_bad_window() -> None:
days_ahead=1,
)
)
@pytest.mark.parametrize("template", ["", "{type} {date}", "{unknown} ({clone})"])
def test_schedule_validation_rejects_invalid_clone_name_template(template: str) -> None:
with pytest.raises(ValueError, match="clone name template|unknown clone"):
validate_schedule_config(
ScheduleConfig(
enabled=True,
interval_minutes=30,
active_window="05:00-22:00",
fixed_times=["05:15"],
days_ahead=1,
clone_name_template=template,
)
)

1
tests/test_sync_service.py
View File

@ -397,6 +397,7 @@ def test_replace_changed_unschedules_matching_calendar_entries(
def test_sync_deletes_generated_clones_older_than_retention(tmp_path, monkeypatch) -> None:
monkeypatch.setattr(sync_service, "_today", lambda settings: date(2026, 6, 16))
monkeypatch.setenv("DELETE_OLD_CLONES", "true")
source = _source("Sprint")
service, _repo, garmin = _service(tmp_path, monkeypatch, source)
old_name = "GCClone 2026-06-10 Sprint"

27
tests/test_workouts.py
View File

@ -7,15 +7,17 @@ from garmin_coach_clone.workouts import (
calendar_entry_date,
calendar_entry_id,
calendar_entry_name,
clone_workout_description,
clone_workout_name,
clone_workout_payload,
estimate_duration,
existing_clone_names,
find_generated_calendar_entry,
find_generated_workout,
generated_clone_date,
generated_calendar_entries,
generated_workouts_older_than,
generated_clone_date,
generated_workouts,
generated_workouts_older_than,
validate_workout_payload,
workout_steps_equal,
)
@ -38,13 +40,32 @@ def test_clone_payload_strips_ids_and_sets_prefix() -> None:
cloned = clone_workout_payload(source, date(2026, 6, 16), "GCClone")
assert cloned["workoutName"] == "GCClone 2026-06-16 Coach Original"
assert cloned["workoutName"] == "Coach Original June 16, 2026 (gcclone)"
assert "workoutId" not in cloned
assert "ownerId" not in cloned
assert "stepId" not in cloned["workoutSegments"][0]["workoutSteps"][0]
assert validate_workout_payload(cloned) == []
def test_clone_title_and_description_use_source_details() -> None:
source = build_dummy_cycling_workout("Tempo")
source["description"] = "3x10:00@180W"
steps = source["workoutSegments"][0]["workoutSteps"]
steps[0]["targetType"] = {"workoutTargetTypeKey": "power.zone"}
steps[0]["targetValueOne"] = 98
steps[0]["targetValueTwo"] = 142
assert clone_workout_name(
"Tempo", date(2026, 7, 15), "GCClone", "{type} {date} ({clone})"
) == "Tempo July 15, 2026 (gcclone)"
assert clone_workout_description(source) == (
"3x10:00@180W\n"
"Intervals: Warm up 5:00 @ 98–142 W | 1× (Interval 2:00 | Recover 2:00) | "
"Cool down 5:00\n"
"GCClone — cloned from Garmin Coach; completing it may not update the Coach plan."
)
def test_workout_steps_equal_ignores_ids_but_compares_steps() -> None:
left = build_dummy_cycling_workout("Left")
right = build_dummy_cycling_workout("Right")