392 lines
20 KiB
Python
392 lines
20 KiB
Python
# -*- 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}
|