Skip to content
GitHub

API Reference

Classes

Section titled “Classes”

Exposes timeline rendering — sync and queued — under `multimedia.compose`.
__init__
def __init__(
    *,
    processor: Any,
    task_manager: Any,
    task_name: str,
    storage: Any,
    path_prefix: str,
    idempotency_manager: Any = None
) -> None
render
async def render(timeline: Timeline) -> Result[MediaAsset, VideoGenerationError]
submit_render
async def submit_render(
    timeline: Timeline,
    idempotency_key: str | None = None
) -> JobHandle

Thin wrapper around lexigram-tasks' IdempotencyResult.

Exists so multimedia callers don’t need to import lexigram-tasks types directly — see design spec ‘Async job model’.

from_idempotency_result
def from_idempotency_result(
    cls,
    result: Any,
    *,
    is_duplicate: bool | None = None
) -> JobHandle

Umbrella configuration for the multimedia subsystem.

Emitted after a generate()/submit() call completes (success or failure).

Audio/video/image generation: TTS, music, video, image, upscale, interpolate, and beat-analysis subsystems.
configure
def configure(
    cls,
    config: MultimediaConfig | None = None
) -> DynamicModule
stub
def stub(
    cls,
    config: MultimediaConfig | None = None
) -> DynamicModule

Provider that registers all seven core multimedia sub-providers.

Hardcodes wiring of the 7 core siblings (matching AIProvider.register()‘s treatment of llm/vector/rag) rather than relying purely on entry-point discovery — each needs its own typed config sub-object from MultimediaConfig, which a generic entry-point loop can’t supply.

__init__
def __init__(config: MultimediaConfig | None = None) -> None
register
async def register(container: ContainerRegistrarProtocol) -> None
boot
async def boot(container: ContainerResolverProtocol) -> None
shutdown
async def shutdown() -> None
health_check
async def health_check(timeout: float = 5.0) -> HealthCheckResult
tts
property tts() -> Any
music
property music() -> Any
video
property video() -> VideoAccessor
compose
property compose() -> ComposeAccessor
image
property image() -> Any
upscale
property upscale() -> Any
interpolate
property interpolate() -> Any
beat
property beat() -> BeatAccessor

Wraps one sub-provider's backend + task manager + storage normalizer.
__init__
def __init__(
    *,
    backend: Any,
    task_manager: Any,
    task_name: str,
    storage: Any,
    path_prefix: str,
    idempotency_manager: Any = None,
    cache_backend: Any = None,
    event_bus: Any = None,
    media_type: str = '',
    backend_method: str = 'generate'
) -> None
generate
async def generate(request: _Req) -> Result[MediaAsset, MultimediaError]
submit
async def submit(
    request: _Req,
    idempotency_key: str | None = None
) -> JobHandle

Builds up a video composition spec; render() executes it via a VideoProcessor.
__init__
def __init__() -> None
add_clip
def add_clip(
    asset: MediaAsset,
    *,
    transition_in: TransitionSpec | None = None
) -> Timeline
set_narration
def set_narration(asset: MediaAsset) -> Timeline
set_music
def set_music(
    asset: MediaAsset,
    *,
    duck_under_narration: bool = False
) -> Timeline
add_captions
def add_captions(cues: list[SubtitleCue]) -> Timeline
add_overlay
def add_overlay(
    asset: MediaAsset,
    *,
    start: float = 0.0,
    end: float | None = None,
    fade_in: float = 0.0,
    fade_out: float = 0.0
) -> Timeline
add_audio
def add_audio(
    asset: MediaAsset,
    *,
    start: float = 0.0,
    volume: float = 1.0
) -> Timeline
set_fade_in
def set_fade_in(duration: float) -> Timeline
set_fade_out
def set_fade_out(duration: float) -> Timeline
set_base_fade_out
def set_base_fade_out(duration: float) -> Timeline
set_encode
def set_encode(spec: EncodeSpec) -> Timeline
render
async def render(
    processor: VideoProcessor,
    *,
    progress_callback: Callable[[float], None] | None = None
) -> Result[MediaAsset, VideoGenerationError]

Execute the composition pipeline via a VideoProcessor.

Assembles clips (single clip passes through directly, multiple clips concat with transitions), then applies the compose stage when any overlay, audio layer, fade, or encode option is set, then narration, music, and captions stages.

Parameters
ParameterTypeDescription
`processor`VideoProcessorProcessor that executes each pipeline operation.
`progress_callback`Callable[[float], None] | NoneOptional sync callback invoked with 0.0..1.0 progress during the compose stage.
Returns
TypeDescription
Result[MediaAsset, VideoGenerationError]Ok(final asset) on success, Err(error) on the first failing stage.

Note

Audio layers mix into the composed audio; a narration asset replaces it. Callers should use one or the other.

clips
property clips() -> list[MediaAsset]
transitions
property transitions() -> list[TransitionSpec]
narration
property narration() -> MediaAsset | None
music
property music() -> MediaAsset | None
duck_under_narration
property duck_under_narration() -> bool
captions
property captions() -> list[SubtitleCue]
overlays
property overlays() -> list[ComposeLayer]
audio_layers
property audio_layers() -> list[ComposeAudioLayer]
fade_in
property fade_in() -> float
fade_out
property fade_out() -> float
base_fade_out
property base_fade_out() -> float
encode
property encode() -> EncodeSpec | None
to_params
def to_params() -> dict[str, Any]
from_params
def from_params(
    cls,
    params: dict[str, Any]
) -> Timeline

__init__
def __init__(processor: Any) -> None
run
async def run(params: dict[str, Any]) -> dict[str, Any]

Exposes both video generation and video processing under `multimedia.video`.
__init__
def __init__(
    *,
    generation: SubsystemAccessor,
    processing: SubsystemAccessor,
    storage: Any,
    path_prefix: str,
    video_upscale_service: Any | None = None,
    video_interpolation_service: Any | None = None
) -> None
generate
async def generate(request: VideoRequest) -> Result[MediaAsset, MultimediaError]
submit
async def submit(
    request: VideoRequest,
    idempotency_key: str | None = None
) -> JobHandle
process
async def process(operation: VideoOperation) -> Result[MediaAsset, MultimediaError]
submit_process
async def submit_process(
    operation: VideoOperation,
    idempotency_key: str | None = None
) -> JobHandle
upscale_video
async def upscale_video(
    asset: MediaAsset,
    *,
    scale_factor: Literal[2, 4] = 4
) -> Result[MediaAsset, MultimediaError]

Upscale a whole video by a factor of 2 or 4.

Requires a whole-video upscaling service, which is only built when a VideoProcessor was configured in the upscale subsystem.

Parameters
ParameterTypeDescription
`asset`MediaAssetSource video asset to upscale.
`scale_factor`Literal[2, 4]Upscale multiplier, 2 or 4. Defaults to 4.
Returns
TypeDescription
Result[MediaAsset, MultimediaError]Ok(upscaled_asset) on success, Err(ProviderNotInstalledError) if no VideoProcessor was configured.

Example

result = await video.upscale_video(asset, scale_factor=2)
if result.is_ok():
    print(result.unwrap().url)
interpolate_video
async def interpolate_video(
    asset: MediaAsset,
    *,
    factor: Literal[2, 4] = 2,
    fps: float
) -> Result[MediaAsset, MultimediaError]

Interpolate a whole video to a higher frame rate (motion smoothing).

Requires a whole-video interpolation service, which is only built when a VideoProcessor was configured in the interpolate subsystem.

Parameters
ParameterTypeDescription
`asset`MediaAssetSource video asset to interpolate.
`factor`Literal[2, 4]Frame multiplier, 2 or 4. Defaults to 2.
`fps`floatOutput target frame rate in frames per second.
Returns
TypeDescription
Result[MediaAsset, MultimediaError]Ok(interpolated_asset) on success, Err(ProviderNotInstalledError) if no VideoProcessor was configured.

Example

result = await video.interpolate_video(asset, factor=2, fps=60.0)
if result.is_ok():
    print(result.unwrap().url)