Skip to content

Open In Colab

How to create a plugin for programmatic QA

Context

This notebook is an end-to-end example that you can follow to: create a project, upload a first plugin and activate it on this project, and finally start monitoring it.

NB: The plugin capabilities of Kili are under active development, and compatible with version 2.125.2 and later of Kili. Don't hesitate to reach out via Github or the Kili support to provide feedback.

Step 1: Instantiate Kili

%pip install kili
from kili.client import Kili

kili = Kili(
    # api_endpoint="https://cloud.kili-technology.com/api/label/v2/graphql",
    # the line above can be uncommented and changed if you are working with an on-premise version of Kili
)

Step 2: Create the project

First, we need to create a new project. In our example, we will use an IMAGE type project with the following jsonInterace:

json_interface = {
    "jobs": {
        "JOB_0": {
            "content": {
                "categories": {
                    "OBJECT_A": {
                        "children": [],
                        "name": "Object A",
                        "color": "#733AFB",
                        "id": "category1",
                    },
                    "OBJECT_B": {
                        "children": [],
                        "name": "Object B",
                        "color": "#3CD876",
                        "id": "category2",
                    },
                },
                "input": "radio",
            },
            "instruction": "Categories",
            "isChild": False,
            "tools": ["rectangle"],
            "mlTask": "OBJECT_DETECTION",
            "models": {},
            "isVisible": True,
            "required": 1,
            "isNew": False,
        }
    }
}
title = "[Kili SDK Notebook]: Plugins test project"
description = "My first project with a plugin"
input_type = "IMAGE"

project = kili.create_project(
    title=title, description=description, input_type=input_type, json_interface=json_interface
)
project_id = project["id"]

print(f"Created project {project_id}")

Upload an asset:

content_array = ["https://storage.googleapis.com/label-public-staging/car/car_1.jpg"]
names_array = ["landscape"]

kili.append_many_to_dataset(
    project_id=project_id, content_array=content_array, external_id_array=names_array
)

asset_id = kili.assets(project_id=project_id, fields=["id"], disable_tqdm=True)[0]["id"]

This project has one job of bounding box creation with two categories.

With our plugin, we want to make sure that the labelers don't create more than one bounding box of category A.

To iterate on the plugin code, you can refer to the plugins_development.ipynb notebook.

Step 3: Write the plugin

from kili.plugins import PluginCore
from typing import Dict, List, Optional

def check_rules_on_label(label: Dict) -> List[Optional[str]]:
    #custom methods
    print('Custom method - checking number of bboxes')

    counter = 0
    for annotation in label['jsonResponse']["JOB_0"]["annotations"]:
        if annotation["categories"][0]["name"] == "OBJECT_A":
            counter += 1

    if counter <= 1:
        return []
    return [f"There are too many BBox ({counter}) - Only 1 BBox of Object A accepted"]


class PluginHandler(PluginCore):
    """
    Custom plugin instance
    """

    def on_submit(self, label: Dict, asset_id: str) -> None:
        """
        Dedicated handler for Submit action
        """
        self.logger.info("On submit called")

        issues_array = check_rules_on_label(label)

        project_id = self.project_id

        if len(issues_array) > 0:
            print("Creating an issue...")

            self.kili.create_issues(
                project_id=project_id,
                label_id_array=[label['id']] * len(issues_array),
                text_array=issues_array,
            )

            print("Issue created!")

            self.kili.send_back_to_queue(asset_ids=[asset_id])
import urllib.request
from pathlib import Path

plugin_folder = "plugin_folder"

Path(plugin_folder).mkdir(parents=True, exist_ok=True)
urllib.request.urlretrieve(
    "https://raw.githubusercontent.com/kili-technology/kili-python-sdk/main/recipes/plugins_library/plugin_image.py",
    "plugin_folder/main.py",
)

Step 4: Upload the plugin from a folder

With the plugin defined in a separate Python file, you can create a folder containing:

  • A main.py file which is the entrypoint of the plugin and must have a PluginHandler class which implements a PluginCore class
  • (optionally) a requirements.txt (if you need specific PyPi packages in your plugin)
folder/
     main.py
     requirements.txt
  • The upload will create the necessary builds to execute the plugin (it will take a few minutes)
  • After the activation, you can start using your plugin right away.

Here is an example of a requirements.txt file: numpy scikit-learn pandas==1.5.1 git+https://github.com/yzhao062/pyod.git

requirements_path = Path(plugin_folder) / "requirements.txt"

packages_list = [
    "numpy\n",
    "scikit-learn\n",
    "pandas==1.5.1\n",
    "git+https://github.com/yzhao062/pyod.git\n",
]

with requirements_path.open("w") as f:
    f.writelines(packages_list)
plugin_name = "Plugin bbox count"
from kili.exceptions import GraphQLError

try:
    kili.upload_plugin(plugin_folder, plugin_name)
except GraphQLError as error:
    print(str(error))
kili.activate_plugin_on_project(plugin_name=plugin_name, project_id=project_id)

Step 4 bis: Upload the plugin from a .py file

Alternatively, you can also create a plugin directly from a .py file.

  • The upload will create the necessary builds to execute the plugin (it will take a few minutes)
  • After the activation, you can start using your plugin right away.
path_to_plugin = Path(plugin_folder) / "main.py"
plugin_name_file = "Plugin bbox count - file"

try:
    kili.upload_plugin(str(path_to_plugin), plugin_name_file)
except GraphQLError as error:
    print(str(error))
kili.activate_plugin_on_project(plugin_name=plugin_name_file, project_id=project_id)

Step 5: Plugin in action

Wait for the plugin to be successfully deployed.

After that, you can test it by labelling in the Kili interface or just by uploading the following label.

When you add the label that contains errors, you will see a new issue automatically created in the Kili app.

json_response = {
    "JOB_0": {
        "annotations": [
            {
                "boundingPoly": [
                    {
                        "normalizedVertices": [
                            {"x": 0.15, "y": 0.84},
                            {"x": 0.15, "y": 0.31},
                            {"x": 0.82, "y": 0.31},
                            {"x": 0.82, "y": 0.84},
                        ]
                    }
                ],
                "categories": [{"name": "OBJECT_A"}],
                "children": {},
                "mid": "20221124161451411-13314",
                "type": "rectangle",
            },
            {
                "boundingPoly": [
                    {
                        "normalizedVertices": [
                            {"x": 0.79, "y": 0.20},
                            {"x": 0.79, "y": 0.13},
                            {"x": 0.91, "y": 0.13},
                            {"x": 0.91, "y": 0.20},
                        ]
                    }
                ],
                "categories": [{"name": "OBJECT_A"}],
                "children": {},
                "mid": "20221124161456406-47055",
                "type": "rectangle",
            },
            {
                "boundingPoly": [
                    {
                        "normalizedVertices": [
                            {"x": 0.87, "y": 0.36},
                            {"x": 0.87, "y": 0.27},
                            {"x": 0.99, "y": 0.27},
                            {"x": 0.99, "y": 0.36},
                        ]
                    }
                ],
                "categories": [{"name": "OBJECT_A"}],
                "children": {},
                "mid": "20221124161459298-45160",
                "type": "rectangle",
            },
        ]
    }
}
kili.append_labels(
    json_response_array=[json_response], asset_id_array=[asset_id], label_type="DEFAULT"
)

If you use the base plugin provided, the plugin should:

  • Create an issue with information that three bboxes were found, instead of one
  • Send the asset back to the labeling queue (status ONGOING)
print(
    kili.assets(project_id=project_id, asset_id=asset_id, fields=["status", "issues.comments.text"])
)

print(
    f"Go to my project: {kili.api_endpoint.split('/api')[0]}/label/projects/{project_id}/menu/queue"
)

Woah! Amazing! Well done :) 🚀

Let's test now to post a proper label, this one for example:

json_response = {
    "JOB_0": {
        "annotations": [
            {
                "boundingPoly": [
                    {
                        "normalizedVertices": [
                            {"x": 0.15, "y": 0.84},
                            {"x": 0.15, "y": 0.31},
                            {"x": 0.82, "y": 0.31},
                            {"x": 0.82, "y": 0.84},
                        ]
                    }
                ],
                "categories": [{"name": "OBJECT_A"}],
                "children": {},
                "mid": "20221124161451411-13314",
                "type": "rectangle",
            }
        ]
    }
}
kili.append_labels(
    json_response_array=[json_response], asset_id_array=[asset_id], label_type="DEFAULT"
)

print(kili.assets(project_id=project_id, asset_id=asset_id, fields=["status"]))

print(
    f"Go to my project: {kili.api_endpoint.split('/api')[0]}/label/projects/{project_id}/menu/queue"
)

The status of your asset should have now changed to LABELED. In this plugin, previous issues remain but you can solve them through the API as well.

Well done! You can now iterate on the script. To learn how to avoid latency when building and deploying your plugin, refer to the plugins development tutorial.

Step 6: Monitor the plugin

To monitor a certain plugin, you can get its logs by using the following command:

import json
from datetime import date, datetime

dt = (
    date.today()
)  # You can change this date if needed, or omit it to set it at the plugin creation date
start_date = datetime.combine(dt, datetime.min.time())

logs = kili.get_plugin_logs(project_id=project_id, plugin_name=plugin_name, start_date=start_date)

logs_json = json.loads(logs)
print(json.dumps(logs_json, indent=4))

Step 7: Manage the plugin

You also have several other methods to manage your plugins.

Get the list of all uploaded plugins in your organization:

plugins = kili.list_plugins()

Update a plugin with new source code:

# Insert the path to the updated plugin
new_path_to_plugin = Path(plugin_folder) / "main.py"

# Change to True if you want to update the plugin
should_update = False

if should_update:
    kili.update_plugin(plugin_name=plugin_name, plugin_path=str(new_path_to_plugin))

Deactivate the plugin on a certain project (the plugin can still be active for other projects):

kili.deactivate_plugin_on_project(plugin_name=plugin_name, project_id=project_id);

Delete the plugin completely (deactivates the plugin from all projects):

if delete_plugin_from_org:
    kili.delete_plugin(plugin_name=plugin_name)