pixano_inference.client

Pixano inference client.

PixanoInferenceClient(**data)

Bases: Settings

Pixano Inference Client.

Source code in pixano_inference/settings.py

def __init__(self, **data: Any):
    """Initialize the settings."""
    if "num_cpus" not in data:
        data["num_cpus"] = os.cpu_count()
    if "num_gpus" not in data:
        data["num_gpus"] = detect_num_gpus()

    super().__init__(**data)

connect(url) staticmethod

Connect to pixano inference.

Parameters:

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

Source code in pixano_inference/client.py

@staticmethod
def connect(url: str) -> "PixanoInferenceClient":
    """Connect to pixano inference.

    Args:
        url: The URL of the pixano inference server.
    """
    server_settings = requests.get(f"{url}/app/settings/").json()
    server_settings = {k: v for k, v in server_settings.items() if v is not None}
    client = PixanoInferenceClient(url=url, **server_settings)
    return client

delete(path, **kwargs) async

Perform a DELETE request to the pixano inference server.

Parameters:

Name	Type	Description	Default
path	str	The path of the request.	required
kwargs	Any	The keyword arguments to pass to the request or httpx client.	{}

Source code in pixano_inference/client.py

async def delete(self, path: str, **kwargs: Any) -> Response:
    """Perform a DELETE request to the pixano inference server.

    Args:
        path: The path of the request.
        kwargs: The keyword arguments to pass to the request or httpx client.
    """
    return await self._rest_call(path=path, method="DELETE", **kwargs)

detection(request) async

Perform an inference to perform zero-shot detection.

Source code in pixano_inference/client.py

async def detection(
    self,
    request: DetectionRequest,
) -> DetectionResponse:
    """Perform an inference to perform zero-shot detection."""
    return await self.inference(
        route="inference/detection/",
        request=request,
        response_type=DetectionResponse,
    )

get(path, **kwargs) async

Perform a GET request to the pixano inference server.

Parameters:

Name	Type	Description	Default
path	str	The path of the request.	required
kwargs	Any	The keyword arguments to pass to the request or httpx client.	{}

Source code in pixano_inference/client.py

async def get(self, path: str, **kwargs: Any) -> Response:
    """Perform a GET request to the pixano inference server.

    Args:
        path: The path of the request.
        kwargs: The keyword arguments to pass to the request or httpx client.
    """
    return await self._rest_call(path=path, method="GET", **kwargs)

get_settings() async

Get the settings for the pixano inference server.

Source code in pixano_inference/client.py

async def get_settings(self) -> Settings:
    """Get the settings for the pixano inference server."""
    response = await self.get("app/settings/")
    raise_if_error(response)
    return Settings(**response.json())

inference(route, request, response_type) async

Perform inference via a POST request to the pixano inference server.

Parameters:

Name	Type	Description	Default
route	str	The route for the request.	required
request	BaseRequest	The request payload.	required
response_type	type[BaseResponse]	The expected response type.	required

Returns:

Type	Description
BaseResponse	The parsed response from the server.

Source code in pixano_inference/client.py

async def inference(
    self,
    route: str,
    request: BaseRequest,
    response_type: type[BaseResponse],
) -> BaseResponse:
    """Perform inference via a POST request to the pixano inference server.

    Args:
        route: The route for the request.
        request: The request payload.
        response_type: The expected response type.

    Returns:
        The parsed response from the server.
    """
    response = await self.post(route, json=request.model_dump())
    return response_type.model_validate(response.json())

list_models() async

List all models.

Source code in pixano_inference/client.py

async def list_models(self) -> list[ModelInfo]:
    """List all models."""
    response = await self.get("app/models/")
    return [ModelInfo.model_construct(**model) for model in response.json()]

post(path, **kwargs) async

Perform a POST request to the pixano inference server.

Parameters:

Name	Type	Description	Default
path	str	The path of the request.	required
kwargs	Any	The keyword arguments to pass to the request or httpx client.	{}

Source code in pixano_inference/client.py

async def post(self, path: str, **kwargs: Any) -> Response:
    """Perform a POST request to the pixano inference server.

    Args:
        path: The path of the request.
        kwargs: The keyword arguments to pass to the request or httpx client.
    """
    return await self._rest_call(path=path, method="POST", **kwargs)

put(path, **kwargs) async

Perform a PUT request to the pixano inference server.

Parameters:

Name	Type	Description	Default
path	str	The path of the request.	required
kwargs	Any	The keyword arguments to pass to the request or httpx client.	{}

Source code in pixano_inference/client.py

async def put(self, path: str, **kwargs: Any) -> Response:
    """Perform a PUT request to the pixano inference server.

    Args:
        path: The path of the request.
        kwargs: The keyword arguments to pass to the request or httpx client.
    """
    return await self._rest_call(path=path, method="PUT", **kwargs)

segmentation(request) async

Perform an inference to perform image segmentation.

Source code in pixano_inference/client.py

async def segmentation(
    self,
    request: SegmentationRequest,
) -> SegmentationResponse:
    """Perform an inference to perform image segmentation."""
    return await self.inference(
        route="inference/segmentation/",
        request=request,
        response_type=SegmentationResponse,
    )

tracking(request) async

Perform an inference to perform video tracking.

Source code in pixano_inference/client.py

async def tracking(
    self,
    request: TrackingRequest,
) -> TrackingResponse:
    """Perform an inference to perform video tracking."""
    return await self.inference(
        route="inference/tracking/",
        request=request,
        response_type=TrackingResponse,
    )

vlm(request) async

Perform an inference for vision-language model generation.

Source code in pixano_inference/client.py

async def vlm(
    self,
    request: VLMRequest,
) -> VLMResponse:
    """Perform an inference for vision-language model generation."""
    return await self.inference(
        route="inference/vlm/",
        request=request,
        response_type=VLMResponse,
    )

raise_if_error(response)

Raise an error from a response.

Source code in pixano_inference/client.py

def raise_if_error(response: Response) -> None:
    """Raise an error from a response."""
    if response.is_success:
        return
    error_out = f"HTTP {response.status_code}: {response.reason_phrase}"
    try:
        json_detail = response.json()
    except Exception:
        json_detail = {}

    detail = json_detail.get("detail", None)
    if detail is not None:
        error_out += f" - {detail}"
    error = json_detail.get("error", None)
    if error is not None:
        error_out += f" - {error}"
    raise HTTPException(response.status_code, detail=error_out)