Media Processing Plugin Interface

Domain-specific plugin interface for media processing (write / file manipulation)

---

MediaProcessingPlugin

def MediaProcessingPlugin(
    args:VAR_POSITIONAL, kwargs:VAR_KEYWORD
):

Abstract base class for plugins that modify, convert, or extract media.

Processing plugins perform write operations that produce new files (format conversion, segment extraction, re-encoding, etc.).

How It Works

Host Process                              Worker Process (Isolated Env)
┌─────────────────────┐                  ┌─────────────────────────────┐
│                     │                  │                             │
│ plugin.execute(     │   HTTP/JSON      │  MediaProcessingPlugin      │
│   action="convert", │ ─────────────────▶    .execute(                │
│   input_path="...", │                  │       action="convert",     │
│   output_format=    │                  │       ...                   │
│     "mp4"           │                  │    )                        │
│ )                   │                  │                             │
│                     │                  │  # Calls internal method    │
│ {                   │  ◀───────────────│  # .convert(...)            │
│   "output_path":    │   (JSON)         │  # Returns new file path    │
│   "/tmp/out.mp4"    │                  │                             │
│ }                   │                  │                             │
└─────────────────────┘                  └─────────────────────────────┘

Processing plugins: - Use action parameter to dispatch to specific methods - Return JSON-serializable results (typically containing output file paths) - May create new files (converted media, extracted segments)

Example Implementation

A minimal processing plugin that demonstrates the interface:

from typing import List

class ExampleFFmpegPlugin(MediaProcessingPlugin):
    """Example FFmpeg-based media processing plugin implementation."""
    
    def __init__(self):
        self._config: Dict[str, Any] = {}

    @property
    def name(self) -> str:
        return "example-ffmpeg"
    
    @property
    def version(self) -> str:
        return "1.0.0"

    def initialize(self, config: Optional[Dict[str, Any]] = None) -> None:
        """Initialize with configuration."""
        self._config = config or {"ffmpeg_path": "ffmpeg"}

    def execute(
        self,
        action: str = "get_info",
        **kwargs
    ) -> Dict[str, Any]:
        """Dispatch to appropriate method based on action."""
        if action == "get_info":
            result = self.get_info(kwargs["file_path"])
            return result.to_dict()
        elif action == "convert":
            output = self.convert(
                kwargs["input_path"],
                kwargs["output_format"],
                **{k: v for k, v in kwargs.items() if k not in ["input_path", "output_format"]}
            )
            return {"output_path": output}
        elif action == "extract_segment":
            output = self.extract_segment(
                kwargs["input_path"],
                kwargs["start"],
                kwargs["end"],
                kwargs.get("output_path")
            )
            return {"output_path": output}
        else:
            raise ValueError(f"Unknown action: {action}")

    def get_info(self, file_path: Union[str, Path]) -> MediaMetadata:
        """Get metadata for a media file (mock implementation)."""
        return MediaMetadata(
            path=str(file_path),
            duration=120.5,
            format="mp4",
            size_bytes=15_000_000,
            video_streams=[{"codec": "h264", "width": 1920, "height": 1080}],
            audio_streams=[{"codec": "aac", "sample_rate": 48000}]
        )

    def convert(
        self,
        input_path: Union[str, Path],
        output_format: str,
        **kwargs
    ) -> str:
        """Convert media format (mock implementation)."""
        # In real implementation, would call ffmpeg here
        input_stem = Path(input_path).stem
        return f"/tmp/{input_stem}.{output_format}"

    def extract_segment(
        self,
        input_path: Union[str, Path],
        start: float,
        end: float,
        output_path: Optional[str] = None
    ) -> str:
        """Extract segment from media (mock implementation)."""
        if output_path:
            return output_path
        # Auto-generate output path
        input_p = Path(input_path)
        return f"/tmp/{input_p.stem}_segment_{start}_{end}{input_p.suffix}"

    def get_config_schema(self) -> Dict[str, Any]:
        """Return JSON Schema for configuration."""
        return {
            "type": "object",
            "properties": {
                "ffmpeg_path": {
                    "type": "string",
                    "default": "ffmpeg",
                    "description": "Path to ffmpeg binary"
                },
                "default_video_codec": {
                    "type": "string",
                    "enum": ["h264", "h265", "vp9"],
                    "default": "h264"
                }
            }
        }

    def get_current_config(self) -> Dict[str, Any]:
        """Return current configuration."""
        return self._config

    def cleanup(self) -> None:
        """Clean up resources."""
        pass

# Test the example plugin
plugin = ExampleFFmpegPlugin()
plugin.initialize({"ffmpeg_path": "/usr/bin/ffmpeg"})

print(f"Plugin: {plugin.name} v{plugin.version}")
print(f"Config schema: {plugin.get_config_schema()}")
print(f"Current config: {plugin.get_current_config()}")

Plugin: example-ffmpeg v1.0.0
Config schema: {'type': 'object', 'properties': {'ffmpeg_path': {'type': 'string', 'default': 'ffmpeg', 'description': 'Path to ffmpeg binary'}, 'default_video_codec': {'type': 'string', 'enum': ['h264', 'h265', 'vp9'], 'default': 'h264'}}}
Current config: {'ffmpeg_path': '/usr/bin/ffmpeg'}

# Test get_info action
info_result = plugin.execute(action="get_info", file_path="/path/to/video.mp4")
print(f"\nget_info result:")
print(f"  Duration: {info_result['duration']}s")
print(f"  Format: {info_result['format']}")
print(f"  Video streams: {info_result['video_streams']}")

get_info result:
  Duration: 120.5s
  Format: mp4
  Video streams: [{'codec': 'h264', 'width': 1920, 'height': 1080}]

# Test convert action
convert_result = plugin.execute(
    action="convert",
    input_path="/path/to/video.mp4",
    output_format="webm"
)
print(f"\nconvert result: {convert_result}")

convert result: {'output_path': '/tmp/video.webm'}

# Test extract_segment action
segment_result = plugin.execute(
    action="extract_segment",
    input_path="/path/to/video.mp4",
    start=10.0,
    end=25.5
)
print(f"\nextract_segment result: {segment_result}")

# Cleanup
plugin.cleanup()

extract_segment result: {'output_path': '/tmp/video_segment_10.0_25.5.mp4'}