PyCS/pycs/frontend/endpoints/pipelines/FitModel.py

from flask import make_response, request, abort
from flask.views import View

from pycs.database.Database import Database
from pycs.database.Project import Project
from pycs.interfaces.MediaStorage import MediaStorage
from pycs.jobs.JobGroupBusyException import JobGroupBusyException
from pycs.jobs.JobRunner import JobRunner
from pycs.util.PipelineCache import PipelineCache


class FitModel(View):
    """
    use annotated data to fit a model
    """
    # pylint: disable=arguments-differ
    methods = ['POST']

    def __init__(self, jobs: JobRunner, pipelines: PipelineCache):
        # pylint: disable=invalid-name
        self.jobs = jobs
        self.pipelines = pipelines

    def dispatch_request(self, project_id):
        # extract request data
        data = request.get_json(force=True)

        if not data.get('fit', False):
            return abort(400)

        # find project
        project = Project.query.get(project_id)
        if project is None:
            return abort(404)

        # create job
        try:
            self.jobs.run(project,
                          'Model Interaction',
                          f'{project.name} (fit model with new data)',
                          f'{project.name}/model-interaction',
                          self.load_and_fit, self.pipelines, project.id)
        except JobGroupBusyException:
            return abort(400)

        return make_response()

    @staticmethod
    def load_and_fit(pipelines: PipelineCache, project_id: int):
        pipeline = None

        project = Project.query.get(project_id)
        model = project.model()
        storage = MediaStorage(project_id)

        # load pipeline
        try:
            pipeline = pipelines.load_from_root_folder(project, model.root_folder)
            yield from pipeline.fit(storage)
        except TypeError:
            pass
        finally:
            if pipeline is not None:
                pipelines.free_instance(model.root_folder)