Video to Music

import base64
import json
import os
import subprocess
import tempfile
from pathlib import Path

import requests

API_KEY = os.environ["SONILO_API_KEY"]
API_URL = "https://api.sonilo.com/v1/video-to-music"


def ensure_demo_video() -> Path:
    """Return a video path so this script succeeds with no CLI arguments.

    Set SONILO_VIDEO_FILE to use your own mp4. Without it, the script creates
    a tiny synthetic mp4 with ffmpeg for smoke testing.
    """
    provided = os.environ.get("SONILO_VIDEO_FILE")
    if provided:
        return Path(provided)

    demo = Path(tempfile.gettempdir()) / "sonilo-demo-video.mp4"
    if demo.exists():
        return demo

    subprocess.run(
        [
            "ffmpeg",
            "-y",
            "-f",
            "lavfi",
            "-i",
            "testsrc=size=320x180:rate=24",
            "-f",
            "lavfi",
            "-i",
            "sine=frequency=440:duration=5",
            "-t",
            "5",
            "-pix_fmt",
            "yuv420p",
            "-c:v",
            "libx264",
            "-c:a",
            "aac",
            str(demo),
        ],
        check=True,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    return demo


video_path = ensure_demo_video()
chunks_by_stream: dict[int, list[bytes]] = {}
completed = False

with video_path.open("rb") as video_file:
    with requests.post(
        API_URL,
        headers={"Authorization": f"Bearer {API_KEY}"},
        files={"video": (video_path.name, video_file, "video/mp4")},
        stream=True,
        timeout=300,
    ) as resp:
        resp.raise_for_status()
        for line in resp.iter_lines(decode_unicode=True):
            if not line:
                continue
            event = json.loads(line)
            if event["type"] == "audio_chunk":
                stream_index = event.get("stream_index", 0)
                chunks_by_stream.setdefault(stream_index, []).append(
                    base64.b64decode(event["data"])
                )
            elif event["type"] == "title":
                print("title:", event["title"])
            elif event["type"] == "error":
                raise RuntimeError(event)
            elif event["type"] == "complete":
                completed = True

if not completed:
    raise RuntimeError("stream ended before complete")

for stream_index, chunks in sorted(chunks_by_stream.items()):
    out = Path(f"sonilo-video-to-music-{stream_index}.m4a")
    out.write_bytes(b"".join(chunks))
    print("saved:", out, out.stat().st_size, "bytes")

import fs from "node:fs";

const form = new FormData();
const blob = await fs.openAsBlob("/path/to/video.mp4");
form.append("video", blob, "video.mp4");

const resp = await fetch("https://api.sonilo.com/v1/video-to-music", {
  method: "POST",
  headers: { Authorization: "Bearer sk_your_api_key" },
  body: form,
});

if (!resp.ok) {
  throw new Error(`HTTP ${resp.status}: ${await resp.text()}`);
}

const reader = resp.body.getReader();
const decoder = new TextDecoder();
let buf = "";
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  buf += decoder.decode(value, { stream: true });
  let nl;
  while ((nl = buf.indexOf("\n")) !== -1) {
    const line = buf.slice(0, nl);
    buf = buf.slice(nl + 1);
    if (!line) continue;
    console.log(JSON.parse(line));
  }
}

import json
import requests

with requests.post(
    "https://api.sonilo.com/v1/video-to-music",
    headers={"Authorization": "Bearer sk_your_api_key"},
    files={
        "video_url": (None, "https://example.com/video.mp4"),
        "segments": (None, json.dumps([
            {"start": 0, "label": "intro", "prompt": "ambient pads"},
            {"start": 12, "label": "chorus", "prompt": "driving drums and bass"},
        ])),
    },
    stream=True,
    timeout=300,
) as resp:
    resp.raise_for_status()
    for line in resp.iter_lines(decode_unicode=True):
        if not line:
            continue
        event = json.loads(line)
        print(event["type"], event)

const form = new FormData();
form.append("video_url", "https://example.com/video.mp4");
form.append(
  "segments",
  JSON.stringify([
    { start: 0, label: "intro", prompt: "ambient pads" },
    { start: 12, label: "chorus", prompt: "driving drums and bass" },
  ]),
);

const resp = await fetch("https://api.sonilo.com/v1/video-to-music", {
  method: "POST",
  headers: { Authorization: "Bearer sk_your_api_key" },
  body: form,
});

if (!resp.ok) {
  throw new Error(`HTTP ${resp.status}: ${await resp.text()}`);
}

const reader = resp.body.getReader();
const decoder = new TextDecoder();
let buf = "";
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  buf += decoder.decode(value, { stream: true });
  let nl;
  while ((nl = buf.indexOf("\n")) !== -1) {
    const line = buf.slice(0, nl);
    buf = buf.slice(nl + 1);
    if (!line) continue;
    console.log(JSON.parse(line));
  }
}