34 Commits
Author SHA1 Message Date
dependabot[bot] ac33956197 Update ruff requirement from >=0.14.0 to >=0.15.10 (#37)
Updates the requirements on [ruff](https://github.com/astral-sh/ruff) to permit the latest version.
- [Release notes](https://github.com/astral-sh/ruff/releases)
- [Changelog](https://github.com/astral-sh/ruff/blob/main/CHANGELOG.md)
- [Commits](https://github.com/astral-sh/ruff/compare/0.14.0...0.15.10)

---
updated-dependencies:
- dependency-name: ruff
  dependency-version: 0.15.10
  dependency-type: direct:development
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-04-13 19:42:58 -07:00
Christian Byrne bb42b2d210 docs: add CHANGELOG.md for v2.0.0 (#46, #47) (#57) 2026-04-13 14:26:12 -07:00
Christian Byrne c7c313e542 docs: add SECURITY.md (#43) (#56) 2026-04-13 14:26:07 -07:00
Christian Byrne cd551161f2 chore: add issue templates, PR template, and release config (#53, #54, #48) (#59) 2026-04-13 14:26:04 -07:00
Christian Byrne 49f095eecb chore: repo essentials — metadata, tooling, and badges (#55)
* docs: add SECURITY.md (#43)

* docs: add CHANGELOG.md for v2.0.0 (#46, #47)
2026-04-13 14:26:01 -07:00
Christian Byrne dcd4dc5ad0 ci: add concurrency, permissions, pip caching, and timeouts (#39, #40, #41, #42) (#58) 2026-04-13 14:25:58 -07:00
Christian Byrne 067ba441ad fix: address code review findings for ChunkResampler tolerance feature (#35)
- Reject zero frequencies (was only rejecting negative, causing ZeroDivisionError)
- Use local vars for clamp values instead of storing on self (only used in __init__)
- Remove forced min margin of 1 in _find_optimal_freq (tiny tolerance should mean no search)
- Fix stale test comments about reduce_ratio using original args (bug was fixed in #34)
- Strengthen clamping tests to verify actual Resample ratio, not just construction
- Fix test_chunk_size_uses_effective_ratio to actually test clamped scenario
- Add tests: zero freq, tiny tolerance edge case (187 total, 95% coverage)
2026-04-10 19:02:48 -07:00
Christian Byrne 403cd1500a feat: add tolerance and configurable clamp parameters to ChunkResampler (#4) (#34)
Add tolerance parameter that searches for a nearby frequency with a large
GCD, producing a smaller resampling kernel for faster computation.

Also make UPPER_CLAMP and LOWER_CLAMP configurable via upper_clamp and
lower_clamp constructor parameters (default to previous hardcoded values).

Fix reduce_ratio call to use potentially-modified self.orig_freq and
self.new_freq instead of the raw constructor arguments.

Closes #4
2026-04-10 18:36:55 -07:00
Christian Byrne fe05f188c6 chore: bump version to 2.0.0 and add Nodes table to README (#33)
* chore: bump version to 2.0.0 and add Nodes table to README

- Bump version from 1.5.0 to 2.0.0 in pyproject.toml
- Add Nodes table documenting all 7 nodes with descriptions
- Keep existing Examples, Stem Mapping, Requirements, Installation,
  and Troubleshooting sections unchanged

Ref: christian-byrne/ticket-to-pr-pipeline#989

* fix: match Audio Combine method order to INPUT_TYPES

Adversarial review caught ordering mismatch between README and code.
2026-04-10 18:11:29 -07:00
Christian Byrne fd28a44ee4 fix: issue triage — corrupted model error (#21), stem mapping docs (#11), troubleshooting (#9) (#32)
- Wrap bundle.get_model() with try/except for BadZipFile/RuntimeError to
  give users a clear message about re-downloading the corrupted checkpoint
- Add Stem Mapping section to README documenting which instruments map to
  which output (guitar → Other stem, addressing #11)
- Add Troubleshooting section with fixes for #21 and #9
- Fix outdated librosa==0.10.2 in README to match actual >=0.10.2,<1 pin
- Add 3 tests for corrupted model error handling

Closes #9, closes #11, closes #21
2026-04-10 17:53:42 -07:00
Christian Byrne d0c76280c4 fix: phase 1 bug fixes (#23 cpu no-op, #16/#22 float cast, #20 librosa pin) (#31)
- Fix #23: move both waveforms to cpu after resampling in combine.py,
  not just the resampled one (avoids device mismatch on arithmetic)
- Fix #16/#22: add waveform.float() after ensure_stereo() in separation.py
  to handle int16 audio and double/float dtype mismatches
- Fix #20: relax librosa pin from ==0.10.2 to >=0.10.2,<1 in both
  pyproject.toml and requirements.txt
- Add tests verifying resampled waveforms are moved to cpu (with cuda inputs)
- Add test verifying float32 dtype invariant after ensure_stereo()
2026-04-10 15:25:39 -07:00
Christian Byrne 201d4d244e phase0: comprehensive test suite (27% → 94% coverage), ruff lint/format, tooling config (#30)
- Add 155 new tests across 9 test files covering all src modules
- Configure pytest-cov with 90% coverage threshold
- Configure ruff lint (E, F, W, I, UP, B, SIM, TCH) and format (line-length=120)
- Add mypy config with ignore_missing_imports for comfy ecosystem
- Fix all lint issues: deprecated typing imports, import sorting, simplifications
- Add from __future__ import annotations for Python 3.9 compat
- Apply ruff format across entire codebase
- Mock infrastructure for testing without torch/torchaudio/comfy dependencies
- No runtime behavior changes
2026-04-10 14:44:26 -07:00
Christian Byrne 12062cacb4 Merge pull request #29 from christian-byrne/feature/video-video-node
Add VIDEO-aware AudioVideoCombine and CI
2025-12-23 17:27:55 -08:00
bymyself 766a6851e7 Use temporary WAV file compatible with Windows 2025-12-23 17:24:46 -08:00
bymyself 59e0314ce3 Guard package imports when dependencies missing 2025-12-23 17:20:36 -08:00
bymyself 665b225b33 Support absolute imports when package context missing 2025-12-23 17:16:37 -08:00
bymyself 271de5faac Stub folder_paths in tests 2025-12-23 17:13:16 -08:00
bymyself c0bc67b7ee Format legacy audio utilities and fix test import loading 2025-12-23 17:10:12 -08:00
bymyself 504248bd04 Add MIT license and bump version to 1.5.0 2025-12-23 17:04:50 -08:00
bymyself 256d51a24f Define license as file reference 2025-12-23 17:03:58 -08:00
bymyself bf63e9159d Fix CI working directory paths 2025-12-23 17:00:51 -08:00
bymyself f0409dfacb Add CI tooling, optional deps, and contributor docs 2025-12-23 16:58:23 -08:00
bymyself f3de07d201 Replace match statement for Python 3.9 support 2025-12-23 16:58:15 -08:00
bymyself 42e3e1e4ed Refactor AudioVideoCombine for VIDEO workflows 2025-12-23 16:58:07 -08:00
Christian Byrne 7a6fb253e8 Merge pull request #17 from ComfyNodePRs/update-publish-yaml
Update Github Action for Publishing to Comfy Registry
2025-12-18 16:21:50 -08:00
Christian Byrne d46191625c Merge pull request #28 from christian-byrne/christian-byrne-patch-1
Bump version
2025-12-09 13:54:02 -08:00
Christian Byrne 8346a6027e Update pyproject.toml 2025-12-09 14:52:59 -07:00
Christian Byrne 88eb956e2d Merge pull request #27 from huntcsg/master
security: mitigate RCE and path traversal risks
2025-12-09 13:52:29 -08:00
Hunter Senft-Grupp 4452f6e0ea security: mitigate RCE and path traversal risks
- Replace os.system() with subprocess.run() using list args to prevent
  shell injection in auto_open feature
- Fix Windows auto_open which was broken (start command misinterpreted
  quoted path as window title)
- Add AUDIO_SEP_DISABLE_AUTO_OPEN env var to disable auto_open feature
  entirely for headless/server deployments
- Add AUDIO_SEP_STRICT_PATHS env var to restrict file access to current
  working directory subtree for multi-tenant environments

All changes are backwards compatible - security restrictions are opt-in
via environment variables.
2025-12-09 16:33:56 -05:00
snomiao 845a33c5c2 chore(publish): update GitHub Actions workflow for publishing
- Add permissions to allow issue writing
- Update condition to check repository owner instead of fork status
- Change action version from `main` to `v1` for stability and consistency
2025-01-25 04:55:59 +00:00
christian-byrne 31a4567726 Update README 2025-01-05 12:12:44 -07:00
christian-byrne 5291831d7c Use example_workflows new frontend feature 2025-01-05 12:10:38 -07:00
christian-byrne 4d38c6f0b5 Remove demo files off master (2/2) 2025-01-05 12:08:48 -07:00
christian-byrne 04454dea39 Remove demo files off master. fix #15 (1/2) 2025-01-05 11:51:50 -07:00
58 changed files with 3596 additions and 253 deletions
+61
View File
@@ -0,0 +1,61 @@
name: Bug Report
description: Report a bug with Audio Separation Nodes
labels: ["bug"]
body:
- type: markdown
attributes:
value: |
Thanks for reporting a bug! Please fill out the sections below.
- type: textarea
id: description
attributes:
label: Description
description: A clear description of the bug.
validations:
required: true
- type: textarea
id: reproduce
attributes:
label: Steps to reproduce
description: How can we reproduce this issue?
placeholder: |
1. Load workflow with ...
2. Connect nodes ...
3. Execute ...
4. See error
validations:
required: true
- type: textarea
id: expected
attributes:
label: Expected behavior
description: What did you expect to happen?
validations:
required: true
- type: textarea
id: actual
attributes:
label: Actual behavior
description: What actually happened? Include error messages or logs if applicable.
validations:
required: true
- type: input
id: comfyui-version
attributes:
label: ComfyUI version
placeholder: "e.g., latest, commit hash, or release tag"
- type: input
id: os
attributes:
label: Operating system
placeholder: "e.g., Windows 11, Ubuntu 24.04, macOS 15"
- type: input
id: python-version
attributes:
label: Python version
placeholder: "e.g., 3.11.9"
- type: textarea
id: additional
attributes:
label: Additional context
description: Any other context, screenshots, or workflow files.
@@ -0,0 +1,32 @@
name: Feature Request
description: Suggest a new feature or improvement
labels: ["enhancement"]
body:
- type: markdown
attributes:
value: |
Thanks for suggesting an improvement!
- type: textarea
id: problem
attributes:
label: Problem
description: What problem does this feature solve?
validations:
required: true
- type: textarea
id: solution
attributes:
label: Proposed solution
description: Describe the solution you'd like.
validations:
required: true
- type: textarea
id: alternatives
attributes:
label: Alternatives considered
description: Any alternative approaches you've considered.
- type: textarea
id: additional
attributes:
label: Additional context
description: Any other context, mockups, or examples.
+18
View File
@@ -0,0 +1,18 @@
## Summary
<!-- Brief description of what this PR does -->
## Related issues
<!-- Link issues: Closes #N, Fixes #N -->
## Changes
-
## Checklist
- [ ] Tests pass locally (`pytest tests`)
- [ ] Linting passes (`ruff check src tests`)
- [ ] Formatting passes (`ruff format --check src tests`)
- [ ] New code has test coverage
+18
View File
@@ -0,0 +1,18 @@
changelog:
categories:
- title: "🚀 Features"
labels:
- enhancement
- title: "🐛 Bug Fixes"
labels:
- bug
- title: "🧹 Maintenance"
labels:
- chore
- dependencies
- title: "📖 Documentation"
labels:
- documentation
- title: "Other Changes"
labels:
- "*"
+5 -3
View File
@@ -7,17 +7,19 @@ on:
paths:
- "pyproject.toml"
permissions:
issues: write
jobs:
publish-node:
name: Publish Custom Node to registry
runs-on: ubuntu-latest
# if this is a forked repository. Skipping the workflow.
if: github.event.repository.fork == false
if: ${{ github.repository_owner == 'christian-byrne' }}
steps:
- name: Check out code
uses: actions/checkout@v4
- name: Publish Custom Node
uses: Comfy-Org/publish-node-action@main
uses: Comfy-Org/publish-node-action@v1
with:
## Add your own personal access token to your Github Repository secrets and reference it here.
personal_access_token: ${{ secrets.REGISTRY_ACCESS_TOKEN }}
+64
View File
@@ -0,0 +1,64 @@
name: Tests
on:
push:
branches: ["master"]
pull_request:
concurrency:
group: ${{ github.workflow }}-${{ github.ref }}
cancel-in-progress: true
permissions:
contents: read
jobs:
ci:
runs-on: ubuntu-latest
timeout-minutes: 10
strategy:
matrix:
python-version: ["3.9", "3.10", "3.11"]
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: 'pip'
- name: Install tooling
run: |
python -m pip install --upgrade pip
python -m pip install -e '.[test,dev]'
- name: Ruff format check
run: ruff format --check src tests
- name: Ruff lint
run: ruff check src tests
- name: Python bytecode compile
run: python -m compileall src
- name: Run tests
run: pytest tests
windows:
runs-on: windows-latest
timeout-minutes: 10
strategy:
matrix:
python-version: ["3.10"]
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: 'pip'
- name: Install tooling
run: |
python -m pip install --upgrade pip
python -m pip install -e ".[test,dev]"
- name: Ruff format check
run: ruff format --check src tests
- name: Ruff lint
run: ruff check src tests
- name: Python bytecode compile
run: python -m compileall src
- name: Run tests
run: pytest tests
+3 -1
View File
@@ -20,4 +20,6 @@ lut
video-editing/
video-editing/**/**
testing-all-nodes-megaworkflow.json
todo.md
todo.md
!tests/
!tests/**
+7
View File
@@ -0,0 +1,7 @@
repos:
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.5.7
hooks:
- id: ruff
args: ["--fix"]
- id: ruff-format
+44
View File
@@ -0,0 +1,44 @@
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [2.0.0] - 2025-04-13
### Added
- Comprehensive test suite with 187 tests and 95% coverage ([#30])
- Ruff linting and formatting with CI enforcement ([#30])
- Windows CI testing on Python 3.10 ([#30])
- Configurable `tolerance` and `clamp` parameters for ChunkResampler ([#34])
- Nodes table in README documenting all seven nodes ([#33])
- Stem mapping reference and troubleshooting guide in README ([#32])
### Fixed
- CPU device handling no longer silently skips separation ([#31], [#23])
- Float-to-int cast errors in audio processing ([#31], [#16], [#22])
- Corrupted model checkpoint error with actionable fix instructions ([#32], [#21])
### Changed
- Pinned librosa to `>=0.10.2,<1` to prevent breaking changes ([#31], [#20])
- Version bumped from 1.x to 2.0.0 ([#33])
[2.0.0]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/releases/tag/v2.0.0
[#4]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/issues/4
[#9]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/issues/9
[#11]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/issues/11
[#16]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/issues/16
[#20]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/issues/20
[#21]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/issues/21
[#22]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/issues/22
[#23]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/issues/23
[#30]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/pull/30
[#31]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/pull/31
[#32]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/pull/32
[#33]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/pull/33
[#34]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/pull/34
[#35]: https://github.com/christian-byrne/audio-separation-nodes-comfyui/pull/35
+11
View File
@@ -0,0 +1,11 @@
# Contributing
```bash
pip install -e '.[test,dev]'
pre-commit install
PYTHONPATH=src pytest tests
```
- The editable install pulls in Ruff, pytest, and other dev tools.
- `pre-commit install` keeps formatting/linting consistent with CI.
- Running pytest with `PYTHONPATH=src` mirrors the CI environment (Linux + Windows, Python 3.9/3.10/3.11).
+21
View File
@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Christian Byrne
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
+80 -38
View File
@@ -1,102 +1,110 @@
![worklow picture](./wiki/pics/Selection_016.png)
![worklow picture](https://github.com/christian-byrne/audio-separation-nodes-comfyui/blob/demo-files/wiki/pics/Selection_016.png?raw=true)
https://github.com/user-attachments/assets/c5cf20de-a17f-438d-81ac-0c392af669cf
# Nodes
| Node | Description |
|------|-------------|
| **Audio Separation** | Separate audio into four stems (bass, drums, other, vocals) using [Hybrid Demucs](https://pytorch.org/audio/stable/tutorials/hybrid_demucs_tutorial.html). |
| **Audio Combine** | Combine two audio tracks by overlaying their waveforms (add, mean, subtract, multiply, divide). |
| **Audio Crop** | Crop (trim) audio to a specific start and end time. |
| **Audio Tempo Match** | Match the tempo of two audio tracks by time-stretching both to their average BPM. |
| **Audio Speed Shift** | Time-stretch or time-compress audio by a given rate. |
| **Audio Get Tempo** | Get the tempo (BPM) of audio using onset detection. |
| **Audio Video Combine** | Replace the audio of a VIDEO input with a new audio track. |
# Examples
#### *`Separating Voices in a Video`*
#### _`Separating Voices in a Video`_
<details>
<summary> &nbsp; Show </summary>
> [!NOTE]
>
> [!NOTE]
>
> In order to load videos into the LoadAudio Node, change [this line](https://github.com/comfyanonymous/ComfyUI/blob/faa57430b0ff882275b1afcf6610e8e9f8a5929b/comfy_extras/nodes_audio.py#L185) in your Comfy install to include the video's extension (e.g., `.mp4`)
[workflow.json](./wiki/workflows/isolate-vocals-video.json)
[workflow.json](./example_workflows/Isolate%20Vocals%20from%20Video.json)
https://github.com/user-attachments/assets/c5af418e-7137-4c36-b86e-3352cf558ea8
</details>
#### *`Replacing BGM with StableAudio-Generated BGM`*
#### _`Replacing BGM with StableAudio-Generated BGM`_
<details>
<summary> &nbsp; Show </summary>
> [!NOTE]
>
> [!NOTE]
>
> In order to load videos into the LoadAudio Node, change [this line](https://github.com/comfyanonymous/ComfyUI/blob/faa57430b0ff882275b1afcf6610e8e9f8a5929b/comfy_extras/nodes_audio.py#L185) in your Comfy install to include the video's extension (e.g., `.mp4`)
You can use this to replace copyrighted BGM in a video with new BGM. You can set the denoise low, so that the new BGM is still stimilar to the original.
[workflow json](./wiki/workflows/replace-bgm.json)
[workflow json](./example_workflows/Replace%20BGM%20with%20Stable-Audio-Generated%20Music.json)
https://github.com/user-attachments/assets/a7d5656b-5f8b-439a-936f-6ebb6a0d538a
</details>
#### *`Remixing Songs with StableAudio`*
#### _`Remixing Songs with StableAudio`_
<details>
<summary> &nbsp; Show </summary>
- [workflow json](./wiki/workflows/remix-songs.json)
- [example output (audio file) with embedded workflow](./wiki/examples/ComfyUI_temp_iaepj_00001_.flac)
- [example output (audio file) with embedded workflow](./wiki/examples/ComfyUI_00002_.flac)
- [workflow json](./example_workflows/Remix%20Song.json)
- [example output (audio file) with embedded workflow](https://github.com/christian-byrne/audio-separation-nodes-comfyui/raw/refs/heads/demo-files/wiki/examples/ComfyUI_temp_iaepj_00001_.flac)
- [example output (audio file) with embedded workflow](https://github.com/christian-byrne/audio-separation-nodes-comfyui/raw/refs/heads/demo-files/wiki/examples/ComfyUI_00002_.flac)
</details>
#### *`Separating Song Vocals`*
#### _`Separating Song Vocals`_
<details>
<summary> &nbsp; Show </summary>
[workflow.json](./wiki/workflows/isolate-vocals-song.json)
[workflow.json](./example_workflows/Isolate%20Vocals%20from%20Audio.json)
https://github.com/user-attachments/assets/c5cf20de-a17f-438d-81ac-0c392af669cf
</details>
#### *`Extracting Instrumentals from Songs`*
#### _`Extracting Instrumentals from Songs`_
<details>
<summary> &nbsp; Show </summary>
- [workflow json](./wiki/workflows/extract-instrumental.json)
- [workflow json](./example_workflows/Extract%20Instrumental%20from%20Song.json)
</details>
&nbsp;
# Stem Mapping
The **Audio Separation** node uses [Hybrid Demucs](https://pytorch.org/audio/stable/tutorials/hybrid_demucs_tutorial.html) to split audio into four stems:
| Output | Contains |
|--------|----------|
| **Bass** | Bass guitar, sub-bass, low-frequency instruments |
| **Drums** | Drums, percussion, hi-hats |
| **Other** | Everything else — guitars, keyboards, synths, strings, etc. |
| **Vocals** | Singing, speech, vocal harmonies |
> **Looking for a specific instrument like guitar?** Guitar is included in the
> **Other** stem. To isolate guitar, separate first, then use the **Audio Combine**
> node to subtract unwanted elements or further process the "Other" output.
# Requirements
```m
librosa==0.10.2
```
librosa>=0.10.2,<1
torchaudio>=2.3.0
numpy
moviepy
@@ -106,5 +114,39 @@ moviepy
1. If you run ComfyUI inside of a virtual environment, make sure it is activated
1. `git clone` this repository in `ComfyUI/custom_nodes` folder
2. `cd` into the cloned repository
3. `pip install -r requirements.txt`
1. `cd` into the cloned repository
1. `pip install -r requirements.txt`
# Troubleshooting
<details>
<summary><b>BadZipFile / "failed finding central directory"</b></summary>
This error means the Hybrid Demucs model checkpoint was corrupted during download.
Delete the cached file and restart ComfyUI to trigger a fresh download:
```bash
# Default location (Linux/macOS)
rm ~/.cache/torch/hub/checkpoints/*.th
# Windows
del %USERPROFILE%\.cache\torch\hub\checkpoints\*.th
```
See [#21](https://github.com/christian-byrne/audio-separation-nodes-comfyui/issues/21).
</details>
<details>
<summary><b>ConnectionResetError on Windows</b></summary>
```
Exception in callback _ProactorBasePipeTransport._call_connection_lost(None)
ConnectionResetError: [WinError 10054]
```
This is harmless Windows asyncio noise — it does not affect audio separation results.
The error comes from Python's `ProactorEventLoop` closing connections and can be
safely ignored. See [#9](https://github.com/christian-byrne/audio-separation-nodes-comfyui/issues/9).
</details>
+29
View File
@@ -0,0 +1,29 @@
# Security Policy
## Reporting a Vulnerability
If you discover a security vulnerability in this project, please report it
responsibly:
1. **Do not** open a public GitHub issue.
2. Email **christian.byrne@comfy.org** with:
- A description of the vulnerability
- Steps to reproduce
- Potential impact
3. You will receive a response within 7 days.
## Scope
This project is a set of ComfyUI custom nodes for audio separation. Security
concerns most likely involve:
- Arbitrary code execution via crafted audio files
- Path traversal in file handling
- Denial of service through resource exhaustion
## Supported Versions
| Version | Supported |
|---------|-----------|
| 2.x | ✅ |
| < 2.0 | ❌ |
+70 -16
View File
@@ -1,18 +1,72 @@
from .src.separation import AudioSeparation
from .src.tempo_match import TempoMatch
from .src.crop import AudioCrop
from .src.combine import AudioCombine
from .src.combine_video_with_audio import AudioVideoCombine
from .src.time_shift import TimeShift
from .src.get_tempo import GetTempo
try: # pragma: no cover - Comfy runtime provides dependencies
from .src.separation import AudioSeparation
except Exception: # pragma: no cover
try:
from src.separation import AudioSeparation
except Exception: # pragma: no cover
AudioSeparation = None
try:
from .src.tempo_match import TempoMatch
except Exception:
try:
from src.tempo_match import TempoMatch
except Exception:
TempoMatch = None
try:
from .src.crop import AudioCrop
except Exception:
try:
from src.crop import AudioCrop
except Exception:
AudioCrop = None
try:
from .src.combine import AudioCombine
except Exception:
try:
from src.combine import AudioCombine
except Exception:
AudioCombine = None
try:
from .src.combine_video_with_audio import AudioVideoCombine
except Exception:
try:
from src.combine_video_with_audio import AudioVideoCombine
except Exception:
AudioVideoCombine = None
try:
from .src.time_shift import TimeShift
except Exception:
try:
from src.time_shift import TimeShift
except Exception:
TimeShift = None
try:
from .src.get_tempo import GetTempo
except Exception:
try:
from src.get_tempo import GetTempo
except Exception:
GetTempo = None
NODE_CLASS_MAPPINGS = {
"AudioSeparation": AudioSeparation,
"AudioCrop": AudioCrop,
"AudioCombine": AudioCombine,
"AudioTempoMatch": TempoMatch,
"AudioVideoCombine": AudioVideoCombine,
"AudioSpeedShift": TimeShift,
"AudioGetTempo": GetTempo,
}
NODE_CLASS_MAPPINGS = {}
if AudioSeparation:
NODE_CLASS_MAPPINGS["AudioSeparation"] = AudioSeparation
if AudioCrop:
NODE_CLASS_MAPPINGS["AudioCrop"] = AudioCrop
if AudioCombine:
NODE_CLASS_MAPPINGS["AudioCombine"] = AudioCombine
if TempoMatch:
NODE_CLASS_MAPPINGS["AudioTempoMatch"] = TempoMatch
if AudioVideoCombine:
NODE_CLASS_MAPPINGS["AudioVideoCombine"] = AudioVideoCombine
if TimeShift:
NODE_CLASS_MAPPINGS["AudioSpeedShift"] = TimeShift
if GetTempo:
NODE_CLASS_MAPPINGS["AudioGetTempo"] = GetTempo
Binary file not shown.

After

Width:  |  Height:  |  Size: 135 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 115 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 610 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 225 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 115 KiB

+28 -3
View File
@@ -1,9 +1,34 @@
[project]
name = "audio-separation-nodes-comfyui"
description = "Separate audio track into stems (vocals, bass, drums, other). Along with tools to recombine, tempo match, slice/crop audio"
version = "1.3.1"
license = "LICENSE"
dependencies = ["librosa==0.10.2", "numpy", "torchaudio>=2.3.0", "moviepy"]
version = "2.0.0"
license = { file = "LICENSE" }
dependencies = ["librosa>=0.10.2,<1", "numpy", "torchaudio>=2.3.0", "moviepy"]
[project.optional-dependencies]
test = ["pytest", "pytest-cov"]
dev = ["ruff>=0.15.10"]
[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "--cov=src --cov-report=term-missing --cov-fail-under=90"
[tool.ruff]
target-version = "py39"
line-length = 120
[tool.ruff.lint]
select = ["E", "F", "W", "I", "UP", "B", "SIM", "TCH"]
per-file-ignores = {"tests/*" = ["E402"]}
[tool.ruff.lint.isort]
known-third-party = ["torch", "torchaudio", "librosa", "numpy", "moviepy", "comfy", "comfy_api", "folder_paths"]
[tool.mypy]
python_version = "3.9"
warn_return_any = true
warn_unused_configs = true
ignore_missing_imports = true
[project.urls]
Repository = "https://github.com/christian-byrne/audio-separation-nodes-comfyui"
+1 -1
View File
@@ -1,4 +1,4 @@
librosa==0.10.2
librosa>=0.10.2,<1
torchaudio>=2.3.0
numpy
moviepy
+1
View File
@@ -1,4 +1,5 @@
from typing import TypedDict
from torch import Tensor
+71
View File
@@ -0,0 +1,71 @@
"""Pure helper functions for the AudioVideoCombine node."""
from __future__ import annotations
class AudioVideoCombineError(ValueError):
"""Custom error to allow targeted unit testing."""
def parse_timestamp(value: str, default: float | None) -> float:
"""Convert ``HH:MM:SS``/``MM:SS`` strings (or seconds) into a float."""
normalized = (value or "").strip()
if not normalized:
if default is None:
raise AudioVideoCombineError(
"AudioVideoCombine: A video end time must be provided when the duration cannot be determined."
)
return float(default)
if ":" not in normalized:
try:
return float(normalized)
except ValueError as exc:
raise AudioVideoCombineError(
f"AudioVideoCombine: Invalid timestamp '{value}'. Expected MM:SS or HH:MM:SS."
) from exc
parts = normalized.split(":")
if len(parts) == 2:
hours = 0
minutes, seconds = parts
elif len(parts) == 3:
hours, minutes, seconds = parts
else:
raise AudioVideoCombineError(f"AudioVideoCombine: Invalid timestamp '{value}'. Expected MM:SS or HH:MM:SS.")
try:
return int(hours) * 3600 + int(minutes) * 60 + float(seconds)
except ValueError as exc:
raise AudioVideoCombineError(
f"AudioVideoCombine: Invalid timestamp '{value}'. Expected MM:SS or HH:MM:SS."
) from exc
def compute_trim_window(
start_time: str,
end_time: str,
duration_seconds: float | None,
) -> tuple[float, float]:
"""Calculate the numeric trim window for the combine node."""
start_seconds = parse_timestamp(start_time, default=0.0)
end_seconds = parse_timestamp(end_time, default=duration_seconds)
if duration_seconds is not None:
end_seconds = min(end_seconds, duration_seconds)
if start_seconds >= end_seconds:
raise AudioVideoCombineError(
"AudioVideoCombine: Start time must be less than end time. Start time cannot be after video ends."
)
return start_seconds, end_seconds
__all__ = [
"AudioVideoCombineError",
"parse_timestamp",
"compute_trim_window",
]
+23 -19
View File
@@ -1,10 +1,14 @@
import torch
from torchaudio.transforms import Resample
from __future__ import annotations
from typing import TYPE_CHECKING
import comfy.model_management
from torchaudio.transforms import Resample
from typing import Tuple
from ._types import AUDIO
if TYPE_CHECKING:
import torch
from ._types import AUDIO
class AudioCombine:
@@ -36,8 +40,7 @@ class AudioCombine:
audio_1: AUDIO,
audio_2: AUDIO,
method: str = "add",
) -> Tuple[AUDIO]:
) -> tuple[AUDIO]:
waveform_1: torch.Tensor = audio_1["waveform"]
input_sample_rate_1: int = audio_1["sample_rate"]
@@ -50,13 +53,13 @@ class AudioCombine:
if input_sample_rate_1 < input_sample_rate_2:
resample = Resample(input_sample_rate_1, input_sample_rate_2).to(device)
waveform_1: torch.Tensor = resample(waveform_1.to(device))
waveform_1.to("cpu")
output_sample_rate = input_sample_rate_2
else:
resample = Resample(input_sample_rate_2, input_sample_rate_1).to(device)
waveform_2: torch.Tensor = resample(waveform_2.to(device))
waveform_2.to("cpu")
output_sample_rate = input_sample_rate_1
waveform_1 = waveform_1.to("cpu")
waveform_2 = waveform_2.to("cpu")
else:
output_sample_rate = input_sample_rate_1
@@ -67,17 +70,18 @@ class AudioCombine:
if waveform_2.shape[-1] != min_length:
waveform_2 = waveform_2[..., :min_length]
match method:
case "add":
waveform = waveform_1 + waveform_2
case "subtract":
waveform = waveform_1 - waveform_2
case "multiply":
waveform = waveform_1 * waveform_2
case "divide":
waveform = waveform_1 / waveform_2
case "mean":
waveform = (waveform_1 + waveform_2) / 2
if method == "add":
waveform = waveform_1 + waveform_2
elif method == "subtract":
waveform = waveform_1 - waveform_2
elif method == "multiply":
waveform = waveform_1 * waveform_2
elif method == "divide":
waveform = waveform_1 / waveform_2
elif method == "mean":
waveform = (waveform_1 + waveform_2) / 2
else:
raise ValueError(f"Unsupported combine method: {method}")
return (
{
+72 -76
View File
@@ -1,24 +1,34 @@
from __future__ import annotations
import os
import platform
import tempfile
import uuid
from pathlib import Path
from typing import TYPE_CHECKING
import torch
import torchaudio
from comfy_api.input_impl import VideoFromFile
try:
# moviepy<=1.0.3
from moviepy.editor import VideoFileClip, AudioFileClip
from moviepy.editor import AudioFileClip, VideoFileClip
except ImportError:
# moviepy>=2.0.0 (Nov. 2024)
from moviepy import VideoFileClip, AudioFileClip
from moviepy import AudioFileClip, VideoFileClip
from typing import Tuple
from ._types import AUDIO
import contextlib
import folder_paths
from .audio_video_logic import compute_trim_window
if TYPE_CHECKING:
import torch
from comfy_api.input.video_types import VideoInput
from ._types import AUDIO
class AudioVideoCombine:
@classmethod
@@ -26,13 +36,7 @@ class AudioVideoCombine:
return {
"required": {
"audio": ("AUDIO",),
"video_path": (
"STRING",
{
"default": "/path/to/video.mp4",
"tooltip": "The absolute file path to the video file to which the audio will be added.",
},
),
"video": ("VIDEO",),
},
"optional": {
"video_start_time": (
@@ -45,79 +49,70 @@ class AudioVideoCombine:
"video_end_time": (
"STRING",
{
"default": "1:00",
"tooltip": "The video will be trimmed to end at this time. The format is MM:SS.",
},
),
"auto_open": (
"BOOLEAN",
{
"default": False,
"label_on": "Auto open after combining",
"description": "Don't auto open after combining",
"tooltip": "Whether to automatically open the combined video with the default video player after processing.",
"default": "",
"tooltip": "The video will be trimmed to end at this time. Leave blank to use the full duration.", # noqa: E501
},
),
},
}
FUNCTION = "main"
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("saved_video_path",)
RETURN_TYPES = ("VIDEO",)
RETURN_NAMES = ("video",)
CATEGORY = "audio"
OUTPUT_NODE = True
OUTPUT_TOOLTIPS = ("The path to the output video.",)
DESCRIPTION = "Replace the audio of a video with a new audio track."
OUTPUT_TOOLTIPS = ("The combined video.",)
DESCRIPTION = "Replace the audio of a VIDEO input with a new audio track."
def main(
self,
audio: AUDIO,
video_path: str = "/path/to/video.mp4",
video: VideoInput,
video_start_time: str = "0:00",
video_end_time: str = "1:00",
auto_open: bool = False,
) -> Tuple[str]:
video_end_time: str = "",
) -> tuple[VideoFromFile]:
waveform: torch.Tensor = audio["waveform"]
sample_rate: int = audio["sample_rate"]
input_path = Path(video_path)
if not input_path.exists():
raise FileNotFoundError(
f"AudioVideoCombine: Video file not found: {video_path}"
)
temp_dir = Path(folder_paths.get_temp_directory())
temp_dir.mkdir(parents=True, exist_ok=True)
# Assume that no ":" in input means that the user is trying to specify seconds
if ":" not in video_start_time:
video_start_time = f"00:{video_start_time}"
if ":" not in video_end_time:
video_end_time = f"00:{video_end_time}"
try:
video_duration = video.get_duration()
except Exception:
video_duration = None
start_seconds_time = 60 * int(video_start_time.split(":")[0]) + int(
video_start_time.split(":")[1]
)
end_seconds_time = 60 * int(video_end_time.split(":")[0]) + int(
video_end_time.split(":")[1]
start_seconds_time, end_seconds_time = compute_trim_window(
video_start_time,
video_end_time,
video_duration,
)
clip_duration = end_seconds_time - start_seconds_time
if start_seconds_time > end_seconds_time:
raise ValueError(
"AudioVideoCombine: Start time must be less than end time. Start time cannot be after video ends."
temp_input_path: Path | None = None
source = video.get_stream_source()
if isinstance(source, (str, os.PathLike)) and Path(source).exists():
video_path = str(source)
else:
temp_input_path = temp_dir / f"audio_video_combine_input_{uuid.uuid4().hex}.mp4"
video.save_to(str(temp_input_path))
video_path = str(temp_input_path)
output_path = temp_dir / f"audio_video_combine_{uuid.uuid4().hex}.mp4"
target_samples = int(round(clip_duration * sample_rate)) if clip_duration > 0 else 0
trimmed_waveform = waveform
if target_samples > 0 and waveform.shape[-1] > target_samples:
trimmed_waveform = waveform[..., :target_samples]
temp_audio_file = tempfile.NamedTemporaryFile(suffix=".wav", delete=False) # noqa: SIM115
temp_audio_file.close()
try:
torchaudio.save(
temp_audio_file.name,
trimmed_waveform.squeeze(0),
sample_rate=sample_rate,
)
output_dir = Path(folder_paths.get_output_directory())
filename = input_path.stem
new_filename = f"{filename}_0_combined.mp4"
index = 0
while new_filename in [f.name for f in output_dir.iterdir()]:
index += 1
new_filename = f"{filename}_{index}_combined.mp4"
new_filepath = str(output_dir / new_filename)
with tempfile.NamedTemporaryFile(suffix=".wav") as f:
torchaudio.save(f.name, waveform.squeeze(0), sample_rate=sample_rate)
video = VideoFileClip(str(video_path), audio=False)
audio = AudioFileClip(f.name)
audio = AudioFileClip(temp_audio_file.name)
try:
# moviepy<=1.0.3
@@ -128,15 +123,16 @@ class AudioVideoCombine:
video = video.subclipped(start_seconds_time, end_seconds_time)
video = video.with_audio(audio)
video.write_videofile(new_filepath, codec="libx264", audio_codec="aac")
video.write_videofile(str(output_path), codec="libx264", audio_codec="aac")
new_filepath = os.path.normpath(new_filepath)
if auto_open:
if platform.system() == "Darwin":
os.system(f'open "{new_filepath}"')
elif platform.system() == "Windows":
os.system(f'start "{new_filepath}"')
else:
os.system(f'xdg-open "{new_filepath}"')
video.close()
audio.close()
finally:
with contextlib.suppress(OSError):
Path(temp_audio_file.name).unlink(missing_ok=True)
return (str(new_filepath),)
if temp_input_path and temp_input_path.exists():
with contextlib.suppress(OSError):
temp_input_path.unlink()
return (VideoFromFile(str(output_path)),)
+11 -14
View File
@@ -1,7 +1,11 @@
import torch
from __future__ import annotations
from typing import Tuple
from ._types import AUDIO
from typing import TYPE_CHECKING
if TYPE_CHECKING:
import torch
from ._types import AUDIO
class AudioCrop:
@@ -35,8 +39,7 @@ class AudioCrop:
audio: AUDIO,
start_time: str = "0:00",
end_time: str = "1:00",
) -> Tuple[AUDIO]:
) -> tuple[AUDIO]:
waveform: torch.Tensor = audio["waveform"]
sample_rate: int = audio["sample_rate"]
@@ -46,16 +49,12 @@ class AudioCrop:
if ":" not in end_time:
end_time = f"00:{end_time}"
start_seconds_time = 60 * int(start_time.split(":")[0]) + int(
start_time.split(":")[1]
)
start_seconds_time = 60 * int(start_time.split(":")[0]) + int(start_time.split(":")[1])
start_frame = start_seconds_time * sample_rate
if start_frame >= waveform.shape[-1]:
start_frame = waveform.shape[-1] - 1
end_seconds_time = 60 * int(end_time.split(":")[0]) + int(
end_time.split(":")[1]
)
end_seconds_time = 60 * int(end_time.split(":")[0]) + int(end_time.split(":")[1])
end_frame = end_seconds_time * sample_rate
if end_frame >= waveform.shape[-1]:
end_frame = waveform.shape[-1] - 1
@@ -65,9 +64,7 @@ class AudioCrop:
end_frame = 0
if start_frame > end_frame:
raise ValueError(
"AudioCrop: Start time must be less than end time and be within the audio length."
)
raise ValueError("AudioCrop: Start time must be less than end time and be within the audio length.")
return (
{
+7 -3
View File
@@ -1,7 +1,11 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from .utils import estimate_tempo
from typing import Tuple
from ._types import AUDIO
if TYPE_CHECKING:
from ._types import AUDIO
class GetTempo:
@@ -22,7 +26,7 @@ class GetTempo:
def main(
self,
audio: AUDIO,
) -> Tuple[AUDIO, AUDIO]:
) -> tuple[AUDIO, AUDIO]:
waveform = audio["waveform"].squeeze(0)
sample_rate = audio["sample_rate"]
tempo = estimate_tempo(waveform, sample_rate)
+65 -24
View File
@@ -1,43 +1,64 @@
import torch
from torchaudio.transforms import Resample
from __future__ import annotations
import math
import comfy.model_management
from typing import Union, Tuple
import torch
from torchaudio.transforms import Resample
class ChunkResampler:
"""
a larger lowpass_filter_width results in a larger resampling kernel, and therefore increases computation time for both the kernel computation and convolution
a larger lowpass_filter_width results in a larger resampling kernel, and therefore increases
computation time for both the kernel computation and convolution
using sinc_interp_kaiser results in longer computation times than the default sinc_interp_hann because it is more complex to compute the intermediate window values
using sinc_interp_kaiser results in longer computation times than the default sinc_interp_hann
because it is more complex to compute the intermediate window values
a large GCD between the sample and resample rate will result in a simplification that allows for a smaller kernel and faster kernel computation.
a large GCD between the sample and resample rate will result in a simplification that allows
for a smaller kernel and faster kernel computation.
"""
DEFAULT_UPPER_CLAMP = 1.1832
DEFAULT_LOWER_CLAMP = 0.945
def __init__(
self,
orig_freq: Union[int, float],
new_freq: Union[int, float],
orig_freq: int | float,
new_freq: int | float,
chunk_size_seconds: int = 2,
tolerance: float = 0.0,
upper_clamp: float | None = None,
lower_clamp: float | None = None,
):
if orig_freq < 0 or new_freq < 0:
if orig_freq <= 0 or new_freq <= 0:
raise ValueError("Frequencies must be positive.")
if tolerance < 0.0 or tolerance > 1.0:
raise ValueError("Tolerance must be between 0.0 and 1.0.")
uc = upper_clamp if upper_clamp is not None else self.DEFAULT_UPPER_CLAMP
lc = lower_clamp if lower_clamp is not None else self.DEFAULT_LOWER_CLAMP
if uc <= 0 or lc <= 0:
raise ValueError("Clamp values must be positive.")
if uc <= lc:
raise ValueError("upper_clamp must be greater than lower_clamp.")
self.UPPER_CLAMP = 1.1832
self.LOWER_CLAMP = 0.945
self.orig_freq = orig_freq
self.new_freq = new_freq
self.chunk_size_seconds = int(chunk_size_seconds)
change_ratio = new_freq / orig_freq
if change_ratio > self.UPPER_CLAMP:
self.new_freq = self.orig_freq * self.UPPER_CLAMP
elif change_ratio < self.LOWER_CLAMP:
self.new_freq = self.orig_freq * self.LOWER_CLAMP
if change_ratio > uc:
self.new_freq = self.orig_freq * uc
elif change_ratio < lc:
self.new_freq = self.orig_freq * lc
diff = abs(1 - change_ratio)
if tolerance > 0.0:
self.new_freq = self._find_optimal_freq(round(self.orig_freq), self.new_freq, tolerance)
effective_ratio = self.new_freq / self.orig_freq
diff = abs(1 - effective_ratio)
if diff > 0.08:
self.chunk_size_seconds = min(self.chunk_size_seconds, 1)
elif diff > 0.002:
@@ -47,26 +68,46 @@ class ChunkResampler:
# If the frequencies are float, try to convert to int while
# maintaining ratio (https://github.com/pytorch/audio/issues/1487).
self.orig_freq, self.new_freq = ChunkResampler.reduce_ratio(orig_freq, new_freq)
self.orig_freq, self.new_freq = ChunkResampler.reduce_ratio(self.orig_freq, self.new_freq)
self.device = comfy.model_management.get_torch_device()
self.resample = Resample(self.orig_freq, self.new_freq).to(self.device)
@staticmethod
def _find_optimal_freq(orig_freq: int, target_freq: float, tolerance: float) -> float:
"""Find a frequency near *target_freq* that shares a large GCD with *orig_freq*.
Searches integer candidates within ``target_freq * (1 ± tolerance)`` and
returns the one whose GCD with *orig_freq* is largest, producing a
smaller resampling kernel.
"""
target = int(round(target_freq))
margin = min(int(target * tolerance), 1000)
lo = max(1, target - margin)
hi = target + margin
best_freq = target
best_gcd = math.gcd(orig_freq, target)
for candidate in range(lo, hi + 1):
g = math.gcd(orig_freq, candidate)
if g > best_gcd:
best_gcd = g
best_freq = candidate
return float(best_freq)
def __call__(self, waveform: torch.Tensor) -> torch.Tensor:
waveform = waveform.to(self.device)
with torch.no_grad():
chunks = torch.split(
waveform, int(self.orig_freq * self.chunk_size_seconds), dim=-1
)
chunks = torch.split(waveform, int(self.orig_freq * self.chunk_size_seconds), dim=-1)
resampled_chunks = [self.resample(chunk) for chunk in chunks]
resampled_waveform = torch.cat(resampled_chunks, dim=-1)
return resampled_waveform.to("cpu")
@staticmethod
def reduce_ratio(
num1: Union[float, int], num2: Union[float, int]
) -> Tuple[int, int]:
def reduce_ratio(num1: float | int, num2: float | int) -> tuple[int, int]:
"""Reduces a ratio to its smallest **integer** form.
Args:
+29 -25
View File
@@ -1,15 +1,20 @@
"""Credit: https://pytorch.org/audio/stable/tutorials/hybrid_demucs_tutorial.html"""
import torch
from torchaudio.transforms import Fade, Resample
from torchaudio.pipelines import HDEMUCS_HIGH_MUSDB_PLUS
from __future__ import annotations
from typing import TYPE_CHECKING
from zipfile import BadZipFile
import comfy.model_management
import torch
from torchaudio.pipelines import HDEMUCS_HIGH_MUSDB_PLUS
from torchaudio.transforms import Fade, Resample
from typing import Dict, Tuple
from ._types import AUDIO
from .utils import ensure_stereo
if TYPE_CHECKING:
from ._types import AUDIO
class AudioSeparation:
@classmethod
@@ -28,21 +33,21 @@ class AudioSeparation:
],
{
"default": "linear",
"tooltip": "Audio is split into segments (chunks) with overlapping areas to ensure smooth transitions. This setting controls the fade effect at these overlaps. Choose Linear for even fading, Half-Sine for a smooth curve, Logarithmic for a quick fade out and slow fade in, or Exponential for a slow fade out and quick fade in.",
"tooltip": "Audio is split into segments (chunks) with overlapping areas to ensure smooth transitions. This setting controls the fade effect at these overlaps. Choose Linear for even fading, Half-Sine for a smooth curve, Logarithmic for a quick fade out and slow fade in, or Exponential for a slow fade out and quick fade in.", # noqa: E501
},
),
"chunk_length": (
"FLOAT",
{
"default": 10.0,
"tooltip": "The length of each segment (chunk) in seconds. Longer chunks may require more memory and MIGHT produce better results.",
"tooltip": "The length of each segment (chunk) in seconds. Longer chunks may require more memory and MIGHT produce better results.", # noqa: E501
},
),
"chunk_overlap": (
"FLOAT",
{
"default": 0.1,
"tooltip": "The overlap between each segment (chunk) in seconds. A higher overlap may be necessary if chunks are too short or the audio changes rapidly.",
"tooltip": "The overlap between each segment (chunk) in seconds. A higher overlap may be necessary if chunks are too short or the audio changes rapidly.", # noqa: E501
},
),
},
@@ -60,24 +65,31 @@ class AudioSeparation:
chunk_fade_shape: str = "linear",
chunk_length: float = 10.0,
chunk_overlap: float = 0.1,
) -> Tuple[AUDIO, AUDIO, AUDIO, AUDIO]:
) -> tuple[AUDIO, AUDIO, AUDIO, AUDIO]:
device: torch.device = comfy.model_management.get_torch_device()
waveform: torch.Tensor = audio["waveform"]
waveform = waveform.squeeze(0).to(device)
self.input_sample_rate_: int = audio["sample_rate"]
bundle = HDEMUCS_HIGH_MUSDB_PLUS
model: torch.nn.Module = bundle.get_model().to(device)
try:
model: torch.nn.Module = bundle.get_model()
except (BadZipFile, RuntimeError) as exc:
raise RuntimeError(
"Failed to load the Hybrid Demucs model — the downloaded checkpoint "
"appears to be corrupted. Delete the cached model file and restart "
"ComfyUI to trigger a fresh download. The cached file is typically "
"located in your torch hub cache directory (~/.cache/torch/hub/checkpoints/)."
) from exc
model = model.to(device)
self.model_sample_rate = bundle.sample_rate
waveform = ensure_stereo(waveform)
waveform = waveform.float()
# Resample to model's expected sample rate
if self.input_sample_rate_ != self.model_sample_rate:
resample = Resample(self.input_sample_rate_, self.model_sample_rate).to(
device
)
resample = Resample(self.input_sample_rate_, self.model_sample_rate).to(device)
waveform = resample(waveform)
ref = waveform.mean(0)
@@ -98,10 +110,7 @@ class AudioSeparation:
return self.sources_to_tuple(dict(zip(sources_list, sources)))
def sources_to_tuple(
self, sources: Dict[str, torch.Tensor]
) -> Tuple[AUDIO, AUDIO, AUDIO, AUDIO]:
def sources_to_tuple(self, sources: dict[str, torch.Tensor]) -> tuple[AUDIO, AUDIO, AUDIO, AUDIO]:
output_order = ["bass", "drums", "other", "vocals"]
outputs = []
for source in output_order:
@@ -137,10 +146,7 @@ class AudioSeparation:
When `device` is different from `mix.device`, only local computations will
be on `device`, while the entire tracks will be stored on `mix.device`.
"""
if device is None:
device = mix.device
else:
device = torch.device(device)
device = mix.device if device is None else torch.device(device)
batch, channels, length = mix.shape
@@ -148,9 +154,7 @@ class AudioSeparation:
start = 0
end = chunk_len
overlap_frames = overlap * sample_rate
fade = Fade(
fade_in_len=0, fade_out_len=int(overlap_frames), fade_shape=chunk_fade_shape
)
fade = Fade(fade_in_len=0, fade_out_len=int(overlap_frames), fade_shape=chunk_fade_shape)
final = torch.zeros(batch, len(model.sources), channels, length, device=device)
+8 -4
View File
@@ -1,7 +1,11 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from .utils import estimate_tempo, time_shift
from typing import Tuple
from ._types import AUDIO
if TYPE_CHECKING:
from ._types import AUDIO
class TempoMatch:
@@ -17,13 +21,13 @@ class TempoMatch:
FUNCTION = "main"
RETURN_TYPES = ("AUDIO", "AUDIO")
CATEGORY = "audio"
DESCRIPTION = "Match the tempo of two audio tracks by time-stretching them both to match the average tempo between them. E.g., if one audio track is 120 BPM and the other is 100 BPM, both will be time-stretched to 110 BPM."
DESCRIPTION = "Match the tempo of two audio tracks by time-stretching them both to match the average tempo between them. E.g., if one audio track is 120 BPM and the other is 100 BPM, both will be time-stretched to 110 BPM." # noqa: E501
def main(
self,
audio_1: AUDIO,
audio_2: AUDIO,
) -> Tuple[AUDIO, AUDIO]:
) -> tuple[AUDIO, AUDIO]:
waveform_1 = audio_1["waveform"].squeeze(0)
input_sample_rate_1 = audio_1["sample_rate"]
+12 -5
View File
@@ -1,7 +1,11 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from .utils import time_shift
from typing import Tuple
from ._types import AUDIO
if TYPE_CHECKING:
from ._types import AUDIO
class TimeShift:
@@ -10,20 +14,23 @@ class TimeShift:
return {
"required": {
"audio": ("AUDIO",),
"rate": ("FLOAT", {"default": 1.0, "min": 0.1, "max": 10.0, "step": 0.1}),
"rate": (
"FLOAT",
{"default": 1.0, "min": 0.1, "max": 10.0, "step": 0.1},
),
},
}
FUNCTION = "main"
RETURN_TYPES = ("AUDIO",)
CATEGORY = "audio"
DESCRIPTION = "Time-stretch or time-compress audio by a given rate. A rate of 2.0 will double the speed of the audio, while a rate of 0.5 will halve the speed."
DESCRIPTION = "Time-stretch or time-compress audio by a given rate. A rate of 2.0 will double the speed of the audio, while a rate of 0.5 will halve the speed." # noqa: E501
def main(
self,
audio: AUDIO,
rate: float,
) -> Tuple[AUDIO, AUDIO]:
) -> tuple[AUDIO, AUDIO]:
waveform = audio["waveform"].squeeze(0)
sample_rate = audio["sample_rate"]
rate = min(max(rate, 0.1), 10.0)
+12 -21
View File
@@ -1,8 +1,9 @@
import librosa
import torch
import math
import torchaudio.functional as F
import librosa
import numpy as np
import torch
import torchaudio.functional as F
def time_shift(
@@ -29,9 +30,7 @@ def time_shift(
if win_length is None:
win_length = fft_size
window = torch.hann_window(
win_length, device=waveform.device
) # shape: [win_length]
window = torch.hann_window(win_length, device=waveform.device) # shape: [win_length]
with torch.no_grad():
complex_spectogram = torch.stft(
@@ -44,13 +43,9 @@ def time_shift(
) # shape: [channels, freq, time]
if complex_spectogram.dtype != torch.cfloat:
raise TypeError(
f"Expected complex-valued STFT for phase vocoder, got dtype {complex_spectogram.dtype}"
)
raise TypeError(f"Expected complex-valued STFT for phase vocoder, got dtype {complex_spectogram.dtype}")
phase_advance = torch.linspace(
0, math.pi * hop_size, complex_spectogram.shape[1]
)[
phase_advance = torch.linspace(0, math.pi * hop_size, complex_spectogram.shape[1])[
..., None
] # shape: [freq, 1]
@@ -59,9 +54,9 @@ def time_shift(
) # shape: [channels, freq, stretched_time]
expected_time = math.ceil(complex_spectogram.shape[2] / rate)
assert (
abs(stretched_spectogram.shape[2] - expected_time) < 3
), f"Expected Time: {expected_time}, Stretched Time: {stretched_spectogram.shape[2]}"
assert abs(stretched_spectogram.shape[2] - expected_time) < 3, (
f"Expected Time: {expected_time}, Stretched Time: {stretched_spectogram.shape[2]}"
)
# Convert back to time basis with inverse STFT
return torch.istft(
@@ -77,9 +72,7 @@ def estimate_tempo(waveform: torch.Tensor, sample_rate: int) -> float:
if waveform.dim() == 3:
waveform = waveform.squeeze(0)
if waveform.dim() != 2:
raise TypeError(
f"Expected waveform to be [channels, frames], got {waveform.shape}"
)
raise TypeError(f"Expected waveform to be [channels, frames], got {waveform.shape}")
onset_env = librosa.onset.onset_strength(
y=waveform.numpy(),
@@ -111,9 +104,7 @@ def ensure_stereo(audio: torch.Tensor) -> torch.Tensor:
torch.Tensor: Stereo audio with the same dimensional format as the input.
"""
if audio.ndim not in (2, 3):
raise ValueError(
"Audio input must have 2 or 3 dimensions: [channels, frames] or [batch, channels, frames]."
)
raise ValueError("Audio input must have 2 or 3 dimensions: [channels, frames] or [batch, channels, frames].")
is_batched = audio.ndim == 3
channels_dim = 1 if is_batched else 0
+208
View File
@@ -0,0 +1,208 @@
from __future__ import annotations
import sys
import types
from pathlib import Path
import numpy as np
import pytest
# ---------------------------------------------------------------------------
# 1. sys.path setup (kept from original)
# ---------------------------------------------------------------------------
PROJECT_ROOT = Path(__file__).resolve().parents[1]
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
# ---------------------------------------------------------------------------
# 2. MockTensor – lightweight stand-in when torch is unavailable
# ---------------------------------------------------------------------------
class MockTensor:
"""Minimal tensor mock that supports shape manipulation, arithmetic, and
conversion methods used by ComfyUI audio nodes."""
def __init__(
self,
shape: tuple[int, ...] = (1, 1, 44100),
dtype: str = "float32",
device: str = "cpu",
):
self._shape = tuple(shape)
self.dtype = dtype
self.device = device
# -- shape / dim --------------------------------------------------------
@property
def shape(self) -> tuple[int, ...]:
return self._shape
@property
def ndim(self) -> int:
return len(self._shape)
def dim(self) -> int:
return len(self._shape)
# -- device / dtype conversions -----------------------------------------
def to(self, *_args, **_kwargs) -> MockTensor:
return self
def float(self) -> MockTensor:
return MockTensor(self._shape, dtype="float32", device=self.device)
def cpu(self) -> MockTensor:
return MockTensor(self._shape, dtype=self.dtype, device="cpu")
# -- shape manipulation -------------------------------------------------
def squeeze(self, dim: int = 0) -> MockTensor:
s = list(self._shape)
if 0 <= dim < len(s) and s[dim] == 1 or dim < 0 and (dim + len(s)) >= 0 and s[dim] == 1:
s.pop(dim)
return MockTensor(tuple(s), dtype=self.dtype, device=self.device)
def unsqueeze(self, dim: int) -> MockTensor:
s = list(self._shape)
if dim < 0:
dim = len(s) + 1 + dim
s.insert(dim, 1)
return MockTensor(tuple(s), dtype=self.dtype, device=self.device)
def mean(self, dim: int | None = None, keepdim: bool = False) -> MockTensor:
if dim is None:
return MockTensor((1,), dtype=self.dtype, device=self.device)
s = list(self._shape)
if keepdim:
s[dim] = 1
else:
s.pop(dim)
return MockTensor(tuple(s), dtype=self.dtype, device=self.device)
# -- numpy conversion ---------------------------------------------------
def numpy(self) -> np.ndarray:
return np.zeros(self._shape, dtype=np.float32)
# -- arithmetic ---------------------------------------------------------
def _binop(self, other: object) -> MockTensor:
return MockTensor(self._shape, dtype=self.dtype, device=self.device)
__add__ = __radd__ = _binop
__sub__ = __rsub__ = _binop
__mul__ = __rmul__ = _binop
__truediv__ = __rtruediv__ = _binop
# -- slicing (e.g. tensor[..., start:end]) ------------------------------
def __getitem__(self, key: object) -> MockTensor:
if not isinstance(key, tuple):
key = (key,)
new_shape = list(self._shape)
real_idx = 0
for k in key:
if k is Ellipsis:
real_idx = len(new_shape) - (len(key) - 1 - list(key).index(Ellipsis))
continue
if isinstance(k, slice):
start = k.start or 0
stop = k.stop if k.stop is not None else new_shape[real_idx]
new_shape[real_idx] = max(stop - start, 0)
real_idx += 1
return MockTensor(tuple(new_shape), dtype=self.dtype, device=self.device)
def __repr__(self) -> str:
return f"MockTensor(shape={self._shape}, dtype={self.dtype})"
# ---------------------------------------------------------------------------
# 3. Autouse session fixture – stub comfy ecosystem modules only when absent
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True, scope="session")
def mock_comfy_modules(tmp_path_factory: pytest.TempPathFactory):
"""Injects lightweight stubs for comfy-ecosystem modules that are not
available in the test environment. Skips any module that is *already*
present in ``sys.modules`` so that test files which set up their own
mocks (e.g. ``test_audio_video_combine_node.py``) keep working."""
_tmp = tmp_path_factory.mktemp("comfy_temp")
installed: list[str] = []
# NOTE: torch, torchaudio, and librosa are NOT mocked here.
# Each test file sets up its own mocks with the fidelity it needs.
# -- comfy + comfy.model_management -------------------------------------
if "comfy" not in sys.modules:
comfy_stub = types.ModuleType("comfy")
comfy_stub.__path__ = []
sys.modules["comfy"] = comfy_stub
installed.append("comfy")
if "comfy.model_management" not in sys.modules:
mm_stub = types.ModuleType("comfy.model_management")
mm_stub.get_torch_device = lambda: "cpu"
sys.modules["comfy.model_management"] = mm_stub
if hasattr(sys.modules.get("comfy", None), "__path__"):
sys.modules["comfy"].model_management = mm_stub
installed.append("comfy.model_management")
# -- comfy_api (and sub-modules) ----------------------------------------
if "comfy_api" not in sys.modules:
comfy_api_stub = types.ModuleType("comfy_api")
comfy_api_stub.__path__ = []
sys.modules["comfy_api"] = comfy_api_stub
installed.append("comfy_api")
# -- folder_paths -------------------------------------------------------
if "folder_paths" not in sys.modules:
fp_stub = types.ModuleType("folder_paths")
fp_stub.get_temp_directory = lambda: str(_tmp)
sys.modules["folder_paths"] = fp_stub
installed.append("folder_paths")
yield
# Teardown: remove only what *we* installed
for mod_name in installed:
sys.modules.pop(mod_name, None)
# ---------------------------------------------------------------------------
# 4. make_audio fixture factory
# ---------------------------------------------------------------------------
def _make_tensor(shape, dtype="float32"):
"""Return a real torch.Tensor if torch is usable, else a MockTensor."""
try:
import torch as _torch
if hasattr(_torch, "zeros") and callable(_torch.zeros):
# Guard against our own stub leaking through
t = _torch.zeros(*shape)
if isinstance(t, MockTensor):
raise TypeError
return t
except Exception:
pass
return MockTensor(shape, dtype=dtype)
@pytest.fixture()
def make_audio():
"""Factory fixture that returns AUDIO TypedDicts.
Usage::
def test_something(make_audio):
audio = make_audio(shape=(1, 2, 44100), sample_rate=44100)
"""
def _factory(
shape: tuple[int, ...] = (1, 1, 44100),
sample_rate: int = 44100,
dtype: str = "float32",
) -> dict:
return {
"waveform": _make_tensor(shape, dtype=dtype),
"sample_rate": sample_rate,
}
return _factory
+136
View File
@@ -0,0 +1,136 @@
from __future__ import annotations
import importlib
import sys
import types
from pathlib import Path
import numpy as np
import pytest
class _DummyVideoFromFile:
def __init__(self, path: str):
self._path = path
def get_stream_source(self):
return self._path
if "torch" not in sys.modules:
sys.modules["torch"] = types.SimpleNamespace(Tensor=object)
if "torchaudio" not in sys.modules:
sys.modules["torchaudio"] = types.SimpleNamespace(save=lambda *_, **__: None)
if "comfy_api" not in sys.modules:
comfy_api = types.ModuleType("comfy_api")
comfy_api.__path__ = []
sys.modules["comfy_api"] = comfy_api
else:
comfy_api = sys.modules["comfy_api"]
input_impl_module = types.ModuleType("comfy_api.input_impl")
input_impl_module.VideoFromFile = _DummyVideoFromFile
sys.modules["comfy_api.input_impl"] = input_impl_module
comfy_api.input_impl = input_impl_module
input_module = types.ModuleType("comfy_api.input")
input_module.__path__ = []
video_types_module = types.ModuleType("comfy_api.input.video_types")
video_types_module.VideoInput = object
sys.modules["comfy_api.input"] = input_module
sys.modules["comfy_api.input.video_types"] = video_types_module
input_module.video_types = video_types_module
comfy_api.input = input_module
moviepy_module = types.ModuleType("moviepy")
editor_module = types.ModuleType("moviepy.editor")
moviepy_module.editor = editor_module
editor_module.VideoFileClip = object
editor_module.AudioFileClip = object
moviepy_module.VideoFileClip = object
moviepy_module.AudioFileClip = object
sys.modules.setdefault("moviepy", moviepy_module)
sys.modules.setdefault("moviepy.editor", editor_module)
folder_paths_module = types.ModuleType("folder_paths")
folder_paths_module.get_temp_directory = lambda: Path.cwd() / "temp"
sys.modules.setdefault("folder_paths", folder_paths_module)
module = importlib.import_module("src.combine_video_with_audio")
class StubVideo:
def __init__(self, source_path: str, duration: float):
self._source_path = source_path
self._duration = duration
def get_stream_source(self):
return self._source_path
def get_duration(self):
return self._duration
def save_to(self, path: str):
Path(path).write_bytes(b"")
@pytest.fixture
def dummy_audio():
return {
"waveform": np.zeros((1, 1, 1000), dtype=float),
"sample_rate": 100,
}
def test_audio_video_combine_replaces_audio(tmp_path, monkeypatch, dummy_audio):
calls = {}
class DummyVideoClip:
def __init__(self, path, audio=False):
calls["video_init"] = (path, audio)
def subclip(self, start, end):
calls["subclip"] = (start, end)
return self
def set_audio(self, audio_clip):
calls["set_audio"] = audio_clip
return self
def write_videofile(self, path, codec, audio_codec):
Path(path).write_bytes(b"video")
calls["write"] = (path, codec, audio_codec)
def close(self):
calls["video_closed"] = True
class DummyAudioClip:
def __init__(self, path):
calls["audio_init"] = path
def close(self):
calls["audio_closed"] = True
monkeypatch.setattr(module, "VideoFileClip", DummyVideoClip)
monkeypatch.setattr(module, "AudioFileClip", DummyAudioClip)
monkeypatch.setattr(module.folder_paths, "get_temp_directory", lambda: tmp_path)
def fake_save(filename, waveform, sample_rate):
Path(filename).write_bytes(b"wav")
calls["audio_samples"] = waveform.shape[-1]
calls["audio_sample_rate"] = sample_rate
monkeypatch.setattr(module.torchaudio, "save", fake_save)
stub_video = StubVideo(source_path=str(tmp_path / "source.mp4"), duration=12.0)
combine = module.AudioVideoCombine()
(result,) = combine.main(dummy_audio, stub_video, "0:05", "0:08")
output_path = result.get_stream_source()
assert Path(output_path).exists()
assert calls["subclip"] == (5.0, 8.0)
assert calls["write"][1:] == ("libx264", "aac")
assert calls["audio_sample_rate"] == 100
assert calls["audio_samples"] == 300
+42
View File
@@ -0,0 +1,42 @@
import math
import pytest
from src.audio_video_logic import (
AudioVideoCombineError,
compute_trim_window,
parse_timestamp,
)
def test_parse_timestamp_supports_seconds_minutes_and_hours():
assert math.isclose(parse_timestamp("42", default=0), 42.0)
assert math.isclose(parse_timestamp("1:30", default=0), 90.0)
assert math.isclose(parse_timestamp("2:01:05", default=0), 2 * 3600 + 65)
def test_parse_timestamp_requires_default_when_blank():
with pytest.raises(AudioVideoCombineError):
parse_timestamp("", default=None)
@pytest.mark.parametrize(
"start,end,duration,expected",
[
("0:00", "1:00", 90.0, (0.0, 60.0)),
("30", "120", 100.0, (30.0, 100.0)),
("0", "", 15.0, (0.0, 15.0)),
],
)
def test_compute_trim_window_clamps_and_defaults(start, end, duration, expected):
assert compute_trim_window(start, end, duration) == expected
def test_compute_trim_window_requires_end_time_if_duration_unknown():
with pytest.raises(AudioVideoCombineError):
compute_trim_window("0", "", None)
def test_compute_trim_window_validates_start_before_end():
with pytest.raises(AudioVideoCombineError):
compute_trim_window("5", "1", 10.0)
+241
View File
@@ -0,0 +1,241 @@
from __future__ import annotations
import importlib
import sys
import types
import numpy as np
import pytest
# ---------------------------------------------------------------------------
# MockTensor – numpy-backed tensor that supports .shape, .to(), slicing, math
# ---------------------------------------------------------------------------
class MockTensor:
def __init__(self, data: np.ndarray, device: str = "cpu"):
self._data = np.asarray(data, dtype=np.float64)
self.device = device
@property
def shape(self):
return self._data.shape
def to(self, device):
return MockTensor(self._data, device=str(device))
def __getitem__(self, key):
return MockTensor(self._data[key], device=self.device)
def __add__(self, other):
return MockTensor(self._data + _unwrap(other), device=self.device)
def __sub__(self, other):
return MockTensor(self._data - _unwrap(other), device=self.device)
def __mul__(self, other):
return MockTensor(self._data * _unwrap(other), device=self.device)
def __truediv__(self, other):
return MockTensor(self._data / _unwrap(other), device=self.device)
def __eq__(self, other):
return np.array_equal(self._data, _unwrap(other))
def numpy(self):
return self._data
def _unwrap(obj):
if isinstance(obj, MockTensor):
return obj._data
return obj
# ---------------------------------------------------------------------------
# Module-level mocks — must be registered before importing src.combine
# ---------------------------------------------------------------------------
# torch
torch_mock = types.ModuleType("torch")
torch_mock.Tensor = MockTensor
sys.modules["torch"] = torch_mock
# torchaudio + torchaudio.transforms
torchaudio_mock = types.ModuleType("torchaudio")
torchaudio_mock.__path__ = []
transforms_mock = types.ModuleType("torchaudio.transforms")
_resample_calls: list[dict] = []
class _MockResample:
def __init__(self, orig_freq: int, new_freq: int):
self.orig_freq = orig_freq
self.new_freq = new_freq
def to(self, device):
return self
def __call__(self, waveform: MockTensor) -> MockTensor:
_resample_calls.append({"orig_freq": self.orig_freq, "new_freq": self.new_freq})
# Return the waveform unchanged (resampling logic is not under test)
return waveform
transforms_mock.Resample = _MockResample
torchaudio_mock.transforms = transforms_mock
sys.modules["torchaudio"] = torchaudio_mock
sys.modules["torchaudio.transforms"] = transforms_mock
# comfy + comfy.model_management
comfy_mock = types.ModuleType("comfy")
comfy_mock.__path__ = []
model_management_mock = types.ModuleType("comfy.model_management")
model_management_mock.get_torch_device = lambda: "cpu"
comfy_mock.model_management = model_management_mock
sys.modules["comfy"] = comfy_mock
sys.modules["comfy.model_management"] = model_management_mock
# Clear any previously-cached src modules so they reimport with our mocks
for _key in list(sys.modules):
if _key.startswith("src."):
del sys.modules[_key]
# Now import the module under test
module = importlib.import_module("src.combine")
AudioCombine = module.AudioCombine
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _audio(values, sample_rate: int = 44100) -> dict:
"""Create an AUDIO dict with a 1×1×N MockTensor waveform."""
arr = np.array(values, dtype=np.float64).reshape(1, 1, -1)
return {"waveform": MockTensor(arr), "sample_rate": sample_rate}
@pytest.fixture(autouse=True)
def _clear_resample_calls():
_resample_calls.clear()
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestInputTypesSchema:
def test_required_keys(self):
schema = AudioCombine.INPUT_TYPES()
assert "audio_1" in schema["required"]
assert "audio_2" in schema["required"]
def test_method_options(self):
schema = AudioCombine.INPUT_TYPES()
method_opts = schema["optional"]["method"]
assert set(method_opts[0]) == {"add", "mean", "subtract", "multiply", "divide"}
assert method_opts[1]["default"] == "add"
class TestClassAttributes:
def test_return_types(self):
assert AudioCombine.RETURN_TYPES == ("AUDIO",)
def test_category(self):
assert AudioCombine.CATEGORY == "audio"
def test_function_name(self):
assert AudioCombine.FUNCTION == "main"
class TestSameSampleRateAdd:
def test_basic_add(self):
a1 = _audio([1.0, 2.0, 3.0])
a2 = _audio([4.0, 5.0, 6.0])
(result,) = AudioCombine().main(a1, a2, method="add")
np.testing.assert_array_equal(result["waveform"].numpy().flatten(), [5.0, 7.0, 9.0])
assert result["sample_rate"] == 44100
class TestAllMethods:
@pytest.mark.parametrize(
"method, expected",
[
("add", [5.0, 7.0, 9.0]),
("subtract", [-3.0, -3.0, -3.0]),
("multiply", [4.0, 10.0, 18.0]),
("divide", [0.25, 0.4, 0.5]),
("mean", [2.5, 3.5, 4.5]),
],
)
def test_method(self, method, expected):
a1 = _audio([1.0, 2.0, 3.0])
a2 = _audio([4.0, 5.0, 6.0])
(result,) = AudioCombine().main(a1, a2, method=method)
np.testing.assert_allclose(result["waveform"].numpy().flatten(), expected)
class TestDifferentLengths:
def test_truncation_to_shorter(self):
a1 = _audio([1.0, 2.0, 3.0, 4.0, 5.0])
a2 = _audio([10.0, 20.0, 30.0])
(result,) = AudioCombine().main(a1, a2, method="add")
out = result["waveform"].numpy().flatten()
assert len(out) == 3
np.testing.assert_array_equal(out, [11.0, 22.0, 33.0])
def test_truncation_second_longer(self):
a1 = _audio([1.0, 2.0])
a2 = _audio([10.0, 20.0, 30.0, 40.0])
(result,) = AudioCombine().main(a1, a2, method="add")
out = result["waveform"].numpy().flatten()
assert len(out) == 2
np.testing.assert_array_equal(out, [11.0, 22.0])
class TestDifferentSampleRates:
def test_lower_rate_gets_resampled_upward(self):
a1 = _audio([1.0, 2.0, 3.0], sample_rate=22050)
a2 = _audio([4.0, 5.0, 6.0], sample_rate=44100)
(result,) = AudioCombine().main(a1, a2, method="add")
assert result["sample_rate"] == 44100
assert len(_resample_calls) == 1
assert _resample_calls[0] == {"orig_freq": 22050, "new_freq": 44100}
def test_higher_rate_stays_when_second_is_lower(self):
a1 = _audio([1.0, 2.0, 3.0], sample_rate=48000)
a2 = _audio([4.0, 5.0, 6.0], sample_rate=16000)
(result,) = AudioCombine().main(a1, a2, method="add")
assert result["sample_rate"] == 48000
assert len(_resample_calls) == 1
assert _resample_calls[0] == {"orig_freq": 16000, "new_freq": 48000}
def test_resampled_waveform_moved_to_cpu_first_lower(self):
"""Fix #23: resampled waveform_1 should be moved back to cpu."""
a1 = _audio([1.0, 2.0, 3.0], sample_rate=22050)
a1["waveform"] = a1["waveform"].to("cuda")
a2 = _audio([4.0, 5.0, 6.0], sample_rate=44100)
a2["waveform"] = a2["waveform"].to("cuda")
(result,) = AudioCombine().main(a1, a2, method="add")
assert result["waveform"].device == "cpu"
def test_resampled_waveform_moved_to_cpu_second_lower(self):
"""Fix #23: resampled waveform_2 should be moved back to cpu."""
a1 = _audio([1.0, 2.0, 3.0], sample_rate=48000)
a1["waveform"] = a1["waveform"].to("cuda")
a2 = _audio([4.0, 5.0, 6.0], sample_rate=16000)
a2["waveform"] = a2["waveform"].to("cuda")
(result,) = AudioCombine().main(a1, a2, method="add")
assert result["waveform"].device == "cpu"
class TestUnsupportedMethod:
def test_raises_value_error(self):
a1 = _audio([1.0])
a2 = _audio([1.0])
with pytest.raises(ValueError, match="Unsupported combine method"):
AudioCombine().main(a1, a2, method="max")
+105
View File
@@ -0,0 +1,105 @@
from __future__ import annotations
import importlib
import sys
import types
import numpy as np
import pytest
# ---------------------------------------------------------------------------
# Mock torch before importing the module under test
# ---------------------------------------------------------------------------
class MockTensor:
"""Numpy-backed tensor supporting .shape and [..., start:end] slicing."""
def __init__(self, data: np.ndarray):
self._data = data
@property
def shape(self):
return self._data.shape
def __getitem__(self, key):
return MockTensor(self._data[key])
def __eq__(self, other):
if isinstance(other, MockTensor):
return np.array_equal(self._data, other._data)
return NotImplemented
if "torch" not in sys.modules:
_torch = types.SimpleNamespace(Tensor=object)
sys.modules["torch"] = _torch
module = importlib.import_module("src.crop")
AudioCrop = module.AudioCrop
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_audio(num_frames: int, sample_rate: int = 100):
waveform = MockTensor(np.arange(num_frames, dtype=np.float32).reshape(1, 1, -1))
return {"waveform": waveform, "sample_rate": sample_rate}
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestAudioCrop:
def test_basic_crop(self):
audio = _make_audio(5000, sample_rate=100)
(result,) = AudioCrop().main(audio, start_time="0:10", end_time="0:30")
assert result["sample_rate"] == 100
assert result["waveform"].shape == (1, 1, 2000)
def test_seconds_only_input(self):
audio = _make_audio(5000, sample_rate=100)
(result,) = AudioCrop().main(audio, start_time="10", end_time="30")
assert result["waveform"].shape == (1, 1, 2000)
def test_start_beyond_audio_length(self):
audio = _make_audio(500, sample_rate=100)
# start = 0:10 → frame 1000, but audio only has 500 frames → clamped to 499
# end = 0:30 → frame 3000 → clamped to 499
# start (499) == end (499) → empty slice
(result,) = AudioCrop().main(audio, start_time="0:10", end_time="0:30")
assert result["waveform"].shape[-1] == 0
def test_end_beyond_audio_length(self):
audio = _make_audio(2000, sample_rate=100)
# start = 0:05 → frame 500
# end = 0:30 → frame 3000 → clamped to 1999
(result,) = AudioCrop().main(audio, start_time="0:05", end_time="0:30")
assert result["waveform"].shape == (1, 1, 1499)
def test_start_greater_than_end_raises(self):
audio = _make_audio(5000, sample_rate=100)
with pytest.raises(ValueError, match="Start time must be less than end time"):
AudioCrop().main(audio, start_time="0:30", end_time="0:10")
def test_full_duration_crop(self):
audio = _make_audio(3000, sample_rate=100)
# end = 0:30 → frame 3000 → clamped to 2999
(result,) = AudioCrop().main(audio, start_time="0:00", end_time="0:30")
assert result["waveform"].shape == (1, 1, 2999)
def test_input_types_schema(self):
schema = AudioCrop.INPUT_TYPES()
required = schema["required"]
assert "audio" in required
assert "start_time" in required
assert "end_time" in required
def test_return_types(self):
assert AudioCrop.RETURN_TYPES == ("AUDIO",)
+137
View File
@@ -0,0 +1,137 @@
from __future__ import annotations
import importlib
import sys
import types
import numpy as np
# ---------------------------------------------------------------------------
# Mock heavy dependencies before importing the module under test
# ---------------------------------------------------------------------------
torch_mock = types.ModuleType("torch")
class MockTensor:
def __init__(self, data):
self._data = np.array(data) if not isinstance(data, np.ndarray) else data
@property
def shape(self):
return self._data.shape
@property
def ndim(self):
return self._data.ndim
def squeeze(self, dim=0):
return MockTensor(np.squeeze(self._data, axis=dim))
def unsqueeze(self, dim=0):
return MockTensor(np.expand_dims(self._data, axis=dim))
def float(self):
return self
def to(self, device):
return self
def numpy(self):
return self._data
torch_mock.Tensor = MockTensor
torch_mock.device = str
torch_mock.hann_window = lambda *a, **kw: MockTensor(np.ones(2048))
torch_mock.stft = lambda *a, **kw: MockTensor(np.zeros((2, 1025, 10)))
torch_mock.istft = lambda *a, **kw: MockTensor(np.zeros((2, 16000)))
torch_mock.linspace = lambda *a, **kw: MockTensor(np.zeros((1025, 1)))
torch_mock.cfloat = "torch.cfloat"
torch_mock.no_grad = type("_NoGrad", (), {"__enter__": lambda s: s, "__exit__": lambda s, *a: None})
sys.modules["torch"] = torch_mock
torchaudio_mock = types.ModuleType("torchaudio")
torchaudio_functional = types.ModuleType("torchaudio.functional")
torchaudio_functional.phase_vocoder = lambda *a, **kw: MockTensor(np.zeros((2, 1025, 10)))
torchaudio_mock.functional = torchaudio_functional
sys.modules["torchaudio"] = torchaudio_mock
sys.modules["torchaudio.functional"] = torchaudio_functional
librosa_mock = types.ModuleType("librosa")
librosa_onset = types.ModuleType("librosa.onset")
librosa_beat = types.ModuleType("librosa.beat")
librosa_mock.onset = librosa_onset
librosa_mock.beat = librosa_beat
sys.modules["librosa"] = librosa_mock
sys.modules["librosa.onset"] = librosa_onset
sys.modules["librosa.beat"] = librosa_beat
# Clear any previously-cached src modules so they reimport with our mocks
for _key in list(sys.modules):
if _key.startswith("src."):
del sys.modules[_key]
# ---------------------------------------------------------------------------
# Import module under test
# ---------------------------------------------------------------------------
module = importlib.import_module("src.get_tempo")
GetTempo = module.GetTempo
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_audio(sample_rate: int = 44100) -> dict:
"""Return a minimal AUDIO dict with a batch-dim waveform."""
waveform = MockTensor(np.random.randn(1, 2, 16000))
return {"waveform": waveform, "sample_rate": sample_rate}
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestGetTempoNode:
def test_input_types_has_required_audio(self):
schema = GetTempo.INPUT_TYPES()
assert "required" in schema
assert "audio" in schema["required"]
assert schema["required"]["audio"] == ("AUDIO",)
def test_return_types(self):
assert GetTempo.RETURN_TYPES == ("STRING", "FLOAT", "INTEGER")
def test_return_names(self):
assert GetTempo.RETURN_NAMES == ("tempo_string", "tempo_float", "tempo_integer")
def test_tempo_120_7(self, monkeypatch):
monkeypatch.setattr(module, "estimate_tempo", lambda w, sr: 120.7)
node = GetTempo()
result = node.main(_make_audio())
assert result == ("121", 120.7, 120)
def test_tempo_85_3(self, monkeypatch):
monkeypatch.setattr(module, "estimate_tempo", lambda w, sr: 85.3)
node = GetTempo()
result = node.main(_make_audio())
assert result == ("85", 85.3, 85)
def test_waveform_is_squeezed_before_call(self, monkeypatch):
"""estimate_tempo should receive a 2-d tensor (batch dim squeezed)."""
received = {}
def spy(waveform, sample_rate):
received["ndim"] = waveform.ndim
received["shape"] = waveform.shape
return 100.0
monkeypatch.setattr(module, "estimate_tempo", spy)
node = GetTempo()
node.main(_make_audio(44100))
assert received["ndim"] == 2, "waveform should be squeezed to 2-d"
+279
View File
@@ -0,0 +1,279 @@
"""Tests for node registration in __init__.py."""
from __future__ import annotations
import sys
import types
import numpy as np
import pytest
# ---------------------------------------------------------------------------
# Mock all heavy dependencies before importing any src modules.
#
# Use sys.modules[...] = ... (not setdefault) so these mocks take precedence
# even if conftest.py or another test file already inserted a partial mock.
# ---------------------------------------------------------------------------
# -- torch (needs nn.Module, device, Tensor, and several top-level functions) --
_torch = types.ModuleType("torch")
_torch.Tensor = object
_torch.device = str # just needs to be a type
_torch.stft = lambda *a, **kw: None
_torch.istft = lambda *a, **kw: None
_torch.hann_window = lambda *a, **kw: None
_torch.cat = lambda *a, **kw: None
_torch.zeros = lambda *a, **kw: None
_torch.linspace = lambda *a, **kw: None
_torch.cfloat = "torch.cfloat"
_torch.no_grad = type("_NoGrad", (), {"__enter__": lambda s: s, "__exit__": lambda s, *a: None})
_nn = types.ModuleType("torch.nn")
_nn.Module = type(
"Module",
(),
{"to": lambda self, d: self, "forward": lambda self, x: x, "sources": ["bass", "drums", "other", "vocals"]}, # noqa: E501
)
_torch.nn = _nn
sys.modules["torch"] = _torch
sys.modules["torch.nn"] = _nn
# -- torchaudio (transforms, pipelines, functional) --
_torchaudio = types.ModuleType("torchaudio")
_torchaudio.save = lambda *a, **kw: None
_transforms = types.ModuleType("torchaudio.transforms")
_transforms.Fade = object
_transforms.Resample = object
_torchaudio.transforms = _transforms
_pipelines = types.ModuleType("torchaudio.pipelines")
_pipelines.HDEMUCS_HIGH_MUSDB_PLUS = types.SimpleNamespace(get_model=lambda: None, sample_rate=44100)
_torchaudio.pipelines = _pipelines
_functional = types.ModuleType("torchaudio.functional")
_functional.phase_vocoder = lambda *a, **kw: None
_torchaudio.functional = _functional
sys.modules["torchaudio"] = _torchaudio
sys.modules["torchaudio.transforms"] = _transforms
sys.modules["torchaudio.pipelines"] = _pipelines
sys.modules["torchaudio.functional"] = _functional
# -- comfy --
_comfy = types.ModuleType("comfy")
_comfy.__path__ = []
_model_mgmt = types.ModuleType("comfy.model_management")
_model_mgmt.get_torch_device = lambda: "cpu"
_comfy.model_management = _model_mgmt
sys.modules["comfy"] = _comfy
sys.modules["comfy.model_management"] = _model_mgmt
# -- comfy_api --
_comfy_api = types.ModuleType("comfy_api")
_comfy_api.__path__ = []
sys.modules["comfy_api"] = _comfy_api
_input_impl = types.ModuleType("comfy_api.input_impl")
_input_impl.VideoFromFile = object
sys.modules["comfy_api.input_impl"] = _input_impl
_comfy_api.input_impl = _input_impl
_input_mod = types.ModuleType("comfy_api.input")
_input_mod.__path__ = []
_video_types = types.ModuleType("comfy_api.input.video_types")
_video_types.VideoInput = object
sys.modules["comfy_api.input"] = _input_mod
sys.modules["comfy_api.input.video_types"] = _video_types
_input_mod.video_types = _video_types
_comfy_api.input = _input_mod
# -- moviepy --
_moviepy = types.ModuleType("moviepy")
_moviepy_editor = types.ModuleType("moviepy.editor")
_moviepy_editor.VideoFileClip = object
_moviepy_editor.AudioFileClip = object
_moviepy.editor = _moviepy_editor
_moviepy.VideoFileClip = object
_moviepy.AudioFileClip = object
sys.modules.setdefault("moviepy", _moviepy)
sys.modules.setdefault("moviepy.editor", _moviepy_editor)
# -- folder_paths --
_folder_paths = types.ModuleType("folder_paths")
_folder_paths.get_temp_directory = lambda: "/tmp"
sys.modules.setdefault("folder_paths", _folder_paths)
# -- librosa --
_librosa = types.ModuleType("librosa")
_librosa_onset = types.ModuleType("librosa.onset")
_librosa_onset.onset_strength = lambda *a, **kw: np.zeros(10)
_librosa.onset = _librosa_onset
_librosa_beat = types.ModuleType("librosa.beat")
_librosa_beat.beat_track = lambda *a, **kw: (np.array([[120.0]]), None)
_librosa.beat = _librosa_beat
sys.modules["librosa"] = _librosa
sys.modules["librosa.onset"] = _librosa_onset
sys.modules["librosa.beat"] = _librosa_beat
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
ALL_NODE_KEYS = [
"AudioSeparation",
"AudioCrop",
"AudioCombine",
"AudioTempoMatch",
"AudioVideoCombine",
"AudioSpeedShift",
"AudioGetTempo",
]
# NODE_CLASS_MAPPINGS key -> fallback source module used by __init__.py
_KEY_TO_SRC = {
"AudioSeparation": "src.separation",
"AudioCrop": "src.crop",
"AudioCombine": "src.combine",
"AudioTempoMatch": "src.tempo_match",
"AudioVideoCombine": "src.combine_video_with_audio",
"AudioSpeedShift": "src.time_shift",
"AudioGetTempo": "src.get_tempo",
}
def _reload_init():
"""Force-reload __init__.py so NODE_CLASS_MAPPINGS is rebuilt."""
# Re-install our mocks — other test files may have overwritten them
sys.modules["torch"] = _torch
sys.modules["torch.nn"] = _nn
sys.modules["torchaudio"] = _torchaudio
sys.modules["torchaudio.transforms"] = _transforms
sys.modules["torchaudio.pipelines"] = _pipelines
sys.modules["torchaudio.functional"] = _functional
sys.modules["librosa"] = _librosa
sys.modules["librosa.onset"] = _librosa_onset
sys.modules["librosa.beat"] = _librosa_beat
# Clear cached src submodules so Python re-executes them on import
for mod_name in list(sys.modules):
if mod_name.startswith("src.") or mod_name == "src":
sys.modules.pop(mod_name, None)
# Remove the top-level __init__ itself so reload actually re-runs it
sys.modules.pop("__init__", None)
import __init__ as pkg # noqa: E0611
return pkg
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestAllNodesRegister:
"""All nodes register when imports succeed."""
@pytest.fixture(autouse=True)
def _setup(self):
self.pkg = _reload_init()
def test_all_keys_present(self):
for key in ALL_NODE_KEYS:
assert key in self.pkg.NODE_CLASS_MAPPINGS, f"{key} missing from NODE_CLASS_MAPPINGS"
def test_no_extra_keys(self):
assert set(self.pkg.NODE_CLASS_MAPPINGS.keys()) == set(ALL_NODE_KEYS)
def test_values_are_not_none(self):
for key in ALL_NODE_KEYS:
assert self.pkg.NODE_CLASS_MAPPINGS[key] is not None
class TestGracefulDegradation:
"""When a specific import fails, that node is missing but others still register."""
@pytest.mark.parametrize("broken_key", ALL_NODE_KEYS)
def test_single_broken_import(self, broken_key):
src_module = _KEY_TO_SRC[broken_key]
# Re-install our mocks — other test files may have overwritten them
sys.modules["torch"] = _torch
sys.modules["torch.nn"] = _nn
sys.modules["torchaudio"] = _torchaudio
sys.modules["torchaudio.transforms"] = _transforms
sys.modules["torchaudio.pipelines"] = _pipelines
sys.modules["torchaudio.functional"] = _functional
sys.modules["librosa"] = _librosa
sys.modules["librosa.onset"] = _librosa_onset
sys.modules["librosa.beat"] = _librosa_beat
# Clear all src modules so nothing is cached
for mod_name in list(sys.modules):
if mod_name.startswith("src.") or mod_name == "src":
sys.modules.pop(mod_name, None)
sys.modules.pop("__init__", None)
# Poison the target module: __getattr__ raises ImportError so
# ``from src.<mod> import <Class>`` fails.
class _Raiser(types.ModuleType):
def __getattr__(self, name):
raise ImportError(f"mocked failure for {src_module}")
sys.modules[src_module] = _Raiser(src_module)
try:
import __init__ as pkg # noqa: E0611
assert broken_key not in pkg.NODE_CLASS_MAPPINGS, (
f"{broken_key} should NOT be registered when its import fails"
)
for key in ALL_NODE_KEYS:
if key == broken_key:
continue
assert key in pkg.NODE_CLASS_MAPPINGS, f"{key} should still register when only {broken_key} is broken"
finally:
# Clean up so other tests get a fresh slate
sys.modules.pop(src_module, None)
sys.modules.pop("__init__", None)
class TestNodeClassAttributes:
"""Each registered class has the attributes ComfyUI expects."""
REQUIRED_ATTRS = ["INPUT_TYPES", "FUNCTION", "RETURN_TYPES", "CATEGORY"]
@pytest.fixture(autouse=True)
def _setup(self):
self.pkg = _reload_init()
@pytest.mark.parametrize("key", ALL_NODE_KEYS)
def test_has_required_attributes(self, key):
cls = self.pkg.NODE_CLASS_MAPPINGS[key]
for attr in self.REQUIRED_ATTRS:
assert hasattr(cls, attr), f"{key} missing attribute {attr}"
@pytest.mark.parametrize("key", ALL_NODE_KEYS)
def test_input_types_is_callable(self, key):
cls = self.pkg.NODE_CLASS_MAPPINGS[key]
assert callable(cls.INPUT_TYPES)
@pytest.mark.parametrize("key", ALL_NODE_KEYS)
def test_function_is_string(self, key):
cls = self.pkg.NODE_CLASS_MAPPINGS[key]
assert isinstance(cls.FUNCTION, str)
@pytest.mark.parametrize("key", ALL_NODE_KEYS)
def test_return_types_is_tuple(self, key):
cls = self.pkg.NODE_CLASS_MAPPINGS[key]
assert isinstance(cls.RETURN_TYPES, tuple)
@pytest.mark.parametrize("key", ALL_NODE_KEYS)
def test_category_is_string(self, key):
cls = self.pkg.NODE_CLASS_MAPPINGS[key]
assert isinstance(cls.CATEGORY, str)
+337
View File
@@ -0,0 +1,337 @@
"""Tests for src.resample.ChunkResampler."""
import sys
import types
import unittest
import numpy as np
# ---------------------------------------------------------------------------
# Mock torch, torchaudio, and comfy *before* importing the module under test
# ---------------------------------------------------------------------------
torch_mock = types.ModuleType("torch")
class MockTensor:
def __init__(self, data):
self._data = np.array(data) if not isinstance(data, np.ndarray) else data
@property
def shape(self):
return self._data.shape
def to(self, device):
self._last_device = device
return self
torch_mock.Tensor = MockTensor
class _NoGrad:
def __enter__(self):
return None
def __exit__(self, *args):
return None
torch_mock.no_grad = _NoGrad
torch_mock.split = lambda tensor, size, dim=-1: [tensor]
torch_mock.cat = lambda tensors, dim=-1: tensors[0]
sys.modules["torch"] = torch_mock
torchaudio_mock = types.ModuleType("torchaudio")
transforms_mock = types.ModuleType("torchaudio.transforms")
class MockResample:
def __init__(self, orig, new):
self.orig = orig
self.new = new
def to(self, device):
return self
def __call__(self, x):
return x
transforms_mock.Resample = MockResample
torchaudio_mock.transforms = transforms_mock
sys.modules["torchaudio"] = torchaudio_mock
sys.modules["torchaudio.transforms"] = transforms_mock
comfy_mock = types.ModuleType("comfy")
mm_mock = types.ModuleType("comfy.model_management")
mm_mock.get_torch_device = lambda: "cpu"
comfy_mock.model_management = mm_mock
sys.modules["comfy"] = comfy_mock
sys.modules["comfy.model_management"] = mm_mock
# Now safe to import -------------------------------------------------------
from src.resample import ChunkResampler # noqa: E402
# ===========================================================================
# reduce_ratio tests
# ===========================================================================
class TestReduceRatio(unittest.TestCase):
"""Tests for the static ChunkResampler.reduce_ratio method."""
def test_simple_halving(self):
self.assertEqual(ChunkResampler.reduce_ratio(44100, 22050), (2, 1))
def test_48000_44100(self):
a, b = ChunkResampler.reduce_ratio(48000, 44100)
# Must be reduced integers smaller than originals
self.assertIsInstance(a, int)
self.assertIsInstance(b, int)
self.assertLessEqual(a, 48000)
self.assertLessEqual(b, 44100)
# The ratio must be preserved
self.assertAlmostEqual(a / b, 48000 / 44100, places=5)
def test_identity(self):
self.assertEqual(ChunkResampler.reduce_ratio(44100, 44100), (1, 1))
def test_float_inputs(self):
a, b = ChunkResampler.reduce_ratio(44100.0, 22050.0)
self.assertEqual((a, b), (2, 1))
def test_large_numbers_fallback(self):
"""Very large coprime-ish numbers should hit max_attempts and fall back."""
big1 = 1_000_003 # large prime
big2 = 1_000_033 # another large prime
a, b = ChunkResampler.reduce_ratio(big1, big2)
# Fallback returns int(originals)
self.assertEqual((a, b), (big1, big2))
# ===========================================================================
# Constructor tests
# ===========================================================================
class TestConstructor(unittest.TestCase):
"""Tests for ChunkResampler.__init__."""
def test_negative_orig_freq_raises(self):
with self.assertRaises(ValueError):
ChunkResampler(-1, 44100)
def test_negative_new_freq_raises(self):
with self.assertRaises(ValueError):
ChunkResampler(44100, -1)
def test_zero_orig_freq_raises(self):
with self.assertRaises(ValueError):
ChunkResampler(0, 44100)
def test_zero_new_freq_raises(self):
with self.assertRaises(ValueError):
ChunkResampler(44100, 0)
# -- clamping -----------------------------------------------------------
def test_ratio_above_upper_clamp(self):
"""change_ratio > UPPER_CLAMP (1.1832) → new_freq is clamped."""
orig, new = 44100, 60000 # ratio ≈ 1.36
r = ChunkResampler(orig, new)
# After clamping, the effective ratio passed to Resample should be ≈ 1.1832
self.assertAlmostEqual(r.resample.new / r.resample.orig, 1.1832, places=2)
def test_ratio_below_lower_clamp(self):
"""change_ratio < LOWER_CLAMP (0.945) → new_freq is clamped."""
orig, new = 44100, 40000 # ratio ≈ 0.907
r = ChunkResampler(orig, new)
# After clamping, the effective ratio passed to Resample should be ≈ 0.945
self.assertAlmostEqual(r.resample.new / r.resample.orig, 0.945, places=2)
def test_ratio_within_bounds(self):
"""Ratio inside [LOWER_CLAMP, UPPER_CLAMP] → no clamping."""
orig, new = 44100, 44100 # ratio = 1.0
r = ChunkResampler(orig, new)
self.assertIsNotNone(r)
# -- chunk_size_seconds adjustment --------------------------------------
def test_chunk_size_large_diff(self):
"""diff > 0.08 → chunk_size_seconds capped at 1."""
# ratio = 48000/44100 ≈ 1.0884 → diff ≈ 0.0884 > 0.08
r = ChunkResampler(44100, 48000)
self.assertEqual(r.chunk_size_seconds, 1)
def test_chunk_size_medium_diff(self):
"""0.002 < diff ≤ 0.08 → chunk_size_seconds capped at 2."""
# ratio = 44100/43000 ≈ 1.0256 → diff ≈ 0.0256
r = ChunkResampler(43000, 44100)
self.assertEqual(r.chunk_size_seconds, 2)
def test_chunk_size_small_diff(self):
"""diff ≤ 0.002 → chunk_size_seconds capped at 4."""
# ratio = 1.0 → diff = 0
r = ChunkResampler(44100, 44100, chunk_size_seconds=10)
self.assertEqual(r.chunk_size_seconds, 4)
def test_chunk_size_seconds_truncated_to_int(self):
r = ChunkResampler(44100, 44100, chunk_size_seconds=3.7)
self.assertIsInstance(r.chunk_size_seconds, int)
# -- custom clamp values ------------------------------------------------
def test_custom_upper_clamp(self):
"""Custom upper_clamp should override the default."""
# Default UPPER_CLAMP is 1.1832; ratio 1.25 would be clamped by default
# With upper_clamp=1.3, it should NOT be clamped
r = ChunkResampler(44100, 55125, upper_clamp=1.3)
# MockResample stores the reduced freqs; verify the ratio is preserved (not clamped)
self.assertAlmostEqual(r.resample.orig / r.resample.new, 44100 / 55125, places=3)
def test_custom_lower_clamp(self):
"""Custom lower_clamp should override the default."""
# Default LOWER_CLAMP is 0.945; ratio 0.9 would be clamped by default
# With lower_clamp=0.8, it should NOT be clamped
r = ChunkResampler(44100, 39690, lower_clamp=0.8)
self.assertAlmostEqual(r.resample.orig / r.resample.new, 44100 / 39690, places=3)
def test_inverted_clamps_raises(self):
"""upper_clamp <= lower_clamp should raise ValueError."""
with self.assertRaises(ValueError):
ChunkResampler(44100, 44100, upper_clamp=0.5, lower_clamp=0.8)
def test_zero_clamp_raises(self):
"""Zero clamp values should raise ValueError."""
with self.assertRaises(ValueError):
ChunkResampler(44100, 44100, upper_clamp=0.0)
def test_negative_clamp_raises(self):
"""Negative clamp values should raise ValueError."""
with self.assertRaises(ValueError):
ChunkResampler(44100, 44100, lower_clamp=-0.5)
# ===========================================================================
# tolerance parameter tests
# ===========================================================================
class TestTolerance(unittest.TestCase):
"""Tests for the tolerance parameter."""
def test_tolerance_negative_raises(self):
with self.assertRaises(ValueError):
ChunkResampler(44100, 48000, tolerance=-0.1)
def test_tolerance_above_one_raises(self):
with self.assertRaises(ValueError):
ChunkResampler(44100, 48000, tolerance=1.5)
def test_tolerance_zero_no_change(self):
"""tolerance=0.0 should behave exactly like no tolerance."""
r = ChunkResampler(44100, 44100, tolerance=0.0)
self.assertIsNotNone(r)
def test_tolerance_finds_better_gcd(self):
"""With tolerance, the chosen freq should have a GCD >= the original target's GCD."""
import math
orig = 44100
target = 48001 # coprime-ish with 44100
original_gcd = math.gcd(orig, target)
ChunkResampler(orig, target, tolerance=0.01)
# _find_optimal_freq should have found a better candidate
# We can't check self.new_freq directly because reduce_ratio transforms it,
# but we can test the static method directly
optimal = ChunkResampler._find_optimal_freq(orig, target, 0.01)
optimal_gcd = math.gcd(orig, int(optimal))
self.assertGreaterEqual(optimal_gcd, original_gcd)
def test_tiny_tolerance_returns_target(self):
"""Very small tolerance should produce zero margin and return the target unchanged."""
result = ChunkResampler._find_optimal_freq(44100, 48000, 1e-8)
# margin = int(48000 * 1e-8) = 0, so only the target itself is checked
self.assertEqual(result, 48000.0)
def test_find_optimal_freq_prefers_44100(self):
"""44100 should be found when searching near 44100 ± tolerance."""
result = ChunkResampler._find_optimal_freq(44100, 44050, 0.01)
# 44100 is in range and gcd(44100, 44100) = 44100 which is maximal
self.assertEqual(result, 44100.0)
def test_find_optimal_freq_exact_when_already_best(self):
"""If the target already has the best GCD, it should be returned as-is."""
result = ChunkResampler._find_optimal_freq(44100, 22050, 0.001)
# 22050 divides 44100 evenly → gcd = 22050, hard to beat
self.assertEqual(result, 22050.0)
def test_tolerance_one_is_max(self):
"""tolerance=1.0 is the maximum allowed value."""
r = ChunkResampler(44100, 44100, tolerance=1.0)
self.assertIsNotNone(r)
def test_margin_capped_at_1000(self):
"""Search margin should be capped at 1000 candidates to avoid O(n) scan."""
# With tolerance=1.0 and target=192000, uncapped margin would be ~192000
# Capped at 1000, so search range is [191000, 193000]
result = ChunkResampler._find_optimal_freq(44100, 192000, 1.0)
# Should complete quickly and return a valid result
self.assertGreater(result, 0)
# Result should be within ±1000 of target
self.assertLessEqual(abs(result - 192000), 1000)
def test_chunk_size_uses_effective_ratio_after_clamp(self):
"""chunk_size_seconds should be based on effective ratio after clamping."""
# orig=44100, new=40000, ratio=0.907 → below LOWER_CLAMP (0.945)
# Clamped to 44100*0.945 = 41674.5, effective diff = |1 - 0.945| = 0.055
# 0.002 < 0.055 ≤ 0.08 → chunk_size_seconds capped at 2
r = ChunkResampler(44100, 40000)
self.assertEqual(r.chunk_size_seconds, 2)
# ===========================================================================
# __call__ tests
# ===========================================================================
class TestCall(unittest.TestCase):
"""Tests for ChunkResampler.__call__."""
def setUp(self):
self.resampler = ChunkResampler(44100, 44100)
def test_waveform_moved_to_device(self):
wav = MockTensor(np.zeros((1, 44100)))
self.resampler(wav)
# .to() was called with the device during __call__
self.assertTrue(hasattr(wav, "_last_device"))
def test_result_on_cpu(self):
wav = MockTensor(np.zeros((1, 44100)))
result = self.resampler(wav)
self.assertEqual(result._last_device, "cpu")
def test_split_and_cat_called(self):
"""Verify the chunk→resample→cat pipeline runs without error."""
calls = {"split": 0, "cat": 0}
orig_split = torch_mock.split
orig_cat = torch_mock.cat
def counting_split(tensor, size, dim=-1):
calls["split"] += 1
return orig_split(tensor, size, dim)
def counting_cat(tensors, dim=-1):
calls["cat"] += 1
return orig_cat(tensors, dim)
torch_mock.split = counting_split
torch_mock.cat = counting_cat
try:
wav = MockTensor(np.zeros((1, 44100)))
self.resampler(wav)
self.assertGreaterEqual(calls["split"], 1)
self.assertGreaterEqual(calls["cat"], 1)
finally:
torch_mock.split = orig_split
torch_mock.cat = orig_cat
if __name__ == "__main__":
unittest.main()
+655
View File
@@ -0,0 +1,655 @@
from __future__ import annotations
import importlib
import sys
import types
from unittest.mock import MagicMock, patch
from zipfile import BadZipFile
import numpy as np
import pytest
# ---------------------------------------------------------------------------
# MockTensor – lightweight numpy-backed stand-in for torch.Tensor
# ---------------------------------------------------------------------------
class MockTensor:
"""Numpy-backed tensor mock supporting the operations used by separation.py."""
def __init__(self, data=None, *, shape=None, device="cpu"):
if data is not None:
self._data = np.array(data, dtype=np.float32)
elif shape is not None:
self._data = np.zeros(shape, dtype=np.float32)
else:
self._data = np.zeros(0, dtype=np.float32)
self.device = device
# -- shape / indexing ---------------------------------------------------
@property
def shape(self):
return self._data.shape
def __len__(self):
return self._data.shape[0]
def __getitem__(self, key):
result = self._data[key]
if isinstance(result, np.ndarray):
t = MockTensor(result, device=self.device)
return t
return result
def __setitem__(self, key, value):
if isinstance(value, MockTensor):
self._data[key] = value._data
else:
self._data[key] = value
def __iter__(self):
for i in range(len(self)):
yield self[i]
# -- device movement ----------------------------------------------------
def to(self, device):
t = MockTensor(self._data.copy(), device=str(device))
return t
def cpu(self):
return MockTensor(self._data.copy(), device="cpu")
# -- shape manipulation -------------------------------------------------
def squeeze(self, dim=None):
return MockTensor(np.squeeze(self._data, axis=dim), device=self.device)
def unsqueeze(self, dim):
return MockTensor(np.expand_dims(self._data, axis=dim), device=self.device)
# -- reductions ---------------------------------------------------------
def mean(self, dim=None, keepdim=False):
if dim is None:
return MockTensor(np.array(self._data.mean(), dtype=np.float32), device=self.device)
return MockTensor(self._data.mean(axis=dim, keepdims=keepdim), device=self.device)
def std(self, *args, **kwargs):
val = self._data.std()
if val == 0:
val = 1.0
return MockTensor(np.array(val, dtype=np.float32), device=self.device)
# -- arithmetic ---------------------------------------------------------
def __sub__(self, other):
o = other._data if isinstance(other, MockTensor) else other
return MockTensor(self._data - o, device=self.device)
def __truediv__(self, other):
o = other._data if isinstance(other, MockTensor) else other
return MockTensor(self._data / o, device=self.device)
def __mul__(self, other):
o = other._data if isinstance(other, MockTensor) else other
return MockTensor(self._data * o, device=self.device)
def __add__(self, other):
o = other._data if isinstance(other, MockTensor) else other
return MockTensor(self._data + o, device=self.device)
def __iadd__(self, other):
o = other._data if isinstance(other, MockTensor) else other
self._data = self._data + o
return self
def __repr__(self):
return f"MockTensor(shape={self.shape}, device={self.device!r})"
def float(self):
return MockTensor(self._data.astype(np.float32), device=self.device)
@property
def ndim(self):
return self._data.ndim
# ---------------------------------------------------------------------------
# Mock torch / torchaudio / comfy – installed before importing separation.py
# ---------------------------------------------------------------------------
def _mock_zeros(*shape_args, device="cpu", **kwargs):
if len(shape_args) == 1 and isinstance(shape_args[0], (tuple, list)):
shape = tuple(shape_args[0])
else:
shape = tuple(shape_args)
return MockTensor(shape=shape, device=device)
# --- torch -----------------------------------------------------------------
torch_mock = types.ModuleType("torch")
torch_mock.Tensor = MockTensor
torch_mock.zeros = _mock_zeros
torch_mock.device = str # torch.device(x) → str(x)
torch_mock.no_grad = MagicMock(
return_value=MagicMock(__enter__=MagicMock(return_value=None), __exit__=MagicMock(return_value=False))
)
torch_mock.nn = types.SimpleNamespace(Module=object)
torch_mock.cfloat = "cfloat"
sys.modules["torch"] = torch_mock
# --- torchaudio & submodules -----------------------------------------------
torchaudio_mock = types.ModuleType("torchaudio")
torchaudio_transforms = types.ModuleType("torchaudio.transforms")
torchaudio_pipelines = types.ModuleType("torchaudio.pipelines")
# Fade: record __init__ args and act as identity on __call__
class _MockFade:
def __init__(self, fade_in_len=0, fade_out_len=0, fade_shape="linear"):
self.fade_in_len = fade_in_len
self.fade_out_len = fade_out_len
self.fade_shape = fade_shape
def __call__(self, x):
return x # identity – keeps tensor shapes unchanged
# Resample: record args, callable, supports .to(device)
class _MockResample:
def __init__(self, orig_freq, new_freq):
self.orig_freq = orig_freq
self.new_freq = new_freq
def to(self, device):
return self
def __call__(self, waveform):
return waveform # identity for simplicity
torchaudio_transforms.Fade = _MockFade
torchaudio_transforms.Resample = _MockResample
# Bundle mock – model w/ .sources and .forward
_MODEL_SR = 44100
class _MockModel:
sources = ["bass", "drums", "other", "vocals"]
def to(self, device):
return self
def forward(self, chunk):
# chunk shape: [batch, channels, chunk_len]
batch = chunk.shape[0]
channels = chunk.shape[1]
chunk_len = chunk.shape[2]
return MockTensor(shape=(batch, len(self.sources), channels, chunk_len))
_mock_bundle = types.SimpleNamespace(
get_model=lambda: _MockModel(),
sample_rate=_MODEL_SR,
)
torchaudio_pipelines.HDEMUCS_HIGH_MUSDB_PLUS = _mock_bundle
torchaudio_mock.transforms = torchaudio_transforms
torchaudio_mock.pipelines = torchaudio_pipelines
sys.modules["torchaudio"] = torchaudio_mock
sys.modules["torchaudio.transforms"] = torchaudio_transforms
sys.modules["torchaudio.pipelines"] = torchaudio_pipelines
# --- comfy -----------------------------------------------------------------
comfy_mod = types.ModuleType("comfy")
comfy_mod.__path__ = []
comfy_mm = types.ModuleType("comfy.model_management")
comfy_mm.get_torch_device = lambda: "cuda"
comfy_mod.model_management = comfy_mm
sys.modules["comfy"] = comfy_mod
sys.modules["comfy.model_management"] = comfy_mm
# --- src._types (AUDIO is just a TypedDict; stub the import) ---------------
src_types = types.ModuleType("src._types")
src_types.AUDIO = dict
sys.modules["src._types"] = src_types
# --- src.utils – provide a real-enough ensure_stereo ----------------------
src_utils = types.ModuleType("src.utils")
def _ensure_stereo(audio):
"""If mono (1, N), duplicate to (2, N). Otherwise pass through."""
if audio.ndim == 2 and audio.shape[0] == 1:
return MockTensor(np.concatenate([audio._data, audio._data], axis=0))
return audio
src_utils.ensure_stereo = _ensure_stereo
sys.modules["src.utils"] = src_utils
# Now import the module under test
separation = importlib.import_module("src.separation")
AudioSeparation = separation.AudioSeparation
# ===========================================================================
# Helpers
# ===========================================================================
def _make_audio(*, channels=2, frames=44100, sample_rate=44100):
"""Return an AUDIO dict with a MockTensor waveform [1, channels, frames]."""
waveform = MockTensor(np.random.randn(1, channels, frames).astype(np.float32))
return {"waveform": waveform, "sample_rate": sample_rate}
# ===========================================================================
# 1. INPUT_TYPES schema
# ===========================================================================
class TestInputTypes:
def test_has_required_audio(self):
schema = AudioSeparation.INPUT_TYPES()
assert "required" in schema
assert "audio" in schema["required"]
assert schema["required"]["audio"] == ("AUDIO",)
def test_has_optional_keys(self):
schema = AudioSeparation.INPUT_TYPES()
optional = schema["optional"]
assert "chunk_fade_shape" in optional
assert "chunk_length" in optional
assert "chunk_overlap" in optional
def test_fade_shape_options(self):
schema = AudioSeparation.INPUT_TYPES()
fade_choices = schema["optional"]["chunk_fade_shape"][0]
assert set(fade_choices) == {"linear", "half_sine", "logarithmic", "exponential"}
def test_chunk_length_is_float(self):
schema = AudioSeparation.INPUT_TYPES()
assert schema["optional"]["chunk_length"][0] == "FLOAT"
def test_chunk_overlap_is_float(self):
schema = AudioSeparation.INPUT_TYPES()
assert schema["optional"]["chunk_overlap"][0] == "FLOAT"
def test_chunk_length_default(self):
schema = AudioSeparation.INPUT_TYPES()
assert schema["optional"]["chunk_length"][1]["default"] == 10.0
def test_chunk_overlap_default(self):
schema = AudioSeparation.INPUT_TYPES()
assert schema["optional"]["chunk_overlap"][1]["default"] == 0.1
def test_fade_shape_default(self):
schema = AudioSeparation.INPUT_TYPES()
assert schema["optional"]["chunk_fade_shape"][1]["default"] == "linear"
# ===========================================================================
# 2. RETURN_TYPES / RETURN_NAMES
# ===========================================================================
class TestReturnMeta:
def test_return_types(self):
assert AudioSeparation.RETURN_TYPES == ("AUDIO", "AUDIO", "AUDIO", "AUDIO")
def test_return_names(self):
assert AudioSeparation.RETURN_NAMES == ("Bass", "Drums", "Other", "Vocals")
def test_function_attr(self):
assert AudioSeparation.FUNCTION == "main"
def test_category(self):
assert AudioSeparation.CATEGORY == "audio"
# ===========================================================================
# 3. sources_to_tuple
# ===========================================================================
class TestSourcesToTuple:
def setup_method(self):
self.node = AudioSeparation()
self.node.model_sample_rate = _MODEL_SR
def _make_sources(self, keys=None):
keys = keys or ["bass", "drums", "other", "vocals"]
return {k: MockTensor(shape=(2, 44100)) for k in keys}
def test_valid_sources_returns_4_tuple(self):
result = self.node.sources_to_tuple(self._make_sources())
assert isinstance(result, tuple)
assert len(result) == 4
def test_output_order(self):
sources = self._make_sources()
result = self.node.sources_to_tuple(sources)
for i, _name in enumerate(["bass", "drums", "other", "vocals"]):
assert result[i]["sample_rate"] == _MODEL_SR
def test_each_output_is_audio_dict(self):
result = self.node.sources_to_tuple(self._make_sources())
for audio in result:
assert "waveform" in audio
assert "sample_rate" in audio
assert audio["sample_rate"] == _MODEL_SR
def test_waveform_has_batch_dim(self):
"""sources_to_tuple calls .unsqueeze(0) → waveform should be 3-D."""
result = self.node.sources_to_tuple(self._make_sources())
for audio in result:
assert audio["waveform"].ndim == 3
assert audio["waveform"].shape[0] == 1
def test_waveform_is_cpu(self):
sources = {k: MockTensor(shape=(2, 44100), device="cuda") for k in ["bass", "drums", "other", "vocals"]}
result = self.node.sources_to_tuple(sources)
for audio in result:
assert audio["waveform"].device == "cpu"
def test_missing_source_raises(self):
sources = self._make_sources(["bass", "drums", "other"]) # no vocals
with pytest.raises(ValueError, match="Missing source vocals"):
self.node.sources_to_tuple(sources)
def test_missing_bass_raises(self):
sources = self._make_sources(["drums", "other", "vocals"])
with pytest.raises(ValueError, match="Missing source bass"):
self.node.sources_to_tuple(sources)
def test_extra_sources_ignored(self):
sources = self._make_sources()
sources["extra"] = MockTensor(shape=(2, 44100))
result = self.node.sources_to_tuple(sources)
assert len(result) == 4
# ===========================================================================
# 4. separate_sources – chunking logic
# ===========================================================================
class TestSeparateSources:
def setup_method(self):
self.node = AudioSeparation()
self.model = _MockModel()
def test_single_chunk_one_forward_call(self):
"""Audio shorter than one chunk → exactly one model.forward call."""
sr = 44100
frames = int(sr * 5) # 5 seconds, segment=10 → fits in one chunk
mix = MockTensor(shape=(1, 2, frames))
self.model.forward = MagicMock(side_effect=lambda c: MockTensor(shape=(c.shape[0], 4, c.shape[1], c.shape[2])))
result = self.node.separate_sources(self.model, mix, sr, segment=10.0, overlap=0.1)
assert self.model.forward.call_count == 1
assert result.shape == (1, 4, 2, frames)
def test_multiple_chunks_multiple_forward_calls(self):
"""Audio much longer than chunk → multiple forward calls."""
sr = 44100
frames = int(sr * 30) # 30 seconds with segment=10
mix = MockTensor(shape=(1, 2, frames))
call_count = 0
def _forward(c):
nonlocal call_count
call_count += 1
return MockTensor(shape=(c.shape[0], 4, c.shape[1], c.shape[2]))
self.model.forward = _forward
self.node.separate_sources(self.model, mix, sr, segment=10.0, overlap=0.1)
assert call_count > 1
def test_fade_parameters_set_correctly(self):
"""Fade should be created with fade_in_len=0 and the given fade_shape."""
sr = 44100
frames = int(sr * 5)
mix = MockTensor(shape=(1, 2, frames))
init_kwargs_list = []
original_fade = _MockFade
class _SpyFade(original_fade):
def __init__(self, **kwargs):
super().__init__(**kwargs)
# Capture the *initial* construction args before any mutation
init_kwargs_list.append(dict(kwargs))
with patch.object(separation, "Fade", _SpyFade):
self.node.separate_sources(
self.model,
mix,
sr,
segment=10.0,
overlap=0.1,
chunk_fade_shape="half_sine",
)
assert len(init_kwargs_list) == 1
init_kw = init_kwargs_list[0]
assert init_kw["fade_in_len"] == 0
assert init_kw["fade_shape"] == "half_sine"
expected_overlap_frames = int(0.1 * sr)
assert init_kw["fade_out_len"] == expected_overlap_frames
def test_fade_in_set_after_first_chunk(self):
"""After the first chunk, fade_in_len should be set to overlap_frames."""
sr = 44100
frames = int(sr * 30) # long enough for multiple chunks
mix = MockTensor(shape=(1, 2, frames))
created_fades = []
original_fade = _MockFade
class _SpyFade(original_fade):
def __init__(self, **kwargs):
super().__init__(**kwargs)
created_fades.append(self)
with patch.object(separation, "Fade", _SpyFade):
self.node.separate_sources(self.model, mix, sr, segment=10.0, overlap=0.1)
fade = created_fades[0]
expected_overlap = int(0.1 * sr)
# After first iteration, fade_in_len should have been set
assert fade.fade_in_len == expected_overlap
def test_output_shape_matches_input(self):
sr = 44100
frames = int(sr * 8)
mix = MockTensor(shape=(1, 2, frames))
result = self.node.separate_sources(self.model, mix, sr, segment=10.0, overlap=0.1)
assert result.shape == (1, 4, 2, frames)
def test_device_none_uses_mix_device(self):
"""When device=None, should use mix.device."""
sr = 44100
frames = int(sr * 2)
mix = MockTensor(shape=(1, 2, frames), device="cpu")
result = self.node.separate_sources(self.model, mix, sr, device=None)
assert result.device == "cpu"
# ===========================================================================
# 5. main flow – end-to-end with mocked pipeline
# ===========================================================================
class TestMainFlow:
def setup_method(self):
self.node = AudioSeparation()
def test_ensure_stereo_called(self, monkeypatch):
"""ensure_stereo should be invoked on the waveform."""
called = {}
def spy_ensure_stereo(wav):
called["yes"] = True
return _ensure_stereo(wav)
monkeypatch.setattr(separation, "ensure_stereo", spy_ensure_stereo)
audio = _make_audio(channels=2, frames=44100, sample_rate=_MODEL_SR)
self.node.main(audio)
assert called.get("yes")
def test_float_called_after_ensure_stereo(self, monkeypatch):
"""Fix #16/#22: waveform passed to the model should be float32, even if input is float64."""
captured = {}
original_sep = self.node.separate_sources
def spy_separate(model, mix, sr, **kw):
captured["dtype"] = mix._data.dtype
return original_sep(model, mix, sr, **kw)
monkeypatch.setattr(self.node, "separate_sources", spy_separate)
# Use float64 input to verify it gets cast to float32
data = np.random.randn(1, 2, 44100).astype(np.float64)
audio = {"waveform": MockTensor(data), "sample_rate": _MODEL_SR}
self.node.main(audio)
assert captured["dtype"] == np.float32, "waveform should be float32 after .float() call"
def test_resample_called_when_rates_differ(self, monkeypatch):
"""Resample should be instantiated when input SR != model SR."""
created_resamplers = []
original = _MockResample
class _SpyResample(original):
def __init__(self, orig, new):
super().__init__(orig, new)
created_resamplers.append((orig, new))
monkeypatch.setattr(separation, "Resample", _SpyResample)
audio = _make_audio(channels=2, frames=22050, sample_rate=22050)
self.node.main(audio)
assert len(created_resamplers) == 1
assert created_resamplers[0] == (22050, _MODEL_SR)
def test_resample_not_called_when_rates_match(self, monkeypatch):
"""Resample should NOT be called when sample rates match."""
created_resamplers = []
original = _MockResample
class _SpyResample(original):
def __init__(self, orig, new):
super().__init__(orig, new)
created_resamplers.append((orig, new))
monkeypatch.setattr(separation, "Resample", _SpyResample)
audio = _make_audio(channels=2, frames=44100, sample_rate=_MODEL_SR)
self.node.main(audio)
assert len(created_resamplers) == 0
def test_normalization_applied(self, monkeypatch):
"""waveform should be normalized using ref = waveform.mean(0)."""
captured = {}
original_sep = self.node.separate_sources
def spy_separate(model, mix, sr, **kw):
# mix is waveform[None] after normalization — capture it
captured["normalized_mix"] = mix
return original_sep(model, mix, sr, **kw)
monkeypatch.setattr(self.node, "separate_sources", spy_separate)
data = np.full((1, 2, 44100), 5.0, dtype=np.float32)
audio = {"waveform": MockTensor(data), "sample_rate": _MODEL_SR}
self.node.main(audio)
mix = captured["normalized_mix"]
# After (waveform - ref.mean()) / ref.std(), the result should be
# zero-centered. With constant input the std is ~0 so our MockTensor
# std() returns 1.0 → result = (5-5)/1 = 0
assert mix.shape[0] == 1 # batch dim from [None]
def test_main_returns_4_tuple(self):
audio = _make_audio(channels=2, frames=44100, sample_rate=_MODEL_SR)
result = self.node.main(audio)
assert isinstance(result, tuple)
assert len(result) == 4
for r in result:
assert "waveform" in r
assert "sample_rate" in r
def test_main_output_sample_rate(self):
audio = _make_audio(channels=2, frames=44100, sample_rate=_MODEL_SR)
result = self.node.main(audio)
for r in result:
assert r["sample_rate"] == _MODEL_SR
# ===========================================================================
# 6. Corrupted model checkpoint handling (#21)
# ===========================================================================
class TestCorruptedModelCheckpoint:
def setup_method(self):
self.node = AudioSeparation()
def test_bad_zip_file_raises_runtime_error(self, monkeypatch):
"""BadZipFile from corrupted checkpoint should be wrapped with a helpful message."""
bad_bundle = types.SimpleNamespace(
get_model=MagicMock(side_effect=BadZipFile("failed finding central directory")),
sample_rate=_MODEL_SR,
)
monkeypatch.setattr(separation, "HDEMUCS_HIGH_MUSDB_PLUS", bad_bundle)
audio = _make_audio(channels=2, frames=44100, sample_rate=_MODEL_SR)
with pytest.raises(RuntimeError, match="corrupted"):
self.node.main(audio)
def test_runtime_error_from_model_load_wrapped(self, monkeypatch):
"""RuntimeError during model load should also produce a helpful message."""
bad_bundle = types.SimpleNamespace(
get_model=MagicMock(side_effect=RuntimeError("invalid load key")),
sample_rate=_MODEL_SR,
)
monkeypatch.setattr(separation, "HDEMUCS_HIGH_MUSDB_PLUS", bad_bundle)
audio = _make_audio(channels=2, frames=44100, sample_rate=_MODEL_SR)
with pytest.raises(RuntimeError, match="Delete the cached model file"):
self.node.main(audio)
def test_corrupted_error_preserves_original_cause(self, monkeypatch):
"""The original exception should be chained via __cause__."""
original = BadZipFile("bad archive")
bad_bundle = types.SimpleNamespace(
get_model=MagicMock(side_effect=original),
sample_rate=_MODEL_SR,
)
monkeypatch.setattr(separation, "HDEMUCS_HIGH_MUSDB_PLUS", bad_bundle)
audio = _make_audio(channels=2, frames=44100, sample_rate=_MODEL_SR)
with pytest.raises(RuntimeError) as exc_info:
self.node.main(audio)
assert exc_info.value.__cause__ is original
def test_to_device_error_not_wrapped(self, monkeypatch):
"""RuntimeError from .to(device) (e.g. CUDA OOM) should NOT be wrapped as 'corrupted'."""
model = _MockModel()
model.to = MagicMock(side_effect=RuntimeError("CUDA out of memory"))
good_bundle = types.SimpleNamespace(
get_model=MagicMock(return_value=model),
sample_rate=_MODEL_SR,
)
monkeypatch.setattr(separation, "HDEMUCS_HIGH_MUSDB_PLUS", good_bundle)
audio = _make_audio(channels=2, frames=44100, sample_rate=_MODEL_SR)
with pytest.raises(RuntimeError, match="CUDA out of memory"):
self.node.main(audio)
+176
View File
@@ -0,0 +1,176 @@
from __future__ import annotations
import importlib
import sys
import types
import numpy as np
import pytest
# ---------------------------------------------------------------------------
# Mock heavy dependencies before importing the module under test
# ---------------------------------------------------------------------------
torch_mock = types.ModuleType("torch")
class MockTensor:
def __init__(self, data):
self._data = np.array(data) if not isinstance(data, np.ndarray) else data
@property
def shape(self):
return self._data.shape
@property
def ndim(self):
return self._data.ndim
def squeeze(self, dim=0):
return MockTensor(np.squeeze(self._data, axis=dim))
def unsqueeze(self, dim=0):
return MockTensor(np.expand_dims(self._data, axis=dim))
def float(self):
return self
def to(self, device):
return self
def numpy(self):
return self._data
torch_mock.Tensor = MockTensor
torch_mock.device = str
torch_mock.hann_window = lambda *a, **kw: MockTensor(np.ones(2048))
torch_mock.stft = lambda *a, **kw: MockTensor(np.zeros((2, 1025, 10)))
torch_mock.istft = lambda *a, **kw: MockTensor(np.zeros((2, 16000)))
torch_mock.linspace = lambda *a, **kw: MockTensor(np.zeros((1025, 1)))
torch_mock.cfloat = "torch.cfloat"
torch_mock.no_grad = type("_NoGrad", (), {"__enter__": lambda s: s, "__exit__": lambda s, *a: None})
sys.modules["torch"] = torch_mock
torchaudio_mock = types.ModuleType("torchaudio")
torchaudio_functional = types.ModuleType("torchaudio.functional")
torchaudio_functional.phase_vocoder = lambda *a, **kw: MockTensor(np.zeros((2, 1025, 10)))
torchaudio_mock.functional = torchaudio_functional
sys.modules["torchaudio"] = torchaudio_mock
sys.modules["torchaudio.functional"] = torchaudio_functional
librosa_mock = types.ModuleType("librosa")
librosa_onset = types.ModuleType("librosa.onset")
librosa_beat = types.ModuleType("librosa.beat")
librosa_mock.onset = librosa_onset
librosa_mock.beat = librosa_beat
sys.modules["librosa"] = librosa_mock
sys.modules["librosa.onset"] = librosa_onset
sys.modules["librosa.beat"] = librosa_beat
# Clear any previously-cached src modules so they reimport with our mocks
for _key in list(sys.modules):
if _key.startswith("src."):
del sys.modules[_key]
# ---------------------------------------------------------------------------
# Import module under test
# ---------------------------------------------------------------------------
module = importlib.import_module("src.tempo_match")
TempoMatch = module.TempoMatch
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_audio(sample_rate: int = 44100) -> dict:
waveform = MockTensor(np.random.randn(1, 2, 16000))
return {"waveform": waveform, "sample_rate": sample_rate}
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestTempoMatchNode:
# -- schema tests -------------------------------------------------------
def test_input_types_has_required_audio_1_and_audio_2(self):
schema = TempoMatch.INPUT_TYPES()
required = schema["required"]
assert "audio_1" in required
assert required["audio_1"] == ("AUDIO",)
assert "audio_2" in required
assert required["audio_2"] == ("AUDIO",)
def test_return_types(self):
assert TempoMatch.RETURN_TYPES == ("AUDIO", "AUDIO")
# -- main logic tests ---------------------------------------------------
def test_different_tempos_120_and_80(self, monkeypatch):
"""120 and 80 → avg 100, rate_1 = 100/120, rate_2 = 100/80."""
tempo_values = iter([120.0, 80.0])
monkeypatch.setattr(module, "estimate_tempo", lambda w, sr: next(tempo_values))
ts_calls = []
def fake_time_shift(waveform, rate):
ts_calls.append(rate)
return waveform # pass-through
monkeypatch.setattr(module, "time_shift", fake_time_shift)
node = TempoMatch()
result_1, result_2 = node.main(_make_audio(44100), _make_audio(22050))
assert ts_calls[0] == pytest.approx(100.0 / 120.0)
assert ts_calls[1] == pytest.approx(100.0 / 80.0)
assert result_1["sample_rate"] == 44100
assert result_2["sample_rate"] == 22050
def test_same_tempo_rates_are_one(self, monkeypatch):
monkeypatch.setattr(module, "estimate_tempo", lambda w, sr: 100.0)
ts_calls = []
def fake_time_shift(waveform, rate):
ts_calls.append(rate)
return waveform
monkeypatch.setattr(module, "time_shift", fake_time_shift)
TempoMatch().main(_make_audio(), _make_audio())
assert ts_calls[0] == pytest.approx(1.0)
assert ts_calls[1] == pytest.approx(1.0)
def test_output_waveforms_unsqueezed(self, monkeypatch):
monkeypatch.setattr(module, "estimate_tempo", lambda w, sr: 100.0)
monkeypatch.setattr(module, "time_shift", lambda w, r: w)
result_1, result_2 = TempoMatch().main(_make_audio(), _make_audio())
assert result_1["waveform"].ndim == 3
assert result_1["waveform"].shape[0] == 1
assert result_2["waveform"].ndim == 3
assert result_2["waveform"].shape[0] == 1
def test_time_shift_receives_squeezed_waveforms(self, monkeypatch):
monkeypatch.setattr(module, "estimate_tempo", lambda w, sr: 100.0)
received_ndims = []
def spy(waveform, rate):
received_ndims.append(waveform.ndim)
return waveform
monkeypatch.setattr(module, "time_shift", spy)
TempoMatch().main(_make_audio(), _make_audio())
assert received_ndims == [2, 2], "both waveforms should be 2-d (batch squeezed)"
+180
View File
@@ -0,0 +1,180 @@
from __future__ import annotations
import importlib
import sys
import types
import numpy as np
import pytest
# ---------------------------------------------------------------------------
# Mock heavy dependencies before importing the module under test
# ---------------------------------------------------------------------------
torch_mock = types.ModuleType("torch")
class MockTensor:
def __init__(self, data):
self._data = np.array(data) if not isinstance(data, np.ndarray) else data
@property
def shape(self):
return self._data.shape
@property
def ndim(self):
return self._data.ndim
def squeeze(self, dim=0):
return MockTensor(np.squeeze(self._data, axis=dim))
def unsqueeze(self, dim=0):
return MockTensor(np.expand_dims(self._data, axis=dim))
def float(self):
return self
def to(self, device):
return self
def numpy(self):
return self._data
torch_mock.Tensor = MockTensor
torch_mock.device = str
torch_mock.hann_window = lambda *a, **kw: MockTensor(np.ones(2048))
torch_mock.stft = lambda *a, **kw: MockTensor(np.zeros((2, 1025, 10)))
torch_mock.istft = lambda *a, **kw: MockTensor(np.zeros((2, 16000)))
torch_mock.linspace = lambda *a, **kw: MockTensor(np.zeros((1025, 1)))
torch_mock.cfloat = "torch.cfloat"
torch_mock.no_grad = type("_NoGrad", (), {"__enter__": lambda s: s, "__exit__": lambda s, *a: None})
sys.modules["torch"] = torch_mock
torchaudio_mock = types.ModuleType("torchaudio")
torchaudio_functional = types.ModuleType("torchaudio.functional")
torchaudio_functional.phase_vocoder = lambda *a, **kw: MockTensor(np.zeros((2, 1025, 10)))
torchaudio_mock.functional = torchaudio_functional
sys.modules["torchaudio"] = torchaudio_mock
sys.modules["torchaudio.functional"] = torchaudio_functional
librosa_mock = types.ModuleType("librosa")
librosa_onset = types.ModuleType("librosa.onset")
librosa_beat = types.ModuleType("librosa.beat")
librosa_mock.onset = librosa_onset
librosa_mock.beat = librosa_beat
sys.modules["librosa"] = librosa_mock
sys.modules["librosa.onset"] = librosa_onset
sys.modules["librosa.beat"] = librosa_beat
# Clear any previously-cached src modules so they reimport with our mocks
for _key in list(sys.modules):
if _key.startswith("src."):
del sys.modules[_key]
# ---------------------------------------------------------------------------
# Import module under test
# ---------------------------------------------------------------------------
module = importlib.import_module("src.time_shift")
TimeShift = module.TimeShift
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_audio(sample_rate: int = 44100) -> dict:
waveform = MockTensor(np.random.randn(1, 2, 16000))
return {"waveform": waveform, "sample_rate": sample_rate}
def _stub_time_shift(returned_data=None):
"""Return a (spy, mock_result) pair for time_shift."""
mock_result = MockTensor(returned_data if returned_data is not None else np.random.randn(2, 8000))
calls = []
def spy(waveform, rate):
calls.append({"waveform_ndim": waveform.ndim, "rate": rate})
return mock_result
return spy, calls, mock_result
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
class TestTimeShiftNode:
# -- schema tests -------------------------------------------------------
def test_input_types_has_required_audio_and_rate(self):
schema = TimeShift.INPUT_TYPES()
required = schema["required"]
assert "audio" in required
assert required["audio"] == ("AUDIO",)
assert "rate" in required
def test_rate_constraints(self):
schema = TimeShift.INPUT_TYPES()
rate_spec = schema["required"]["rate"]
assert rate_spec[0] == "FLOAT"
opts = rate_spec[1]
assert opts["default"] == 1.0
assert opts["min"] == 0.1
assert opts["max"] == 10.0
def test_return_types(self):
assert TimeShift.RETURN_TYPES == ("AUDIO",)
# -- main logic tests ---------------------------------------------------
def test_basic_call(self, monkeypatch):
spy, calls, mock_result = _stub_time_shift()
monkeypatch.setattr(module, "time_shift", spy)
node = TimeShift()
(result,) = node.main(_make_audio(22050), rate=1.5)
assert calls[0]["rate"] == 1.5
assert result["sample_rate"] == 22050
def test_rate_clamped_below_min(self, monkeypatch):
spy, calls, _ = _stub_time_shift()
monkeypatch.setattr(module, "time_shift", spy)
TimeShift().main(_make_audio(), rate=0.05)
assert calls[0]["rate"] == pytest.approx(0.1)
def test_rate_clamped_above_max(self, monkeypatch):
spy, calls, _ = _stub_time_shift()
monkeypatch.setattr(module, "time_shift", spy)
TimeShift().main(_make_audio(), rate=15.0)
assert calls[0]["rate"] == pytest.approx(10.0)
def test_rate_within_range_unchanged(self, monkeypatch):
spy, calls, _ = _stub_time_shift()
monkeypatch.setattr(module, "time_shift", spy)
TimeShift().main(_make_audio(), rate=1.5)
assert calls[0]["rate"] == pytest.approx(1.5)
def test_waveform_squeezed_before_call(self, monkeypatch):
spy, calls, _ = _stub_time_shift()
monkeypatch.setattr(module, "time_shift", spy)
TimeShift().main(_make_audio(), rate=1.0)
assert calls[0]["waveform_ndim"] == 2, "batch dim should be squeezed"
def test_output_waveform_unsqueezed(self, monkeypatch):
data_2d = np.random.randn(2, 8000)
spy, _, _ = _stub_time_shift(data_2d)
monkeypatch.setattr(module, "time_shift", spy)
(result,) = TimeShift().main(_make_audio(), rate=1.0)
assert result["waveform"].ndim == 3, "output should have batch dim restored"
assert result["waveform"].shape[0] == 1
+297
View File
@@ -0,0 +1,297 @@
import math
import sys
import types
from unittest.mock import MagicMock
import numpy as np
import pytest
# ---------------------------------------------------------------------------
# Mock torch, torchaudio, and librosa BEFORE importing src.utils
# ---------------------------------------------------------------------------
class MockTensor:
"""Wraps a numpy array to emulate torch.Tensor for ensure_stereo tests."""
def __init__(self, data):
self._data = np.array(data, dtype=np.float32) if not isinstance(data, np.ndarray) else data
self._dtype = "torch.cfloat"
@property
def ndim(self):
return self._data.ndim
@property
def shape(self):
return self._data.shape
@property
def device(self):
return "cpu"
@property
def dtype(self):
return self._dtype
@dtype.setter
def dtype(self, value):
self._dtype = value
def repeat(self, *sizes):
return MockTensor(np.tile(self._data, sizes))
def narrow(self, dim, start, length):
slices = [slice(None)] * self._data.ndim
slices[dim] = slice(start, start + length)
return MockTensor(self._data[tuple(slices)])
def mean(self, dim=None, keepdim=False):
return MockTensor(np.mean(self._data, axis=dim, keepdims=keepdim))
def __getitem__(self, key):
return MockTensor(self._data[key])
def dim(self):
return self._data.ndim
def squeeze(self, dim):
return MockTensor(np.squeeze(self._data, axis=dim))
def numpy(self):
return self._data
def __eq__(self, other):
if isinstance(other, MockTensor):
return np.array_equal(self._data, other._data)
return NotImplemented
# -- torch mock -------------------------------------------------------------
torch_mock = types.ModuleType("torch")
torch_mock.Tensor = MockTensor
torch_mock.cfloat = "torch.cfloat"
torch_mock.hann_window = MagicMock(return_value=MockTensor(np.ones(2048)))
torch_mock.stft = MagicMock()
torch_mock.istft = MagicMock()
torch_mock.linspace = MagicMock(return_value=MockTensor(np.zeros((1025, 1))))
class _NoGrad:
def __enter__(self):
return self
def __exit__(self, *args):
pass
torch_mock.no_grad = _NoGrad
sys.modules["torch"] = torch_mock
# -- torchaudio mock --------------------------------------------------------
torchaudio_mock = types.ModuleType("torchaudio")
torchaudio_functional_mock = types.ModuleType("torchaudio.functional")
torchaudio_functional_mock.phase_vocoder = MagicMock()
torchaudio_mock.functional = torchaudio_functional_mock
sys.modules["torchaudio"] = torchaudio_mock
sys.modules["torchaudio.functional"] = torchaudio_functional_mock
# -- librosa mock -----------------------------------------------------------
librosa_mock = types.ModuleType("librosa")
librosa_onset_mock = types.ModuleType("librosa.onset")
librosa_beat_mock = types.ModuleType("librosa.beat")
librosa_onset_mock.onset_strength = MagicMock()
librosa_beat_mock.beat_track = MagicMock()
librosa_mock.onset = librosa_onset_mock
librosa_mock.beat = librosa_beat_mock
sys.modules["librosa"] = librosa_mock
sys.modules["librosa.onset"] = librosa_onset_mock
sys.modules["librosa.beat"] = librosa_beat_mock
# Clear any previously-cached src modules so they reimport with our mocks
for _key in list(sys.modules):
if _key.startswith("src."):
del sys.modules[_key]
# ---------------------------------------------------------------------------
# Now import the module under test
# ---------------------------------------------------------------------------
from src.utils import ensure_stereo, estimate_tempo, time_shift
# ===========================================================================
# ensure_stereo
# ===========================================================================
class TestEnsureStereo:
def test_already_stereo_2d(self):
audio = MockTensor(np.random.rand(2, 100))
result = ensure_stereo(audio)
assert result.shape == (2, 100)
assert np.array_equal(result._data, audio._data)
def test_already_stereo_3d(self):
audio = MockTensor(np.random.rand(1, 2, 100))
result = ensure_stereo(audio)
assert result.shape == (1, 2, 100)
assert np.array_equal(result._data, audio._data)
def test_mono_2d_duplicated(self):
audio = MockTensor(np.ones((1, 100)))
result = ensure_stereo(audio)
assert result.shape == (2, 100)
assert np.array_equal(result._data[0], result._data[1])
def test_mono_3d_duplicated(self):
audio = MockTensor(np.ones((1, 1, 100)))
result = ensure_stereo(audio)
assert result.shape == (1, 2, 100)
assert np.array_equal(result._data[0, 0], result._data[0, 1])
def test_multichannel_2d_downmixed(self):
data = np.array(
[
[1.0, 2.0, 3.0],
[5.0, 6.0, 7.0],
[9.0, 10.0, 11.0],
[13.0, 14.0, 15.0],
]
)
audio = MockTensor(data)
result = ensure_stereo(audio)
assert result.shape == (2, 3)
expected_mean = np.mean(data[:2], axis=0)
np.testing.assert_allclose(result._data[0], expected_mean)
np.testing.assert_allclose(result._data[1], expected_mean)
def test_multichannel_3d_downmixed(self):
data = np.random.rand(1, 4, 50).astype(np.float32)
audio = MockTensor(data)
result = ensure_stereo(audio)
assert result.shape == (1, 2, 50)
expected_mean = np.mean(data[:, :2, :], axis=1, keepdims=True)
np.testing.assert_allclose(result._data[0, 0], expected_mean[0, 0])
np.testing.assert_allclose(result._data[0, 1], expected_mean[0, 0])
def test_invalid_1d_raises(self):
audio = MockTensor(np.ones(100))
with pytest.raises(ValueError, match="2 or 3 dimensions"):
ensure_stereo(audio)
def test_invalid_4d_raises(self):
audio = MockTensor(np.ones((1, 1, 2, 100)))
with pytest.raises(ValueError, match="2 or 3 dimensions"):
ensure_stereo(audio)
# ===========================================================================
# estimate_tempo
# ===========================================================================
class TestEstimateTempo:
def test_returns_tempo(self):
librosa_onset_mock.onset_strength.return_value = np.ones(10)
librosa_beat_mock.beat_track.return_value = (np.array([[120.0]]), None)
waveform = MockTensor(np.random.rand(2, 22050))
tempo = estimate_tempo(waveform, 22050)
assert tempo == 120.0
librosa_onset_mock.onset_strength.assert_called_once()
librosa_beat_mock.beat_track.assert_called_once()
def test_3d_input_gets_squeezed(self):
librosa_onset_mock.onset_strength.reset_mock()
librosa_beat_mock.beat_track.reset_mock()
librosa_onset_mock.onset_strength.return_value = np.ones(10)
librosa_beat_mock.beat_track.return_value = (np.array([[90.0]]), None)
waveform = MockTensor(np.random.rand(1, 2, 22050))
tempo = estimate_tempo(waveform, 22050)
assert tempo == 90.0
call_kwargs = librosa_onset_mock.onset_strength.call_args
# After squeeze(0) the array should be 2D
assert call_kwargs.kwargs["y"].ndim == 2
def test_min_clamp(self):
librosa_onset_mock.onset_strength.return_value = np.ones(10)
librosa_beat_mock.beat_track.return_value = (np.array([[0.5]]), None)
waveform = MockTensor(np.random.rand(2, 22050))
tempo = estimate_tempo(waveform, 22050)
assert tempo == 1.0
def test_wrong_ndim_raises(self):
waveform = MockTensor(np.random.rand(22050))
with pytest.raises(TypeError, match="Expected waveform"):
estimate_tempo(waveform, 22050)
# ===========================================================================
# time_shift
# ===========================================================================
class TestTimeShift:
def _setup_stft_mocks(self, channels=2, freq=1025, time_frames=10, rate=1.0):
"""Configure stft/phase_vocoder/istft mocks for a call."""
stft_result = MockTensor(np.zeros((channels, freq, time_frames)))
# Ensure dtype matches torch.cfloat so the TypeError check passes
stft_result.dtype = torch_mock.cfloat
torch_mock.stft.reset_mock()
torch_mock.istft.reset_mock()
torchaudio_functional_mock.phase_vocoder.reset_mock()
torch_mock.linspace.reset_mock()
torch_mock.stft.return_value = stft_result
stretched_time = math.ceil(time_frames / rate)
stretched = MockTensor(np.zeros((channels, freq, stretched_time)))
torchaudio_functional_mock.phase_vocoder.return_value = stretched
output = MockTensor(np.zeros((channels, 44100)))
torch_mock.istft.return_value = output
return output
def test_calls_phase_vocoder_with_rate(self):
rate = 1.5
self._setup_stft_mocks(rate=rate)
waveform = MockTensor(np.zeros((2, 44100)))
time_shift(waveform, rate)
call_args = torchaudio_functional_mock.phase_vocoder.call_args
assert call_args[0][1] == rate
def test_default_hop_size(self):
fft_size = 2048
expected_hop = fft_size // 4
self._setup_stft_mocks(rate=1.0)
waveform = MockTensor(np.zeros((2, 44100)))
time_shift(waveform, 1.0, fft_size=fft_size)
stft_kwargs = torch_mock.stft.call_args
assert stft_kwargs.kwargs["hop_length"] == expected_hop
def test_custom_hop_size(self):
self._setup_stft_mocks(rate=1.0)
waveform = MockTensor(np.zeros((2, 44100)))
time_shift(waveform, 1.0, hop_size=256)
stft_kwargs = torch_mock.stft.call_args
assert stft_kwargs.kwargs["hop_length"] == 256
def test_non_complex_dtype_raises(self):
stft_result = MockTensor(np.zeros((2, 1025, 10)))
stft_result.dtype = "torch.float32"
torch_mock.stft.return_value = stft_result
waveform = MockTensor(np.zeros((2, 44100)))
with pytest.raises(TypeError, match="Expected complex-valued STFT"):
time_shift(waveform, 1.0)
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.

Before

Width:  |  Height:  |  Size: 22 MiB

Binary file not shown.
Binary file not shown.
Binary file not shown.