API Reference

Classes

`HatUpscaleProvider`

Talks to a hat_server.py reference server via UpscaleProvider.

__init__

def __init__(
    base_url: str,
    timeout: float = 30.0,
    retry: RetryPolicyProtocol | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None
) -> None

source

upscale

async def upscale(request: UpscaleRequest) -> Result[MediaAsset, UpscaleError]

source

`RealEsrganUpscaleProvider`

source

Talks to a real_esrgan_server.py reference server via UpscaleProvider.

__init__

def __init__(
    base_url: str,
    timeout: float = 30.0,
    retry: RetryPolicyProtocol | None = None,
    circuit_breaker: CircuitBreakerProtocol | None = None
) -> None

source

upscale

async def upscale(request: UpscaleRequest) -> Result[MediaAsset, UpscaleError]

source

`UpscaleConfig`

source

Configuration for the upscale generation subsystem.

`UpscaleGenerationProvider`

source

Provider that registers a configured UpscaleProvider backend.

__init__

def __init__(config: UpscaleConfig | None = None) -> None

source

async def register(container: ContainerRegistrarProtocol) -> None

source

boot

async def boot(container: ContainerResolverProtocol) -> None

source

No async I/O needed at boot beyond what register() already did.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

`UpscaleModule`

source

Image/video upscale integration.

configure

def configure(
    cls,
    config: UpscaleConfig | None = None
) -> DynamicModule

source

stub

def stub(
    cls,
    config: UpscaleConfig | None = None
) -> DynamicModule

source

`UpscaleTask`

source

Callable task handler — wraps an UpscaleProvider backend for lexigram-tasks.

Returns a plain dict (never raw MediaAsset bytes) because lexigram-tasks’ result store JSON-serializes JobResult. The umbrella wraps this handler during MultimediaProvider.register() to persist any bytes to lexigram-storage BEFORE this dict is constructed.

__init__

def __init__(backend: UpscaleProvider) -> None

source

run

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

source

`VideoUpscaleService`

source

Extracts a video's frames, upscales each, then reassembles them.

Not itself an UpscaleProvider — a different method name (upscale_video, not upscale) and a different signature (a whole-video MediaAsset plus scale_factor, not an UpscaleRequest), since video upscaling isn’t a drop-in single-frame operation (design spec §6.3).

__init__

def __init__(
    upscale_provider: UpscaleProvider,
    video_processor: VideoProcessor
) -> None

source

upscale_video

async def upscale_video(
    asset: MediaAsset,
    *,
    scale_factor: Literal[2, 4] = 4
) -> Result[MediaAsset, MultimediaError]

source