from typing import List from pycs.interfaces.AnnotatedMediaFile import AnnotatedMediaFile from pycs.interfaces.MediaFile import MediaFile class Pipeline: """ pipeline interface that should be implemented by model developers """ def __init__(self, root_folder: str, distribution: dict): """ prepare everything needed to run jobs later :param root_folder: relative path to model folder :param distribution: dict parsed from distribution.json """ raise NotImplementedError def close(self): """ is called everytime a pipeline is not needed anymore and should be used to free native resources :return: """ raise NotImplementedError def execute(self, file: MediaFile) -> List[dict]: """ receive a job, execute it and return the predicted result :param file: which should be analyzed :return: """ raise NotImplementedError def fit(self, files: List[AnnotatedMediaFile]): """ receive a list of annotated media files and adapt the underlying model :param files: list of annotated media files :return: """ raise NotImplementedError @staticmethod def create_labeled_image_result(label: int): """ create a labeled-image result dictionary :param label: label identifier :return: dict """ return { 'type': 'labeled-image', 'label': label } @staticmethod def create_bounding_box_result(x: float, y: float, w: float, h: float, label=None, frame=None): # pylint: disable=too-many-arguments # pylint: disable=invalid-name """ create a bounding-box result dictionary :param x: relative x coordinate [0, 1] :param y: relative y coordinate [0, 1] :param w: relative width [0, 1] :param h: relative height [0, 1] :param label: label identifier :param frame: frame index :return: dict """ result = { 'type': 'bounding-box', 'x': x, 'y': y, 'w': w, 'h': h } if label is not None: result['label'] = label if frame is not None: result['frame'] = frame return result