# -*- coding: utf-8 -*- """Versionierter Datenvertrag zwischen Trainingsplan-Editor und Session-Tracker.""" from __future__ import annotations import copy import hashlib import re from typing import Any PLAN_SCHEMA_VERSION = 3 CONTRACT_VERSION = 2 RESULT_SCHEMA_VERSION = 2 VALID_RESULT_MODES = {"auto", "reps", "seconds", "minutes", "none"} VALID_WEIGHT_MODES = {"none", "optional", "required"} VALID_LATERALITY = {"bilateral", "unilateral"} VALID_SIDES_MODES = {"same", "separate"} VALID_TRAINING_MODES = {"auto", "sets_reps", "tabata", "fixed_interval"} def slug(value: Any) -> str: text = str(value or "").strip().lower() text = text.replace("ä", "ae").replace("ö", "oe").replace("ü", "ue").replace("ß", "ss") return re.sub(r"[^a-z0-9]+", "-", text).strip("-") or "item" def stable_id(prefix: str, *parts: Any) -> str: seed = "|".join(str(part or "") for part in parts) digest = hashlib.sha1(seed.encode("utf-8")).hexdigest()[:10] return f"{prefix}-{slug(parts[-1] if parts else prefix)[:30]}-{digest}" def normalize_result_schema(value: Any, partial: bool = False) -> dict[str, Any] | None: if value is None: return None if not isinstance(value, dict): value = {} mode = str(value.get("mode") or ("auto" if not partial else "")) weight_mode = str(value.get("weight_mode") or ("none" if not partial else "")) laterality = str(value.get("laterality") or ("bilateral" if not partial else "")) sides_mode = str(value.get("sides_mode") or ("same" if not partial else "")) result: dict[str, Any] = {} if mode in VALID_RESULT_MODES: result["mode"] = mode elif not partial: result["mode"] = "auto" if weight_mode in VALID_WEIGHT_MODES: result["weight_mode"] = weight_mode elif not partial: result["weight_mode"] = "none" if laterality in VALID_LATERALITY: result["laterality"] = laterality elif not partial: result["laterality"] = "bilateral" if sides_mode in VALID_SIDES_MODES: result["sides_mode"] = sides_mode elif not partial: result["sides_mode"] = "same" try: sets = int(value.get("sets")) except (TypeError, ValueError): sets = 0 if 1 <= sets <= 40: result["sets"] = sets if "locked_sets" in value: result["locked_sets"] = bool(value.get("locked_sets")) if result.get("laterality") == "bilateral": result["sides_mode"] = "same" return result or None def infer_result_schema(name: Any, cue: Any = "", *, stage: bool = False) -> dict[str, Any]: """Leitet für Legacy- und neue Standardpläne ein explizites Basisschema ab. Die Ableitung ist nur eine Migrationshilfe. Nach dem Speichern liegt das Ergebnis fest im Plan und kann im Editor jederzeit korrigiert werden. """ text = f"{name or ''} {cue or ''}".casefold() no_measure = re.search( r"(?:regeneration|schlaf|schmerzcheck|pausieren|protokollieren|prüfen|check|" r"mobility|mobilität|dehnen|stretch|cool.?down|warm.?up|spaziergang locker)$", text, ) if no_measure: mode = "none" elif re.search(r"(?:spaziergang|gehen|walk).*(?:min|minute)|(?:min|minute).*(?:spaziergang|gehen|walk)", text): mode = "minutes" elif re.search(r"(?:hold|plank|hang|wall\s*sit|wandsitz|isometr|halte|stützposition|deep\s+squat)", text): mode = "seconds" elif re.search(r"(?:sek(?:unde)?|\d+\s*s)", text) and not re.search(r"(?:reps?|wiederhol|negativ|push|pull|squat|row|rudern|dip)", text): mode = "seconds" else: mode = "reps" has_load = bool(re.search(r"(?:kb|kettlebell|kurzhantel|dumbbell|rucksack|gewicht|last)", text)) bodyweight_mix = bool(re.search(r"(?:optional|wahlweise|ggf\.?|bodyweight|bw|ungewichtet|körpergewicht)", text)) weight_mode = "optional" if has_load and bodyweight_mix else ("required" if has_load else "none") # Bei generischen Stufen mit „gewichtet“ bleibt Gewicht optional, wenn eine # ungewichtete Variante im selben Text genannt wird. if "gewichtet" in text and ("ungewichtet" in text or "bw" in text): weight_mode = "optional" unilateral = bool(re.search(r"(?:einarm|einbein|je\s+seite|pro\s+seite|links.?/?rechts|seitenwechsel|side\s+plank|wechsel)", text)) schema = { "mode": mode, "weight_mode": weight_mode, "laterality": "unilateral" if unilateral else "bilateral", "sides_mode": "separate" if unilateral else "same", } # Nicht messbare Checklistenpunkte benötigen nie Gewicht oder Seitenfelder. if mode == "none": schema.update({"weight_mode": "none", "laterality": "bilateral", "sides_mode": "same"}) return normalize_result_schema(schema, partial=False) or { "mode": "auto", "weight_mode": "none", "laterality": "bilateral", "sides_mode": "same" } def infer_cluster(name: str, key: str = "") -> tuple[str, str]: text = f"{key} {name}".lower() rules = [ (r"liegest|push.?up|\bls\b", "horizontal_push", "Push-up"), (r"dip", "vertical_push", "Dip"), (r"pike|overhead|floor press", "vertical_push", "Schulterdrücken"), (r"pull.?up|klimm|hang", "vertical_pull", "Pull-up / Hang"), (r"rudern|row", "horizontal_pull", "Rudern"), (r"squat|kniebeuge|goblet|lunge|ausfallschritt|wall sit", "knee_dominant", "Kniebeuge / Ausfallschritt"), (r"deadlift|rdl|hinge|swing|glute bridge", "hip_hinge", "Hip Hinge / Hüftstreckung"), (r"plank|hollow|dead bug|side plank|core", "core", "Core"), (r"curl", "elbow_flexion", "Bizeps"), (r"waden|calf", "calf", "Waden"), (r"halo|around|carry|bottoms", "loaded_mobility", "Kettlebell-Kontrolle"), ] for pattern, cid, label in rules: if re.search(pattern, text): return cid, label return "general", "Allgemein" def infer_factor(name: str, cluster_id: str) -> float: text = str(name or "").lower() if cluster_id == "horizontal_push": if "wand" in text: return 0.05 if "hoch" in text and ("incline" in text or "inkline" in text): return 0.10 if "mittel" in text and ("incline" in text or "inkline" in text): return 0.20 if "tief" in text and ("incline" in text or "inkline" in text): return 0.35 if "negativ" in text and ("voll" in text or "erste" in text): return 0.85 if "negativ" in text: return 0.70 if "assist" in text: return 0.55 if "knie" in text: return 0.45 if "teil" in text or "partial" in text: return 0.35 if "voll" in text or "standard" in text or "normal" in text: return 1.00 if cluster_id == "vertical_pull": if "active hang" in text or "hang" in text: return 0.20 if "scapular" in text: return 0.25 if "negativ" in text: return 0.70 if "leicht assist" in text: return 0.80 if "assist" in text: return 0.55 if "voll" in text or "klimm" in text or "pull-up" in text: return 1.00 if cluster_id == "horizontal_pull": if "aufrecht" in text or "stark assist" in text: return 0.25 if "bent" in text or "gebeugt" in text: return 0.50 if "straight" in text or "gestreckt" in text: return 1.00 if "füße erhöht" in text: return 1.15 if cluster_id == "knee_dominant": if "hold" in text: return 1.00 if "chair" in text or "box" in text or "stuhl" in text: return 0.35 if "negativ" in text: return 0.75 if "assist" in text or "festhalten" in text: return 0.55 if "voll" in text or "goblet" in text or "körpergewicht" in text: return 1.00 if cluster_id == "vertical_push": if "hold" in text: return 0.30 if "assist" in text: return 0.60 if "negativ" in text: return 0.80 if "voll" in text: return 1.00 if cluster_id == "core": if "knie" in text or "tuck" in text: return 0.65 if "halb" in text: return 0.82 if "voll" in text: return 1.00 return 1.00 def migrate_training_config(config: Any, plan_id: str | None = None) -> dict[str, Any]: cfg = copy.deepcopy(config or {}) if cfg.get("type") == "recipe": return cfg cfg["schema_version"] = PLAN_SCHEMA_VERSION cfg["contract_version"] = CONTRACT_VERSION meta = cfg.setdefault("meta", {}) resolved_plan_id = str(cfg.get("plan_id") or plan_id or stable_id("plan", meta.get("title") or "training")) cfg["plan_id"] = resolved_plan_id phases = cfg.setdefault("phases", {}).setdefault("items", []) for index, phase in enumerate(phases): if not isinstance(phase, dict): phases[index] = phase = {"name": str(phase or "")} phase.setdefault("id", stable_id("phase", resolved_plan_id, index, phase.get("name"))) phase_ids = [phase.get("id") for phase in phases] fmt = cfg.get("training_format") if isinstance(cfg.get("training_format"), dict) else {} mode = str(fmt.get("mode") or "auto") if mode not in VALID_TRAINING_MODES: mode = "auto" fmt["mode"] = mode fmt["fixed_interval"] = bool(fmt.get("fixed_interval") or mode in {"tabata", "fixed_interval"}) fmt["rounds_scope"] = str(fmt.get("rounds_scope") or "block") fmt["schema_version"] = 1 cfg["training_format"] = fmt days = cfg.get("days") if isinstance(cfg.get("days"), list) else [] for di, day in enumerate(days): if not isinstance(day, dict): continue day.setdefault("id", stable_id("day", resolved_plan_id, day.get("num", di + 1), day.get("focus"))) rotations = day.get("rotations") if isinstance(day.get("rotations"), list) else [] for ri, rotation in enumerate(rotations): if not isinstance(rotation, dict): continue rotation.setdefault("id", stable_id("rotation", day["id"], ri, rotation.get("label"))) exercises = rotation.get("exercises") if isinstance(rotation.get("exercises"), list) else [] day_num = day.get("num", di + 1) for ei, exercise in enumerate(exercises): if not isinstance(exercise, dict): continue # Persist the original positional key once so legacy session data can # still be migrated safely even after the published plan is reordered. exercise.setdefault("legacy_id", f"d{day_num}-r{ri}-e{ei}") explicit_progression = str(exercise.get("progression_id") or exercise.get("key") or "").strip() progression_id = explicit_progression or f"name:{slug(exercise.get('name'))}" exercise["progression_id"] = progression_id exercise["key"] = str(exercise.get("key") or progression_id) exercise.setdefault("id", stable_id("exercise", rotation["id"], progression_id, exercise.get("name"), ei)) exercise.setdefault("exercise_id", stable_id("movement", progression_id, exercise.get("name"))) cid, label = infer_cluster(str(exercise.get("name") or ""), progression_id) exercise.setdefault("movement_cluster", cid) exercise.setdefault("movement_label", label) if exercise.get("result_schema") is not None: exercise["result_schema"] = normalize_result_schema(exercise.get("result_schema"), partial=False) else: exercise["result_schema"] = infer_result_schema(exercise.get("name"), exercise.get("cue")) cfg["days"] = days old_stages = cfg.get("stages") if isinstance(cfg.get("stages"), dict) else {} new_stages: dict[str, Any] = {} for key, raw_stage in old_stages.items(): stage = copy.deepcopy(raw_stage) if isinstance(raw_stage, dict) else {"steps": raw_stage if isinstance(raw_stage, list) else []} stage_id = str(stage.get("id") or stable_id("progression", resolved_plan_id, key, stage.get("name"))) stage["id"] = stage_id stage["key"] = str(stage.get("key") or key) raw_steps = stage.get("steps") if isinstance(stage.get("steps"), list) else [] schemas = stage.get("result_schemas") if isinstance(stage.get("result_schemas"), list) else [] steps: list[dict[str, Any]] = [] for index, raw_step in enumerate(raw_steps): if isinstance(raw_step, dict): step = copy.deepcopy(raw_step) step_name = str(step.get("name") or step.get("label") or "") else: step_name = str(raw_step or "") step = {"name": step_name} step.setdefault("id", stable_id("step", stage_id, index, step_name)) step.setdefault("phase_id", phase_ids[index] if index < len(phase_ids) else "") schema = step.get("result_schema") if schema is None and index < len(schemas): schema = schemas[index] if schema is not None: step["result_schema"] = normalize_result_schema(schema, partial=False) else: step["result_schema"] = infer_result_schema(step_name, stage.get("name"), stage=True) cid, _ = infer_cluster(str(stage.get("name") or step_name), key) step.setdefault("movement_cluster", cid) step.setdefault("factor", infer_factor(step_name, cid)) steps.append(step) stage["steps"] = steps stage.pop("result_schemas", None) new_stages[str(key)] = stage cfg["stages"] = new_stages # Zentrale, planbezogene Übungsbibliothek. Der Tracker nutzt diese vor Regex-Fallbacks. catalog: dict[str, Any] = {} for day in days: for rotation in day.get("rotations", []): for exercise in rotation.get("exercises", []): eid = str(exercise.get("exercise_id")) pid = str(exercise.get("progression_id") or "") entry = catalog.setdefault(eid, { "id": eid, "name": str(exercise.get("name") or ""), "progression_id": pid, "movement_cluster": exercise.get("movement_cluster") or "general", "movement_label": exercise.get("movement_label") or "Allgemein", "result_schema": copy.deepcopy(exercise.get("result_schema")), "variants": [], }) stage = new_stages.get(pid) if stage: entry["variants"] = [ { "id": step.get("id"), "name": step.get("name", ""), "factor": step.get("factor", 1.0), "movement_cluster": step.get("movement_cluster") or entry["movement_cluster"], "result_schema": copy.deepcopy(step.get("result_schema")), } for step in stage.get("steps", []) ] cfg["exercise_catalog"] = catalog return cfg def validate_training_config(config: Any) -> dict[str, list[dict[str, Any]]]: cfg = migrate_training_config(config) errors: list[dict[str, Any]] = [] warnings: list[dict[str, Any]] = [] ids: dict[str, str] = {} def register(value: Any, location: str) -> None: sid = str(value or "") if not sid: errors.append({"code": "missing_id", "location": location, "message": "Stabile ID fehlt."}) elif sid in ids: errors.append({"code": "duplicate_id", "location": location, "message": f"ID wird bereits bei {ids[sid]} verwendet."}) else: ids[sid] = location def validate_result_schema(schema: Any, location: str, *, required: bool) -> None: if not isinstance(schema, dict): target = errors if required else warnings target.append({"code": "missing_result_schema", "location": location, "message": "Ergebniserfassung ist nicht eindeutig festgelegt."}) return mode = str(schema.get("mode") or "auto") weight = str(schema.get("weight_mode") or "none") laterality = str(schema.get("laterality") or "bilateral") sides = str(schema.get("sides_mode") or "same") if mode == "none" and weight != "none": errors.append({"code": "invalid_result_schema", "location": location, "message": "Ohne Messwert darf kein Gewicht verlangt werden."}) if laterality == "bilateral" and sides == "separate": errors.append({"code": "invalid_sides", "location": location, "message": "Getrennte Seitenwerte sind nur bei einseitigen Übungen möglich."}) if laterality == "unilateral" and sides not in {"same", "separate"}: errors.append({"code": "invalid_sides", "location": location, "message": "Für einseitige Übungen muss die Seitenlogik festgelegt sein."}) try: sets = int(schema.get("sets") or 0) except (TypeError, ValueError): sets = 0 if "sets" in schema and not 1 <= sets <= 40: errors.append({"code": "invalid_sets", "location": location, "message": "Die feste Anzahl muss zwischen 1 und 40 liegen."}) for pi, phase in enumerate(cfg.get("phases", {}).get("items", [])): register(phase.get("id"), f"Phase {pi + 1}") if not str(phase.get("name") or "").strip(): errors.append({"code": "empty_phase", "location": f"Phase {pi + 1}", "message": "Phasenname fehlt."}) for di, day in enumerate(cfg.get("days", [])): register(day.get("id"), f"Tag {di + 1}") for ri, rotation in enumerate(day.get("rotations", [])): register(rotation.get("id"), f"Tag {di + 1}, Rotation {ri + 1}") exercises = rotation.get("exercises", []) if not exercises: warnings.append({"code": "empty_rotation", "location": f"Tag {di + 1}, Rotation {ri + 1}", "message": "Rotation enthält keine Übungen."}) for ei, exercise in enumerate(exercises): location = f"Tag {di + 1}, Rotation {ri + 1}, Übung {ei + 1}" register(exercise.get("id"), location) if not str(exercise.get("name") or "").strip(): errors.append({"code": "empty_exercise", "location": location, "message": "Übungsname fehlt."}) pid = str(exercise.get("progression_id") or "") if pid and not pid.startswith("name:") and pid not in cfg.get("stages", {}): warnings.append({"code": "unknown_progression", "location": location, "message": f"Progression „{pid}“ ist nicht definiert."}) validate_result_schema(exercise.get("result_schema"), f"{location}, Ergebniserfassung", required=True) for key, stage in cfg.get("stages", {}).items(): register(stage.get("id"), f"Progression {key}") for si, step in enumerate(stage.get("steps", [])): register(step.get("id"), f"Progression {key}, Stufe {si + 1}") step_location = f"Progression {key}, Stufe {si + 1}" if not str(step.get("name") or "").strip(): warnings.append({"code": "empty_step", "location": step_location, "message": "Stufenname ist leer."}) validate_result_schema(step.get("result_schema"), f"{step_location}, Ergebniserfassung", required=False) try: factor = float(step.get("factor") or 0) except (TypeError, ValueError): factor = 0 if factor <= 0: errors.append({"code": "invalid_factor", "location": step_location, "message": "Der Analysefaktor muss größer als 0 sein."}) fmt = cfg.get("training_format", {}) if fmt.get("mode") in {"tabata", "fixed_interval"}: if not fmt.get("work_seconds"): errors.append({"code": "missing_work_seconds", "location": "Trainingsformat", "message": "Arbeitszeit fehlt."}) if not fmt.get("rounds"): errors.append({"code": "missing_rounds", "location": "Trainingsformat", "message": "Intervallzahl fehlt."}) rounds = int(fmt.get("rounds") or 0) for di, day in enumerate(cfg.get("days", [])): for ri, rotation in enumerate(day.get("rotations", [])): count = len(rotation.get("exercises", [])) if rounds and count and rounds % count: warnings.append({"code": "uneven_intervals", "location": f"Tag {di + 1}, Rotation {ri + 1}", "message": f"{rounds} Intervalle sind nicht durch {count} Übungen teilbar. Feste Satzanzahl im Ergebnisschema setzen."}) return {"errors": errors, "warnings": warnings, "config": cfg}