Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 50 additions & 4 deletions src/voxkit/engines/mfa_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,12 @@
--------
Stored at ``~/.voxkit/MFAENGINE/{tool}/settings.json``:

- **align**: dictionary, file_type
- **train**: epochs, use_gpu
- **align**: dictionary, file_type, conda_path
- **train**: dictionary, num_iterations, use_gpu, conda_path

``conda_path`` lets users (mainly on Windows) point VoxKit at a specific
``conda``/``conda.exe`` when it is not discoverable on PATH. Leaving it blank
keeps the existing auto-detection behavior.

Notes
-----
Expand All @@ -32,6 +36,29 @@
from voxkit.storage import alignments, datasets, models

from .base import AlignmentEngine
from .constants import AVAILABLE_TOOLS


def _conda_path_field() -> FieldConfig:
"""Build the shared 'Conda Path' settings field used by every MFA tool.

The field is optional: a blank value preserves the existing auto-detection
in ``voxkit.services.mfa._find_conda``. It exists primarily so Windows users
whose conda install is not on PATH can point VoxKit straight at their
``conda.exe``.
"""
return FieldConfig(
name="conda_path",
label="Conda Path",
field_type=FieldType.LINEEDIT,
default_value="",
placeholder="auto-detect",
tooltip=(
"Full path to the conda executable. Leave blank to auto-detect. "
"Mainly useful on Windows when conda.exe is not on PATH "
r"(e.g. C:\Users\me\miniconda3\Scripts\conda.exe)."
),
)


class MFAEngine(AlignmentEngine):
Expand All @@ -44,7 +71,7 @@ def __init__(self, id: str | None = None):
settings_configurations={
"align": SettingsConfig(
title="MFA Aligner Settings",
dimensions=(400, 350),
dimensions=(460, 380),
apply_blur=True,
fields=[
FieldConfig(
Expand All @@ -61,12 +88,13 @@ def __init__(self, id: str | None = None):
default_value="wav",
tooltip="Specify the audio file type (e.g., wav, flac).",
),
_conda_path_field(),
],
store_file="MFAENGINE/align/settings.json",
),
"train": SettingsConfig(
title="MFA Trainer Settings",
dimensions=(400, 350),
dimensions=(460, 440),
apply_blur=True,
fields=[
FieldConfig(
Expand All @@ -92,6 +120,7 @@ def __init__(self, id: str | None = None):
default_value=False,
tooltip="Enable GPU acceleration for faster training.",
),
_conda_path_field(),
],
store_file="MFAENGINE/train/settings.json",
),
Expand Down Expand Up @@ -147,6 +176,7 @@ def align(self, dataset_id: str, model_id: str) -> None:
corpus_dir=str(corpus_path),
model_path=str(model_path),
output_dir=str(alignment_output_path),
conda_path=self._configured_conda_path("align"),
)
alignments.update_alignment(
dataset_id=dataset_id,
Expand Down Expand Up @@ -234,10 +264,26 @@ def train_aligner(
corpus_dir=str(audio_root),
base_model_path=str(base_model_path),
output_model_path=str(new_model_path),
conda_path=self._configured_conda_path("train"),
)
except Exception as e:
raise RuntimeError(f"MFA model training failed: {e}")

def _configured_conda_path(self, tool_type: AVAILABLE_TOOLS) -> str | None:
"""Return the user-configured conda path for a tool, or None if unset.

Reads the persisted settings for ``tool_type`` and normalizes a blank
or whitespace-only value to ``None`` so the service layer falls back to
auto-detection.
"""
try:
value = self.get_settings(tool_type).get("conda_path")
except Exception:
return None
if isinstance(value, str) and value.strip():
return value.strip()
return None

def _validate_align_settings(self, settings: dict) -> bool:
return True # Implement validation logic for align settings here

Expand Down
66 changes: 52 additions & 14 deletions src/voxkit/services/mfa.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,8 +12,33 @@ def _no_window() -> dict:
return {}


def _find_conda() -> str:
"""Return the conda executable path, checking common install locations on Windows."""
def _find_conda(conda_path: str | None = None) -> str:
"""Return the conda executable path.

Resolution order:
1. ``conda_path`` argument, when it points to an existing file. This is
the path a user configures in the MFA engine settings, primarily so
Windows users with a non-standard Anaconda/Miniconda install can
point VoxKit straight at their ``conda.exe``.
2. The ``VOXKIT_CONDA_PATH`` environment variable, when it points to an
existing file.
3. ``conda`` on the system PATH.
4. Common install locations on Windows.

Args:
conda_path: Optional user-configured path to the conda executable.

Raises:
FileNotFoundError: If conda cannot be located through any of the above.
"""
# User-configured path (settings dialog) or environment override take
# precedence over auto-detection so a deliberate choice always wins.
for candidate in (conda_path, os.environ.get("VOXKIT_CONDA_PATH")):
if candidate:
resolved = Path(candidate).expanduser()
if resolved.exists():
return str(resolved)

# Fast path: conda is already on PATH
if shutil.which("conda"):
return "conda"
Expand Down Expand Up @@ -44,20 +69,25 @@ def _find_conda() -> str:
raise FileNotFoundError(
"conda not found. Install Miniconda from https://docs.conda.io/en/latest/miniconda.html "
"and create the aligner environment with: "
"conda create -n aligner -c conda-forge montreal-forced-aligner"
"conda create -n aligner -c conda-forge montreal-forced-aligner. "
"If conda is installed but not on PATH (common on Windows), set its full path in the "
"MFA engine settings ('Conda Path') or via the VOXKIT_CONDA_PATH environment variable."
)


def ensure_dictionary_downloaded(dictionary_name: str = "english_us_arpa") -> None:
def ensure_dictionary_downloaded(
dictionary_name: str = "english_us_arpa", conda_path: str | None = None
) -> None:
"""Ensure the specified MFA dictionary is downloaded.

Args:
dictionary_name: Name of the dictionary to download (default: "english_us_arpa").
conda_path: Optional user-configured path to the conda executable.

Raises:
AssertionError: If dictionary download fails and dictionary is not available.
"""
conda = _find_conda()
conda = _find_conda(conda_path)
download_cmd = [
conda,
"run",
Expand Down Expand Up @@ -97,7 +127,7 @@ def ensure_dictionary_downloaded(dictionary_name: str = "english_us_arpa") -> No
print(f"[mfa] Dictionary '{dictionary_name}' is ready.")


def _ensure_mfa_server_running() -> None:
def _ensure_mfa_server_running(conda_path: str | None = None) -> None:
"""Start MFA's bundled Postgres server on Windows. No-op elsewhere.

Why: MFA 3.3.x's SQLite backend has a multiprocessing race in
Expand All @@ -111,7 +141,7 @@ def _ensure_mfa_server_running() -> None:
if sys.platform != "win32":
return

conda = _find_conda()
conda = _find_conda(conda_path)
# Both calls are idempotent: `init` errors if the server dir already exists,
# `start` errors if it's already running. Either error state is the goal,
# so we ignore returncodes and only guard against true failures (timeout,
Expand All @@ -131,7 +161,12 @@ def _ensure_mfa_server_running() -> None:


def run_mfa_align(
corpus_dir, model_path, output_dir, dictionary_name="english_us_arpa", eval_dir=None
corpus_dir,
model_path,
output_dir,
dictionary_name="english_us_arpa",
eval_dir=None,
conda_path=None,
) -> None:
"""
Run MFA align command with the provided arguments.
Expand All @@ -142,16 +177,17 @@ def run_mfa_align(
output_dir: Path to output TextGrids.
dictionary_name: MFA dictionary name (default: "english_us_arpa").
eval_dir: Optional path to reference alignments for evaluation.
conda_path: Optional user-configured path to the conda executable.

Raises:
AssertionError: If dictionary is not available.
subprocess.CalledProcessError: If MFA alignment fails.
"""
# Ensure dictionary is downloaded
ensure_dictionary_downloaded(dictionary_name)
_ensure_mfa_server_running()
ensure_dictionary_downloaded(dictionary_name, conda_path=conda_path)
_ensure_mfa_server_running(conda_path=conda_path)

conda = _find_conda()
conda = _find_conda(conda_path)
cmd = [
conda,
"run",
Expand Down Expand Up @@ -186,6 +222,7 @@ def run_mfa_adapt(
output_model_path,
dictionary_name="english_us_arpa",
num_iterations=1,
conda_path=None,
) -> None:
"""
Run MFA adapt command with the provided arguments.
Expand All @@ -196,16 +233,17 @@ def run_mfa_adapt(
output_model_path (str): Path where the adapted model will be saved.
dictionary_name (str): Name of the dictionary to use (default: "english_us_arpa").
num_iterations (int): Number of adaptation iterations.
conda_path (str | None): Optional user-configured path to the conda executable.

Raises:
AssertionError: If dictionary is not available.
subprocess.CalledProcessError: If MFA adaptation fails.
"""
# Ensure dictionary is downloaded
ensure_dictionary_downloaded(dictionary_name)
_ensure_mfa_server_running()
ensure_dictionary_downloaded(dictionary_name, conda_path=conda_path)
_ensure_mfa_server_running(conda_path=conda_path)

conda = _find_conda()
conda = _find_conda(conda_path)
cmd = [
conda,
"run",
Expand Down
Empty file added tests/services/__init__.py
Empty file.
56 changes: 56 additions & 0 deletions tests/services/test_mfa.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,56 @@
"""Tests for conda executable resolution in voxkit.services.mfa.

Only the pure ``_find_conda`` resolution logic is covered here; the subprocess
wrappers shell out to MFA and conda and are exercised by integration runs.
"""

from voxkit.services import mfa


def test_find_conda_prefers_explicit_path(tmp_path, monkeypatch):
"""An existing explicit conda_path wins over PATH and the env var."""
conda = tmp_path / "conda.exe"
conda.write_text("")
# Make sure the auto-detect fast path would otherwise succeed.
monkeypatch.setattr(mfa.shutil, "which", lambda _: "conda")
monkeypatch.delenv("VOXKIT_CONDA_PATH", raising=False)

assert mfa._find_conda(str(conda)) == str(conda)


def test_find_conda_expands_user_home(tmp_path, monkeypatch):
"""A ``~``-prefixed configured path is expanded before the existence check."""
conda = tmp_path / "conda.exe"
conda.write_text("")
monkeypatch.setenv("HOME", str(tmp_path))
monkeypatch.setenv("USERPROFILE", str(tmp_path)) # Windows home for expanduser
monkeypatch.setattr(mfa.shutil, "which", lambda _: None)
monkeypatch.delenv("VOXKIT_CONDA_PATH", raising=False)

assert mfa._find_conda("~/conda.exe") == str(conda)


def test_find_conda_uses_env_var(tmp_path, monkeypatch):
"""VOXKIT_CONDA_PATH is honored when no explicit path is given."""
conda = tmp_path / "conda.exe"
conda.write_text("")
monkeypatch.setattr(mfa.shutil, "which", lambda _: None)
monkeypatch.setenv("VOXKIT_CONDA_PATH", str(conda))

assert mfa._find_conda() == str(conda)


def test_find_conda_ignores_nonexistent_configured_path(monkeypatch):
"""A configured path that does not exist falls back to PATH detection."""
monkeypatch.setattr(mfa.shutil, "which", lambda _: "conda")
monkeypatch.delenv("VOXKIT_CONDA_PATH", raising=False)

assert mfa._find_conda("/no/such/conda") == "conda"


def test_find_conda_falls_back_to_path(monkeypatch):
"""With no override, conda on PATH is returned."""
monkeypatch.setattr(mfa.shutil, "which", lambda _: "conda")
monkeypatch.delenv("VOXKIT_CONDA_PATH", raising=False)

assert mfa._find_conda() == "conda"
Loading