feat: add isolate_vocals (async) support to video-to-music

README.md

@@ -32,6 +32,39 @@ track = client.video_to_music.generate(video="my_video.mp4", prompt="upbeat")
 track = client.video_to_music.generate(video_url="https://example.com/clip.mp4")
 ```
 
+### Vocal isolation (async)
+
+Pass `isolate_vocals=True` to also get an isolated vocal stem and a muxed
+video, alongside the scored audio. This requires async processing — submit
+returns a `task_id` immediately, and `generate_async()` wraps submit + poll:
+
+```python
+result = client.video_to_music.generate_async(
+    video="my_video.mp4",
+    prompt="upbeat",
+    isolate_vocals=True,  # implies mode="async"; omit mode to let it auto-select
+)
+result.save("mix.m4a")           # result.audio[0] — the full mix
+result.save("vocals.m4a", which="vocals")
+result.save("video.mp4", which="mux")  # video muxed with the generated audio
+print(result.title.title if result.title else None)
+```
+
+Or control submission and polling yourself:
+
+```python
+from sonilo.resources.tasks import parse_music_result
+
+task = client.video_to_music.submit(video_url="https://example.com/clip.mp4", isolate_vocals=True)
+result = client.tasks.wait(
+    task.task_id,
+    parser=parse_music_result,  # required: tasks.wait()/get() default to the SFX parser
+)
+```
+
+`isolate_vocals=True` with an explicit non-async `mode` raises `SoniloError`
+locally before any request is sent.
+
 ## Streaming
 
 ```python

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "sonilo"
-version = "0.2.1"
+version = "0.3.0"
 description = "Official Python client for the Sonilo API"
 readme = "README.md"
 license = "MIT"

src/sonilo/__init__.py

@@ -13,6 +13,9 @@
     TaskTimeoutError,
 )
 from sonilo.types import (
+    MusicAudioMedia,
+    MusicResult,
+    MusicTitle,
     Segment,
     SfxMedia,
     SfxResult,
@@ -28,6 +31,9 @@
     "AuthenticationError",
     "BadRequestError",
     "GenerationError",
+    "MusicAudioMedia",
+    "MusicResult",
+    "MusicTitle",
     "PaymentRequiredError",
     "RateLimitError",
     "Segment",

src/sonilo/_requests.py

@@ -66,6 +66,38 @@ def build_v2m_parts(
     return data, files, opened
 
 
+def _resolve_music_mode(mode: Optional[str], isolate_vocals: Optional[bool]) -> str:
+    """isolate_vocals only works with async processing: auto-select mode
+    "async" when the caller didn't specify one, but fail fast (mirroring the
+    video/video_url XOR check above) if they explicitly asked for anything
+    else. Without isolate_vocals, submit() still needs an async response
+    (a task_id ack, not a stream), so "async" is also the default there.
+    """
+    if isolate_vocals:
+        if mode is not None and mode != "async":
+            raise SoniloError("isolate_vocals=True requires mode='async'")
+        return "async"
+    return mode or "async"
+
+
+def build_v2m_async_parts(
+    video: Any,
+    video_url: Optional[str],
+    prompt: Optional[str],
+    segments: Optional[List[Segment]],
+    mode: Optional[str],
+    isolate_vocals: Optional[bool],
+) -> Tuple[Dict[str, str], Optional[Dict[str, tuple]], bool]:
+    """Like build_v2m_parts, plus the async-only `mode`/`isolate_vocals`
+    fields used by the video-to-music submit()/generate_async() path."""
+    resolved_mode = _resolve_music_mode(mode, isolate_vocals)
+    data, files, opened = build_v2m_parts(video, video_url, prompt, segments)
+    data["mode"] = resolved_mode
+    if isolate_vocals is not None:
+        data["isolate_vocals"] = "true" if isolate_vocals else "false"
+    return data, files, opened
+
+
 def build_sfx_t2s_data(
     prompt: str, duration: int, audio_format: Optional[str]
 ) -> Dict[str, str]:

src/sonilo/_version.py

@@ -1 +1 @@
-__version__ = "0.2.0"
+__version__ = "0.3.0"

src/sonilo/resources/tasks.py

@@ -2,11 +2,11 @@
 
 import asyncio
 import time
-from typing import TYPE_CHECKING, Any, Dict, Optional
+from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Protocol, TypeVar
 from urllib.parse import quote
 
 from sonilo.errors import SoniloError, TaskFailedError, TaskTimeoutError
-from sonilo.types import SfxMedia, SfxResult, SfxTask
+from sonilo.types import MusicAudioMedia, MusicResult, MusicTitle, SfxMedia, SfxResult, SfxTask
 
 if TYPE_CHECKING:
     from sonilo._async_client import AsyncSonilo
@@ -21,6 +21,19 @@
 _monotonic = time.monotonic
 
 
+class _PollableResult(Protocol):
+    """Structural shape Tasks.get()/wait() need from any parsed result,
+    regardless of which endpoint produced it (SFX vs. music)."""
+
+    task_id: str
+    status: str
+    error: Optional[Dict[str, Any]]
+    refunded: Optional[bool]
+
+
+ResultT = TypeVar("ResultT", bound=_PollableResult)
+
+
 def _media_from(data: Any) -> Optional[SfxMedia]:
     if not isinstance(data, dict) or "url" not in data:
         return None
@@ -48,6 +61,61 @@ def parse_sfx_result(body: Dict[str, Any]) -> SfxResult:
         raise SoniloError(f"Malformed task response: missing {e.args[0]!r}") from e
 
 
+def _music_audio_from(data: Any) -> Optional[MusicAudioMedia]:
+    if not isinstance(data, dict) or "url" not in data:
+        return None
+    return MusicAudioMedia(
+        stream_index=data.get("stream_index", 0),
+        url=data["url"],
+        content_type=data.get("content_type"),
+        file_size=data.get("file_size"),
+        sample_rate=data.get("sample_rate"),
+        channels=data.get("channels"),
+    )
+
+
+def _music_audio_list_from(data: Any) -> Optional[List[MusicAudioMedia]]:
+    if not isinstance(data, list):
+        return None
+    items = [item for item in (_music_audio_from(entry) for entry in data) if item is not None]
+    return items
+
+
+def _music_title_from(data: Any) -> Optional[MusicTitle]:
+    if not isinstance(data, dict):
+        return None
+    return MusicTitle(
+        title=data.get("title"),
+        summary=data.get("summary"),
+        display_tags=data.get("display_tags"),
+    )
+
+
+def parse_music_result(body: Dict[str, Any]) -> MusicResult:
+    """Map a GET /v1/tasks/{id} body for a video-to-music task to
+    MusicResult; unknown fields are ignored.
+
+    `audio` is always a list; `vocals`/`mux` are only populated when the
+    task was submitted with isolate_vocals=True.
+    """
+    try:
+        return MusicResult(
+            task_id=body["task_id"],
+            status=body["status"],
+            type=body.get("type"),
+            audio=_music_audio_list_from(body.get("audio")),
+            vocals=_media_from(body.get("vocals")),
+            mux=_music_audio_list_from(body.get("mux")),
+            title=_music_title_from(body.get("title")),
+            duration_seconds=body.get("duration_seconds"),
+            cost=body.get("cost"),
+            error=body.get("error"),
+            refunded=body.get("refunded"),
+        )
+    except KeyError as e:
+        raise SoniloError(f"Malformed task response: missing {e.args[0]!r}") from e
+
+
 def parse_sfx_task(body: Dict[str, Any]) -> SfxTask:
     """Map a submission ack to SfxTask."""
     try:
@@ -56,7 +124,7 @@ def parse_sfx_task(body: Dict[str, Any]) -> SfxTask:
         raise SoniloError(f"Malformed task response: missing {e.args[0]!r}") from e
 
 
-def _raise_if_failed(result: SfxResult) -> None:
+def _raise_if_failed(result: _PollableResult) -> None:
     if result.status == "failed":
         error = result.error if isinstance(result.error, dict) else {}
         message = error.get("message") or "Generation failed"
@@ -87,24 +155,33 @@ class Tasks:
     def __init__(self, client: "Sonilo") -> None:
         self._client = client
 
-    def get(self, task_id: str) -> SfxResult:
-        """Fetch current task state. Never raises on a failed status."""
-        return parse_sfx_result(
-            self._client._get_json(f"/v1/tasks/{quote(task_id, safe='')}")
-        )
+    def get(
+        self,
+        task_id: str,
+        *,
+        parser: Callable[[Dict[str, Any]], ResultT] = parse_sfx_result,  # type: ignore[assignment]
+    ) -> ResultT:
+        """Fetch current task state. Never raises on a failed status.
+
+        `parser` maps the raw response body to a result type; it defaults to
+        the SFX parser for back-compat. Pass `parse_music_result` for
+        video-to-music async tasks.
+        """
+        return parser(self._client._get_json(f"/v1/tasks/{quote(task_id, safe='')}"))
 
     def wait(
         self,
         task_id: str,
         *,
         poll_interval: float = DEFAULT_POLL_INTERVAL,
         timeout: float = DEFAULT_WAIT_TIMEOUT,
-    ) -> SfxResult:
+        parser: Callable[[Dict[str, Any]], ResultT] = parse_sfx_result,  # type: ignore[assignment]
+    ) -> ResultT:
         """Poll until the task is terminal; raise on failure or deadline."""
         _validate_wait_args(poll_interval, timeout)
         deadline = _monotonic() + timeout
         while True:
-            result = self.get(task_id)
+            result = self.get(task_id, parser=parser)
             if result.status == "succeeded":
                 return result
             _raise_if_failed(result)
@@ -118,24 +195,33 @@ class AsyncTasks:
     def __init__(self, client: "AsyncSonilo") -> None:
         self._client = client
 
-    async def get(self, task_id: str) -> SfxResult:
-        """Fetch current task state. Never raises on a failed status."""
-        return parse_sfx_result(
-            await self._client._get_json(f"/v1/tasks/{quote(task_id, safe='')}")
-        )
+    async def get(
+        self,
+        task_id: str,
+        *,
+        parser: Callable[[Dict[str, Any]], ResultT] = parse_sfx_result,  # type: ignore[assignment]
+    ) -> ResultT:
+        """Fetch current task state. Never raises on a failed status.
+
+        `parser` maps the raw response body to a result type; it defaults to
+        the SFX parser for back-compat. Pass `parse_music_result` for
+        video-to-music async tasks.
+        """
+        return parser(await self._client._get_json(f"/v1/tasks/{quote(task_id, safe='')}"))
 
     async def wait(
         self,
         task_id: str,
         *,
         poll_interval: float = DEFAULT_POLL_INTERVAL,
         timeout: float = DEFAULT_WAIT_TIMEOUT,
-    ) -> SfxResult:
+        parser: Callable[[Dict[str, Any]], ResultT] = parse_sfx_result,  # type: ignore[assignment]
+    ) -> ResultT:
         """Poll until the task is terminal; raise on failure or deadline."""
         _validate_wait_args(poll_interval, timeout)
         deadline = _monotonic() + timeout
         while True:
-            result = await self.get(task_id)
+            result = await self.get(task_id, parser=parser)
             if result.status == "succeeded":
                 return result
             _raise_if_failed(result)