pixano.app.routers.inference.mask_generation

ImageMaskGenerationOutput(**data)

Bases: BaseModel

Image mask generation output.

Source code in .venv/lib/python3.12/site-packages/pydantic/main.py

def __init__(self, /, **data: Any) -> None:
    """Create a new model by parsing and validating input data from keyword arguments.

    Raises [`ValidationError`][pydantic_core.ValidationError] if the input data cannot be
    validated to form a valid model.

    `self` is explicitly positional-only to allow `self` as a field name.
    """
    # `__tracebackhide__` tells pytest and some other tools to omit this function from tracebacks
    __tracebackhide__ = True
    validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self)
    if self is not validated_self:
        warnings.warn(
            'A custom validator is returning a value other than `self`.\n'
            "Returning anything other than `self` from a top level model validator isn't supported when validating via `__init__`.\n"
            'See the `model_validator` docs (https://docs.pydantic.dev/latest/concepts/validators/#model-validators) for more details.',
            stacklevel=2,
        )

VideoMaskGenerationOutput(**data)

Bases: BaseModel

Video masks generation output.

Source code in .venv/lib/python3.12/site-packages/pydantic/main.py

def __init__(self, /, **data: Any) -> None:
    """Create a new model by parsing and validating input data from keyword arguments.

    Raises [`ValidationError`][pydantic_core.ValidationError] if the input data cannot be
    validated to form a valid model.

    `self` is explicitly positional-only to allow `self` as a field name.
    """
    # `__tracebackhide__` tells pytest and some other tools to omit this function from tracebacks
    __tracebackhide__ = True
    validated_self = self.__pydantic_validator__.validate_python(data, self_instance=self)
    if self is not validated_self:
        warnings.warn(
            'A custom validator is returning a value other than `self`.\n'
            "Returning anything other than `self` from a top level model validator isn't supported when validating via `__init__`.\n"
            'See the `model_validator` docs (https://docs.pydantic.dev/latest/concepts/validators/#model-validators) for more details.',
            stacklevel=2,
        )

call_image_mask_generation(dataset_id, image, model, mask_table_name, settings, entity=None, bbox=None, points=None, labels=None, provider_name=None) async

Perform image mask generation on an image.

Parameters:

Name	Type	Description	Default
dataset_id	Annotated[str, Body(embed=True)]	The ID of the dataset to use.	required
image	Annotated[ViewModel, Body(embed=True)]	The image to use for detection.	required
entity	Annotated[EntityModel | None, Body(embed=True)]	The entity to use for detection.	None
model	Annotated[str, Body(embed=True)]	The name of the model to use.	required
mask_table_name	Annotated[str, Body(embed=True)]	The name of the table to use for masks in dataset.	required
settings	Annotated[Settings, Depends(get_settings)]	App settings.	required
bbox	Annotated[BBox | None, Body(embed=True)]	Input bounding box or None.	None
points	Annotated[list[list[int]] | None, Body(embed=True)]	Input points or None.	None
labels	Annotated[list[int] | None, Body(embed=True)]	Labels for input points, or None.	None
provider_name	Annotated[str | None, Body(embed=True)]	Optional provider name to route the request to.	None

Returns:

Type	Description
ImageMaskGenerationOutput	The generated mask.

Source code in pixano/app/routers/inference/mask_generation.py

@router.post(
    "/image",
    response_model=ImageMaskGenerationOutput,
)
async def call_image_mask_generation(
    dataset_id: Annotated[str, Body(embed=True)],
    image: Annotated[ViewModel, Body(embed=True)],
    model: Annotated[str, Body(embed=True)],
    mask_table_name: Annotated[str, Body(embed=True)],
    settings: Annotated[Settings, Depends(get_settings)],
    entity: Annotated[EntityModel | None, Body(embed=True)] = None,
    bbox: Annotated[BBox | None, Body(embed=True)] = None,
    points: Annotated[list[list[int]] | None, Body(embed=True)] = None,
    labels: Annotated[list[int] | None, Body(embed=True)] = None,
    provider_name: Annotated[str | None, Body(embed=True)] = None,
) -> ImageMaskGenerationOutput:
    """Perform image mask generation on an image.

    Args:
        dataset_id: The ID of the dataset to use.
        image: The image to use for detection.
        entity: The entity to use for detection.
        model: The name of the model to use.
        mask_table_name: The name of the table to use for masks in dataset.
        settings: App settings.
        bbox: Input bounding box or None.
        points: Input points or None.
        labels: Labels for input points, or None.
        provider_name: Optional provider name to route the request to.

    Returns:
        The generated mask.
    """
    global store_last_image_id

    dataset = get_dataset(dataset_id=dataset_id, dir=settings.library_dir, media_dir=settings.media_dir)
    provider = get_provider_from_settings(settings=settings, provider_name=provider_name)

    if not is_image(dataset.schema.schemas[image.table_info.name]):
        raise HTTPException(status_code=400, detail="Image must be an image.")

    image_row: Image = image.to_row(dataset)
    entity_row: Entity | None = entity.to_row(dataset) if entity else None
    source = get_model_source(dataset=dataset, model=model)

    try:
        mask_output: tuple[
            CompressedRLE, float, NDArrayFloat | None, list[NDArrayFloat] | None
        ] = await image_mask_generation(
            provider=provider,
            source=source,
            media_dir=settings.media_dir,
            image=image_row,
            entity=entity_row,
            image_embedding=None,
            high_resolution_features=None,
            reset_predictor=store_last_image_id != image_row.id,
            bbox=bbox,
            points=points,
            labels=labels,
        )
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e)) from e

    # store image_id to check against for next generation
    store_last_image_id = image_row.id

    output: ImageMaskGenerationOutput = ImageMaskGenerationOutput(
        mask=AnnotationModel.from_row(
            row=mask_output[0],  # right now we don't use other ouputs (score, ...)
            table_info=TableInfo(name=mask_table_name, group="annotations", base_schema="CompressedRLE"),
        )
    )
    return output

call_video_mask_generation(dataset_id, video, model, settings, bbox=None, points=None, labels=None, provider_name=None) async

Perform video mask generation on a video.

Parameters:

Name	Type	Description	Default
dataset_id	Annotated[str, Body(embed=True)]	The ID of the dataset to use.	required
video	Annotated[list[ViewModel], Body(embed=True)]	The video as a list of SequenceFrame (or a subset).	required
model	Annotated[str, Body(embed=True)]	The name of the model to use.	required
settings	Annotated[Settings, Depends(get_settings)]	App settings.	required
bbox	Annotated[BBox | None, Body(embed=True)]	Input bounding box or None.	None
points	Annotated[list[list[int]] | None, Body(embed=True)]	Input points or None.	None
labels	Annotated[list[int] | None, Body(embed=True)]	Labels for input points, or None.	None
provider_name	Annotated[str | None, Body(embed=True)]	Optional provider name to route the request to.	None

Returns:

Type	Description
VideoMaskGenerationOutput	The generated masks.

Source code in pixano/app/routers/inference/mask_generation.py

@router.post(
    "/video",
    response_model=VideoMaskGenerationOutput,
)
async def call_video_mask_generation(
    dataset_id: Annotated[str, Body(embed=True)],
    video: Annotated[list[ViewModel], Body(embed=True)],
    model: Annotated[str, Body(embed=True)],
    settings: Annotated[Settings, Depends(get_settings)],
    bbox: Annotated[BBox | None, Body(embed=True)] = None,
    points: Annotated[list[list[int]] | None, Body(embed=True)] = None,
    labels: Annotated[list[int] | None, Body(embed=True)] = None,
    provider_name: Annotated[str | None, Body(embed=True)] = None,
) -> VideoMaskGenerationOutput:
    """Perform video mask generation on a video.

    Args:
        dataset_id: The ID of the dataset to use.
        video: The video as a list of SequenceFrame (or a subset).
        model: The name of the model to use.
        settings: App settings.
        bbox: Input bounding box or None.
        points: Input points or None.
        labels: Labels for input points, or None.
        provider_name: Optional provider name to route the request to.

    Returns:
        The generated masks.
    """
    dataset = get_dataset(dataset_id=dataset_id, dir=settings.library_dir, media_dir=settings.media_dir)
    provider = get_provider_from_settings(settings=settings, provider_name=provider_name)

    if not is_sequence_frame(dataset.schema.schemas[video[0].table_info.name]):
        raise HTTPException(status_code=400, detail="Video must be a list of SequenceFrame.")

    video_row: list[SequenceFrame] = [vm.to_row(dataset) for vm in video]
    source = get_model_source(dataset=dataset, model=model)

    try:
        mask_output: tuple[list[CompressedRLE], list[int], list[int]] = await video_mask_generation(
            provider=provider,
            source=source,
            media_dir=settings.media_dir,
            video=video_row,
            bbox=bbox,
            points=points,
            labels=labels,
        )
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e)) from e

    output: VideoMaskGenerationOutput = VideoMaskGenerationOutput(
        masks=[
            AnnotationModel.from_row(
                row=m,  # only use mask (don't send obj_ids & frame_idx => viewRef has the frame_index info)
                table_info=TableInfo(name="mask_table_name", group="annotations", base_schema="CompressedRLE"),
            )
            for m in mask_output[0]
        ]
    )
    return output