LocalAI/backend/python/mlx-video/backend.py

#!/usr/bin/env python3
# SPDX-License-Identifier: MIT

import argparse
import os
import platform
import signal
import subprocess
import sys
import traceback
from concurrent import futures

import grpc

import backend_pb2
import backend_pb2_grpc
from mlx_video_utils import build_generation_command, classify_model, parse_options

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "common"))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "common"))
from grpc_auth import get_auth_interceptors  # noqa: E402


MAX_WORKERS = int(os.environ.get("PYTHON_GRPC_MAX_WORKERS", "1"))


class BackendServicer(backend_pb2_grpc.BackendServicer):
    def __init__(self):
        self.family = None
        self.model = None
        self.options = {}

    def Health(self, request, context):
        return backend_pb2.Reply(message=b"OK")

    def LoadModel(self, request, context):
        if platform.system() != "Darwin" or platform.machine() != "arm64":
            return self._fail(
                context,
                grpc.StatusCode.FAILED_PRECONDITION,
                "mlx-video requires macOS on Apple Silicon",
            )

        model = (
            request.ModelFile
            if request.ModelFile and os.path.isdir(request.ModelFile)
            else request.Model
        )
        family = classify_model(model)
        if family is None:
            return self._fail(
                context,
                grpc.StatusCode.INVALID_ARGUMENT,
                "mlx-video supports LTX-2 and converted Wan2.1/Wan2.2 MLX checkpoints",
            )
        if family == "wan" and not os.path.isdir(model):
            return self._fail(
                context,
                grpc.StatusCode.INVALID_ARGUMENT,
                "Wan models must be converted to MLX and configured as a local model directory",
            )

        self.family = family
        self.model = model
        self.options = parse_options(request.Options)
        return backend_pb2.Result(message="mlx-video model configured", success=True)

    def Free(self, request, context):
        self.family = None
        self.model = None
        self.options = {}
        return backend_pb2.Result(message="Model released", success=True)

    def GenerateVideo(self, request, context):
        if not self.family or not self.model:
            return self._fail(
                context, grpc.StatusCode.FAILED_PRECONDITION, "model is not loaded"
            )
        if not request.prompt.strip():
            return self._fail(context, grpc.StatusCode.INVALID_ARGUMENT, "prompt is required")
        if not request.dst:
            return self._fail(
                context,
                grpc.StatusCode.INVALID_ARGUMENT,
                "output destination is required",
            )

        process = None
        try:
            command = build_generation_command(
                family=self.family,
                model=self.model,
                prompt=request.prompt,
                destination=request.dst,
                negative_prompt=request.negative_prompt,
                start_image=request.start_image,
                end_image=request.end_image,
                audio=request.audio,
                width=request.width,
                height=request.height,
                num_frames=request.num_frames,
                fps=request.fps,
                seed=request.seed,
                steps=request.step,
                cfg_scale=request.cfg_scale,
                params={**self.options, **dict(request.params)},
            )
            os.makedirs(
                os.path.dirname(request.dst) or ".", mode=0o750, exist_ok=True
            )
            process = subprocess.Popen(command)
            if hasattr(context, "add_callback"):
                context.add_callback(
                    lambda: process.poll() is None and process.terminate()
                )
            return_code = process.wait()
            if return_code != 0:
                raise RuntimeError(
                    f"upstream generator exited with status {return_code}"
                )
            if not os.path.isfile(request.dst) or os.path.getsize(request.dst) == 0:
                raise RuntimeError("upstream generator did not create the requested video")
            return backend_pb2.Result(message="Video generated successfully", success=True)
        except ValueError as err:
            return self._fail(context, grpc.StatusCode.INVALID_ARGUMENT, str(err))
        except Exception as err:
            print(f"mlx-video generation failed: {err}", file=sys.stderr)
            traceback.print_exc()
            return self._fail(
                context,
                grpc.StatusCode.INTERNAL,
                f"mlx-video generation failed: {err}",
            )

    @staticmethod
    def _fail(context, code, message):
        context.set_code(code)
        context.set_details(message)
        return backend_pb2.Result(message=message, success=False)


def serve(address):
    server = grpc.server(
        futures.ThreadPoolExecutor(max_workers=MAX_WORKERS),
        options=[
            ("grpc.max_message_length", 64 * 1024 * 1024),
            ("grpc.max_send_message_length", 64 * 1024 * 1024),
            ("grpc.max_receive_message_length", 64 * 1024 * 1024),
        ],
        interceptors=get_auth_interceptors(),
    )
    backend_pb2_grpc.add_BackendServicer_to_server(BackendServicer(), server)
    server.add_insecure_port(address)
    server.start()
    print(f"MLX Video backend listening on {address}", file=sys.stderr)

    def stop_server(signum, frame):
        del signum, frame
        server.stop(0)

    signal.signal(signal.SIGINT, stop_server)
    signal.signal(signal.SIGTERM, stop_server)
    server.wait_for_termination()


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Run the MLX Video gRPC backend")
    parser.add_argument("--addr", default="localhost:50051")
    serve(parser.parse_args().addr)