PyCS/pycs/interfaces/Pipeline.py

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