PyCS/pycs/pipeline/PipelineManager.py

from os import path

from eventlet import tpool

from pycs.pipeline.Job import Job
from pycs.projects.Project import Project


class PipelineManager:
    def __init__(self, project: Project):
        code_path = path.join(project['model']['path'], project['model']['code']['module'])
        module_name = code_path.replace('/', '.').replace('\\', '.')
        class_name = project['model']['code']['class']

        mod = __import__(module_name, fromlist=[class_name])
        cl = getattr(mod, class_name)

        self.project = project
        self.pipeline = cl(project['model']['path'], project['model'])

    def close(self):
        print('PipelineManager', 'close')
        self.pipeline.close()

    def run(self, media_file):
        # create job list
        # TODO update job progress
        job = Job('detect-faces', self.project['id'], media_file)
        result = tpool.execute(lambda p, j: p.execute(j), self.pipeline, job)

        # remove existing pipeline predictions from media_fle
        media_file.remove_pipeline_results()

        # add new predictions
        for prediction in result.predictions:
            media_file.add_result(prediction, origin='pipeline')

    def fit(self):
        print('PipelineManager', 'fit')

        data = []
        for identifier in self.project['data']:
            obj = self.project['data'][identifier]
            media = {
                'name': obj['name'],
                'extension': obj['extension'],
                'size': obj['size'],
                'path': path.join('projects', self.project['id'], 'data', identifier + obj['extension']),
                'predictionResults': []
            }

            for prediction_identifier in obj['predictionResults']:
                prediction = obj['predictionResults'][prediction_identifier]
                if prediction['origin'] != 'user':
                    continue

                if 'x' not in prediction:
                    media['predictionResults'].append({
                        'type': 'labeled-image',
                        'label': prediction['label']
                    })
                else:
                    if 'label' in prediction:
                        media['predictionResults'].append({
                            'type': 'labeled-bounding-box',
                            'x': prediction['x'],
                            'y': prediction['y'],
                            'w': prediction['w'],
                            'h': prediction['h'],
                            'label': prediction['label']
                        })
                    else:
                        media['predictionResults'].append({
                            'type': 'bounding-box',
                            'x': prediction['x'],
                            'y': prediction['y'],
                            'w': prediction['w'],
                            'h': prediction['h']
                        })

            data.append(media)

        self.pipeline.fit(data)