Kili plugin example: 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
import os
import time
from pathlib import Path
kili = 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 = "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 = list(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
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/master/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.pyfile which is the entrypoint of the plugin and must have aPluginHandlerclass which implements aPluginCoreclass - (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(
"Go to my project:"
f" {kili.auth.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(
"Go to my project:"
f" {kili.auth.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
from datetime import 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)