Source code
src/scripts/presentation/pipeline.py
This file is included in this documentation build. It is displayed as code and is not executed.
"""Build narrated showcase videos from versioned slides and editable narration.
@author Laurent Declercq l.declercq@agon-innovation.ch
@version 20260921
"""
import asyncio
import hashlib
import html
import json
import math
import os
from pathlib import Path
import re
import struct
import subprocess
import tempfile
ROOT = Path(__file__).resolve().parents[3]
SOURCE = ROOT / "docs/presentation"
CACHE_VERSION = 1
def digest(value):
"""Return a stable digest of JSON-compatible build inputs."""
encoded = json.dumps(
value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
)
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
def file_hash(path):
"""Hash file contents so edits invalidate cached artifacts."""
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def read_json(path):
"""Read a UTF-8 JSON document."""
return json.loads(Path(path).read_text(encoding="utf-8"))
def write_json(path, value):
"""Write deterministic metadata after the associated build succeeds."""
Path(path).write_text(
json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
def parse_narration(text, expected):
"""Read bracketed slide sections, rejecting missing, duplicate, or unknown IDs."""
sections = {}
current = None
lines = []
for line in text.splitlines():
match = re.fullmatch(r"\[([a-z][a-z0-9-]*)\]", line.strip())
if match:
if current is not None:
sections[current] = " ".join(lines).strip()
current = match[1]
if current in sections:
raise ValueError(f"Duplicate narration section [{current}].")
lines = []
elif current is None and line.strip():
raise ValueError("Narration must start with a [slide-id] section.")
else:
lines.append(line.strip())
if current is not None:
sections[current] = " ".join(lines).strip()
if set(sections) != set(expected):
raise ValueError(
f'Narration sections must match slide IDs: {", ".join(expected)}.'
)
for name, value in sections.items():
if not value:
raise ValueError(f"Narration section [{name}] is empty.")
return sections
def visual_hash(slide):
"""Hash visible slide copy with the same encoding as the deck renderer."""
data = [slide["id"], slide["title"], slide["body"], slide["screenshots"]]
return hashlib.sha256(
json.dumps(data, ensure_ascii=False, separators=(",", ":")).encode()
).hexdigest()
def asset_hash(slide, source):
"""Detect refreshed artwork, branding, or screenshots that need new slide renders."""
values = [file_hash(ROOT / "src/showcase/public/favicon.svg")]
values.append(
file_hash(source / "source/cover.png")
if slide["id"] in ("cover", "close")
else None
)
for name in slide["screenshots"]:
if not re.fullmatch(r"[a-z][a-z0-9-]*", name):
raise ValueError("Screenshot names must be simple lowercase identifiers.")
values.append(
[
name,
file_hash(
source / "source/screenshots" / f'{slide["locale"]}-{name}.png'
),
]
)
return hashlib.sha256(
json.dumps(values, ensure_ascii=False, separators=(",", ":")).encode()
).hexdigest()
def edge_options(settings, locale):
"""Validate locale-specific Edge speed and reversible single-word pronunciations."""
rate = settings.get("rates", {}).get(locale, settings["rate"])
if not isinstance(rate, str) or not re.fullmatch(r"[+-]\d+%", rate):
raise ValueError("Speech rate must be a signed percentage such as +0%.")
aliases = settings.get("pronunciations", {}).get(locale, {})
if not isinstance(aliases, dict) or any(
not isinstance(word, str)
or not isinstance(spoken, str)
or not word.isalpha()
or not spoken.isalpha()
for word, spoken in aliases.items()
):
raise ValueError(
"Pronunciation entries must map one alphabetic word to another."
)
if len(set(aliases.values())) != len(aliases) or any(
spoken in aliases and spoken != word for word, spoken in aliases.items()
):
raise ValueError(
"Pronunciation entries must have unique, non-overlapping spellings."
)
return rate, aliases
def pronunciation_text(text, aliases):
"""Replace exact whole words once without cascading into another spelling."""
if not aliases:
return text
pattern = r"(?<!\w)(?:" + "|".join(re.escape(word) for word in aliases) + r")(?!\w)"
def replacement(match):
"""Resolve one matched word to its configured pronunciation."""
return aliases[match.group()]
return re.sub(pattern, replacement, text)
def load_project(source, locales):
"""Validate all selected inputs before making network calls or changing output."""
story = read_json(source / "storyboard.json")
settings = read_json(source / "settings.json")
manifest = read_json(source / "source/slides/manifest.json")
if settings["schemaVersion"] != 1:
raise ValueError("Unsupported presentation settings schema.")
if not re.fullmatch(r"[+-]\d+%", settings["rate"]):
raise ValueError("Speech rate must be a signed percentage such as -4%.")
for name, minimum, maximum in [
("fps", 1, 60),
("leadSeconds", 0, 10),
("tailSeconds", 0, 10),
("fadeSeconds", 0, 5),
("crf", 0, 63),
]:
number = settings[name]
if (
not isinstance(number, (int, float))
or not math.isfinite(number)
or not minimum <= number <= maximum
):
raise ValueError(f"Invalid {name} setting.")
if not re.fullmatch(r"\d+k", settings["audioBitrate"]):
raise ValueError("Audio bitrate must be expressed in kilobits, such as 96k.")
selected = locales or list(story["languages"])
result = {}
for locale in selected:
if locale not in story["languages"] or not re.fullmatch(r"[a-z]{2}", locale):
raise ValueError(f"Unknown presentation language: {locale}.")
if not settings["voices"].get(locale):
raise ValueError(f"Missing voice for {locale}.")
edge_options(settings, locale)
slides = story["languages"][locale]
ids = [slide["id"] for slide in slides]
if (
not ids
or len(ids) != len(set(ids))
or any(not re.fullmatch(r"[a-z][a-z0-9-]*", key) for key in ids)
):
raise ValueError(f"Invalid or duplicate slide IDs for {locale}.")
narration = parse_narration(
(source / "narration" / f"{locale}.txt").read_text(encoding="utf-8"), ids
)
entries = []
for index, slide in enumerate(slides, 1):
key = f"{locale}-{index:02d}"
image = source / "source/slides" / f"{key}.png"
header = image.read_bytes()[:24]
if (
len(header) < 24
or header[:8] != b"\x89PNG\r\n\x1a\n"
or struct.unpack(">II", header[16:24]) != (1920, 1080)
):
raise ValueError(f"{image} must be a 1920 × 1080 PNG.")
record = manifest.get(key, {})
if (
record.get("visualHash") != visual_hash(slide)
or record.get("imageHash") != file_hash(image)
or record.get("assetHash")
!= asset_hash({**slide, "locale": locale}, source)
):
raise ValueError(
f"{key} has stale slide images. Run presentation:slides before building the video."
)
entries.append(
{**slide, "narration": narration[slide["id"]], "image": image}
)
result[locale] = entries
return settings, result
def cached(paths, metadata):
"""Accept cached files only when their recorded content digests still match."""
try:
hashes = read_json(metadata)
return all(file_hash(path) == hashes[path.suffix] for path in paths)
except (OSError, ValueError, KeyError):
return False
def record_cache(paths, metadata):
"""Commit a cache record after every output file has been written."""
write_json(metadata, {path.suffix: file_hash(path) for path in paths})
async def speech(text, voice, rate, audio, boundaries):
"""Generate one slide's MP3 and word timings, retrying transient service failures."""
import edge_tts
for attempt in range(3):
try:
cues = []
communicate = edge_tts.Communicate(
text, voice, rate=rate, boundary="WordBoundary"
)
with audio.open("wb") as output:
async for chunk in communicate.stream():
if chunk["type"] == "audio":
output.write(chunk["data"])
elif chunk["type"] == "WordBoundary":
cues.append(
{key: chunk[key] for key in ("offset", "duration", "text")}
)
if not cues or audio.stat().st_size == 0:
raise ValueError(
"The speech service returned no usable audio or word timings."
)
write_json(boundaries, cues)
return
except Exception:
if attempt == 2:
raise
await asyncio.sleep(2)
def ensure_audio(slide, locale, settings, cache, offline, force):
"""Regenerate Edge speech when text, pronunciation, voice, or rate changes."""
rate, aliases = edge_options(settings, locale)
spoken = pronunciation_text(slide["narration"], aliases)
identity = {
"version": CACHE_VERSION,
"text": slide["narration"],
"voice": settings["voices"][locale],
"rate": rate,
"requirements": file_hash(Path(__file__).with_name("requirements.txt")),
}
if aliases:
identity["spokenText"] = spoken
identity["pronunciations"] = aliases
key = digest(identity)
folder = cache / "audio"
folder.mkdir(parents=True, exist_ok=True)
audio, boundaries, metadata = [
folder / f"{key}{suffix}" for suffix in (".mp3", ".json", ".cache")
]
if not force and cached([audio, boundaries], metadata):
return audio, read_json(boundaries)
if offline:
raise ValueError(
f'Offline speech cache missing or stale for {locale}/{slide["id"]}. Run without --offline to synthesize it.'
)
print(f'Speech: {locale}/{slide["id"]}', flush=True)
with tempfile.TemporaryDirectory(dir=folder) as temporary:
temporary = Path(temporary)
produced_audio, produced_boundaries = (
temporary / "speech.mp3",
temporary / "timings.json",
)
asyncio.run(
speech(
spoken,
settings["voices"][locale],
rate,
produced_audio,
produced_boundaries,
)
)
if aliases:
reverse = {spelling: word for word, spelling in aliases.items()}
words = read_json(produced_boundaries)
for word in words:
word["text"] = pronunciation_text(word["text"], reverse)
write_json(produced_boundaries, words)
os.replace(produced_audio, audio)
os.replace(produced_boundaries, boundaries)
record_cache([audio, boundaries], metadata)
return audio, read_json(boundaries)
def execute(ffmpeg, args):
"""Run FFmpeg without shell interpolation and surface bounded error output."""
result = subprocess.run(
[ffmpeg, "-hide_banner", "-loglevel", "error", "-y", *map(str, args)],
capture_output=True,
text=True,
)
if result.returncode:
raise RuntimeError(f"FFmpeg failed:\n{result.stderr[-4000:]}")
def audio_duration(ffmpeg, audio):
"""Read speech duration from the same FFmpeg binary used for encoding."""
result = subprocess.run(
[ffmpeg, "-hide_banner", "-i", str(audio)], capture_output=True, text=True
)
match = re.search(r"Duration: (\d+):(\d+):(\d+\.\d+)", result.stderr)
if not match:
raise ValueError(f"Cannot decode speech duration: {audio}.")
return int(match[1]) * 3600 + int(match[2]) * 60 + float(match[3])
def ensure_segment(slide, audio, length, settings, cache, ffmpeg, force):
"""Encode a content-addressed slide clip with normalized speech and short fades."""
key = digest(
{
"version": CACHE_VERSION,
"image": file_hash(slide["image"]),
"audio": file_hash(audio),
"settings": {
name: settings[name]
for name in (
"fps",
"leadSeconds",
"tailSeconds",
"fadeSeconds",
"crf",
"audioBitrate",
)
},
"length": length,
"ffmpeg": file_hash(ffmpeg),
}
)
folder = cache / "segments"
folder.mkdir(parents=True, exist_ok=True)
output, metadata = folder / f"{key}.webm", folder / f"{key}.cache"
if not force and cached([output], metadata):
return output
fade = min(settings["fadeSeconds"], length / 2)
video_filter = (
f"fade=t=in:st=0:d={fade},fade=t=out:st={length - fade}:d={fade},format=yuv420p"
)
audio_filter = f'loudnorm=I=-16:TP=-1.5:LRA=9,adelay={round(settings["leadSeconds"] * 1000)},apad=whole_dur={length}'
print(f'Encode: {slide["id"]}', flush=True)
with tempfile.TemporaryDirectory(dir=folder) as temporary:
candidate = Path(temporary) / "clip.webm"
execute(
ffmpeg,
[
"-loop",
"1",
"-framerate",
settings["fps"],
"-i",
slide["image"],
"-i",
audio,
"-t",
length,
"-vf",
video_filter,
"-af",
audio_filter,
"-c:v",
"libvpx-vp9",
"-crf",
settings["crf"],
"-b:v",
"0",
"-cpu-used",
"5",
"-row-mt",
"1",
"-threads",
"4",
"-g",
240,
"-c:a",
"libopus",
"-b:a",
settings["audioBitrate"],
candidate,
],
)
os.replace(candidate, output)
record_cache([output], metadata)
return output
def timestamp(seconds):
"""Format nonnegative seconds as an exact WebVTT timestamp."""
milliseconds = max(0, round(seconds * 1000))
hours, milliseconds = divmod(milliseconds, 3600000)
minutes, milliseconds = divmod(milliseconds, 60000)
seconds, milliseconds = divmod(milliseconds, 1000)
return f"{hours:02d}:{minutes:02d}:{seconds:02d}.{milliseconds:03d}"
def caption_cues(words, original, offset):
"""Restore narration punctuation and group service timings into readable captions."""
positions = [
index for index, character in enumerate(original) if character.isalnum()
]
normalized = "".join(original[index].lower() for index in positions)
cursor = 0
result, group = [], []
for index, word in enumerate(words):
word = dict(word)
token = "".join(
character.lower() for character in word["text"] if character.isalnum()
)
match = normalized.find(token, cursor)
if token and match >= 0:
finish = match + len(token)
stop = positions[finish] if finish < len(positions) else len(original)
word["text"] = original[positions[match] : stop].strip()
cursor = finish
group.append(word)
text = " ".join(item["text"] for item in group)
if (
len(text) > 66
or len(group) >= 10
or text.endswith((".", "?", "!"))
or index == len(words) - 1
):
start = offset + group[0]["offset"] / 1e7
end = offset + (group[-1]["offset"] + group[-1]["duration"]) / 1e7
if end <= start or start < 0:
raise ValueError("Speech service returned invalid caption timings.")
text = (
text.replace("A G", "AG")
.replace("no V N C", "noVNC")
.replace("R F B", "RFB")
)
spaces = [
position for position, character in enumerate(text) if character == " "
]
if len(text) > 44 and spaces:
middle = min(spaces, key=lambda position: abs(position - len(text) / 2))
text = text[:middle] + "\n" + text[middle + 1 :]
result.extend(
[
f"{timestamp(start)} --> {timestamp(end)}",
html.escape(text, quote=False),
"",
]
)
group = []
return result
def concat_path(path):
"""Escape a local filename for FFmpeg's concat demuxer, including apostrophes."""
return (
"file '" + str(path.resolve()).replace("\\", "/").replace("'", "'\\''") + "'\n"
)
def output_paths(source, output=None, artifacts=None):
"""Separate released videos from project artifacts, including isolated review builds."""
videos = (output or ROOT / "src/showcase/assets").resolve()
project = (
artifacts
or (
videos.with_name(videos.name + "-artifacts")
if output
else source / "artifacts"
)
).resolve()
if project == videos or videos in project.parents or project in videos.parents:
raise ValueError(
"Video and artifact output directories must be separate, non-nested directories."
)
return videos, project
def build_language(
locale, slides, settings, output, artifacts, cache, ffmpeg, offline=False, force=False
):
"""Stage a video and project-only artifacts before replacing successful outputs."""
output, artifacts = output_paths(SOURCE, output, artifacts)
segments, timeline, cues = [], [], ["WEBVTT", ""]
offset = 0
for slide in slides:
audio, words = ensure_audio(slide, locale, settings, cache, offline, force)
length = (
math.ceil(
(
audio_duration(ffmpeg, audio)
+ settings["leadSeconds"]
+ settings["tailSeconds"]
)
* settings["fps"]
)
/ settings["fps"]
)
segment = ensure_segment(slide, audio, length, settings, cache, ffmpeg, force)
segments.append(segment)
timeline.append({"id": slide["id"], "start": offset, "duration": length})
cues.extend(
caption_cues(
words,
slide["narration"],
offset + settings["leadSeconds"],
)
)
offset += length
output.mkdir(parents=True, exist_ok=True)
artifacts.mkdir(parents=True, exist_ok=True)
stem = f"atrium-presentation-{locale}"
with tempfile.TemporaryDirectory(
prefix=".presentation-", dir=output
) as temporary, tempfile.TemporaryDirectory(
prefix=".presentation-", dir=artifacts
) as auxiliary:
stage = Path(temporary)
project_stage = Path(auxiliary)
playlist = project_stage / "concat.txt"
playlist.write_text(
"".join(concat_path(segment) for segment in segments), encoding="utf-8"
)
execute(
ffmpeg,
[
"-f",
"concat",
"-safe",
"0",
"-i",
playlist,
"-c",
"copy",
stage / f"{stem}.webm",
],
)
execute(
ffmpeg,
[
"-i",
slides[0]["image"],
"-frames:v",
"1",
"-q:v",
"2",
project_stage / f"{stem}.jpg",
],
)
(project_stage / f"{stem}.vtt").write_text(
"\n".join(cues) + "\n", encoding="utf-8"
)
transcript = (
"Agon Ātrium — Agon Partners Innovation AG\n\n"
+ "\n\n".join(
f'{index:02d}. {slide["title"].replace(chr(10), " ")}\n\n{slide["narration"]}'
for index, slide in enumerate(slides, 1)
)
+ "\n"
)
(project_stage / f"{stem}.txt").write_text(transcript, encoding="utf-8")
for extension, folder in (
("jpg", "posters"), ("vtt", "captions"), ("txt", "transcripts")
):
destination = artifacts / folder
destination.mkdir(parents=True, exist_ok=True)
os.replace(
project_stage / f"{stem}.{extension}",
destination / f"{stem}.{extension}",
)
os.replace(stage / f"{stem}.webm", output / f"{stem}.webm")
write_json(
cache / f"{locale}-timeline.json", {"duration": offset, "slides": timeline}
)
print(
f'Built {locale}: {len(slides)} slides, {offset:.1f} seconds, {output / (stem + ".webm")}',
flush=True,
)