pixano.inference.providers.pixano_inference

PixanoInferenceProvider(url)

Bases: HTTPProvider

Provider for pixano-inference server.

This provider communicates with a pixano-inference server using its HTTP API. It translates between Pixano's types and the pixano-inference API format.

Example

from pixano.inference import get_provider

provider = get_provider("pixano-inference", url="http://localhost:8000")
models = await provider.list_models()

Source code in pixano/inference/providers/base.py

def __init__(self, url: str):
    """Initialize the HTTP provider.

    Args:
        url: The base URL of the inference server.
    """
    self._url = url.rstrip("/")
    self._client: httpx.AsyncClient | None = None

name property

Provider name.

connect(url) async classmethod

Connect to a pixano-inference server.

Parameters:

Name	Type	Description	Default
url	str	The URL of the pixano-inference server.	required

Returns:

Type	Description
PixanoInferenceProvider	A connected PixanoInferenceProvider instance.

Raises:

Type	Description
ProviderConnectionError	If connection fails.

Source code in pixano/inference/providers/pixano_inference.py

@classmethod
async def connect(cls, url: str) -> "PixanoInferenceProvider":
    """Connect to a pixano-inference server.

    Args:
        url: The URL of the pixano-inference server.

    Returns:
        A connected PixanoInferenceProvider instance.

    Raises:
        ProviderConnectionError: If connection fails.
    """
    provider = cls(url=url)
    try:
        await provider.get("ready")
    except Exception:
        try:
            await provider.get("health")
        except Exception:
            try:
                await provider.get("app/settings/")
            except Exception as e:
                raise ProviderConnectionError(f"Failed to connect to pixano-inference at {url}: {e}") from e
    return provider

get_capabilities() async

Return provider capabilities.

Source code in pixano/inference/providers/pixano_inference.py

async def get_capabilities(self) -> ProviderCapabilities:
    """Return provider capabilities."""
    return ProviderCapabilities(
        tasks=[
            InferenceTask.MASK_GENERATION,
            InferenceTask.VIDEO_MASK_GENERATION,
            InferenceTask.OBJECT_DETECTION,
            InferenceTask.TEXT_GENERATION,
        ],
        supports_batching=True,
        supports_streaming=False,
    )

get_server_info() async

Get server information.

Source code in pixano/inference/providers/pixano_inference.py

async def get_server_info(self) -> ServerInfo:
    """Get server information."""
    response = await self.get("app/settings/")
    data = response.json()
    return ServerInfo(
        app_name=data.get("app_name", ""),
        app_version=data.get("app_version", "unknown"),
        app_description=data.get("app_description", ""),
        num_cpus=data.get("num_cpus"),
        num_gpus=data.get("num_gpus", 0),
        num_nodes=data.get("num_nodes", 1),
        gpus_used=data.get("gpus_used", []),
        gpu_to_model=data.get("gpu_to_model", {}),
        models=data.get("models", []),
        models_to_task=data.get("models_to_task", {}),
    )

image_mask_generation(input_data, timeout=60.0) async

Generate masks for an image.

Source code in pixano/inference/providers/pixano_inference.py

async def image_mask_generation(
    self,
    input_data: ImageMaskGenerationInput,
    timeout: float = 60.0,
) -> ImageMaskGenerationResult:
    """Generate masks for an image."""
    request_data = self._build_mask_request(input_data)
    response = await self.post("tasks/image/mask_generation/", json=request_data, timeout=timeout)
    return self._parse_mask_response(response.json())

image_zero_shot_detection(input_data, timeout=60.0) async

Detect objects using zero-shot detection.

Source code in pixano/inference/providers/pixano_inference.py

async def image_zero_shot_detection(
    self,
    input_data: ImageZeroShotDetectionInput,
    timeout: float = 60.0,
) -> ImageZeroShotDetectionResult:
    """Detect objects using zero-shot detection."""
    request_data = self._build_detection_request(input_data)
    response = await self.post("tasks/image/zero_shot_detection/", json=request_data, timeout=timeout)
    return self._parse_detection_response(response.json())

list_models(task=None) async

List available models.

Source code in pixano/inference/providers/pixano_inference.py

async def list_models(self, task: InferenceTask | None = None) -> list[ModelInfo]:
    """List available models."""
    response = await self.get("app/models/")
    models_data = response.json()
    models = [
        ModelInfo(
            name=m["name"],
            task=m["task"],
            model_path=m.get("model_path"),
            model_class=m.get("model_class"),
            provider=m.get("provider"),
        )
        for m in models_data
    ]

    if task is not None:
        models = [m for m in models if m.task == task.value]

    return models

text_image_conditional_generation(input_data, timeout=60.0) async

Generate text conditioned on images.

Source code in pixano/inference/providers/pixano_inference.py

async def text_image_conditional_generation(
    self,
    input_data: TextImageConditionalGenerationInput,
    timeout: float = 60.0,
) -> TextImageConditionalGenerationResult:
    """Generate text conditioned on images."""
    request_data = self._build_text_generation_request(input_data)
    response = await self.post(
        "tasks/multimodal/text-image/conditional_generation/", json=request_data, timeout=timeout
    )
    return self._parse_text_generation_response(response.json())

video_mask_generation(input_data, timeout=120.0) async

Generate masks for video frames.

Source code in pixano/inference/providers/pixano_inference.py

async def video_mask_generation(
    self,
    input_data: VideoMaskGenerationInput,
    timeout: float = 120.0,
) -> VideoMaskGenerationResult:
    """Generate masks for video frames."""
    request_data = self._build_video_mask_request(input_data)
    response = await self.post("tasks/video/mask_generation/", json=request_data, timeout=timeout)
    return self._parse_video_mask_response(response.json())