Pipeline & generation

The end-to-end orchestration plus the input stages that feed it: the naive aligner and Montreal Forced Aligner parser, grapheme-to-phoneme conversion, the ARPAbet phoneme inventory, the transcript-free energy fallback, the prosody tracker that turns pitch and loudness into typed events, and the edit-preservation layer that carries hand-tweaks across a regeneration.

openfacefx.pipeline

End-to-end pipeline: (audio, text) -> FaceTrack.

Two entry points:

• generate_from_alignment -- you already have time-stamped phonemes (from MFA, or Gentle/Whisper/WhisperX via the built-in :mod:openfacefx.aligners adapters, issue #54). This is the accurate path.

• generate_naive -- you only have text and an audio duration. Uses G2P + NaiveAligner. Fast, dependency-free, approximate lip-sync for prototyping.

wav_duration(path: str) -> float

Duration of a PCM WAV in seconds, using only the stdlib.

Source code in src/openfacefx/pipeline.py

def wav_duration(path: str) -> float:
    """Duration of a PCM WAV in seconds, using only the stdlib."""
    with contextlib.closing(wave.open(path, "rb")) as w:
        return w.getnframes() / float(w.getframerate())

generate_from_alignment(segments: List[PhonemeSegment], fps: float = 60.0, epsilon: float = 0.015, mapping=None, params=None, gestures=None, wav: Optional[str] = None) -> FaceTrack

gestures opts into the non-verbal gesture layer (issue #5): pass a GestureParams (or True for defaults) to append blink/brow/head/eye channels after viseme reduction. Off (None) leaves output byte-identical. wav supplies the audio those energy-driven brows/nods read; without it they degrade gracefully (stress still comes from the segments).

Source code in src/openfacefx/pipeline.py

def generate_from_alignment(
    segments: List[PhonemeSegment],
    fps: float = 60.0,
    epsilon: float = 0.015,
    mapping=None,
    params=None,
    gestures=None,
    wav: Optional[str] = None,
) -> FaceTrack:
    """``gestures`` opts into the non-verbal gesture layer (issue #5): pass a
    ``GestureParams`` (or ``True`` for defaults) to append blink/brow/head/eye
    channels after viseme reduction. Off (``None``) leaves output byte-identical.
    ``wav`` supplies the audio those energy-driven brows/nods read; without it
    they degrade gracefully (stress still comes from the segments)."""
    times, matrix = build_viseme_curves(segments, fps=fps, mapping=mapping,
                                        params=params)
    targets = mapping.targets if mapping is not None else None
    track = reduce_to_track(times, matrix, fps=fps, epsilon=epsilon,
                            targets=targets)
    # Lag/lead post-process (issue #10): slide the reduced viseme keyframes in
    # time before the gesture layer is attached, so only the mouth leads/trails
    # the audio (blinks/brows keep their own timing). Off (lag=0) => untouched.
    if params is not None and params.lag:
        from .postprocess import time_shift
        time_shift(track, params.lag)
    if gestures:
        _attach_gestures(track, segments, wav, gestures)
    return track

derive_events(segments: Optional[List[PhonemeSegment]] = None, env_times=None, env=None, emphasis: bool = True, phrase: bool = True, energy_thresh: float = 0.55, min_prominence: float = 0.15, min_spacing: float = 0.4) -> list

Auto-author a typed event layer from the speech itself (issue #6) -- emphasis events on stressed syllables / loudness peaks and phrase boundary markers at pauses -- WITHOUT touching the gesture channels.

It reuses the exact stress/pause/peak detectors behind :mod:openfacefx.gestures (gestures_layers), so an emphasis event and a head-nod gesture agree on where the accents are, yet the two layers stay independent and separately opt-in. Determinism is inherited from those numpy detectors (no RNG here): identical inputs give identical events.

segments supply ARPABET stress digits and sil pauses; env_times / env are an :func:openfacefx.energy.energy_envelope result. With segments, stress drives emphasis and sil spans drive phrase markers; with only an envelope (energy mode), loudness peaks drive emphasis and quiet runs drive phrase markers. Returns a time-sorted list[Event].

Source code in src/openfacefx/pipeline.py

def derive_events(
    segments: Optional[List[PhonemeSegment]] = None,
    env_times=None,
    env=None,
    emphasis: bool = True,
    phrase: bool = True,
    energy_thresh: float = 0.55,
    min_prominence: float = 0.15,
    min_spacing: float = 0.40,
) -> list:
    """Auto-author a typed event layer from the speech itself (issue #6) --
    ``emphasis`` events on stressed syllables / loudness peaks and ``phrase``
    boundary markers at pauses -- WITHOUT touching the gesture *channels*.

    It reuses the exact stress/pause/peak detectors behind
    :mod:`openfacefx.gestures` (``gestures_layers``), so an emphasis event and a
    head-nod gesture agree on where the accents are, yet the two layers stay
    independent and separately opt-in. Determinism is inherited from those
    numpy detectors (no RNG here): identical inputs give identical events.

    ``segments`` supply ARPABET stress digits and ``sil`` pauses; ``env_times`` /
    ``env`` are an :func:`openfacefx.energy.energy_envelope` result. With
    segments, stress drives emphasis and ``sil`` spans drive phrase markers; with
    only an envelope (energy mode), loudness peaks drive emphasis and quiet runs
    drive phrase markers. Returns a time-sorted ``list[Event]``."""
    import numpy as np
    from . import gestures_layers as _gl
    from .events import Event

    et = np.asarray(env_times, dtype=float) if env_times is not None else np.zeros(0)
    ev = np.asarray(env, dtype=float) if env is not None else np.zeros(0)
    out: list = []
    if emphasis:
        stresses = _gl.stress_events(segments, et, ev) if segments else []
        if stresses:
            out += [Event(t, "emphasis", "beat",
                          payload={"strength": round(float(s), 4)})
                    for t, s in stresses]
        elif len(ev):
            peaks = _gl.energy_peaks(et, ev, energy_thresh, min_prominence,
                                     min_spacing)
            out += [Event(t, "emphasis", "beat",
                          payload={"level": round(float(p), 4)})
                    for t, p in peaks]
    if phrase:
        out += [Event(t, "marker", "phrase") for t in _gl.pause_times(segments, et, ev)]
    out.sort(key=lambda e: e.t)
    return out

naive_segments(text: str, duration: float, g2p: Optional[G2P] = None) -> List[PhonemeSegment]

Time-stamped phonemes for text spread over duration seconds.

This is the phoneme-timing layer the curve solver consumes; exporters that need phonemes rather than visemes (e.g. Bethesda .LIP) start here.

Source code in src/openfacefx/pipeline.py

def naive_segments(
    text: str,
    duration: float,
    g2p: Optional[G2P] = None,
) -> List[PhonemeSegment]:
    """Time-stamped phonemes for ``text`` spread over ``duration`` seconds.

    This is the phoneme-timing layer the curve solver consumes; exporters that
    need phonemes rather than visemes (e.g. Bethesda .LIP) start here.
    """
    g2p = g2p or G2P()
    # Pad with silence at both ends so the mouth starts and ends relaxed.
    phones = [SILENCE] + g2p.phrase(text) + [SILENCE]
    return NaiveAligner().align(phones, total_duration=duration)

generate_naive(text: str, duration: float, fps: float = 60.0, epsilon: float = 0.015, g2p: Optional[G2P] = None, mapping=None, params=None, gestures=None, wav: Optional[str] = None, preprocess: Optional[Callable[[str], str]] = None, parse_tags: bool = False) -> FaceTrack

preprocess (issue #7) is an optional callable(text) -> text run on the transcript before anything else, so a registered auto-tagger can insert tags programmatically; injecting a tag this way is identical to hand-writing it. parse_tags enables the transcript text-tag stage (see :mod:openfacefx.texttags): curve tags become extra channels, event tags the event layer, [emphasis] a local articulation boost and <T>/[pause] chunk/silence the timeline. Both default off, so the plain naive path is byte-identical; a parse_tags run on a transcript that carries no tags is byte-identical too.

Source code in src/openfacefx/pipeline.py

def generate_naive(
    text: str,
    duration: float,
    fps: float = 60.0,
    epsilon: float = 0.015,
    g2p: Optional[G2P] = None,
    mapping=None,
    params=None,
    gestures=None,
    wav: Optional[str] = None,
    preprocess: Optional[Callable[[str], str]] = None,
    parse_tags: bool = False,
) -> FaceTrack:
    """``preprocess`` (issue #7) is an optional ``callable(text) -> text`` run on
    the transcript before anything else, so a registered auto-tagger can insert
    tags programmatically; injecting a tag this way is identical to hand-writing
    it. ``parse_tags`` enables the transcript text-tag stage (see
    :mod:`openfacefx.texttags`): curve tags become extra channels, event tags the
    event layer, ``[emphasis]`` a local articulation boost and ``<T>``/``[pause]``
    chunk/silence the timeline. Both default off, so the plain naive path is
    byte-identical; a ``parse_tags`` run on a transcript that carries no tags is
    byte-identical too."""
    if preprocess is not None:
        text = preprocess(text)
    if parse_tags:
        from .texttags import (parse_tagged_transcript, resolve_tagged_segments,
                               curve_channels, tag_events, emphasis_params)
        clean, tags = parse_tagged_transcript(text)
        if not tags:
            segs = naive_segments(clean, duration, g2p=g2p)
            return generate_from_alignment(segs, fps=fps, epsilon=epsilon,
                                           mapping=mapping, params=params,
                                           gestures=gestures, wav=wav)
        segs, spans, windows = resolve_tagged_segments(clean, duration, tags,
                                                       g2p or G2P())
        track = generate_from_alignment(
            segs, fps=fps, epsilon=epsilon, mapping=mapping,
            params=emphasis_params(params, windows), gestures=gestures, wav=wav)
        track.channels.extend(curve_channels(tags, spans, duration))
        track.events.extend(tag_events(tags, spans, duration))
        return track
    segs = naive_segments(text, duration, g2p=g2p)
    return generate_from_alignment(segs, fps=fps, epsilon=epsilon,
                                   mapping=mapping, params=params,
                                   gestures=gestures, wav=wav)

openfacefx.edits

Edit preservation: hand-tweaks that survive regeneration (issue #9).

The pipeline is a pure function (audio, text) -> FaceTrack; re-running it with new dials (intensity, coarticulation, energy gain) throws away any manual curve edits an animator made. This module mirrors FaceFX's two-layer ownership model so it doesn't have to: analysis owns the generated curves, while a user keeps a small, separate record of what they changed. FaceFX offers two ways to edit an "Owned by Analysis" curve -- add an OFFSET curve (keys added to the analysis value, then clamped to the node's min/max), or take full manual ownership (regeneration skips the curve). We store that as a sidecar *.edits.json rather than inline flags, so the .track stays clean, versioned interchange and the solver stays untouched.

Regeneration is then: re-run the pipeline to a fresh :class:FaceTrack, then :func:apply_edits overlays the sidecar back on. The inverse, :func:diff_edits, captures what a user changed (a hand-edited track vs the baseline it came from) into a sidecar.

Two per-channel edit modes, each optionally restricted to a time span (the "locked region"):

• offset -- FaceFX offset curve: keys are deltas added to whatever the solver now produces, then clamped. Being relative, an offset survives dial changes (intensity/coarticulation/energy), which is the primary use case.
• replace -- FaceFX manual ownership: keys are absolute values that replace the generated channel (whole-channel, or just inside span).

Conflict handling is conservative: an edit on a channel the regeneration dropped is preserved and flagged (keep-edit, the default -- a hand-edit is never silently lost) or discarded (take-generated); a locked region always wins over the regenerated content inside its span while the fresh curve shows through everywhere else.

Deterministic and dependency-light: only numpy (interp/clip and the existing :func:openfacefx.curves._rdp thinner), hashlib and json. No RNG, stable 4-dp rounding, identical output across Python 3.9/3.13. Additive: a track generated without a sidecar is byte-identical to previous releases -- the whole layer is opt-in.

FORMAT = 'openfacefx.edits' module-attribute

VERSION = 1 module-attribute

EditsDoc(channels: Dict[str, dict] = dict(), fps: float = 60.0, source_id: Optional[str] = None, base_hash: Optional[str] = None, viseme_set: Optional[List[str]] = None) dataclass

A parsed openfacefx.edits sidecar.

channels maps a channel name to an edit record -- a validated dict {"mode": "offset"|"replace", "keys": [[t, v], ...], "clamp"?: [lo, hi], "span"?: [t0, t1]}. base_hash / source_id are provenance (the baseline the edits were captured from, and the audio id); fps / viseme_set echo the track for reference.

channels: Dict[str, dict] = field(default_factory=dict) class-attribute instance-attribute

fps: float = 60.0 class-attribute instance-attribute

source_id: Optional[str] = None class-attribute instance-attribute

base_hash: Optional[str] = None class-attribute instance-attribute

viseme_set: Optional[List[str]] = None class-attribute instance-attribute

to_dict() -> Dict

Source code in src/openfacefx/edits.py

def to_dict(self) -> Dict:
    d: Dict = {"format": FORMAT, "version": VERSION}
    if self.source_id is not None:
        d["source_id"] = self.source_id
    if self.base_hash is not None:
        d["base_hash"] = self.base_hash
    d["fps"] = self.fps
    if self.viseme_set is not None:
        d["viseme_set"] = list(self.viseme_set)
    d["channels"] = self.channels
    return d

from_dict(d: Dict) -> 'EditsDoc' classmethod

Parse and validate a sidecar dict, raising ValueError with a clear, channel-named message on any malformed field.

Source code in src/openfacefx/edits.py

@classmethod
def from_dict(cls, d: Dict) -> "EditsDoc":
    """Parse and validate a sidecar dict, raising ``ValueError`` with a clear,
    channel-named message on any malformed field."""
    if d.get("format") != FORMAT or d.get("version") != VERSION:
        raise ValueError(
            f"expected format {FORMAT!r} version {VERSION}, got "
            f"{d.get('format')!r} version {d.get('version')!r}")
    fps = d.get("fps", 60.0)
    if not _finite(fps) or fps <= 0.0:
        raise ValueError(f"edits 'fps' must be a positive number, got {fps!r}")
    return cls(
        channels=_validate_channels(d.get("channels")),
        fps=float(fps),
        source_id=d.get("source_id"),
        base_hash=d.get("base_hash"),
        viseme_set=d.get("viseme_set"),
    )

sample(keys, T) -> np.ndarray

Values of a keyframe list at times T by linear interpolation.

Uses np.interp's endpoint-hold (flat before the first key / after the last), matching how a curve editor reads a piecewise-linear channel. An empty key list samples to zeros. keys may be a :class:Channel, a list of :class:Keyframe, or [time, value] pairs (offset/replace records).

Source code in src/openfacefx/edits.py

def sample(keys, T) -> np.ndarray:
    """Values of a keyframe list at times ``T`` by linear interpolation.

    Uses ``np.interp``'s endpoint-hold (flat before the first key / after the
    last), matching how a curve editor reads a piecewise-linear channel. An empty
    key list samples to zeros. ``keys`` may be a :class:`Channel`, a list of
    :class:`Keyframe`, or ``[time, value]`` pairs (offset/replace records)."""
    Ta = np.asarray(T, dtype=float)
    kt, kv = _key_arrays(keys)
    if kt.size == 0:
        return np.zeros(Ta.shape, dtype=float)
    return np.interp(Ta, kt, kv)

save_edits(doc: EditsDoc, path: str) -> None

Write a sidecar as pretty JSON (2-space indent, trailing newline).

Source code in src/openfacefx/edits.py

def save_edits(doc: EditsDoc, path: str) -> None:
    """Write a sidecar as pretty JSON (2-space indent, trailing newline)."""
    with open(path, "w", encoding="utf-8") as fh:
        json.dump(doc.to_dict(), fh, indent=2)
        fh.write("\n")

load_edits(path: str) -> EditsDoc

Read and validate a sidecar from path.

Source code in src/openfacefx/edits.py

def load_edits(path: str) -> EditsDoc:
    """Read and validate a sidecar from ``path``."""
    with open(path, encoding="utf-8") as fh:
        try:
            d = json.load(fh)
        except json.JSONDecodeError as e:
            raise ValueError(f"{path}: not valid JSON ({e})") from None
    return EditsDoc.from_dict(d)

diff_edits(base: FaceTrack, edited: FaceTrack, mode: str = 'offset', span: Optional[Tuple[float, float]] = None, tol: float = 0.01, eps: float = 0.01, clamps: Optional[Dict[str, Tuple[float, float]]] = None, source_id: Optional[str] = None) -> EditsDoc

Build a sidecar from a hand-edited track and the base it was edited from. A channel whose curves differ by more than tol anywhere is recorded; identical channels are skipped (so an untouched track yields an empty sidecar).

mode='offset' (default) stores the delta edited - base (RDP-thinned at eps) plus the per-channel clamp -- the relative form that survives later dial changes. mode='replace' stores the absolute edited values. span=(t0, t1) restricts capture to a time window (a locked region). clamps maps channel -> (lo, hi) (e.g. from mapping.Target); missing entries default to [0, 1].

Source code in src/openfacefx/edits.py

def diff_edits(base: FaceTrack, edited: FaceTrack, mode: str = "offset",
               span: Optional[Tuple[float, float]] = None, tol: float = 0.01,
               eps: float = 0.01, clamps: Optional[Dict[str, Tuple[float, float]]] = None,
               source_id: Optional[str] = None) -> EditsDoc:
    """Build a sidecar from a hand-``edited`` track and the ``base`` it was edited
    from. A channel whose curves differ by more than ``tol`` anywhere is recorded;
    identical channels are skipped (so an untouched track yields an empty sidecar).

    ``mode='offset'`` (default) stores the delta ``edited - base`` (RDP-thinned at
    ``eps``) plus the per-channel ``clamp`` -- the relative form that survives
    later dial changes. ``mode='replace'`` stores the absolute edited values.
    ``span=(t0, t1)`` restricts capture to a time window (a locked region).
    ``clamps`` maps channel -> (lo, hi) (e.g. from ``mapping.Target``); missing
    entries default to ``[0, 1]``."""
    if mode not in _MODES:
        raise ValueError(f"diff mode must be one of {_MODES}, got {mode!r}")
    bmap = {c.name: c for c in base.channels}
    umap = {c.name: c for c in edited.channels}
    channels: Dict[str, dict] = {}
    for name in sorted(set(bmap) | set(umap)):
        B, U = bmap.get(name), umap.get(name)
        clamp = list(clamps[name]) if (clamps and name in clamps) else [0.0, 1.0]
        if U is not None and B is None:                       # user added a channel
            channels[name] = {"mode": "replace",
                              "keys": [[round(k.time, 4), round(k.value, 4)]
                                       for k in U.keys]}
            continue
        if B is not None and U is None:                       # user silenced it
            channels[name] = {"mode": "replace",
                              "keys": [[round(B.keys[0].time, 4), 0.0],
                                       [round(B.keys[-1].time, 4), 0.0]]}
            continue
        times = sorted({k.time for k in B.keys} | {k.time for k in U.keys})
        if span is not None:
            times = [t for t in times if span[0] <= t <= span[1]]
        if not times:
            continue
        Ta = np.asarray(times, dtype=float)
        delta = sample(U, Ta) - sample(B, Ta)
        if float(np.max(np.abs(delta))) <= tol:               # untouched channel
            continue
        channels[name] = _diff_record(mode, U, times, Ta, delta, clamp, span, eps)
    return EditsDoc(
        channels=channels, fps=float(base.fps), source_id=source_id,
        base_hash=_sha1_track(base),
        viseme_set=list(base.target_set) if base.target_set is not None else None)

apply_edits(gen: FaceTrack, edits: EditsDoc, eps: float = 0.015, on_conflict: str = 'keep-edit') -> Tuple[FaceTrack, List[dict]]

Overlay edits onto a freshly generated gen track and return (merged_track, conflicts).

A generated channel with no record passes through untouched. An offset record adds its (interpolated) deltas to the current curve and clamps; a replace record substitutes its values. A span confines the edit to a window -- the fresh curve is preserved exactly outside it, so a locked region wins only where it was drawn.

conflicts lists edits whose channel is absent from gen (renamed, or its word removed on re-alignment). on_conflict='keep-edit' (default) re-injects them so a hand-edit is never silently lost; 'take-generated' drops them for the fresh output. Either way the conflict is reported so the caller can warn.

Source code in src/openfacefx/edits.py

def apply_edits(gen: FaceTrack, edits: EditsDoc, eps: float = 0.015,
                on_conflict: str = "keep-edit") -> Tuple[FaceTrack, List[dict]]:
    """Overlay ``edits`` onto a freshly generated ``gen`` track and return
    ``(merged_track, conflicts)``.

    A generated channel with no record passes through untouched. An ``offset``
    record adds its (interpolated) deltas to the current curve and clamps; a
    ``replace`` record substitutes its values. A ``span`` confines the edit to a
    window -- the fresh curve is preserved exactly outside it, so a locked region
    wins only where it was drawn.

    ``conflicts`` lists edits whose channel is absent from ``gen`` (renamed,
    or its word removed on re-alignment). ``on_conflict='keep-edit'`` (default)
    re-injects them so a hand-edit is never silently lost;
    ``'take-generated'`` drops them for the fresh output. Either way the conflict
    is reported so the caller can warn."""
    if on_conflict not in _ON_CONFLICT:
        raise ValueError(
            f"on_conflict must be one of {_ON_CONFLICT}, got {on_conflict!r}")
    gen_names = {c.name for c in gen.channels}
    out: List[Channel] = []
    conflicts: List[dict] = []
    for c in gen.channels:
        rec = edits.channels.get(c.name)
        out.append(_apply_record(c, rec, eps) if rec is not None else c)
    target_set = list(gen.target_set) if gen.target_set is not None else None
    for name in sorted(edits.channels):                       # edits on dropped channels
        if name in gen_names:
            continue
        action = ("dropped (take-generated)" if on_conflict == "take-generated"
                  else "preserved (keep-edit)")
        conflicts.append({
            "channel": name, "reason": "absent-from-regen",
            "detail": f"channel not in the regenerated track; edit {action} "
                      f"-- verify it was not renamed or its word removed"})
        if on_conflict == "take-generated":
            continue
        ch = _apply_record(Channel(name, []), edits.channels[name], eps)
        if ch.keys:
            out.append(ch)
            if target_set is None:            # None == the built-in viseme vocab;
                target_set = list(VISEMES)    # make it explicit before extending
            if name not in target_set:
                target_set.append(name)
    merged = FaceTrack(gen.fps, out, target_set)
    # Curve edits leave the fresh generation's event/take layer (issue #6) intact,
    # so apply order relative to the event layer does not matter.
    merged.events = list(getattr(gen, "events", None) or [])
    merged.variants = getattr(gen, "variants", None)
    return merged, conflicts

openfacefx.alignment

Forced alignment: producing time-stamped phonemes.

This is the model-heavy stage of the pipeline. Rather than reinvent a speech recogniser, OpenFaceFX defines a small PhonemeSegment contract and provides:

• NaiveAligner -- no acoustic model. Given the phonemes for an utterance and the utterance duration (or word timings), it distributes phonemes over time using per-phoneme duration priors. Good enough to see the pipeline end to end; not accurate lip-sync.

• load_mfa_textgrid -- parse the output of the Montreal Forced Aligner (the recommended production aligner). MFA gives real acoustic alignment.

Built-in adapters for the open-source aligners live in :mod:openfacefx.aligners (Whisper / WhisperX word timings, Gentle word and phone timings, issue #54); any other source just needs to produce a list of PhonemeSegment and the rest of the pipeline is unchanged.

PhonemeSegment(phoneme: str, start: float, end: float, confidence: Optional[float] = None) dataclass

phoneme: str instance-attribute

start: float instance-attribute

end: float instance-attribute

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

dur: float property

NaiveAligner

Distribute phonemes across a time span using duration priors.

align(phonemes: List[str], total_duration: float, start: float = 0.0) -> List[PhonemeSegment]

Source code in src/openfacefx/alignment.py

def align(
    self,
    phonemes: List[str],
    total_duration: float,
    start: float = 0.0,
) -> List[PhonemeSegment]:
    if not phonemes:
        return []
    weights = [_prior(p) for p in phonemes]
    wsum = sum(weights) or 1.0
    segs: List[PhonemeSegment] = []
    t = start
    for p, w in zip(phonemes, weights):
        d = total_duration * (w / wsum)
        segs.append(PhonemeSegment(p, t, t + d))
        t += d
    return segs

align_words(words: List[tuple]) -> List[PhonemeSegment]

When you already have word-level timings (common from ASR), align phonemes within each word span. Much better than utterance-level.

Source code in src/openfacefx/alignment.py

def align_words(
    self,
    words: List[tuple],  # (phoneme_list, word_start, word_end)
) -> List[PhonemeSegment]:
    """When you already have word-level timings (common from ASR), align
    phonemes within each word span. Much better than utterance-level."""
    segs: List[PhonemeSegment] = []
    for phones, ws, we in words:
        segs.extend(self.align(phones, we - ws, start=ws))
    return segs

dump_segments(segments: List['PhonemeSegment']) -> List[dict]

Serialise phoneme segments to the plain-JSON shape the HTML previewer consumes (tools/build_preview.py --segments): a list of {"phoneme", "start", "end"[, "confidence"]}. confidence is emitted only when the aligner supplied one.

Source code in src/openfacefx/alignment.py

def dump_segments(segments: List["PhonemeSegment"]) -> List[dict]:
    """Serialise phoneme segments to the plain-JSON shape the HTML previewer
    consumes (``tools/build_preview.py --segments``): a list of
    ``{"phoneme", "start", "end"[, "confidence"]}``. ``confidence`` is emitted
    only when the aligner supplied one."""
    out = []
    for s in segments:
        d = {"phoneme": s.phoneme, "start": round(s.start, 4),
             "end": round(s.end, 4)}
        if s.confidence is not None:
            d["confidence"] = round(s.confidence, 4)
        out.append(d)
    return out

load_mfa_textgrid(path: str, tier: str = 'phones') -> List[PhonemeSegment]

Parse a Praat TextGrid produced by the Montreal Forced Aligner.

Only the interval tier named tier is read. Empty / silence intervals become the sil phoneme so the mouth relaxes between words.

Source code in src/openfacefx/alignment.py

def load_mfa_textgrid(path: str, tier: str = "phones") -> List[PhonemeSegment]:
    """Parse a Praat TextGrid produced by the Montreal Forced Aligner.

    Only the interval tier named ``tier`` is read. Empty / silence intervals
    become the ``sil`` phoneme so the mouth relaxes between words.
    """
    text = open(path, "r", encoding="utf-8").read()
    # Find the requested interval tier block.
    tier_pat = re.compile(
        r'name = "%s".*?intervals: size = \d+(.*?)(?:item \[\d+\]:|$)' % re.escape(tier),
        re.DOTALL,
    )
    m = tier_pat.search(text)
    if not m:
        raise ValueError(f"tier {tier!r} not found in TextGrid")
    block = m.group(1)
    seg_pat = re.compile(
        r"xmin = ([\d.]+)\s*xmax = ([\d.]+)\s*text = \"([^\"]*)\"",
        re.DOTALL,
    )
    segs: List[PhonemeSegment] = []
    for xmin, xmax, label in seg_pat.findall(block):
        label = label.strip()
        ph = label if label else SILENCE
        segs.append(PhonemeSegment(ph, float(xmin), float(xmax)))
    return segs

openfacefx.g2p

Grapheme-to-phoneme (word -> ARPAbet phonemes).

Priority order

1. A pronunciation dictionary (CMUdict format) if one is loaded.
2. A small built-in dictionary so the demo runs with no downloads.
3. A crude rule-based fallback for out-of-vocabulary words.

For production accuracy, load the full CMU Pronouncing Dictionary via G2P.load_cmudict(path) or plug in a neural G2P model. The fallback exists only so nothing crashes on unknown words.

G2P(pronouncer=None, tokenizer=None)

Source code in src/openfacefx/g2p.py

def __init__(self, pronouncer=None, tokenizer=None) -> None:
    self._dict: Dict[str, List[str]] = dict(_BUILTIN)
    # i18n hooks (issue #8), both opt-in — with both ``None`` (the default)
    # the English tokenizer + lookup-then-rules path is byte-identical.
    # ``pronouncer(word, prev, next) -> phonemes | None`` is consulted
    # between dictionary lookup and the rule fallback; ``tokenizer(text) ->
    # words`` replaces the default ``[A-Za-z']+`` split.
    self.pronouncer = pronouncer
    self.tokenizer = tokenizer

pronouncer = pronouncer instance-attribute

tokenizer = tokenizer instance-attribute

load_cmudict(path: str) -> int

Load a CMUdict-format file. Returns the number of entries added.

Format per line: WORD P1 P2 P3 ... (alt pronunciations as WORD(2))

Source code in src/openfacefx/g2p.py

def load_cmudict(self, path: str) -> int:
    """Load a CMUdict-format file. Returns the number of entries added.

    Format per line:  WORD  P1 P2 P3 ...   (alt pronunciations as WORD(2))
    """
    added = 0
    with open(path, "r", encoding="latin-1") as fh:
        for line in fh:
            if line.startswith(";;;") or not line.strip():
                continue
            parts = line.split()
            word = re.sub(r"\(\d+\)$", "", parts[0]).lower()
            if word not in self._dict:  # keep first (primary) pronunciation
                self._dict[word] = parts[1:]
                added += 1
    return added

word(w: str) -> List[str]

Source code in src/openfacefx/g2p.py

def word(self, w: str) -> List[str]:
    key = re.sub(r"[^a-z']", "", w.lower())
    if not key:
        return []
    if key in self._dict:
        return list(self._dict[key])
    return self._fallback(key)

tokenize(text: str) -> List[str]

Split text into word tokens. The English default is [A-Za-z']+ (which drops non-Latin script); set tokenizer to a per-language callable so other scripts survive (issue #8).

Source code in src/openfacefx/g2p.py

def tokenize(self, text: str) -> List[str]:
    """Split ``text`` into word tokens. The English default is
    ``[A-Za-z']+`` (which drops non-Latin script); set ``tokenizer`` to a
    per-language callable so other scripts survive (issue #8)."""
    if self.tokenizer is not None:
        return list(self.tokenizer(text))
    return re.findall(r"[A-Za-z']+", text)

phrase(text: str) -> List[str]

Source code in src/openfacefx/g2p.py

def phrase(self, text: str) -> List[str]:
    words = self.tokenize(text)
    out: List[str] = []
    for i, w in enumerate(words):
        if self.pronouncer is None:          # unchanged English resolution
            out.extend(self.word(w))
        else:                                # lookup -> pronouncer -> rules
            prev = words[i - 1] if i > 0 else None
            nxt = words[i + 1] if i + 1 < len(words) else None
            out.extend(self._resolve(w, prev, nxt))
    return out

load_dictionary(path: str) -> int

Load a locale dictionary (issue #8): a file declaring its locale and phoneme alphabet (arpabet / ipa / sampa), each entry mapped into the internal inventory via the alias tables. Returns the number of entries added (the first spelling of a word wins, as with load_cmudict).

Source code in src/openfacefx/g2p.py

def load_dictionary(self, path: str) -> int:
    """Load a locale dictionary (issue #8): a file declaring its ``locale``
    and phoneme ``alphabet`` (arpabet / ipa / sampa), each entry mapped into
    the internal inventory via the alias tables. Returns the number of
    entries added (the first spelling of a word wins, as with ``load_cmudict``)."""
    from .pronounce import read_dictionary
    added = 0
    for word, phones in read_dictionary(path).entries.items():
        if word not in self._dict:
            self._dict[word] = phones
            added += 1
    return added

oov_words(text: str) -> List[str]

Words in text that would fall through to the rule fallback — the ones worth adding to a pronunciation dictionary (QA reporting).

Source code in src/openfacefx/g2p.py

def oov_words(self, text: str) -> List[str]:
    """Words in ``text`` that would fall through to the rule fallback —
    the ones worth adding to a pronunciation dictionary (QA reporting)."""
    oov = []
    for w in re.findall(r"[A-Za-z']+", text):
        key = re.sub(r"[^a-z']", "", w.lower())
        if key and key not in self._dict and key not in oov:
            oov.append(key)
    return oov

openfacefx.phonemes

ARPAbet phoneme set.

We use the CMU/ARPAbet inventory because it is what the CMU Pronouncing Dictionary, Montreal Forced Aligner, and most English G2P tools emit. Stress digits (0/1/2) on vowels are stripped before mapping to visemes.

ARPABET_VOWELS = {'AA', 'AE', 'AH', 'AO', 'AW', 'AY', 'EH', 'ER', 'EY', 'IH', 'IY', 'OW', 'OY', 'UH', 'UW'} module-attribute

ARPABET_CONSONANTS = {'B', 'CH', 'D', 'DH', 'F', 'G', 'HH', 'JH', 'K', 'L', 'M', 'N', 'NG', 'P', 'R', 'S', 'SH', 'T', 'TH', 'V', 'W', 'Y', 'Z', 'ZH'} module-attribute

ARPABET = ARPABET_VOWELS | ARPABET_CONSONANTS module-attribute

SILENCE = 'sil' module-attribute

IPA_ALIASES = {'AA': 'ɑ', 'AE': 'æ', 'AH': 'ʌ', 'AO': 'ɔ', 'AW': 'aʊ', 'AY': 'aɪ', 'EH': 'ɛ', 'ER': 'ɝ', 'EY': 'eɪ', 'IH': 'ɪ', 'IY': 'i', 'OW': 'oʊ', 'OY': 'ɔɪ', 'UH': 'ʊ', 'UW': 'u', 'B': 'b', 'CH': 'tʃ', 'D': 'd', 'DH': 'ð', 'F': 'f', 'G': 'ɡ', 'HH': 'h', 'JH': 'dʒ', 'K': 'k', 'L': 'l', 'M': 'm', 'N': 'n', 'NG': 'ŋ', 'P': 'p', 'R': 'ɹ', 'S': 's', 'SH': 'ʃ', 'T': 't', 'TH': 'θ', 'V': 'v', 'W': 'w', 'Y': 'j', 'Z': 'z', 'ZH': 'ʒ'} module-attribute

SAMPA_ALIASES = {'AA': 'A', 'AE': '{', 'AH': 'V', 'AO': 'O', 'AW': 'aU', 'AY': 'aI', 'EH': 'E', 'ER': '3`', 'EY': 'eI', 'IH': 'I', 'IY': 'i', 'OW': 'oU', 'OY': 'OI', 'UH': 'U', 'UW': 'u', 'B': 'b', 'CH': 'tS', 'D': 'd', 'DH': 'D', 'F': 'f', 'G': 'g', 'HH': 'h', 'JH': 'dZ', 'K': 'k', 'L': 'l', 'M': 'm', 'N': 'n', 'NG': 'N', 'P': 'p', 'R': 'r\\', 'S': 's', 'SH': 'S', 'T': 't', 'TH': 'T', 'V': 'v', 'W': 'w', 'Y': 'j', 'Z': 'z', 'ZH': 'Z'} module-attribute

ALPHABETS = ('arpabet', 'ipa', 'sampa') module-attribute

strip_stress(phoneme: str) -> str

Remove the trailing stress digit ARPAbet attaches to vowels (e.g. AH0).

Source code in src/openfacefx/phonemes.py

def strip_stress(phoneme: str) -> str:
    """Remove the trailing stress digit ARPAbet attaches to vowels (e.g. AH0)."""
    if phoneme and phoneme[-1].isdigit():
        return phoneme[:-1]
    return phoneme

is_vowel(phoneme: str) -> bool

Source code in src/openfacefx/phonemes.py

def is_vowel(phoneme: str) -> bool:
    return strip_stress(phoneme) in ARPABET_VOWELS

to_ipa(phoneme: str) -> str

Internal phoneme -> IPA (stress digit dropped); unknown symbols pass through unchanged so a mixed transcript never crashes.

Source code in src/openfacefx/phonemes.py

def to_ipa(phoneme: str) -> str:
    """Internal phoneme -> IPA (stress digit dropped); unknown symbols pass
    through unchanged so a mixed transcript never crashes."""
    return IPA_ALIASES.get(strip_stress(phoneme), phoneme)

to_sampa(phoneme: str) -> str

Internal phoneme -> X-SAMPA (stress digit dropped); unknown pass through.

Source code in src/openfacefx/phonemes.py

def to_sampa(phoneme: str) -> str:
    """Internal phoneme -> X-SAMPA (stress digit dropped); unknown pass through."""
    return SAMPA_ALIASES.get(strip_stress(phoneme), phoneme)

from_ipa(symbol: str) -> str

IPA -> internal ARPAbet; an unrecognised symbol passes through (and will fall to sil at the viseme stage, the documented behaviour).

Source code in src/openfacefx/phonemes.py

def from_ipa(symbol: str) -> str:
    """IPA -> internal ARPAbet; an unrecognised symbol passes through (and will
    fall to ``sil`` at the viseme stage, the documented behaviour)."""
    return _IPA_TO_INTERNAL.get(symbol, symbol)

from_sampa(symbol: str) -> str

X-SAMPA -> internal ARPAbet; an unrecognised symbol passes through.

Source code in src/openfacefx/phonemes.py

def from_sampa(symbol: str) -> str:
    """X-SAMPA -> internal ARPAbet; an unrecognised symbol passes through."""
    return _SAMPA_TO_INTERNAL.get(symbol, symbol)

from_alphabet(symbol: str, alphabet: str) -> str

Map a phoneme symbol written in alphabet (arpabet | ipa | sampa) into the internal inventory. arpabet is the identity.

Source code in src/openfacefx/phonemes.py

def from_alphabet(symbol: str, alphabet: str) -> str:
    """Map a phoneme ``symbol`` written in ``alphabet`` (``arpabet`` | ``ipa`` |
    ``sampa``) into the internal inventory. ``arpabet`` is the identity."""
    if alphabet == "ipa":
        return from_ipa(symbol)
    if alphabet == "sampa":
        return from_sampa(symbol)
    if alphabet == "arpabet":
        return symbol
    raise ValueError(f"unknown phoneme alphabet {alphabet!r} (use {ALPHABETS})")

openfacefx.energy

Audio-energy lip-sync fallback for untranscripted speech.

When there is no transcript and no aligner output, you can still drive a believable flapping mouth straight from the loudness of the audio. This is the single most common non-ML lip-sync mechanism in the wild — SALSA remaps RMS amplitude through low/high cutoffs, Moho drives openness from instantaneous loudness, Live2D bakes RMS volume into a mouth-open parameter. This module is OpenFaceFX's version of that idea.

PCM samples -> per-frame RMS -> noise-gated, robustly-normalized envelope
            -> asymmetric attack/release smoothing -> mouth-open curves

This is an energy fallback, not lip-sync. It knows nothing about phonemes or visemes: it cannot tell a /m/ from an /aa/, and it will happily open the mouth on a cough. It drives one primary jaw-open channel plus a small, purely aesthetic spread into two secondary mouth shapes so the result does not read as a single channel flapping on and off. Use the transcript-based pipeline (generate_naive / generate_from_alignment) whenever you have text; use this only when you have nothing but a WAV.

Only numpy and the stdlib wave module are used. Input must be 16-bit PCM WAV (mono or stereo — stereo is downmixed); convert other codecs first, e.g. ffmpeg -i in.mp3 -c:a pcm_s16le -ar 16000 out.wav.

energy_envelope(wav_path: str, fps: float = 60.0, window: Optional[float] = None, gate: float = 0.06, smoothing: Tuple[float, float] = (0.012, 0.09)) -> Tuple[np.ndarray, np.ndarray]

Loudness envelope of a WAV, sampled at fps, as (times, envelope).

envelope is in [0, 1]: per-frame RMS, noise-gated and normalized against a high percentile (not the max — see _REF_PERCENTILE), then run through an asymmetric attack/release follower.

Parameters

window : RMS analysis window in seconds (default 2 / fps — a mild overlap between frames). Larger = smoother, blurrier. gate : noise-floor gate as a fraction of the reference level in [0, 1). Frames quieter than gate * reference read as full silence; the remaining range is stretched back to [0, 1] (a SALSA-style low/high cutoff remap (v - lo) / (hi - lo)). smoothing : (attack_seconds, release_seconds) for the envelope follower. The default opens fast and closes ~7x slower.

A clip whose reference level is below _SILENCE_REF is all silence and returns an all-zero envelope.

Source code in src/openfacefx/energy.py

def energy_envelope(
    wav_path: str,
    fps: float = 60.0,
    window: Optional[float] = None,
    gate: float = 0.06,
    smoothing: Tuple[float, float] = (0.012, 0.09),
) -> Tuple[np.ndarray, np.ndarray]:
    """Loudness envelope of a WAV, sampled at ``fps``, as ``(times, envelope)``.

    ``envelope`` is in ``[0, 1]``: per-frame RMS, noise-gated and normalized
    against a high percentile (not the max — see ``_REF_PERCENTILE``), then run
    through an asymmetric attack/release follower.

    Parameters
    ----------
    window : RMS analysis window in seconds (default ``2 / fps`` — a mild
        overlap between frames). Larger = smoother, blurrier.
    gate : noise-floor gate as a fraction of the reference level in ``[0, 1)``.
        Frames quieter than ``gate * reference`` read as full silence; the
        remaining range is stretched back to ``[0, 1]`` (a SALSA-style
        low/high cutoff remap ``(v - lo) / (hi - lo)``).
    smoothing : ``(attack_seconds, release_seconds)`` for the envelope
        follower. The default opens fast and closes ~7x slower.

    A clip whose reference level is below ``_SILENCE_REF`` is all silence and
    returns an all-zero envelope.
    """
    if not (0.0 <= gate < 1.0):
        raise ValueError(f"gate must be in [0, 1), got {gate}")
    signal, rate = _read_wav_mono(wav_path)
    win = window if window is not None else 2.0 / fps
    rms = _frame_rms(signal, rate, fps, win)
    times = np.arange(len(rms)) / fps

    ref = float(np.percentile(rms, _REF_PERCENTILE)) if len(rms) else 0.0
    if ref < _SILENCE_REF:
        return times, np.zeros_like(rms)

    # Cutoff remap: floor (noise gate) -> 0, reference percentile -> 1.
    floor = gate * ref
    env = (rms - floor) / max(ref - floor, 1e-9)
    np.clip(env, 0.0, 1.0, out=env)

    attack, release = smoothing
    env = _attack_release(env, fps, attack, release)
    return times, env

generate_from_energy(wav_path: str, fps: float = 60.0, epsilon: float = 0.015, mapping: Optional[Mapping] = None, intensity: float = 1.0, spread: float = 0.25, window: Optional[float] = None, gate: float = 0.06, smoothing: Tuple[float, float] = (0.012, 0.09), gestures=None, smooth: float = 0.0, lag: float = 0.0) -> FaceTrack

Build a FaceTrack from audio loudness alone — no transcript needed.

Energy fallback, not viseme detection. The loudness envelope (:func:energy_envelope) drives one primary jaw-open channel (aa); a small spread fraction of that opening is bled into two secondary mouth shapes purely so the motion does not look like a single channel flapping. The spread rule is a deliberate aesthetic heuristic, not phoneme recognition: louder frames lean rounded/open (O), quieter frames lean mid-spread (E). Silent frames go to rest (sil). No claim is made that any channel matches the sound actually being spoken.

Each frame partitions unit weight across sil and the mouth shapes (sil + aa + E + O == 1), so sil is a true "mouth closed" amount. Output is an ordinary Oculus-viseme track, so --retarget, .anim and the cue exporters all compose downstream exactly as for the other modes.

Parameters

intensity : gain on the mouth opening (1.0 as-is; >1 opens wider on quiet speech, <1 is subtler). Clamped into [0, 1] per frame. spread : fraction of the opening lent to the secondary E/O shapes (0 = pure jaw flap; default 0.25). mapping : optional target vocabulary supplying channel names and per-target clamps (like the other generators). Must contain aa and sil; defaults to the built-in Oculus-15 set. epsilon, window, gate, smoothing : forwarded to keyframe reduction and :func:energy_envelope. gestures : opt-in non-verbal gesture layer (issue #5); a GestureParams (or True for defaults) appends blink/brow/head/eye channels driven by this same envelope. None (default) leaves output byte-identical. smooth, lag : opt-in post-solve curve conditioning (issue #10). smooth is a temporal Gaussian sigma in seconds run over the mouth partition before keyframe reduction (the partition aa + E + O + sil == 1 survives it); lag slides the reduced keyframes in time (seconds; >0 lags, <0 leads the audio, clamped into the clip). Both 0.0 (default) are a byte-identical no-op. No closures exist here, so nothing is re-sealed.

Output is deterministic: identical audio and parameters give an identical track (the mouth path has no RNG; the gesture layer is seeded from GestureParams.seed).

Source code in src/openfacefx/energy.py

def generate_from_energy(
    wav_path: str,
    fps: float = 60.0,
    epsilon: float = 0.015,
    mapping: Optional[Mapping] = None,
    intensity: float = 1.0,
    spread: float = 0.25,
    window: Optional[float] = None,
    gate: float = 0.06,
    smoothing: Tuple[float, float] = (0.012, 0.09),
    gestures=None,
    smooth: float = 0.0,
    lag: float = 0.0,
) -> FaceTrack:
    """Build a ``FaceTrack`` from audio loudness alone — no transcript needed.

    **Energy fallback, not viseme detection.** The loudness envelope
    (:func:`energy_envelope`) drives one primary jaw-open channel (``aa``); a
    small ``spread`` fraction of that opening is bled into two secondary mouth
    shapes purely so the motion does not look like a single channel flapping.
    The spread rule is a deliberate *aesthetic* heuristic, **not** phoneme
    recognition: louder frames lean rounded/open (``O``), quieter frames lean
    mid-spread (``E``). Silent frames go to rest (``sil``). No claim is made
    that any channel matches the sound actually being spoken.

    Each frame partitions unit weight across ``sil`` and the mouth shapes
    (``sil + aa + E + O == 1``), so ``sil`` is a true "mouth closed" amount.
    Output is an ordinary Oculus-viseme track, so ``--retarget``, ``.anim`` and
    the cue exporters all compose downstream exactly as for the other modes.

    Parameters
    ----------
    intensity : gain on the mouth opening (``1.0`` as-is; ``>1`` opens wider on
        quiet speech, ``<1`` is subtler). Clamped into ``[0, 1]`` per frame.
    spread : fraction of the opening lent to the secondary ``E``/``O`` shapes
        (``0`` = pure jaw flap; default ``0.25``).
    mapping : optional target vocabulary supplying channel names and per-target
        clamps (like the other generators). Must contain ``aa`` and ``sil``;
        defaults to the built-in Oculus-15 set.
    epsilon, window, gate, smoothing : forwarded to keyframe reduction and
        :func:`energy_envelope`.
    gestures : opt-in non-verbal gesture layer (issue #5); a ``GestureParams``
        (or ``True`` for defaults) appends blink/brow/head/eye channels driven
        by this same envelope. ``None`` (default) leaves output byte-identical.
    smooth, lag : opt-in post-solve curve conditioning (issue #10). ``smooth`` is
        a temporal Gaussian sigma in seconds run over the mouth partition before
        keyframe reduction (the partition ``aa + E + O + sil == 1`` survives it);
        ``lag`` slides the reduced keyframes in time (seconds; ``>0`` lags, ``<0``
        leads the audio, clamped into the clip). Both ``0.0`` (default) are a
        byte-identical no-op. No closures exist here, so nothing is re-sealed.

    Output is deterministic: identical audio and parameters give an identical
    track (the mouth path has no RNG; the gesture layer is seeded from
    ``GestureParams.seed``).
    """
    times, env = energy_envelope(wav_path, fps=fps, window=window, gate=gate,
                                 smoothing=smoothing)
    openness = np.clip(env * intensity, 0.0, 1.0)
    spread = float(np.clip(spread, 0.0, 1.0))

    names = mapping.target_names if mapping is not None else list(VISEMES)
    targets = mapping.targets if mapping is not None else None
    index = {n: i for i, n in enumerate(names)}
    if "aa" not in index or "sil" not in index:
        raise ValueError(
            "energy mode drives the built-in viseme roles (aa, E, O, sil); "
            "the given mapping has no 'aa'/'sil' target to drive")

    # Partition unit weight: jaw-open keeps (1 - spread) of the opening; the
    # spread budget splits between mid (E, quiet) and round (O, loud); the rest
    # is silence. aa + E + O == openness, so sil + aa + E + O == 1.
    channels = {
        "aa": openness * (1.0 - spread),
        "E": openness * spread * (1.0 - openness),
        "O": openness * spread * openness,
        "sil": 1.0 - openness,
    }
    matrix = np.zeros((len(times), len(names)))
    for name, col in channels.items():
        if name in index:
            matrix[:, index[name]] = col
    if smooth > 0.0:
        from .postprocess import smooth_matrix
        matrix = smooth_matrix(matrix, smooth, fps)
    track = reduce_to_track(times, matrix, fps=fps, epsilon=epsilon,
                            targets=targets)
    if lag:
        from .postprocess import time_shift
        time_shift(track, lag)
    if gestures:
        # No phonemes here, so stress/pauses/peaks come from the envelope itself.
        from .gestures import GestureParams, add_gestures_to_track
        gp = gestures if isinstance(gestures, GestureParams) else GestureParams()
        duration = float(times[-1]) if len(times) else 0.0
        add_gestures_to_track(track, duration, times, env, None, gp)
    return track

openfacefx.prosody

Prosody extraction: pitch, loudness and speaking-rate -> typed events.

Where :mod:openfacefx.energy follows how loud the voice is, this module also follows how high it is: a short-time autocorrelation pitch tracker recovers a per-frame fundamental frequency (F0) and a voiced/unvoiced flag, and — reusing energy.py's WAV reader and RMS follower for loudness — turns the two tracks into the prosodic events an animator wants: emphasis (pitch and loudness spike together), phrase_boundary (a silent pause or the utterance end) and question_rise (a rising terminal F0, the yes/no-question cue), plus a global speaking rate (a syllable-ish proxy).

PCM samples -> framed short-time autocorrelation -> F0 + voicing + clarity
            -> robust pitch/energy z-scores -> emphasis / boundary / question
               events (:class:`openfacefx.events.Event`)

This is DSP heuristics, not an ML prosody model. The tracker is the standard non-neural pipeline (windowed autocorrelation debiased by the window's own autocorrelation, à la Boersma/Praat; parabolic peak interpolation; an octave-cost period pick), so expect it to behave accordingly:

• Accuracy / voicing. On clean speech F0 lands within a few percent on voiced frames but makes occasional octave errors and mislabels voicing on whispered/breathy/creaky voice or low SNR (the voicing gate is clarity+energy, not a trained VAD); an ML tracker (CREPE/pYIN) is cleaner. Fine here because the events need only relative pitch movement, not calibrated Hz — an octave-shifted F0 still lands the emphasis and question-rise correctly.
• Prominence / questions are rule-based cue layers, not ToBI labelling: emphasis keys on coincident pitch+energy peaks (~75-80 % vs ~85 % human agreement), and question detection keys purely on terminal F0 rise, so it misses falling wh-questions and can false-fire on list/uptalk intonation.
• Will misbehave on music/singing, background noise, overlapping speakers and heavy reverb. Input must be 16-bit PCM WAV (same as energy.py); convert first, e.g. ffmpeg -i in.mp3 -c:a pcm_s16le out.wav.

Events are ordinary :class:openfacefx.events.Event records (issue #6), so they attach to a :class:~openfacefx.curves.FaceTrack and serialise through the JSON / Unity .anim / Unreal-notify exporters like an authored layer. numpy + stdlib wave only; no RNG, so identical audio gives identical events across runs, platforms and Python versions.

Source = Union[str, np.ndarray] module-attribute

ProsodyParams(fmin: float = 80.0, fmax: float = 400.0, voicing_threshold: float = 0.45, silence_frac: float = 0.03, octave_cost: float = 0.01, emph_thresh: float = 1.0, min_emph: float = 0.05, emph_merge: float = 0.12, min_pause: float = 0.18, clause_max: float = 0.45, q_look: float = 0.3, q_min_voiced: int = 4, q_rise: float = 2.0, rate_smooth: float = 0.05, rate_min_spacing: float = 0.12) dataclass

Every threshold the tracker and event derivation use, in one dataclass so the defaults live in one place and callers can retune without touching the algorithm. Defaults follow Praat's raw-autocorrelation pitch settings and the prosodic-prominence literature (see the module docstring).

fmin: float = 80.0 class-attribute instance-attribute

fmax: float = 400.0 class-attribute instance-attribute

voicing_threshold: float = 0.45 class-attribute instance-attribute

silence_frac: float = 0.03 class-attribute instance-attribute

octave_cost: float = 0.01 class-attribute instance-attribute

emph_thresh: float = 1.0 class-attribute instance-attribute

min_emph: float = 0.05 class-attribute instance-attribute

emph_merge: float = 0.12 class-attribute instance-attribute

min_pause: float = 0.18 class-attribute instance-attribute

clause_max: float = 0.45 class-attribute instance-attribute

q_look: float = 0.3 class-attribute instance-attribute

q_min_voiced: int = 4 class-attribute instance-attribute

q_rise: float = 2.0 class-attribute instance-attribute

rate_smooth: float = 0.05 class-attribute instance-attribute

rate_min_spacing: float = 0.12 class-attribute instance-attribute

ProsodyTrack(fps: float, times: np.ndarray, f0: np.ndarray, voiced: np.ndarray, rms: np.ndarray, clarity: np.ndarray, speaking_rate: float = 0.0) dataclass

Per-frame prosody bundle at the analysis frame rate fps. f0 is Hz on voiced frames and nan on unvoiced ones; voiced is the boolean gate; rms is the loudness follower (same units as energy._frame_rms); clarity is the 0..1 autocorrelation periodicity; speaking_rate is a global syllable-per-(voiced-)second estimate.

fps: float instance-attribute

times: np.ndarray instance-attribute

f0: np.ndarray instance-attribute

voiced: np.ndarray instance-attribute

rms: np.ndarray instance-attribute

clarity: np.ndarray instance-attribute

speaking_rate: float = 0.0 class-attribute instance-attribute

pitch_track(source: Source, rate: Optional[int] = None, fps: float = 100.0, fmin: float = 80.0, fmax: float = 400.0, params: Optional[ProsodyParams] = None) -> Tuple[np.ndarray, np.ndarray, np.ndarray]

Per-frame fundamental frequency of speech, as (times, f0, voiced).

source is a 16-bit PCM WAV path or a float sample array in [-1, 1] (then rate is required). fps is the analysis frame rate (hop) in Hz — 100 by default, i.e. a 10 ms step; pitch accuracy is set by the window (which holds ~2.5 of the lowest period), not by fps. f0 is Hz on voiced frames and np.nan on unvoiced ones; voiced is the boolean gate. See the module docstring for accuracy caveats — this is a heuristic tracker, not an ML pitch model.

Source code in src/openfacefx/prosody.py

def pitch_track(
    source: Source,
    rate: Optional[int] = None,
    fps: float = 100.0,
    fmin: float = 80.0,
    fmax: float = 400.0,
    params: Optional[ProsodyParams] = None,
) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Per-frame fundamental frequency of speech, as ``(times, f0, voiced)``.

    ``source`` is a 16-bit PCM WAV path *or* a float sample array in ``[-1, 1]``
    (then ``rate`` is required). ``fps`` is the analysis frame rate (hop) in Hz —
    100 by default, i.e. a 10 ms step; pitch accuracy is set by the window (which
    holds ~2.5 of the lowest period), not by ``fps``. ``f0`` is Hz on voiced
    frames and ``np.nan`` on unvoiced ones; ``voiced`` is the boolean gate. See
    the module docstring for accuracy caveats — this is a heuristic tracker, not
    an ML pitch model.
    """
    p = params or ProsodyParams()
    fmin = float(fmin if fmin is not None else p.fmin)
    fmax = float(fmax if fmax is not None else p.fmax)
    if fmin <= 0 or fmax <= fmin:
        raise ValueError(f"need 0 < fmin < fmax, got fmin={fmin}, fmax={fmax}")
    if isinstance(source, str):
        signal, rate = _read_wav_mono(source)
    else:
        if rate is None:
            raise ValueError("pitch_track needs the sample rate when given samples")
        signal = np.asarray(source, dtype=np.float64)
    times, f0, voiced, _clar, _rms = _analyze(signal, int(rate), fps,
                                              replace(p, fmin=fmin, fmax=fmax))
    return times, f0, voiced

prosody_features(wav_path: str, fps: float = 100.0, params: Optional[ProsodyParams] = None) -> ProsodyTrack

Bundle a WAV's pitch, loudness, voicing/clarity and speaking rate into a :class:ProsodyTrack at the fps analysis rate. Reuses :func:openfacefx.energy._frame_rms for the loudness track, so prosody and the energy fallback measure loudness the same way.

Source code in src/openfacefx/prosody.py

def prosody_features(
    wav_path: str,
    fps: float = 100.0,
    params: Optional[ProsodyParams] = None,
) -> ProsodyTrack:
    """Bundle a WAV's pitch, loudness, voicing/clarity and speaking rate into a
    :class:`ProsodyTrack` at the ``fps`` analysis rate. Reuses
    :func:`openfacefx.energy._frame_rms` for the loudness track, so prosody and
    the energy fallback measure loudness the same way."""
    p = params or ProsodyParams()
    signal, rate = _read_wav_mono(wav_path)
    times, f0, voiced, clarity, rms = _analyze(signal, rate, fps, p)
    rate_hz = _speaking_rate(times, rms, voiced, fps, p)
    return ProsodyTrack(fps=fps, times=times, f0=f0, voiced=voiced, rms=rms,
                        clarity=clarity, speaking_rate=rate_hz)

detect_events(track: ProsodyTrack, params: Optional[ProsodyParams] = None, segments=None) -> List[Event]

Derive the typed prosodic events from a :class:ProsodyTrack: emphasis (pitch+loudness beats), phrase_boundary (pauses + utterance end) and question_rise (rising terminal F0). Returns a time-sorted list[Event].

segments (optional phoneme segments) currently only sharpen the utterance end used for the trailing phrase boundary; pitch/energy drive everything else. All events are plain :class:openfacefx.events.Event, so they attach and serialise exactly like an authored event layer.

Source code in src/openfacefx/prosody.py

def detect_events(track: ProsodyTrack, params: Optional[ProsodyParams] = None,
                  segments=None) -> List[Event]:
    """Derive the typed prosodic events from a :class:`ProsodyTrack`: emphasis
    (pitch+loudness beats), phrase_boundary (pauses + utterance end) and
    question_rise (rising terminal F0). Returns a time-sorted ``list[Event]``.

    ``segments`` (optional phoneme segments) currently only sharpen the utterance
    end used for the trailing phrase boundary; pitch/energy drive everything else.
    All events are plain :class:`openfacefx.events.Event`, so they attach and
    serialise exactly like an authored event layer."""
    p = params or ProsodyParams()
    if len(track.times) == 0:
        return []
    end_t = float(segments[-1].end) if segments else float(track.times[-1])
    emphasis = _emphasis_events(track, p)
    spans = _boundary_spans(track, p)
    boundaries = _phrase_events(spans, end_t, p)
    questions = _question_events(track, boundaries, p)
    events = emphasis + boundaries + questions
    events.sort(key=lambda e: e.t)
    return events

prosody_events(wav_path: str, fps: float = 100.0, segments=None, params: Optional[ProsodyParams] = None) -> List[Event]

One call from a WAV to typed prosodic events: extract the :class:ProsodyTrack then :func:detect_events. fps is the analysis rate (event times are absolute seconds, independent of any render fps). Reuses energy.py's reader, so the same 16-bit-PCM-WAV constraint applies.

Source code in src/openfacefx/prosody.py

def prosody_events(
    wav_path: str,
    fps: float = 100.0,
    segments=None,
    params: Optional[ProsodyParams] = None,
) -> List[Event]:
    """One call from a WAV to typed prosodic events: extract the
    :class:`ProsodyTrack` then :func:`detect_events`. ``fps`` is the analysis rate
    (event times are absolute seconds, independent of any render fps). Reuses
    ``energy.py``'s reader, so the same 16-bit-PCM-WAV constraint applies."""
    p = params or ProsodyParams()
    track = prosody_features(wav_path, fps=fps, params=p)
    return detect_events(track, p, segments=segments)