diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml deleted file mode 100644 index ba93a87f..00000000 --- a/.github/workflows/test.yaml +++ /dev/null @@ -1,111 +0,0 @@ -name: Python Tests - -on: - pull_request: - -jobs: - unit: - runs-on: ${{ matrix.os }} - strategy: - fail-fast: false - matrix: - include: - - {os: ubuntu-latest, architecture: x64, python-version: '3.10'} - - {os: ubuntu-latest, architecture: x64, python-version: '3.11'} - - {os: ubuntu-latest, architecture: x64, python-version: '3.12'} - - {os: macos-latest, architecture: x64, python-version: '3.10'} - - {os: macos-latest, architecture: arm64, python-version: '3.10'} - - {os: macos-latest, architecture: x64, python-version: '3.11'} - - {os: macos-latest, architecture: arm64, python-version: '3.11'} - - {os: macos-latest, architecture: x64, python-version: '3.12'} - - {os: macos-latest, architecture: arm64, python-version: '3.12'} - # - {os: windows-latest, architecture: x64, python-version: '3.10'} - # - {os: windows-latest, architecture: x64, python-version: '3.11'} - env: - GITHUB_ACTIONS: true - steps: - - uses: actions/checkout@v4 - with: # no need for the history - fetch-depth: 1 - - name: Set up Python ${{ matrix.python-version }} - uses: actions/setup-python@v5 - with: - python-version: ${{ matrix.python-version }} - - - name: Install ffmpeg (Ubuntu) - if: startsWith(matrix.os, 'ubuntu') - run: sudo apt-get update && sudo apt-get install -y ffmpeg - - name: Install ffmpeg (macOS) - if: startsWith(matrix.os, 'macos') - run: brew install ffmpeg - - name: Install ffmpeg (Windows) - if: startsWith(matrix.os, 'windows') - run: choco install ffmpeg - - - name: Install pipx and ensure it's up to date - run: | - python -m pip install --upgrade pipx - pipx ensurepath - shell: bash - - name: Install poetry - run: pipx install poetry==1.7.1 - shell: bash - - name: Install dependencies with Poetry - run: | - poetry run pip install iso-639 - poetry install --with dev - shell: bash - - name: Run unit tests - id: run-tests - env: - HF_TOKEN: ${{ secrets.HF_TOKEN }} - run: > - poetry run pytest \ - --junitxml=pytest.xml \ - --cov-report=term-missing:skip-covered \ - --cov-report=xml:coverage.xml \ - --cov=src src/tests \ - --log-level=DEBUG \ - --verbose - shell: bash - - name: Upload coverage to Codecov - uses: codecov/codecov-action@v4 - with: - token: ${{ secrets.CODECOV_TOKEN }} - - pre-commit: - runs-on: ${{ matrix.os }} - strategy: - matrix: - os: [ubuntu-latest] # For demonstration, other OSes are commented out: macos-latest, windows-latest - python-version: ['3.10'] # For speeding up the process we removed "3.11" for now - steps: - - uses: actions/checkout@v4 - with: # no need for the history - fetch-depth: 1 - - name: Set up Python - uses: actions/setup-python@v5 - with: - python-version: ${{ matrix.python-version }} - - name: Install pipx and ensure it's up to date - run: | - python -m pip install --upgrade pipx - pipx ensurepath - shell: bash - - name: Install poetry - run: pipx install poetry==1.7.1 - shell: bash - - name: Install dependencies with Poetry - run: | - poetry run pip install iso-639 - poetry install --with dev - shell: bash - - name: Install pre-commit - run: pipx install pre-commit - shell: bash - - name: Run pre-commit - env: - SKIP: pytest - run: | - poetry run pre-commit run --all-files - shell: bash diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml new file mode 100644 index 00000000..36bb4da2 --- /dev/null +++ b/.github/workflows/tests.yaml @@ -0,0 +1,387 @@ +name: github-runner-tests + +on: + pull_request: + types: [opened, synchronize, reopened, labeled] + +jobs: + macos-tests: + if: github.event.pull_request.draft == false && contains(github.event.pull_request.labels.*.name, 'to-test') + name: macOS-tests + runs-on: ${{ matrix.os }} + strategy: + fail-fast: true + matrix: + include: + - {os: macos-latest, architecture: arm64, python-version: '3.10'} + # - {os: macos-latest, architecture: arm64, python-version: '3.11'} + # the reason why we commented out 3.11 is that it hits github rate limit for some modules (e.g., knn-vc, Camb-ai/mars5-tts) + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 1 # no need for the history + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install ffmpeg (Ubuntu) + if: startsWith(matrix.os, 'ubuntu') + run: sudo apt-get update && sudo apt-get install -y ffmpeg + shell: bash + - name: Install ffmpeg (macOS) + if: startsWith(matrix.os, 'macos') + run: brew install ffmpeg + shell: bash + - name: Install Poetry + uses: snok/install-poetry@v1 + with: + version: 1.7.1 + virtualenvs-create: true + virtualenvs-in-project: true + - name: Install dependencies with Poetry + run: | + poetry run pip install iso-639 + poetry install --with dev + shell: bash + - name: Run unit tests + id: run-tests + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + HF_TOKEN: ${{ secrets.HF_TOKEN }} + run: > + poetry run pytest -n auto \ + --junitxml=pytest.xml \ + --cov-report=term-missing:skip-covered \ + --cov-report=xml:coverage.xml \ + --cov=src src/tests \ + --log-level=DEBUG \ + --verbose + shell: bash + - name: Upload coverage to Codecov + uses: codecov/codecov-action@v4 + with: + token: ${{ secrets.CODECOV_TOKEN }} + + pre-commit: + if: github.event.pull_request.draft == false && contains(github.event.pull_request.labels.*.name, 'to-test') + runs-on: ${{ matrix.os }} + strategy: + matrix: + os: [ubuntu-latest] + python-version: ['3.10'] + steps: + - uses: actions/checkout@v4 + with: # no need for the history + fetch-depth: 1 + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install Poetry + uses: snok/install-poetry@v1 + with: + version: 1.7.1 + virtualenvs-create: true + virtualenvs-in-project: true + - name: Install dependencies with Poetry + run: | + poetry run pip install iso-639 + poetry install --with dev + shell: bash + - name: Install pre-commit + run: pipx install pre-commit + shell: bash + - name: Run pre-commit + run: | + poetry run pre-commit run --all-files + shell: bash + + start-runner-310: + if: github.event.pull_request.draft == false && contains(github.event.pull_request.labels.*.name, 'to-test-gpu') && success() + needs: + - pre-commit + - macos-tests + name: start-runner-310 + runs-on: ubuntu-latest + outputs: + label: ${{ steps.start-ec2-runner.outputs.label }} + ec2-instance-id: ${{ steps.start-ec2-runner.outputs.ec2-instance-id }} + job-ran: ${{ steps.set-ran.outputs.ran }} + steps: + - id: set-ran + run: echo "::set-output name=ran::true" + - name: Configure AWS credentials + uses: aws-actions/configure-aws-credentials@v1 + with: + aws-access-key-id: ${{ secrets.AWS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_KEY_SECRET }} + aws-region: ${{ vars.AWS_REGION }} + - name: Start EC2 runner + id: start-ec2-runner + uses: machulav/ec2-github-runner@v2 + with: + mode: start + github-token: ${{ secrets.GH_TOKEN }} + ec2-image-id: ${{ vars.AWS_IMAGE_ID }} + ec2-instance-type: ${{ vars.AWS_INSTANCE_TYPE }} + subnet-id: ${{ vars.AWS_SUBNET }} + security-group-id: ${{ vars.AWS_SECURITY_GROUP }} + + ubuntu-tests-310: + name: ubuntu-tests-310 + needs: start-runner-310 + runs-on: ${{ needs.start-runner-310.outputs.label }} + defaults: + run: + shell: bash + working-directory: ${{ vars.WORKING_DIR }} + strategy: + matrix: + python-version: ['3.10'] + env: + WORKING_DIR: ${{ vars.WORKING_DIR }} + POETRY_CACHE_DIR: ${{ vars.WORKING_DIR }} + outputs: + job-ran: ${{ steps.set-ran.outputs.ran }} + steps: + - id: set-ran + run: echo "::set-output name=ran::true" + - uses: actions/checkout@v4 + with: + fetch-depth: 1 # no need for the history + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install ffmpeg (Ubuntu) + if: startsWith(matrix.os, 'ubuntu') + run: sudo apt-get update && sudo apt-get install -y ffmpeg + shell: bash + - name: Install Poetry + uses: snok/install-poetry@v1 + with: + version: 1.7.1 + virtualenvs-create: true + virtualenvs-in-project: true + - name: Check available space + run: | + df -h + shell: bash + - name: Echo python info + run: | + python --version + which python + shell: bash + - name: Copy senselab directory to current directory + run: | + cp -r /actions-runner/_work/senselab/senselab . + - name: Install dependencies with Poetry + run: | + cd senselab + poetry env use ${{ matrix.python-version }} + poetry run pip install iso-639 + poetry install --with dev + shell: bash + - name: Check poetry info + run: | + cd senselab + poetry env info + poetry --version + shell: bash + - name: Check NVIDIA SMI details + run: | + cd senselab + poetry run nvidia-smi + poetry run nvidia-smi -L + poetry run nvidia-smi -q -d Memory + shell: bash + - name: Prepare cache folder for pytest + run: mkdir -p $WORKING_DIR/pytest/temp + shell: bash + - name: Run unit tests + id: run-tests + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + HF_TOKEN: ${{ secrets.HF_TOKEN }} + run: > + cd senselab && poetry run pytest \ + --rootdir=$WORKING_DIR/pytest \ + --basetemp=$WORKING_DIR/pytest/temp \ + --junitxml=pytest.xml \ + --cov-report=term-missing:skip-covered \ + --cov-report=xml:coverage.xml \ + --cov=src src/tests \ + --log-level=DEBUG \ + --verbose + shell: bash + + stop-runner-310: + name: stop-runner-310 + needs: + - start-runner-310 # waits for the EC2 instance to be created + - ubuntu-tests-310 # waits for the actual job to finish + runs-on: ubuntu-latest + if: ${{ needs.start-runner-310.outputs.job-ran == 'true' && needs.ubuntu-tests-310.outputs.job-ran == 'true' || failure() }} # required to stop the runner even if an error occurred in previous jobs + steps: + - name: Check available space + run: | + df -h + shell: bash + - name: Configure AWS credentials + uses: aws-actions/configure-aws-credentials@v1 + with: + aws-access-key-id: ${{ secrets.AWS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_KEY_SECRET }} + aws-region: ${{ vars.AWS_REGION }} + - name: Stop EC2 runner + uses: machulav/ec2-github-runner@v2 + with: + mode: stop + github-token: ${{ secrets.GH_TOKEN }} + label: ${{ needs.start-runner-310.outputs.label }} + ec2-instance-id: ${{ needs.start-runner-310.outputs.ec2-instance-id }} + + start-runner-311: + if: github.event.pull_request.draft == false && contains(github.event.pull_request.labels.*.name, 'to-test-gpu') && success() + needs: + - pre-commit + - macos-tests + name: start-runner-311 + runs-on: ubuntu-latest + outputs: + label: ${{ steps.start-ec2-runner.outputs.label }} + ec2-instance-id: ${{ steps.start-ec2-runner.outputs.ec2-instance-id }} + job-ran: ${{ steps.set-ran.outputs.ran }} + steps: + - id: set-ran + run: echo "::set-output name=ran::true" + - name: Configure AWS credentials + uses: aws-actions/configure-aws-credentials@v1 + with: + aws-access-key-id: ${{ secrets.AWS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_KEY_SECRET }} + aws-region: ${{ vars.AWS_REGION }} + - name: Start EC2 runner + id: start-ec2-runner + uses: machulav/ec2-github-runner@v2 + with: + mode: start + github-token: ${{ secrets.GH_TOKEN }} + ec2-image-id: ${{ vars.AWS_IMAGE_ID }} + ec2-instance-type: ${{ vars.AWS_INSTANCE_TYPE }} + subnet-id: ${{ vars.AWS_SUBNET }} + security-group-id: ${{ vars.AWS_SECURITY_GROUP }} + + ubuntu-tests-311: + name: ubuntu-tests-311 + needs: start-runner-311 + runs-on: ${{ needs.start-runner-311.outputs.label }} + defaults: + run: + shell: bash + working-directory: ${{ vars.WORKING_DIR }} + strategy: + matrix: + python-version: ['3.11'] + env: + WORKING_DIR: ${{ vars.WORKING_DIR }} + POETRY_CACHE_DIR: ${{ vars.WORKING_DIR }} + outputs: + job-ran: ${{ steps.set-ran.outputs.ran }} + steps: + - id: set-ran + run: echo "::set-output name=ran::true" + - uses: actions/checkout@v4 + with: + fetch-depth: 1 # no need for the history + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install ffmpeg (Ubuntu) + if: startsWith(matrix.os, 'ubuntu') + run: sudo apt-get update && sudo apt-get install -y ffmpeg + shell: bash + - name: Install Poetry + uses: snok/install-poetry@v1 + with: + version: 1.7.1 + virtualenvs-create: true + virtualenvs-in-project: true + - name: Check available space + run: | + df -h + shell: bash + - name: Echo python info + run: | + python --version + which python + shell: bash + - name: Copy senselab directory to current directory + run: | + cp -r /actions-runner/_work/senselab/senselab . + - name: Install dependencies with Poetry + run: | + cd senselab + poetry env use ${{ matrix.python-version }} + poetry run pip install iso-639 + poetry install --with dev + shell: bash + - name: Check poetry info + run: | + cd senselab + poetry env info + poetry --version + shell: bash + - name: Check NVIDIA SMI details + run: | + cd senselab + poetry run nvidia-smi + poetry run nvidia-smi -L + poetry run nvidia-smi -q -d Memory + shell: bash + - name: Prepare cache folder for pytest + run: mkdir -p $WORKING_DIR/pytest/temp + shell: bash + - name: Run unit tests + id: run-tests + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + HF_TOKEN: ${{ secrets.HF_TOKEN }} + run: > + cd senselab && poetry run pytest \ + --rootdir=$WORKING_DIR/pytest \ + --basetemp=$WORKING_DIR/pytest/temp \ + --junitxml=pytest.xml \ + --cov-report=term-missing:skip-covered \ + --cov-report=xml:coverage.xml \ + --cov=src src/tests \ + --log-level=DEBUG \ + --verbose + shell: bash + + stop-runner-311: + name: stop-runner-311 + needs: + - start-runner-311 # waits for the EC2 instance to be created + - ubuntu-tests-311 # waits for the actual job to finish + runs-on: ubuntu-latest + if: ${{ needs.start-runner-311.outputs.job-ran == 'true' && needs.ubuntu-tests-311.outputs.job-ran == 'true' || failure() }} # required to stop the runner even if an error occurred in previous jobs + steps: + - name: Check available space + run: | + df -h + shell: bash + - name: Configure AWS credentials + uses: aws-actions/configure-aws-credentials@v1 + with: + aws-access-key-id: ${{ secrets.AWS_KEY_ID }} + aws-secret-access-key: ${{ secrets.AWS_KEY_SECRET }} + aws-region: ${{ vars.AWS_REGION }} + - name: Stop EC2 runner + uses: machulav/ec2-github-runner@v2 + with: + mode: stop + github-token: ${{ secrets.GH_TOKEN }} + label: ${{ needs.start-runner-311.outputs.label }} + ec2-instance-id: ${{ needs.start-runner-311.outputs.ec2-instance-id }} diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 77e87ebc..c2500214 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -73,13 +73,3 @@ repos: entry: YAML files must have .yaml extension. language: fail files: \.yml$ - -- repo: local - hooks: - - id: pytest - name: pytest - entry: poetry run pytest --testmon - language: system - types: [python] - pass_filenames: false - always_run: true diff --git a/CHANGELOG.md b/CHANGELOG.md index be94ca37..d91d7598 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,16 @@ +# 0.13.0 (Fri Sep 20 2024) + +#### 🚀 Enhancement + +- GitHub wf experiments [#163](https://github.com/sensein/senselab/pull/163) ([@fabiocat93](https://github.com/fabiocat93) [@wilke0818](https://github.com/wilke0818)) + +#### Authors: 2 + +- [@wilke0818](https://github.com/wilke0818) +- Fabio Catania ([@fabiocat93](https://github.com/fabiocat93)) + +--- + # 0.12.0 (Thu Sep 12 2024) #### 🚀 Enhancement diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 34713b35..2058bdfc 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -55,7 +55,7 @@ If you feel that the functionality you have added to senselab requires some extr ### An example of well documented function following Google-style -```` +```python import statistics from typing import Dict, List @@ -99,4 +99,4 @@ def calculate_statistics(data: List[float]) -> Dict[str, float]: 'variance': variance, 'std_dev': std_dev } -```` +``` diff --git a/pyproject.toml b/pyproject.toml index 3df48c7e..3785793b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -20,7 +20,6 @@ classifiers = [ "Development Status :: 3 - Alpha", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", - "Programming Language :: Python :: 3.12", "License :: OSI Approved :: Apache Software License", "Operating System :: OS Independent" ] @@ -62,16 +61,15 @@ vocos = "~=0.1" optional = true [tool.poetry.group.dev.dependencies] -pytest = "~=8.2" +pytest-xdist = {version = "~=3.6.1", extras = ["psutil"]} pytest-mock = "~=3.14" +pytest-cov = "~=5.0" mypy = "~=1.9" pre-commit = "~=3.7" -pytest-cov = "~=5.0" ruff = "~=0.3" codespell = "~=2.3" jupyter = "~=1.0" ipywidgets = "~=8.1" -pytest-testmon = "~=2.1.1" [tool.poetry.group.docs] optional = true diff --git a/src/senselab/audio/workflows/transcribe_timestamped/__init__.py b/src/senselab/audio/workflows/transcribe_timestamped/__init__.py index aa7553a4..cbee06b7 100644 --- a/src/senselab/audio/workflows/transcribe_timestamped/__init__.py +++ b/src/senselab/audio/workflows/transcribe_timestamped/__init__.py @@ -1,5 +1,8 @@ """Workflow for timestamped transcription.""" +""" +# TODO: Please double-check this because tests are failing from senselab.audio.workflows.transcribe_timestamped.transcribe_timestamped import transcribe_timestamped __all__ = ["transcribe_timestamped"] +""" diff --git a/src/senselab/audio/workflows/transcribe_timestamped/transcribe_timestamped.py b/src/senselab/audio/workflows/transcribe_timestamped/transcribe_timestamped.py index 4ebde0ea..39ebd7c9 100644 --- a/src/senselab/audio/workflows/transcribe_timestamped/transcribe_timestamped.py +++ b/src/senselab/audio/workflows/transcribe_timestamped/transcribe_timestamped.py @@ -1,5 +1,7 @@ """Transcribes audio files with timestamps.""" +''' +# TODO: Please double-check this because tests are failing from typing import List import pydra @@ -77,7 +79,7 @@ def transcribe_task(audios: List[Audio], model: HFModel, language: Language) -> model=wf.lzin.model, language=wf.lzin.language, ) - ).split("batched_audios", batched_audios=wf.inputs.batched_audios) + ).split("batched_audios", batched_audios=wf.transcribe.lzin.batched_audios) align_transcriptions_task = pydra.mark.task(align_transcriptions) wf.add( @@ -99,3 +101,4 @@ def transcribe_task(audios: List[Audio], model: HFModel, language: Language) -> sub(wf) return wf.result()[0].output.aligned_transcriptions +''' diff --git a/src/senselab/text/tasks/embeddings_extraction/huggingface.py b/src/senselab/text/tasks/embeddings_extraction/huggingface.py index 0c365eb0..ed725362 100644 --- a/src/senselab/text/tasks/embeddings_extraction/huggingface.py +++ b/src/senselab/text/tasks/embeddings_extraction/huggingface.py @@ -78,6 +78,9 @@ def extract_text_embeddings( device, _ = _select_device_and_dtype( user_preference=device, compatible_devices=[DeviceType.CUDA, DeviceType.CPU] ) + + print(f"Using device: {device}") + # Load tokenizer and model tokenizer = cls._get_tokenizer(model=model) ssl_model = cls._load_model(model=model, device=device) @@ -87,13 +90,15 @@ def extract_text_embeddings( # Process each piece of text individually for text in pieces_of_text: # Tokenize sentence - encoded_input = tokenizer(text, return_tensors="pt").to(device) + encoded_input = tokenizer(text, return_tensors="pt").to(device.value) # Compute token embeddings with torch.no_grad(): model_output = ssl_model(**encoded_input, output_hidden_states=True) hidden_states = model_output.hidden_states - concatenated_hidden_states = torch.cat([state.unsqueeze(0) for state in hidden_states], dim=0) + concatenated_hidden_states = torch.cat( + [state.to(device.value).unsqueeze(0) for state in hidden_states], dim=0 + ) embeddings.append(concatenated_hidden_states.squeeze()) return embeddings diff --git a/src/senselab/utils/data_structures/model.py b/src/senselab/utils/data_structures/model.py index 770e6c9b..2163f521 100644 --- a/src/senselab/utils/data_structures/model.py +++ b/src/senselab/utils/data_structures/model.py @@ -1,6 +1,7 @@ """This module implements some utilities for the model class.""" import os +from functools import lru_cache from pathlib import Path from typing import Optional, Union @@ -155,14 +156,25 @@ def check_hf_repo_exists(repo_id: str, revision: str = "main", repo_type: str = return False +@lru_cache(maxsize=128) def check_github_repo_exists(repo_id: str, branch: str = "main") -> bool: - """Private function to check if a GitHub repository exists.""" + """Private function to check if a GitHub repository exists with caching and authentication.""" url = f"https://api.github.com/repos/{repo_id}/branches/{branch}" - response = requests.get(url, timeout=10) + token = os.getenv("GITHUB_TOKEN") or None + + headers = {} + if token: + headers = {"Authorization": f"token {token}"} + + response = requests.get(url, headers=headers, timeout=10) + if response.status_code == 200: return True elif response.status_code == 404: return False + elif response.status_code == 403: # Handle rate limit exceeded + print("GitHub API rate limit exceeded. Please try again later.") + return False else: response.raise_for_status() return False diff --git a/src/senselab/utils/tasks/plotting.py b/src/senselab/utils/tasks/plotting.py index 00668cc4..c8398c9a 100644 --- a/src/senselab/utils/tasks/plotting.py +++ b/src/senselab/utils/tasks/plotting.py @@ -45,7 +45,8 @@ def plot_transcript(transcript: ScriptLine) -> None: for i, text in enumerate(texts): if start_times[i] is not None and end_times[i] is not None: ax.plot([start_times[i], end_times[i]], [i, i], marker="o") - ax.text((start_times[i] + end_times[i]) / 2, i, text, ha="center", va="bottom") + if text: + ax.text((start_times[i] + end_times[i]) / 2, i, text, ha="center", va="bottom") # Setting labels and title ax.set_yticks(range(len(texts))) diff --git a/src/tests/audio/tasks/classification_test.py b/src/tests/audio/tasks/classification_test.py index ffedd5d1..6d414273 100644 --- a/src/tests/audio/tasks/classification_test.py +++ b/src/tests/audio/tasks/classification_test.py @@ -1,6 +1,7 @@ """Test audio classification APIs.""" -import os +import pytest +import torch from senselab.audio.data_structures.audio import Audio from senselab.audio.tasks.classification.speech_emotion_recognition import speech_emotion_recognition_with_hf_models @@ -8,30 +9,30 @@ from senselab.utils.data_structures.model import HFModel from tests.audio.conftest import MONO_AUDIO_PATH -if os.getenv("GITHUB_ACTIONS") != "true": - - def test_speech_emotion_recognition() -> None: - """Tests speech emotion recognition.""" - audio_dataset = [Audio.from_filepath(MONO_AUDIO_PATH)] - - resampled_audios = resample_audios(audio_dataset, 16000) # some pipelines resample for us but can't guarantee - - # Discrete test - result = speech_emotion_recognition_with_hf_models( - resampled_audios, HFModel(path_or_uri="ehcalabres/wav2vec2-lg-xlsr-en-speech-emotion-recognition") - ) - top_emotion, emotion_probs = result[0] - rav_emotions = ["angry", "calm", "disgust", "fearful", "happy", "neutral", "sad", "surprised"] - assert top_emotion in rav_emotions, "Top emotion should be in RAVDESS Dataset" - - for emotion in emotion_probs: - assert emotion in rav_emotions - - # Continuous test - result = speech_emotion_recognition_with_hf_models( - resampled_audios, HFModel(path_or_uri="audeering/wav2vec2-large-robust-12-ft-emotion-msp-dim") - ) - emotion, continuous_values = result[0] - assert emotion in ["arousal", "valence", "dominance"], "No emotion here but rather is one of \ - arousal, valence, or dominance" - assert set(continuous_values.keys()) == set(["arousal", "valence", "dominance"]) + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_speech_emotion_recognition() -> None: + """Tests speech emotion recognition.""" + audio_dataset = [Audio.from_filepath(MONO_AUDIO_PATH)] + + resampled_audios = resample_audios(audio_dataset, 16000) # some pipelines resample for us but can't guarantee + + # Discrete test + result = speech_emotion_recognition_with_hf_models( + resampled_audios, HFModel(path_or_uri="ehcalabres/wav2vec2-lg-xlsr-en-speech-emotion-recognition") + ) + top_emotion, emotion_probs = result[0] + rav_emotions = ["angry", "calm", "disgust", "fearful", "happy", "neutral", "sad", "surprised"] + assert top_emotion in rav_emotions, "Top emotion should be in RAVDESS Dataset" + + for emotion in emotion_probs: + assert emotion in rav_emotions + + # Continuous test + result = speech_emotion_recognition_with_hf_models( + resampled_audios, HFModel(path_or_uri="audeering/wav2vec2-large-robust-12-ft-emotion-msp-dim") + ) + emotion, continuous_values = result[0] + assert emotion in ["arousal", "valence", "dominance"], "No emotion here but rather is one of \ + arousal, valence, or dominance" + assert set(continuous_values.keys()) == set(["arousal", "valence", "dominance"]) diff --git a/src/tests/audio/tasks/features_extraction_test.py b/src/tests/audio/tasks/features_extraction_test.py index d613d0bc..949b18f2 100644 --- a/src/tests/audio/tasks/features_extraction_test.py +++ b/src/tests/audio/tasks/features_extraction_test.py @@ -1,7 +1,5 @@ """This script contains unit tests for the features extraction tasks.""" -import os - import pytest import torch @@ -21,6 +19,10 @@ extract_pitch_from_audios, extract_spectrogram_from_audios, ) +from senselab.audio.tasks.features_extraction.torchaudio_squim import ( + extract_objective_quality_features_from_audios, + extract_subjective_quality_features_from_audios, +) def test_extract_spectrogram_from_audios(resampled_mono_audio_sample: Audio) -> None: @@ -168,40 +170,41 @@ def test_extract_opensmile_features_from_audios(resampled_mono_audio_sample: Aud assert all(isinstance(value, (float, int)) for value in features.values()) -if os.getenv("GITHUB_ACTIONS") != "true": - from senselab.audio.tasks.features_extraction.torchaudio_squim import ( - extract_objective_quality_features_from_audios, - extract_subjective_quality_features_from_audios, +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_extract_objective_quality_features_from_audios(resampled_mono_audio_sample: Audio) -> None: + """Test extraction of objective quality features from audio.""" + result = extract_objective_quality_features_from_audios([resampled_mono_audio_sample]) + assert isinstance(result, dict) + assert "stoi" in result + assert "pesq" in result + assert "si_sdr" in result + assert all(isinstance(feature, float) for feature in result["stoi"]) + assert all(isinstance(feature, float) for feature in result["pesq"]) + assert all(isinstance(feature, float) for feature in result["si_sdr"]) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_extract_objective_quality_features_from_audios_invalid_audio(mono_audio_sample: Audio) -> None: + """Test extraction of objective quality features from invalid audio.""" + with pytest.raises(ValueError, match="Only 16000 Hz sampling rate is supported by Torchaudio-Squim model."): + extract_objective_quality_features_from_audios([mono_audio_sample]) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_extract_subjective_quality_features_from_audios(resampled_mono_audio_sample: Audio) -> None: + """Test extraction of subjective quality features from audio.""" + result = extract_subjective_quality_features_from_audios( + audio_list=[resampled_mono_audio_sample], non_matching_references=[resampled_mono_audio_sample] ) + assert isinstance(result, dict) + assert "mos" in result + assert all(isinstance(feature, float) for feature in result["mos"]) + - def test_extract_objective_quality_features_from_audios(resampled_mono_audio_sample: Audio) -> None: - """Test extraction of objective quality features from audio.""" - result = extract_objective_quality_features_from_audios([resampled_mono_audio_sample]) - assert isinstance(result, dict) - assert "stoi" in result - assert "pesq" in result - assert "si_sdr" in result - assert all(isinstance(feature, float) for feature in result["stoi"]) - assert all(isinstance(feature, float) for feature in result["pesq"]) - assert all(isinstance(feature, float) for feature in result["si_sdr"]) - - def test_extract_objective_quality_features_from_audios_invalid_audio(mono_audio_sample: Audio) -> None: - """Test extraction of objective quality features from invalid audio.""" - with pytest.raises(ValueError, match="Only 16000 Hz sampling rate is supported by Torchaudio-Squim model."): - extract_objective_quality_features_from_audios([mono_audio_sample]) - - def test_extract_subjective_quality_features_from_audios(resampled_mono_audio_sample: Audio) -> None: - """Test extraction of subjective quality features from audio.""" - result = extract_subjective_quality_features_from_audios( - audio_list=[resampled_mono_audio_sample], non_matching_references=[resampled_mono_audio_sample] +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_extract_subjective_quality_features_invalid_audio(mono_audio_sample: Audio) -> None: + """Test extraction of subjective quality features from invalid audio.""" + with pytest.raises(ValueError, match="Only 16000 Hz sampling rate is supported by Torchaudio-Squim model."): + extract_subjective_quality_features_from_audios( + audio_list=[mono_audio_sample], non_matching_references=[mono_audio_sample] ) - assert isinstance(result, dict) - assert "mos" in result - assert all(isinstance(feature, float) for feature in result["mos"]) - - def test_extract_subjective_quality_features_invalid_audio(mono_audio_sample: Audio) -> None: - """Test extraction of subjective quality features from invalid audio.""" - with pytest.raises(ValueError, match="Only 16000 Hz sampling rate is supported by Torchaudio-Squim model."): - extract_subjective_quality_features_from_audios( - audio_list=[mono_audio_sample], non_matching_references=[mono_audio_sample] - ) diff --git a/src/tests/audio/tasks/forced_alignment_test.py b/src/tests/audio/tasks/forced_alignment_test.py index 53f8c9b3..882da2f2 100644 --- a/src/tests/audio/tasks/forced_alignment_test.py +++ b/src/tests/audio/tasks/forced_alignment_test.py @@ -1,7 +1,5 @@ """Tests for forced alignment functions.""" -import os - import numpy as np import pandas as pd import pytest @@ -102,106 +100,111 @@ def test_interpolate_nans() -> None: assert interpolated_series.isnull().sum() == 0 -if os.getenv("GITHUB_ACTIONS") != "true": +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_get_prediction_matrix(dummy_model: tuple) -> None: + """Test generation of prediction matrix.""" + model, _ = dummy_model + waveform_segment = torch.randn(1, 16000) + prediction_matrix = _get_prediction_matrix(model, waveform_segment, None, "huggingface", DeviceType.CPU) + assert prediction_matrix.shape[0] > 0 - def test_get_prediction_matrix(dummy_model: tuple) -> None: - """Test generation of prediction matrix.""" - model, _ = dummy_model - waveform_segment = torch.randn(1, 16000) - prediction_matrix = _get_prediction_matrix(model, waveform_segment, None, "huggingface", DeviceType.CPU) - assert prediction_matrix.shape[0] > 0 - def test_align_segments(mono_audio_sample: Audio, dummy_model: tuple) -> None: - """Test alignment of segments.""" - model, processor = dummy_model - model_dictionary = processor.tokenizer.get_vocab() +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_align_segments(mono_audio_sample: Audio, dummy_model: tuple) -> None: + """Test alignment of segments.""" + model, processor = dummy_model + model_dictionary = processor.tokenizer.get_vocab() - # Create a sample transcript - transcript = [SingleSegment(start=0.0, end=1.0, text="test")] + # Create a sample transcript + transcript = [SingleSegment(start=0.0, end=1.0, text="test")] - # Preprocess the transcript segments - preprocessed_transcript = _preprocess_segments( - transcript, - model_dictionary, - model_lang=Language(language_code="en"), - print_progress=False, - combined_progress=False, - ) + # Preprocess the transcript segments + preprocessed_transcript = _preprocess_segments( + transcript, + model_dictionary, + model_lang=Language(language_code="en"), + print_progress=False, + combined_progress=False, + ) - # Ensure the model dictionary has the necessary keys - for char in "test": - if char not in model_dictionary: - model_dictionary[char] = len(model_dictionary) - - aligned_segments, word_segments = _align_segments( - transcript=preprocessed_transcript, - model=model, - model_dictionary=model_dictionary, - model_lang=Language(language_code="en"), - model_type="huggingface", - audio=mono_audio_sample, - device=DeviceType.CPU, - max_duration=10.0, - return_char_alignments=False, - interpolate_method="nearest", - ) - assert isinstance(aligned_segments, list) - assert isinstance(word_segments, list) - - def test_align_transcription_faked(resampled_mono_audio_sample: Audio, dummy_model: tuple) -> None: - """Test alignment of transcription.""" - model, processor = dummy_model - transcript = [ - SingleSegment( - start=0.0, - end=1.0, - text="test", - clean_char=["t", "e", "s", "t"], - clean_cdx=[0, 1, 2, 3], - clean_wdx=[0], - sentence_spans=None, - ) - ] - aligned_result = _align_transcription( - transcript=transcript, - model=model, - align_model_metadata={ - "dictionary": processor.tokenizer.get_vocab(), - "language": Language(language_code="en"), - "type": "huggingface", - }, - audio=resampled_mono_audio_sample, - device=DeviceType.CPU, + # Ensure the model dictionary has the necessary keys + for char in "test": + if char not in model_dictionary: + model_dictionary[char] = len(model_dictionary) + + aligned_segments, word_segments = _align_segments( + transcript=preprocessed_transcript, + model=model, + model_dictionary=model_dictionary, + model_lang=Language(language_code="en"), + model_type="huggingface", + audio=mono_audio_sample, + device=DeviceType.CPU, + max_duration=10.0, + return_char_alignments=False, + interpolate_method="nearest", + ) + assert isinstance(aligned_segments, list) + assert isinstance(word_segments, list) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_align_transcription_faked(resampled_mono_audio_sample: Audio, dummy_model: tuple) -> None: + """Test alignment of transcription.""" + model, processor = dummy_model + transcript = [ + SingleSegment( + start=0.0, + end=1.0, + text="test", + clean_char=["t", "e", "s", "t"], + clean_cdx=[0, 1, 2, 3], + clean_wdx=[0], + sentence_spans=None, ) - assert "segments" in aligned_result - assert "word_segments" in aligned_result - - def test_align_transcriptions_fixture(resampled_mono_audio_sample: Audio, script_line_fixture: ScriptLine) -> None: - """Test alignment of transcriptions.""" + ] + aligned_result = _align_transcription( + transcript=transcript, + model=model, + align_model_metadata={ + "dictionary": processor.tokenizer.get_vocab(), + "language": Language(language_code="en"), + "type": "huggingface", + }, + audio=resampled_mono_audio_sample, + device=DeviceType.CPU, + ) + assert "segments" in aligned_result + assert "word_segments" in aligned_result + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_align_transcriptions_fixture(resampled_mono_audio_sample: Audio, script_line_fixture: ScriptLine) -> None: + """Test alignment of transcriptions.""" + audios_and_transcriptions_and_language = [ + (resampled_mono_audio_sample, script_line_fixture, Language(language_code="en")), + (resampled_mono_audio_sample, script_line_fixture, Language(language_code="fr")), + ] + aligned_transcriptions = align_transcriptions(audios_and_transcriptions_and_language) + assert len(aligned_transcriptions) == 2 + assert len(aligned_transcriptions[0]) == 1 + assert aligned_transcriptions[0][0].text == "test" + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_align_transcriptions_multilingual(resampled_mono_audio_sample: Audio, script_line_fixture: ScriptLine) -> None: + """Test alignment of transcriptions.""" + languages = ["de", "es"] + expected_text = "test" # Replace with the appropriate expected text for your fixtures + + for lang in languages: audios_and_transcriptions_and_language = [ - (resampled_mono_audio_sample, script_line_fixture, Language(language_code="en")), - (resampled_mono_audio_sample, script_line_fixture, Language(language_code="fr")), + (resampled_mono_audio_sample, script_line_fixture, Language(language_code=lang)) ] aligned_transcriptions = align_transcriptions(audios_and_transcriptions_and_language) - assert len(aligned_transcriptions) == 2 - assert len(aligned_transcriptions[0]) == 1 - assert aligned_transcriptions[0][0].text == "test" - - def test_align_transcriptions_multilingual( - resampled_mono_audio_sample: Audio, script_line_fixture: ScriptLine - ) -> None: - """Test alignment of transcriptions.""" - languages = ["de", "es"] - expected_text = "test" # Replace with the appropriate expected text for your fixtures - - for lang in languages: - audios_and_transcriptions_and_language = [ - (resampled_mono_audio_sample, script_line_fixture, Language(language_code=lang)) - ] - aligned_transcriptions = align_transcriptions(audios_and_transcriptions_and_language) - assert len(aligned_transcriptions) == 1, f"Failed for language: {lang}" - assert len(aligned_transcriptions[0]) == 1, f"Failed for language: {lang}" - assert aligned_transcriptions[0][0].text == expected_text, f"Failed for language: {lang}" + assert len(aligned_transcriptions) == 1, f"Failed for language: {lang}" + assert len(aligned_transcriptions[0]) == 1, f"Failed for language: {lang}" + assert aligned_transcriptions[0][0].text == expected_text, f"Failed for language: {lang}" if __name__ == "__main__": diff --git a/src/tests/audio/tasks/speaker_diarization_test.py b/src/tests/audio/tasks/speaker_diarization_test.py index 48d5b07e..b0b5f6c8 100644 --- a/src/tests/audio/tasks/speaker_diarization_test.py +++ b/src/tests/audio/tasks/speaker_diarization_test.py @@ -1,60 +1,68 @@ """Tests for speaker diarization.""" -import os - -if os.getenv("GITHUB_ACTIONS") != "true": - import pytest - - from senselab.audio.data_structures.audio import Audio - from senselab.audio.tasks.speaker_diarization.api import diarize_audios - from senselab.audio.tasks.speaker_diarization.pyannote import PyannoteDiarization, diarize_audios_with_pyannote - from senselab.utils.data_structures.device import DeviceType - from senselab.utils.data_structures.model import PyannoteAudioModel - from senselab.utils.data_structures.script_line import ScriptLine - - @pytest.fixture - def pyannote_model() -> PyannoteAudioModel: - """Fixture for Pyannote model.""" - return PyannoteAudioModel(path_or_uri="pyannote/speaker-diarization-3.1") - - def test_diarize_audios(resampled_mono_audio_sample: Audio, pyannote_model: PyannoteAudioModel) -> None: - """Test diarizing audios.""" - results = diarize_audios(audios=[resampled_mono_audio_sample], model=pyannote_model) - assert len(results) == 1 - assert isinstance(results[0][0], ScriptLine) - - def test_diarize_audios_with_pyannote( - resampled_mono_audio_sample: Audio, pyannote_model: PyannoteAudioModel - ) -> None: - """Test diarizing audios with Pyannote.""" - results = diarize_audios_with_pyannote( - audios=[resampled_mono_audio_sample], model=pyannote_model, device=DeviceType.CPU, num_speakers=2 - ) - assert len(results) == 1 - assert isinstance(results[0][0], ScriptLine) - - def test_pyannote_pipeline_factory(pyannote_model: PyannoteAudioModel) -> None: - """Test Pyannote pipeline factory.""" - pipeline1 = PyannoteDiarization._get_pyannote_diarization_pipeline( - model=pyannote_model, - device=DeviceType.CPU, - ) - pipeline2 = PyannoteDiarization._get_pyannote_diarization_pipeline( - model=pyannote_model, - device=DeviceType.CPU, - ) - assert pipeline1 is pipeline2 # Check if the same instance is returned - - def test_diarize_audios_with_pyannote_invalid_sampling_rate( - mono_audio_sample: Audio, pyannote_model: PyannoteAudioModel - ) -> None: - """Test diarizing audios with unsupported sampling_rate.""" - with pytest.raises(ValueError): - diarize_audios(audios=[mono_audio_sample], model=pyannote_model) - - def test_diarize_stereo_audios_with_pyannote_invalid( - resampled_stereo_audio_sample: Audio, pyannote_model: PyannoteAudioModel - ) -> None: - """Test diarizing audios with unsupported number of channels.""" - with pytest.raises(ValueError): - diarize_audios(audios=[resampled_stereo_audio_sample], model=pyannote_model) +import pytest +import torch + +from senselab.audio.data_structures.audio import Audio +from senselab.audio.tasks.speaker_diarization.api import diarize_audios +from senselab.audio.tasks.speaker_diarization.pyannote import PyannoteDiarization, diarize_audios_with_pyannote +from senselab.utils.data_structures.device import DeviceType +from senselab.utils.data_structures.model import PyannoteAudioModel +from senselab.utils.data_structures.script_line import ScriptLine + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def pyannote_model() -> PyannoteAudioModel: + """Fixture for Pyannote model.""" + return PyannoteAudioModel(path_or_uri="pyannote/speaker-diarization-3.1") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_diarize_audios(resampled_mono_audio_sample: Audio, pyannote_model: PyannoteAudioModel) -> None: + """Test diarizing audios.""" + results = diarize_audios(audios=[resampled_mono_audio_sample], model=pyannote_model) + assert len(results) == 1 + assert isinstance(results[0][0], ScriptLine) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_diarize_audios_with_pyannote(resampled_mono_audio_sample: Audio, pyannote_model: PyannoteAudioModel) -> None: + """Test diarizing audios with Pyannote.""" + results = diarize_audios_with_pyannote( + audios=[resampled_mono_audio_sample], model=pyannote_model, device=DeviceType.CPU, num_speakers=2 + ) + assert len(results) == 1 + assert isinstance(results[0][0], ScriptLine) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_pyannote_pipeline_factory(pyannote_model: PyannoteAudioModel) -> None: + """Test Pyannote pipeline factory.""" + pipeline1 = PyannoteDiarization._get_pyannote_diarization_pipeline( + model=pyannote_model, + device=DeviceType.CPU, + ) + pipeline2 = PyannoteDiarization._get_pyannote_diarization_pipeline( + model=pyannote_model, + device=DeviceType.CPU, + ) + assert pipeline1 is pipeline2 # Check if the same instance is returned + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_diarize_audios_with_pyannote_invalid_sampling_rate( + mono_audio_sample: Audio, pyannote_model: PyannoteAudioModel +) -> None: + """Test diarizing audios with unsupported sampling_rate.""" + with pytest.raises(ValueError): + diarize_audios(audios=[mono_audio_sample], model=pyannote_model) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_diarize_stereo_audios_with_pyannote_invalid( + resampled_stereo_audio_sample: Audio, pyannote_model: PyannoteAudioModel +) -> None: + """Test diarizing audios with unsupported number of channels.""" + with pytest.raises(ValueError): + diarize_audios(audios=[resampled_stereo_audio_sample], model=pyannote_model) diff --git a/src/tests/audio/tasks/speaker_embeddings_test.py b/src/tests/audio/tasks/speaker_embeddings_test.py index 52258549..d16a4fb5 100644 --- a/src/tests/audio/tasks/speaker_embeddings_test.py +++ b/src/tests/audio/tasks/speaker_embeddings_test.py @@ -1,145 +1,161 @@ """Tests for speaker_embeddings.py.""" -import os - -if os.getenv("GITHUB_ACTIONS") != "true": - import pytest - from torch import Tensor - - from senselab.audio.data_structures.audio import Audio - from senselab.audio.tasks.speaker_embeddings.api import extract_speaker_embeddings_from_audios - from senselab.utils.data_structures.model import SenselabModel, SpeechBrainModel - - @pytest.fixture - def ecapa_model() -> SpeechBrainModel: - """Fixture for the ECAPA-TDNN model.""" - return SpeechBrainModel(path_or_uri="speechbrain/spkrec-ecapa-voxceleb", revision="main") - - @pytest.fixture - def xvector_model() -> SpeechBrainModel: - """Fixture for the xvector model.""" - return SpeechBrainModel(path_or_uri="speechbrain/spkrec-xvect-voxceleb", revision="main") - - @pytest.fixture - def resnet_model() -> SpeechBrainModel: - """Fixture for the ResNet model.""" - return SpeechBrainModel(path_or_uri="speechbrain/spkrec-resnet-voxceleb", revision="main") - - def test_extract_speaker_embeddings_from_audio( - resampled_mono_audio_sample: Audio, - ecapa_model: SpeechBrainModel, - xvector_model: SpeechBrainModel, - resnet_model: SpeechBrainModel, - ) -> None: - """Test extracting speaker embeddings from audio.""" - embeddings = extract_speaker_embeddings_from_audios(audios=[resampled_mono_audio_sample], model=ecapa_model) - assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) - assert all(embedding.size(0) == 192 for embedding in embeddings) - - embeddings = extract_speaker_embeddings_from_audios(audios=[resampled_mono_audio_sample], model=xvector_model) - assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) - assert all(embedding.size(0) == 512 for embedding in embeddings) - - embeddings = extract_speaker_embeddings_from_audios(audios=[resampled_mono_audio_sample], model=resnet_model) - assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) - assert all(embedding.size(0) == 256 for embedding in embeddings) - - def test_extract_speaker_embeddings_from_multiple_audios( - resampled_mono_audio_sample: Audio, - ecapa_model: SpeechBrainModel, - xvector_model: SpeechBrainModel, - resnet_model: SpeechBrainModel, - ) -> None: - """Test extracting speaker embeddings from multiple audios.""" - embeddings = extract_speaker_embeddings_from_audios( - audios=[resampled_mono_audio_sample, resampled_mono_audio_sample], model=ecapa_model +import pytest +import torch +from torch import Tensor + +from senselab.audio.data_structures.audio import Audio +from senselab.audio.tasks.speaker_embeddings.api import extract_speaker_embeddings_from_audios +from senselab.utils.data_structures.model import SenselabModel, SpeechBrainModel + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def ecapa_model() -> SpeechBrainModel: + """Fixture for the ECAPA-TDNN model.""" + return SpeechBrainModel(path_or_uri="speechbrain/spkrec-ecapa-voxceleb", revision="main") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def xvector_model() -> SpeechBrainModel: + """Fixture for the xvector model.""" + return SpeechBrainModel(path_or_uri="speechbrain/spkrec-xvect-voxceleb", revision="main") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def resnet_model() -> SpeechBrainModel: + """Fixture for the ResNet model.""" + return SpeechBrainModel(path_or_uri="speechbrain/spkrec-resnet-voxceleb", revision="main") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_extract_speaker_embeddings_from_audio( + resampled_mono_audio_sample: Audio, + ecapa_model: SpeechBrainModel, + xvector_model: SpeechBrainModel, + resnet_model: SpeechBrainModel, +) -> None: + """Test extracting speaker embeddings from audio.""" + embeddings = extract_speaker_embeddings_from_audios(audios=[resampled_mono_audio_sample], model=ecapa_model) + assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) + assert all(embedding.size(0) == 192 for embedding in embeddings) + + embeddings = extract_speaker_embeddings_from_audios(audios=[resampled_mono_audio_sample], model=xvector_model) + assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) + assert all(embedding.size(0) == 512 for embedding in embeddings) + + embeddings = extract_speaker_embeddings_from_audios(audios=[resampled_mono_audio_sample], model=resnet_model) + assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) + assert all(embedding.size(0) == 256 for embedding in embeddings) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_extract_speaker_embeddings_from_multiple_audios( + resampled_mono_audio_sample: Audio, + ecapa_model: SpeechBrainModel, + xvector_model: SpeechBrainModel, + resnet_model: SpeechBrainModel, +) -> None: + """Test extracting speaker embeddings from multiple audios.""" + embeddings = extract_speaker_embeddings_from_audios( + audios=[resampled_mono_audio_sample, resampled_mono_audio_sample], model=ecapa_model + ) + assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) + assert all(embedding.size(0) == 192 for embedding in embeddings) + + embeddings = extract_speaker_embeddings_from_audios( + audios=[resampled_mono_audio_sample, resampled_mono_audio_sample], model=xvector_model + ) + assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) + assert all(embedding.size(0) == 512 for embedding in embeddings) + + embeddings = extract_speaker_embeddings_from_audios( + audios=[resampled_mono_audio_sample, resampled_mono_audio_sample], model=resnet_model + ) + assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) + assert all(embedding.size(0) == 256 for embedding in embeddings) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_extract_speaker_embeddings_from_multiple_audios_different_sizes( + resampled_mono_audio_sample: Audio, + resampled_mono_audio_sample_x2: Audio, + ecapa_model: SpeechBrainModel, + xvector_model: SpeechBrainModel, + resnet_model: SpeechBrainModel, +) -> None: + """Test extracting speaker embeddings from multiple audios of differing lengths.""" + embeddings = extract_speaker_embeddings_from_audios( + audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], model=ecapa_model + ) + assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) + assert all(embedding.size(0) == 192 for embedding in embeddings) + + embeddings = extract_speaker_embeddings_from_audios( + audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], model=xvector_model + ) + assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) + assert all(embedding.size(0) == 512 for embedding in embeddings) + + embeddings = extract_speaker_embeddings_from_audios( + audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], model=resnet_model + ) + assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) + assert all(embedding.size(0) == 256 for embedding in embeddings) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_error_wrong_model(resampled_mono_audio_sample: Audio) -> None: + """Test raising error when using a non-existent model.""" + with pytest.raises(ValueError): + extract_speaker_embeddings_from_audios( + audios=[resampled_mono_audio_sample], model=SpeechBrainModel(path_or_uri="nonexistent---") ) - assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) - assert all(embedding.size(0) == 192 for embedding in embeddings) - - embeddings = extract_speaker_embeddings_from_audios( - audios=[resampled_mono_audio_sample, resampled_mono_audio_sample], model=xvector_model - ) - assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) - assert all(embedding.size(0) == 512 for embedding in embeddings) - - embeddings = extract_speaker_embeddings_from_audios( - audios=[resampled_mono_audio_sample, resampled_mono_audio_sample], model=resnet_model + with pytest.raises(NotImplementedError): + extract_speaker_embeddings_from_audios( + audios=[resampled_mono_audio_sample], model=SenselabModel(path_or_uri="nonexistent---") ) - assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) - assert all(embedding.size(0) == 256 for embedding in embeddings) - - def test_extract_speaker_embeddings_from_multiple_audios_different_sizes( - resampled_mono_audio_sample: Audio, - resampled_mono_audio_sample_x2: Audio, - ecapa_model: SpeechBrainModel, - xvector_model: SpeechBrainModel, - resnet_model: SpeechBrainModel, - ) -> None: - """Test extracting speaker embeddings from multiple audios of differing lengths.""" - embeddings = extract_speaker_embeddings_from_audios( - audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], model=ecapa_model - ) - assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) - assert all(embedding.size(0) == 192 for embedding in embeddings) - embeddings = extract_speaker_embeddings_from_audios( - audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], model=xvector_model - ) - assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) - assert all(embedding.size(0) == 512 for embedding in embeddings) - embeddings = extract_speaker_embeddings_from_audios( - audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], model=resnet_model - ) - assert isinstance(embeddings, list) and all(isinstance(embedding, Tensor) for embedding in embeddings) - assert all(embedding.size(0) == 256 for embedding in embeddings) - - def test_error_wrong_model(resampled_mono_audio_sample: Audio) -> None: - """Test raising error when using a non-existent model.""" - with pytest.raises(ValueError): - extract_speaker_embeddings_from_audios( - audios=[resampled_mono_audio_sample], model=SpeechBrainModel(path_or_uri="nonexistent---") - ) - with pytest.raises(NotImplementedError): - extract_speaker_embeddings_from_audios( - audios=[resampled_mono_audio_sample], model=SenselabModel(path_or_uri="nonexistent---") - ) - - def test_extract_speechbrain_speaker_embeddings_from_audio_resampled( - mono_audio_sample: Audio, - ecapa_model: SpeechBrainModel, - xvector_model: SpeechBrainModel, - resnet_model: SpeechBrainModel, - ) -> None: - """Test extracting speaker embeddings from audio.""" - # Testing with the ecapa model - with pytest.raises(ValueError): - extract_speaker_embeddings_from_audios(audios=[mono_audio_sample], model=ecapa_model) - - # Testing with the xvector model - with pytest.raises(ValueError): - extract_speaker_embeddings_from_audios(audios=[mono_audio_sample], model=xvector_model) - - # Testing with the resnet model - with pytest.raises(ValueError): - extract_speaker_embeddings_from_audios(audios=[mono_audio_sample], model=resnet_model) - - def test_extract_speechbrain_speaker_embeddings_from_stereo_audio( - stereo_audio_sample: Audio, - ecapa_model: SpeechBrainModel, - xvector_model: SpeechBrainModel, - resnet_model: SpeechBrainModel, - ) -> None: - """Test extracting speaker embeddings from audio.""" - # Testing with the ecapa model - with pytest.raises(ValueError): - extract_speaker_embeddings_from_audios(audios=[stereo_audio_sample], model=ecapa_model) - - # Testing with the xvector model - with pytest.raises(ValueError): - extract_speaker_embeddings_from_audios(audios=[stereo_audio_sample], model=xvector_model) - - # Testing with the resnet model - with pytest.raises(ValueError): - extract_speaker_embeddings_from_audios(audios=[stereo_audio_sample], model=resnet_model) +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_extract_speechbrain_speaker_embeddings_from_audio_resampled( + mono_audio_sample: Audio, + ecapa_model: SpeechBrainModel, + xvector_model: SpeechBrainModel, + resnet_model: SpeechBrainModel, +) -> None: + """Test extracting speaker embeddings from audio.""" + # Testing with the ecapa model + with pytest.raises(ValueError): + extract_speaker_embeddings_from_audios(audios=[mono_audio_sample], model=ecapa_model) + + # Testing with the xvector model + with pytest.raises(ValueError): + extract_speaker_embeddings_from_audios(audios=[mono_audio_sample], model=xvector_model) + + # Testing with the resnet model + with pytest.raises(ValueError): + extract_speaker_embeddings_from_audios(audios=[mono_audio_sample], model=resnet_model) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_extract_speechbrain_speaker_embeddings_from_stereo_audio( + stereo_audio_sample: Audio, + ecapa_model: SpeechBrainModel, + xvector_model: SpeechBrainModel, + resnet_model: SpeechBrainModel, +) -> None: + """Test extracting speaker embeddings from audio.""" + # Testing with the ecapa model + with pytest.raises(ValueError): + extract_speaker_embeddings_from_audios(audios=[stereo_audio_sample], model=ecapa_model) + + # Testing with the xvector model + with pytest.raises(ValueError): + extract_speaker_embeddings_from_audios(audios=[stereo_audio_sample], model=xvector_model) + + # Testing with the resnet model + with pytest.raises(ValueError): + extract_speaker_embeddings_from_audios(audios=[stereo_audio_sample], model=resnet_model) diff --git a/src/tests/audio/tasks/speaker_verification_test.py b/src/tests/audio/tasks/speaker_verification_test.py index f42ccd38..761fc700 100644 --- a/src/tests/audio/tasks/speaker_verification_test.py +++ b/src/tests/audio/tasks/speaker_verification_test.py @@ -8,9 +8,8 @@ - test_verify_speaker_from_files: Tests the verify_speaker_from_files function. """ -import os - import pytest +import torch from senselab.audio.data_structures.audio import Audio from senselab.audio.tasks.preprocessing.preprocessing import resample_audios @@ -18,23 +17,23 @@ verify_speaker, ) -if os.getenv("GITHUB_ACTIONS") != "true": - - @pytest.mark.large_model - def test_verify_speaker(mono_audio_sample: Audio) -> None: - """Tests the verify_speaker function to ensure it does not fail. - - Args: - mono_audio_sample (Audio): The mono audio sample to use for testing. - - Returns: - None - """ - mono_audio_sample = resample_audios([mono_audio_sample], 16000)[0] - assert mono_audio_sample.sampling_rate == 16000 - mono_audio_samples = [(mono_audio_sample, mono_audio_sample)] * 3 - scores_and_predictions = verify_speaker(mono_audio_samples) - assert scores_and_predictions - assert len(scores_and_predictions[0]) == 2 - assert isinstance(scores_and_predictions[0][0], float) - assert isinstance(scores_and_predictions[0][1], bool) + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.mark.large_model +def test_verify_speaker(mono_audio_sample: Audio) -> None: + """Tests the verify_speaker function to ensure it does not fail. + + Args: + mono_audio_sample (Audio): The mono audio sample to use for testing. + + Returns: + None + """ + mono_audio_sample = resample_audios([mono_audio_sample], 16000)[0] + assert mono_audio_sample.sampling_rate == 16000 + mono_audio_samples = [(mono_audio_sample, mono_audio_sample)] * 3 + scores_and_predictions = verify_speaker(mono_audio_samples) + assert scores_and_predictions + assert len(scores_and_predictions[0]) == 2 + assert isinstance(scores_and_predictions[0][0], float) + assert isinstance(scores_and_predictions[0][1], bool) diff --git a/src/tests/audio/tasks/speech_enhancement_test.py b/src/tests/audio/tasks/speech_enhancement_test.py index 38e9cf07..ef282b7c 100644 --- a/src/tests/audio/tasks/speech_enhancement_test.py +++ b/src/tests/audio/tasks/speech_enhancement_test.py @@ -1,101 +1,121 @@ """Tests for the speech enhancement task.""" -import os from typing import List -if os.getenv("GITHUB_ACTIONS") != "true": - import pytest - from speechbrain.inference.separation import SepformerSeparation as separator - - from senselab.audio.data_structures.audio import Audio - from senselab.audio.tasks.speech_enhancement.api import enhance_audios - from senselab.audio.tasks.speech_enhancement.speechbrain import SpeechBrainEnhancer - from senselab.utils.data_structures.device import DeviceType - from senselab.utils.data_structures.model import SpeechBrainModel - - @pytest.fixture - def speechbrain_model() -> SpeechBrainModel: - """Fixture for Hugging Face model.""" - return SpeechBrainModel(path_or_uri="speechbrain/sepformer-wham16k-enhancement") - - def test_enhance_audios_stereo_audio( - resampled_stereo_audio_sample: Audio, speechbrain_model: SpeechBrainModel - ) -> None: - """Test that enhancing stereo audios raises a ValueError.""" - with pytest.raises(ValueError, match="Audio waveform must be mono"): - SpeechBrainEnhancer.enhance_audios_with_speechbrain( - audios=[resampled_stereo_audio_sample], model=speechbrain_model - ) - - def test_enhance_audios( - resampled_mono_audio_sample: Audio, resampled_mono_audio_sample_x2: Audio, speechbrain_model: SpeechBrainModel - ) -> None: - """Test enhancing audios.""" - enhanced_audios = enhance_audios( - audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], model=speechbrain_model - ) - assert len(enhanced_audios) == 2 - assert isinstance(enhanced_audios[0], Audio) - assert enhanced_audios[0].waveform.shape == resampled_mono_audio_sample.waveform.shape - - def test_speechbrain_enhancer_get_model(speechbrain_model: SpeechBrainModel) -> None: - """Test getting SpeechBrain model.""" - # TODO: add tests like these but with multithreading - model, _, _ = SpeechBrainEnhancer._get_speechbrain_model(model=speechbrain_model, device=DeviceType.CPU) - assert model is not None - assert isinstance(model, separator) - assert ( - model - == SpeechBrainEnhancer._models[ - f"{speechbrain_model.path_or_uri}-{speechbrain_model.revision}-{DeviceType.CPU.value}" - ] - ) +import pytest +import torch +from speechbrain.inference.separation import SepformerSeparation as separator + +from senselab.audio.data_structures.audio import Audio +from senselab.audio.tasks.speech_enhancement.api import enhance_audios +from senselab.audio.tasks.speech_enhancement.speechbrain import SpeechBrainEnhancer +from senselab.utils.data_structures.device import DeviceType +from senselab.utils.data_structures.model import SpeechBrainModel + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def speechbrain_model() -> SpeechBrainModel: + """Fixture for Hugging Face model.""" + return SpeechBrainModel(path_or_uri="speechbrain/sepformer-wham16k-enhancement") + - def test_enhance_audios_with_speechbrain( - resampled_mono_audio_sample: Audio, resampled_mono_audio_sample_x2: Audio, speechbrain_model: SpeechBrainModel - ) -> None: - """Test enhancing audios with SpeechBrain.""" - enhanced_audios = SpeechBrainEnhancer.enhance_audios_with_speechbrain( - audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], model=speechbrain_model +@pytest.fixture(autouse=True) +def clear_cache() -> None: + """Fixture for clearing the cached models between pytest runs.""" + SpeechBrainEnhancer._models = {} + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_enhance_audios_stereo_audio(resampled_stereo_audio_sample: Audio, speechbrain_model: SpeechBrainModel) -> None: + """Test that enhancing stereo audios raises a ValueError.""" + with pytest.raises(ValueError, match="Audio waveform must be mono"): + SpeechBrainEnhancer.enhance_audios_with_speechbrain( + audios=[resampled_stereo_audio_sample], model=speechbrain_model ) - assert len(enhanced_audios) == 2 - assert isinstance(enhanced_audios[0], Audio) - assert enhanced_audios[0].waveform.shape == resampled_mono_audio_sample.waveform.shape - assert enhanced_audios[1].waveform.shape == resampled_mono_audio_sample_x2.waveform.shape - - def test_enhance_audios_incorrect_sampling_rate( - mono_audio_sample: Audio, speechbrain_model: SpeechBrainModel - ) -> None: - """Test enhancing audios with incorrect sampling rate.""" - mono_audio_sample.sampling_rate = 8000 # Incorrect sample rate for this model - with pytest.raises(ValueError, match="Audio sampling rate 8000 does not match expected 16000"): - SpeechBrainEnhancer.enhance_audios_with_speechbrain(audios=[mono_audio_sample], model=speechbrain_model) - - def test_enhance_audios_with_different_bit_depths(audio_with_different_bit_depths: List[Audio]) -> None: - """Test enhancing audios with different bit depths.""" - enhanced_audios = SpeechBrainEnhancer.enhance_audios_with_speechbrain(audios=audio_with_different_bit_depths) - assert len(enhanced_audios) == 2 - for audio in enhanced_audios: - assert isinstance(audio, Audio) - assert audio.waveform.shape == audio_with_different_bit_depths[0].waveform.shape - - def test_enhance_audios_with_metadata(audio_with_metadata: Audio) -> None: - """Test enhancing audios with metadata.""" - enhanced_audios = SpeechBrainEnhancer.enhance_audios_with_speechbrain(audios=[audio_with_metadata]) - assert len(enhanced_audios) == 1 - assert isinstance(enhanced_audios[0], Audio) - assert enhanced_audios[0].metadata == audio_with_metadata.metadata - - def test_enhance_audios_with_extreme_amplitude(audio_with_extreme_amplitude: Audio) -> None: - """Test enhancing audios with extreme amplitude values.""" - enhanced_audios = SpeechBrainEnhancer.enhance_audios_with_speechbrain(audios=[audio_with_extreme_amplitude]) - assert len(enhanced_audios) == 1 - assert isinstance(enhanced_audios[0], Audio) - assert enhanced_audios[0].waveform.shape == audio_with_extreme_amplitude.waveform.shape - - def test_model_caching(resampled_mono_audio_sample: Audio) -> None: - """Test model caching by enhancing audios with the same model multiple times.""" - SpeechBrainEnhancer.enhance_audios_with_speechbrain(audios=[resampled_mono_audio_sample]) - assert len(SpeechBrainEnhancer._models) == 1 - SpeechBrainEnhancer.enhance_audios_with_speechbrain(audios=[resampled_mono_audio_sample]) - assert len(SpeechBrainEnhancer._models) == 1 + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_enhance_audios( + resampled_mono_audio_sample: Audio, resampled_mono_audio_sample_x2: Audio, speechbrain_model: SpeechBrainModel +) -> None: + """Test enhancing audios.""" + enhanced_audios = enhance_audios( + audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], model=speechbrain_model + ) + assert len(enhanced_audios) == 2 + assert isinstance(enhanced_audios[0], Audio) + assert enhanced_audios[0].waveform.shape == resampled_mono_audio_sample.waveform.shape + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_speechbrain_enhancer_get_model(speechbrain_model: SpeechBrainModel) -> None: + """Test getting SpeechBrain model.""" + # TODO: add tests like these but with multithreading + model, _, _ = SpeechBrainEnhancer._get_speechbrain_model(model=speechbrain_model, device=DeviceType.CPU) + assert model is not None + assert isinstance(model, separator) + assert ( + model + == SpeechBrainEnhancer._models[ + f"{speechbrain_model.path_or_uri}-{speechbrain_model.revision}-{DeviceType.CPU.value}" + ] + ) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_enhance_audios_with_speechbrain( + resampled_mono_audio_sample: Audio, resampled_mono_audio_sample_x2: Audio, speechbrain_model: SpeechBrainModel +) -> None: + """Test enhancing audios with SpeechBrain.""" + enhanced_audios = SpeechBrainEnhancer.enhance_audios_with_speechbrain( + audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], model=speechbrain_model + ) + assert len(enhanced_audios) == 2 + assert isinstance(enhanced_audios[0], Audio) + assert enhanced_audios[0].waveform.shape == resampled_mono_audio_sample.waveform.shape + assert enhanced_audios[1].waveform.shape == resampled_mono_audio_sample_x2.waveform.shape + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_enhance_audios_incorrect_sampling_rate(mono_audio_sample: Audio, speechbrain_model: SpeechBrainModel) -> None: + """Test enhancing audios with incorrect sampling rate.""" + mono_audio_sample.sampling_rate = 8000 # Incorrect sample rate for this model + with pytest.raises(ValueError, match="Audio sampling rate 8000 does not match expected 16000"): + SpeechBrainEnhancer.enhance_audios_with_speechbrain(audios=[mono_audio_sample], model=speechbrain_model) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_enhance_audios_with_different_bit_depths(audio_with_different_bit_depths: List[Audio]) -> None: + """Test enhancing audios with different bit depths.""" + enhanced_audios = SpeechBrainEnhancer.enhance_audios_with_speechbrain(audios=audio_with_different_bit_depths) + assert len(enhanced_audios) == 2 + for audio in enhanced_audios: + assert isinstance(audio, Audio) + assert audio.waveform.shape == audio_with_different_bit_depths[0].waveform.shape + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_enhance_audios_with_metadata(audio_with_metadata: Audio) -> None: + """Test enhancing audios with metadata.""" + enhanced_audios = SpeechBrainEnhancer.enhance_audios_with_speechbrain(audios=[audio_with_metadata]) + assert len(enhanced_audios) == 1 + assert isinstance(enhanced_audios[0], Audio) + assert enhanced_audios[0].metadata == audio_with_metadata.metadata + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_enhance_audios_with_extreme_amplitude(audio_with_extreme_amplitude: Audio) -> None: + """Test enhancing audios with extreme amplitude values.""" + enhanced_audios = SpeechBrainEnhancer.enhance_audios_with_speechbrain(audios=[audio_with_extreme_amplitude]) + assert len(enhanced_audios) == 1 + assert isinstance(enhanced_audios[0], Audio) + assert enhanced_audios[0].waveform.shape == audio_with_extreme_amplitude.waveform.shape + + +def test_model_caching(resampled_mono_audio_sample: Audio) -> None: + """Test model caching by enhancing audios with the same model multiple times.""" + SpeechBrainEnhancer.enhance_audios_with_speechbrain(audios=[resampled_mono_audio_sample], device=DeviceType.CPU) + assert len(list(SpeechBrainEnhancer._models.keys())) == 1 + SpeechBrainEnhancer.enhance_audios_with_speechbrain(audios=[resampled_mono_audio_sample], device=DeviceType.CPU) + assert len(list(SpeechBrainEnhancer._models.keys())) == 1 diff --git a/src/tests/audio/tasks/speech_to_text_test.py b/src/tests/audio/tasks/speech_to_text_test.py index cb0c26ad..16104786 100644 --- a/src/tests/audio/tasks/speech_to_text_test.py +++ b/src/tests/audio/tasks/speech_to_text_test.py @@ -1,10 +1,16 @@ """Tests for the speech to text task.""" -import os from typing import Callable import pytest - +import torch + +from senselab.audio.data_structures.audio import Audio +from senselab.audio.tasks.speech_to_text import transcribe_audios +from senselab.audio.tasks.speech_to_text.huggingface import HuggingFaceASR +from senselab.utils.data_structures.device import DeviceType +from senselab.utils.data_structures.language import Language +from senselab.utils.data_structures.model import HFModel from senselab.utils.data_structures.script_line import ScriptLine @@ -28,98 +34,105 @@ def test_scriptline_from_dict() -> None: assert scriptline.chunks[1].get_timestamps()[1] == 2.0 -if os.getenv("GITHUB_ACTIONS") != "true": - from senselab.audio.data_structures.audio import Audio - from senselab.audio.tasks.speech_to_text import transcribe_audios - from senselab.audio.tasks.speech_to_text.huggingface import HuggingFaceASR - from senselab.utils.data_structures.device import DeviceType - from senselab.utils.data_structures.language import Language - from senselab.utils.data_structures.model import HFModel - - @pytest.fixture - def hf_model() -> HFModel: - """Fixture for Hugging Face model.""" - return HFModel(path_or_uri="openai/whisper-tiny") +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def hf_model() -> HFModel: + """Fixture for Hugging Face model.""" + return HFModel(path_or_uri="openai/whisper-tiny") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def hf_model2() -> HFModel: + """Fixture for Hugging Face model.""" + return HFModel(path_or_uri="facebook/seamless-m4t-unity-small") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.mark.parametrize("device", [DeviceType.CPU, DeviceType.CUDA]) # MPS is not available for now +def test_hf_asr_pipeline_factory(hf_model: HFModel, device: DeviceType, is_device_available: Callable) -> None: + """Test ASR pipeline factory.""" + if not is_device_available(device): + pytest.skip(f"{device} is not available") + + pipeline1 = HuggingFaceASR._get_hf_asr_pipeline( + model=hf_model, + return_timestamps="word", + max_new_tokens=128, + chunk_length_s=30, + batch_size=1, + device=device, + ) + pipeline2 = HuggingFaceASR._get_hf_asr_pipeline( + model=hf_model, + return_timestamps="word", + max_new_tokens=128, + chunk_length_s=30, + batch_size=1, + device=device, + ) + assert pipeline1 is pipeline2 # Check if the same instance is returned (this is the case for serial execution) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.mark.parametrize("hf_model", ["hf_model", "hf_model2"], indirect=True) +def test_transcribe_audios( + resampled_mono_audio_sample: Audio, resampled_mono_audio_sample_x2: Audio, hf_model: HFModel +) -> None: + """Test transcribing audios.""" + transcripts = transcribe_audios( + audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], model=hf_model + ) + assert len(transcripts) == 2 + assert isinstance(transcripts[0], ScriptLine) + assert ( + transcripts[0].text + == "This is Peter. This is Johnny. Kenny. And Joe. We just wanted to take a minute to thank you." + ) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.mark.parametrize("hf_model", ["hf_model", "hf_model2"], indirect=True) +def test_transcribe_audios_with_params( + resampled_mono_audio_sample: Audio, resampled_mono_audio_sample_x2: Audio, hf_model: HFModel +) -> None: + """Test transcribing audios.""" + transcripts = transcribe_audios( + audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], + model=hf_model, + language=Language(language_code="English"), + return_timestamps=False, + ) + assert len(transcripts) == 2 + assert isinstance(transcripts[0], ScriptLine) + # Note: we don't check the transcript because we have noticed that by specifying the language, + # the transcript is not correct with our sample audio + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_transcribe_audios_with_unsupported_params( + resampled_mono_audio_sample: Audio, resampled_mono_audio_sample_x2: Audio, hf_model: HFModel +) -> None: + """Test transcribing audios with an unsupported param.""" + with pytest.raises(TypeError, match="got an unexpected keyword argument"): + transcribe_audios( + audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], + model=hf_model, + unsupported_param="unsupported_param", + ) - @pytest.fixture - def hf_model2() -> HFModel: - """Fixture for Hugging Face model.""" - return HFModel(path_or_uri="facebook/seamless-m4t-unity-small") - @pytest.mark.parametrize("device", [DeviceType.CPU, DeviceType.CUDA]) # MPS is not available for now - def test_hf_asr_pipeline_factory(hf_model: HFModel, device: DeviceType, is_device_available: Callable) -> None: - """Test ASR pipeline factory.""" - if not is_device_available(device): - pytest.skip(f"{device} is not available") +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_transcribe_stereo_audio(resampled_stereo_audio_sample: Audio, hf_model: HFModel) -> None: + """Test transcribing stereo audio.""" + # Create a mock stereo audio sample + with pytest.raises(ValueError, match="Stereo audio is not supported"): + transcribe_audios(audios=[resampled_stereo_audio_sample], model=hf_model) - pipeline1 = HuggingFaceASR._get_hf_asr_pipeline( - model=hf_model, - return_timestamps="word", - max_new_tokens=128, - chunk_length_s=30, - batch_size=1, - device=device, - ) - pipeline2 = HuggingFaceASR._get_hf_asr_pipeline( - model=hf_model, - return_timestamps="word", - max_new_tokens=128, - chunk_length_s=30, - batch_size=1, - device=device, - ) - assert pipeline1 is pipeline2 # Check if the same instance is returned (this is the case for serial execution) - - @pytest.mark.parametrize("hf_model", ["hf_model", "hf_model2"], indirect=True) - def test_transcribe_audios( - resampled_mono_audio_sample: Audio, resampled_mono_audio_sample_x2: Audio, hf_model: HFModel - ) -> None: - """Test transcribing audios.""" - transcripts = transcribe_audios( - audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], model=hf_model - ) - assert len(transcripts) == 2 - assert isinstance(transcripts[0], ScriptLine) - assert ( - transcripts[0].text - == "This is Peter. This is Johnny. Kenny. And Joe. We just wanted to take a minute to thank you." - ) - @pytest.mark.parametrize("hf_model", ["hf_model", "hf_model2"], indirect=True) - def test_transcribe_audios_with_params( - resampled_mono_audio_sample: Audio, resampled_mono_audio_sample_x2: Audio, hf_model: HFModel - ) -> None: - """Test transcribing audios.""" - transcripts = transcribe_audios( - audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], - model=hf_model, - language=Language(language_code="English"), - return_timestamps=False, - ) - assert len(transcripts) == 2 - assert isinstance(transcripts[0], ScriptLine) - # Note: we don't check the transcript because we have noticed that by specifying the language, - # the transcript is not correct with our sample audio - - def test_transcribe_audios_with_unsupported_params( - resampled_mono_audio_sample: Audio, resampled_mono_audio_sample_x2: Audio, hf_model: HFModel - ) -> None: - """Test transcribing audios with an unsupported param.""" - with pytest.raises(TypeError, match="got an unexpected keyword argument"): - transcribe_audios( - audios=[resampled_mono_audio_sample, resampled_mono_audio_sample_x2], - model=hf_model, - unsupported_param="unsupported_param", - ) - - def test_transcribe_stereo_audio(resampled_stereo_audio_sample: Audio, hf_model: HFModel) -> None: - """Test transcribing stereo audio.""" - # Create a mock stereo audio sample - with pytest.raises(ValueError, match="Stereo audio is not supported"): - transcribe_audios(audios=[resampled_stereo_audio_sample], model=hf_model) - - def test_transcribe_audio_with_wrong_sampling_rate(mono_audio_sample: Audio, hf_model: HFModel) -> None: - """Test transcribing stereo audio.""" - # Create a mock stereo audio sample - with pytest.raises(ValueError, match="Incorrect sampling rate."): - transcribe_audios(audios=[mono_audio_sample], model=hf_model) +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_transcribe_audio_with_wrong_sampling_rate(mono_audio_sample: Audio, hf_model: HFModel) -> None: + """Test transcribing stereo audio.""" + # Create a mock stereo audio sample + with pytest.raises(ValueError, match="Incorrect sampling rate."): + transcribe_audios(audios=[mono_audio_sample], model=hf_model) diff --git a/src/tests/audio/tasks/text_to_speech_test.py b/src/tests/audio/tasks/text_to_speech_test.py index 8e94f0f4..7ab50b7b 100644 --- a/src/tests/audio/tasks/text_to_speech_test.py +++ b/src/tests/audio/tasks/text_to_speech_test.py @@ -1,115 +1,129 @@ """Tests for the text to speech task.""" -import os - -if os.getenv("GITHUB_ACTIONS") != "true": - from typing import Callable - - import pytest - - from senselab.audio.data_structures.audio import Audio - from senselab.audio.tasks.preprocessing.preprocessing import extract_segments, resample_audios - from senselab.audio.tasks.text_to_speech.api import HuggingFaceTTS, synthesize_texts - from senselab.utils.data_structures.device import DeviceType - from senselab.utils.data_structures.language import Language - from senselab.utils.data_structures.model import HFModel, SenselabModel, TorchModel - - @pytest.fixture - def hf_model() -> HFModel: - """Fixture for the HF model.""" - return HFModel(path_or_uri="suno/bark-small", revision="main") - - @pytest.fixture - def hf_model2() -> HFModel: - """Fixture for HF model.""" - return HFModel(path_or_uri="facebook/mms-tts-eng", revision="main") - - @pytest.fixture - def mars5_model() -> TorchModel: - """Fixture for MARS5 model.""" - return TorchModel(path_or_uri="Camb-ai/mars5-tts", revision="master") - - @pytest.fixture - def style_tts2() -> TorchModel: - """Fixture for StyleTTS2 model.""" - return TorchModel(path_or_uri="wilke0818/StyleTTS2-TorchHub", revision="main") - - @pytest.mark.parametrize("hf_model", ["hf_model", "hf_model2"], indirect=True) - def test_synthesize_texts_with_hf_model(hf_model: HFModel) -> None: - """Test synthesizing texts.""" - texts = ["Hello world", "Hello world again."] - audios = synthesize_texts(texts=texts, model=hf_model) - - assert len(audios) == 2 - assert isinstance(audios[0], Audio) - assert audios[0].waveform is not None - assert audios[0].sampling_rate > 0 - - # TODO: create support for StyleTTS2 which currently has some dependency issues - # def test_synthesize_texts_with_styletts2_model(style_tts2: TorchModel, mono_audio_sample: Audio) -> None: - # """Test synthesizing texts.""" - # texts_to_synthesize = ["Hello world", "Hello world again."] - # terget_audio_resampling_rate = 24000 - # target_audio_ground_truth = "This is Peter." - # language = Language(language_code="en") - - # resampled_mono_audio_sample = resample_audios([mono_audio_sample], terget_audio_resampling_rate)[0] - # target_audio = extract_segments([(resampled_mono_audio_sample, [(0.0, 1.0)])])[0][0] - # audios = synthesize_texts( - # texts=texts_to_synthesize, - # target=[(target_audio, target_audio_ground_truth), (target_audio, target_audio_ground_truth)], - # model=style_tts2, - # language=language, - # force_reload=True, - # ) - - # assert len(audios) == 2 - # assert isinstance(audios[0], Audio) - # assert audios[0].waveform is not None - # assert audios[0].sampling_rate == terget_audio_resampling_rate - - def test_synthesize_texts_with_mars5_model(mars5_model: TorchModel, mono_audio_sample: Audio) -> None: - """Test synthesizing texts.""" - texts_to_synthesize = ["Hello world", "Hello world again."] - terget_audio_resampling_rate = 24000 - target_audio_ground_truth = "This is Peter." - language = Language(language_code="en") - - resampled_mono_audio_sample = resample_audios([mono_audio_sample], terget_audio_resampling_rate)[0] - target_audio = extract_segments([(resampled_mono_audio_sample, [(0.0, 1.0)])])[0][0] - audios = synthesize_texts( - texts=texts_to_synthesize, - targets=[(target_audio, target_audio_ground_truth), (target_audio, target_audio_ground_truth)], - model=mars5_model, - language=language, - ) - - assert len(audios) == 2 - assert isinstance(audios[0], Audio) - assert audios[0].waveform is not None - assert audios[0].sampling_rate == terget_audio_resampling_rate - - @pytest.mark.parametrize("device", [DeviceType.CPU, DeviceType.CUDA]) # MPS is not available for now - def test_huggingface_tts_pipeline_factory( - hf_model: HFModel, device: DeviceType, is_device_available: Callable - ) -> None: - """Test Hugging Face TTS pipeline factory.""" - if not is_device_available(device): - pytest.skip(f"{device} is not available") - - pipeline1 = HuggingFaceTTS._get_hf_tts_pipeline(model=hf_model, device=device) - pipeline2 = HuggingFaceTTS._get_hf_tts_pipeline(model=hf_model, device=device) - - assert pipeline1 is pipeline2 # Check if the same instance is returned - - def test_invalid_model() -> None: - """Test synthesize_texts with invalid model.""" - texts = ["Hello world"] - model = SenselabModel(path_or_uri="-----", revision="main") - - # TODO Texts like these should be stored in a common utils/constants file such that - # they only need to be changed in one place - with pytest.raises( - NotImplementedError, match="Only Hugging Face models and select Torch models are supported for now." - ): - synthesize_texts(texts=texts, model=model) +from typing import Callable + +import pytest +import torch + +from senselab.audio.data_structures.audio import Audio +from senselab.audio.tasks.preprocessing.preprocessing import extract_segments, resample_audios +from senselab.audio.tasks.text_to_speech.api import HuggingFaceTTS, synthesize_texts +from senselab.utils.data_structures.device import DeviceType +from senselab.utils.data_structures.language import Language +from senselab.utils.data_structures.model import HFModel, SenselabModel, TorchModel + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def hf_model() -> HFModel: + """Fixture for the HF model.""" + return HFModel(path_or_uri="suno/bark-small", revision="main") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def hf_model2() -> HFModel: + """Fixture for HF model.""" + return HFModel(path_or_uri="facebook/mms-tts-eng", revision="main") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def mars5_model() -> TorchModel: + """Fixture for MARS5 model.""" + return TorchModel(path_or_uri="Camb-ai/mars5-tts", revision="master") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def style_tts2() -> TorchModel: + """Fixture for StyleTTS2 model.""" + return TorchModel(path_or_uri="wilke0818/StyleTTS2-TorchHub", revision="main") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.mark.parametrize("hf_model", ["hf_model", "hf_model2"], indirect=True) +def test_synthesize_texts_with_hf_model(hf_model: HFModel) -> None: + """Test synthesizing texts.""" + texts = ["Hello world", "Hello world again."] + audios = synthesize_texts(texts=texts, model=hf_model, device=DeviceType.CUDA) + + assert len(audios) == 2 + assert isinstance(audios[0], Audio) + assert audios[0].waveform is not None + assert audios[0].sampling_rate > 0 + + +# TODO: create support for StyleTTS2 which currently has some dependency issues +# def test_synthesize_texts_with_styletts2_model(style_tts2: TorchModel, mono_audio_sample: Audio) -> None: +# """Test synthesizing texts.""" +# texts_to_synthesize = ["Hello world", "Hello world again."] +# terget_audio_resampling_rate = 24000 +# target_audio_ground_truth = "This is Peter." +# language = Language(language_code="en") + +# resampled_mono_audio_sample = resample_audios([mono_audio_sample], terget_audio_resampling_rate)[0] +# target_audio = extract_segments([(resampled_mono_audio_sample, [(0.0, 1.0)])])[0][0] +# audios = synthesize_texts( +# texts=texts_to_synthesize, +# target=[(target_audio, target_audio_ground_truth), (target_audio, target_audio_ground_truth)], +# model=style_tts2, +# language=language, +# force_reload=True, +# ) + +# assert len(audios) == 2 +# assert isinstance(audios[0], Audio) +# assert audios[0].waveform is not None +# assert audios[0].sampling_rate == terget_audio_resampling_rate + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_synthesize_texts_with_mars5_model(mars5_model: TorchModel, mono_audio_sample: Audio) -> None: + """Test synthesizing texts.""" + texts_to_synthesize = ["Hello world", "Hello world again."] + terget_audio_resampling_rate = 24000 + target_audio_ground_truth = "This is Peter." + language = Language(language_code="en") + + resampled_mono_audio_sample = resample_audios([mono_audio_sample], terget_audio_resampling_rate)[0] + target_audio = extract_segments([(resampled_mono_audio_sample, [(0.0, 1.0)])])[0][0] + audios = synthesize_texts( + texts=texts_to_synthesize, + targets=[(target_audio, target_audio_ground_truth), (target_audio, target_audio_ground_truth)], + model=mars5_model, + language=language, + device=DeviceType.CUDA, + ) + + assert len(audios) == 2 + assert isinstance(audios[0], Audio) + assert audios[0].waveform is not None + assert audios[0].sampling_rate == terget_audio_resampling_rate + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.mark.parametrize("device", [DeviceType.CPU, DeviceType.CUDA]) # MPS is not available for now +def test_huggingface_tts_pipeline_factory(hf_model: HFModel, device: DeviceType, is_device_available: Callable) -> None: + """Test Hugging Face TTS pipeline factory.""" + if not is_device_available(device): + pytest.skip(f"{device} is not available") + + pipeline1 = HuggingFaceTTS._get_hf_tts_pipeline(model=hf_model, device=device) + pipeline2 = HuggingFaceTTS._get_hf_tts_pipeline(model=hf_model, device=device) + + assert pipeline1 is pipeline2 # Check if the same instance is returned + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_invalid_model() -> None: + """Test synthesize_texts with invalid model.""" + texts = ["Hello world"] + model = SenselabModel(path_or_uri="-----", revision="main") + + # TODO Texts like these should be stored in a common utils/constants file such that + # they only need to be changed in one place + with pytest.raises( + NotImplementedError, match="Only Hugging Face models and select Torch models are supported for now." + ): + synthesize_texts(texts=texts, model=model) diff --git a/src/tests/audio/tasks/voice_activity_detection_test.py b/src/tests/audio/tasks/voice_activity_detection_test.py index 2b9c1ced..7869d579 100644 --- a/src/tests/audio/tasks/voice_activity_detection_test.py +++ b/src/tests/audio/tasks/voice_activity_detection_test.py @@ -1,12 +1,11 @@ """Tests for voice activity detection.""" -import os - import pytest +import torch from senselab.audio.data_structures.audio import Audio from senselab.audio.tasks.voice_activity_detection.api import detect_human_voice_activity_in_audios -from senselab.utils.data_structures.model import SenselabModel +from senselab.utils.data_structures.model import PyannoteAudioModel, SenselabModel def test_detect_human_voice_activity_in_audios_with_invalid_model(mono_audio_sample: Audio) -> None: @@ -17,18 +16,18 @@ def test_detect_human_voice_activity_in_audios_with_invalid_model(mono_audio_sam ) -if os.getenv("GITHUB_ACTIONS") != "true": - from senselab.utils.data_structures.model import PyannoteAudioModel +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def pyannote_model() -> PyannoteAudioModel: + """Fixture for Pyannote model.""" + return PyannoteAudioModel(path_or_uri="pyannote/speaker-diarization-3.1") - @pytest.fixture - def pyannote_model() -> PyannoteAudioModel: - """Fixture for Pyannote model.""" - return PyannoteAudioModel(path_or_uri="pyannote/speaker-diarization-3.1") - def test_detect_human_voice_activity_in_audios( - resampled_mono_audio_sample: Audio, pyannote_model: PyannoteAudioModel - ) -> None: - """Test detecting human voice activity in audios.""" - results = detect_human_voice_activity_in_audios(audios=[resampled_mono_audio_sample], model=pyannote_model) - assert len(results) == 1 - assert all(chunk.speaker == "VOICE" for chunk in results[0]) +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_detect_human_voice_activity_in_audios( + resampled_mono_audio_sample: Audio, pyannote_model: PyannoteAudioModel +) -> None: + """Test detecting human voice activity in audios.""" + results = detect_human_voice_activity_in_audios(audios=[resampled_mono_audio_sample], model=pyannote_model) + assert len(results) == 1 + assert all(chunk.speaker == "VOICE" for chunk in results[0]) diff --git a/src/tests/audio/tasks/voice_cloning_test.py b/src/tests/audio/tasks/voice_cloning_test.py index 6bbfeb64..6e6cb53b 100644 --- a/src/tests/audio/tasks/voice_cloning_test.py +++ b/src/tests/audio/tasks/voice_cloning_test.py @@ -1,120 +1,132 @@ """This script is for testing the voice cloning API.""" -import os - -if os.getenv("GITHUB_ACTIONS") != "true": - import pytest - - from senselab.audio.data_structures.audio import Audio - from senselab.audio.tasks.voice_cloning.api import clone_voices - from senselab.utils.data_structures.device import DeviceType - from senselab.utils.data_structures.model import TorchModel - - @pytest.fixture - def torch_model() -> TorchModel: - """Fixture for torch model.""" - return TorchModel(path_or_uri="bshall/knn-vc", revision="master") - - def test_clone_voices_length_mismatch(resampled_mono_audio_sample: Audio, torch_model: TorchModel) -> None: - """Test length mismatch in source and target audios.""" - source_audios = [resampled_mono_audio_sample] - target_audios = [resampled_mono_audio_sample, resampled_mono_audio_sample] - - with pytest.raises(ValueError, match="The list of source and target audios must have the same length"): - clone_voices( - source_audios=source_audios, target_audios=target_audios, model=torch_model, device=DeviceType.CPU - ) - - def test_clone_voices_invalid_topk(resampled_mono_audio_sample: Audio, torch_model: TorchModel) -> None: - """Test invalid topk value.""" - source_audios = [resampled_mono_audio_sample] - target_audios = [resampled_mono_audio_sample] - - with pytest.raises(TypeError, match="argument 'k' must be int, not str"): - clone_voices( - source_audios=source_audios, - target_audios=target_audios, - model=torch_model, - device=DeviceType.CPU, - topk="invalid", # type: ignore[arg-type] - ) - - def test_clone_voices_invalid_prematched_vocoder( - resampled_mono_audio_sample: Audio, torch_model: TorchModel - ) -> None: - """Test invalid prematched_vocoder value.""" - source_audios = [resampled_mono_audio_sample] - target_audios = [resampled_mono_audio_sample] - - with pytest.raises(TypeError, match="prematched_vocoder must be a boolean."): - clone_voices( - source_audios=source_audios, - target_audios=target_audios, - model=torch_model, - device=DeviceType.CPU, - prematched_vocoder="invalid", # type: ignore[arg-type] - ) - - def test_clone_voices_valid_input(resampled_mono_audio_sample: Audio, torch_model: TorchModel) -> None: - """Test cloning voices with valid input.""" - source_audios = [resampled_mono_audio_sample, resampled_mono_audio_sample] - target_audios = [resampled_mono_audio_sample, resampled_mono_audio_sample] - - try: - cloned_output = clone_voices( - source_audios=source_audios, - target_audios=target_audios, - model=torch_model, - device=DeviceType.CPU, - topk=5, - prematched_vocoder=False, - ) - assert isinstance(cloned_output, list), "Output must be a list." - assert len(cloned_output) == 2, "Output list should contain exactly two audio samples." - assert isinstance(cloned_output[0], Audio), "Each item in the output list should be an instance of Audio." - source_duration = source_audios[0].waveform.shape[1] - cloned_duration = cloned_output[0].waveform.shape[1] - - # Set tolerance to 1% of source duration - tolerance = 0.01 * source_duration - - # Check if the absolute difference is within the tolerance - assert abs(source_duration - cloned_duration) <= tolerance, ( - f"Cloned audio duration is not within acceptable range. Source: {source_duration}, " - f"Cloned: {cloned_duration}" - ) - - except Exception as e: - pytest.fail(f"An unexpected exception occurred: {e}") - - def test_clone_voices_unsupported_model(resampled_mono_audio_sample: Audio) -> None: - """Test unsupported model.""" - source_audios = [resampled_mono_audio_sample] - target_audios = [resampled_mono_audio_sample] - # this uri doesn't exist - unsupported_model = TorchModel(path_or_uri="sensein/senselab", revision="main") - - with pytest.raises(NotImplementedError, match="Only KNNVC is supported for now."): - clone_voices( - source_audios=source_audios, target_audios=target_audios, model=unsupported_model, device=DeviceType.CPU - ) - - def test_clone_voices_stereo_audio(resampled_stereo_audio_sample: Audio, torch_model: TorchModel) -> None: - """Test unsupported stereo audio.""" - source_audios = [resampled_stereo_audio_sample] - target_audios = [resampled_stereo_audio_sample] - - with pytest.raises(ValueError, match="Only mono audio files are supported."): - clone_voices( - source_audios=source_audios, target_audios=target_audios, model=torch_model, device=DeviceType.CPU - ) - - def test_clone_voices_invalid_sampling_rate(mono_audio_sample: Audio, torch_model: TorchModel) -> None: - """Test unsupported sampling rate.""" - source_audios = [mono_audio_sample] - target_audios = [mono_audio_sample] - - with pytest.raises(ValueError, match="Only 16000 sampling rate is supported."): - clone_voices( - source_audios=source_audios, target_audios=target_audios, model=torch_model, device=DeviceType.CPU - ) +import pytest +import torch + +from senselab.audio.data_structures.audio import Audio +from senselab.audio.tasks.voice_cloning.api import clone_voices +from senselab.utils.data_structures.device import DeviceType +from senselab.utils.data_structures.model import TorchModel + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def torch_model() -> TorchModel: + """Fixture for torch model.""" + return TorchModel(path_or_uri="bshall/knn-vc", revision="master") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_clone_voices_length_mismatch(resampled_mono_audio_sample: Audio, torch_model: TorchModel) -> None: + """Test length mismatch in source and target audios.""" + source_audios = [resampled_mono_audio_sample] + target_audios = [resampled_mono_audio_sample, resampled_mono_audio_sample] + + with pytest.raises(ValueError, match="The list of source and target audios must have the same length"): + clone_voices( + source_audios=source_audios, target_audios=target_audios, model=torch_model, device=DeviceType.CUDA + ) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_clone_voices_invalid_topk(resampled_mono_audio_sample: Audio, torch_model: TorchModel) -> None: + """Test invalid topk value.""" + source_audios = [resampled_mono_audio_sample] + target_audios = [resampled_mono_audio_sample] + + with pytest.raises(TypeError, match="argument 'k' must be int, not str"): + clone_voices( + source_audios=source_audios, + target_audios=target_audios, + model=torch_model, + device=DeviceType.CUDA, + topk="invalid", # type: ignore[arg-type] + ) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_clone_voices_invalid_prematched_vocoder(resampled_mono_audio_sample: Audio, torch_model: TorchModel) -> None: + """Test invalid prematched_vocoder value.""" + source_audios = [resampled_mono_audio_sample] + target_audios = [resampled_mono_audio_sample] + + with pytest.raises(TypeError, match="prematched_vocoder must be a boolean."): + clone_voices( + source_audios=source_audios, + target_audios=target_audios, + model=torch_model, + device=DeviceType.CUDA, + prematched_vocoder="invalid", # type: ignore[arg-type] + ) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_clone_voices_valid_input(resampled_mono_audio_sample: Audio, torch_model: TorchModel) -> None: + """Test cloning voices with valid input.""" + source_audios = [resampled_mono_audio_sample, resampled_mono_audio_sample] + target_audios = [resampled_mono_audio_sample, resampled_mono_audio_sample] + + try: + cloned_output = clone_voices( + source_audios=source_audios, + target_audios=target_audios, + model=torch_model, + device=DeviceType.CUDA, + topk=5, + prematched_vocoder=False, + ) + assert isinstance(cloned_output, list), "Output must be a list." + assert len(cloned_output) == 2, "Output list should contain exactly two audio samples." + assert isinstance(cloned_output[0], Audio), "Each item in the output list should be an instance of Audio." + source_duration = source_audios[0].waveform.shape[1] + cloned_duration = cloned_output[0].waveform.shape[1] + + # Set tolerance to 1% of source duration + tolerance = 0.01 * source_duration + + # Check if the absolute difference is within the tolerance + assert abs(source_duration - cloned_duration) <= tolerance, ( + f"Cloned audio duration is not within acceptable range. Source: {source_duration}, " + f"Cloned: {cloned_duration}" + ) + + except Exception as e: + pytest.fail(f"An unexpected exception occurred: {e}") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_clone_voices_unsupported_model(resampled_mono_audio_sample: Audio) -> None: + """Test unsupported model.""" + source_audios = [resampled_mono_audio_sample] + target_audios = [resampled_mono_audio_sample] + # this uri doesn't exist + unsupported_model = TorchModel(path_or_uri="sensein/senselab", revision="main") + + with pytest.raises(NotImplementedError, match="Only KNNVC is supported for now."): + clone_voices( + source_audios=source_audios, target_audios=target_audios, model=unsupported_model, device=DeviceType.CUDA + ) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_clone_voices_stereo_audio(resampled_stereo_audio_sample: Audio, torch_model: TorchModel) -> None: + """Test unsupported stereo audio.""" + source_audios = [resampled_stereo_audio_sample] + target_audios = [resampled_stereo_audio_sample] + + with pytest.raises(ValueError, match="Only mono audio files are supported."): + clone_voices( + source_audios=source_audios, target_audios=target_audios, model=torch_model, device=DeviceType.CUDA + ) + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_clone_voices_invalid_sampling_rate(mono_audio_sample: Audio, torch_model: TorchModel) -> None: + """Test unsupported sampling rate.""" + source_audios = [mono_audio_sample] + target_audios = [mono_audio_sample] + + with pytest.raises(ValueError, match="Only 16000 sampling rate is supported."): + clone_voices( + source_audios=source_audios, target_audios=target_audios, model=torch_model, device=DeviceType.CUDA + ) diff --git a/src/tests/audio/workflows/transcribe_timestamped_test.py b/src/tests/audio/workflows/transcribe_timestamped_test.py index 0ee028bf..e07f8ed2 100644 --- a/src/tests/audio/workflows/transcribe_timestamped_test.py +++ b/src/tests/audio/workflows/transcribe_timestamped_test.py @@ -1,79 +1,75 @@ """Tests the transcribe_timestamped module.""" -import os +''' +# TODO: Please double-check this because tests are failing +from senselab.audio.data_structures.audio import Audio +from senselab.audio.workflows.transcribe_timestamped import transcribe_timestamped -if os.getenv("GITHUB_ACTIONS") != "true": - from senselab.audio.data_structures.audio import Audio - from senselab.audio.workflows.transcribe_timestamped import transcribe_timestamped - def test_transcribe_timestamped_mono(mono_audio_sample: Audio) -> None: - """Runs the transcribe_timestamped function.""" - assert transcribe_timestamped(audios=[mono_audio_sample]) +def test_transcribe_timestamped_mono(mono_audio_sample: Audio) -> None: + """Runs the transcribe_timestamped function.""" + assert transcribe_timestamped(audios=[mono_audio_sample]) - def test_transcribe_timestamped_stereo(stereo_audio_sample: Audio) -> None: - """Test transcribe_timestamped with a stereo audio sample.""" - result = transcribe_timestamped(audios=[stereo_audio_sample]) - assert isinstance(result, list), "The result should be a list of ScriptLine lists." - assert len(result) > 0, "The result should not be empty." - assert all( - isinstance(script_lines, list) for script_lines in result - ), "Each item in the result should be a list." - assert all( - len(script_lines) > 0 for script_lines in result - ), "Each list in the result should contain ScriptLine objects." - def test_transcribe_timestamped_resampled_mono( - resampled_mono_audio_sample: Audio, - ) -> None: - """Test transcribe_timestamped with a resampled mono audio sample.""" - result = transcribe_timestamped(audios=[resampled_mono_audio_sample]) - assert isinstance(result, list), "The result should be a list of ScriptLine lists." - assert len(result) > 0, "The result should not be empty." - assert all( - isinstance(script_lines, list) for script_lines in result - ), "Each item in the result should be a list." - assert all( - len(script_lines) > 0 for script_lines in result - ), "Each list in the result should contain ScriptLine objects." +def test_transcribe_timestamped_stereo(stereo_audio_sample: Audio) -> None: + """Test transcribe_timestamped with a stereo audio sample.""" + result = transcribe_timestamped(audios=[stereo_audio_sample]) + assert isinstance(result, list), "The result should be a list of ScriptLine lists." + assert len(result) > 0, "The result should not be empty." + assert all(isinstance(script_lines, list) for script_lines in result), "Each item in the result should be a list." + assert all( + len(script_lines) > 0 for script_lines in result + ), "Each list in the result should contain ScriptLine objects." - def test_transcribe_timestamped_resampled_stereo( - resampled_stereo_audio_sample: Audio, - ) -> None: - """Test transcribe_timestamped with a resampled stereo audio sample.""" - result = transcribe_timestamped(audios=[resampled_stereo_audio_sample]) - assert isinstance(result, list), "The result should be a list of ScriptLine lists." - assert len(result) > 0, "The result should not be empty." - assert all( - isinstance(script_lines, list) for script_lines in result - ), "Each item in the result should be a list." - assert all( - len(script_lines) > 0 for script_lines in result - ), "Each list in the result should contain ScriptLine objects." - def test_transcribe_timestamped_noise(audio_with_metadata: Audio) -> None: - """Test transcribe_timestamped with a noisy audio sample.""" - result = transcribe_timestamped(audios=[audio_with_metadata]) - assert isinstance(result, list), "The result should be a list of ScriptLine lists." - assert len(result) > 0, "The result should not be empty." - assert all( - isinstance(script_lines, list) for script_lines in result - ), "Each item in the result should be a list." - assert all( - len(script_lines) > 0 for script_lines in result - ), "Each list in the result should contain ScriptLine objects." +def test_transcribe_timestamped_resampled_mono( + resampled_mono_audio_sample: Audio, +) -> None: + """Test transcribe_timestamped with a resampled mono audio sample.""" + result = transcribe_timestamped(audios=[resampled_mono_audio_sample]) + assert isinstance(result, list), "The result should be a list of ScriptLine lists." + assert len(result) > 0, "The result should not be empty." + assert all(isinstance(script_lines, list) for script_lines in result), "Each item in the result should be a list." + assert all( + len(script_lines) > 0 for script_lines in result + ), "Each list in the result should contain ScriptLine objects." - def test_transcribe_timestamped_different_bit_depths( - audio_with_different_bit_depths: list[Audio], - ) -> None: - """Test transcribe_timestamped with audio samples of different bit depths.""" - result = transcribe_timestamped(audios=audio_with_different_bit_depths) - assert isinstance(result, list), "The result should be a list of ScriptLine lists." - assert len(result) == len( - audio_with_different_bit_depths - ), "The result should have the same number of elements as the input audio." - assert all( - isinstance(script_lines, list) for script_lines in result - ), "Each item in the result should be a list." - assert all( - len(script_lines) > 0 for script_lines in result - ), "Each list in the result should contain ScriptLine objects." + +def test_transcribe_timestamped_resampled_stereo( + resampled_stereo_audio_sample: Audio, +) -> None: + """Test transcribe_timestamped with a resampled stereo audio sample.""" + result = transcribe_timestamped(audios=[resampled_stereo_audio_sample]) + assert isinstance(result, list), "The result should be a list of ScriptLine lists." + assert len(result) > 0, "The result should not be empty." + assert all(isinstance(script_lines, list) for script_lines in result), "Each item in the result should be a list." + assert all( + len(script_lines) > 0 for script_lines in result + ), "Each list in the result should contain ScriptLine objects." + + +def test_transcribe_timestamped_noise(audio_with_metadata: Audio) -> None: + """Test transcribe_timestamped with a noisy audio sample.""" + result = transcribe_timestamped(audios=[audio_with_metadata]) + assert isinstance(result, list), "The result should be a list of ScriptLine lists." + assert len(result) > 0, "The result should not be empty." + assert all(isinstance(script_lines, list) for script_lines in result), "Each item in the result should be a list." + assert all( + len(script_lines) > 0 for script_lines in result + ), "Each list in the result should contain ScriptLine objects." + + +def test_transcribe_timestamped_different_bit_depths( + audio_with_different_bit_depths: list[Audio], +) -> None: + """Test transcribe_timestamped with audio samples of different bit depths.""" + result = transcribe_timestamped(audios=audio_with_different_bit_depths) + assert isinstance(result, list), "The result should be a list of ScriptLine lists." + assert len(result) == len( + audio_with_different_bit_depths + ), "The result should have the same number of elements as the input audio." + assert all(isinstance(script_lines, list) for script_lines in result), "Each item in the result should be a list." + assert all( + len(script_lines) > 0 for script_lines in result + ), "Each list in the result should contain ScriptLine objects." +''' diff --git a/src/tests/text/tasks/embeddings_extraction_test.py b/src/tests/text/tasks/embeddings_extraction_test.py index 236eda09..7546e16b 100644 --- a/src/tests/text/tasks/embeddings_extraction_test.py +++ b/src/tests/text/tasks/embeddings_extraction_test.py @@ -1,40 +1,45 @@ """This module is for extracting deep learning embeddings from text.""" -import os - -if os.getenv("GITHUB_ACTIONS") != "true": - from typing import List - - import pytest - import torch - - from senselab.text.tasks.embeddings_extraction.api import extract_embeddings_from_text - from senselab.utils.data_structures.model import HFModel, SentenceTransformersModel - - @pytest.fixture - def hf_model() -> HFModel: - """Fixture for our default embeddings extraction Hugging Face model.""" - return HFModel(path_or_uri="sentence-transformers/all-MiniLM-L6-v2", revision="main") - - @pytest.fixture - def sentencetransformers_model() -> SentenceTransformersModel: - """Fixture for our default embeddings extraction SentenceTransformer model.""" - return SentenceTransformersModel(path_or_uri="sentence-transformers/all-MiniLM-L6-v2", revision="main") - - def test_extract_sentencetransformers_embeddings_from_text( - sample_texts: List[str], sentencetransformers_model: SentenceTransformersModel - ) -> None: - """Test extract_embeddings_from_text.""" - embeddings = extract_embeddings_from_text(sample_texts, sentencetransformers_model) - assert isinstance(embeddings, List) - assert embeddings[0].shape == torch.Size([384]) # shape of "sentence-transformers/all-MiniLM-L6-v2" - - def test_extract_huggingface_embeddings_from_text(sample_texts: List[str], hf_model: HFModel) -> None: - """Test extract_embeddings_from_text.""" - embeddings = extract_embeddings_from_text(sample_texts, hf_model) - assert isinstance(embeddings, List) - print(embeddings[0].shape) - # 7 layers for "sentence-transformers/all-MiniLM-L6-v2" (6 is the sequence Length in this case) - assert embeddings[0].shape[0] == 7 - # 384 as Hidden Size for shape of "sentence-transformers/all-MiniLM-L6-v2" - assert embeddings[0].shape[2] == 384 +from typing import List + +import pytest +import torch + +from senselab.text.tasks.embeddings_extraction.api import extract_embeddings_from_text +from senselab.utils.data_structures.model import HFModel, SentenceTransformersModel + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def hf_model() -> HFModel: + """Fixture for our default embeddings extraction Hugging Face model.""" + return HFModel(path_or_uri="sentence-transformers/all-MiniLM-L6-v2", revision="main") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +@pytest.fixture +def sentencetransformers_model() -> SentenceTransformersModel: + """Fixture for our default embeddings extraction SentenceTransformer model.""" + return SentenceTransformersModel(path_or_uri="sentence-transformers/all-MiniLM-L6-v2", revision="main") + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_extract_sentencetransformers_embeddings_from_text( + sample_texts: List[str], sentencetransformers_model: SentenceTransformersModel +) -> None: + """Test extract_embeddings_from_text.""" + embeddings = extract_embeddings_from_text(sample_texts, sentencetransformers_model) + assert isinstance(embeddings, List) + assert embeddings[0].shape == torch.Size([384]) # shape of "sentence-transformers/all-MiniLM-L6-v2" + + +@pytest.mark.skipif(not torch.cuda.is_available(), reason="GPU is not available") +def test_extract_huggingface_embeddings_from_text(sample_texts: List[str], hf_model: HFModel) -> None: + """Test extract_embeddings_from_text.""" + embeddings = extract_embeddings_from_text(sample_texts, hf_model) + assert isinstance(embeddings, List) + print(embeddings[0].shape) + # 7 layers for "sentence-transformers/all-MiniLM-L6-v2" (6 is the sequence Length in this case) + assert embeddings[0].shape[0] == 7 + # 384 as Hidden Size for shape of "sentence-transformers/all-MiniLM-L6-v2" + assert embeddings[0].shape[2] == 384 diff --git a/tutorials/getting_started.ipynb b/tutorials/00_getting_started.ipynb similarity index 99% rename from tutorials/getting_started.ipynb rename to tutorials/00_getting_started.ipynb index d7d8d57f..eac46138 100644 --- a/tutorials/getting_started.ipynb +++ b/tutorials/00_getting_started.ipynb @@ -155,6 +155,7 @@ ], "source": [ "from senselab.audio.tasks.plotting.plotting import play_audio\n", + "\n", "play_audio(audio1)" ] }, @@ -176,6 +177,7 @@ ], "source": [ "from senselab.audio.tasks.plotting.plotting import plot_waveform\n", + "\n", "plot_waveform(audio1)" ] }, @@ -432,6 +434,7 @@ ], "source": [ "from torch_audiomentations import Compose, PolarityInversion\n", + "\n", "from senselab.audio.tasks.data_augmentation.data_augmentation import augment_audios\n", "\n", "apply_augmentation = Compose(transforms=[PolarityInversion(p=1, output_type=\"dict\")], output_type=\"dict\")\n", diff --git a/tutorials/dimensionality_reduction_tutorial.ipynb b/tutorials/dimensionality_reduction.ipynb similarity index 88% rename from tutorials/dimensionality_reduction_tutorial.ipynb rename to tutorials/dimensionality_reduction.ipynb index b638df72..94c818d6 100644 --- a/tutorials/dimensionality_reduction_tutorial.ipynb +++ b/tutorials/dimensionality_reduction.ipynb @@ -6,7 +6,7 @@ "source": [ "# Dimensionality Reduction Tutorial\n", "\n", - "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/sensein/senselab/blob/main/tutorials/dimensionality_reduction_tutorial.ipynb)\n", + "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/sensein/senselab/blob/main/tutorials/dimensionality_reduction.ipynb)\n", "\n", "\n", "In this tutorial, we'll explore how to use dsenselab to emplot dimensionality reduction techniques to visualize high-dimensional data. We'll use the scikit-learn digits dataset as our example, but this can be extended to any high-dimensionality dataset, such as a set of speaker embeddings.\n", @@ -27,12 +27,11 @@ "outputs": [], "source": [ "# %pip install senselab\n", + "import matplotlib.pyplot as plt\n", + "import torch\n", "from sklearn.datasets import load_digits\n", - "from senselab.utils.tasks.dimensionality_reduction import compute_dimensionality_reduction\n", "\n", - "import torch\n", - "import senselab\n", - "import matplotlib.pyplot as plt" + "from senselab.utils.tasks.dimensionality_reduction import compute_dimensionality_reduction" ] }, { @@ -86,7 +85,10 @@ "\n", "# Plot the reduced data\n", "plt.figure(figsize=(12, 8))\n", - "scatter = plt.scatter(reduced_data_pca[:, 0], reduced_data_pca[:, 1], c=digits_target, cmap='tab10')\n", + "scatter = plt.scatter(reduced_data_pca[:, 0], \n", + " reduced_data_pca[:, 1], \n", + " c=digits_target, \n", + " cmap='tab10')\n", "plt.title(\"PCA Analysis of Digits Dataset\")\n", "plt.xlabel(\"PCA Component 1\")\n", "plt.ylabel(\"PCA Component 2\")\n", @@ -115,7 +117,10 @@ "\n", "# Plot the reduced data\n", "plt.figure(figsize=(12, 8))\n", - "scatter = plt.scatter(reduced_data_tsne[:, 0], reduced_data_tsne[:, 1], c=digits_target, cmap='tab10')\n", + "scatter = plt.scatter(reduced_data_tsne[:, 0], \n", + " reduced_data_tsne[:, 1], \n", + " c=digits_target, \n", + " cmap='tab10')\n", "plt.title(\"t-SNE of Digits Dataset\")\n", "plt.xlabel(\"t-SNE Component 1\")\n", "plt.ylabel(\"t-SNE Component 2\")\n", @@ -144,7 +149,10 @@ "\n", "# Plot the reduced data\n", "plt.figure(figsize=(12, 8))\n", - "scatter = plt.scatter(reduced_data_umap[:, 0], reduced_data_umap[:, 1], c=digits_target, cmap='tab10')\n", + "scatter = plt.scatter(reduced_data_umap[:, 0], \n", + " reduced_data_umap[:, 1], \n", + " c=digits_target, \n", + " cmap='tab10')\n", "plt.title(\"UMAP of Digits Dataset\")\n", "plt.xlabel(\"UMAP Component 1\")\n", "plt.ylabel(\"UMAP Component 2\")\n", @@ -168,12 +176,18 @@ "metadata": {}, "outputs": [], "source": [ - "reduced_data_tsne_custom_params = compute_dimensionality_reduction(digits_data, model=\"tsne\", n_components=2, perplexity=5)\n", + "reduced_data_tsne_custom_params = compute_dimensionality_reduction(digits_data, \n", + " model=\"tsne\", \n", + " n_components=2, \n", + " perplexity=5)\n", "\n", "print(\"Reduced data shape:\", reduced_data_tsne_custom_params.shape)\n", "\n", "plt.figure(figsize=(12, 8))\n", - "scatter = plt.scatter(reduced_data_tsne_custom_params[:, 0], reduced_data_tsne_custom_params[:, 1], c=digits_target, cmap='tab10')\n", + "scatter = plt.scatter(reduced_data_tsne_custom_params[:, 0], \n", + " reduced_data_tsne_custom_params[:, 1], \n", + " c=digits_target, \n", + " cmap='tab10')\n", "plt.title(\"t-SNE of Digits Dataset (perplexity=5)\")\n", "plt.xlabel(\"t-SNE Component 1\")\n", "plt.ylabel(\"t-SNE Component 2\")\n", diff --git a/tutorials/extract_speaker_embeddings.ipynb b/tutorials/extract_speaker_embeddings.ipynb index 1f741b4c..cf22ef7e 100644 --- a/tutorials/extract_speaker_embeddings.ipynb +++ b/tutorials/extract_speaker_embeddings.ipynb @@ -6,7 +6,7 @@ "source": [ "# Speaker Embeddings Extraction Tutorial\n", "\n", - "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/sensein/senselab/blob/main/tutorials/extract_speaker_embeddings_tutorial.ipynb)\n", + "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/sensein/senselab/blob/main/tutorials/extract_speaker_embeddings.ipynb)\n", "\n", "\n", "## Introduction\n", @@ -23,16 +23,17 @@ "metadata": {}, "outputs": [], "source": [ - "import torch\n", - "import numpy as np\n", - "import matplotlib.pyplot as plt\n", "from typing import List\n", + "\n", + "import matplotlib.pyplot as plt\n", + "import numpy as np\n", + "import torch\n", + "\n", "from senselab.audio.data_structures.audio import Audio\n", - "from senselab.utils.data_structures.model import SpeechBrainModel\n", - "from senselab.utils.data_structures.device import DeviceType\n", + "from senselab.audio.tasks.preprocessing.preprocessing import downmix_audios_to_mono, resample_audios\n", "from senselab.audio.tasks.speaker_embeddings.api import extract_speaker_embeddings_from_audios\n", - "from senselab.audio.tasks.preprocessing.preprocessing import downmix_audios_to_mono\n", - "from senselab.audio.tasks.preprocessing.preprocessing import resample_audios" + "from senselab.utils.data_structures.device import DeviceType\n", + "from senselab.utils.data_structures.model import SpeechBrainModel" ] }, { @@ -132,8 +133,9 @@ "source": [ "from senselab.utils.tasks.cosine_similarity import cosine_similarity\n", "\n", + "\n", "# DIRECTLY PLOT THE EMBEDDINGS FOR THE TWO FILES\n", - "def plot_embedding_heatmap(embeddings: List[torch.Tensor], titles: List[str]):\n", + "def plot_embedding_heatmap(embeddings: List[torch.Tensor], titles: List[str]) -> None:\n", " \"\"\"Plot a heatmap of a list of speaker embeddings.\"\"\"\n", " fig, axes = plt.subplots(len(embeddings), 1, figsize=(10, 5 * len(embeddings)))\n", " if len(embeddings) == 1:\n", @@ -152,7 +154,7 @@ "\n", "\n", "# PLOT THE COSINE SIMILARITY MATRIX FOR THE TWO FILES\n", - "def plot_similarity_matrix(embeddings: List[torch.Tensor], labels: List[str]):\n", + "def plot_similarity_matrix(embeddings: List[torch.Tensor], labels: List[str]) -> None:\n", " \"\"\"Plot a similarity matrix for a list of embeddings.\"\"\"\n", " n = len(embeddings)\n", " similarity_matrix = np.zeros((n, n))\n", diff --git a/tutorials/speaker_diarization.ipynb b/tutorials/speaker_diarization.ipynb index 0aa676c6..7b860bea 100644 --- a/tutorials/speaker_diarization.ipynb +++ b/tutorials/speaker_diarization.ipynb @@ -19,11 +19,11 @@ "source": [ "# Import necessary modules\n", "from senselab.audio.data_structures.audio import Audio\n", - "from senselab.audio.tasks.speaker_diarization import diarize_audios\n", - "from senselab.utils.data_structures.model import PyannoteAudioModel\n", - "from senselab.utils.data_structures.device import DeviceType\n", "from senselab.audio.tasks.plotting.plotting import play_audio\n", "from senselab.audio.tasks.preprocessing.preprocessing import resample_audios\n", + "from senselab.audio.tasks.speaker_diarization import diarize_audios\n", + "from senselab.utils.data_structures.device import DeviceType\n", + "from senselab.utils.data_structures.model import PyannoteAudioModel\n", "from senselab.utils.tasks.plotting import plot_segment" ] }, diff --git a/tutorials/speech_enhancement.ipynb b/tutorials/speech_enhancement.ipynb index dabac8fc..ff56253d 100644 --- a/tutorials/speech_enhancement.ipynb +++ b/tutorials/speech_enhancement.ipynb @@ -21,10 +21,10 @@ "source": [ "# Import the necessary modules from the Senselab package for audio processing\n", "from senselab.audio.data_structures.audio import Audio\n", - "from senselab.audio.tasks.speech_enhancement.api import enhance_audios\n", - "from senselab.utils.data_structures.device import DeviceType\n", "from senselab.audio.tasks.plotting.plotting import play_audio\n", "from senselab.audio.tasks.preprocessing.preprocessing import resample_audios\n", + "from senselab.audio.tasks.speech_enhancement.api import enhance_audios\n", + "from senselab.utils.data_structures.device import DeviceType\n", "from senselab.utils.data_structures.model import SpeechBrainModel\n" ] }, diff --git a/tutorials/speech_to_text.ipynb b/tutorials/speech_to_text.ipynb index 371837a7..2d6dc904 100644 --- a/tutorials/speech_to_text.ipynb +++ b/tutorials/speech_to_text.ipynb @@ -26,13 +26,12 @@ "outputs": [], "source": [ "from senselab.audio.data_structures.audio import Audio\n", - "from senselab.utils.data_structures.model import HFModel\n", - "from senselab.utils.data_structures.device import DeviceType\n", - "from senselab.audio.tasks.preprocessing.preprocessing import downmix_audios_to_mono\n", - "from senselab.audio.tasks.preprocessing.preprocessing import resample_audios\n", + "from senselab.audio.tasks.preprocessing.preprocessing import downmix_audios_to_mono, resample_audios\n", "from senselab.audio.tasks.speech_to_text import transcribe_audios\n", - "from senselab.utils.tasks.plotting import plot_transcript\n", - "from senselab.audio.tasks.speech_to_text_evaluation import calculate_wer" + "from senselab.audio.tasks.speech_to_text_evaluation import calculate_wer\n", + "from senselab.utils.data_structures.device import DeviceType\n", + "from senselab.utils.data_structures.model import HFModel\n", + "from senselab.utils.tasks.plotting import plot_transcript" ] }, { diff --git a/tutorials/text_to_speech.ipynb b/tutorials/text_to_speech.ipynb index b9561f64..328efdc4 100644 --- a/tutorials/text_to_speech.ipynb +++ b/tutorials/text_to_speech.ipynb @@ -30,12 +30,13 @@ "# Model: facebook/mms-tts-eng (https://huggingface.co/facebook/mms-tts-eng)\n", "\n", "# Import the Hugging Face model\n", - "from senselab.utils.data_structures.model import HFModel\n", - "# Import the text-to-speech function\n", - "from senselab.audio.tasks.text_to_speech import synthesize_texts\n", "# Import the audio player\n", "from senselab.audio.tasks.plotting.plotting import play_audio\n", "\n", + "# Import the text-to-speech function\n", + "from senselab.audio.tasks.text_to_speech import synthesize_texts\n", + "from senselab.utils.data_structures.model import HFModel\n", + "\n", "# Initialize the model\n", "hf_model = HFModel(path_or_uri=\"facebook/mms-tts-eng\", revision=\"main\")\n", "# Write the text to be synthesized\n", @@ -64,12 +65,13 @@ "# Model: suno/bark-small (https://huggingface.co/suno/bark-small)\n", "\n", "# Import the Hugging Face model\n", - "from senselab.utils.data_structures.model import HFModel\n", - "# Import the text-to-speech function\n", - "from senselab.audio.tasks.text_to_speech import synthesize_texts\n", "# Import the audio player\n", "from senselab.audio.tasks.plotting.plotting import play_audio\n", "\n", + "# Import the text-to-speech function\n", + "from senselab.audio.tasks.text_to_speech import synthesize_texts\n", + "from senselab.utils.data_structures.model import HFModel\n", + "\n", "# Initialize the model\n", "hf_model = HFModel(path_or_uri=\"suno/bark-small\", revision=\"main\")\n", "# Write the text to be synthesized\n", @@ -94,8 +96,9 @@ "metadata": {}, "outputs": [], "source": [ - "from datasets import load_dataset\n", "import torch\n", + "from datasets import load_dataset\n", + "\n", "embeddings_dataset = load_dataset(\"Matthijs/cmu-arctic-xvectors\", split=\"validation\")\n", "speaker_embedding = torch.tensor(embeddings_dataset[7306][\"xvector\"]).unsqueeze(0)\n", "\n", @@ -137,13 +140,13 @@ "metadata": {}, "outputs": [], "source": [ - "from senselab.utils.data_structures.model import TorchModel\n", - "from senselab.utils.data_structures.language import Language\n", - "from senselab.utils.data_structures.device import DeviceType\n", "from senselab.audio.data_structures.audio import Audio\n", - "from senselab.audio.tasks.preprocessing.preprocessing import resample_audios, downmix_audios_to_mono, extract_segments\n", "from senselab.audio.tasks.plotting.plotting import play_audio\n", - "from senselab.audio.tasks.text_to_speech import synthesize_texts" + "from senselab.audio.tasks.preprocessing.preprocessing import downmix_audios_to_mono, extract_segments, resample_audios\n", + "from senselab.audio.tasks.text_to_speech import synthesize_texts\n", + "from senselab.utils.data_structures.device import DeviceType\n", + "from senselab.utils.data_structures.language import Language\n", + "from senselab.utils.data_structures.model import TorchModel" ] }, { diff --git a/tutorials/transcribe_timestamped_tutorial.ipynb b/tutorials/transcribe_timestamped.ipynb similarity index 98% rename from tutorials/transcribe_timestamped_tutorial.ipynb rename to tutorials/transcribe_timestamped.ipynb index 3b72b401..e546bc32 100644 --- a/tutorials/transcribe_timestamped_tutorial.ipynb +++ b/tutorials/transcribe_timestamped.ipynb @@ -5,7 +5,7 @@ "metadata": {}, "source": [ "### Transcribe Timestamped Workflow Tutorial\n", - "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/sensein/senselab/blob/main/tutorials/transcribe_timestamped_tutorial.ipynb)\n", + "[![Open In Colab](https://colab.research.google.com/assets/colab-badge.svg)](https://colab.research.google.com/github/sensein/senselab/blob/main/tutorials/transcribe_timestamped.ipynb)\n", "\n", "This notebook provides a step-by-step guide on how to use the transcribe_timestamped function to transcribe audio files and obtain timestamped transcriptions.\n", "\n", @@ -18,11 +18,10 @@ "metadata": {}, "outputs": [], "source": [ - "\n", "# Import the necessary modules from the Senselab package for audio processing\n", + "from senselab.audio.data_structures.audio import Audio\n", "from senselab.audio.tasks.plotting.plotting import play_audio\n", "from senselab.audio.tasks.preprocessing.preprocessing import resample_audios\n", - "from senselab.audio.data_structures.audio import Audio\n", "from senselab.audio.workflows.transcribe_timestamped import transcribe_timestamped\n", "from senselab.utils.data_structures.language import Language\n", "from senselab.utils.data_structures.model import HFModel" diff --git a/tutorials/voice_activity_detection.ipynb b/tutorials/voice_activity_detection.ipynb index 973273f6..c6955809 100644 --- a/tutorials/voice_activity_detection.ipynb +++ b/tutorials/voice_activity_detection.ipynb @@ -21,11 +21,11 @@ "source": [ "# Import necessary modules\n", "from senselab.audio.data_structures.audio import Audio\n", - "from senselab.audio.tasks.voice_activity_detection import detect_human_voice_activity_in_audios\n", - "from senselab.utils.data_structures.model import PyannoteAudioModel\n", - "from senselab.utils.data_structures.device import DeviceType\n", "from senselab.audio.tasks.plotting.plotting import play_audio\n", "from senselab.audio.tasks.preprocessing.preprocessing import resample_audios\n", + "from senselab.audio.tasks.voice_activity_detection import detect_human_voice_activity_in_audios\n", + "from senselab.utils.data_structures.device import DeviceType\n", + "from senselab.utils.data_structures.model import PyannoteAudioModel\n", "from senselab.utils.tasks.plotting import plot_segment" ] }, diff --git a/tutorials/voice_cloning.ipynb b/tutorials/voice_cloning.ipynb index b578c3b7..78ebb096 100644 --- a/tutorials/voice_cloning.ipynb +++ b/tutorials/voice_cloning.ipynb @@ -25,11 +25,11 @@ "outputs": [], "source": [ "from senselab.audio.data_structures.audio import Audio\n", + "from senselab.audio.tasks.plotting.plotting import play_audio\n", + "from senselab.audio.tasks.preprocessing.preprocessing import extract_segments, resample_audios\n", "from senselab.audio.tasks.voice_cloning.api import clone_voices\n", "from senselab.utils.data_structures.device import DeviceType\n", - "from senselab.utils.data_structures.model import TorchModel\n", - "from senselab.audio.tasks.preprocessing.preprocessing import resample_audios, extract_segments\n", - "from senselab.audio.tasks.plotting.plotting import play_audio" + "from senselab.utils.data_structures.model import TorchModel" ] }, {