Streaming / real-time generation

The offline pipeline solves a whole clip at once. A live pipeline — a TTS engine emitting phonemes as it speaks — needs to emit animation incrementally, with memory that stays constant no matter how long the stream runs. StreamingGenerator (issue #43) does that: push(chunk_of_segments) returns the keyframe frames that just became final, flush() (alias close()) emits the tail. It reuses the exact offline component math — coarticulation._blend, the shared core of build_viseme_curves — over a bounded segment window.

from openfacefx import StreamingGenerator, frames_to_track

gen = StreamingGenerator(fps=60.0, look_ahead=0.5)   # look_ahead = latency dial
frames = []
for chunk in phoneme_segment_chunks:                 # e.g. from a live aligner
    frames += gen.push(chunk)
frames += gen.flush()
track = frames_to_track(frames, 60.0)                # -> a normal FaceTrack

Honesty: reproduces the offline solve within tolerance, not bit-exactly

This is important and stated plainly. The coarticulation dominance is a Laplacian bump D_i(t) = alpha·exp(-theta·|t − c_i|) — exponential, infinite support — and the blend normalizes over every segment. Bounded memory (pruning old segments) and a finite look-ahead (dropping far-future ones) therefore both omit exponentially small tails, so no finite window is bit-identical to generate_from_alignment. That is fundamental to this dominance model, not an implementation gap.

It converges fast, though. The per-frame error from a window W seconds wide is bounded by O(exp(-theta·W)) (slowest θ ≈ 2.9/s for a long vowel):

look-ahead W	~ max per-frame error	≈
1.5 s	1e-2	the RDP epsilon (0.015)
3 s	1e-4	the 4-dp keyframe grid
4.5 s	1e-6	10⁴× below perceptual / storage precision

look_ahead is the single latency ↔ fidelity dial: 0 = zero latency, no anticipatory coarticulation (causal only); larger = more anticipation, tighter to offline. There is one exact case: when the window covers the whole clip (look_ahead and back_span ≥ clip duration) the per-frame blend is bit-identical to offline, and frames_to_track then reduces to the same keyframes as generate_from_alignment.

Guarantees

• Chunk boundaries never matter. The same clip pushed in 1 or K chunks yields bit-identical frames — windows are selected by frame time, not by arrival.
• Bounded memory. The cooked-segment ring buffer stays O(look_ahead + back_span), independent of stream length.
• Causal. A frame, once emitted, is never revised; its value depends only on inputs within its window, so a later chunk cannot alter an already-emitted keyframe (the flip side of the tolerance). The optional causal_smooth is a past-only one-pole filter, deliberately distinct from the offline symmetric postprocess.smooth_matrix (which reads both directions and so is not reproducible causally).
• Not offline RDP. Frames are emitted incrementally (dense), so the generator does not reproduce offline's global Ramer–Douglas–Peucker keyframe reduction (that needs the whole column = O(stream)); frames_to_track applies the same RDP once at the end for storage.

Network transport (gRPC / WebSocket) is out of scope — this is an in-process generator only. Deterministic; numpy + stdlib.

openfacefx.streaming

Streaming / real-time generation (issue #43): a long-lived coarticulation generator that carries state across pushed chunks in bounded memory.

The offline pipeline solves a whole clip at once. A live pipeline (a TTS engine emitting phonemes as it speaks) needs to emit animation incrementally, with constant memory regardless of stream length. :class:StreamingGenerator does that: push(chunk_of_segments) returns the keyframe frames that just became final, flush() emits the tail. It reuses the exact offline component math — :func:openfacefx.coarticulation._blend (the shared core of build_viseme_curves) — over a bounded segment window, so the streamed frames reproduce the offline solve.

Honesty — reproduces the offline solve WITHIN TOLERANCE, not bit-exactly. The coarticulation dominance is a Laplacian bump D_i(t) = alpha·exp(-theta·|t − c_i|) (coarticulation.py), exponential/infinite support, and the blend normalizes over every segment. Bounded memory (pruning old segments) and a finite look-ahead (dropping far-future ones) therefore both omit exponentially small tails, so no finite window is bit-identical to generate_from_alignment — that is fundamental to this dominance model, not an implementation gap. It converges fast, though: the per-frame error from a window W seconds wide is bounded by O(exp(-theta·W)) (slowest θ≈2.9/s for a long vowel), so W≈1.5 s → ~1e‑2 (≈ the RDP epsilon), W≈3 s → ~1e‑4 (≈ the 4‑dp keyframe grid), W≈4.5 s → ~1e‑6. look_ahead is the single latency↔fidelity dial: 0 = zero latency / no anticipatory coarticulation (causal only); larger = more anticipation, tighter to offline. There is one exact case: when the window covers the whole clip (look_ahead and back_span ≥ clip duration) the per-frame blend is bit-identical to offline. Chunk boundaries never matter — the same clip pushed in 1 or K chunks yields bit-identical frames (windows are selected by frame time, not by arrival).

Causal only: a frame, once emitted, is never revised, and its value depends only on inputs within its window — a later chunk cannot alter an already-emitted keyframe (that immutability is the flip side of the tolerance). The optional causal_smooth is a past-only one-pole filter, distinct from the offline symmetric :func:openfacefx.postprocess.smooth_matrix (which reads both ways and so is not reproducible causally). Network transport is out of scope — this is an in-process generator.

Frame = Tuple[float, np.ndarray] module-attribute

DEFAULT_LOOK_AHEAD = 0.5 module-attribute

DEFAULT_BACK_SPAN = 2.0 module-attribute

StreamingGenerator(fps: float = 60.0, mapping=None, params: Optional[CoartParams] = None, *, look_ahead: float = DEFAULT_LOOK_AHEAD, back_span: float = DEFAULT_BACK_SPAN, causal_smooth: float = 0.0)

Incremental coarticulation over pushed chunks with O(window) memory.

push(segments) cooks the new :class:~openfacefx.alignment.PhonemeSegment chunk (the same silence-absorb + diphthong-split _preprocess does, applied causally) and returns the frames now covered by the look-ahead; flush() (alias close()) emits the tail. look_ahead / back_span are the future / past window extents in seconds; causal_smooth is a past-only one-pole time constant (0 = off). Reuse :func:frames_to_track to assemble the emitted frames into a :class:~openfacefx.curves.FaceTrack.

Source code in src/openfacefx/streaming.py

def __init__(self, fps: float = 60.0, mapping=None,
             params: Optional[CoartParams] = None, *,
             look_ahead: float = DEFAULT_LOOK_AHEAD,
             back_span: float = DEFAULT_BACK_SPAN,
             causal_smooth: float = 0.0) -> None:
    self.fps = float(fps)
    self.mapping = mapping
    self.params = params or CoartParams()
    # streaming does its own (causal) conditioning: never the offline
    # symmetric smoother, so strip it from the blend params.
    self._bp = replace(self.params, smooth=0.0)
    self.look_ahead = max(float(look_ahead), 0.0)
    self.back_span = max(float(back_span), 0.0)
    self.causal_smooth = max(float(causal_smooth), 0.0)
    self.target_names = (list(mapping.target_names) if mapping is not None
                         else list(VISEMES))
    self.n_targets = len(self.target_names)
    # cooking state (mirrors coarticulation._preprocess, incrementally)
    self._raw_tail: Optional[PhonemeSegment] = None
    self._raw_idx = -1
    self._next_idx = 0
    self._pending: Optional[PhonemeSegment] = None    # open cooked segment
    self._cooked: List[PhonemeSegment] = []           # bounded ring buffer
    self._cooked_end = 0.0
    # grid + emission
    self._started = False
    self._t0 = 0.0
    self._emit_idx = 0
    self._sm_prev: Optional[np.ndarray] = None        # causal filter state
    self._closed = False

fps = float(fps) instance-attribute

mapping = mapping instance-attribute

params = params or CoartParams() instance-attribute

look_ahead = max(float(look_ahead), 0.0) instance-attribute

back_span = max(float(back_span), 0.0) instance-attribute

causal_smooth = max(float(causal_smooth), 0.0) instance-attribute

target_names = list(mapping.target_names) if mapping is not None else list(VISEMES) instance-attribute

n_targets = len(self.target_names) instance-attribute

close = flush class-attribute instance-attribute

buffered_segments: int property

Cooked segments currently retained — O(window), for the memory test.

push(segments) -> List[Frame]

Feed a chunk of segments; return the frames finalized by it.

Source code in src/openfacefx/streaming.py

def push(self, segments) -> List[Frame]:
    """Feed a chunk of segments; return the frames finalized by it."""
    if self._closed:
        raise RuntimeError("StreamingGenerator: push() after flush()/close()")
    out: List[Frame] = []
    for s in segments:
        if not self._started:
            self._t0 = self._grid_origin(s)
            self._started = True
        if self._raw_tail is not None:          # its successor is known: not
            self._cook(self._raw_tail, self._raw_idx, is_last=False)  # last
            out += self._emit_ready()
        self._raw_tail, self._raw_idx = s, self._next_idx
        self._next_idx += 1
    return out

flush() -> List[Frame]

Cook the held-back last segment, emit every remaining frame.

Source code in src/openfacefx/streaming.py

def flush(self) -> List[Frame]:
    """Cook the held-back last segment, emit every remaining frame."""
    out: List[Frame] = []
    if self._raw_tail is not None:
        self._cook(self._raw_tail, self._raw_idx, is_last=True)
        self._raw_tail = None
    self._finalize_pending()
    out += self._emit_ready(final=True)
    self._closed = True
    return out

frames_to_track(frames: List[Frame], fps: float, mapping=None, epsilon: float = 0.015) -> FaceTrack

Assemble emitted (time, values) frames into a :class:~openfacefx.curves.FaceTrack via the same reduce_to_track the offline pipeline uses — so a full-window stream reduces to the same keyframes as generate_from_alignment.

Source code in src/openfacefx/streaming.py

def frames_to_track(frames: List[Frame], fps: float, mapping=None,
                    epsilon: float = 0.015) -> FaceTrack:
    """Assemble emitted ``(time, values)`` frames into a
    :class:`~openfacefx.curves.FaceTrack` via the same ``reduce_to_track`` the
    offline pipeline uses — so a full-window stream reduces to the same keyframes
    as ``generate_from_alignment``."""
    if not frames:
        target_set = None if mapping is None else list(mapping.target_names)
        return FaceTrack(fps=fps, channels=[], target_set=target_set)
    times = np.array([t for t, _ in frames])
    matrix = np.vstack([v for _, v in frames])
    targets = mapping.targets if mapping is not None else None
    return reduce_to_track(times, matrix, fps, epsilon, targets)