File size: 3,321 Bytes
038574d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | from __future__ import annotations
import importlib.util
import os
import sys
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Iterator
from fastapi import FastAPI
@dataclass(frozen=True)
class ServiceDefinition:
key: str
title: str
path: str
directory: str
module_file: str
app_object: str
description: str
enabled_env: str
SERVICES = [
ServiceDefinition(
key="ktts",
title="Kokoro TTS",
path="/services/ktts",
directory="ktts",
module_file="app.py",
app_object="app",
description="ONNX Kokoro text-to-speech service.",
enabled_env="MAESTER_ENABLE_KTTS",
),
ServiceDefinition(
key="musicgen",
title="MusicGen",
path="/services/musicgen",
directory="musicgen",
module_file="app.py",
app_object="app",
description="Prompt-to-music generation service.",
enabled_env="MAESTER_ENABLE_MUSICGEN",
),
ServiceDefinition(
key="whisper",
title="Whisper Operator",
path="/services/whisper",
directory="whisper",
module_file="main.py",
app_object="app",
description="Transcription, clipping, publishing, and video automation service.",
enabled_env="MAESTER_ENABLE_WHISPER",
),
ServiceDefinition(
key="ffmpeg_automation",
title="FFmpeg Automation Hub",
path="/services/ffmpeg",
directory="ffmpeg_automation",
module_file="app.py",
app_object="app",
description="Common FFmpeg task automation API.",
enabled_env="MAESTER_ENABLE_FFMPEG_AUTOMATION",
),
ServiceDefinition(
key="render_engine",
title="Ava2lon Render Engine",
path="/services/render",
directory="render_engine",
module_file="app.py",
app_object="app",
description="CPU-first rendering engine and studio dashboard.",
enabled_env="MAESTER_ENABLE_RENDER_ENGINE",
),
]
@contextmanager
def _service_import_path(service_root: Path) -> Iterator[None]:
original_path = list(sys.path)
sys.path.insert(0, str(service_root))
try:
yield
finally:
sys.path[:] = original_path
def service_enabled(service: ServiceDefinition) -> bool:
return os.getenv(service.enabled_env, "true").strip().lower() in {"1", "true", "yes", "on"}
def load_service_app(base_services_dir: Path, service: ServiceDefinition) -> FastAPI:
service_root = base_services_dir / service.directory
module_path = service_root / service.module_file
if not module_path.exists():
raise RuntimeError(f"Missing module for {service.key}: {module_path}")
module_name = f"maester_embedded_{service.key}"
spec = importlib.util.spec_from_file_location(module_name, module_path)
if not spec or not spec.loader:
raise RuntimeError(f"Cannot import {service.key} from {module_path}")
module = importlib.util.module_from_spec(spec)
with _service_import_path(service_root):
spec.loader.exec_module(module)
app = getattr(module, service.app_object, None)
if app is None:
raise RuntimeError(f"{service.key} does not expose {service.app_object}")
return app
|