feat(skills): add zwift-workout skill (.zwo builder, validator, preview, publishing)

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
This commit is contained in:
IT Guy
2026-09-20 07:47:05 +00:00
parent 42f37a89f8
commit 109162889c
6 changed files with 993 additions and 0 deletions
+178
View File
@@ -0,0 +1,178 @@
#!/usr/bin/env python3
"""zwo.py — CLI for creating, validating, previewing and publishing Zwift .zwo workouts.
Commands
--------
build <spec.json> [-o out.zwo] [--preview] [--publish]
from-template <name> [--title T] [-o out.zwo] [--preview] [--publish]
validate <file.zwo>
parse <file.zwo> [--json]
render <file.zwo|spec.json>
templates
Exit codes: 0 = ok, 1 = validation/usage error.
"""
from __future__ import annotations
import argparse
import json
import os
import re
import subprocess
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import zwo_builder as zb # noqa: E402
HOSTING_SCRIPT = Path("/home/node/mop-skills/mop-skills/skills/hosting/scripts/publish_file.py")
def _slug(text: str) -> str:
s = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-")
return s or "workout"
def _load_spec(path: str) -> dict:
with open(path, encoding="utf-8") as fh:
return json.load(fh)
def _publish(zwo_path: Path, title: str) -> str | None:
"""Publish the .zwo via the hosting skill. Returns the public URL or None."""
if not HOSTING_SCRIPT.exists():
print(f"⚠️ hosting publisher not found at {HOSTING_SCRIPT}", file=sys.stderr)
return None
cmd = [sys.executable, str(HOSTING_SCRIPT), "--file", str(zwo_path),
"--category", "docs", "--slug", _slug(title)]
try:
out = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
except Exception as exc: # noqa: BLE001
print(f"⚠️ hosting publish failed: {exc}", file=sys.stderr)
return None
url = None
for line in out.stdout.splitlines():
if line.startswith("URL:"):
url = line.split("URL:", 1)[1].strip()
if not url:
print(out.stdout, out.stderr, file=sys.stderr)
return url
def cmd_build(args: argparse.Namespace) -> int:
spec = _load_spec(args.spec)
warnings = zb.validate_spec(spec)
out = Path(args.output or f"{_slug(spec['name'])}.zwo")
zb.write_zwo(spec, str(out))
print(f"✅ {out} ({out.stat().st_size} bytes)")
for w in warnings:
print(f"⚠️ {w}")
if args.preview:
print()
print(zb.render_text(spec))
if args.publish:
url = _publish(out, spec["name"])
if url:
print(f"🔗 {url}")
return 0
def cmd_from_template(args: argparse.Namespace) -> int:
spec = zb.from_template(args.name, args.title)
out = Path(args.output or f"{_slug(spec['name'])}.zwo")
zb.write_zwo(spec, str(out))
print(f"✅ {out} ({out.stat().st_size} bytes)")
print()
print(zb.render_text(spec))
if args.publish:
url = _publish(out, spec["name"])
if url:
print(f"🔗 {url}")
return 0
def cmd_validate(args: argparse.Namespace) -> int:
ok, messages = zb.validate_file(args.file)
if ok:
print(f"✅ {args.file} is a valid .zwo file")
for m in messages:
print(f"⚠️ {m}")
return 0
print(f"❌ {args.file} is invalid:")
for m in messages:
print(f" - {m}")
return 1
def cmd_parse(args: argparse.Namespace) -> int:
spec = zb.parse_file(args.file)
if args.json:
print(json.dumps(spec, ensure_ascii=False, indent=2))
else:
print(zb.render_text(spec))
return 0
def cmd_render(args: argparse.Namespace) -> int:
if args.file.endswith(".json"):
spec = _load_spec(args.file)
else:
spec = zb.parse_file(args.file)
print(zb.render_text(spec))
return 0
def cmd_templates(_: argparse.Namespace) -> int:
for t in zb.templates():
print(t)
return 0
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(prog="zwo.py", description="Zwift .zwo workout toolkit (bike-only)")
sub = parser.add_subparsers(dest="command", required=True)
p = sub.add_parser("build", help="build a .zwo from a JSON spec")
p.add_argument("spec")
p.add_argument("-o", "--output")
p.add_argument("--preview", action="store_true", help="print the human-readable preview")
p.add_argument("--publish", action="store_true", help="upload via the hosting skill")
p.set_defaults(func=cmd_build)
p = sub.add_parser("from-template", help="build from a built-in deterministic template")
p.add_argument("name")
p.add_argument("--title")
p.add_argument("-o", "--output")
p.add_argument("--publish", action="store_true")
p.set_defaults(func=cmd_from_template)
p = sub.add_parser("validate", help="validate an existing .zwo file")
p.add_argument("file")
p.set_defaults(func=cmd_validate)
p = sub.add_parser("parse", help="parse a .zwo into a JSON spec")
p.add_argument("file")
p.add_argument("--json", action="store_true")
p.set_defaults(func=cmd_parse)
p = sub.add_parser("render", help="render a human-readable preview")
p.add_argument("file")
p.set_defaults(func=cmd_render)
p = sub.add_parser("templates", help="list built-in templates")
p.set_defaults(func=cmd_templates)
args = parser.parse_args(argv)
try:
return args.func(args)
except zb.ZwoValidationError as exc:
print(f"❌ validation error: {exc}", file=sys.stderr)
return 1
except (OSError, json.JSONDecodeError) as exc:
print(f"❌ {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())
+484
View File
@@ -0,0 +1,484 @@
"""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)