Add David AI redelivered MFA pipeline from draco#2186
Conversation
Sync current pipeline (RAM-by-session stages, glued-OOV heuristic detector, lexicon build, cluster submission scripts, ffmpeg timeout hardening) from the draco working copy.
timofeeva16tanya
requested review from
sarahyurick
and removed request for
a team
Greptile SummaryThis PR adds a new David AI redelivered MFA alignment tutorial pipeline, porting the current working copy from a private cluster branch into the public repository. It introduces the full RAM-by-session processing flow — text normalization, 16 kHz Opus encoding, Montreal Forced Aligner invocation, Lhotse cutset construction, and per-session RTTM/mixed-audio generation — along with cluster submission scripts, a lexicon build tool, and supporting documentation.
Confidence Score: 2/5The Two bugs in
Important Files Changed
Flowchart%%{init: {'theme': 'neutral'}}%%
flowchart TD
A[Source audio + transcript] --> B[stage0: Normalize text in RAM]
B --> C[Lexicon build\npreprocess_build_lexicon.py]
C --> D[16 kHz Opus encode per speaker\naudio_16k/]
D --> E{MFA align needed?}
E -- yes --> F[stage2: mfa align per segment\nstage2_mfa_align_textgrids.py]
E -- no --> G[Load existing TextGrids]
F --> H[Session + recording TextGrids\ntextgrids/]
G --> H
H --> I[Build Lhotse cuts\ndavid_ai_ram_lhotse.py]
H --> J[Build session RTTM\nbuild_session_rttm_lines_from_words]
I --> K[Per-session cuts\nlhotse/sessions/]
J --> L[Prepare per-speaker pause noise audio]
L --> M[Mix speakers to session Opus\naudio_mixed/]
M --> N[Mark session done\n.done/sessions/]
N --> O{All sessions done?}
O -- yes --> P[stage_ram_merge_lhotse:\nMerge global Lhotse manifests]
O -- no --> Q[Resubmit shard]
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
flowchart TD
A[Source audio + transcript] --> B[stage0: Normalize text in RAM]
B --> C[Lexicon build\npreprocess_build_lexicon.py]
C --> D[16 kHz Opus encode per speaker\naudio_16k/]
D --> E{MFA align needed?}
E -- yes --> F[stage2: mfa align per segment\nstage2_mfa_align_textgrids.py]
E -- no --> G[Load existing TextGrids]
F --> H[Session + recording TextGrids\ntextgrids/]
G --> H
H --> I[Build Lhotse cuts\ndavid_ai_ram_lhotse.py]
H --> J[Build session RTTM\nbuild_session_rttm_lines_from_words]
I --> K[Per-session cuts\nlhotse/sessions/]
J --> L[Prepare per-speaker pause noise audio]
L --> M[Mix speakers to session Opus\naudio_mixed/]
M --> N[Mark session done\n.done/sessions/]
N --> O{All sessions done?}
O -- yes --> P[stage_ram_merge_lhotse:\nMerge global Lhotse manifests]
O -- no --> Q[Resubmit shard]
Reviews (2): Last reviewed commit: "scripts" | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| start = stage_names.index(stage_name) | ||
| except ValueError: | ||
| return | ||
| for name in stage_names[stage:]: |
There was a problem hiding this comment.
NameError on every call: the local variable is start but the slice uses the undefined name stage. Any caller of clear_stage_done_from will crash with NameError: name 'stage' is not defined at runtime.
| for name in stage_names[stage:]: | |
| for name in stage_names[start:]: |
| for session_dir in sessions | ||
| ] |
There was a problem hiding this comment.
sessions (all sessions) instead of todo_sessions (only the ones that need processing). The todo_sessions filter was correctly computed and logged, but is never used here. Every already-finished session is submitted to the process pool, where process_session_ram parses manifests and re-checks outputs before returning skipped=True — a significant waste at scale (e.g. a 200-node run re-processes every completed session on restart).
| for session_dir in sessions | |
| ] | |
| for session_dir in todo_sessions | |
| ] |
| def recording_textgrid_paths(textgrid_dir: Path, recording_id: str) -> list[Path]: | ||
| ordinary = recording_textgrid_path(textgrid_dir, recording_id, variant="ordinary") | ||
| if ordinary.is_file(): | ||
| return [ordinary] | ||
| fb_path = recording_textgrid_path(textgrid_dir, recording_id, variant="fb") | ||
| if fb_path.is_file(): | ||
| return [ordinary, fb_path] | ||
| return [ordinary] |
There was a problem hiding this comment.
ordinary does not exist as a file but fb_path does, the function returns [ordinary, fb_path] — a list that includes the non-existent ordinary path. build_recording_rttm_lines iterates that list and calls parse_textgrid_words(ordinary), which raises an exception that is silently swallowed; only the fb path is parsed. The intent appears to be returning just the fb path (or both when both exist), but the current code returns a stale, missing path.
| def recording_textgrid_paths(textgrid_dir: Path, recording_id: str) -> list[Path]: | |
| ordinary = recording_textgrid_path(textgrid_dir, recording_id, variant="ordinary") | |
| if ordinary.is_file(): | |
| return [ordinary] | |
| fb_path = recording_textgrid_path(textgrid_dir, recording_id, variant="fb") | |
| if fb_path.is_file(): | |
| return [ordinary, fb_path] | |
| return [ordinary] | |
| def recording_textgrid_paths(textgrid_dir: Path, recording_id: str) -> list[Path]: | |
| ordinary = recording_textgrid_path(textgrid_dir, recording_id, variant="ordinary") | |
| fb_path = recording_textgrid_path(textgrid_dir, recording_id, variant="fb") | |
| if ordinary.is_file() and fb_path.is_file(): | |
| return [ordinary, fb_path] | |
| if ordinary.is_file(): | |
| return [ordinary] | |
| if fb_path.is_file(): | |
| return [fb_path] | |
| return [ordinary] |
| from pathlib import Path | ||
| from typing import TypeVar | ||
|
|
||
| MFA_ROOT_DIR_DEFAULT = "/home/ttimofeeva/MFA_models" |
There was a problem hiding this comment.
MFA_ROOT_DIR will get a FileNotFoundError when resolving models, with a confusing path pointing to ttimofeeva's home. A generic fallback like ~/MFA_models is more appropriate.
| MFA_ROOT_DIR_DEFAULT = "/home/ttimofeeva/MFA_models" | |
| MFA_ROOT_DIR_DEFAULT = "~/MFA_models" |
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
| try: | ||
| result = subprocess.run(align_cmd, capture_output=True, text=True, env=mfa_env) | ||
| except OSError as exc: | ||
| logger.error("%s: mfa align failed to start: %s", recording_id, exc) | ||
| mfa_failed_globally = True | ||
| detail = str(exc) |
There was a problem hiding this comment.
mfa align subprocess
subprocess.run(align_cmd, …) has no timeout= argument. The codebase already documents this exact failure mode: FFMPEG_TIMEOUT_S = 600 exists precisely because "a wedged ffmpeg (e.g. an internal futex deadlock seen when many run in a worker pool) blocks its caller forever and hangs the whole shard." MFA runs as another pooled subprocess with the same risk — if mfa align hangs (e.g. SQLite or pynini lock contention), the entire process pool worker blocks indefinitely with no recovery path. The outer try/except Exception in align_recording would catch a subprocess.TimeoutExpired and return RecordingAlignResult(ok=False), so adding a timeout here is safe and consistent.
| try: | ||
| start = stage_names.index(stage_name) | ||
| except ValueError: | ||
| return | ||
| for name in stage_names[stage:]: |
There was a problem hiding this comment.
NameError in clear_stage_done_from: stage vs start
The loop slices on stage, but the variable assigned by list.index() is named start. Any caller will crash with NameError: name 'stage' is not defined at runtime.
| try: | |
| start = stage_names.index(stage_name) | |
| except ValueError: | |
| return | |
| for name in stage_names[stage:]: | |
| try: | |
| start = stage_names.index(stage_name) | |
| except ValueError: | |
| return | |
| for name in stage_names[start:]: |
1 participant
Sync current pipeline (RAM-by-session stages, glued-OOV heuristic detector, lexicon build, cluster submission scripts, ffmpeg timeout hardening) from the draco working copy.
Description
Usage
# Add snippet demonstrating usageChecklist