Skip to content

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, and the prosody tracker that turns pitch and loudness into typed events.

openfacefx.pipeline

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

Two entry points:

  • generate_from_alignment -- you already have time-stamped phonemes (from MFA, Gentle, wav2vec2, Whisper...). 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 
25
26
27
28
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 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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)
    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 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
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 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
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) -> FaceTrack

Source code in src/openfacefx/pipeline.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
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,
) -> FaceTrack:
    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.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.

You can add adapters for Gentle, wav2vec2, or Whisper the same way: 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 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
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 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
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 
39
40
41
42
43
44
45
46
47
48
49
50
51
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 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
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()

Source code in src/openfacefx/g2p.py 
56
57
def __init__(self) -> None:
    self._dict: Dict[str, List[str]] = dict(_BUILTIN)

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 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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 
76
77
78
79
80
81
82
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)

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

Source code in src/openfacefx/g2p.py 
84
85
86
87
88
def phrase(self, text: str) -> List[str]:
    out: List[str] = []
    for w in re.findall(r"[A-Za-z']+", text):
        out.extend(self.word(w))
    return out

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 
90
91
92
93
94
95
96
97
98
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

strip_stress(phoneme: str) -> str

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

Source code in src/openfacefx/phonemes.py 
28
29
30
31
32
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 
35
36
def is_vowel(phoneme: str) -> bool:
    return strip_stress(phoneme) in ARPABET_VOWELS

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 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
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) -> 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.

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 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
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,
) -> 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.

    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
    track = reduce_to_track(times, matrix, fps=fps, epsilon=epsilon,
                            targets=targets)
    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 
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
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 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
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 
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
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 
513
514
515
516
517
518
519
520
521
522
523
524
525
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)