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
        """
        pass

    def close(self):
        """
        is called everytime a pipeline is not needed anymore and should be used
        to free native resources

        :return:
        """
        pass

    def collections(self) -> List[dict]:
        """
        is called while initializing a pipeline to receive available
        collections

        :return: list of collections or None
        """
        return []

    @staticmethod
    def create_collection(reference: str,
                          name: str,
                          description: str = None,
                          autoselect: bool = False) -> dict:
        """
        create a collection dict

        :param reference: unique reference
        :param name: collection name
        :param description: collection description
        :param autoselect: show this collection by default if it contains elements
        :return: collection dict
        """
        return {
            'reference': reference,
            'name': name,
            'description': description,
            'autoselect': autoselect
        }

    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

    @staticmethod
    def create_collection_result(reference: str) -> dict:
        """
        create a collection result dictionary

        :param reference: use None to remove this file's collection
        :return: dict
        """
        return {
            'type': 'collection',
            'reference': reference
        }

    @staticmethod
    def create_labeled_image_result(label: int) -> dict:
        """
        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) -> dict:
        # 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

    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