IT Guy
109162889c
Bike-only Zwift .zwo generator with human-readable preview + confirmation loop. - scripts/zwo_builder.py: spec<->.zwo build/validate/parse/render/TSS estimate - scripts/zwo.py: CLI (build, from-template, validate, parse, render, templates) - references/zwo-format.md: condensed offline format reference - assets/example-vo2max.zwo + tests/test_zwo.py (9 tests) - Publishes via hosting skill; files result into Knowledge Wiki Format ground truth: h4l/zwift-workout-file-reference
484 lines
18 KiB
Python
484 lines
18 KiB
Python
"""zwo_builder.py — Build, validate, parse and render Zwift ``.zwo`` workout files.
|
||
|
||
Bike-only implementation following the community format reference
|
||
(https://github.com/h4l/zwift-workout-file-reference). No official Zwift XSD
|
||
exists; this module encodes the subset that Zwift reliably accepts.
|
||
|
||
Public API
|
||
----------
|
||
build_spec(spec) -> xml.etree.ElementTree.Element (raises on error)
|
||
spec_to_xml(spec) -> str
|
||
validate_spec(spec) -> list[str] (non-fatal warnings)
|
||
validate_file(path) -> (ok: bool, messages: list[str])
|
||
parse_file(path) -> dict (spec)
|
||
render_text(spec) -> str (human-readable preview)
|
||
estimate(spec) -> dict (duration / NP / IF / TSS)
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import xml.etree.ElementTree as ET
|
||
from typing import Any
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Constants / guardrails
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
VALID_SPORT_TYPES = {"bike"}
|
||
VALID_DURATION_TYPES = {"time"}
|
||
|
||
POWER_MIN = 0.20 # 20 % FTP — below this is almost certainly a mistake
|
||
POWER_MAX = 1.50 # 150 % FTP — above this is very unusual for a *.zwo target
|
||
DURATION_MIN = 5 # seconds; shorter steps are usually noise
|
||
DURATION_MAX = 4 * 3600 # 4 h single step sanity cap
|
||
|
||
# spec key -> (XML attribute name, coercion)
|
||
_STEP_SPECS: dict[str, dict[str, tuple[str, type]]] = {
|
||
"Warmup": {
|
||
"duration": ("Duration", int),
|
||
"powerLow": ("PowerLow", float),
|
||
"powerHigh": ("PowerHigh", float),
|
||
"cadence": ("Cadence", int),
|
||
},
|
||
"Cooldown": {
|
||
"duration": ("Duration", int),
|
||
"powerLow": ("PowerLow", float),
|
||
"powerHigh": ("PowerHigh", float),
|
||
"cadence": ("Cadence", int),
|
||
},
|
||
"SteadyState": {
|
||
"duration": ("Duration", int),
|
||
"power": ("Power", float),
|
||
"cadence": ("Cadence", int),
|
||
},
|
||
"Ramp": {
|
||
"duration": ("Duration", int),
|
||
"powerLow": ("PowerLow", float),
|
||
"powerHigh": ("PowerHigh", float),
|
||
},
|
||
"IntervalsT": {
|
||
"repeat": ("Repeat", int),
|
||
"onDuration": ("OnDuration", int),
|
||
"offDuration": ("OffDuration", int),
|
||
"onPower": ("OnPower", float),
|
||
"offPower": ("OffPower", float),
|
||
"cadence": ("Cadence", int),
|
||
"cadenceResting": ("CadenceResting", int),
|
||
},
|
||
"FreeRide": {
|
||
"duration": ("Duration", int),
|
||
"flatRoad": ("FlatRoad", int),
|
||
},
|
||
"MaxEffort": {
|
||
"duration": ("Duration", int),
|
||
},
|
||
"SolidState": {
|
||
"duration": ("Duration", int),
|
||
"power": ("Power", float),
|
||
},
|
||
"RestDay": {},
|
||
"TextEvent": {
|
||
"timeoffset": ("timeoffset", int),
|
||
"message": ("message", str),
|
||
"duration": ("duration", int),
|
||
},
|
||
}
|
||
|
||
# Elements that carry a target power (used for the TSS estimate)
|
||
_POWER_STEPS = {"Warmup", "Cooldown", "SteadyState", "Ramp", "IntervalsT", "SolidState"}
|
||
|
||
REQUIRED_STEP_ATTRS: dict[str, tuple[str, ...]] = {
|
||
"Warmup": ("duration", "powerLow", "powerHigh"),
|
||
"Cooldown": ("duration", "powerLow", "powerHigh"),
|
||
"SteadyState": ("duration", "power"),
|
||
"Ramp": ("duration", "powerLow", "powerHigh"),
|
||
"IntervalsT": ("repeat", "onDuration", "offDuration", "onPower", "offPower"),
|
||
"FreeRide": ("duration",),
|
||
"MaxEffort": ("duration",),
|
||
"SolidState": ("duration", "power"),
|
||
"RestDay": (),
|
||
"TextEvent": ("timeoffset", "message"),
|
||
}
|
||
|
||
|
||
class ZwoValidationError(ValueError):
|
||
"""Raised when a spec cannot produce a valid .zwo file."""
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Validation
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _check_power(value: float, where: str, errors: list[str], warnings: list[str]) -> None:
|
||
if value < POWER_MIN:
|
||
errors.append(f"{where}: power {value:.2f} (< {POWER_MIN:.2f} FTP) is implausible")
|
||
elif value > POWER_MAX:
|
||
errors.append(f"{where}: power {value:.2f} (> {POWER_MAX:.2f} FTP) is implausible")
|
||
|
||
|
||
def _check_duration(value: int, where: str, errors: list[str], warnings: list[str]) -> None:
|
||
if value <= 0:
|
||
errors.append(f"{where}: duration must be > 0 seconds")
|
||
elif value < DURATION_MIN:
|
||
warnings.append(f"{where}: duration {value}s is very short")
|
||
elif value > DURATION_MAX:
|
||
warnings.append(f"{where}: duration {value}s exceeds the {DURATION_MAX}s single-step cap")
|
||
|
||
|
||
def validate_spec(spec: dict[str, Any]) -> list[str]:
|
||
"""Validate a workout spec. Raise on fatal errors, return warnings."""
|
||
errors: list[str] = []
|
||
warnings: list[str] = []
|
||
|
||
if not isinstance(spec, dict):
|
||
raise ZwoValidationError("spec must be a JSON object")
|
||
|
||
name = spec.get("name")
|
||
if not isinstance(name, str) or not name.strip():
|
||
errors.append("missing required field: name")
|
||
|
||
sport = spec.get("sportType", "bike")
|
||
if sport not in VALID_SPORT_TYPES:
|
||
errors.append(f"sportType {sport!r} not supported (bike-only): {sorted(VALID_SPORT_TYPES)}")
|
||
|
||
steps = spec.get("steps")
|
||
if not isinstance(steps, list) or not steps:
|
||
errors.append("spec must contain a non-empty 'steps' list")
|
||
steps = []
|
||
|
||
for i, step in enumerate(steps, start=1):
|
||
where = f"step {i}"
|
||
if not isinstance(step, dict):
|
||
errors.append(f"{where}: must be an object")
|
||
continue
|
||
stype = step.get("type")
|
||
if stype not in _STEP_SPECS:
|
||
errors.append(f"{where}: unknown step type {stype!r}")
|
||
continue
|
||
for key in REQUIRED_STEP_ATTRS.get(stype, ()):
|
||
if step.get(key) is None:
|
||
errors.append(f"{where} ({stype}): missing required field {key!r}")
|
||
# power sanity
|
||
for key in ("power", "powerLow", "powerHigh", "onPower", "offPower"):
|
||
if key in step and step[key] is not None:
|
||
try:
|
||
_check_power(float(step[key]), f"{where} ({stype}).{key}", errors, warnings)
|
||
except (TypeError, ValueError):
|
||
errors.append(f"{where} ({stype}).{key}: not a number")
|
||
# ramp direction is only a hint
|
||
if stype in ("Warmup",) and step.get("powerHigh") is not None and step.get("powerLow") is not None:
|
||
if float(step["powerHigh"]) < float(step["powerLow"]):
|
||
warnings.append(f"{where} (Warmup): powerHigh < powerLow")
|
||
for key in ("duration", "onDuration", "offDuration"):
|
||
if key in step and step[key] is not None:
|
||
try:
|
||
_check_duration(int(step[key]), f"{where} ({stype}).{key}", errors, warnings)
|
||
except (TypeError, ValueError):
|
||
errors.append(f"{where} ({stype}).{key}: not an integer")
|
||
if stype == "IntervalsT" and step.get("repeat") is not None:
|
||
try:
|
||
if int(step["repeat"]) < 1:
|
||
errors.append(f"{where} (IntervalsT): repeat must be >= 1")
|
||
except (TypeError, ValueError):
|
||
errors.append(f"{where} (IntervalsT).repeat: not an integer")
|
||
|
||
if errors:
|
||
raise ZwoValidationError("; ".join(errors))
|
||
return warnings
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Build
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _coerce(value: Any, typ: type) -> str:
|
||
if typ is int:
|
||
return str(int(value))
|
||
if typ is float:
|
||
# Zwift accepts plain decimals; trim trailing zeros
|
||
return f"{float(value):g}"
|
||
return str(value)
|
||
|
||
|
||
def build_spec(spec: dict[str, Any]) -> ET.Element:
|
||
"""Build and re-validate the XML element tree for ``spec``."""
|
||
validate_spec(spec)
|
||
|
||
root = ET.Element("workout_file")
|
||
ET.SubElement(root, "author").text = str(spec.get("author", "IT Guy"))
|
||
ET.SubElement(root, "name").text = str(spec["name"])
|
||
if spec.get("description"):
|
||
ET.SubElement(root, "description").text = str(spec["description"])
|
||
ET.SubElement(root, "sportType").text = spec.get("sportType", "bike")
|
||
|
||
tags = spec.get("tags") or []
|
||
if tags:
|
||
tag_el = ET.SubElement(root, "tags")
|
||
for t in tags:
|
||
ET.SubElement(tag_el, "tag", {"name": str(t)})
|
||
|
||
workout = ET.SubElement(root, "workout")
|
||
for step in spec["steps"]:
|
||
stype = step["type"]
|
||
attrs: dict[str, str] = {}
|
||
for key, (attr, typ) in _STEP_SPECS[stype].items():
|
||
if step.get(key) is not None:
|
||
attrs[attr] = _coerce(step[key], typ)
|
||
ET.SubElement(workout, stype, attrs)
|
||
|
||
return root
|
||
|
||
|
||
def spec_to_xml(spec: dict[str, Any]) -> str:
|
||
"""Serialize a spec to a pretty-printed .zwo XML string (with declaration)."""
|
||
root = build_spec(spec)
|
||
ET.indent(root, space=" ")
|
||
body = ET.tostring(root, encoding="unicode")
|
||
return '<?xml version="1.0" encoding="UTF-8"?>\n' + body + "\n"
|
||
|
||
|
||
def write_zwo(spec: dict[str, Any], path: str) -> str:
|
||
"""Write ``spec`` to ``path`` after a round-trip parse check. Returns path."""
|
||
xml = spec_to_xml(spec)
|
||
# round-trip: refuse to emit XML that cannot be re-parsed
|
||
try:
|
||
ET.fromstring(xml)
|
||
except ET.ParseError as exc: # pragma: no cover - defensive
|
||
raise ZwoValidationError(f"generated XML is not well-formed: {exc}") from exc
|
||
with open(path, "w", encoding="utf-8") as fh:
|
||
fh.write(xml)
|
||
return path
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Parse / validate files
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _step_from_element(el: ET.Element) -> dict[str, Any]:
|
||
stype = el.tag
|
||
spec = {"type": stype}
|
||
for key, (attr, typ) in _STEP_SPECS.get(stype, {}).items():
|
||
if attr in el.attrib:
|
||
try:
|
||
spec[key] = typ(el.attrib[attr])
|
||
except ValueError:
|
||
spec[key] = el.attrib[attr]
|
||
return spec
|
||
|
||
|
||
def parse_file(path: str) -> dict[str, Any]:
|
||
"""Parse an existing .zwo file into our spec representation."""
|
||
tree = ET.parse(path)
|
||
root = tree.getroot()
|
||
if root.tag != "workout_file":
|
||
raise ZwoValidationError(f"root element is <{root.tag}>, expected <workout_file>")
|
||
|
||
def text(tag: str) -> str | None:
|
||
el = root.find(tag)
|
||
return el.text.strip() if el is not None and el.text else None
|
||
|
||
tags: list[str] = []
|
||
tags_el = root.find("tags")
|
||
if tags_el is not None:
|
||
tags = [t.get("name", "") for t in tags_el.findall("tag")]
|
||
|
||
workout = root.find("workout")
|
||
steps = [_step_from_element(el) for el in (workout or [])]
|
||
|
||
return {
|
||
"name": text("name") or "Unnamed",
|
||
"author": text("author") or "",
|
||
"description": text("description") or "",
|
||
"sportType": text("sportType") or "bike",
|
||
"tags": tags,
|
||
"steps": steps,
|
||
}
|
||
|
||
|
||
def validate_file(path: str) -> tuple[bool, list[str]]:
|
||
"""Validate an existing .zwo file. Returns (ok, messages)."""
|
||
try:
|
||
spec = parse_file(path)
|
||
except (ET.ParseError, ZwoValidationError, OSError) as exc:
|
||
return False, [f"parse error: {exc}"]
|
||
try:
|
||
warnings = validate_spec(spec)
|
||
except ZwoValidationError as exc:
|
||
return False, str(exc).split("; ")
|
||
return True, warnings
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Render (human-readable preview)
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
|
||
def _fmt_time(seconds: int) -> str:
|
||
m, s = divmod(int(seconds), 60)
|
||
return f"{m}:{s:02d}"
|
||
|
||
|
||
def _pct(value: float) -> str:
|
||
return f"{value * 100:g}%"
|
||
|
||
|
||
def estimate(spec: dict[str, Any]) -> dict[str, Any]:
|
||
"""Estimate total duration, normalised power fraction and TSS."""
|
||
total = 0
|
||
weighted = 0.0 # sum(sec * f^2)
|
||
untracked = 0
|
||
for step in spec.get("steps", []):
|
||
stype = step.get("type")
|
||
if stype == "IntervalsT":
|
||
reps = int(step.get("repeat", 1))
|
||
on_d, off_d = int(step["onDuration"]), int(step["offDuration"])
|
||
total += reps * (on_d + off_d)
|
||
weighted += reps * (on_d * float(step["onPower"]) ** 2 + off_d * float(step["offPower"]) ** 2)
|
||
elif stype in _POWER_STEPS:
|
||
dur = int(step["duration"])
|
||
if stype in ("Warmup", "Cooldown", "Ramp"):
|
||
frac = (float(step["powerLow"]) + float(step["powerHigh"])) / 2
|
||
else:
|
||
frac = float(step["power"])
|
||
total += dur
|
||
weighted += dur * frac ** 2
|
||
elif stype in ("FreeRide", "MaxEffort"):
|
||
total += int(step["duration"])
|
||
untracked += int(step["duration"])
|
||
tss = (100 * weighted / 3600) if total else 0.0
|
||
np_frac = (weighted / total) ** 0.5 if total else 0.0
|
||
return {
|
||
"duration_s": total,
|
||
"duration": _fmt_time(total),
|
||
"np_frac": round(np_frac, 3),
|
||
"if": round(np_frac, 3),
|
||
"tss": round(tss, 1),
|
||
"untracked_s": untracked,
|
||
}
|
||
|
||
|
||
def _bar(frac: float, width: int = 40, scale: float = 1.5) -> str:
|
||
filled = max(1, min(width, round(frac / scale * width)))
|
||
return "█" * filled
|
||
|
||
|
||
def render_text(spec: dict[str, Any]) -> str:
|
||
"""Human-readable workout preview for confirmation before publishing."""
|
||
est = estimate(spec)
|
||
lines: list[str] = []
|
||
lines.append(f"🏋️ {spec['name']}")
|
||
if spec.get("description"):
|
||
lines.append(f" {spec['description']}")
|
||
lines.append(f" Sport: {spec.get('sportType', 'bike')} Autor: {spec.get('author', '')}")
|
||
lines.append("")
|
||
lines.append(f" # {'Step':<13}{'Dauer':>7} Details")
|
||
lines.append(" " + "-" * 62)
|
||
|
||
for i, step in enumerate(spec["steps"], start=1):
|
||
stype = step["type"]
|
||
if stype == "IntervalsT":
|
||
reps = int(step["repeat"])
|
||
total = reps * (int(step["onDuration"]) + int(step["offDuration"]))
|
||
detail = (f"{reps}× {_fmt_time(step['onDuration'])} @{_pct(step['onPower'])}"
|
||
f" / {_fmt_time(step['offDuration'])} @{_pct(step['offPower'])}")
|
||
dur = _fmt_time(total)
|
||
elif stype in ("Warmup", "Cooldown", "Ramp"):
|
||
detail = f"{_pct(step['powerLow'])} → {_pct(step['powerHigh'])}"
|
||
dur = _fmt_time(int(step["duration"]))
|
||
elif stype in ("SteadyState", "SolidState"):
|
||
detail = f"konstant @{_pct(step['power'])}"
|
||
dur = _fmt_time(int(step["duration"]))
|
||
elif stype == "FreeRide":
|
||
detail = "freie Fahrt (ERG aus)"
|
||
dur = _fmt_time(int(step["duration"]))
|
||
elif stype == "MaxEffort":
|
||
detail = "All-out (ERG aus)"
|
||
dur = _fmt_time(int(step["duration"]))
|
||
elif stype == "RestDay":
|
||
detail, dur = "Ruhetag", "—"
|
||
elif stype == "TextEvent":
|
||
detail, dur = f"Text: {step.get('message', '')}", "—"
|
||
else:
|
||
detail, dur = "", "—"
|
||
lines.append(f" {i:<3}{stype:<13}{dur:>7} {detail}")
|
||
|
||
lines.append("")
|
||
lines.append(" Power-Profil:")
|
||
for step in spec["steps"]:
|
||
stype = step["type"]
|
||
if stype == "IntervalsT":
|
||
lines.append(f" ON {_bar(float(step['onPower']))} {_pct(step['onPower'])}")
|
||
lines.append(f" OFF {_bar(float(step['offPower']))} {_pct(step['offPower'])}")
|
||
elif stype in ("Warmup", "Cooldown", "Ramp"):
|
||
lines.append(f" {_bar((float(step['powerLow']) + float(step['powerHigh'])) / 2):<40} "
|
||
f"{_pct(step['powerLow'])}→{_pct(step['powerHigh'])}")
|
||
elif stype in ("SteadyState", "SolidState"):
|
||
lines.append(f" {_bar(float(step['power'])):<40} {_pct(step['power'])}")
|
||
|
||
lines.append("")
|
||
lines.append(f" Gesamt: {est['duration']} NP ≈ {est['np_frac'] * 100:g}% FTP "
|
||
f"IF {est['if']:.2f} TSS ≈ {est['tss']}")
|
||
if est["untracked_s"]:
|
||
lines.append(f" ⚠️ {_fmt_time(est['untracked_s'])} FreeRide/MaxEffort sind nicht in der TSS-Schätzung enthalten.")
|
||
return "\n".join(lines)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# Deterministic templates ("from-text")
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
_TEMPLATES: dict[str, dict[str, Any]] = {
|
||
"endurance": {
|
||
"description": "Ruhige Grundlagenfahrt (Z2).",
|
||
"steps": [
|
||
{"type": "Warmup", "duration": 600, "powerLow": 0.50, "powerHigh": 0.65},
|
||
{"type": "SteadyState", "duration": 3000, "power": 0.68},
|
||
{"type": "Cooldown", "duration": 300, "powerLow": 0.65, "powerHigh": 0.45},
|
||
],
|
||
},
|
||
"sweetspot": {
|
||
"description": "3 × 12 min Sweet Spot (88 % FTP).",
|
||
"steps": [
|
||
{"type": "Warmup", "duration": 600, "powerLow": 0.50, "powerHigh": 0.75},
|
||
{"type": "IntervalsT", "repeat": 3, "onDuration": 720, "offDuration": 240,
|
||
"onPower": 0.88, "offPower": 0.55},
|
||
{"type": "Cooldown", "duration": 300, "powerLow": 0.65, "powerHigh": 0.45},
|
||
],
|
||
},
|
||
"vo2max": {
|
||
"description": "5 × 3 min VO2max (115 % FTP).",
|
||
"steps": [
|
||
{"type": "Warmup", "duration": 600, "powerLow": 0.50, "powerHigh": 0.75},
|
||
{"type": "IntervalsT", "repeat": 5, "onDuration": 180, "offDuration": 180,
|
||
"onPower": 1.15, "offPower": 0.55},
|
||
{"type": "Cooldown", "duration": 300, "powerLow": 0.65, "powerHigh": 0.45},
|
||
],
|
||
},
|
||
"sprint": {
|
||
"description": "SIT: 6 × 30 s all-out, 4 min Pause (wiki-verankertes Protokoll).",
|
||
"steps": [
|
||
{"type": "Warmup", "duration": 900, "powerLow": 0.50, "powerHigh": 0.80},
|
||
{"type": "IntervalsT", "repeat": 6, "onDuration": 30, "offDuration": 240,
|
||
"onPower": 1.50, "offPower": 0.40},
|
||
{"type": "Cooldown", "duration": 420, "powerLow": 0.60, "powerHigh": 0.40},
|
||
],
|
||
},
|
||
}
|
||
|
||
|
||
def from_template(name: str, title: str | None = None) -> dict[str, Any]:
|
||
"""Return a deterministic workout spec from the built-in template set."""
|
||
key = name.strip().lower()
|
||
if key not in _TEMPLATES:
|
||
raise ZwoValidationError(f"unknown template {name!r}; available: {sorted(_TEMPLATES)}")
|
||
spec = {k: (v.copy() if isinstance(v, list) else v) for k, v in _TEMPLATES[key].items()}
|
||
spec.update({"name": title or key.title(), "author": "IT Guy", "sportType": "bike",
|
||
"tags": ["zwift", key]})
|
||
return spec
|
||
|
||
|
||
def templates() -> list[str]:
|
||
return sorted(_TEMPLATES) |