backend_worker.py

#!/usr/bin/env python3
"""One-shot BeatNet backend runner for the Python 3.9 environment.

The frontend invokes this script as a subprocess and passes uploaded audio by
file path, avoiding cross-interpreter object sharing without keeping a backend
worker process alive.
"""

from __future__ import annotations

import contextlib
import json
import sys
from typing import Any


def run_beatnet(audio_path: str) -> list[dict[str, float]]:
    with contextlib.redirect_stdout(sys.stderr):
        from BeatNet.BeatNet import BeatNet

        estimator = BeatNet(1, mode="offline", inference_model="DBN", plot=[], thread=False)
        beats = estimator.process(audio_path)

    return [
        {"time": float(timestamp), "beat": float(beat)}
        for timestamp, beat in beats
    ]


def main(argv: list[str]) -> int:
    if len(argv) != 2:
        print(
            json.dumps({"ok": False, "error": "Usage: backend_worker.py <audio_path>"}),
            flush=True,
        )
        return 2

    try:
        payload: dict[str, Any] = {"ok": True, "beats": run_beatnet(argv[1])}
    except Exception as exc:
        payload = {"ok": False, "error": repr(exc)}

    print(json.dumps(payload), flush=True)
    return 0 if payload["ok"] else 1


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))