diff --git a/CHANGELOG.md b/CHANGELOG.md index c06e8be..44f1736 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,10 +22,20 @@ after its first published distribution. ### Changed +- Audio read failures now raise the typed `AudioIntegrityError` and + `AudioDecodeError` instead of a bare `OSError`, and preserve the underlying + cause. Both subclass `OSError`, so existing handling keeps working. +- Preparing CREMA-D now requires Git and Git LFS up front and fails with an + actionable message when either is missing. - README Python API guidance now directs workflow users to `ser.api`. ### Fixed +- Unmaterialized Git LFS pointer files are now reported as a dataset integrity + problem naming the repair steps, instead of surfacing as an opaque audio + decode failure during feature extraction. +- CREMA-D audio is validated before manifest registration, so an incomplete + checkout fails during dataset preparation rather than mid-training. - Import lint path collection now works on shells without `mapfile`, such as the Bash 3.2 that ships with macOS. diff --git a/ser/_internal/data/adapters/crema_d.py b/ser/_internal/data/adapters/crema_d.py index c1f426e..4135bc6 100644 --- a/ser/_internal/data/adapters/crema_d.py +++ b/ser/_internal/data/adapters/crema_d.py @@ -3,9 +3,12 @@ from __future__ import annotations import os +import time from collections.abc import Mapping from pathlib import Path +import soundfile as sf + from ser._internal.data.manifest import Utterance from ser._internal.data.manifest_jsonl import write_manifest_jsonl from ser._internal.data.ontology import LabelOntology, remap_label @@ -17,6 +20,127 @@ CREMA_D_DATASET_POLICY_ID = "share_alike" CREMA_D_DATASET_LICENSE_ID = "odbl-1.0" CREMA_D_SOURCE_URL = "https://github.com/CheyneyComputerScience/CREMA-D" +_GIT_LFS_POINTER_PREFIX = b"version https://git-lfs.github.com/spec/v1" +_MIN_WAV_FILE_BYTES = 44 + + +class CremaDDatasetIntegrityError(RuntimeError): + """Raised when a CREMA-D tree contains missing or invalid audio.""" + + +def _audio_integrity_error(audio_path: Path) -> str | None: + """Returns an integrity diagnostic for one CREMA-D audio file.""" + try: + if not audio_path.is_file(): + return "not a regular file" + file_size = audio_path.stat().st_size + with audio_path.open("rb") as audio_file: + prefix = audio_file.read(len(_GIT_LFS_POINTER_PREFIX)) + if prefix == _GIT_LFS_POINTER_PREFIX: + return "unmaterialized Git LFS pointer" + if file_size == 0: + return "empty file" + if file_size < _MIN_WAV_FILE_BYTES: + return f"implausibly small file ({file_size} bytes)" + metadata = sf.info(str(audio_path)) + if metadata.samplerate <= 0: + return "invalid sample rate" + if metadata.channels <= 0: + return "invalid channel count" + if metadata.frames <= 0: + return "audio contains no frames" + except Exception as err: + detail = str(err).strip() or type(err).__name__ + return f"audio metadata decode failed ({detail})" + return None + + +def validate_crema_d_audio_files( + *, + dataset_root: Path, + dataset_glob_pattern: str, +) -> tuple[Path, ...]: + """Validates every candidate CREMA-D WAV before manifest registration. + + Args: + dataset_root: Root of the CREMA-D checkout. + dataset_glob_pattern: Glob used to discover candidate WAV files. + + Returns: + Validated audio paths in deterministic order. + + Raises: + CremaDDatasetIntegrityError: If audio is missing, unmaterialized, or invalid. + """ + started_at = time.perf_counter() + files = tuple(sorted(dataset_root.glob(dataset_glob_pattern))) + if not files: + raise CremaDDatasetIntegrityError( + "CREMA-D audio integrity check failed: no WAV files were found. " + "The dataset checkout is incomplete." + ) + logger.info( + "DATASET_INTEGRITY_START dataset_id=%s files=%d root=%s", + CREMA_D_CORPUS_ID, + len(files), + dataset_root, + ) + + invalid_count = 0 + samples: list[str] = [] + interval = max(1, len(files) // 10) + last_progress_at = started_at + for processed, audio_path in enumerate(files, start=1): + diagnostic = _audio_integrity_error(audio_path) + if diagnostic is None: + pass + else: + invalid_count += 1 + if len(samples) < 5: + try: + display_path = audio_path.relative_to(dataset_root).as_posix() + except ValueError: + display_path = str(audio_path) + samples.append(f"{display_path}: {diagnostic}") + now = time.perf_counter() + if ( + processed == 1 + or processed == len(files) + or processed % interval == 0 + or (now - last_progress_at) >= 30.0 + ): + logger.info( + "DATASET_INTEGRITY_PROGRESS dataset_id=%s checked=%d total=%d invalid=%d elapsed=%.1fs", + CREMA_D_CORPUS_ID, + processed, + len(files), + invalid_count, + now - started_at, + ) + last_progress_at = now + + if invalid_count: + sample_text = "; ".join(samples) + logger.error( + "DATASET_INTEGRITY_DONE dataset_id=%s files=%d invalid=%d elapsed=%.1fs", + CREMA_D_CORPUS_ID, + len(files), + invalid_count, + time.perf_counter() - started_at, + ) + raise CremaDDatasetIntegrityError( + "CREMA-D audio integrity check failed: " + f"{invalid_count}/{len(files)} invalid file(s). Examples: {sample_text}. " + "CREMA-D audio must be materialized with Git LFS; run `git lfs pull` and " + "`git lfs checkout`, or move the incomplete dataset aside and download it again." + ) + logger.info( + "DATASET_INTEGRITY_DONE dataset_id=%s files=%d invalid=0 elapsed=%.1fs", + CREMA_D_CORPUS_ID, + len(files), + time.perf_counter() - started_at, + ) + return files def _extract_actor_id(file_name: str) -> str | None: @@ -58,22 +182,20 @@ def build_crema_d_utterances( A list of utterances or ``None`` when no usable samples exist. """ - files = sorted(dataset_root.glob(dataset_glob_pattern)) - if not files: - logger.warning("No dataset files found for pattern: %s", dataset_glob_pattern) - return None + files = validate_crema_d_audio_files( + dataset_root=dataset_root, + dataset_glob_pattern=dataset_glob_pattern, + ) utterances: list[Utterance] = [] parse_errors: list[str] = [] root = dataset_root.expanduser() for audio_path in files: - if audio_path.is_dir(): - continue file_name = os.path.basename(str(audio_path)) emotion_code = _extract_emotion_code(file_name) if emotion_code is None: parse_errors.append( - "Skipping file with unexpected name format " f"(missing emotion code): {file_name}" + f"Skipping file with unexpected name format (missing emotion code): {file_name}" ) continue mapped_label = remap_label( diff --git a/ser/_internal/data/dataset_prepare.py b/ser/_internal/data/dataset_prepare.py index 3aa67ca..da124bf 100644 --- a/ser/_internal/data/dataset_prepare.py +++ b/ser/_internal/data/dataset_prepare.py @@ -15,6 +15,10 @@ from pathlib import Path from time import perf_counter +from ser._internal.data.adapters.crema_d import ( + CremaDDatasetIntegrityError, + validate_crema_d_audio_files, +) from ser._internal.data.dataset_registry import ( DatasetRegistryEntry, load_dataset_registry, @@ -57,6 +61,31 @@ class DatasetRegistryHealthIssue: message: str +def validate_registered_dataset_integrity(entry: DatasetRegistryEntry) -> None: + """Validates media required by a registered dataset before it is consumed. + + Args: + entry: Registered dataset paths and metadata. + + Raises: + CremaDDatasetIntegrityError: If registered CREMA-D media is incomplete or invalid. + """ + if entry.dataset_id != "crema-d": + return + dataset_root = entry.dataset_root.expanduser() + pattern = "AudioWAV/**/*.wav" if (dataset_root / "AudioWAV").exists() else "**/*.wav" + try: + validate_crema_d_audio_files( + dataset_root=dataset_root, + dataset_glob_pattern=pattern, + ) + except CremaDDatasetIntegrityError as err: + raise CremaDDatasetIntegrityError( + f"Registered CREMA-D dataset is not ready: {err} " + "After installing Git LFS, run `ser data download --dataset crema-d` to repair it." + ) from err + + SUPPORTED_DATASETS: dict[str, DatasetDescriptor] = { "ravdess": DatasetDescriptor( dataset_id="ravdess", @@ -205,7 +234,7 @@ def _normalize_source_overrides( char.isspace() for char in normalized_source_repo_id ): raise ValueError( - "Invalid --source value. Expected Hugging Face dataset id like " "`namespace/name`." + "Invalid --source value. Expected Hugging Face dataset id like `namespace/name`." ) if normalized_source_revision is not None and any( char.isspace() for char in normalized_source_revision @@ -294,7 +323,7 @@ def _build_dataset_strategy_registry( supported_dataset_ids=resolved_supported_dataset_ids, ) except ValueError as err: - raise ValueError("Dataset strategy registry initialization failed. " f"{err}") from err + raise ValueError(f"Dataset strategy registry initialization failed. {err}") from err _DATASET_STRATEGY_REGISTRY = _build_dataset_strategy_registry() @@ -305,7 +334,7 @@ def _resolve_dataset_strategy(dataset_id: str) -> DatasetStrategy: return _DATASET_STRATEGY_REGISTRY.resolve(dataset_id) except ValueError as err: raise ValueError( - "Dataset strategy resolution failed for " f"dataset_id={dataset_id!r}. {err}" + f"Dataset strategy resolution failed for dataset_id={dataset_id!r}. {err}" ) from err @@ -506,7 +535,9 @@ def collect_dataset_registry_health_issues( ) -> tuple[DatasetRegistryHealthIssue, ...]: """Collects deterministic dataset registry health issues.""" + started_at = perf_counter() registry = load_dataset_registry(settings=settings) + logger.info("REGISTRY_HEALTH_START datasets=%d", len(registry)) issues: list[DatasetRegistryHealthIssue] = [] for entry in sorted(registry.values(), key=lambda item: item.dataset_id): try: @@ -552,6 +583,23 @@ def collect_dataset_registry_health_issues( message=str(err), ) ) + continue + try: + validate_registered_dataset_integrity(entry) + except CremaDDatasetIntegrityError as err: + issues.append( + DatasetRegistryHealthIssue( + dataset_id=entry.dataset_id, + code="dataset_media_invalid", + message=str(err), + ) + ) + logger.info( + "REGISTRY_HEALTH_DONE datasets=%d issues=%d elapsed=%.1fs", + len(registry), + len(issues), + perf_counter() - started_at, + ) return tuple(issues) diff --git a/ser/_internal/data/strategies/default.py b/ser/_internal/data/strategies/default.py index fd8e38d..91f8c59 100644 --- a/ser/_internal/data/strategies/default.py +++ b/ser/_internal/data/strategies/default.py @@ -2,13 +2,19 @@ from __future__ import annotations +import os import shutil import subprocess +import tempfile from pathlib import Path from typing import TYPE_CHECKING from ser._internal.data.adapters.biic_podcast import build_biic_podcast_manifest_jsonl -from ser._internal.data.adapters.crema_d import build_crema_d_manifest_jsonl +from ser._internal.data.adapters.crema_d import ( + CremaDDatasetIntegrityError, + build_crema_d_manifest_jsonl, + validate_crema_d_audio_files, +) from ser._internal.data.adapters.msp_podcast import build_msp_podcast_manifest_jsonl from ser._internal.data.adapters.ravdess import build_ravdess_manifest_jsonl from ser._internal.data.msp_podcast_mirror import ( @@ -46,6 +52,60 @@ logger = get_logger(__name__) +_CREMA_D_AUDIO_PATTERN = "AudioWAV/**/*.wav" +_SUBPROCESS_ERROR_DETAIL_LIMIT = 2_000 + + +def _crema_d_audio_pattern(dataset_root: Path) -> str: + """Returns the narrowest useful audio glob for one CREMA-D tree.""" + return _CREMA_D_AUDIO_PATTERN if (dataset_root / "AudioWAV").exists() else "**/*.wav" + + +def _require_crema_d_git_tools() -> str: + """Resolves the mandatory Git and Git LFS executables for CREMA-D.""" + git_bin = shutil.which("git") + if git_bin is None: + raise RuntimeError( + "git is required to download CREMA-D. Install Git and retry dataset preparation." + ) + if shutil.which("git-lfs") is None: + raise RuntimeError( + "git-lfs is required to download CREMA-D audio. Install it with " + "`brew install git-lfs` or your operating system package manager, then retry." + ) + return git_bin + + +def _run_crema_d_git_command( + command: list[str], + *, + operation: str, + cwd: Path | None = None, +) -> None: + """Runs one mandatory CREMA-D Git command with bounded diagnostics.""" + try: + subprocess.run( + command, + check=True, + capture_output=True, + text=True, + cwd=str(cwd) if cwd is not None else None, + ) + except (OSError, subprocess.CalledProcessError) as err: + stderr = getattr(err, "stderr", None) + detail = stderr.strip() if isinstance(stderr, str) else str(err) + if len(detail) > _SUBPROCESS_ERROR_DETAIL_LIMIT: + detail = f"...{detail[-_SUBPROCESS_ERROR_DETAIL_LIMIT:]}" + raise RuntimeError(f"CREMA-D {operation} failed: {detail}") from err + + +def _validate_crema_d_tree(dataset_root: Path) -> None: + """Validates all CREMA-D audio without decoding complete waveforms.""" + validate_crema_d_audio_files( + dataset_root=dataset_root, + dataset_glob_pattern=_crema_d_audio_pattern(dataset_root), + ) + def _log_manual_download_instructions( *, @@ -136,28 +196,68 @@ def download( source_revision: str | None, ) -> tuple[str | None, str | None]: del source_repo_id, source_revision - if any(dataset_root.iterdir()): - logger.info("CREMA-D target directory is not empty; skipping download.") + dataset_root.mkdir(parents=True, exist_ok=True) + has_existing_content = any(dataset_root.iterdir()) + if has_existing_content: + try: + _validate_crema_d_tree(dataset_root) + except CremaDDatasetIntegrityError as integrity_error: + if not (dataset_root / ".git").exists(): + raise RuntimeError( + f"Existing CREMA-D directory is incomplete: {integrity_error} " + f"Move it aside and retry: {dataset_root}" + ) from integrity_error + git_bin = _require_crema_d_git_tools() + logger.info("Repairing incomplete CREMA-D Git LFS checkout at %s", dataset_root) + for args, operation in ( + (("lfs", "install", "--local"), "Git LFS initialization"), + (("lfs", "pull"), "Git LFS pull"), + (("lfs", "checkout"), "Git LFS checkout"), + ): + _run_crema_d_git_command( + [git_bin, *args], + operation=operation, + cwd=dataset_root, + ) + _validate_crema_d_tree(dataset_root) + return (None, None) + logger.info("Existing CREMA-D dataset passed integrity checks; skipping download.") return (None, None) - git_bin = shutil.which("git") - if git_bin is None: - raise RuntimeError("git is required to download CREMA-D (git-lfs recommended).") - logger.info("Cloning CREMA-D into %s", dataset_root) - subprocess.run( - [ - git_bin, - "clone", - "--depth", - "1", - descriptor.source_url, - str(dataset_root), - ], - check=True, + + git_bin = _require_crema_d_git_tools() + staging_root = Path( + tempfile.mkdtemp(prefix=f".{dataset_root.name}.staging-", dir=dataset_root.parent) ) - git_lfs = shutil.which("git-lfs") - if git_lfs is not None: - logger.info("Running git-lfs pull for CREMA-D") - subprocess.run([git_lfs, "pull"], check=False, cwd=str(dataset_root)) + staging_root.rmdir() + try: + logger.info("Cloning CREMA-D into staging directory %s", staging_root) + _run_crema_d_git_command( + [ + git_bin, + "clone", + "--depth", + "1", + descriptor.source_url, + str(staging_root), + ], + operation="clone", + ) + for args, operation in ( + (("lfs", "install", "--local"), "Git LFS initialization"), + (("lfs", "pull"), "Git LFS pull"), + (("lfs", "checkout"), "Git LFS checkout"), + ): + _run_crema_d_git_command( + [git_bin, *args], + operation=operation, + cwd=staging_root, + ) + _validate_crema_d_tree(staging_root) + dataset_root.rmdir() + os.replace(staging_root, dataset_root) + finally: + if staging_root.exists(): + shutil.rmtree(staging_root) return (None, None) def prepare_manifest( @@ -182,7 +282,7 @@ def prepare_manifest( "NEU": "neutral", "SAD": "sad", } - pattern = "AudioWAV/**/*.wav" if (dataset_root / "AudioWAV").exists() else "**/*.wav" + pattern = _crema_d_audio_pattern(dataset_root) utterances = build_crema_d_manifest_jsonl( dataset_root=dataset_root, dataset_glob_pattern=pattern, diff --git a/ser/_internal/utils/audio_utils.py b/ser/_internal/utils/audio_utils.py index be291b4..0c16888 100644 --- a/ser/_internal/utils/audio_utils.py +++ b/ser/_internal/utils/audio_utils.py @@ -14,6 +14,15 @@ from ser.config import AudioReadConfig logger: logging.Logger = get_logger(__name__) +_GIT_LFS_POINTER_PREFIX = b"version https://git-lfs.github.com/spec/v1" + + +class AudioIntegrityError(OSError): + """Raised when a path contains metadata in place of audio bytes.""" + + +class AudioDecodeError(OSError): + """Raised when one otherwise regular media file cannot be decoded locally.""" def _normalize_audio(audiofile: NDArray[np.float32]) -> NDArray[np.float32]: @@ -77,8 +86,16 @@ def read_audio_file( raise FileNotFoundError(f"Audio file not found: {file_path}") if not path.is_file(): raise OSError(f"Path is not a regular file: {file_path}") + with path.open("rb") as audio_file: + if audio_file.read(len(_GIT_LFS_POINTER_PREFIX)) == _GIT_LFS_POINTER_PREFIX: + raise AudioIntegrityError( + f"Audio file is an unmaterialized Git LFS pointer: {file_path}. " + "Install Git LFS, then run `git lfs pull` and `git lfs checkout` " + "in the dataset checkout." + ) logger.debug(msg=f"Starting to read audio file: {file_path}") + last_error: Exception | None = None for attempt in range(active_config.max_retries): logger.debug(msg=f"Attempt {attempt + 1} to read audio file using librosa.") try: @@ -96,7 +113,9 @@ def read_audio_file( return normalized_audio, int(current_sample_rate) except Exception as err: - logger.warning(msg=f"Librosa failed to read audio file: {err}") + last_error = err + detail = str(err).strip() or type(err).__name__ + logger.warning(msg=f"Librosa failed to read audio file: {detail}") # Segment reads rely on librosa offset/duration. if start_seconds is not None or duration_seconds is not None: @@ -122,7 +141,9 @@ def read_audio_file( return normalized_audio, current_sample_rate except Exception as err: - logger.warning(msg=f"Soundfile also failed: {err}") + last_error = err + detail = str(err).strip() or type(err).__name__ + logger.warning(msg=f"Soundfile also failed: {detail}") if attempt < active_config.max_retries - 1: logger.info( msg=( @@ -133,8 +154,9 @@ def read_audio_file( time.sleep(active_config.retry_delay_seconds) logger.error( - msg=( - f"Failed to read audio file {file_path} " f"after {active_config.max_retries} retries." - ) + msg=(f"Failed to read audio file {file_path} after {active_config.max_retries} retries.") ) - raise OSError(f"Error reading {file_path}") + error = AudioDecodeError(f"Error reading {file_path}") + if last_error is None: + raise error + raise error from last_error diff --git a/tests/suites/integration/data/test_dataset_prepare.py b/tests/suites/integration/data/test_dataset_prepare.py index ba02658..3058f26 100644 --- a/tests/suites/integration/data/test_dataset_prepare.py +++ b/tests/suites/integration/data/test_dataset_prepare.py @@ -1071,6 +1071,41 @@ def test_collect_registry_health_reports_unpinned_manifest_source( assert "missing source pin" in issues[0].message +def test_collect_registry_health_reports_crema_d_lfs_pointer( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + """Startup health checks should block an existing unmaterialized CREMA-D checkout.""" + settings = _settings(tmp_path) + dataset_root = tmp_path / "datasets" / "crema-d" + audio_path = dataset_root / "AudioWAV" / "1077_WSI_ANG_XX.wav" + audio_path.parent.mkdir(parents=True) + audio_path.write_bytes( + b"version https://git-lfs.github.com/spec/v1\n" + b"oid sha256:be5c849653d28aaed49fbca687812f5352f295b1e1a66c269e4a3f2b7ae46489\n" + b"size 85462\n" + ) + entry = DatasetRegistryEntry( + dataset_id="crema-d", + dataset_root=dataset_root, + manifest_path=tmp_path / "manifests" / "crema-d.jsonl", + options={}, + ) + monkeypatch.setattr( + dp, + "load_dataset_registry", + lambda **kwargs: {"crema-d": entry}, + ) + + issues = dp.collect_dataset_registry_health_issues(settings=settings) + + assert len(issues) == 1 + assert issues[0].dataset_id == "crema-d" + assert issues[0].code == "dataset_media_invalid" + assert "unmaterialized Git LFS pointer" in issues[0].message + assert "ser data download --dataset crema-d" in issues[0].message + + def test_prepare_from_registry_entry_detects_msp_mirror_artifact_path_mismatch( tmp_path: Path, ) -> None: diff --git a/tests/suites/unit/data/test_dataset_adapters_crema_d.py b/tests/suites/unit/data/test_dataset_adapters_crema_d.py index e957fc3..e84bbcd 100644 --- a/tests/suites/unit/data/test_dataset_adapters_crema_d.py +++ b/tests/suites/unit/data/test_dataset_adapters_crema_d.py @@ -2,13 +2,21 @@ from __future__ import annotations +import logging from pathlib import Path +import numpy as np +import pytest +import soundfile as sf + from ser._internal.data.adapters.crema_d import ( CREMA_D_CORPUS_ID, CREMA_D_DATASET_LICENSE_ID, CREMA_D_DATASET_POLICY_ID, + CremaDDatasetIntegrityError, + build_crema_d_manifest_jsonl, build_crema_d_utterances, + validate_crema_d_audio_files, ) from ser._internal.data.ontology import LabelOntology @@ -20,12 +28,17 @@ def _ontology() -> LabelOntology: ) +def _write_wav(path: Path, *, frames: int = 4) -> None: + """Writes a minimal valid mono WAV fixture.""" + sf.write(path, np.linspace(-0.5, 0.5, frames, dtype=np.float32), 16_000) + + def test_build_crema_d_utterances_maps_codes_and_metadata(tmp_path: Path) -> None: """CREMA-D adapter should parse filename code and attach metadata.""" audio_root = tmp_path / "AudioWAV" audio_root.mkdir(parents=True, exist_ok=True) - (audio_root / "1001_IEO_HAP_LO.wav").write_bytes(b"fake") - (audio_root / "1002_IEO_SAD_LO.wav").write_bytes(b"fake") + _write_wav(audio_root / "1001_IEO_HAP_LO.wav") + _write_wav(audio_root / "1002_IEO_SAD_LO.wav") utterances = build_crema_d_utterances( dataset_root=tmp_path, @@ -41,3 +54,65 @@ def test_build_crema_d_utterances_maps_codes_and_metadata(tmp_path: Path) -> Non assert {item.corpus for item in utterances} == {CREMA_D_CORPUS_ID} assert {item.dataset_policy_id for item in utterances} == {CREMA_D_DATASET_POLICY_ID} assert {item.dataset_license_id for item in utterances} == {CREMA_D_DATASET_LICENSE_ID} + + +def test_validate_crema_d_audio_accepts_a_one_frame_wav(tmp_path: Path) -> None: + """Metadata validation should accept a genuinely decodable tiny WAV.""" + audio_path = tmp_path / "AudioWAV" / "1001_IEO_HAP_LO.wav" + audio_path.parent.mkdir(parents=True) + _write_wav(audio_path, frames=1) + + files = validate_crema_d_audio_files( + dataset_root=tmp_path, + dataset_glob_pattern="AudioWAV/**/*.wav", + ) + + assert files == (audio_path,) + + +def test_validate_crema_d_audio_logs_integrity_progress( + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + """CREMA-D validation should report progress around metadata scans.""" + audio_root = tmp_path / "AudioWAV" + audio_root.mkdir(parents=True) + _write_wav(audio_root / "1001_IEO_HAP_LO.wav", frames=1) + _write_wav(audio_root / "1002_IEO_HAP_LO.wav", frames=1) + caplog.set_level(logging.INFO, logger="ser._internal.data.adapters.crema_d") + + files = validate_crema_d_audio_files( + dataset_root=tmp_path, + dataset_glob_pattern="AudioWAV/**/*.wav", + ) + + messages = [record.getMessage() for record in caplog.records] + assert len(files) == 2 + assert any(message.startswith("DATASET_INTEGRITY_START") for message in messages) + assert any(message.startswith("DATASET_INTEGRITY_PROGRESS") for message in messages) + assert any(message.startswith("DATASET_INTEGRITY_DONE") for message in messages) + + +def test_crema_d_lfs_pointer_aborts_before_manifest_write(tmp_path: Path) -> None: + """An unmaterialized LFS object must never be registered as ready audio.""" + audio_path = tmp_path / "AudioWAV" / "1077_WSI_ANG_XX.wav" + audio_path.parent.mkdir(parents=True) + audio_path.write_bytes( + b"version https://git-lfs.github.com/spec/v1\n" + b"oid sha256:be5c849653d28aaed49fbca687812f5352f295b1e1a66c269e4a3f2b7ae46489\n" + b"size 85462\n" + ) + manifest_path = tmp_path / "manifests" / "crema-d.jsonl" + + with pytest.raises(CremaDDatasetIntegrityError, match="Git LFS pointer"): + build_crema_d_manifest_jsonl( + dataset_root=tmp_path, + dataset_glob_pattern="AudioWAV/**/*.wav", + emotion_code_map={"ANG": "angry"}, + default_language="en", + ontology=_ontology(), + max_failed_file_ratio=0.5, + output_path=manifest_path, + ) + + assert not manifest_path.exists() diff --git a/tests/suites/unit/data/test_dataset_strategies.py b/tests/suites/unit/data/test_dataset_strategies.py index c294dd3..a5def2a 100644 --- a/tests/suites/unit/data/test_dataset_strategies.py +++ b/tests/suites/unit/data/test_dataset_strategies.py @@ -2,14 +2,19 @@ from __future__ import annotations +import subprocess from pathlib import Path from types import MappingProxyType from typing import cast +import numpy as np import pytest +import soundfile as sf import ser._internal.data.dataset_prepare as dataset_prepare +import ser._internal.data.strategies.default as default_strategies from ser._internal.data.strategies import ( + CremaDDatasetStrategy, DatasetStrategyRegistry, Emodb2DatasetStrategy, PreparedManifestResult, @@ -19,6 +24,24 @@ ) +def _write_crema_wav(dataset_root: Path) -> Path: + """Writes one minimal valid CREMA-D WAV fixture.""" + audio_path = dataset_root / "AudioWAV" / "1001_IEO_HAP_LO.wav" + audio_path.parent.mkdir(parents=True, exist_ok=True) + sf.write(audio_path, np.asarray([0.0, 0.25], dtype=np.float32), 16_000) + return audio_path + + +def _download_crema_d(strategy: CremaDDatasetStrategy, dataset_root: Path) -> None: + """Invokes the CREMA-D strategy with its static descriptor.""" + strategy.download( + descriptor=dataset_prepare.SUPPORTED_DATASETS["crema-d"], + dataset_root=dataset_root, + source_repo_id=None, + source_revision=None, + ) + + def test_build_default_dataset_strategies_contains_expected_ids() -> None: """Default strategy factory should register every supported dataset id.""" strategies = build_default_dataset_strategies() @@ -85,6 +108,165 @@ def test_source_override_support_is_only_enabled_for_msp() -> None: assert strategies["biic-podcast"].supports_source_overrides is False +def test_crema_d_download_requires_git_lfs_before_clone( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Missing Git LFS should fail before acquisition mutates the target.""" + commands: list[list[str]] = [] + monkeypatch.setattr( + default_strategies.shutil, + "which", + lambda executable: "/fake/git" if executable == "git" else None, + ) + monkeypatch.setattr( + default_strategies.subprocess, + "run", + lambda command, **_kwargs: commands.append(command), + ) + dataset_root = tmp_path / "crema-d" + + with pytest.raises(RuntimeError, match=r"git-lfs.*brew install git-lfs"): + _download_crema_d(CremaDDatasetStrategy(), dataset_root) + + assert commands == [] + assert dataset_root.exists() + assert not any(dataset_root.iterdir()) + + +def test_crema_d_lfs_pull_failure_is_fatal_and_cleans_staging( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """A failed LFS transfer must not publish a partial checkout.""" + staging_paths: list[Path] = [] + monkeypatch.setattr(default_strategies.shutil, "which", lambda name: f"/fake/{name}") + + def _run(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: + if command[1] == "clone": + staging_root = Path(command[-1]) + staging_paths.append(staging_root) + (staging_root / ".git").mkdir(parents=True) + if command[1:] == ["lfs", "pull"]: + raise subprocess.CalledProcessError( + returncode=1, + cmd=command, + stderr="quota exceeded", + ) + return subprocess.CompletedProcess(command, 0, "", "") + + monkeypatch.setattr(default_strategies.subprocess, "run", _run) + dataset_root = tmp_path / "crema-d" + + with pytest.raises(RuntimeError, match="Git LFS pull failed: quota exceeded"): + _download_crema_d(CremaDDatasetStrategy(), dataset_root) + + assert dataset_root.exists() + assert not any(dataset_root.iterdir()) + assert staging_paths and all(not path.exists() for path in staging_paths) + + +def test_crema_d_pointer_checkout_is_not_published( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Successful commands still require validated audio before atomic publication.""" + staging_paths: list[Path] = [] + monkeypatch.setattr(default_strategies.shutil, "which", lambda name: f"/fake/{name}") + + def _run(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: + if command[1] == "clone": + staging_root = Path(command[-1]) + staging_paths.append(staging_root) + audio_path = staging_root / "AudioWAV" / "1077_WSI_ANG_XX.wav" + audio_path.parent.mkdir(parents=True) + audio_path.write_bytes(b"version https://git-lfs.github.com/spec/v1\nsize 85462\n") + return subprocess.CompletedProcess(command, 0, "", "") + + monkeypatch.setattr(default_strategies.subprocess, "run", _run) + dataset_root = tmp_path / "crema-d" + + with pytest.raises(RuntimeError, match="unmaterialized Git LFS pointer"): + _download_crema_d(CremaDDatasetStrategy(), dataset_root) + + assert dataset_root.exists() + assert not any(dataset_root.iterdir()) + assert staging_paths and all(not path.exists() for path in staging_paths) + + +def test_crema_d_download_publishes_only_validated_staging_tree( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """A validated checkout should atomically replace the pre-created empty root.""" + monkeypatch.setattr(default_strategies.shutil, "which", lambda name: f"/fake/{name}") + + def _run(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: + if command[1] == "clone": + staging_root = Path(command[-1]) + (staging_root / ".git").mkdir(parents=True) + _write_crema_wav(staging_root) + return subprocess.CompletedProcess(command, 0, "", "") + + monkeypatch.setattr(default_strategies.subprocess, "run", _run) + dataset_root = tmp_path / "crema-d" + + _download_crema_d(CremaDDatasetStrategy(), dataset_root) + + assert (dataset_root / ".git").is_dir() + assert (dataset_root / "AudioWAV" / "1001_IEO_HAP_LO.wav").is_file() + assert not list(tmp_path.glob(".crema-d.staging-*")) + + +def test_crema_d_existing_non_checkout_is_preserved( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Invalid user data outside a Git checkout must never be deleted or overwritten.""" + dataset_root = tmp_path / "crema-d" + pointer_path = dataset_root / "AudioWAV" / "1077_WSI_ANG_XX.wav" + pointer_path.parent.mkdir(parents=True) + pointer_bytes = b"version https://git-lfs.github.com/spec/v1\nsize 85462\n" + pointer_path.write_bytes(pointer_bytes) + monkeypatch.setattr( + default_strategies.subprocess, + "run", + lambda *_args, **_kwargs: pytest.fail("subprocess should not run"), + ) + + with pytest.raises(RuntimeError, match="Move it aside and retry"): + _download_crema_d(CremaDDatasetStrategy(), dataset_root) + + assert pointer_path.read_bytes() == pointer_bytes + + +def test_crema_d_existing_git_checkout_is_repaired_and_revalidated( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """A compatible partial checkout should run explicit LFS repair commands.""" + dataset_root = tmp_path / "crema-d" + (dataset_root / ".git").mkdir(parents=True) + pointer_path = dataset_root / "AudioWAV" / "1077_WSI_ANG_XX.wav" + pointer_path.parent.mkdir(parents=True) + pointer_path.write_bytes(b"version https://git-lfs.github.com/spec/v1\nsize 85462\n") + commands: list[list[str]] = [] + monkeypatch.setattr(default_strategies.shutil, "which", lambda name: f"/fake/{name}") + + def _run(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: + commands.append(command) + if command[1:] == ["lfs", "checkout"]: + _write_crema_wav(dataset_root) + pointer_path.unlink() + return subprocess.CompletedProcess(command, 0, "", "") + + monkeypatch.setattr(default_strategies.subprocess, "run", _run) + + _download_crema_d(CremaDDatasetStrategy(), dataset_root) + + assert [command[1:] for command in commands] == [ + ["lfs", "install", "--local"], + ["lfs", "pull"], + ["lfs", "checkout"], + ] + assert (dataset_root / ".git").is_dir() + assert (dataset_root / "AudioWAV" / "1001_IEO_HAP_LO.wav").is_file() + + def test_dataset_strategy_registry_rejects_missing_supported_ids() -> None: """Registry construction should fail when supported ids are missing mappings.""" with pytest.raises(ValueError, match="Missing strategy ids"): diff --git a/tests/suites/unit/utils/test_audio_utils.py b/tests/suites/unit/utils/test_audio_utils.py index fd02368..b0d8386 100644 --- a/tests/suites/unit/utils/test_audio_utils.py +++ b/tests/suites/unit/utils/test_audio_utils.py @@ -116,3 +116,53 @@ def _fake_librosa_load( assert captured["sr"] is None assert captured["offset"] == pytest.approx(1.5) assert captured["duration"] == pytest.approx(0.75) + + +def test_read_audio_file_rejects_lfs_pointer_without_decoder_retries( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """LFS pointers should produce an actionable permanent integrity failure immediately.""" + audio_path = tmp_path / "sample.wav" + audio_path.write_bytes( + b"version https://git-lfs.github.com/spec/v1\noid sha256:0123456789abcdef\nsize 1234\n" + ) + + def _unexpected(*_args: object, **_kwargs: object) -> None: + raise AssertionError("decoder or retry invoked") + + monkeypatch.setattr(au.librosa, "load", _unexpected) + monkeypatch.setattr(au.sf, "SoundFile", _unexpected) + monkeypatch.setattr(au.time, "sleep", _unexpected) + + with pytest.raises(au.AudioIntegrityError, match="git lfs pull"): + au.read_audio_file( + str(audio_path), + audio_read_config=AudioReadConfig(max_retries=3, retry_delay_seconds=1.0), + ) + + +def test_read_audio_file_chains_the_terminal_decoder_failure( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Generic read errors should retain the decoder cause for diagnostics.""" + audio_path = tmp_path / "sample.wav" + audio_path.write_bytes(b"not-audio") + terminal_error = RuntimeError("soundfile decoder failed") + monkeypatch.setattr( + au.librosa, + "load", + lambda *_args, **_kwargs: (_ for _ in ()).throw(ValueError("librosa failed")), + ) + monkeypatch.setattr( + au.sf, + "SoundFile", + lambda *_args, **_kwargs: (_ for _ in ()).throw(terminal_error), + ) + + with pytest.raises(OSError, match="Error reading") as exc_info: + au.read_audio_file( + str(audio_path), + audio_read_config=AudioReadConfig(max_retries=1, retry_delay_seconds=0.0), + ) + + assert exc_info.value.__cause__ is terminal_error