summaryrefslogtreecommitdiff
path: root/src/mcp_sd15/pipeline_factory.py
diff options
context:
space:
mode:
authortanaka takashi <isazaka@gmail.com>2026-09-12 18:04:56 +0900
committertanaka takashi <isazaka@gmail.com>2026-09-12 18:04:56 +0900
commitcba15a62829a0fe57a03c8031f981e8f790e2efc (patch)
tree3bc0ef1a243f356f55c2fa97f63c3bfd5c77e8fc /src/mcp_sd15/pipeline_factory.py
feat: Stable Diffusion 1.x MCPサーバを実装
仕様書・設計書に基づき、ローカルCUDA GPU上のStable Diffusion 1.x互換モデルをMCP経由で利用するサーバを新規実装。 実装内容: - generate_image、generate_guided_image、inpaint_imageの3ツールをMCP SDK v2で公開 - TOMLからcheckpoint、VAE、複数LoRAと重み、ControlNet、生成設定を読み込み - モデル種別ごとの固定ディレクトリ解決と起動時の設定・ファイル存在検証を追加 - Depth/MiDaS、OpenPose body/full、ControlNet Inpaintの前処理とパイプライン切替を実装 - FP16/CUDA、1枚固定生成、ControlNetオンデマンドロード、単一GPU直列実行を実装 - Base64 PNG/JPEG入力の形式・サイズ検証と任意のdata URI入力に対応 - seed、サイズ、モード、時刻、処理時間をPNG metadataへ記録 - 任意のローカルPNG保存、生成エラー分類、CUDA OOM後のキャッシュ回復を追加 - --configと--versionを備えたstdio CLIエントリポイントを追加 開発・運用: - Python 3.11およびuvによる依存・lock管理を構成 - config.example.tomlとセットアップ・運用手順をREADMEへ追加 - Ruff、Pyright、pytestの設定を追加 - 設定、パス解決、schema、画像検証、サービスルーティング、排他制御、MCP画像返却をテスト 検証結果: - pytest: 30 passed - Ruff lint/format: passed - Pyright: 0 errors, 0 warnings - CLI versionおよびMCPクライアント経由のPNG返却を確認 - 実GPU smoke testはCUDA GPUとモデルファイルがないため対象外
Diffstat (limited to 'src/mcp_sd15/pipeline_factory.py')
-rw-r--r--src/mcp_sd15/pipeline_factory.py133
1 files changed, 133 insertions, 0 deletions
diff --git a/src/mcp_sd15/pipeline_factory.py b/src/mcp_sd15/pipeline_factory.py
new file mode 100644
index 0000000..cd92690
--- /dev/null
+++ b/src/mcp_sd15/pipeline_factory.py
@@ -0,0 +1,133 @@
+"""Lazy Diffusers construction kept outside the MCP layer."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from .config import AppConfig
+from .errors import ConfigError
+
+
+class PipelineFactory:
+ """Build SD1.x pipelines in FP16 without network access."""
+
+ def __init__(self, config: AppConfig) -> None:
+ self.config = config
+
+ @staticmethod
+ def _runtime() -> tuple[Any, Any]:
+ try:
+ import diffusers
+ import torch
+ except ImportError as exc:
+ raise ConfigError(f"inference dependency is unavailable: {exc}") from exc
+ if not torch.cuda.is_available():
+ raise ConfigError("CUDA GPU is required but torch.cuda.is_available() is false")
+ return torch, diffusers
+
+ def create_base(self) -> Any:
+ torch, diffusers = self._runtime()
+ try:
+ pipeline = diffusers.StableDiffusionPipeline.from_single_file(
+ str(self.config.model.checkpoint_path()),
+ torch_dtype=torch.float16,
+ local_files_only=True,
+ )
+ vae_path = self.config.model.vae_path()
+ if vae_path is not None:
+ pipeline.vae = diffusers.AutoencoderKL.from_single_file(
+ str(vae_path),
+ torch_dtype=torch.float16,
+ local_files_only=True,
+ )
+ adapter_names: list[str] = []
+ adapter_weights: list[float] = []
+ for index, (item, path) in enumerate(
+ zip(self.config.model.loras, self.config.model.lora_paths(), strict=True)
+ ):
+ name = f"configured_lora_{index}"
+ pipeline.load_lora_weights(
+ str(path.parent),
+ weight_name=path.name,
+ adapter_name=name,
+ local_files_only=True,
+ )
+ adapter_names.append(name)
+ adapter_weights.append(item.weight)
+ if adapter_names:
+ pipeline.set_adapters(adapter_names, adapter_weights=adapter_weights)
+ pipeline.scheduler = self._scheduler(pipeline.scheduler, diffusers)
+ pipeline.set_progress_bar_config(disable=True)
+ return pipeline.to("cuda")
+ except ConfigError:
+ raise
+ except Exception as exc:
+ raise ConfigError(f"failed to load SD1.x model configuration: {exc}") from exc
+
+ def create_controlnet(self, path: Path) -> Any:
+ torch, diffusers = self._runtime()
+ try:
+ return diffusers.ControlNetModel.from_single_file(
+ str(path), torch_dtype=torch.float16, local_files_only=True
+ ).to("cuda")
+ except Exception as exc:
+ raise ConfigError(f"failed to load ControlNet model {path}: {exc}") from exc
+
+ @staticmethod
+ def create_guided_pipeline(base: Any, controlnet: Any) -> Any:
+ from diffusers.pipelines.controlnet.pipeline_controlnet import (
+ StableDiffusionControlNetPipeline,
+ )
+
+ components: dict[str, Any] = dict(
+ vae=base.vae,
+ text_encoder=base.text_encoder,
+ tokenizer=base.tokenizer,
+ unet=base.unet,
+ controlnet=controlnet,
+ scheduler=base.scheduler,
+ safety_checker=base.safety_checker,
+ feature_extractor=base.feature_extractor,
+ requires_safety_checker=getattr(base, "requires_safety_checker", False),
+ )
+ if getattr(base, "image_encoder", None) is not None:
+ components["image_encoder"] = base.image_encoder
+ pipeline = StableDiffusionControlNetPipeline(**components)
+ pipeline.set_progress_bar_config(disable=True)
+ return pipeline.to("cuda")
+
+ @staticmethod
+ def create_inpaint_pipeline(base: Any, controlnet: Any) -> Any:
+ from diffusers.pipelines.controlnet.pipeline_controlnet_inpaint import (
+ StableDiffusionControlNetInpaintPipeline,
+ )
+
+ components: dict[str, Any] = dict(
+ vae=base.vae,
+ text_encoder=base.text_encoder,
+ tokenizer=base.tokenizer,
+ unet=base.unet,
+ controlnet=controlnet,
+ scheduler=base.scheduler,
+ safety_checker=base.safety_checker,
+ feature_extractor=base.feature_extractor,
+ requires_safety_checker=getattr(base, "requires_safety_checker", False),
+ )
+ if getattr(base, "image_encoder", None) is not None:
+ components["image_encoder"] = base.image_encoder
+ pipeline = StableDiffusionControlNetInpaintPipeline(**components)
+ pipeline.set_progress_bar_config(disable=True)
+ return pipeline.to("cuda")
+
+ def _scheduler(self, scheduler: Any, diffusers: Any) -> Any:
+ classes = {
+ "dpmpp_2m": diffusers.DPMSolverMultistepScheduler,
+ "ddim": diffusers.DDIMScheduler,
+ "euler": diffusers.EulerDiscreteScheduler,
+ "euler_a": diffusers.EulerAncestralDiscreteScheduler,
+ }
+ options = dict(scheduler.config)
+ if self.config.generation.scheduler == "dpmpp_2m":
+ options.update(algorithm_type="dpmsolver++", solver_order=2)
+ return classes[self.config.generation.scheduler].from_config(options)