PyCS/pycs/interfaces/Pipeline.py

from typing import List

from pycs.interfaces.MediaFile import MediaFile
from pycs.interfaces.MediaBoundingBox import MediaBoundingBox
from pycs.interfaces.MediaStorage import MediaStorage


class Pipeline:
    """
    pipeline interface that should be implemented by model developers
    """

    #pylint: disable=unnecessary-pass
    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

    #pylint: disable=unnecessary-pass
    def close(self):
        """
        is called everytime a pipeline is not needed anymore and should be used
        to free native resources

        :return:
        """
        pass

    #pylint: disable=no-self-use
    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, storage: MediaStorage, file: MediaFile):
        """
        receive a file, create predictions and add them to the object

        :param storage: database abstraction object
        :param file: which should be analyzed
        """
        raise NotImplementedError

    def pure_inference(self, storage: MediaStorage, file: MediaFile, bounding_boxes: List[MediaBoundingBox]):
        """
        receive a file and a list of bounding boxes and only create a
        classification for the given bounding boxes.

        :param storage: database abstraction object
        :param file: which should be analyzed
        :param bounding_boxes: only perform inference for the given bounding boxes

        :return: labels for the given bounding boxes
        """
        raise NotImplementedError

    def fit(self, storage: MediaStorage):
        """
        receive a list of annotated media files and adapt the underlying model

        :param storage: database abstraction object
        """
        raise NotImplementedError