How to Extend Pixelle-Video with Custom TTS Engines: A Complete Integration Guide

You can extend Pixelle-Video with custom TTS engines by implementing an async function matching the edge_tts signature, registering it in TTSService, and optionally exposing it via a new inference_mode in your configuration.

Pixelle-Video's pluggable TTS subsystem enables runtime selection of speech synthesis engines without modifying downstream video generation code. This guide walks you through the complete integration process using the actual source code from the AIDC-AI/Pixelle-Video repository.

---

Understanding the TTS Architecture in Pixelle-Video

Before extending, you need to understand how Pixelle-Video routes TTS requests. The system uses three core components working together.

TTSService: The Central Router

Located at pixelle_video/services/tts_service.py, the TTSService class acts as the public façade. Its __call__ method reads the inference_mode flag—either from the call site or global config—and dispatches to the appropriate implementation.

Built-in Engine Implementations

The repository ships with two built-in paths:

• Local Edge-TTS (pixelle_video/utils/tts_util.py): Async function calling Microsoft's free Edge TTS service with retry logic and rate limiting
• ComfyUI Workflows (workflows/*.json): JSON-described graphs that can invoke external HTTP TTS services or locally-hosted models

Voice Catalog and Configuration

The EDGE_TTS_VOICES dictionary in pixelle_video/tts_voices.py supplies UI voice options, while config.example.yaml stores defaults for inference mode, workflow selection, and engine-specific parameters.

---

Step-by-Step: Adding a Custom TTS Engine

Follow these five steps to integrate any HTTP-based, locally-hosted, or third-party TTS engine into Pixelle-Video.

Step 1: Implement the Async TTS Function

Create a new file in pixelle_video/utils/ with a function matching the edge_tts signature. This contract ensures compatibility with TTSService.

# pixelle_video/utils/my_custom_tts.py

import aiohttp
from pathlib import Path
from loguru import logger


async def my_custom_tts(
    text: str,
    voice: str = "default",
    rate: str = "+0%",
    output_path: str | None = None,
) -> bytes:
    """
    Call your own HTTP-based TTS endpoint.
    
    The endpoint must accept JSON {text, voice, rate} and return raw MP3 bytes.
    Matches the signature of edge_tts in tts_util.py for drop-in compatibility.
    """
    api_url = "https://my-tts.example.com/api/synthesize"
    payload = {"text": text, "voice": voice, "rate": rate}
    
    async with aiohttp.ClientSession() as session:
        async with session.post(api_url, json=payload) as resp:
            resp.raise_for_status()
            audio = await resp.read()

    if output_path:
        Path(output_path).parent.mkdir(parents=True, exist_ok=True)
        with open(output_path, "wb") as f:
            f.write(audio)

    logger.info(f"✅ Custom TTS produced {len(audio)} bytes")
    return audio

Key requirements for your function:

• Accepts: text, voice, rate, and optional output_path parameters
• Returns: Raw audio bytes (typically MP3)
• Writes to disk: When output_path is provided, creating parent directories as needed
• Raises exceptions: On failure so TTSService can surface errors appropriately

Step 2: Register Your Engine in TTSService

Modify pixelle_video/services/tts_service.py to import your function and add a new dispatch branch.

# pixelle_video/services/tts_service.py (excerpt)

from pixelle_video.utils.my_custom_tts import my_custom_tts   # ← new import


class TTSService(ComfyBaseService):
    ...
    
    async def __call__(self, *, inference_mode: Optional[str] = None, **kw):
        mode = inference_mode or self.config.get("inference_mode", "local")
        
        if mode == "local":
            return await self._call_local_tts(**kw)
        elif mode == "custom":          # ← new branch

            return await self._call_custom_tts(**kw)
        else:                           # existing comfyui path

            return await self._call_comfyui_tts(inference_mode=mode, **kw)
    
    async def _call_custom_tts(
        self,
        text: str,
        voice: Optional[str] = None,
        speed: Optional[float] = None,
        output_path: Optional[str] = None,
    ) -> str:
        # Convert speed multiplier to rate string (reuse existing helper)

        rate = speed_to_rate(speed or 1.0)
        
        audio_bytes = await my_custom_tts(
            text=text,
            voice=voice or "default",
            rate=rate,
            output_path=output_path,
        )
        
        # Return path expected by downstream processors

        return output_path or "/tmp/custom_tts.mp3"

The speed_to_rate helper converts a float multiplier (e.g., 1.5 for 1.5× speed) to the string format your TTS API expects, such as "+50%".

Step 3: Add Voice Catalog Entries (Optional)

If your engine supports multiple voices and you want them visible in the UI, extend pixelle_video/tts_voices.py:

# pixelle_video/tts_voices.py (excerpt)

CUSTOM_TTS_VOICES = {
    "my-voice-1": "custom-voice-id-123",
    "my-voice-2": "custom-voice-id-456",
    "my-voice-3": "custom-voice-id-789",
}

# Or merge with existing catalog

ALL_VOICES = {**EDGE_TTS_VOICES, **CUSTOM_TTS_VOICES}

The UI reads this dictionary to populate voice selection dropdowns.

Step 4: Update Configuration Schema

Register your new inference_mode in pixelle_video/config/schema.py so it validates correctly in config.yaml:

# pixelle_video/config/schema.py (excerpt)

from pydantic import BaseModel, Field
from typing import Literal, Optional


class TTSConfig(BaseModel):
    inference_mode: Literal["local", "comfyui", "custom"] = "local"
    default_voice: str = "en-US-Standard-A"
    default_speed: float = 1.0
    
    # Custom engine-specific settings

    custom: Optional[dict] = Field(default_factory=lambda: {
        "api_url": "https://my-tts.example.com/api/synthesize",
        "api_key": None,
        "timeout": 30,
    })

Then set your new mode as default in config.yaml:

# config.yaml (derived from config.example.yaml)

comfyui:
  tts:
    inference_mode: custom          # new default

    default_voice: "my-voice-1"
    default_speed: 1.0
    custom:
      api_url: "https://my-tts.example.com/api/synthesize"
      api_key: "${MY_TTS_API_KEY}"  # loaded from environment

      timeout: 30

Step 5: (Optional) Create a ComfyUI Workflow

For engines that work better within ComfyUI's node graph, create a workflow JSON in workflows/selfhost/:

// workflows/selfhost/tts_myengine.json
{
  "nodes": [
    {
      "type": "LoadAudioConfig",
      "params": {
        "text": "${text}",
        "voice": "${voice}",
        "rate": "${rate}"
      },
      "id": "config"
    },
    {
      "type": "HTTPRequest",
      "params": {
        "url": "https://my-tts.example.com/api/synthesize",
        "method": "POST",
        "headers": { "Content-Type": "application/json" },
        "body": "{ \"text\": \"${config.text}\", \"voice\": \"${config.voice}\", \"rate\": \"${config.rate}\" }",
        "output_type": "binary"
      },
      "id": "tts_req"
    },
    {
      "type": "SaveFile",
      "params": {
        "filename": "output/${uuid()}.mp3",
        "input": "tts_req"
      },
      "id": "save"
    }
  ],
  "outputs": { "audio": "save" }
}

Reference this workflow in config.yaml:

comfyui:
  tts:
    default_workflow: "selfhost/tts_myengine.json"

---

Complete Usage Example

With your custom TTS engine fully integrated, invoke it from application code:

import asyncio
from pixelle_video import tts


async def demo():
    audio_path = await tts(
        text="Welcome to Pixelle-Video with custom TTS!",
        inference_mode="custom",          # selects your engine

        voice="my-voice-1",
        speed=1.1,                        # 10% faster

    )
    print("Generated audio at:", audio_path)


asyncio.run(demo())

The frame_processor and video generation pipeline remain unchanged because they consume the file path returned by TTSService.

---

Summary

Extending Pixelle-Video with custom TTS engines requires five key steps:

• Implement an async function matching the edge_tts signature in pixelle_video/utils/
• Register a new dispatch branch in TTSService.__call__ at pixelle_video/services/tts_service.py
• Add voice catalog entries in pixelle_video/tts_voices.py for UI visibility
• Update the Pydantic schema in pixelle_video/config/schema.py to validate your new inference_mode
• Optionally create a ComfyUI workflow JSON for node-graph-based execution

The pluggable architecture in pixelle_video/services/tts_service.py ensures that video generation, frame processing, and UI components remain decoupled from specific TTS implementations.

---

Frequently Asked Questions

What async signature must my custom TTS function implement?

Your function must accept text: str, voice: str, rate: str, and optional output_path: str | None, returning bytes of audio data. This matches the edge_tts function in pixelle_video/utils/tts_util.py. The TTSService._call_custom_tts helper handles speed-to-rate conversion and path management.

Can I use a local model instead of an HTTP API?

Yes. Instead of aiohttp requests, your utility function can load a local model with libraries like transformers, torch, or TTS (Coqui). Ensure the function remains async—use asyncio.to_thread or torch.inference_mode context managers to prevent blocking the event loop. The output must still be MP3 bytes or written to output_path.

How do I expose my engine in the Pixelle-Video UI?

Add your voices to pixelle_video/tts_voices.py or create a separate catalog dictionary. The UI populates dropdowns from this module. Ensure your inference_mode value appears in the config schema at pixelle_video/config/schema.py so the settings panel can persist user selections. The TTSService reads the mode from config when users don't specify it explicitly in API calls.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}