diff --git a/README.md b/README.md index ef265ad..6807c19 100644 --- a/README.md +++ b/README.md @@ -2,10 +2,25 @@ _If you are using FastLabel prototype, please install version 0.2.2._ +## Table of Contents + +- [Installation](#installation) +- [Usage](#usage) + - [Limitation](#limitation) +- [Task](#task) + - [Image](#image) + - [Image Classification](#image-classification) + - [Multi Image](#multi-image) + - [Video](#video) + - [Common](#common) +- [Annotation](#annotation) +- [Converter](#converter) + - [COCO](#coco) + ## Installation ```bash -$ pip install --upgrade fastlabel +pip install --upgrade fastlabel ``` > Python version 3.7 or greater is required @@ -25,7 +40,7 @@ import fastlabel client = fastlabel.Client() ``` -## Limitation +### Limitation API is allowed to call 10000 times per 10 minutes. If you create/delete a large size of tasks, please wait a second for every requests. @@ -90,6 +105,12 @@ task_id = client.create_image_task( task = client.find_image_task(task_id="YOUR_TASK_ID") ``` +- Find a single task by name. + +```python +tasks = client.find_image_task_by_name(project="YOUR_PROJECT_SLUG", task_name="YOUR_TASK_NAME") +``` + #### Get Tasks - Get tasks. (Up to 1000 tasks) @@ -466,6 +487,180 @@ task_id = client.update_task( client.delete_task(task_id="YOUR_TASK_ID") ``` +#### Get Tasks Id and Name map + +```python +map = client.get_task_id_name_map(project="YOUR_PROJECT_SLUG") +``` + +## Annotation + +### Create Annotaion + +- Create a new annotation. + +```python +annotation_id = client.create_annotation( + project="YOUR_PROJECT_SLUG", type="bbox", value="cat", title="Cat", color="#FF0000") +``` + +- Create a new annotation with attributes. + +```python +attributes = [ + { + "type": "text", + "name": "Kind", + "key": "kind" + }, + { + "type": "select", + "name": "Size", + "key": "size", + "options": [ # select, radio and checkbox type requires options + { + "title": "Large", + "value": "large" + }, + { + "title": "Small", + "value": "small" + }, + ] + }, +] +annotation_id = client.create_annotation( + project="YOUR_PROJECT_SLUG", type="bbox", value="cat", title="Cat", color="#FF0000", attributes=attributes) +``` + +- Create a new classification annotation. + +```python +annotation_id = client.create_classification_annotation( + project="YOUR_PROJECT_SLUG", attributes=attributes) +``` + +### Find Annotation + +- Find an annotation. + +```python +annotaion = client.find_annotation(annotation_id="YOUR_ANNOTATIPN_ID") +``` + +- Find an annotation by value. + +```python +annotaion = client.find_annotation_by_value(project="YOUR_PROJECT_SLUG", value="cat") +``` + +- Find an annotation by value in classification project. + +```python +annotaion = client.find_annotation_by_value( + project="YOUR_PROJECT_SLUG", value="classification") # "classification" is fixed value +``` + +### Get Annotations + +- Get annotations. (Up to 1000 annotations) + +```python +annotatios = client.get_annotations(project="YOUR_PROJECT_SLUG") +``` + +### Response + +- Example of an annotation object + +```python +{ + "id": "YOUR_ANNOTATION_ID", + "type": "bbox", + "value": "cat", + "title": "Cat", + "color": "#FF0000", + "attributes": [ + { + "id": "YOUR_ATTRIBUTE_ID", + "key": "kind", + "name": "Kind", + "options": [], + "type": "text", + "value": "" + }, + { + "id": "YOUR_ATTRIBUTE_ID", + "key": "size", + "name": "Size", + "options": [ + {"title": "Large", "value": "large"}, + {"title": "Small", "value": "small"} + ], + "type": "select", + "value": "" + } + ], + "createdAt": "2021-05-25T05:36:50.459Z", + "updatedAt": "2021-05-25T05:36:50.459Z" +} +``` + +### Update Annotation + +- Update an annotation. + +```python +annotation_id = client.update_annotation( + annotation_id="YOUR_ANNOTATION_ID", value="cat2", title="Cat2", color="#FF0000") +``` + +- Update an annotation with attributes. + +```python +attributes = [ + { + "id": "YOUR_ATTRIBUTE_ID", # check by sdk get methods + "type": "text", + "name": "Kind2", + "key": "kind2" + }, + { + "id": "YOUR_ATTRIBUTE_ID", + "type": "select", + "name": "Size2", + "key": "size2", + "options": [ + { + "title": "Large2", + "value": "large2" + }, + { + "title": "Small2", + "value": "small2" + }, + ] + }, +] +annotation_id = client.update_annotation( + annotation_id="YOUR_ANNOTATION_ID", value="cat2", title="Cat2", color="#FF0000", attributes=attributes) +``` + +- Update a classification annotation. + +```python +annotation_id = client.update_classification_annotation( + project="YOUR_PROJECT_SLUG", attributes=attributes) +``` + +### Delete Annotation + +- Delete an annotation. + +```python +client.delete_annotation(annotation_id="YOUR_ANNOTATIPN_ID") +``` + ## Converter ### COCO diff --git a/fastlabel/__init__.py b/fastlabel/__init__.py index d237a23..f3f80bd 100644 --- a/fastlabel/__init__.py +++ b/fastlabel/__init__.py @@ -1,165 +1,110 @@ import os import glob -from enum import Enum from logging import getLogger -from concurrent.futures import ThreadPoolExecutor -import requests -import base64 -import numpy as np -import geojson +from .exceptions import FastLabelInvalidException +from .api import Api +from fastlabel import converters, utils logger = getLogger(__name__) -FASTLABEL_ENDPOINT = "https://api.fastlabel.ai/v1/" - class Client: - access_token = None + api = None - def __init__(self) -> None: - if not os.environ.get("FASTLABEL_ACCESS_TOKEN"): - raise ValueError("FASTLABEL_ACCESS_TOKEN is not configured.") - self.access_token = "Bearer " + \ - os.environ.get("FASTLABEL_ACCESS_TOKEN") + def __init__(self): + self.api = Api() - def __getrequest(self, endpoint: str, params=None) -> dict: - """Makes a get request to an endpoint. - If an error occurs, assumes that endpoint returns JSON as: - { 'statusCode': XXX, - 'error': 'I failed' } - """ - params = params or {} - headers = { - "Content-Type": "application/json", - "Authorization": self.access_token, - } - r = requests.get(FASTLABEL_ENDPOINT + endpoint, - headers=headers, params=params) - - if r.status_code == 200: - return r.json() - else: - try: - error = r.json()["message"] - except ValueError: - error = r.text - if str(r.status_code).startswith("4"): - raise FastLabelInvalidException(error, r.status_code) - else: - raise FastLabelException(error, r.status_code) - - def __deleterequest(self, endpoint: str, params=None) -> dict: - """Makes a delete request to an endpoint. - If an error occurs, assumes that endpoint returns JSON as: - { 'statusCode': XXX, - 'error': 'I failed' } - """ - params = params or {} - headers = { - "Content-Type": "application/json", - "Authorization": self.access_token, - } - r = requests.delete( - FASTLABEL_ENDPOINT + endpoint, headers=headers, params=params - ) - - if r.status_code != 204: - try: - error = r.json()["message"] - except ValueError: - error = r.text - if str(r.status_code).startswith("4"): - raise FastLabelInvalidException(error, r.status_code) - else: - raise FastLabelException(error, r.status_code) - - def __postrequest(self, endpoint, payload=None): - """Makes a post request to an endpoint. - If an error occurs, assumes that endpoint returns JSON as: - { 'statusCode': XXX, - 'error': 'I failed' } - """ - payload = payload or {} - headers = { - "Content-Type": "application/json", - "Authorization": self.access_token, - } - r = requests.post(FASTLABEL_ENDPOINT + endpoint, - json=payload, headers=headers) - - if r.status_code == 200: - return r.json() - else: - try: - error = r.json()["message"] - except ValueError: - error = r.text - if str(r.status_code).startswith("4"): - raise FastLabelInvalidException(error, r.status_code) - else: - raise FastLabelException(error, r.status_code) - - def __putrequest(self, endpoint, payload=None): - """Makes a put request to an endpoint. - If an error occurs, assumes that endpoint returns JSON as: - { 'statusCode': XXX, - 'error': 'I failed' } - """ - payload = payload or {} - headers = { - "Content-Type": "application/json", - "Authorization": self.access_token, - } - r = requests.put(FASTLABEL_ENDPOINT + endpoint, - json=payload, headers=headers) - - if r.status_code == 200: - return r.json() - else: - try: - error = r.json()["message"] - except ValueError: - error = r.text - if str(r.status_code).startswith("4"): - raise FastLabelInvalidException(error, r.status_code) - else: - raise FastLabelException(error, r.status_code) + # Task Find def find_image_task(self, task_id: str) -> dict: """ Find a signle image task. """ endpoint = "tasks/image/" + task_id - return self.__getrequest(endpoint) + return self.api.get_request(endpoint) + + def find_image_task_by_name(self, project: str, task_name: str) -> dict: + """ + Find a signle image task by name. + + project is slug of your project. (Required) + task_name is a task name. (Required) + """ + tasks = self.get_image_tasks(project=project, task_name=task_name) + if not tasks: + return None + return tasks[0] def find_image_classification_task(self, task_id: str) -> dict: """ Find a signle image classification task. """ endpoint = "tasks/image/classification/" + task_id - return self.__getrequest(endpoint) + return self.api.get_request(endpoint) + + def find_image_classification_task_by_name(self, project: str, task_name: str) -> dict: + """ + Find a signle image classification task by name. + + project is slug of your project. (Required) + task_name is a task name. (Required) + """ + tasks = self.get_image_classification_tasks( + project=project, task_name=task_name) + if not tasks: + return None + return tasks[0] def find_multi_image_task(self, task_id: str) -> dict: """ Find a signle multi image task. """ endpoint = "tasks/multi-image/" + task_id - return self.__getrequest(endpoint) + return self.api.get_request(endpoint) + + def find_multi_image_task_by_name(self, project: str, task_name: str) -> dict: + """ + Find a signle multi image task by name. + + project is slug of your project. (Required) + task_name is a task name. (Required) + """ + tasks = self.get_multi_image_tasks( + project=project, task_name=task_name) + if not tasks: + return None + return tasks[0] def find_video_task(self, task_id: str) -> dict: """ Find a signle video task. """ endpoint = "tasks/video/" + task_id - return self.__getrequest(endpoint) + return self.api.get_request(endpoint) + + def find_video_task_by_name(self, project: str, task_name: str) -> dict: + """ + Find a signle video task by name. + + project is slug of your project. (Required) + task_name is a task name. (Required) + """ + tasks = self.get_video_tasks( + project=project, task_name=task_name) + if not tasks: + return None + return tasks[0] + + # Task Get def get_image_tasks( self, project: str, status: str = None, tags: list = [], + task_name: str = None, offset: int = None, limit: int = 100, ) -> list: @@ -170,26 +115,33 @@ def get_image_tasks( project is slug of your project. (Required) status can be 'registered', 'in_progress', 'completed', 'skipped', 'in_review', 'send_backed', 'approved', 'customer_in_review', 'customer_send_backed', 'customer_approved'. (Optional) tags is a list of tag. (Optional) + task_name is a task name. (Optional) offset is the starting position number to fetch. (Optional) limit is the max number to fetch. (Optional) """ + if limit > 1000: + raise FastLabelInvalidException( + "Limit must be less than or equal to 1000.", 422) endpoint = "tasks/image" params = {"project": project} if status: params["status"] = status if tags: params["tags"] = tags + if task_name: + params["taskName"] = task_name if offset: params["offset"] = offset if limit: params["limit"] = limit - return self.__getrequest(endpoint, params=params) + return self.api.get_request(endpoint, params=params) def get_image_classification_tasks( self, project: str, status: str = None, tags: list = [], + task_name: str = None, offset: int = None, limit: int = 100, ) -> list: @@ -209,20 +161,23 @@ def get_image_classification_tasks( params["status"] = status if tags: params["tags"] = tags + if task_name: + params["taskName"] = task_name if offset: params["offset"] = offset if limit: params["limit"] = limit - return self.__getrequest(endpoint, params=params) + return self.api.get_request(endpoint, params=params) def get_multi_image_tasks( self, project: str, status: str = None, tags: list = [], + task_name: str = None, offset: int = None, limit: int = 10, - ) -> dict: + ) -> list: """ Returns a list of multi image tasks. Returns up to 10 at a time, to get more, set offset as the starting position to fetch. @@ -242,20 +197,23 @@ def get_multi_image_tasks( params["status"] = status if tags: params["tags"] = tags + if task_name: + params["taskName"] = task_name if offset: params["offset"] = offset if limit: params["limit"] = limit - return self.__getrequest(endpoint, params=params) + return self.api.get_request(endpoint, params=params) def get_video_tasks( self, project: str, status: str = None, tags: list = [], + task_name: str = None, offset: int = None, limit: int = 10, - ) -> dict: + ) -> list: """ Returns a list of video tasks. Returns up to 10 at a time, to get more, set offset as the starting position to fetch. @@ -263,6 +221,7 @@ def get_video_tasks( project is slug of your project. (Required) status can be 'registered', 'in_progress', 'completed', 'skipped', 'in_review', 'send_backed', 'approved', 'customer_in_review', 'customer_send_backed', 'customer_approved'. (Optional) tags is a list of tag. (Optional) + task_name is a task name. (Optional) offset is the starting position number to fetch. (Optional) limit is the max number to fetch. (Optional) """ @@ -275,11 +234,43 @@ def get_video_tasks( params["status"] = status if tags: params["tags"] = tags + if task_name: + params["taskName"] = task_name + if offset: + params["offset"] = offset + if limit: + params["limit"] = limit + return self.api.get_request(endpoint, params=params) + + def get_task_id_name_map( + self, project: str, + offset: int = None, + limit: int = 1000, + ) -> list: + """ + Returns a map of task ids and names. + e.g.) { + "88e74507-07b5-4607-a130-cb6316ca872c", "01_cat.jpg", + "fe2c24a4-8270-46eb-9c78-bb7281c8bdgs", "02_cat.jpg" + } + Returns up to 1000 at a time, to get more, set offset as the starting position to fetch. + + project is slug of your project. (Required) + offset is the starting position number to fetch. (Optional) + limit is the max number to fetch. (Optional) + """ + if limit > 1000: + raise FastLabelInvalidException( + "Limit must be less than or equal to 1000.", 422) + endpoint = "tasks/map/id-name" + params = {"project": project} if offset: params["offset"] = offset if limit: params["limit"] = limit - return self.__getrequest(endpoint, params=params) + return self.api.get_request(endpoint, params=params) + + # Task Create def create_image_task( self, @@ -301,10 +292,10 @@ def create_image_task( tags is a list of tag to be set in advance. (Optional) """ endpoint = "tasks/image" - if not self.__is_image_supported_ext(file_path): + if not utils.is_image_supported_ext(file_path): raise FastLabelInvalidException( "Supported extensions are png, jpg, jpeg.", 422) - file = self.__base64_encode(file_path) + file = utils.base64_encode(file_path) payload = {"project": project, "name": name, "file": file} if status: payload["status"] = status @@ -314,7 +305,7 @@ def create_image_task( payload["annotations"] = annotations if tags: payload["tags"] = tags - return self.__postrequest(endpoint, payload=payload) + return self.api.post_request(endpoint, payload=payload) def create_image_classification_task( self, @@ -336,10 +327,10 @@ def create_image_classification_task( tags is a list of tag to be set in advance. (Optional) """ endpoint = "tasks/image/classification" - if not self.__is_image_supported_ext(file_path): + if not utils.is_image_supported_ext(file_path): raise FastLabelInvalidException( "Supported extensions are png, jpg, jpeg.", 422) - file = self.__base64_encode(file_path) + file = utils.base64_encode(file_path) payload = {"project": project, "name": name, "file": file} if status: payload["status"] = status @@ -347,7 +338,7 @@ def create_image_classification_task( payload["attributes"] = attributes if tags: payload["tags"] = tags - return self.__postrequest(endpoint, payload=payload) + return self.api.post_request(endpoint, payload=payload) def create_multi_image_task( self, @@ -376,10 +367,10 @@ def create_multi_image_task( file_paths = glob.glob(os.path.join(folder_path, "*")) contents = [] for file_path in file_paths: - if not self.__is_image_supported_ext(file_path): + if not utils.is_image_supported_ext(file_path): raise FastLabelInvalidException( "Supported extensions are png, jpg, jpeg.", 422) - file = self.__base64_encode(file_path) + file = utils.base64_encode(file_path) contents.append({ "name": os.path.basename(file_path), "file": file @@ -391,7 +382,7 @@ def create_multi_image_task( payload["annotations"] = annotations if tags: payload["tags"] = tags - return self.__postrequest(endpoint, payload=payload) + return self.api.post_request(endpoint, payload=payload) def create_video_task( self, @@ -411,16 +402,18 @@ def create_video_task( tags is a list of tag to be set in advance. (Optional) """ endpoint = "tasks/video" - if not self.__is_video_supported_ext(file_path): + if not utils.is_video_supported_ext(file_path): raise FastLabelInvalidException( "Supported extensions are mp4.", 422) - file = self.__base64_encode(file_path) + file = utils.base64_encode(file_path) payload = {"project": project, "name": name, "file": file} if status: payload["status"] = status if tags: payload["tags"] = tags - return self.__postrequest(endpoint, payload=payload) + return self.api.post_request(endpoint, payload=payload) + + # Task Update def update_task( self, @@ -441,167 +434,166 @@ def update_task( payload["status"] = status if tags: payload["tags"] = tags - return self.__putrequest(endpoint, payload=payload) + return self.api.put_request(endpoint, payload=payload) + + # Task Delete def delete_task(self, task_id: str) -> None: """ Delete a single task. """ endpoint = "tasks/" + task_id - self.__deleterequest(endpoint) + self.api.delete_request(endpoint) + + # Task Convert def to_coco(self, tasks: list) -> dict: - # Get categories - categories = self.__get_categories(tasks) - - # Get images and annotations - images = [] - annotations = [] - annotation_id = 0 - image_id = 0 - for task in tasks: - if task["height"] == 0 or task["width"] == 0: - continue - - image_id += 1 - image = { - "file_name": task["name"], - "height": task["height"], - "width": task["width"], - "id": image_id, - } - images.append(image) - - data = [{"annotation": annotation, "categories": categories, - "image": image} for annotation in task["annotations"]] - with ThreadPoolExecutor(max_workers=8) as executor: - results = executor.map(self.__to_annotation, data) - - for result in results: - annotation_id += 1 - if not result: - continue - result["id"] = annotation_id - annotations.append(result) - - return { - "images": images, - "categories": categories, - "annotations": annotations, - } + """ + Convert tasks to COCO format. + """ - def __base64_encode(self, file_path: str) -> str: - with open(file_path, "rb") as f: - return base64.b64encode(f.read()).decode() - - def __is_image_supported_ext(self, file_path: str) -> bool: - return file_path.lower().endswith(('.png', '.jpg', '.jpeg')) - - def __is_video_supported_ext(self, file_path: str) -> bool: - return file_path.lower().endswith(('.mp4')) - - def __get_categories(self, tasks: list) -> list: - values = [] - for task in tasks: - for annotation in task["annotations"]: - if annotation["type"] != AnnotationType.bbox.value and annotation["type"] != AnnotationType.polygon.value: - continue - values.append(annotation["value"]) - values = list(set(values)) - - categories = [] - for index, value in enumerate(values): - category = { - "supercategory": value, - "id": index + 1, - "name": value - } - categories.append(category) - return categories - - def __to_annotation(self, data: dict) -> dict: - annotation = data["annotation"] - categories = data["categories"] - image = data["image"] - points = annotation["points"] - annotation_type = annotation["type"] - annotation_id = 0 - - if annotation_type != AnnotationType.bbox.value and annotation_type != AnnotationType.polygon.value: - return None - if not points or len(points) == 0: - return None - if annotation_type == AnnotationType.bbox.value and (int(points[0]) == int(points[2]) or int(points[1]) == int(points[3])): + return converters.to_coco(tasks) + + # Annotation + + def find_annotation(self, annotation_id: str) -> dict: + """ + Find an annotation. + """ + endpoint = "annotations/" + annotation_id + return self.api.get_request(endpoint) + + def find_annotation_by_value(self, project: str, value: str) -> dict: + """ + Find an annotation by value. + """ + annotations = self.get_annotations(project=project, value=value) + if not annotations: return None + return annotations[0] + + def get_annotations( + self, + project: str, + value: str = None, + offset: int = None, + limit: int = 10, + ) -> list: + """ + Returns a list of annotations. + Returns up to 1000 at a time, to get more, set offset as the starting position to fetch. + + project is slug of your project. (Required) + value is an unique identifier of annotation in your project. (Required) + offset is the starting position number to fetch. (Optional) + limit is the max number to fetch. (Optional) + """ + if limit > 1000: + raise FastLabelInvalidException( + "Limit must be less than or equal to 1000.", 422) + endpoint = "annotations" + params = {"project": project} + if value: + params["value"] = value + if offset: + params["offset"] = offset + if limit: + params["limit"] = limit + return self.api.get_request(endpoint, params=params) + + def create_annotation( + self, + project: str, + type: str, + value: str, + title: str, + color: str, + attributes: list = [] + ) -> str: + """ + Create an annotation. - category = self.__get_category_by_name(categories, annotation["value"]) - - return self.__get_annotation( - annotation_id, points, category["id"], image, annotation_type) - - def __get_category_by_name(self, categories: list, name: str) -> str: - category = [ - category for category in categories if category["name"] == name][0] - return category - - def __get_annotation(self, id_: int, points: list, category_id: int, image: dict, annotation_type: str) -> dict: - annotation = {} - annotation["segmentation"] = [points] - annotation["iscrowd"] = 0 - annotation["area"] = self.__calc_area(annotation_type, points) - annotation["image_id"] = image["id"] - annotation["bbox"] = self.__to_bbox(points) - annotation["category_id"] = category_id - annotation["id"] = id_ - return annotation - - def __to_bbox(self, points: list) -> list: - points_splitted = [points[idx:idx + 2] - for idx in range(0, len(points), 2)] - polygon_geo = geojson.Polygon(points_splitted) - coords = np.array(list(geojson.utils.coords(polygon_geo))) - left_top_x = coords[:, 0].min() - left_top_y = coords[:, 1].min() - right_bottom_x = coords[:, 0].max() - right_bottom_y = coords[:, 1].max() - - return [ - left_top_x, # x - left_top_y, # y - right_bottom_x - left_top_x, # width - right_bottom_y - left_top_y, # height - ] - - def __calc_area(self, annotation_type: str, points: list) -> float: - area = 0 - if annotation_type == AnnotationType.bbox.value: - width = points[0] - points[2] - height = points[1] - points[3] - area = width * height - elif annotation_type == AnnotationType.polygon.value: - x = points[0::2] - y = points[1::2] - area = 0.5 * np.abs(np.dot(x, np.roll(y, 1)) - - np.dot(y, np.roll(x, 1))) - return area - - -class AnnotationType(Enum): - bbox = "bbox" - polygon = "polygon" - keypoint = "keypoint" - classification = "classification" - line = "line" - segmentation = "segmentation" - - -class FastLabelException(Exception): - def __init__(self, message, errcode): - super(FastLabelException, self).__init__( - " {}".format(errcode, message) - ) - self.code = errcode - - -class FastLabelInvalidException(FastLabelException, ValueError): - pass + project is slug of your project. (Required) + type can be 'bbox', 'polygon', 'keypoint', 'classification', 'line', 'segmentation'. (Required) + value is an unique identifier of annotation in your project. (Required) + title is a display name of value. (Required) + color is hex color code like #ffffff. (Required) + attributes is a list of attribute. (Optional) + """ + endpoint = "annotations" + payload = { + "project": project, + "type": type, + "value": value, + "title": title, + "color": color + } + if attributes: + payload["attributes"] = attributes + return self.api.post_request(endpoint, payload=payload) + + def create_classification_annotation( + self, + project: str, + attributes: list + ) -> str: + """ + Create a classification annotation. + + project is slug of your project. (Required) + attributes is a list of attribute. (Required) + """ + endpoint = "annotations/classification" + payload = {"project": project, "attributes": attributes} + return self.api.post_request(endpoint, payload=payload) + + def update_annotation( + self, + annotation_id: str, + value: str = None, + title: str = None, + color: str = None, + attributes: list = [] + ) -> str: + """ + Update an annotation. + + annotation_id is an id of the annotation. (Required) + value is an unique identifier of annotation in your project. (Optional) + title is a display name of value. (Optional) + color is hex color code like #ffffff. (Optional) + attributes is a list of attribute. (Optional) + """ + endpoint = "annotations/" + annotation_id + payload = {} + if value: + payload["value"] = value + if title: + payload["title"] = title + if color: + payload["color"] = color + if attributes: + payload["attributes"] = attributes + return self.api.put_request(endpoint, payload=payload) + + def update_classification_annotation( + self, + annotation_id: str, + attributes: list + ) -> str: + """ + Update a classification annotation. + + annotation_id is an id of the annotation. (Required) + attributes is a list of attribute. (Required) + """ + endpoint = "annotations/classification/" + annotation_id + payload = {"attributes": attributes} + return self.api.put_request(endpoint, payload=payload) + + def delete_annotation(self, annotation_id: str) -> None: + """ + Delete an annotation. + """ + endpoint = "annotations/" + annotation_id + self.api.delete_request(endpoint) diff --git a/fastlabel/api.py b/fastlabel/api.py new file mode 100644 index 0000000..fafca89 --- /dev/null +++ b/fastlabel/api.py @@ -0,0 +1,120 @@ +import os +import requests + +from .exceptions import FastLabelException, FastLabelInvalidException + +FASTLABEL_ENDPOINT = "https://api.fastlabel.ai/v1/" + + +class Api: + + access_token = None + + def __init__(self): + if not os.environ.get("FASTLABEL_ACCESS_TOKEN"): + raise ValueError("FASTLABEL_ACCESS_TOKEN is not configured.") + self.access_token = "Bearer " + \ + os.environ.get("FASTLABEL_ACCESS_TOKEN") + + def get_request(self, endpoint: str, params=None) -> dict: + """Makes a get request to an endpoint. + If an error occurs, assumes that endpoint returns JSON as: + { 'statusCode': XXX, + 'error': 'I failed' } + """ + params = params or {} + headers = { + "Content-Type": "application/json", + "Authorization": self.access_token, + } + r = requests.get(FASTLABEL_ENDPOINT + endpoint, + headers=headers, params=params) + + if r.status_code == 200: + return r.json() + else: + try: + error = r.json()["message"] + except ValueError: + error = r.text + if str(r.status_code).startswith("4"): + raise FastLabelInvalidException(error, r.status_code) + else: + raise FastLabelException(error, r.status_code) + + def delete_request(self, endpoint: str, params=None) -> dict: + """Makes a delete request to an endpoint. + If an error occurs, assumes that endpoint returns JSON as: + { 'statusCode': XXX, + 'error': 'I failed' } + """ + params = params or {} + headers = { + "Content-Type": "application/json", + "Authorization": self.access_token, + } + r = requests.delete( + FASTLABEL_ENDPOINT + endpoint, headers=headers, params=params + ) + + if r.status_code != 204: + try: + error = r.json()["message"] + except ValueError: + error = r.text + if str(r.status_code).startswith("4"): + raise FastLabelInvalidException(error, r.status_code) + else: + raise FastLabelException(error, r.status_code) + + def post_request(self, endpoint, payload=None): + """Makes a post request to an endpoint. + If an error occurs, assumes that endpoint returns JSON as: + { 'statusCode': XXX, + 'error': 'I failed' } + """ + payload = payload or {} + headers = { + "Content-Type": "application/json", + "Authorization": self.access_token, + } + r = requests.post(FASTLABEL_ENDPOINT + endpoint, + json=payload, headers=headers) + + if r.status_code == 200: + return r.json() + else: + try: + error = r.json()["message"] + except ValueError: + error = r.text + if str(r.status_code).startswith("4"): + raise FastLabelInvalidException(error, r.status_code) + else: + raise FastLabelException(error, r.status_code) + + def put_request(self, endpoint, payload=None): + """Makes a put request to an endpoint. + If an error occurs, assumes that endpoint returns JSON as: + { 'statusCode': XXX, + 'error': 'I failed' } + """ + payload = payload or {} + headers = { + "Content-Type": "application/json", + "Authorization": self.access_token, + } + r = requests.put(FASTLABEL_ENDPOINT + endpoint, + json=payload, headers=headers) + + if r.status_code == 200: + return r.json() + else: + try: + error = r.json()["message"] + except ValueError: + error = r.text + if str(r.status_code).startswith("4"): + raise FastLabelInvalidException(error, r.status_code) + else: + raise FastLabelException(error, r.status_code) diff --git a/fastlabel/converters.py b/fastlabel/converters.py new file mode 100644 index 0000000..c554756 --- /dev/null +++ b/fastlabel/converters.py @@ -0,0 +1,149 @@ +from enum import Enum +from concurrent.futures import ThreadPoolExecutor + +import geojson +import numpy as np + + +class AnnotationType(Enum): + bbox = "bbox" + polygon = "polygon" + keypoint = "keypoint" + classification = "classification" + line = "line" + segmentation = "segmentation" + + +def to_coco(tasks: list) -> dict: + """ + Convert tasks to COCO format. + """ + # Get categories + categories = __get_categories(tasks) + + # Get images and annotations + images = [] + annotations = [] + annotation_id = 0 + image_id = 0 + for task in tasks: + if task["height"] == 0 or task["width"] == 0: + continue + + image_id += 1 + image = { + "file_name": task["name"], + "height": task["height"], + "width": task["width"], + "id": image_id, + } + images.append(image) + + data = [{"annotation": annotation, "categories": categories, + "image": image} for annotation in task["annotations"]] + with ThreadPoolExecutor(max_workers=8) as executor: + results = executor.map(__to_annotation, data) + + for result in results: + annotation_id += 1 + if not result: + continue + result["id"] = annotation_id + annotations.append(result) + + return { + "images": images, + "categories": categories, + "annotations": annotations, + } + + +def __get_categories(tasks: list) -> list: + values = [] + for task in tasks: + for annotation in task["annotations"]: + if annotation["type"] != AnnotationType.bbox.value and annotation["type"] != AnnotationType.polygon.value: + continue + values.append(annotation["value"]) + values = list(set(values)) + + categories = [] + for index, value in enumerate(values): + category = { + "supercategory": value, + "id": index + 1, + "name": value + } + categories.append(category) + return categories + + +def __to_annotation(data: dict) -> dict: + annotation = data["annotation"] + categories = data["categories"] + image = data["image"] + points = annotation["points"] + annotation_type = annotation["type"] + annotation_id = 0 + + if annotation_type != AnnotationType.bbox.value and annotation_type != AnnotationType.polygon.value: + return None + if not points or len(points) == 0: + return None + if annotation_type == AnnotationType.bbox.value and (int(points[0]) == int(points[2]) or int(points[1]) == int(points[3])): + return None + + category = __get_category_by_name(categories, annotation["value"]) + + return __get_annotation( + annotation_id, points, category["id"], image, annotation_type) + + +def __get_category_by_name(categories: list, name: str) -> str: + category = [ + category for category in categories if category["name"] == name][0] + return category + + +def __get_annotation(id_: int, points: list, category_id: int, image: dict, annotation_type: str) -> dict: + annotation = {} + annotation["segmentation"] = [points] + annotation["iscrowd"] = 0 + annotation["area"] = __calc_area(annotation_type, points) + annotation["image_id"] = image["id"] + annotation["bbox"] = __to_bbox(points) + annotation["category_id"] = category_id + annotation["id"] = id_ + return annotation + + +def __to_bbox(points: list) -> list: + points_splitted = [points[idx:idx + 2] + for idx in range(0, len(points), 2)] + polygon_geo = geojson.Polygon(points_splitted) + coords = np.array(list(geojson.utils.coords(polygon_geo))) + left_top_x = coords[:, 0].min() + left_top_y = coords[:, 1].min() + right_bottom_x = coords[:, 0].max() + right_bottom_y = coords[:, 1].max() + + return [ + left_top_x, # x + left_top_y, # y + right_bottom_x - left_top_x, # width + right_bottom_y - left_top_y, # height + ] + + +def __calc_area(annotation_type: str, points: list) -> float: + area = 0 + if annotation_type == AnnotationType.bbox.value: + width = points[0] - points[2] + height = points[1] - points[3] + area = width * height + elif annotation_type == AnnotationType.polygon.value: + x = points[0::2] + y = points[1::2] + area = 0.5 * np.abs(np.dot(x, np.roll(y, 1)) - + np.dot(y, np.roll(x, 1))) + return area diff --git a/fastlabel/exceptions.py b/fastlabel/exceptions.py new file mode 100644 index 0000000..7f0a4ba --- /dev/null +++ b/fastlabel/exceptions.py @@ -0,0 +1,10 @@ +class FastLabelException(Exception): + def __init__(self, message, errcode): + super(FastLabelException, self).__init__( + " {}".format(errcode, message) + ) + self.code = errcode + + +class FastLabelInvalidException(FastLabelException, ValueError): + pass diff --git a/fastlabel/utils.py b/fastlabel/utils.py new file mode 100644 index 0000000..abaf5ea --- /dev/null +++ b/fastlabel/utils.py @@ -0,0 +1,14 @@ +import base64 + + +def base64_encode(file_path: str) -> str: + with open(file_path, "rb") as f: + return base64.b64encode(f.read()).decode() + + +def is_image_supported_ext(file_path: str) -> bool: + return file_path.lower().endswith(('.png', '.jpg', '.jpeg')) + + +def is_video_supported_ext(file_path: str) -> bool: + return file_path.lower().endswith(('.mp4')) diff --git a/setup.py b/setup.py index 9779540..ab109ed 100644 --- a/setup.py +++ b/setup.py @@ -8,7 +8,7 @@ setuptools.setup( name="fastlabel", - version="0.7.0", + version="0.7.1", author="eisuke-ueta", author_email="eisuke.ueta@fastlabel.ai", description="The official Python SDK for FastLabel API, the Data Platform for AI",