User Guide

In the following sections, we will describe how to create an algorithm module for Compox.

How to create an algorithm module

The algorithm module is a Python package that contains the algorithm code and assets. The algorithm module should be structured in a specific way in order to work properly with Compox.

The algorithm should be structured as follows:

algorithm_name/
    ├── __init__.py
    ├── Runner.py
    ├── pyproject.toml
    └── files/
        ├── file1
        └── file2
    └── some_internal_submodule/
        ├── __init__.py
        ├── module1.py
        └── module2.py

The Runner.py file

The Runner.py file is a mandatory component of the algorithm module. It serves as the entry point for Compox to run the algorithm. It must define a class named Runner. The Runner class should inherit either from the BaseRunner class, or a Runner class specific to the algorithm type (see more in the algorithm types section). The Runner classes can be imported from the compox.algorithm_utils module. There are several mandatory methods that the Runner class must implement in order to work properly. The Runner class can also implement additional methods and functions that are not mandatory. But these can also be included as a submodule in the algorithm directory.

Algorithm types

There are several types of algorithms that can be implemented in Compox. The main difference between the algorithm types is in the way input and output data is handled. The algorihtm type is defined in the algorithm’s pyproject.toml file and the class, from which the Runner class should inherit. An example of algorithm type is e.g. and Image2Image algorithm, which receives an image as input and returns an image as output. The pyproject.toml file should thus contain the following line:

[tool.compox]
algorithm_type = "Image2Image"

Furthermore the algotihm’s Runner class should inherit from the Image2ImageRunner class, which is imported from the compox.algorithm_utils module:

from compox.algorithm_utils.Image2ImageRunner import Image2ImageRunner

class Runner(Image2ImageRunner):
    """
    The runner class for the denoiser algorithm.
    """

The following algorithm types are currently supported:

Image2Image: The algorithm receives an image as input and returns an image as output.
Image2Embedding: The algorithm receives an image as input and returns a deep learning embedding as output.
Image2Segmentation: The algorithm receives an image as input and returns a segmentation mask as output.
Image2Alignment: The algorithm receives two images as input and returns an alignment transformation matrix as output.

If the algorithm does not fit into any of the above categories, the Runner class can either inherit from the BaseRunner class, and the data schemas can be defined manually (more on that in the next section), or a new algorithm type can be defined in the compox.algorithm_utils module.

Algorithm tags

The algorithm tags are a useful tool to categorize the algorithms for the users in the frontend application. Each algorithm type has a set of predefined tags, which are used to categorize the algorithms. This is important, because when several algorithms are tagged by a specific tag, the frontend developer can then operate with the assumption, that the algorithms with the same tag have the same input and output data schemas.

The `preprocess`, `inference` and `postprocess` methods

The methods are preprocess, inference and postprocess. In general it does not matter which part of the algorithm is implemented in which method, because the run method will call them sequentially in the order they are listed above. The separation is mostly for readability and maintainability.

Each of the methods must have exactly two inputs. The first input is some data that is passed between the methods (except for the preprocess method, which receives the input data identifiers defined by the user). The second input is a dictionary of arguments that the user has passed to the algorithm.

The preprocess method in general should serve for fetching the data using the fetch_data method, processing the data and returning the result. The preprocess method will receive the input_data and args as arguments. The first input is a dictionary currently containing only one key: input_dataset_ids. The value of the input_dataset_ids key is a list of dataset ids. The args is a dictionary containing the arguments that the user has passed to the algorithm. The output of the preprocess method does not have a set format, but it will be passed to the inference method, so it must be compatible with the inference method. The inference method should serve for running the actual algorithm on the data. The input of the inference method is the output of the preprocess method and the output is the result of the algorithm’s inference phase. The postprocess method should postprocess the result of the algorithm and pass it to the post_data method. The output of the postprocess method is a list strings, where each string is a dataset identifier of the posted data. This list will be returned to the user, who can then use these dataset identifiers to fetch the posted data from the database.

The `fetch_data` method for BaseRunner

Because the preprocess does not directly receive any data, only the data identifiers, the Runner class can use the fetch_data method to retrieve the data from the database. The fetch_data method receives a list of dataset identifiers and a pydantic model as arguments. The pydantic model is used to validate the fetched data. The output of the fetch_data method is a list of dictionaries, where each dictionary contains the data of one dataset.

Example of fetching data:

embeddings = self.fetch_data(input_data["input_dataset_ids"], EmbeddingSchema)

The EmbeddingSchema is a pydantic model that is used to validate the fetched data. The pydantic schemas are defined in the fastapi/app/algorithms/io_schemas.py file. The EmbeddingSchema is defined as:

class EmbeddingSchema(DataSchema):
    features: np.ndarray
    input_size: tuple
    original_size: tuple

The `fetch_data` method for specific algorithm types

The fetch_data for Runners inherited from specific algorithm types is a bit different. The fetch_data method does not accept a pydantic model as an argument, because the data schemas are predefined for each algorithm type. The fetch_data method receives a list of dataset identifiers as an argument. The output of the fetch_data method is a list of dictionaries, where each dictionary contains the data of one dataset.

Example of fetching data with a Runner inherited from the Image2ImageRunner class:

input_data = self.fetch_data(input_data["input_dataset_ids"])

The `post_data` method for BaseRunner

After the computation is done, the Runner class can use the post_data method of the class to post the result of the computation to the database. The post_data method receives the result of the computation and a pydantic model as arguments. The result is expected to be a list of dictionaries, where each dictionary contains the data of one dataset. The pydantic model is used to validate the result before posting it to the database. Output of the post_data method is a list of dataset identifiers of the posted data.

Example of posting data:

output_dataset_ids = self.post_data(output, MaskSchema)

The output is a list of dictionaries, where each dictionary contains the data of one dataset. The MaskSchema is a pydantic model that is used to validate the posted data. The MaskSchema is defined as:

class MaskSchema(DataSchema):
    mask: np.ndarray

The `post_data` method for specific algorithm types

The post_data method for Runners inherited from specific algorithm types follows the same pattern as the fetch_data method. The post_data method does not accept a pydantic model as an argument, because the data schemas are predefined for each algorithm type. The post_data method receives the result of the computation as an argument. The result is expected to be a list of dictionaries, where each dictionary contains the data of one dataset. The output of the post_data method is a list of dataset identifiers of the posted data.

Example of posting data with a Runner inherited from the Image2ImageRunner class:

output_dataset_ids = self.post_data(output)

The `load_assets` method

It is expected that the Runner class will work with a machine learning model. Because loading of model weights can be time consuming, the BaseRunner gives the developer an option to implement the load_assets method. The load_assets method is called during the instantiation of the Runner class in the Compox process and the attributes set in the load_assets will get cached together with the Runner instance. This will make repeated calls to the Runner class faster, because the model weights will not have to be loaded again as long as the cache is not invalidated.

Any file present in the algorithm directory can be loaded in the load_assets method (other than .py files). The load_assets should receive a relative path to the file that should be loaded as an argument. A bytes object will be returned, which can be loaded e.g. using the the torch.load method in the case of PyTorch state dicts.

Example of loading a PyTorch state dict:

state_dict = self.fetch_asset("files/vit_b.pt")
state_dict = torch.load(state_dict)

The `log_message` method

The log_message method can be used to log messages to Compox. The log_message method receives a message and a logging level as arguments. The logging level can be one of the following: “DEBUG”, “INFO”, “WARNING”, “ERROR”. The default logging level is “INFO”.

Example of logging a message:

self.log_message("This is an info message.", logging_level="INFO")

The `set_progress` method

The set_progress method can be used to report the progress of the algorithm to Compox. The set_progress method receives a float value between 0 and 1 as an argument. The starting progress is automatically set to 0 and if the computation is done, or fails, the progress is automatically set to 1.

Example of reporting the progress:

self.set_progress(0.5)

The `pyproject.toml` file

The pyproject.toml is a file that contains the algorithm metadata. This file is used by compox to properly deploy the algorithm as a service. The pyproject.toml file should be placed in the root directory of the algorithm.

Mandatory fields

The pyproject.toml file should contain the following fields mandatory fields:

[project]
name = "algorithm_name"
version = "major.minor.patch"

Even though the following fields are not mandatory, it is recommended to include them in the pyproject.toml file to make the algorithm as user-friendly and compatible with compox as possible.

The algorithm type should be specified in the tool.compox section. The algorithm type is used to specify the general algorithm functionality. The algorithm type is used to determine the input and output data schemas, and the general algorithm behavior. The algorithm type is defined in the compox.algorithm_utils module.

Algorithm type, tags, and description

[tool.compox]
algorithm_type = "AlgorithmType"

Each algorithm type has a set of potential tags, which are used to specify the general algorithm functionality. Multiple tags can be provided for one algorithm. For image denoising algorithms, we will use the image-denoising tag.

tags = ["tag1", "tag2", "tag3", ...]

The description field should contain a brief description of the algorithm.

description = "This is a super cool algorithm that does super cool things."

Algorithm supported devices

The server allows algorithms to be run on both CPU and GPU devices. The supported_devices field is used to specify which devices the algorithm supports. The supported_devices field should contain a list of strings, where each string specifies a device that the algorithm supports, [“cpu”], [“gpu”], or [“cpu”, “gpu”]. Additionally, a default_device field must be specified, which specifies the default device that the algorithm will run on. The default_device field should be a string, either “cpu” or “gpu”. Do not specify a default_device that’s not included in the supported_devices list as this will cause a warning and the algorithm will run on the CPU by default.

This setting will cause the algorithm to run on the CPU by default, but the user can override this setting in the execution request and run the algorithm on the GPU.

supported_devices = ["cpu", "gpu"]
default_device = "cpu"

Additional parameters

We can also specify additional parameters in the additional_parameters field. The additional_parameters field should contain a list of dictionaries, where each dictionary contains the parameter name, type, default value, and description. The additional_parameters field is optional, and can be omitted if the algorithm does not require any additional parameters.

Each additional parameter must contain the following fields:

name: The name of the parameter.
description: The description of the parameter. This field is used to describe the purpose of the parameter. It should be a short, human-readable description of the parameter that can be displayed by the client application as a tooltip or help text.
config: The configuration of parameter type, default, and adjustable. The type field specifies the type of the parameter. The default field specifies the default value of the parameter. The adjustable field specifies whether the parameter can be adjusted by the user. If set to true, the parameter can be adjusted by the user. If set to false, the parameter is to be specified from within the client application without exposing it to the user.

The following parameter types and configuration fields are supported:

Parameter type	Configuration fields
string	type, default, adjustable
int	type, default, adjustable
float	type, default, adjustable
bool	type, default, adjustable
int_range	type, default, min, max, step, adjustable
float_range	type, default, min, max, step, adjustable
string_enum	type, default, options, adjustable
int_enum	type, default, options, adjustable
float_enum	type, default, options, adjustable
string_list	type, default, adjustable
int_list	type, default, adjustable
float_list	type, default, adjustable
bool_list	type, default, adjustable

Here you can see an example of how to define additional parameters in the pyproject.toml file:

To define an user-adjustable string parameter, use the following configuration:

additional_parameters = [
    {name = "some_string_parameter", description = "This parameter strings.", config = {type = "string", default = "hello", adjustable = true}},
]

To define an user-adjustable int parameter, use the following configuration:

{name = "some_int_parameter", description = "This paramete ints.", config = {type = "int", default = 42, adjustable = true}},

To define an user-adjustable float parameter, use the following configuration:

{name = "some_float_parameter", description = "This parameter floats.", config = {type = "float", default = 3.14, adjustable = true}},

To define an user-adjustable bool parameter, use the following configuration:

{name = "some_bool_parameter", description = "This parameter bools.", config = {type = "bool", default = true, adjustable = true}},

To define an user-adjustable int_range parameter, use the following configuration:

{name = "some_int_range_parameter", description = "This parameter ranges ints.", config = {type = "int_range", default = 42, min = 0, max = 100, step = 1, adjustable = true}},

To define an user-adjustable float_range parameter, use the following configuration:

{name = "some_float_range_parameter", description = "This parameter ranges floats.", config = {type = "float_range", default = 3.14, min = 0.0, max = 10.0, step = 0.1, adjustable = true}},

To define an user-adjustable string_enum parameter, use the following configuration:

{name = "some_string_enum_parameter", description = "This parameter enums strings.", config = {type = "string_enum", default = "hello", options = ["hello", "world"], adjustable = true}},

To define an user-adjustable int_enum parameter, use the following configuration:

{name = "some_int_enum_parameter", description = "This parameter enums ints.", config = {type = "int_enum", default = 42, options = [42, 43, 44], adjustable = true}},

To define an user-adjustable float_enum parameter, use the following configuration:

{name = "some_float_enum_parameter", description = "This parameter enums floats.", config = {type = "float_enum", default = 3.14, options = [3.14, 3.15, 3.16], adjustable = true}},

To define an user-adjustable string_list parameter, use the following configuration:

{name = "some_string_list_parameter", description = "This parameter lists strings.", config = {type = "string_list", default = ["hello", "world"], adjustable = true}},

To define an user-adjustable int_list parameter, use the following configuration:

{name = "some_int_list_parameter", description = "This parameter lists ints.", config = {type = "int_list", default = [42, 43, 44], adjustable = true}},

To define an user-adjustable float_list parameter, use the following configuration:

{name = "some_float_list_parameter", description = "This parameter lists floats.", config = {type = "float_list", default = [3.14, 3.15, 3.16], adjustable = true}},

To define an user-adjustable bool_list parameter, use the following configuration:

{name = "some_bool_list_parameter", description = "This parameter lists bools.", config = {type = "bool_list", default = [true, false, true], adjustable = true}},

Other fields

The check_importable field is used to check if the algorithm can be imported. If set to true, compox will check if the algorithm can be imported before deploying it as a service.

check_importable = false

The obfuscate field is used to obfuscate the algorithm code. If set to true, compox will obfuscate the algorithm code before deploying it as a service. The obfuscation is currently implemented as minimization of the code. It is recommended to set this field to true to reasonably protect the algorithm code.

obfuscate = true

You can use the hash_module and hash_assets fields to check if the algorithm module or assets have already been deployed. If they have been deployed, compox will not redeploy them, but reuse them for the current algorithm deployment. This can reduce the deployment time and the amount of data that needs to be stored.

hash_module = true
hash_assets = true

The `files` directory

The files directory is an optional component of the algorithm module. It should contain any files that the algorithm needs to run. The files directory can contain any type of file, such as images, text files, etc. The files directory can be accessed by calling the self.fetch_asset method from anywhere in the Runner class. The fetch_asset method receives a relative path to the file that should be fetched as an argument. A bytes object will be returned, which can be used to load the file.

The `some_internal_submodule` directory

The some_internal_submodule directory is an optional component of the algorithm module. It should contain any internal submodules that the algorithm needs to run. The some_internal_submodule directory can contain any number of Python files. The some_internal_submodule directory can be accessed by importing the submodule from the Runner class. Note that the submodule will be deployed together with the algorithm, so it should not contain any sensitive information (even though the submodule is not accessible from the outside and there is an option to obfuscate the code).

Example of a dummy algorithm

The file structure should look like this (this is just an untested example):

algorithm_name/
    ├── __init__.py
    ├── Runner.py
    ├── pyproject.toml
    └── files/
        ├── some_heavy_model.pt
    └── my_big_model/
        ├── __init__.py
        ├── utils.py

The Runner.py file should look like this:

from my_big_model.utils import MyBigModel
from algorithms.BaseRunner import BaseRunner
from algorithms.io_schemas import MyDataSchema, MyOutputSchema
import numpy as np
import torch
class Runner(BaseRunner):
    """
    The runner class for the foo algorithm.
    """

    def load_assets(self):
        """
        The assets to load for the foo algorithm.
        """
        some_model = MyBigModel()
        self.log_message("Loading the Foo assets.")
        some_big_state_dict = self.fetch_asset("files/some_heavy_model.pt")¨
        some_big_state_dict = torch.load(some_big_state_dict)
        some_model.load_state_dict(some_big_state_dict)
        self.my_big_model = some_model


    def preprocess(self, input_data: dict, args: dict = {}) -> np.ndarray:
        """Preprocess the request data before feeding into model for inference.

        Parameters
        ----------
        input_data : ImageSchema
            The input data.
        args : dict, optional
            The arguments, by default None

        Returns
        -------
        tuple
            The preprocessed data.
        """
        self.log_message("Preprocessing the Foo input data.")
        my_data = self.fetch_data(input_data["input_dataset_ids"], MyDataSchema)
        input_data = np.array(my_data[0])
        return input_data

    def inference(self, data: np.ndarray,  args: dict = {}) -> torch.tensor:
        """Run the inference on the preprocessed data.

        Parameters
        ----------
        data : np.ndarray
            The preprocessed data.
        args : dict, optional
            The arguments, by default None


        Returns
        -------
        torch.tensor
            The inference output.
        """

        self.log_message("Running the Foo inference.")

        some_user_defined_args = args.get("some_user_defined_args", None)
        if some_user_defined_args is not None:
            self.log_message(f"User defined args: {some_user_defined_args}")
        output = self.my_big_model(input_data, some_user_defined_args)
        self.set_progress(0.5)
        self.log_message("The Foo inference is done.")
        return output

    def postprocess(self, inference_output: torch.tensor, args: dict = {}) -> list[str]:
        """Postprocess the inference output.

        Parameters
        ----------
        inference_output : dict
            The inference output.
        args : dict, optional
            The arguments, by default None

        Returns
        -------
        list[str]
            The output dataset ids.
        """

        self.log_message("Postprocessing the Foo output.")
        output = inference_output.detach().numpy()
        output = [
            {
                "output": output
            }
        ]
        output_dataset_ids = self.post_data(output, MyOutputSchema)
        return output_dataset_ids

The pyproject.toml file should look like this:

[project]
name = "foo"
version = "0.1.0"

[tool.compox]
algorithm_type = "Generic"
tags = ["foo", "bar"]
description = "This algorithm does foo and bar."
additional_parameters = [
    {name = "some_user_defined_args", type = "str", default = "hello", description = "This is a user defined argument."},
]
check_importable = false
obfuscate = true
hash_module = true
hash_assets = true

Denoising algorithm template

Here a working template for developing a denoising algorithm will be presented. This guide will cover the specifics needed to develop an image denoising algorithm. To see how compox algorithm should generally be structured, please refer to the algorithms/readme.md file.

The algorithm folder is structured as follows:

template_denoising_algorithm/
    ├── __init__.py
    ├── Runner.py
    ├── pyproject.toml
    └── image_denoising/
        ├── __init__.py
        └── denoising_utils.py
    └── README.md

The pyproject.toml file

The pyproject.toml is a file that contains the algorithm metadata. This file is used by compox to properly deploy the algorithm as a service. The pyproject.toml file should be placed in the root directory of the algorithm.

First, let’s create the pyproject.toml file. Under the [project] section, you should provide the name and version of the algorithm. The name should be unique and should not contain any spaces. The version should be in the format major.minor.patch. The algorithm name and versions is used to identify the algorithm in compox so it is important to provide a unique name and version.

[project]
name = "template_denosing_algorithm"
version = "1.0.0"

Next, you should fill out the [tool.compox] section. This section contains the metadata that compox uses to deploy the algorithm as a service. algorithm_type defines the algorithm input and output types, you may either use some predefined algorithm types or define your own. The predefined algorithm types are located in compox.algorithm_utils. For an image denoising algorithm, we will use the the Image2Image type. This type is suitable for image denoising as both our input and output is an image (or a sequence of images).

[tool.compox]
algorithm_type = "Image2Image"

Each algorithm type has a set of potential tags, which are used to specify the general algorithm functionality. Mutliple tags can be provided for one algorithm. For image denoising algorithms, we will use the image-denoising tag.

tags = ["image-denoising"]

The description field should contain a brief description of the algorithm.

description = "Denoises a sequence of images using the total variation denoising algorithm."

For the denoising algorithm, we will add a denoising_weight parameter that will control the denoising strength. Because we want to set a range for the denoising weight, we will use the float_range parameter type. The default field should contain the default value of the parameter. The min and max fields should contain the minimum and maximum values of the parameter. The step field should contain the step size of the parameter. The adjustable field should be set to true if the parameter should be exposed to the user to adjust.

additional_parameters = [
    {name = "denoising_weight", description = "The weight of the denoising term between 0 and 1. Higher values will result in more denoising, but can distort the image.", config = {type = "float_range", default = 0.1, min = 0.0, max = 1.0, step = 0.05, adjustable = true}}
]

To see more information about the possible parameter types see the How to create an algorithm module section.

The algorithm dependencies

The algorithm can use any libraries from the global compox environment. Additional dependencies can be provided as python submodules. Here we will use the numpy library to handle the image data. We also implemented a simple image_denoising module that contains an __init__.py file and a denosing_utils.py file. The denoising_utils.py file contains the denoise_image function that performs the denoising of the images. The image_denoising module should be placed in the root directory of the algorithm.

from skimage.restoration import (
    denoise_tv_chambolle,
)

def denoise_image(image, weight=0.1):
    """
    Denoise the image using the total variation denoising algorithm.

    Parameters
    ----------
    image : np.ndarray
        The image to denoise.
    weight: float
        The weight parameter for the denoising algorithm.
    Returns
    -------
    np.ndarray
        The denoised image.
    """

    return denoise_tv_chambolle(image, weight=weight)

The Runner.py file

The Runner.py file is the main file of the algorithm. This file should contain the algorithm implementation. The Runner.py file should be placed in the root directory of the algorithm.

Because we specified the algorithm type as Image2Image, the Runner.py file should contain a class that inherits from the Image2ImageRunner class. The Image2ImageRunner class is located in the compox.algorithm_utils module. The Image2ImageRunner class contains the necessary methods to handle the input and output of the algorithm.

from compox.algorithm_utils.Image2ImageRunner import Image2ImageRunner

class Runner(Image2ImageRunner):
    """
    The runner class for the denoiser algorithm.
    """

    def __init__(self, task_handler, device: str = "cpu"):
        """
        The denoising runner.
        """
        super().__init__(task_handler, device)

We can implement a load_assets method to load any assets that the algorithm requires upon initilaization of the Runner. The important bit is that the attributes that are loaded in the load_assets method are cached with the algorithm and do not have to be reloaded for each algorithm call. This can greatly speed up the algorithm execution. Since we do not need any assets for the denoising algorithm, we can leave the load_assets method empty.

def load_assets(self):
    """
    Here you can load the assets needed for the algorithm. This can be
    the model, the weights, etc. The assets are loaded upon the first
    call of the algorithm and are cached with the algorithm instance.
    """
    pass

Next, we can implement the inference method, where we perform the denoising of the images. The data will be passed to the inference method as a numpy array. The inference method should return a numpy array with the denoised images of the same shape as the input images. You can use the self.log_message method to log messages to the compox log. The self.set_progress method can be used to update the progress with a float value between 0 and 1.

def inference(self, data: np.ndarray, args: dict = {}) -> np.ndarray:
    """
    Run the inference.

    Parameters
    ----------
    input_data : dict
        The input data.

    Returns
    -------
    np.ndarray
        The denoised images.
    """
    self.log_message("Starting inference.")
    # now we retrieve the input data
    # we will min max normalize the images
    min_val = np.min(data)
    max_val = np.max(data)
    images = (data - min_val) / (max_val - min_val)

    # here we will get the optional argument of denoising weight
    denosing_weight = args.get("denoising_weight", 0.1)

    # we can post messages to the log
    self.log_message(
        f"Starting denoising of {images.shape[0]} images with weight {denosing_weight}."
    )

    # we will denoise the images
    denoised_images = np.zeros_like(images)
    for i in range(images.shape[0]):
        denoised_images[i] = denoise_image(
            images[i], weight=denosing_weight
        )
        # this will update the progress bar
        self.set_progress(i / images.shape[0])

    # we will nromalize the output
    denoised_images = (denoised_images - denoised_images.min()) / (
        denoised_images.max() - denoised_images.min()
    )
    denoised_images = denoised_images.astype(np.float32)

    # we will pass the denoised images to the postprocess method
    return denoised_images

To customize the behavior of fetching and processing the input data, and postprocessing and uploading the output data, we can implement the preprocess and postprocess methods. The preprocess method is called before the inference method and is used to fetch the input data. The postprocess method is called after the inference method and is used to process the output data. In our case, we will not implement any custom behavior for these methods. You can refer to the compoxorithm_utils.Image2ImageRunner class for more information about these methods.

Deploying the algorithm

To deploy the finished algorithm, you can use the pdm run deployment template_denoising_algorithm command. This command will deploy the algorithm to the compox. The algorithm can also be added through compox systray interface by clicking the “Add Algorithm” button and selecting the algorithm directory.

Segmentation algorithm template

This guide will cover the specifics needed to develop an image segmentation algorithm. To see how compox algorithm should generally be structured, please refer to the algorithms/readme.md file.

The algorithm folder is structured as follows:

template_segmentation_algorithm/
    ├── __init__.py
    ├── Runner.py
    ├── pyproject.toml
    └── image_segmentation/
        ├── __init__.py
        └── segmentation_utils.py
    └── README.md

The pyproject.toml file

The pyproject.toml is a file that contains the algorithm metadata. This file is used by compox to properly deploy the algorithm as a service. The pyproject.toml file should be placed in the root directory of the algorithm.

First, let’s create the pyproject.toml file. Under the [project] section, you should provide the name and version of the algorithm. The name should be unique and should not contain any spaces. The version should be in the format major.minor.patch. The algorithm name and versions is used to identify the algorithm in compox so it is important to provide a unique name and version.

[project]
name = "template_segmentation_algorithm"
version = "1.0.0"

Next, we will fill out the [tool.compox] section. This section contains the metadata that compox uses to deploy the algorithm as a service. algorithm_type defines the algorithm input and output types, you may either use some predefined algorithm types or define your own. The predefined algorithm types are located in compox.algorithm_utils. For an image segmentation algorithm, we will use the the Image2Segmentation type. This type is suitable for image segmentation as the input is a sequence of images and the output is a sequence of segmentation masks.

[tool.compox]
algorithm_type = "Image2Segmentation"

Each algorithm type has a set of potential tags, which are used to specify the general algorithm functionality. Mutliple tags can be provided for one algorithm. For image segmentation algorithms, we will use the image-segmentation tag.

tags = ["image-segmenation"]

The description field should contain a brief description of the algorithm.

description = "Performs a binary segmentation of a 3-D image using a skimage filter."

Here we will add a thresholding_algorithm parameter that will allow the user to select the thresholding algorithm to use. The type field is set to string_enum to specify that the parameter is a string with a predefined set of values. The default field is set to otsu to specify the default value of the parameter. The options field is set to a list of strings that specify the possible values of the parameter. The adjustable field is set to true to specify that the user should be able to select the thresholding algorithm to apply.

additional_parameters = [
    {name = "thresholding_algorithm", description = "The thresholding algorithm to use.", config = {type = "string_enum", default = "otsu", options = ["otsu", "yen", "li", "minimum", "mean", "triangle", "isodata", "local"], adjustable = true}},
]

To see more information about the possible parameter types see the How to create an algorithm module section.

The algorithm dependencies

The algorithm can use any libraries from the global compox environment. Additional dependencies can be provided as python submodules. Here we will use the numpy library to handle the image data. We also implemented a simple image_segmentation module that contains an __init__.py file and a segmentation_utils.py file. The segmentation_utils.py file contains the threshold_image function that performs segmentation of an image using a selected algorithm. The image_segmentation module should be placed in the root directory of the algorithm.

import skimage.filters as skif


def threshold_image(image, thresholding_algorithm):
    """
    Threshold the image using the specified thresholding algorithm.

    Parameters
    ----------
    image : np.ndarray
        The image to threshold.
    thresholding_algorithm : str
        The thresholding algorithm to use.

    Returns
    -------
    np.ndarray
        The thresholded image.
    """
    if thresholding_algorithm == "otsu":
        threshold = skif.threshold_otsu(image)
    elif thresholding_algorithm == "yen":
        threshold = skif.threshold_yen(image)
    elif thresholding_algorithm == "li":
        threshold = skif.threshold_li(image)
    elif thresholding_algorithm == "minimum":
        threshold = skif.threshold_minimum(image)
    elif thresholding_algorithm == "mean":
        threshold = skif.threshold_mean(image)
    elif thresholding_algorithm == "triangle":
        threshold = skif.threshold_triangle(image)
    elif thresholding_algorithm == "isodata":
        threshold = skif.threshold_isodata(image)
    elif thresholding_algorithm == "local":
        threshold = skif.threshold_local(image)
    else:
        raise ValueError(
            f"Invalid thresholding algorithm: {thresholding_algorithm}"
        )

    return image > threshold

The Runner.py file

The Runner.py file is the main file of the algorithm. This file should contain the algorithm implementation. The Runner.py file should be placed in the root directory of the algorithm.

Because we specified the algorithm type as Image2Segmentation, the Runner.py file should contain a class that inherits from the Image2SegmentationRunner class. The Image2SegmentationRunner class is located in the compox.algorithm_utils module. The Image2SegmentationRunner class contains the necessary methods to handle the input and output of the algorithm.

import numpy as np
from compox.algorithm_utils.Image2SegmentationRunner import (
    Image2SegmentationRunner,
)
from image_segmentation.segmentation_utils import threshold_image


class Runner(Image2SegmentationRunner):
    """
    The runner class for the image segmentation algorithm.
    """

    def __init__(self, task_handler, device: str = "cpu") -> None:
        """
        The aligner runner.
        """
        super().__init__(task_handler, device=device)

We can implement a load_assets method to load any assets that the algorithm requires upon initilaization of the Runner. The important bit is that the attributes that are loaded in the load_assets method are cached with the algorithm and do not have to be reloaded for each algorithm call. This can greatly speed up the algorithm execution. Since we do not need any assets for the segmentation algorithm, we can leave the load_assets method empty.

def load_assets(self):
    """
    Here you can load the assets needed for the algorithm. This can be
    the model, the weights, etc. The assets are loaded upon the first
    call of the algorithm and are cached with the algorithm instance.
    """
    pass

Next, we can implement the inference method, where we perform the segmentation of the images. The inference will receive a numpy array with the images to be segmented. The inference method must return a numpy array with the segmentation masks of the same shape as the input images. The inference method can also receive a dictionary with the arguments for the algorithm. The arguments are passed to the algorithm from compox and can be used to customize the behavior of the algorithm. In our case, we will use the thresholding_algorithm argument to specify the thresholding algorithm to use. You can also report the progress of the algorithm by calling the set_progress method. The set_progress method takes a float value between 0 and 1, where 0 is the start of the algorithm and 1 is the end of the algorithm. The log_message method can be used to log messages to compox log.

def inference(self, data: np.ndarray, args: dict = {}) -> np.ndarray:
    """
    Run the inference.

    Parameters
    ----------
    data : np.ndarray
        The images to be segmented.
    args : dict
        The arguments for the algorithm.

    Returns
    -------
    np.ndarray
        The segmented images.
    """

    # now we retrieve the input data
    thresholding_algorithm = args.get("thresholding_algorithm", "otsu")
    # we can post messages to the log
    self.log_message(
        f"Starting inference with thresholding algorithm: {thresholding_algorithm}"
    )

    # here we will threshold the images
    mask = threshold_image(data, thresholding_algorithm)

    # we can also log progress
    self.set_progress(0.5)

    # pass the mask to the postprocess
    return mask

To customize the behavior of fetching and processing the input data, and postprocessing and uploading the output data, we can implement the preprocess and postprocess methods. The preprocess method is called before the inference method and is used to fetch the input data. The postprocess method is called after the inference method and is used to process the output data. In our case, we will not implement any custom behavior for these methods. You can refer to the compox.algorithm_utils.Image2SegmentationRunner class for more information about these methods.

Deploying the algorithm

To deploy the finished algorithm, you can use the pdm run deployment template_segmentation_algorithm command. This command will deploy the algorithm to the compox. The algorithm can also be added through compox systray interface by clicking the “Add Algorithm” button and selecting the algorithm directory.

Registration algorithm template

Here a working template for developing an image registration algorithm will be presented. To see how compox algorithm should generally be structured, please refer to the algorithms/readme.md file.

The algorithm folder is structured as follows:

template_registration_algorithm/
    ├── __init__.py
    ├── Runner.py
    ├── pyproject.toml
    └── image_registration/
        ├── __init__.py
        └── registration_utils.py
    └── README.md

The pyproject.toml file

The pyproject.toml is a file that contains the algorithm metadata. This file is used by compox to properly deploy the algorithm as a service. The pyproject.toml file should be placed in the root directory of the algorithm.

First, let’s create the pyproject.toml file. Under the [project] section, you should provide the name and version of the algorithm. The name should be unique and should not contain any spaces. The version should be in the format major.minor.patch. The algorithm name and versions is used to identify the algorithm in compox so it is important to provide a unique name and version.

[project]
name = "template_registration_algorithm"
version = "1.0.0"

Next, we will fill out the [tool.compox] section. This section contains the metadata that compox uses to deploy the algorithm as a service. algorithm_type defines the algorithm input and output types, you may either use some predefined algorithm types or define your own. The predefined algorithm types are located in compox.algorithm_utils. For an image registration algorithm, we will use the the Image2Alignment type. This type is suitable for image segmentation as the input is a sequence of images and the output is a sequence homography matrices.

[tool.compox]
algorithm_type = "Image2Alignment"

Each algorithm type has a set of potential tags, which are used to specify the general algorithm functionality. Mutliple tags can be provided for one algorithm. For image registration algorithms, we will use the image-alignment tag.

tags = ["image-alignment"]

The description field should contain a brief description of the algorithm.

description = "Generates homography matrices for aligning a sequence of images."

Here we will add a max_translation parameter that defines the maximum translation as a fraction of the image size. Because we want to set a range for the parameter, we will use the float_range type. The default field should contain the default value of the parameter. The min and max fields should contain the minimum and maximum values of the parameter. The step field should contain the step size of the parameter. The adjustable field should be set to true if we want to expose the parameter to the user to adjust.

 {name = "max_translation", description = "Maximum translation as a fraction of the image size.", config = {type = "float_range", default = 0.25, min = 0.0, max = 1.0, step = 0.05, adjustable = true}}

To see more information about the possible parameter types see the How to create an algorithm module section.

The algorithm dependencies

The algorithm can use any libraries from the global compox environment. Additional dependencies can be provided as python submodules. Here we will use the numpy library to handle the image data. We also implemented a simple image_registration module that contains an __init__.py file and a registration_utils.py file. The registration_utils.py file contains the get_random_translation function that generates a random homography matrix with a maximum translation defined by the max_translation parameters as a fraction of the input image size. The image_registration module should be placed in the root directory of the algorithm.

import numpy as np

def get_random_translation(image: np.ndarray, max_translation: float = 0.25):
    """
    Get a random translation matrix.

    Parameters
    ----------
    image : np.ndarray
        The image.
    max_translation : float
        The maximum translation.

    Returns
    -------
    np.ndarray
        The translation matrix.
    """

    # get the image dimensions
    height, width = image.shape[:2]
    h = np.eye(3)

    # random translation
    h[0, 2] = np.random.uniform(
        -max_translation * width, max_translation * width
    )
    h[1, 2] = np.random.uniform(
        -max_translation * height, max_translation * height
    )

    return h

The Runner.py file

The Runner.py file is the main file of the algorithm. This file should contain the algorithm implementation. The Runner.py file should be placed in the root directory of the algorithm.

Because we specified the algorithm type as Image2Alignment, the Runner.py file should contain a class that inherits from the Image2AlignmentRunner class. The Image2AlignmentRunner class is located in the compox.algorithm_utils module. The Image2AlignmentRunner class contains the necessary methods to handle the input and output of the algorithm.

import numpy as np

from compox.algorithm_utils.Image2AlignmentRunner import (
    Image2AlignmentRunner,
)
from image_registration.registration_utils import get_random_translation

class Runner(Image2AlignmentRunner):
    """
    The runner class for the denoiser algorithm.
    """

    def __init__(self, task_handler, device: str = "cpu"):
        """
        The image registration runner.
        """
        super().__init__(task_handler, device)

We can implement a load_assets method to load any assets that the algorithm requires upon initilaization of the Runner. The important bit is that the attributes that are loaded in the load_assets method are cached with the algorithm and do not have to be reloaded for each algorithm call. This can greatly speed up the algorithm execution. Since we do not need any assets for the image registration algorithm, we can leave the load_assets method empty.

def load_assets(self):
    """
    Here you can load the assets needed for the algorithm. This can be
    the model, the weights, etc. The assets are loaded upon the first
    call of the algorithm and are cached with the algorithm instance.
    """
    pass

Next, we can implement the inference method, where we perform the registration of the images. The data will be passed to the inference method as a numpy array. The inference method return a list of homography matrices represented by numpy arrays. You can also report the progress of the algorithm by calling the set_progress method. The set_progress method takes a float value between 0 and 1, where 0 is the start of the algorithm and 1 is the end of the algorithm. The log_message method can be used to log messages to the compox log.

def inference(self, data: np.ndarray, args: dict = {}) -> list[np.ndarray]:
    """
    Run the inference.

    Parameters
    ----------
    data : np.ndarray
        The input images

    Returns
    -------
    list[np.ndarray]
        The output homography matrices.
    """
    self.log_message("Starting inference.")
    # now we retrieve the input data
    max_translation = args.get("max_translation", 0.25)
    # we can post messages to the log
    self.log_message(f"Registering {data.shape[0]} images.")

    # we will denoise the images
    matrices = []
    for i in range(data.shape[0] - 1):
        matrix = get_random_translation(
            data[i], max_translation=max_translation
        )
        matrices.append(matrix)
        self.set_progress(i / data.shape[0])
    # we will pass the homography matrices to the output
    return matrices

To customize the behavior of fetching and processing the input data, and postprocessing and uploading the output data, we can implement the preprocess and postprocess methods. The preprocess method is called before the inference method and is used to fetch the input data. The postprocess method is called after the inference method and is used to process the output data. In our case, we will not implement any custom behavior for these methods. You can refer to the compox.algorithm_utils.Image2AlignmentRunner class for more information about these methods.

Deploying the algorithm

To deploy the finished algorithm, you can use the pdm run deployment template_registration_algorithm command. This command will deploy the algorithm to the compox. The algorithm can also be added through compox systray interface by clicking the “Add Algorithm” button and selecting the algorithm directory.

User Guide

How to create an algorithm module

The Runner.py file

Algorithm types

Algorithm tags

The preprocess, inference and postprocess methods

The fetch_data method for BaseRunner

The fetch_data method for specific algorithm types

The post_data method for BaseRunner

The post_data method for specific algorithm types

The load_assets method

The log_message method

The set_progress method

The pyproject.toml file

Mandatory fields

Algorithm type, tags, and description

Algorithm supported devices

Additional parameters

Other fields

The files directory

The some_internal_submodule directory

Example of a dummy algorithm

Denoising algorithm template

The pyproject.toml file

The algorithm dependencies

The Runner.py file

Deploying the algorithm

Segmentation algorithm template

The pyproject.toml file

The algorithm dependencies

The Runner.py file

Deploying the algorithm

Registration algorithm template

The pyproject.toml file

The algorithm dependencies

The Runner.py file

Deploying the algorithm

The `preprocess`, `inference` and `postprocess` methods

The `fetch_data` method for BaseRunner

The `fetch_data` method for specific algorithm types

The `post_data` method for BaseRunner

The `post_data` method for specific algorithm types

The `load_assets` method

The `log_message` method

The `set_progress` method

The `pyproject.toml` file

The `files` directory

The `some_internal_submodule` directory