Timing & anchors

Skip the aligner: ingest a TTS engine's own phoneme or viseme timing (espeak / MBROLA, Piper, Cartesia, Azure, Polly), or pin the naive aligner at known word/segment boundaries from subtitle cues and TTS word timestamps.

openfacefx.timing

Normalized TTS timing schema + vendor adapters (issue #14).

TTS engines already know exactly when each phoneme or viseme happens, so they can replace the aligner entirely. This module defines one intermediate schema -- TimingEvent(unit, symbol, start, end) -- and parsers that turn each vendor's dump into a list of them. From there:

• phoneme-unit events -> to_segments -> the existing weighted mapping and coarticulation, unchanged. The symbol is used verbatim as the phoneme label, so the source's alphabet must match the active mapping's: the default pipeline expects ARPABET, while Piper/Cartesia emit IPA and MBROLA emits its voice's SAMPA -- feed those a matching --mapping (which may itself set allow_custom_symbols) for meaningful visemes.
• viseme-unit events (Azure, Polly) skip phoneme->target mapping: the vendor's small fixed symbol set is remapped straight onto the Oculus-15 targets via AZURE_VISEME_TO_TARGET / POLLY_VISEME_TO_TARGET and fed to coarticulation through a Mapping(allow_custom_symbols=True).

Only start times are guaranteed. resolve_ends fills any missing end from the next event's start (the start-time-only sources, Azure and Polly), with a configurable duration for the final event.

Time units per vendor, all converted to float seconds: * MBROLA .pho -- per-phoneme duration in milliseconds, cumulative. * Piper -- per-phoneme audio sample counts / voice sample_rate. * Cartesia -- explicit start/end seconds. * Azure viseme -- audio offset in 100-ns ticks (ticks / 10000 = ms). * Polly marks -- time in integer milliseconds.

Parsers are pure stdlib text/JSON (numpy is not imported here) and reject malformed input with a clear ValueError. Field names for the SDK-event sources (Azure viseme JSON; the espeak C API phoneme path) are set by the ~10-line capture scripts in docs/timing.md; the parsers also accept the obvious aliases.

AZURE_VISEME_TO_TARGET: Dict[int, str] = {0: 'sil', 1: 'aa', 2: 'aa', 3: 'O', 4: 'E', 5: 'RR', 6: 'I', 7: 'U', 8: 'O', 9: 'O', 10: 'O', 11: 'aa', 12: 'kk', 13: 'RR', 14: 'DD', 15: 'SS', 16: 'CH', 17: 'TH', 18: 'FF', 19: 'DD', 20: 'kk', 21: 'PP'} module-attribute

POLLY_VISEME_TO_TARGET: Dict[str, str] = {'sil': 'sil', 'p': 'PP', 't': 'DD', 'S': 'CH', 'T': 'TH', 'f': 'FF', 'k': 'kk', 'i': 'I', 'r': 'RR', 's': 'SS', 'u': 'U', '@': 'aa', 'a': 'aa', 'e': 'E', 'E': 'E', 'o': 'O', 'O': 'O', 'l': 'DD'} module-attribute

TimingEvent(unit: str, symbol: str, start: float, end: Optional[float] = None) dataclass

unit: str instance-attribute

symbol: str instance-attribute

start: float instance-attribute

end: Optional[float] = None class-attribute instance-attribute

resolve_ends(events: List[TimingEvent], final_duration: float = 0.08) -> List[TimingEvent]

Fill every end is None from the next event's start; the last such event gets start + final_duration. Events that already carry an end (the duration / explicit-end sources) are left untouched. Order is preserved.

Source code in src/openfacefx/timing.py

def resolve_ends(events: List[TimingEvent],
                 final_duration: float = 0.08) -> List[TimingEvent]:
    """Fill every ``end is None`` from the next event's start; the last such
    event gets ``start + final_duration``. Events that already carry an end (the
    duration / explicit-end sources) are left untouched. Order is preserved."""
    if final_duration <= 0.0:
        raise ValueError("final_duration must be > 0")
    out: List[TimingEvent] = []
    n = len(events)
    for i, e in enumerate(events):
        end = e.end
        if end is None:
            end = events[i + 1].start if i + 1 < n else e.start + final_duration
        if end < e.start:                # guard out-of-order / coincident starts
            end = e.start
        out.append(TimingEvent(e.unit, e.symbol, e.start, end))
    return out

to_segments(events: List[TimingEvent]) -> List[PhonemeSegment]

Phoneme-unit events -> PhonemeSegment list (ends must be resolved). The symbol becomes the phoneme label verbatim.

Source code in src/openfacefx/timing.py

def to_segments(events: List[TimingEvent]) -> List[PhonemeSegment]:
    """Phoneme-unit events -> ``PhonemeSegment`` list (ends must be resolved).
    The symbol becomes the phoneme label verbatim."""
    segs: List[PhonemeSegment] = []
    for e in events:
        if e.unit != "phoneme":
            raise ValueError(
                f"to_segments: expected phoneme-unit events, got {e.unit!r}")
        if e.end is None:
            raise ValueError(
                "to_segments: unresolved end (call resolve_ends first)")
        segs.append(PhonemeSegment(e.symbol, e.start, e.end))
    return segs

parse_pho(text: str) -> List[TimingEvent]

MBROLA .pho: one PHONEME DURATION_MS [pos% pitch_hz ...] per line. ; starts a comment; blank and comment lines are ignored. Durations are cumulative from t=0; the trailing pitch-target pairs are ignored (only the timing matters). Symbols are the MBROLA voice's alphabet (a SAMPA variant), not ARPABET.

Source code in src/openfacefx/timing.py

def parse_pho(text: str) -> List[TimingEvent]:
    """MBROLA .pho: one ``PHONEME DURATION_MS [pos% pitch_hz ...]`` per line.
    ``;`` starts a comment; blank and comment lines are ignored. Durations are
    cumulative from t=0; the trailing pitch-target pairs are ignored (only the
    timing matters). Symbols are the MBROLA voice's alphabet (a SAMPA variant),
    not ARPABET."""
    events: List[TimingEvent] = []
    t = 0.0
    for lineno, raw in enumerate(text.splitlines(), 1):
        line = raw.split(";", 1)[0].strip()
        if not line:
            continue
        parts = line.split()
        if len(parts) < 2:
            raise ValueError(
                f".pho line {lineno}: expected 'PHONEME DURATION_MS', got {raw!r}")
        try:
            dur_ms = float(parts[1])
        except ValueError:
            raise ValueError(
                f".pho line {lineno}: duration {parts[1]!r} is not a number"
            ) from None
        if dur_ms < 0.0:
            raise ValueError(f".pho line {lineno}: negative duration {dur_ms}")
        start = t
        t += dur_ms / 1000.0
        events.append(TimingEvent("phoneme", parts[0], start, t))
    if not events:
        raise ValueError(".pho: no phoneme lines found")
    return events

parse_piper_alignments(json_text: str, sample_rate: int) -> List[TimingEvent]

Piper per-phoneme alignments: audio sample counts -> seconds via the voice sample_rate. Accepts either parallel arrays {"phonemes": [...], "phoneme_id_samples": [...]} (the Python AudioChunk attribute names; alignments / samples are accepted aliases) or a list {"alignments": [{"phoneme": "h", "num_samples": 2205}, ...]}. Counts are cumulative from t=0. Piper's docs name the fields but ship no example JSON, so both shapes are supported; the sample-count semantics are documented.

Source code in src/openfacefx/timing.py

def parse_piper_alignments(json_text: str, sample_rate: int) -> List[TimingEvent]:
    """Piper per-phoneme alignments: audio **sample counts** -> seconds via the
    voice ``sample_rate``. Accepts either parallel arrays
    ``{"phonemes": [...], "phoneme_id_samples": [...]}`` (the Python AudioChunk
    attribute names; ``alignments`` / ``samples`` are accepted aliases) or a list
    ``{"alignments": [{"phoneme": "h", "num_samples": 2205}, ...]}``. Counts are
    cumulative from t=0. Piper's docs name the fields but ship no example JSON, so
    both shapes are supported; the sample-count semantics are documented."""
    if sample_rate <= 0:
        raise ValueError("sample_rate must be a positive integer")
    try:
        d = json.loads(json_text)
    except json.JSONDecodeError as e:
        raise ValueError(f"piper: not valid JSON ({e})") from None
    phonemes, samples = _piper_arrays(d)
    if len(phonemes) != len(samples):
        raise ValueError(
            f"piper: {len(phonemes)} phonemes but {len(samples)} sample counts")
    if not phonemes:
        raise ValueError("piper: no phonemes in alignment dump")
    events: List[TimingEvent] = []
    t = 0.0
    for ph, ns in zip(phonemes, samples):
        if isinstance(ns, bool) or not isinstance(ns, (int, float)) or ns < 0:
            raise ValueError(f"piper: bad sample count {ns!r} for phoneme {ph!r}")
        start = t
        t += ns / sample_rate
        events.append(TimingEvent("phoneme", str(ph), start, t))
    return events

parse_cartesia(json_text: str) -> List[TimingEvent]

Cartesia phoneme_timestamps: parallel phonemes / start / end arrays already in seconds. Accepts the full stream message {"phoneme_timestamps": {...}} or a bare object carrying the three arrays. Symbols are IPA.

Source code in src/openfacefx/timing.py

def parse_cartesia(json_text: str) -> List[TimingEvent]:
    """Cartesia ``phoneme_timestamps``: parallel ``phonemes`` / ``start`` /
    ``end`` arrays already in **seconds**. Accepts the full stream message
    ``{"phoneme_timestamps": {...}}`` or a bare object carrying the three
    arrays. Symbols are IPA."""
    try:
        d = json.loads(json_text)
    except json.JSONDecodeError as e:
        raise ValueError(f"cartesia: not valid JSON ({e})") from None
    if isinstance(d, dict) and "phoneme_timestamps" in d:
        d = d["phoneme_timestamps"]
    if not isinstance(d, dict):
        raise ValueError("cartesia: expected a 'phoneme_timestamps' object")
    phonemes, starts, ends = d.get("phonemes"), d.get("start"), d.get("end")
    if not (isinstance(phonemes, list) and isinstance(starts, list)
            and isinstance(ends, list)):
        raise ValueError(
            "cartesia: 'phonemes', 'start' and 'end' arrays are required")
    if not len(phonemes) == len(starts) == len(ends):
        raise ValueError("cartesia: phonemes/start/end arrays differ in length")
    if not phonemes:
        raise ValueError("cartesia: empty phoneme_timestamps")
    events: List[TimingEvent] = []
    for ph, s, e in zip(phonemes, starts, ends):
        try:
            s, e = float(s), float(e)
        except (TypeError, ValueError):
            raise ValueError(
                f"cartesia: non-numeric time for phoneme {ph!r}") from None
        events.append(TimingEvent("phoneme", str(ph), s, e))
    return events

parse_azure_visemes(json_text: str) -> List[TimingEvent]

Azure VisemeReceived events: audio offset in 100-ns ticks (ticks / 10000 = ms) + integer viseme ID (documented range 0-21). Accepts a bare JSON array of events or {"visemes"|"events": [...]}; each event needs an offset (audio_offset) and an id (viseme_id). Ends come from the next event (resolve_ends). Symbol is str(id); IDs outside the documented table are kept here and flagged as unknown at mapping time.

Source code in src/openfacefx/timing.py

def parse_azure_visemes(json_text: str) -> List[TimingEvent]:
    """Azure VisemeReceived events: audio offset in **100-ns ticks**
    (ticks / 10000 = ms) + integer viseme ID (documented range 0-21). Accepts a
    bare JSON array of events or ``{"visemes"|"events": [...]}``; each event
    needs an offset (``audio_offset``) and an id (``viseme_id``). Ends come from
    the next event (``resolve_ends``). Symbol is ``str(id)``; IDs outside the
    documented table are kept here and flagged as unknown at mapping time."""
    try:
        d = json.loads(json_text)
    except json.JSONDecodeError as e:
        raise ValueError(f"azure: not valid JSON ({e})") from None
    items = _event_list(d, ("visemes", "events"), "azure")
    events: List[TimingEvent] = []
    for i, ev in enumerate(items):
        if not isinstance(ev, dict):
            raise ValueError(f"azure: event {i} is not an object")
        ticks = _first_key(ev, _AZURE_OFFSET_KEYS)
        vid = _as_int(_first_key(ev, _AZURE_ID_KEYS))
        if ticks is None or vid is None:
            raise ValueError(
                f"azure: event {i} needs a numeric offset and integer id, got {ev!r}")
        try:
            ticks = float(ticks)
        except (TypeError, ValueError):
            raise ValueError(
                f"azure: event {i} offset {ticks!r} not numeric") from None
        events.append(TimingEvent("viseme", str(vid), ticks / 1e7, None))
    if not events:
        raise ValueError("azure: no viseme events found")
    return events

parse_polly_marks(text: str) -> List[TimingEvent]

Amazon Polly speech marks: newline-delimited JSON, one object per line. Only type == 'viseme' lines are kept; time is integer milliseconds and value the viseme symbol. A mark's start/end are byte offsets into the input text, not time, and are ignored. Ends come from the next viseme mark (resolve_ends).

Source code in src/openfacefx/timing.py

def parse_polly_marks(text: str) -> List[TimingEvent]:
    """Amazon Polly speech marks: newline-delimited JSON, one object per line.
    Only ``type == 'viseme'`` lines are kept; ``time`` is integer
    **milliseconds** and ``value`` the viseme symbol. A mark's ``start``/``end``
    are byte offsets into the input text, not time, and are ignored. Ends come
    from the next viseme mark (``resolve_ends``)."""
    events: List[TimingEvent] = []
    for lineno, raw in enumerate(text.splitlines(), 1):
        line = raw.strip()
        if not line:
            continue
        try:
            mark = json.loads(line)
        except json.JSONDecodeError as e:
            raise ValueError(f"polly line {lineno}: not valid JSON ({e})") from None
        if not isinstance(mark, dict) or "type" not in mark:
            raise ValueError(f"polly line {lineno}: not a speech-mark object")
        if mark["type"] != "viseme":
            continue
        if "time" not in mark or "value" not in mark:
            raise ValueError(
                f"polly line {lineno}: viseme mark needs 'time' and 'value'")
        try:
            ms = float(mark["time"])
        except (TypeError, ValueError):
            raise ValueError(
                f"polly line {lineno}: time {mark['time']!r} not numeric") from None
        events.append(TimingEvent("viseme", str(mark["value"]), ms / 1000.0, None))
    if not events:
        raise ValueError("polly: no viseme marks found")
    return events

build_vendor_mapping(table: Dict) -> Mapping

A Mapping over the Oculus-15 targets whose rows are the vendor viseme symbols, each driving its target at weight 1.0. allow_custom_symbols lets numeric Azure IDs and case-significant Polly letters through verbatim. A sil fallback row absorbs any symbol routed to silence.

Source code in src/openfacefx/timing.py

def build_vendor_mapping(table: Dict) -> Mapping:
    """A ``Mapping`` over the Oculus-15 targets whose rows are the vendor viseme
    symbols, each driving its target at weight 1.0. ``allow_custom_symbols`` lets
    numeric Azure IDs and case-significant Polly letters through verbatim. A
    ``sil`` fallback row absorbs any symbol routed to silence."""
    table = _str_table(table)
    targets = [Target(v, _DEFAULT_CLASSES.get(v, "basic")) for v in VISEMES]
    rows: Dict[str, Dict[str, float]] = {s: {t: 1.0} for s, t in table.items()}
    rows[SILENCE] = {"sil": 1.0}
    return Mapping(targets, rows, allow_custom_symbols=True)

viseme_events_to_segments(events: List[TimingEvent], table: Dict) -> Tuple[List[PhonemeSegment], List[str]]

Resolved viseme events -> (segments, warnings). Each segment's phoneme is the vendor symbol, consumed by build_vendor_mapping(table). Symbols absent from table are routed to silence and reported as QA warnings (never a crash), per the acceptance criteria.

Source code in src/openfacefx/timing.py

def viseme_events_to_segments(
    events: List[TimingEvent], table: Dict,
) -> Tuple[List[PhonemeSegment], List[str]]:
    """Resolved viseme events -> (segments, warnings). Each segment's phoneme is
    the vendor symbol, consumed by ``build_vendor_mapping(table)``. Symbols
    absent from ``table`` are routed to silence and reported as QA warnings
    (never a crash), per the acceptance criteria."""
    known = _str_table(table)
    segs: List[PhonemeSegment] = []
    unknown: Dict[str, int] = {}
    for e in events:
        if e.unit != "viseme":
            raise ValueError(
                f"viseme_events_to_segments: expected viseme-unit events, "
                f"got {e.unit!r}")
        if e.end is None:
            raise ValueError(
                "viseme_events_to_segments: unresolved end (call resolve_ends first)")
        if e.symbol in known:
            segs.append(PhonemeSegment(e.symbol, e.start, e.end))
        else:
            unknown[e.symbol] = unknown.get(e.symbol, 0) + 1
            segs.append(PhonemeSegment(SILENCE, e.start, e.end))
    warnings = [f"unknown viseme symbol {s!r} ({n}x) routed to silence"
                for s, n in sorted(unknown.items())]
    return segs, warnings

openfacefx.anchors

Word/segment-anchored alignment (issue #15).

The naive aligner spreads phonemes across a whole utterance by duration priors. That is only as good as the assumption that speech fills the clip uniformly -- it does not. Anything that already knows when words happen (subtitle cue times, TTS word timestamps) can pin the aligner at those boundaries and let it distribute phonemes within each span. Same zero-ML machinery, far better sync.

This module adds:

• Anchor(text, start, end=None) -- one or more words pinned to a time span. end=None means "unknown": the span runs to the next anchor's start (or a duration-proportional share when uncovered words sit in that gap).

• anchored_segments(transcript, duration, anchors, g2p=None) -- the naive aligner, anchored. Anchor words are matched against the transcript sequentially (case/punctuation-insensitive, the regex G2P.phrase uses); their phonemes stay inside the anchor span; transcript words no anchor covers fill the time gaps between anchors; wordless gaps longer than ~0.15 s relax to sil; utterance edges are padded with sil exactly like pipeline.naive_segments (with no anchors the output is byte-identical).

• Parsers/converters, each pure stdlib and rejecting malformed input with a clear ValueError:

  • parse_srt -- SubRip cues (segment anchors), multi-line, HH:MM:SS,mmm.
  • parse_word_anchors -- generic [{text, start, end?}] (this project's own schema).
  • from_azure_word_boundaries -- Azure WordBoundary events.
  • from_elevenlabs_alignment -- character arrays grouped into words.
  • from_kokoro_tokens -- per-token start_ts/end_ts (None-tolerant).
  • google_ssml_with_marks / from_google_timepoints -- inject one <mark/> per word, then map returned timepoints back to words.

Field names are verified against vendor docs where a vendor exists (see each docstring: VERIFIED vs assumed); the snake_case aliases and object wrappers are the tolerant extras, mirroring timing.py. Times land in float seconds: Azure offsets/durations are 100-ns ticks (÷1e7); ElevenLabs, Kokoro and Google are already seconds.

Anchor(text: str, start: float, end: Optional[float] = None) dataclass

text: str instance-attribute

start: float instance-attribute

end: Optional[float] = None class-attribute instance-attribute

anchored_segments(transcript: str, duration: float, anchors: Optional[List[Anchor]] = None, g2p: Optional[G2P] = None) -> List[PhonemeSegment]

Time-stamped phonemes for transcript over duration seconds, pinned at anchors. With no anchors this is exactly pipeline.naive_segments.

Source code in src/openfacefx/anchors.py

def anchored_segments(
    transcript: str,
    duration: float,
    anchors: Optional[List[Anchor]] = None,
    g2p: Optional[G2P] = None,
) -> List[PhonemeSegment]:
    """Time-stamped phonemes for ``transcript`` over ``duration`` seconds, pinned
    at ``anchors``. With no anchors this is exactly ``pipeline.naive_segments``."""
    if duration <= 0.0:
        raise ValueError("duration must be > 0")
    g2p = g2p or G2P()
    anchors = list(anchors or [])
    words = _WORD_RE.findall(transcript)
    phones = [g2p.word(w) for w in words]
    keys = [_key(w) for w in words]

    _validate_anchors(anchors, duration)
    runs = _match_runs(anchors, keys)
    ends = _resolve_ends(anchors, runs, phones, len(words), duration)

    segs: List[PhonemeSegment] = []
    prev_t, prev_w = 0.0, 0
    n = len(anchors)
    for i, a in enumerate(anchors):
        s = max(0.0, min(a.start, duration))
        e = max(s, min(ends[i], duration))
        ws, we = runs[i]
        # Uncovered words sitting before this anchor share the gap time.
        segs += _gap(prev_t, s, phones[prev_w:ws], leading=(i == 0), trailing=False)
        aphones = [p for k in range(ws, we) for p in phones[k]]
        if aphones and e > s:
            segs += NaiveAligner().align(aphones, e - s, start=s)
        elif not aphones and e - s > 1e-9:
            segs.append(PhonemeSegment(SILENCE, s, e))   # non-lexical anchor (e.g. "♪")
        prev_t, prev_w = e, we
    segs += _gap(prev_t, duration, phones[prev_w:], leading=(n == 0), trailing=True)
    return segs

anchors_transcript(anchors: List[Anchor]) -> str

Concatenate cue text into a transcript -- what --anchors-format srt uses when --text is omitted.

Source code in src/openfacefx/anchors.py

def anchors_transcript(anchors: List[Anchor]) -> str:
    """Concatenate cue text into a transcript -- what ``--anchors-format srt``
    uses when ``--text`` is omitted."""
    return " ".join(a.text for a in anchors if a.text).strip()

parse_srt(text: str) -> List[Anchor]

SubRip subtitles -> segment anchors. Cues are blank-line separated; the HH:MM:SS,mmm --> HH:MM:SS,mmm line sets the span (. also accepted as the decimal mark) and every following line is the cue text, joined with spaces. A leading index line is ignored. Formatting tags (<i>, {\an8}) are stripped so they never leak into word matching.

Source code in src/openfacefx/anchors.py

def parse_srt(text: str) -> List[Anchor]:
    """SubRip subtitles -> segment anchors. Cues are blank-line separated; the
    ``HH:MM:SS,mmm --> HH:MM:SS,mmm`` line sets the span (``.`` also accepted as
    the decimal mark) and every following line is the cue text, joined with
    spaces. A leading index line is ignored. Formatting tags (``<i>``, ``{\\an8}``)
    are stripped so they never leak into word matching."""
    out: List[Anchor] = []
    for block in re.split(r"\r?\n[ \t]*\r?\n", text.strip()):
        lines = block.splitlines()
        ti = next((k for k, ln in enumerate(lines) if "-->" in ln), None)
        if ti is None:
            continue
        stamps = list(_SRT_TIME.finditer(lines[ti]))
        if len(stamps) < 2:
            raise ValueError(f"srt: malformed timecode line {lines[ti]!r}")
        cue = " ".join(lines[ti + 1:])
        cue = re.sub(r"<[^>]+>", "", cue)          # <i>, <b>, <font ...>
        cue = re.sub(r"\{[^}]*\}", "", cue)        # {\an8}, {\pos(..)}
        cue = re.sub(r"\s+", " ", cue).strip()
        out.append(Anchor(cue, _srt_seconds(stamps[0]), _srt_seconds(stamps[1])))
    if not out:
        raise ValueError("srt: no cues found")
    return out

parse_word_anchors(json_text: str) -> List[Anchor]

Generic word/segment anchors: a JSON array (or {"anchors"|"words": [...]}) of {"text": str, "start": seconds, "end": seconds|null}. This is OpenFaceFX's own normalized schema -- the target every converter below produces -- so start/end are plain float seconds. end may be omitted or null (inferred from the next anchor).

Source code in src/openfacefx/anchors.py

def parse_word_anchors(json_text: str) -> List[Anchor]:
    """Generic word/segment anchors: a JSON array (or ``{"anchors"|"words": [...]}``)
    of ``{"text": str, "start": seconds, "end": seconds|null}``. This is
    OpenFaceFX's own normalized schema -- the target every converter below
    produces -- so ``start``/``end`` are plain float seconds. ``end`` may be
    omitted or null (inferred from the next anchor)."""
    items = _obj_list(_json(json_text, "word-anchors"), ("anchors", "words"),
                      "word-anchors")
    out: List[Anchor] = []
    for i, it in enumerate(items):
        if not isinstance(it, dict):
            raise ValueError(f"word-anchors: item {i} is not an object")
        text = it.get("text", it.get("word"))
        if not isinstance(text, str):
            raise ValueError(f"word-anchors: item {i} needs a string 'text'")
        start = _num(it.get("start", it.get("start_s")), f"word-anchors {i} 'start'")
        raw_end = it.get("end", it.get("end_s"))
        end = None if raw_end is None else _num(raw_end, f"word-anchors {i} 'end'")
        out.append(Anchor(text, start, end))
    if not out:
        raise ValueError("word-anchors: no anchors found")
    return out

from_azure_word_boundaries(json_text: str) -> List[Anchor]

Azure Speech WordBoundary events -> anchors. VERIFIED field names (learn.microsoft.com/.../how-to-speech-synthesis): Text, AudioOffset and Duration in 100-ns ticks (÷1e7 = seconds), BoundaryType in {Word, Punctuation, Sentence}. The Python SDK lower-cases these (text/audio_offset/duration/boundary_type); both, plus a {"words"|"events": [...]} wrapper, are accepted. Punctuation boundaries (and any tokens without letters) are skipped so they do not offset matching.

Source code in src/openfacefx/anchors.py

def from_azure_word_boundaries(json_text: str) -> List[Anchor]:
    """Azure Speech ``WordBoundary`` events -> anchors. VERIFIED field names
    (learn.microsoft.com/.../how-to-speech-synthesis): ``Text``, ``AudioOffset``
    and ``Duration`` in **100-ns ticks** (÷1e7 = seconds), ``BoundaryType`` in
    {Word, Punctuation, Sentence}. The Python SDK lower-cases these
    (``text``/``audio_offset``/``duration``/``boundary_type``); both, plus a
    ``{"words"|"events": [...]}`` wrapper, are accepted. Punctuation boundaries
    (and any tokens without letters) are skipped so they do not offset matching."""
    items = _obj_list(_json(json_text, "azure-words"),
                      ("words", "events", "word_boundaries"), "azure-words")
    out: List[Anchor] = []
    for i, ev in enumerate(items):
        if not isinstance(ev, dict):
            raise ValueError(f"azure-words: event {i} is not an object")
        bt = _first(ev, _AZ_BOUNDARY)
        if isinstance(bt, str) and ("punct" in bt.lower() or "sentence" in bt.lower()):
            continue
        text = _first(ev, _AZ_TEXT)
        off = _first(ev, _AZ_OFFSET)
        if not isinstance(text, str) or off is None:
            raise ValueError(
                f"azure-words: event {i} needs 'Text' and 'AudioOffset', got {ev!r}")
        if not _WORD_RE.search(text):          # punctuation-only token
            continue
        start = _num(off, f"azure-words {i} offset") / 1e7
        dur = _first(ev, _AZ_DUR)
        end = None if dur is None else start + _num(dur, f"azure-words {i} duration") / 1e7
        out.append(Anchor(text, start, end))
    if not out:
        raise ValueError("azure-words: no word-boundary events found")
    return out

from_elevenlabs_alignment(json_text: str) -> List[Anchor]

ElevenLabs timestamped TTS -> anchors. VERIFIED field names (elevenlabs.io/docs/.../convert-with-timestamps): a top-level alignment and/or normalized_alignment object, each with parallel characters, character_start_times_seconds and character_end_times_seconds arrays (already seconds). normalized_alignment is preferred when present. A bare alignment object (the three arrays at top level) is also accepted. Characters are grouped into words at whitespace; a word spans its first character's start to its last character's end.

Source code in src/openfacefx/anchors.py

def from_elevenlabs_alignment(json_text: str) -> List[Anchor]:
    """ElevenLabs timestamped TTS -> anchors. VERIFIED field names
    (elevenlabs.io/docs/.../convert-with-timestamps): a top-level ``alignment``
    and/or ``normalized_alignment`` object, each with parallel ``characters``,
    ``character_start_times_seconds`` and ``character_end_times_seconds`` arrays
    (already **seconds**). ``normalized_alignment`` is preferred when present. A
    bare alignment object (the three arrays at top level) is also accepted.
    Characters are grouped into words at whitespace; a word spans its first
    character's start to its last character's end."""
    d = _json(json_text, "elevenlabs")
    if not isinstance(d, dict):
        raise ValueError("elevenlabs: expected a JSON object")
    al = d.get("normalized_alignment") or d.get("alignment")
    if al is None and "characters" in d:
        al = d
    if not isinstance(al, dict):
        raise ValueError(
            "elevenlabs: no 'alignment' or 'normalized_alignment' object")
    chars = al.get("characters")
    starts = al.get("character_start_times_seconds")
    ends = al.get("character_end_times_seconds")
    if not (isinstance(chars, list) and isinstance(starts, list)
            and isinstance(ends, list)):
        raise ValueError(
            "elevenlabs: alignment needs 'characters', "
            "'character_start_times_seconds', 'character_end_times_seconds' arrays")
    if not len(chars) == len(starts) == len(ends):
        raise ValueError("elevenlabs: character arrays differ in length")
    if not chars:
        raise ValueError("elevenlabs: empty alignment")
    out: List[Anchor] = []
    cur: List[str] = []
    cs = ce = 0.0
    for ch, s, e in zip(chars, starts, ends):
        if not isinstance(ch, str):
            raise ValueError("elevenlabs: non-string character in 'characters'")
        if ch.strip() == "":                    # whitespace closes the word
            if cur:
                out.append(Anchor("".join(cur), cs, ce))
                cur = []
            continue
        if not cur:
            cs = _num(s, "elevenlabs start")
        cur.append(ch)
        ce = _num(e, "elevenlabs end")
    if cur:
        out.append(Anchor("".join(cur), cs, ce))
    if not out:
        raise ValueError("elevenlabs: no words after grouping at whitespace")
    return out

from_kokoro_tokens(json_text: str) -> List[Anchor]

Kokoro per-token timestamps -> anchors. Field names from the community write-up (ryanwelch.co.uk/blog/kokoro-word-timestamps; not a vendor doc): tokens carry text, start_ts and end_ts (float seconds, either may be None). A JSON array or {"tokens": [...]} is accepted. Tokens with start_ts is None are dropped (that word falls back to the surrounding gap distribution); end_ts is None leaves the span open to the next token.

Source code in src/openfacefx/anchors.py

def from_kokoro_tokens(json_text: str) -> List[Anchor]:
    """Kokoro per-token timestamps -> anchors. Field names from the community
    write-up (ryanwelch.co.uk/blog/kokoro-word-timestamps; not a vendor doc):
    tokens carry ``text``, ``start_ts`` and ``end_ts`` (float **seconds**, either
    may be ``None``). A JSON array or ``{"tokens": [...]}`` is accepted. Tokens
    with ``start_ts is None`` are dropped (that word falls back to the surrounding
    gap distribution); ``end_ts is None`` leaves the span open to the next token."""
    items = _obj_list(_json(json_text, "kokoro"), ("tokens", "words"), "kokoro")
    out: List[Anchor] = []
    for i, tok in enumerate(items):
        if not isinstance(tok, dict):
            raise ValueError(f"kokoro: token {i} is not an object")
        text = tok.get("text", tok.get("graphemes"))
        if not isinstance(text, str) or not _WORD_RE.search(text):
            continue                            # whitespace / punctuation token
        st = tok.get("start_ts")
        if st is None:                          # no start -> cannot anchor it
            continue
        start = _num(st, f"kokoro token {i} start_ts")
        et = tok.get("end_ts")
        end = None if et is None else _num(et, f"kokoro token {i} end_ts")
        out.append(Anchor(text.strip(), start, end))
    if not out:
        raise ValueError("kokoro: no timestamped word tokens found")
    return out

google_ssml_with_marks(transcript: str) -> str

Insert an SSML <mark name="wN"/> before each transcript word so Google Cloud TTS (synthesized with enableTimePointing=SSML_MARK) returns one timepoint per word. Pure text transform, no SDK: the original text is kept intact (punctuation and spacing preserved for natural prosody) with XML metacharacters escaped; only the marks are added. from_google_timepoints maps the wN marks back to words using the same tokenisation.

Source code in src/openfacefx/anchors.py

def google_ssml_with_marks(transcript: str) -> str:
    """Insert an SSML ``<mark name="wN"/>`` before each transcript word so Google
    Cloud TTS (synthesized with ``enableTimePointing=SSML_MARK``) returns one
    timepoint per word. Pure text transform, no SDK: the original text is kept
    intact (punctuation and spacing preserved for natural prosody) with XML
    metacharacters escaped; only the marks are added. ``from_google_timepoints``
    maps the ``wN`` marks back to words using the same tokenisation."""
    out: List[str] = []
    pos = 0
    for i, m in enumerate(_WORD_RE.finditer(transcript)):
        out.append(_xml_escape(transcript[pos:m.start()]))   # inter-word text
        out.append(f'<mark name="w{i}"/>')
        out.append(_xml_escape(m.group()))
        pos = m.end()
    out.append(_xml_escape(transcript[pos:]))                # trailing text
    return "<speak>" + "".join(out) + "</speak>"

from_google_timepoints(json_text: str, transcript: str) -> List[Anchor]

Google Cloud TTS timepoints -> anchors. VERIFIED field names (cloud.google.com/.../v1beta1/text/synthesize): a top-level timepoints array of {"markName": str, "timeSeconds": number} (already seconds). A bare array is also accepted. Each wN mark (from google_ssml_with_marks) identifies transcript word N and pins its start; ends are left open (Google marks give word starts only), so the next word's start closes each span.

Source code in src/openfacefx/anchors.py

def from_google_timepoints(json_text: str, transcript: str) -> List[Anchor]:
    """Google Cloud TTS timepoints -> anchors. VERIFIED field names
    (cloud.google.com/.../v1beta1/text/synthesize): a top-level ``timepoints``
    array of ``{"markName": str, "timeSeconds": number}`` (already **seconds**). A
    bare array is also accepted. Each ``wN`` mark (from ``google_ssml_with_marks``)
    identifies transcript word N and pins its start; ends are left open (Google
    marks give word starts only), so the next word's start closes each span."""
    items = _obj_list(_json(json_text, "google"), ("timepoints",), "google")
    words = _WORD_RE.findall(transcript)
    out: List[Anchor] = []
    for i, tp in enumerate(items):
        if not isinstance(tp, dict):
            raise ValueError(f"google: timepoint {i} is not an object")
        name = tp.get("markName", tp.get("mark_name"))
        if not isinstance(name, str):
            raise ValueError(f"google: timepoint {i} needs a string 'markName'")
        mt = re.fullmatch(r"w(\d+)", name.strip())
        if not mt:
            raise ValueError(
                f"google: markName {name!r} is not 'w<index>' "
                "(use google_ssml_with_marks to inject marks)")
        idx = int(mt.group(1))
        if idx >= len(words):
            raise ValueError(
                f"google: mark {name!r} index {idx} is beyond the transcript "
                f"({len(words)} words)")
        ts = tp.get("timeSeconds", tp.get("time_seconds"))
        out.append(Anchor(words[idx], _num(ts, f"google timepoint {i} timeSeconds"),
                          None))
    if not out:
        raise ValueError("google: no timepoints found")
    return out